PandasTA 源码解析（七）（2）https://developer.aliyun.com/article/1506112
.\pandas-ta\pandas_ta\momentum\tsi.py
# -*- coding: utf-8 -*- # 从 pandas 库导入 DataFrame 类 from pandas import DataFrame # 从 pandas_ta.overlap 模块导入 ema, ma 函数 from pandas_ta.overlap import ema, ma # 从 pandas_ta.utils 模块导入 get_drift, get_offset, verify_series 函数 from pandas_ta.utils import get_drift, get_offset, verify_series # 定义 True Strength Index (TSI) 指标函数 def tsi(close, fast=None, slow=None, signal=None, scalar=None, mamode=None, drift=None, offset=None, **kwargs): """Indicator: True Strength Index (TSI)""" # 验证参数有效性 # 如果 fast 参数存在且大于 0，则将其转换为整数，否则设为默认值 13 fast = int(fast) if fast and fast > 0 else 13 # 如果 slow 参数存在且大于 0，则将其转换为整数，否则设为默认值 25 slow = int(slow) if slow and slow > 0 else 25 # 如果 signal 参数存在且大于 0，则将其转换为整数，否则设为默认值 13 signal = int(signal) if signal and signal > 0 else 13 # 如果 close 序列为 None，则返回 None if close is None: return # 如果 scalar 存在，则将其转换为浮点数，否则设为默认值 100 scalar = float(scalar) if scalar else 100 # 获取漂移值，用于处理偏移 drift = get_drift(drift) # 获取偏移量，用于处理偏移 offset = get_offset(offset) # 如果 mamode 不是字符串类型，则将其设为默认值 "ema" mamode = mamode if isinstance(mamode, str) else "ema" # 如果 kwargs 中包含 "length" 键，则将其移除 if "length" in kwargs: kwargs.pop("length") # 计算结果 # 计算 close 序列的一阶差分 diff = close.diff(drift) # 计算 slow 期 EMA slow_ema = ema(close=diff, length=slow, **kwargs) # 计算 fast 期 EMA fast_slow_ema = ema(close=slow_ema, length=fast, **kwargs) # 计算绝对差分 abs_diff = diff.abs() # 计算 slow 期绝对差分的 EMA abs_slow_ema = ema(close=abs_diff, length=slow, **kwargs) # 计算 fast 期绝对差分的 EMA abs_fast_slow_ema = ema(close=abs_slow_ema, length=fast, **kwargs) # 计算 TSI tsi = scalar * fast_slow_ema / abs_fast_slow_ema # 计算 TSI 的信号线 tsi_signal = ma(mamode, tsi, length=signal) # 处理偏移 if offset != 0: tsi = tsi.shift(offset) tsi_signal = tsi_signal.shift(offset) # 处理填充 if "fillna" in kwargs: tsi.fillna(kwargs["fillna"], inplace=True) tsi_signal.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: tsi.fillna(method=kwargs["fill_method"], inplace=True) tsi_signal.fillna(method=kwargs["fill_method"], inplace=True) # 命名并分类化指标 tsi.name = f"TSI_{fast}_{slow}_{signal}" tsi_signal.name = f"TSIs_{fast}_{slow}_{signal}" tsi.category = tsi_signal.category = "momentum" # 准备返回的 DataFrame df = DataFrame({tsi.name: tsi, tsi_signal.name: tsi_signal}) df.name = f"TSI_{fast}_{slow}_{signal}" df.category = "momentum" return df # 设置 tsi 函数的文档字符串 tsi.__doc__ = \ """True Strength Index (TSI) The True Strength Index is a momentum indicator used to identify short-term swings while in the direction of the trend as well as determining overbought and oversold conditions. Sources: https://www.investopedia.com/terms/t/tsi.asp Calculation: Default Inputs: fast=13, slow=25, signal=13, scalar=100, drift=1 EMA = Exponential Moving Average diff = close.diff(drift) slow_ema = EMA(diff, slow) fast_slow_ema = EMA(slow_ema, slow) abs_diff_slow_ema = absolute_diff_ema = EMA(ABS(diff), slow) abema = abs_diff_fast_slow_ema = EMA(abs_diff_slow_ema, fast) TSI = scalar * fast_slow_ema / abema Signal = EMA(TSI, signal) Args: close (pd.Series): Series of 'close's fast (int): The short period. Default: 13 slow (int): The long period. Default: 25 signal (int): The signal period. Default: 13 """ scalar (float): How much to magnify. Default: 100 # 定义一个浮点数变量 scalar，表示放大倍数，默认值为 100 mamode (str): Moving Average of TSI Signal Line. See ```help(ta.ma)```py. Default: 'ema' # 定义一个字符串变量 mamode，表示 TSI 信号线的移动平均方式，默认值为 'ema'，可查看 ta.ma 的帮助文档 drift (int): The difference period. Default: 1 # 定义一个整数变量 drift，表示差分周期，默认值为 1 offset (int): How many periods to offset the result. Default: 0 # 定义一个整数变量 offset，表示结果的偏移周期数，默认值为 0 # 函数参数说明部分，kwargs表示可变关键字参数 Kwargs: # fillna参数，用于填充缺失值的值，类型为任意 fillna (value, optional): pd.DataFrame.fillna(value) # fill_method参数，填充方法的类型 fill_method (value, optional): Type of fill method # 返回值说明部分，返回一个pandas DataFrame对象，包含tsi和signal两列 Returns: # 返回的pandas DataFrame对象，包含tsi和signal两列数据 pd.DataFrame: tsi, signal.
.\pandas-ta\pandas_ta\momentum\uo.py
# -*- coding: utf-8 -*- # 导入 DataFrame 类 from pandas import DataFrame # 导入 Imports 类 from pandas_ta import Imports # 导入 get_drift 和 get_offset 函数 from pandas_ta.utils import get_drift, get_offset, verify_series # 定义 Ultimate Oscillator（UO）指标函数 def uo(high, low, close, fast=None, medium=None, slow=None, fast_w=None, medium_w=None, slow_w=None, talib=None, drift=None, offset=None, **kwargs): """Indicator: Ultimate Oscillator (UO)""" # 验证参数 fast = int(fast) if fast and fast > 0 else 7 fast_w = float(fast_w) if fast_w and fast_w > 0 else 4.0 medium = int(medium) if medium and medium > 0 else 14 medium_w = float(medium_w) if medium_w and medium_w > 0 else 2.0 slow = int(slow) if slow and slow > 0 else 28 slow_w = float(slow_w) if slow_w and slow_w > 0 else 1.0 _length = max(fast, medium, slow) # 验证 high、low、close 序列长度 high = verify_series(high, _length) low = verify_series(low, _length) close = verify_series(close, _length) # 获取漂移和偏移量 drift = get_drift(drift) offset = get_offset(offset) # 设置是否使用 talib 模式 mode_tal = bool(talib) if isinstance(talib, bool) else True # 如果 high、low、close 有任何一个为 None，则返回空 if high is None or low is None or close is None: return # 计算结果 if Imports["talib"] and mode_tal: # 使用 talib 计算 UO 指标 from talib import ULTOSC uo = ULTOSC(high, low, close, fast, medium, slow) else: # 否则，使用自定义计算方法 tdf = DataFrame({ "high": high, "low": low, f"close_{drift}": close.shift(drift) }) # 获取最大最小值 max_h_or_pc = tdf.loc[:, ["high", f"close_{drift}"]].max(axis=1) min_l_or_pc = tdf.loc[:, ["low", f"close_{drift}"]].min(axis=1) del tdf # 计算 buying pressure（bp）和 true range（tr） bp = close - min_l_or_pc tr = max_h_or_pc - min_l_or_pc # 计算 fast、medium、slow 平均值 fast_avg = bp.rolling(fast).sum() / tr.rolling(fast).sum() medium_avg = bp.rolling(medium).sum() / tr.rolling(medium).sum() slow_avg = bp.rolling(slow).sum() / tr.rolling(slow).sum() # 计算总权重和加权平均值 total_weight = fast_w + medium_w + slow_w weights = (fast_w * fast_avg) + (medium_w * medium_avg) + (slow_w * slow_avg) uo = 100 * weights / total_weight # 考虑偏移量 if offset != 0: uo = uo.shift(offset) # 处理填充 if "fillna" in kwargs: uo.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: uo.fillna(method=kwargs["fill_method"], inplace=True) # 设置指标名称和分类 uo.name = f"UO_{fast}_{medium}_{slow}" uo.category = "momentum" # 返回 Ultimate Oscillator（UO）指标 return uo # 设置 Ultimate Oscillator（UO）指标的文档字符串 uo.__doc__ = \ """Ultimate Oscillator (UO) The Ultimate Oscillator is a momentum indicator over three different periods. It attempts to correct false divergence trading signals. Sources: https://www.tradingview.com/wiki/Ultimate_Oscillator_(UO) Calculation: Default Inputs: fast=7, medium=14, slow=28, fast_w=4.0, medium_w=2.0, slow_w=1.0, drift=1 min_low_or_pc = close.shift(drift).combine(low, min) max_high_or_pc = close.shift(drift).combine(high, max) bp = buying pressure = close - min_low_or_pc tr = true range = max_high_or_pc - min_low_or_pc fast_avg = SUM(bp, fast) / SUM(tr, fast) """ # 计算中等速度的平均值，中等速度报告的总和除以中等速度报告的数量 medium_avg = SUM(bp, medium) / SUM(tr, medium) # 计算慢速度的平均值，慢速度报告的总和除以慢速度报告的数量 slow_avg = SUM(bp, slow) / SUM(tr, slow) # 计算所有速度权重的总和 total_weight = fast_w + medium_w + slow_w # 计算加权平均值，每个速度报告的权重乘以其对应的平均值，然后相加 weights = (fast_w * fast_avg) + (medium_w * medium_avg) + (slow_w * slow_avg) # 计算不确定性指标（UO），即权重总和乘以100除以所有权重的总和 UO = 100 * weights / total_weight # 参数说明： # high (pd.Series): 'high' 数据序列 # low (pd.Series): 'low' 数据序列 # close (pd.Series): 'close' 数据序列 # fast (int): 快速 %K 周期。默认值：7 # medium (int): 慢速 %K 周期。默认值：14 # slow (int): 慢速 %D 周期。默认值：28 # fast_w (float): 快速 %K 周期。默认值：4.0 # medium_w (float): 慢速 %K 周期。默认值：2.0 # slow_w (float): 慢速 %D 周期。默认值：1.0 # talib (bool): 如果安装了 TA Lib 并且 talib 为 True，则返回 TA Lib 版本。默认值：True # drift (int): 差异周期。默认值：1 # offset (int): 结果的偏移周期数。默认值：0 # 可选参数： # fillna (value, optional): pd.DataFrame.fillna(value) 的填充值 # fill_method (value, optional): 填充方法的类型 # 返回值： # pd.Series: 生成的新特征序列
.\pandas-ta\pandas_ta\momentum\willr.py
# -*- coding: utf-8 -*- # 从 pandas_ta 库中导入 Imports 类 from pandas_ta import Imports # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series def willr(high, low, close, length=None, talib=None, offset=None, **kwargs): """Indicator: William's Percent R (WILLR)""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则默认为 14 length = int(length) if length and length > 0 else 14 # 如果 kwargs 中存在 "min_periods"，并且其值不为 None，则将其转换为整数，否则使用 length 的值 min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length # 确定最终使用的长度为 length 和 min_periods 中的较大值 _length = max(length, min_periods) # 验证 high、low、close 系列，并设置它们的长度为 _length high = verify_series(high, _length) low = verify_series(low, _length) close = verify_series(close, _length) # 获取偏移量 offset = get_offset(offset) # 确定是否使用 TA-Lib mode_tal = bool(talib) if isinstance(talib, bool) else True # 如果 high、low、close 中有任何一个为 None，则返回空 if high is None or low is None or close is None: return # 计算结果 if Imports["talib"] and mode_tal: # 如果 TA-Lib 可用且 mode_tal 为 True，则使用 TA-Lib 中的 WILLR 函数计算 from talib import WILLR willr = WILLR(high, low, close, length) else: # 否则，使用自定义方法计算 WILLR # 计算长度为 length 的最低低点 lowest_low = low.rolling(length, min_periods=min_periods).min() # 计算长度为 length 的最高高点 highest_high = high.rolling(length, min_periods=min_periods).max() # 计算 WILLR willr = 100 * ((close - lowest_low) / (highest_high - lowest_low) - 1) # 根据偏移量对结果进行偏移 if offset != 0: willr = willr.shift(offset) # 处理填充值 if "fillna" in kwargs: willr.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: willr.fillna(method=kwargs["fill_method"], inplace=True) # 设置指标名称和类别 willr.name = f"WILLR_{length}" willr.category = "momentum" # 返回计算结果 return willr # 设置函数文档字符串 willr.__doc__ = \ """William's Percent R (WILLR) William's Percent R is a momentum oscillator similar to the RSI that attempts to identify overbought and oversold conditions. Sources: https://www.tradingview.com/wiki/Williams_%25R_(%25R) Calculation: Default Inputs: length=20 LL = low.rolling(length).min() HH = high.rolling(length).max() WILLR = 100 * ((close - LL) / (HH - LL) - 1) 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 talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib version. Default: True 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. """
