PandasTA 源码解析（十二）（3）

# -*- coding: utf-8 -*-
# 从 numpy 库中导入 log10 函数并重命名为 npLog10
from numpy import log10 as npLog10
# 从 numpy 库中导入 log 函数并重命名为 npLn
from numpy import log as npLn
# 从 pandas_ta 库中导入 volatility 模块中的 atr 函数
from pandas_ta.volatility import atr
# 从 pandas_ta 库中导入 utils 模块中的 get_drift, get_offset, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, verify_series
# 定义函数 chop，计算 Choppiness Index (CHOP)
def chop(high, low, close, length=None, atr_length=None, ln=None, scalar=None, drift=None, offset=None, **kwargs):
    """Indicator: Choppiness Index (CHOP)"""
    # 验证参数
    length = int(length) if length and length > 0 else 14
    atr_length = int(atr_length) if atr_length is not None and atr_length > 0 else 1
    ln = bool(ln) if isinstance(ln, bool) else False
    scalar = float(scalar) if scalar else 100
    high = verify_series(high, length)
    low = verify_series(low, length)
    close = verify_series(close, length)
    drift = get_drift(drift)
    offset = get_offset(offset)
    # 如果 high、low、close 中有任何一个为 None，则返回
    if high is None or low is None or close is None: return
    # 计算结果
    diff = high.rolling(length).max() - low.rolling(length).min()
    atr_ = atr(high=high, low=low, close=close, length=atr_length)
    atr_sum = atr_.rolling(length).sum()
    chop = scalar
    if ln:
        chop *= (npLn(atr_sum) - npLn(diff)) / npLn(length)
    else:
        chop *= (npLog10(atr_sum) - npLog10(diff)) / npLog10(length)
    # 偏移结果
    if offset != 0:
        chop = chop.shift(offset)
    # 处理填充值
    if "fillna" in kwargs:
        chop.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        chop.fillna(method=kwargs["fill_method"], inplace=True)
    # 命名和分类
    chop.name = f"CHOP{'ln' if ln else ''}_{length}_{atr_length}_{scalar}"
    chop.category = "trend"
    return chop
# 设置 chop 函数的文档字符串
chop.__doc__ = \
"""Choppiness Index (CHOP)
The Choppiness Index was created by Australian commodity trader
E.W. Dreiss and is designed to determine if the market is choppy
(trading sideways) or not choppy (trading within a trend in either
direction). Values closer to 100 implies the underlying is choppier
whereas values closer to 0 implies the underlying is trending.
Sources:
    https://www.tradingview.com/scripts/choppinessindex/
    https://www.motivewave.com/studies/choppiness_index.htm
Calculation:
    Default Inputs:
        length=14, scalar=100, drift=1
    HH = high.rolling(length).max()
    LL = low.rolling(length).min()
    ATR_SUM = SUM(ATR(drift), length)
    CHOP = scalar * (LOG10(ATR_SUM) - LOG10(HH - LL))
    CHOP /= LOG10(length)
Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 14
    atr_length (int): Length for ATR. Default: 1
    ln (bool): If True, uses ln otherwise log10. Default: False
    scalar (float): How much to magnify. Default: 100
    drift (int): The difference period. Default: 1
    offset (int): How many periods to offset the result. Default: 0
Kwargs:
    fillna (value, optional): pd.DataFrame.fillna(value)
    fill_method (value, optional): Type of fill method
Returns:
"""
    # 创建一个 Pandas Series 对象，表示生成了一个新特征
    pd.Series: New feature generated.
# 这是一个空的字符串，通常用作多行注释的起始

# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta 库中导入 atr 函数
from pandas_ta.volatility import atr
# 从 pandas_ta 库中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义函数 cksp，计算 Chande Kroll Stop (CKSP) 指标
def cksp(high, low, close, p=None, x=None, q=None, tvmode=None, offset=None, **kwargs):
    """Indicator: Chande Kroll Stop (CKSP)"""
    
    # 验证参数
    p = int(p) if p and p > 0 else 10
    x = float(x) if x and x > 0 else 1 if tvmode is True else 3
    q = int(q) if q and q > 0 else 9 if tvmode is True else 20
    _length = max(p, q, x)
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    close = verify_series(close, _length)
    if high is None or low is None or close is None: return
    offset = get_offset(offset)
    tvmode = tvmode if isinstance(tvmode, bool) else True
    mamode = "rma" if tvmode is True else "sma"
    # 计算结果
    atr_ = atr(high=high, low=low, close=close, length=p, mamode=mamode)
    long_stop_ = high.rolling(p).max() - x * atr_
    long_stop = long_stop_.rolling(q).max()
    short_stop_ = low.rolling(p).min() + x * atr_
    short_stop = short_stop_.rolling(q).min()
    # 偏移
    if offset != 0:
        long_stop = long_stop.shift(offset)
        short_stop = short_stop.shift(offset)
    # 处理填充
    if "fillna" in kwargs:
        long_stop.fillna(kwargs["fillna"], inplace=True)
        short_stop.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        long_stop.fillna(method=kwargs["fill_method"], inplace=True)
        short_stop.fillna(method=kwargs["fill_method"], inplace=True)
    # 命名和分类
    _props = f"_{p}_{x}_{q}"
    long_stop.name = f"CKSPl{_props}"
    short_stop.name = f"CKSPs{_props}"
    long_stop.category = short_stop.category = "trend"
    # 准备返回的 DataFrame
    ckspdf = DataFrame({long_stop.name: long_stop, short_stop.name: short_stop})
    ckspdf.name = f"CKSP{_props}"
    ckspdf.category = long_stop.category
    return ckspdf
# 设置函数 cksp 的文档字符串
cksp.__doc__ = \
"""Chande Kroll Stop (CKSP)
The Tushar Chande and Stanley Kroll in their book
“The New Technical Trader”. It is a trend-following indicator,
identifying your stop by calculating the average true range of
the recent market volatility. The indicator defaults to the implementation
found on tradingview but it provides the original book implementation as well,
which differs by the default periods and moving average mode. While the trading
view implementation uses the Welles Wilder moving average, the book uses a
simple moving average.
Sources:
    https://www.multicharts.com/discussion/viewtopic.php?t=48914
    "The New Technical Trader", Wikey 1st ed. ISBN 9780471597803, page 95
Calculation:
    Default Inputs:
        p=10, x=1, q=9, tvmode=True
    ATR = Average True Range
    LS0 = high.rolling(p).max() - x * ATR(length=p)
    LS = LS0.rolling(q).max()
    SS0 = high.rolling(p).min() + x * ATR(length=p)
    SS = SS0.rolling(q).min()
Args:
"""
    # 'close'是一个包含收盘价的Series对象
    # p是ATR和第一个停止期的值，以整数表示。在两种模式下默认值均为10
    # x是ATR的标量值，在Trading View模式下默认值为1，在其他模式下默认值为3
    # q是第二个停止期的值，以整数表示。在Trading View模式下默认值为9，在其他模式下默认值为20
    # tvmode是一个布尔值，表示是否使用Trading View模式或书中实现模式。默认为True表示使用Trading View模式
    # offset是结果的偏移周期数。默认值为0
# 定义函数的参数列表，这里使用了可选参数
Kwargs:
    fillna (value, optional): pd.DataFrame.fillna(value)  # 填充缺失值的数值
    fill_method (value, optional): Type of fill method  # 填充方法的类型
# 返回值说明
Returns:
    pd.DataFrame: long and short columns.  # 返回一个包含长列和短列的 Pandas 数据帧

# -*- coding: utf-8 -*-
# 从 numpy 导入 exp 函数并重命名为 npExp
from numpy import exp as npExp
# 从 pandas 导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.utils 导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
def decay(close, kind=None, length=None, mode=None, offset=None, **kwargs):
    """Indicator: Decay"""
    # 验证参数
    # 如果 length 存在且大于 0，则转换为整数，否则设置为默认值 5
    length = int(length) if length and length > 0 else 5
    # 如果 mode 是字符串，则转换为小写，否则设置为默认值 "linear"
    mode = mode.lower() if isinstance(mode, str) else "linear"
    # 验证 close 是否是有效的 Series，长度为 length
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)
    # 如果 close 为空，则返回 None
    if close is None: return
    # 计算结果
    # 默认模式为线性模式
    _mode = "L"
    # 如果 mode 是 "exp" 或 kind 是 "exponential"，则使用指数模式
    if mode == "exp" or kind == "exponential":
        _mode = "EXP"
        # 计算差异，利用指数函数 exp(-length)
        diff = close.shift(1) - npExp(-length)
    else:  # 默认为 "linear"
        # 计算差异，利用线性函数 (1 / length)
        diff = close.shift(1) - (1 / length)
    # 将第一个元素设置为 close 的第一个值
    diff[0] = close[0]
    # 创建 DataFrame，包含 close、diff 和 0 列
    tdf = DataFrame({"close": close, "diff": diff, "0": 0})
    # 计算最大值
    ld = tdf.max(axis=1)
    # 偏移结果
    if offset != 0:
        ld = ld.shift(offset)
    # 处理填充
    # 如果 kwargs 中包含 "fillna"，则使用指定值填充缺失值
    if "fillna" in kwargs:
        ld.fillna(kwargs["fillna"], inplace=True)
    # 如果 kwargs 中包含 "fill_method"，则使用指定的填充方法
    if "fill_method" in kwargs:
        ld.fillna(method=kwargs["fill_method"], inplace=True)
    # 命名和分类
    # 设置 Series 的名称为模式和长度的组合
    ld.name = f"{_mode}DECAY_{length}"
    # 设置 Series 的分类为 "trend"
    ld.category = "trend"
    return ld
# 设置 decay 函数的文档字符串
decay.__doc__ = \
"""Decay
Creates a decay moving forward from prior signals like crosses. The default is
"linear". Exponential is optional as "exponential" or "exp".
Sources:
    https://tulipindicators.org/decay
Calculation:
    Default Inputs:
        length=5, mode=None
    if mode == "exponential" or mode == "exp":
        max(close, close[-1] - exp(-length), 0)
    else:
        max(close, close[-1] - (1 / length), 0)
Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 1
    mode (str): If 'exp' then "exponential" decay. Default: 'linear'
    offset (int): How many periods to offset the result. Default: 0
Kwargs:
    fillna (value, optional): pd.DataFrame.fillna(value)
    fill_method (value, optional): Type of fill method
Returns:
    pd.Series: New feature generated.
"""