PandasTA 源码解析（七）（1）https://developer.aliyun.com/article/1506110
.\pandas-ta\pandas_ta\momentum\stochrsi.py
# -*- coding: utf-8 -*- # 从 pandas 库中导入 DataFrame 类 from pandas import DataFrame # 从 .rsi 模块中导入 rsi 函数 from .rsi import rsi # 从 pandas_ta.overlap 模块中导入 ma 函数 from pandas_ta.overlap import ma # 从 pandas_ta.utils 模块中导入 get_offset, non_zero_range, verify_series 函数 from pandas_ta.utils import get_offset, non_zero_range, verify_series # 定义函数 stochrsi，计算 Stochastic RSI Oscillator (STOCHRSI) def stochrsi(close, length=None, rsi_length=None, k=None, d=None, mamode=None, offset=None, **kwargs): """Indicator: Stochastic RSI Oscillator (STOCHRSI)""" # 校验参数 length = length if length and length > 0 else 14 rsi_length = rsi_length if rsi_length and rsi_length > 0 else 14 k = k if k and k > 0 else 3 d = d if d and d > 0 else 3 # 校验 close 序列 close = verify_series(close, max(length, rsi_length, k, d)) offset = get_offset(offset) # 确定 mamode 默认为 "sma"，如果 mamode 不是字符串则设为 "sma" mamode = mamode if isinstance(mamode, str) else "sma" # 如果 close 为 None，返回空值 if close is None: return # 计算结果 # 计算 RSI rsi_ = rsi(close, length=rsi_length) # 计算最低 RSI lowest_rsi = rsi_.rolling(length).min() # 计算最高 RSI highest_rsi = rsi_.rolling(length).max() # 计算 stoch 值 stoch = 100 * (rsi_ - lowest_rsi) stoch /= non_zero_range(highest_rsi, lowest_rsi) # 计算 STOCHRSI 的 %K 线和 %D 线 stochrsi_k = ma(mamode, stoch, length=k) stochrsi_d = ma(mamode, stochrsi_k, length=d) # 偏移 if offset != 0: stochrsi_k = stochrsi_k.shift(offset) stochrsi_d = stochrsi_d.shift(offset) # 处理填充值 if "fillna" in kwargs: stochrsi_k.fillna(kwargs["fillna"], inplace=True) stochrsi_d.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: stochrsi_k.fillna(method=kwargs["fill_method"], inplace=True) stochrsi_d.fillna(method=kwargs["fill_method"], inplace=True) # 命名并分类 _name = "STOCHRSI" _props = f"_{length}_{rsi_length}_{k}_{d}" stochrsi_k.name = f"{_name}k{_props}" stochrsi_d.name = f"{_name}d{_props}" stochrsi_k.category = stochrsi_d.category = "momentum" # 准备返回的 DataFrame data = {stochrsi_k.name: stochrsi_k, stochrsi_d.name: stochrsi_d} df = DataFrame(data) df.name = f"{_name}{_props}" df.category = stochrsi_k.category return df # 设置 stochrsi 函数的文档字符串 stochrsi.__doc__ = \ """Stochastic (STOCHRSI) "Stochastic RSI and Dynamic Momentum Index" was created by Tushar Chande and Stanley Kroll and published in Stock & Commodities V.11:5 (189-199) It is a range-bound oscillator with two lines moving between 0 and 100. The first line (%K) displays the current RSI in relation to the period's high/low range. The second line (%D) is a Simple Moving Average of the %K line. The most common choices are a 14 period %K and a 3 period SMA for %D. Sources: https://www.tradingview.com/wiki/Stochastic_(STOCH) Calculation: Default Inputs: length=14, rsi_length=14, k=3, d=3 RSI = Relative Strength Index SMA = Simple Moving Average RSI = RSI(high, low, close, rsi_length) LL = lowest RSI for last rsi_length periods HH = highest RSI for last rsi_length periods STOCHRSI = 100 * (RSI - LL) / (HH - LL) STOCHRSIk = SMA(STOCHRSI, k) STOCHRSId = SMA(STOCHRSIk, d) Args: high (pd.Series): Series of 'high's """ low (pd.Series): 存储股价的最低价序列 close (pd.Series): 存储股价的收盘价序列 length (int): STOCHRSI 的周期。默认为 14 rsi_length (int): RSI 的周期。默认为 14 k (int): 快速 %K 的周期。默认为 3 d (int): 慢速 %K 的周期。默认为 3 mamode (str): 查看 ```help(ta.ma)```py。默认为 'sma'（简单移动平均） offset (int): 结果偏移的周期数。默认为 0 # 参数说明部分，描述函数的参数和返回值 Kwargs: fillna (value, optional): pd.DataFrame.fillna(value) fill_method (value, optional): Type of fill method # 返回值说明部分，描述函数返回的数据类型和列名 Returns: pd.DataFrame: RSI %K, RSI %D columns.
.\pandas-ta\pandas_ta\momentum\td_seq.py
# -*- coding: utf-8 -*- # 从 numpy 库中导入 where 函数并重命名为 npWhere from numpy import where as npWhere # 从 pandas 库中导入 DataFrame 和 Series 类 from pandas import DataFrame, Series # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义函数 td_seq，用于计算 Tom Demark Sequential（TD_SEQ）指标 def td_seq(close, asint=None, offset=None, **kwargs): """Indicator: Tom Demark Sequential (TD_SEQ)""" # 验证参数 close 是否为有效的 Series 对象 close = verify_series(close) # 获取偏移量 offset = get_offset(offset) # 如果 asint 不为布尔值，则设置为 False asint = asint if isinstance(asint, bool) else False # 获取参数中的 show_all，如果不存在则设置默认值为 True show_all = kwargs.setdefault("show_all", True) # 定义函数 true_sequence_count，用于计算连续的真值序列数量 def true_sequence_count(series: Series): # 找到最后一个为 False 的索引 index = series.where(series == False).last_valid_index() if index is None: # 如果索引为空，则返回序列的总数 return series.count() else: # 否则，返回索引之后的序列数量 s = series[series.index > index] return s.count() # 定义函数 calc_td，用于计算 TD_SEQ def calc_td(series: Series, direction: str, show_all: bool): # 计算 TD_SEQ 的布尔值 td_bool = series.diff(4) > 0 if direction=="up" else series.diff(4) < 0 # 根据布尔值计算 TD_SEQ 数值 td_num = npWhere( td_bool, td_bool.rolling(13, min_periods=0).apply(true_sequence_count), 0 ) td_num = Series(td_num) if show_all: # 如果 show_all 为 True，则保留所有 TD_SEQ 值 td_num = td_num.mask(td_num == 0) else: # 否则，只保留在 6 到 9 之间的 TD_SEQ 值 td_num = td_num.mask(~td_num.between(6,9)) return td_num # 计算上升序列的 TD_SEQ up_seq = calc_td(close, "up", show_all) # 计算下降序列的 TD_SEQ down_seq = calc_td(close, "down", show_all) # 如果需要将结果转换为整数 if asint: if up_seq.hasnans and down_seq.hasnans: # 填充缺失值为 0 up_seq.fillna(0, inplace=True) down_seq.fillna(0, inplace=True) # 转换结果为整数类型 up_seq = up_seq.astype(int) down_seq = down_seq.astype(int) # 如果偏移量不为 0 if offset != 0: # 对结果进行偏移 up_seq = up_seq.shift(offset) down_seq = down_seq.shift(offset) # 处理填充值 if "fillna" in kwargs: up_seq.fillna(kwargs["fillna"], inplace=True) down_seq.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: up_seq.fillna(method=kwargs["fill_method"], inplace=True) down_seq.fillna(method=kwargs["fill_method"], inplace=True) # 设置上升序列和下降序列的名称和分类 up_seq.name = f"TD_SEQ_UPa" if show_all else f"TD_SEQ_UP" down_seq.name = f"TD_SEQ_DNa" if show_all else f"TD_SEQ_DN" up_seq.category = down_seq.category = "momentum" # 准备要返回的 DataFrame df = DataFrame({up_seq.name: up_seq, down_seq.name: down_seq}) df.name = "TD_SEQ" df.category = up_seq.category return df # 设置函数文档字符串 td_seq.__doc__ = \ """TD Sequential (TD_SEQ) Tom DeMark's Sequential indicator attempts to identify a price point where an uptrend or a downtrend exhausts itself and reverses. Sources: https://tradetrekker.wordpress.com/tdsequential/ Calculation: Compare current close price with 4 days ago price, up to 13 days. For the consecutive ascending or descending price sequence, display 6th to 9th day value. Args: close (pd.Series): Series of 'close's asint (bool): If True, fillnas with 0 and change type to int. Default: False offset (int): How many periods to offset the result. Default: 0 Kwargs: """ # 定义函数参数show_all，用于控制展示范围，默认为True，即展示1到13；如果设置为False，仅展示6到9。 show_all (bool): Show 1 - 13. If set to False, show 6 - 9. Default: True # 定义函数参数fillna，用于填充缺失值，参数value为填充的数值，默认为空。 fillna (value, optional): pd.DataFrame.fillna(value) # 返回类型说明：返回的是一个 Pandas DataFrame，其中包含了生成的新特征。
.\pandas-ta\pandas_ta\momentum\trix.py
# -*- coding: utf-8 -*- # 从 pandas 库中导入 DataFrame 类 from pandas import DataFrame # 从 pandas_ta 库中导入 overlap 模块下的 ema 函数 from pandas_ta.overlap.ema import ema # 从 pandas_ta 库中导入 utils 模块 from pandas_ta.utils import get_drift, get_offset, verify_series # 定义 Trix 指标函数，用于计算 Trix (TRIX) 指标 def trix(close, length=None, signal=None, scalar=None, drift=None, offset=None, **kwargs): """Indicator: Trix (TRIX)""" # 验证参数 length = int(length) if length and length > 0 else 30 signal = int(signal) if signal and signal > 0 else 9 scalar = float(scalar) if scalar else 100 close = verify_series(close, max(length, signal)) drift = get_drift(drift) offset = get_offset(offset) # 如果 close 为空，则返回空值 if close is None: return # 计算结果 ema1 = ema(close=close, length=length, **kwargs) ema2 = ema(close=ema1, length=length, **kwargs) ema3 = ema(close=ema2, length=length, **kwargs) trix = scalar * ema3.pct_change(drift) trix_signal = trix.rolling(signal).mean() # 偏移 if offset != 0: trix = trix.shift(offset) trix_signal = trix_signal.shift(offset) # 处理填充 if "fillna" in kwargs: trix.fillna(kwargs["fillna"], inplace=True) trix_signal.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: trix.fillna(method=kwargs["fill_method"], inplace=True) trix_signal.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和类别 trix.name = f"TRIX_{length}_{signal}" trix_signal.name = f"TRIXs_{length}_{signal}" trix.category = trix_signal.category = "momentum" # 准备返回的 DataFrame df = DataFrame({trix.name: trix, trix_signal.name: trix_signal}) df.name = f"TRIX_{length}_{signal}" df.category = "momentum" return df # 设置 trix 函数的文档字符串 trix.__doc__ = \ """Trix (TRIX) TRIX is a momentum oscillator to identify divergences. Sources: https://www.tradingview.com/wiki/TRIX Calculation: Default Inputs: length=18, drift=1 EMA = Exponential Moving Average ROC = Rate of Change ema1 = EMA(close, length) ema2 = EMA(ema1, length) ema3 = EMA(ema2, length) TRIX = 100 * ROC(ema3, drift) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 18 signal (int): It's period. Default: 9 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: pd.Series: New feature generated. """
PandasTA 源码解析（七）（3）https://developer.aliyun.com/article/1506115
