PandasTA 源码解析（十二）（2）https://developer.aliyun.com/article/1506222
.\pandas-ta\pandas_ta\trend\chop.py
# -*- 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. # 这是一个空的字符串，通常用作多行注释的起始
.\pandas-ta\pandas_ta\trend\cksp.py
# -*- 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 数据帧
.\pandas-ta\pandas_ta\trend\decay.py
# -*- 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. """
