PandasTA 源码解析（十六）（1）https://developer.aliyun.com/article/1506251
.\pandas-ta\pandas_ta\volatility\rvi.py
# -*- coding: utf-8 -*- # 从pandas_ta.overlap模块导入ma函数 from pandas_ta.overlap import ma # 从pandas_ta.statistics模块导入stdev函数 from pandas_ta.statistics import stdev # 从pandas_ta.utils模块导入get_drift和get_offset函数 from pandas_ta.utils import get_drift, get_offset # 从pandas_ta.utils模块导入unsigned_differences和verify_series函数 from pandas_ta.utils import unsigned_differences, verify_series # 定义Relative Volatility Index (RVI)指标函数 def rvi(close, high=None, low=None, length=None, scalar=None, refined=None, thirds=None, mamode=None, drift=None, offset=None, **kwargs): """Indicator: Relative Volatility Index (RVI)""" # 验证参数 length = int(length) if length and length > 0 else 14 # 如果length为正整数则保留，否则默认为14 scalar = float(scalar) if scalar and scalar > 0 else 100 # 如果scalar为正浮点数则保留，否则默认为100 refined = False if refined is None else refined # 如果refined为None则默认为False thirds = False if thirds is None else thirds # 如果thirds为None则默认为False mamode = mamode if isinstance(mamode, str) else "ema" # 如果mamode为字符串则保留，否则默认为"ema" close = verify_series(close, length) # 验证close是否为有效序列 drift = get_drift(drift) # 获取漂移参数 offset = get_offset(offset) # 获取偏移参数 if close is None: return if refined or thirds: high = verify_series(high) # 验证high是否为有效序列 low = verify_series(low) # 验证low是否为有效序列 # 计算结果 def _rvi(source, length, scalar, mode, drift): """RVI""" # 计算标准差 std = stdev(source, length) # 获取正差值和负差值 pos, neg = unsigned_differences(source, amount=drift) # 计算正差值的标准差加权平均 pos_std = pos * std pos_avg = ma(mode, pos_std, length=length) # 计算负差值的标准差加权平均 neg_std = neg * std neg_avg = ma(mode, neg_std, length=length) # 计算RVI指标 result = scalar * pos_avg result /= pos_avg + neg_avg return result _mode = "" if refined: # 如果使用了refined模式 # 计算高价RVI high_rvi = _rvi(high, length, scalar, mamode, drift) # 计算低价RVI low_rvi = _rvi(low, length, scalar, mamode, drift) # 计算RVI rvi = 0.5 * (high_rvi + low_rvi) _mode = "r" # 设置模式为"r" elif thirds: # 如果使用了thirds模式 # 计算高价RVI high_rvi = _rvi(high, length, scalar, mamode, drift) # 计算低价RVI low_rvi = _rvi(low, length, scalar, mamode, drift) # 计算收盘价RVI close_rvi = _rvi(close, length, scalar, mamode, drift) # 计算RVI rvi = (high_rvi + low_rvi + close_rvi) / 3.0 _mode = "t" # 设置模式为"t" else: # 如果未使用refined和thirds模式 # 计算RVI rvi = _rvi(close, length, scalar, mamode, drift) # 偏移 if offset != 0: rvi = rvi.shift(offset) # 处理填充 if "fillna" in kwargs: rvi.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: rvi.fillna(method=kwargs["fill_method"], inplace=True) # 命名和归类 rvi.name = f"RVI{_mode}_{length}" rvi.category = "volatility" return rvi # 设置RVI指标的文档字符串 rvi.__doc__ = \ """Relative Volatility Index (RVI) The Relative Volatility Index (RVI) was created in 1993 and revised in 1995. Instead of adding up price changes like RSI based on price direction, the RVI adds up standard deviations based on price direction. Sources: https://www.tradingview.com/wiki/Keltner_Channels_(KC) Calculation: Default Inputs: length=14, scalar=100, refined=None, thirds=None EMA = Exponential Moving Average STDEV = Standard Deviation UP = STDEV(src, length) IF src.diff() > 0 ELSE 0 DOWN = STDEV(src, length) IF src.diff() <= 0 ELSE 0 UPSUM = EMA(UP, length) DOWNSUM = EMA(DOWN, length """ # 计算相对强度指数（RSI），其计算公式为 RVI = scalar * (UPSUM / (UPSUM + DOWNSUM)) RVI = scalar * (UPSUM / (UPSUM + DOWNSUM)) # 定义一个函数，用于计算布林带指标（Bollinger Bands） def bollinger_hband_indicator(high, low, close, length=14, scalar=100, refined=False, thirds=False, mamode='ema', offset=0, fillna=None, fill_method=None): """ Args: high (pd.Series): 最高价的序列 low (pd.Series): 最低价的序列 close (pd.Series): 收盘价的序列 length (int): 短周期。默认值为14 scalar (float): 缩放带的正浮点数。默认值为100 refined (bool): 使用“精炼”计算，即 RVI(high) 和 RVI(low) 的平均值，而不是 RVI(close)。默认值为False thirds (bool): 使用最高价、最低价和收盘价的平均值。默认值为False mamode (str): 参见 ```help(ta.ma)```py。默认值为'ema' offset (int): 结果的偏移周期数。默认值为0 Kwargs: fillna (value, optional): pd.DataFrame.fillna(value) 的参数 fill_method (value, optional): 填充方法的类型 Returns: pd.DataFrame: 包含 lower、basis、upper 列的数据框 """
.\pandas-ta\pandas_ta\volatility\thermo.py
# -*- coding: utf-8 -*- # 从 pandas 库中导入 DataFrame 类 from pandas import DataFrame # 从 pandas_ta 库中的 overlap 模块导入 ma 函数 from pandas_ta.overlap import ma # 从 pandas_ta 库中的 utils 模块导入 get_offset、verify_series、get_drift 函数 from pandas_ta.utils import get_offset, verify_series, get_drift # 定义函数 thermo，计算 Elder's Thermometer 指标 def thermo(high, low, length=None, long=None, short=None, mamode=None, drift=None, offset=None, **kwargs): """Indicator: Elders Thermometer (THERMO)""" # 验证参数 # 如果 length 存在且大于 0，则转换为整数，否则默认为 20 length = int(length) if length and length > 0 else 20 # 如果 long 存在且大于 0，则转换为浮点数，否则默认为 2 long = float(long) if long and long > 0 else 2 # 如果 short 存在且大于 0，则转换为浮点数，否则默认为 0.5 short = float(short) if short and short > 0 else 0.5 # 如果 mamode 是字符串，则保持不变，否则默认为 "ema" mamode = mamode if isinstance(mamode, str) else "ema" # 验证 high 和 low 系列数据，长度为 length high = verify_series(high, length) low = verify_series(low, length) # 获取 drift 和 offset drift = get_drift(drift) offset = get_offset(offset) # 从 kwargs 中弹出 asint 参数，默认为 True asint = kwargs.pop("asint", True) # 如果 high 或 low 为 None，则返回 if high is None or low is None: return # 计算结果 # 计算 Elder's Thermometer 的下限 thermoL = (low.shift(drift) - low).abs() # 计算 Elder's Thermometer 的上限 thermoH = (high - high.shift(drift)).abs() # 取较小的值作为 Elder's Thermometer thermo = thermoL thermo = thermo.where(thermoH < thermoL, thermoH) # 索引设置为 high 的索引 thermo.index = high.index # 计算 Elder's Thermometer 的移动平均线 thermo_ma = ma(mamode, thermo, length=length) # 生成信号 thermo_long = thermo < (thermo_ma * long) thermo_short = thermo > (thermo_ma * short) # 二进制输出，用于信号 if asint: thermo_long = thermo_long.astype(int) thermo_short = thermo_short.astype(int) # 调整偏移量 if offset != 0: thermo = thermo.shift(offset) thermo_ma = thermo_ma.shift(offset) thermo_long = thermo_long.shift(offset) thermo_short = thermo_short.shift(offset) # 处理填充 if "fillna" in kwargs: thermo.fillna(kwargs["fillna"], inplace=True) thermo_ma.fillna(kwargs["fillna"], inplace=True) thermo_long.fillna(kwargs["fillna"], inplace=True) thermo_short.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: thermo.fillna(method=kwargs["fill_method"], inplace=True) thermo_ma.fillna(method=kwargs["fill_method"], inplace=True) thermo_long.fillna(method=kwargs["fill_method"], inplace=True) thermo_short.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和分类 _props = f"_{length}_{long}_{short}" thermo.name = f"THERMO{_props}" thermo_ma.name = f"THERMOma{_props}" thermo_long.name = f"THERMOl{_props}" thermo_short.name = f"THERMOs{_props}" thermo.category = thermo_ma.category = thermo_long.category = thermo_short.category = "volatility" # 准备返回的 DataFrame data = { thermo.name: thermo, thermo_ma.name: thermo_ma, thermo_long.name: thermo_long, thermo_short.name: thermo_short } df = DataFrame(data) df.name = f"THERMO{_props}" df.category = thermo.category return df # 为 thermo 函数添加文档字符串 thermo.__doc__ = \ """Elders Thermometer (THERMO) Elder's Thermometer measures price volatility. Sources: https://www.motivewave.com/studies/elders_thermometer.htm # 导入所需的库 import requests # 定义函数`get_tradingview_script`用于获取TradingView上的脚本内容 def get_tradingview_script(url): # 发送GET请求获取指定URL的页面内容 response = requests.get(url) # 返回页面内容的文本 return response.text # 定义变量`script_url`，存储TradingView脚本的URL script_url = "https://www.tradingview.com/script/HqvTuEMW-Elder-s-Market-Thermometer-LazyBear/" # 调用`get_tradingview_script`函数，获取指定URL的脚本内容 script_content = get_tradingview_script(script_url) # 计算热力指标（thermo）和相关指标 Calculation: # 默认输入参数 length=20, drift=1, mamode=EMA, long=2, short=0.5 # EMA为指数移动平均 # 计算低价的漂移 thermoL = (low.shift(drift) - low).abs() # 计算高价的漂移 thermoH = (high - high.shift(drift)).abs() # 选择较大的漂移值 thermo = np.where(thermoH > thermoL, thermoH, thermoL) # 对漂移值进行指数移动平均 thermo_ma = ema(thermo, length) # 判断是否满足买入条件 thermo_long = thermo < (thermo_ma * long) # 判断是否满足卖出条件 thermo_short = thermo > (thermo_ma * short) # 将布尔值转换为整数 thermo_long = thermo_long.astype(int) thermo_short = thermo_short.astype(int) Args: high (pd.Series): 'high' 的序列 low (pd.Series): 'low' 的序列 long(int): 买入因子 short(float): 卖出因子 length (int): 周期。默认值：20 mamode (str): 参见 ```help(ta.ma)```py。默认值：'ema' drift (int): 漂移周期。默认值：1 offset (int): 结果的偏移周期数。默认值：0 Kwargs: fillna (value, optional): pd.DataFrame.fillna(value) fill_method (value, optional): 填充方法的类型 Returns: pd.DataFrame: 包含 thermo, thermo_ma, thermo_long, thermo_short 列的数据框
PandasTA 源码解析（十六）（3）https://developer.aliyun.com/article/1506254
