PandasTA 源码解析（十二）（1）https://developer.aliyun.com/article/1506220
.\pandas-ta\pandas_ta\trend\adx.py
# -*- coding: utf-8 -*- # 导入所需的库和模块 from pandas import DataFrame from pandas_ta.overlap import ma from pandas_ta.volatility import atr from pandas_ta.utils import get_drift, get_offset, verify_series, zero # 定义 ADX 指标函数 def adx(high, low, close, length=None, lensig=None, scalar=None, mamode=None, drift=None, offset=None, **kwargs): """Indicator: ADX""" # 验证参数 length = length if length and length > 0 else 14 lensig = lensig if lensig and lensig > 0 else length mamode = mamode if isinstance(mamode, str) else "rma" 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) if high is None or low is None or close is None: return # 计算 ATR 指标 atr_ = atr(high=high, low=low, close=close, length=length) # 计算上升和下降动向线 up = high - high.shift(drift) # high.diff(drift) dn = low.shift(drift) - low # low.diff(-drift).shift(drift) pos = ((up > dn) & (up > 0)) * up neg = ((dn > up) & (dn > 0)) * dn pos = pos.apply(zero) neg = neg.apply(zero) k = scalar / atr_ dmp = k * ma(mamode, pos, length=length) dmn = k * ma(mamode, neg, length=length) dx = scalar * (dmp - dmn).abs() / (dmp + dmn) adx = ma(mamode, dx, length=lensig) # 偏移 if offset != 0: dmp = dmp.shift(offset) dmn = dmn.shift(offset) adx = adx.shift(offset) # 处理填充值 if "fillna" in kwargs: adx.fillna(kwargs["fillna"], inplace=True) dmp.fillna(kwargs["fillna"], inplace=True) dmn.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: adx.fillna(method=kwargs["fill_method"], inplace=True) dmp.fillna(method=kwargs["fill_method"], inplace=True) dmn.fillna(method=kwargs["fill_method"], inplace=True) # 命名和分类 adx.name = f"ADX_{lensig}" dmp.name = f"DMP_{length}" dmn.name = f"DMN_{length}" adx.category = dmp.category = dmn.category = "trend" # 准备返回的 DataFrame data = {adx.name: adx, dmp.name: dmp, dmn.name: dmn} adxdf = DataFrame(data) adxdf.name = f"ADX_{lensig}" adxdf.category = "trend" return adxdf # 设置 ADX 函数的文档字符串 adx.__doc__ = \ """Average Directional Movement (ADX) Average Directional Movement is meant to quantify trend strength by measuring the amount of movement in a single direction. Sources: https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/average-directional-movement-adx/ TA Lib Correlation: >99% Calculation: DMI ADX TREND 2.0 by @TraderR0BERT, NETWORTHIE.COM // 由 @TraderR0BERT, NETWORTHIE.COM 创建，最后更新日期为 01/26/2016 // DMI 指标 // 分辨率输入选项，用于更高/更低的时间框架 study(title="DMI ADX TREND 2.0", shorttitle="ADX TREND 2.0") adxlen = input(14, title="ADX Smoothing") dilen = input(14, title="DI Length") thold = input(20, title="Threshold") threshold = thold //Script for Indicator dirmov(len) => up = change(high) down = -change(low) truerange = rma(tr, len) plus = fixnan(100 * rma(up > down and up > 0 ? up : 0, len) / truerange) minus = fixnan(100 * rma(down > up and down > 0 ? down : 0, len) / truerange) [plus, minus] adx(dilen, adxlen) => [plus, minus] = dirmov(dilen) sum = plus + minus adx = 100 * rma(abs(plus - minus) / (sum == 0 ? 1 : sum), adxlen) [adx, plus, minus] [sig, up, down] = adx(dilen, adxlen) osob=input(40,title="Exhaustion Level for ADX, default = 40") col = sig >= sig[1] ? green : sig <= sig[1] ? red : gray //Plot Definitions Current Timeframe p1 = plot(sig, color=col, linewidth = 3, title="ADX") p2 = plot(sig, color=col, style=circles, linewidth=3, title="ADX") p3 = plot(up, color=blue, linewidth = 3, title="+DI") p4 = plot(up, color=blue, style=circles, linewidth=3, title="+DI") p5 = plot(down, color=fuchsia, linewidth = 3, title="-DI") p6 = plot(down, color=fuchsia, style=circles, linewidth=3, title="-DI") h1 = plot(threshold, color=black, linewidth =3, title="Threshold") trender = (sig >= up or sig >= down) ? 1 : 0 bgcolor(trender>0?black:gray, transp=85) //Alert Function for ADX crossing Threshold Up_Cross = crossover(up, threshold) alertcondition(Up_Cross, title="DMI+ cross", message="DMI+ Crossing Threshold") Down_Cross = crossover(down, threshold) alertcondition(Down_Cross, title="DMI- cross", message="DMI- Crossing Threshold") # 定义函数参数 Args: high (pd.Series): 'high' 数据序列 low (pd.Series): 'low' 数据序列 close (pd.Series): 'close' 数据序列 length (int): 周期长度，默认为14 lensig (int): 信号长度，类似于 TradingView 的默认 ADX 长度，默认为 length scalar (float): 放大倍数，默认为100 mamode (str): 参考 ```help(ta.ma)```py，默认为 'rma' drift (int): 差异周期，默认为1 offset (int): 结果偏移周期数，默认为0 Kwargs: fillna (value, optional): pd.DataFrame.fillna(value) 的填充值 fill_method (value, optional): 填充方法类型 Returns: pd.DataFrame: 包含 adx、dmp、dmn 列的数据框
.\pandas-ta\pandas_ta\trend\amat.py
# -*- coding: utf-8 -*- # 从 pandas 库中导入 DataFrame 类 from pandas import DataFrame # 从当前目录下的 long_run 模块中导入 long_run 函数 from .long_run import long_run # 从当前目录下的 short_run 模块中导入 short_run 函数 from .short_run import short_run # 从 pandas_ta 包中的 overlap 模块中导入 ma 函数 from pandas_ta.overlap import ma # 从 pandas_ta 包中的 utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义函数 amat，用于计算 Archer Moving Averages Trends (AMAT) 指标 def amat(close=None, fast=None, slow=None, lookback=None, mamode=None, offset=None, **kwargs): """Indicator: Archer Moving Averages Trends (AMAT)""" # 验证参数的有效性，如果未提供则使用默认值 fast = int(fast) if fast and fast > 0 else 8 slow = int(slow) if slow and slow > 0 else 21 lookback = int(lookback) if lookback and lookback > 0 else 2 # 将 mamode 转换为小写字符串，如果未提供则使用默认值 "ema" mamode = mamode.lower() if isinstance(mamode, str) else "ema" # 验证 close 参数，确保长度足够用于计算指标 close = verify_series(close, max(fast, slow, lookback)) # 获取偏移量 offset = get_offset(offset) # 如果 kwargs 中包含 "length" 键，则移除它 if "length" in kwargs: kwargs.pop("length") # 如果未提供 close 参数，则返回空值 if close is None: return # 计算快速移动平均线和慢速移动平均线 fast_ma = ma(mamode, close, length=fast, **kwargs) slow_ma = ma(mamode, close, length=slow, **kwargs) # 计算长期和短期运行趋势 mas_long = long_run(fast_ma, slow_ma, length=lookback) mas_short = short_run(fast_ma, slow_ma, length=lookback) # 对结果进行偏移处理 if offset != 0: mas_long = mas_long.shift(offset) mas_short = mas_short.shift(offset) # 如果 kwargs 中包含 "fillna" 键，则使用指定的值填充缺失值 if "fillna" in kwargs: mas_long.fillna(kwargs["fillna"], inplace=True) mas_short.fillna(kwargs["fillna"], inplace=True) # 如果 kwargs 中包含 "fill_method" 键，则使用指定的填充方法填充缺失值 if "fill_method" in kwargs: mas_long.fillna(method=kwargs["fill_method"], inplace=True) mas_short.fillna(method=kwargs["fill_method"], inplace=True) # 准备要返回的 DataFrame amatdf = DataFrame({ f"AMAT{mamode[0]}_LR_{fast}_{slow}_{lookback}": mas_long, f"AMAT{mamode[0]}_SR_{fast}_{slow}_{lookback}": mas_short }) # 设置 DataFrame 的名称和类别 amatdf.name = f"AMAT{mamode[0]}_{fast}_{slow}_{lookback}" amatdf.category = "trend" # 返回结果 DataFrame return amatdf
.\pandas-ta\pandas_ta\trend\aroon.py
# -*- coding: utf-8 -*- # 从 pandas 库中导入 DataFrame 类 from pandas import DataFrame # 从 pandas_ta 库中导入 Imports 模块 from pandas_ta import Imports # 从 pandas_ta.utils 模块中导入 get_offset, verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 从 pandas_ta.utils 模块中导入 recent_maximum_index, recent_minimum_index 函数 from pandas_ta.utils import recent_maximum_index, recent_minimum_index # 定义函数 aroon，计算 Aroon 和 Aroon Oscillator 指标 def aroon(high, low, length=None, scalar=None, talib=None, offset=None, **kwargs): """Indicator: Aroon & Aroon Oscillator""" # 验证参数 length = length if length and length > 0 else 14 scalar = float(scalar) if scalar else 100 high = verify_series(high, length) low = verify_series(low, length) offset = get_offset(offset) mode_tal = bool(talib) if isinstance(talib, bool) else True if high is None or low is None: return # 计算结果 if Imports["talib"] and mode_tal: from talib import AROON, AROONOSC aroon_down, aroon_up = AROON(high, low, length) aroon_osc = AROONOSC(high, low, length) else: periods_from_hh = high.rolling(length + 1).apply(recent_maximum_index, raw=True) periods_from_ll = low.rolling(length + 1).apply(recent_minimum_index, raw=True) aroon_up = aroon_down = scalar aroon_up *= 1 - (periods_from_hh / length) aroon_down *= 1 - (periods_from_ll / length) aroon_osc = aroon_up - aroon_down # 处理填充 if "fillna" in kwargs: aroon_up.fillna(kwargs["fillna"], inplace=True) aroon_down.fillna(kwargs["fillna"], inplace=True) aroon_osc.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: aroon_up.fillna(method=kwargs["fill_method"], inplace=True) aroon_down.fillna(method=kwargs["fill_method"], inplace=True) aroon_osc.fillna(method=kwargs["fill_method"], inplace=True) # 偏移 if offset != 0: aroon_up = aroon_up.shift(offset) aroon_down = aroon_down.shift(offset) aroon_osc = aroon_osc.shift(offset) # 命名和分类 aroon_up.name = f"AROONU_{length}" aroon_down.name = f"AROOND_{length}" aroon_osc.name = f"AROONOSC_{length}" aroon_down.category = aroon_up.category = aroon_osc.category = "trend" # 准备要返回的 DataFrame data = { aroon_down.name: aroon_down, aroon_up.name: aroon_up, aroon_osc.name: aroon_osc, } aroondf = DataFrame(data) aroondf.name = f"AROON_{length}" aroondf.category = aroon_down.category return aroondf # 设置函数 aroon 的文档字符串 aroon.__doc__ = \ """Aroon & Aroon Oscillator (AROON) Aroon attempts to identify if a security is trending and how strong. Sources: https://www.tradingview.com/wiki/Aroon https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/aroon-ar/ Calculation: Default Inputs: length=1, scalar=100 recent_maximum_index(x): return int(np.argmax(x[::-1])) recent_minimum_index(x): return int(np.argmin(x[::-1])) periods_from_hh = high.rolling(length + 1).apply(recent_maximum_index, raw=True) """ # 计算 Aroon 指标中的上升线，使用公式：scalar * (1 - (periods_from_hh / length)) AROON_UP = scalar * (1 - (periods_from_hh / length)) # 计算 Aroon 指标中的下降线，使用公式：scalar * (1 - (periods_from_ll / length)) periods_from_ll = low.rolling(length + 1).apply(recent_minimum_index, raw=True) AROON_DN = scalar * (1 - (periods_from_ll / length)) # 计算 Aroon 指标的震荡值，使用公式：AROON_UP - AROON_DN AROON_OSC = AROON_UP - AROON_DN # 定义函数参数 Args: close (pd.Series): 包含'close'价格数据的Series length (int): 计算指标的周期，默认为14 scalar (float): 放大倍数，默认为100 talib (bool): 如果安装了TA Lib并且talib为True，则返回TA Lib版本，默认为True offset (int): 结果的偏移周期数，默认为0 # 定义函数关键字参数 Kwargs: fillna (value, optional): pd.DataFrame.fillna(value)的填充值 fill_method (value, optional): 填充方法的类型 # 返回值 Returns: pd.DataFrame: 包含aroon_up、aroon_down、aroon_osc列的DataFrame
PandasTA 源码解析（十二）（3）https://developer.aliyun.com/article/1506223
