PandasTA 源码解析（十三）（1）https://developer.aliyun.com/article/1506227
.\pandas-ta\pandas_ta\trend\psar.py
# -*- coding: utf-8 -*- # 导入 numpy 库中的 nan 作为 npNaN from numpy import nan as npNaN # 导入 DataFrame 和 Series 类 from pandas import DataFrame, Series # 导入 get_offset, verify_series, zero 函数 from pandas_ta.utils import get_offset, verify_series, zero # 定义 PSAR 指标函数，参数包括 high, low, close, af0, af, max_af, offset def psar(high, low, close=None, af0=None, af=None, max_af=None, offset=None, **kwargs): """Indicator: Parabolic Stop and Reverse (PSAR)""" # 验证参数 high = verify_series(high) low = verify_series(low) af = float(af) if af and af > 0 else 0.02 af0 = float(af0) if af0 and af0 > 0 else af max_af = float(max_af) if max_af and max_af > 0 else 0.2 offset = get_offset(offset) # 定义 _falling 函数，用于返回最后一个 -DM 值 def _falling(high, low, drift:int=1): """Returns the last -DM value""" # 不要与 ta.falling() 混淆 up = high - high.shift(drift) dn = low.shift(drift) - low _dmn = (((dn > up) & (dn > 0)) * dn).apply(zero).iloc[-1] return _dmn > 0 # 如果第一个 NaN -DM 为正，则为下降趋势 falling = _falling(high.iloc[:2], low.iloc[:2]) if falling: sar = high.iloc[0] ep = low.iloc[0] else: sar = low.iloc[0] ep = high.iloc[0] # 如果存在 close 参数，则使用 close 的值 if close is not None: close = verify_series(close) sar = close.iloc[0] # 初始化 long, short, reversal, _af long = Series(npNaN, index=high.index) short = long.copy() reversal = Series(0, index=high.index) _af = long.copy() _af.iloc[0:2] = af0 # 计算结果 m = high.shape[0] for row in range(1, m): high_ = high.iloc[row] low_ = low.iloc[row] if falling: _sar = sar + af * (ep - sar) reverse = high_ > _sar if low_ < ep: ep = low_ af = min(af + af0, max_af) _sar = max(high.iloc[row - 1], high.iloc[row - 2], _sar) else: _sar = sar + af * (ep - sar) reverse = low_ < _sar if high_ > ep: ep = high_ af = min(af + af0, max_af) _sar = min(low.iloc[row - 1], low.iloc[row - 2], _sar) if reverse: _sar = ep af = af0 falling = not falling ep = low_ if falling else high_ sar = _sar if falling: short.iloc[row] = sar else: long.iloc[row] = sar _af.iloc[row] = af reversal.iloc[row] = int(reverse) # 偏移 if offset != 0: _af = _af.shift(offset) long = long.shift(offset) short = short.shift(offset) reversal = reversal.shift(offset) # 处理填充 if "fillna" in kwargs: _af.fillna(kwargs["fillna"], inplace=True) long.fillna(kwargs["fillna"], inplace=True) short.fillna(kwargs["fillna"], inplace=True) reversal.fillna(kwargs["fillna"], inplace=True) # 检查参数中是否包含'fill_method'关键字 if "fill_method" in kwargs: # 使用指定的填充方法填充数据，直接修改原数据，无需返回值 _af.fillna(method=kwargs["fill_method"], inplace=True) long.fillna(method=kwargs["fill_method"], inplace=True) short.fillna(method=kwargs["fill_method"], inplace=True) reversal.fillna(method=kwargs["fill_method"], inplace=True) # 准备要返回的 DataFrame # 根据参数构建用于命名列的字符串 _params = f"_{af0}_{max_af}" # 构建包含 PSAR 数据的字典 data = { f"PSARl{_params}": long, f"PSARs{_params}": short, f"PSARaf{_params}": _af, f"PSARr{_params}": reversal, } # 从字典创建 DataFrame psardf = DataFrame(data) # 设置 DataFrame 的名称 psardf.name = f"PSAR{_params}" # 设置 DataFrame 及其列的类别属性为 'trend' psardf.category = long.category = short.category = "trend" # 返回 PSAR 数据的 DataFrame return psardf # 设置 psar 函数的文档字符串，用于描述 Parabolic Stop and Reverse (PSAR) 指标的作用、计算方式和参数说明 psar.__doc__ = \ """Parabolic Stop and Reverse (psar) Parabolic Stop and Reverse (PSAR) was developed by J. Wells Wilder, that is used to determine trend direction and it's potential reversals in price. PSAR uses a trailing stop and reverse method called "SAR," or stop and reverse, to identify possible entries and exits. It is also known as SAR. PSAR indicator typically appears on a chart as a series of dots, either above or below an asset's price, depending on the direction the price is moving. A dot is placed below the price when it is trending upward, and above the price when it is trending downward. Sources: https://www.tradingview.com/pine-script-reference/#fun_sar https://www.sierrachart.com/index.php?page=doc/StudiesReference.php&ID=66&Name=Parabolic Calculation: Default Inputs: af0=0.02, af=0.02, max_af=0.2 See Source links Args: high (pd.Series): Series of 'high's low (pd.Series): Series of 'low's close (pd.Series, optional): Series of 'close's. Optional af0 (float): Initial Acceleration Factor. Default: 0.02 af (float): Acceleration Factor. Default: 0.02 max_af (float): Maximum Acceleration Factor. Default: 0.2 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.DataFrame: long, short, af, and reversal columns. """
.\pandas-ta\pandas_ta\trend\qstick.py
# -*- coding: utf-8 -*- # 从 pandas_ta.overlap 模块导入 dema, ema, hma, rma, sma 函数 from pandas_ta.overlap import dema, ema, hma, rma, sma # 从 pandas_ta.utils 模块导入 get_offset, non_zero_range, verify_series 函数 from pandas_ta.utils import get_offset, non_zero_range, verify_series # 定义 Q Stick 指标函数 def qstick(open_, close, length=None, offset=None, **kwargs): """Indicator: Q Stick""" # 验证参数 length = int(length) if length and length > 0 else 10 ma = kwargs.pop("ma", "sma") open_ = verify_series(open_, length) close = verify_series(close, length) offset = get_offset(offset) if open_ is None or close is None: return # 计算结果 diff = non_zero_range(close, open_) if ma == "dema": qstick = dema(diff, length=length, **kwargs) elif ma == "ema": qstick = ema(diff, length=length, **kwargs) elif ma == "hma": qstick = hma(diff, length=length) elif ma == "rma": qstick = rma(diff, length=length) else: # "sma" qstick = sma(diff, length=length) # 偏移 if offset != 0: qstick = qstick.shift(offset) # 处理填充 if "fillna" in kwargs: qstick.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: qstick.fillna(method=kwargs["fill_method"], inplace=True) # 命名和分类 qstick.name = f"QS_{length}" qstick.category = "trend" return qstick # 设置 Q Stick 函数的文档字符串 qstick.__doc__ = \ """Q Stick The Q Stick indicator, developed by Tushar Chande, attempts to quantify and identify trends in candlestick charts. Sources: https://library.tradingtechnologies.com/trade/chrt-ti-qstick.html Calculation: Default Inputs: length=10 xMA is one of: sma (default), dema, ema, hma, rma qstick = xMA(close - open, length) Args: open (pd.Series): Series of 'open's close (pd.Series): Series of 'close's length (int): It's period. Default: 1 ma (str): The type of moving average to use. Default: None, which is 'sma' 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. """
.\pandas-ta\pandas_ta\trend\short_run.py
# -*- coding: utf-8 -*- # 导入必要的模块和函数 from .decreasing import decreasing # 从当前目录下的 decreasing 模块中导入 decreasing 函数 from .increasing import increasing # 从当前目录下的 increasing 模块中导入 increasing 函数 from pandas_ta.utils import get_offset, verify_series # 从 pandas_ta 包中的 utils 模块导入 get_offset 和 verify_series 函数 # 定义一个名为 short_run 的函数，用于计算短期趋势 def short_run(fast, slow, length=None, offset=None, **kwargs): """Indicator: Short Run""" # 验证参数 length = int(length) if length and length > 0 else 2 # 如果 length 存在且大于0，则将其转换为整数类型，否则设为默认值2 fast = verify_series(fast, length) # 验证 fast 参数，确保其为有效的序列，并可能调整长度 slow = verify_series(slow, length) # 验证 slow 参数，确保其为有效的序列，并可能调整长度 offset = get_offset(offset) # 获取偏移量 if fast is None or slow is None: return # 如果 fast 或 slow 为空，则返回空值 # 计算结果 pt = decreasing(fast, length) & increasing(slow, length) # 潜在顶部或顶部的条件 bd = decreasing(fast, length) & decreasing(slow, length) # 快速和慢速都在下降的条件 short_run = pt | bd # 判断是否出现短期趋势 # 偏移结果 if offset != 0: short_run = short_run.shift(offset) # 处理填充值 if "fillna" in kwargs: short_run.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: short_run.fillna(method=kwargs["fill_method"], inplace=True) # 命名和分类 short_run.name = f"SR_{length}" # 设置结果的名称 short_run.category = "trend" # 设置结果的分类为趋势 return short_run # 返回计算结果
.\pandas-ta\pandas_ta\trend\tsignals.py
# -*- coding: utf-8 -*- # 从 pandas 库导入 DataFrame 类 from pandas import DataFrame # 从 pandas_ta.utils 模块导入 get_drift, get_offset, verify_series 函数 from pandas_ta.utils import get_drift, get_offset, verify_series # 定义 tsignals 函数，用于计算趋势信号 def tsignals(trend, asbool=None, trend_reset=0, trade_offset=None, drift=None, offset=None, **kwargs): """Indicator: Trend Signals""" # 验证参数 # 将 trend 参数转换为 pandas Series 对象 trend = verify_series(trend) # 将 asbool 参数转换为布尔值，默认为 False asbool = bool(asbool) if isinstance(asbool, bool) else False # 将 trend_reset 参数转换为整数，默认为 0 trend_reset = int(trend_reset) if trend_reset and isinstance(trend_reset, int) else 0 # 如果 trade_offset 不为 0，则将其转换为整数，默认为 0 if trade_offset != 0: trade_offset = int(trade_offset) if trade_offset and isinstance(trade_offset, int) else 0 # 获取 drift 和 offset 参数的值 drift = get_drift(drift) offset = get_offset(offset) # 计算结果 # 将趋势值转换为整数类型 trends = trend.astype(int) # 计算交易信号 trades = trends.diff(drift).shift(trade_offset).fillna(0).astype(int) # 计算进入交易的信号 entries = (trades > 0).astype(int) # 计算退出交易的信号 exits = (trades < 0).abs().astype(int) # 如果 asbool 为 True，则将结果转换为布尔值 if asbool: trends = trends.astype(bool) entries = entries.astype(bool) exits = exits.astype(bool) # 构建结果数据 data = { f"TS_Trends": trends, f"TS_Trades": trades, f"TS_Entries": entries, f"TS_Exits": exits, } # 创建 DataFrame 对象 df = DataFrame(data, index=trends.index) # 处理偏移 if offset != 0: df = df.shift(offset) # 处理填充 if "fillna" in kwargs: df.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: df.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和类别 df.name = f"TS" df.category = "trend" return df # 设置 tsignals 函数的文档字符串 tsignals.__doc__ = \ """Trend Signals Given a Trend, Trend Signals returns the Trend, Trades, Entries and Exits as boolean integers. When 'asbool=True', it returns Trends, Entries and Exits as boolean values which is helpful when combined with the vectorbt backtesting package. A Trend can be a simple as: 'close' > 'moving average' or something more complex whose values are boolean or integers (0 or 1). Examples: ta.tsignals(close > ta.sma(close, 50), asbool=False) ta.tsignals(ta.ema(close, 8) > ta.ema(close, 21), asbool=True) Source: Kevin Johnson Calculation: Default Inputs: asbool=False, trend_reset=0, trade_offset=0, drift=1 trades = trends.diff().shift(trade_offset).fillna(0).astype(int) entries = (trades > 0).astype(int) exits = (trades < 0).abs().astype(int) Args: trend (pd.Series): Series of 'trend's. The trend can be either a boolean or integer series of '0's and '1's asbool (bool): If True, it converts the Trends, Entries and Exits columns to booleans. When boolean, it is also useful for backtesting with vectorbt's Portfolio.from_signal(close, entries, exits) Default: False trend_reset (value): Value used to identify if a trend has ended. Default: 0 trade_offset (value): Value used shift the trade entries/exits Use 1 for backtesting and 0 for live. Default: 0 drift (int): The difference period. Default: 1 """ offset (int): How many periods to offset the result. Default: 0 # 函数参数说明部分，描述了函数的参数和返回值 Kwargs: # fillna 参数，用于填充缺失值，使用 pd.DataFrame.fillna 函数 fillna (value, optional): pd.DataFrame.fillna(value) # fill_method 参数，填充方法的类型说明 fill_method (value, optional): Type of fill method # 返回值说明部分，描述了函数返回的 DataFrame 的列 Returns: # 返回一个 pandas DataFrame，包含以下列： pd.DataFrame with columns: # Trends 列，表示趋势，有趋势为 1，无趋势为 0 Trends (trend: 1, no trend: 0), # Trades 列，表示交易，进入为 1，退出为 -1，其他情况为 0 Trades (Enter: 1, Exit: -1, Otherwise: 0), # Entries 列，表示进入，进入为 1，无操作为 0 Entries (entry: 1, nothing: 0), # Exits 列，表示退出，退出为 1，无操作为 0 Exits (exit: 1, nothing: 0)
PandasTA 源码解析（十三）（3）https://developer.aliyun.com/article/1506230
