.\pandas-ta\pandas_ta\trend\decreasing.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 模块中导入所需函数和类 from pandas_ta.utils import get_drift, get_offset, is_percent, verify_series # 定义一个名为 decreasing 的函数，用于计算序列是否递减 def decreasing(close, length=None, strict=None, asint=None, percent=None, drift=None, offset=None, **kwargs): """Indicator: Decreasing""" # 验证参数 # 如果 length 参数存在且大于 0，则将其转换为整数，否则设为 1 length = int(length) if length and length > 0 else 1 # 如果 strict 参数是布尔类型，则保持不变，否则设为 False strict = strict if isinstance(strict, bool) else False # 如果 asint 参数是布尔类型，则保持不变，否则设为 True asint = asint if isinstance(asint, bool) else True # 对 close 序列进行验证，并设定长度为 length close = verify_series(close, length) # 获取 drift 和 offset 参数的值 drift = get_drift(drift) offset = get_offset(offset) # 如果 percent 参数是百分比，则将其转换为浮点数，否则设为 False percent = float(percent) if is_percent(percent) else False # 如果 close 为 None，则返回 None if close is None: return # 计算结果 # 如果 percent 存在，则对 close 序列进行缩放 close_ = (1 - 0.01 * percent) * close if percent else close # 如果 strict 为 True，则进行严格递减的计算 if strict: # 使用循环检查连续递减的情况 decreasing = close < close_.shift(drift) for x in range(3, length + 1): decreasing = decreasing & (close.shift(x - (drift + 1)) < close_.shift(x - drift)) # 填充缺失值为 0，并将结果转换为布尔类型 decreasing.fillna(0, inplace=True) decreasing = decreasing.astype(bool) else: # 否则，使用简单的递减计算 decreasing = close_.diff(length) < 0 # 如果 asint 为 True，则将结果转换为整数 if asint: decreasing = decreasing.astype(int) # 对结果进行偏移 if offset != 0: decreasing = decreasing.shift(offset) # 处理填充值 if "fillna" in kwargs: decreasing.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: decreasing.fillna(method=kwargs["fill_method"], inplace=True) # 设定结果的名称和类别 _percent = f"_{0.01 * percent}" if percent else '' _props = f"{'S' if strict else ''}DEC{'p' if percent else ''}" decreasing.name = f"{_props}_{length}{_percent}" decreasing.category = "trend" return decreasing # 为 decreasing 函数添加文档字符串 decreasing.__doc__ = \ """Decreasing Returns True if the series is decreasing over a period, False otherwise. If the kwarg 'strict' is True, it returns True if it is continuously decreasing over the period. When using the kwarg 'asint', then it returns 1 for True or 0 for False. Calculation: if strict: decreasing = all(i > j for i, j in zip(close[-length:], close[1:])) else: decreasing = close.diff(length) < 0 if asint: decreasing = decreasing.astype(int) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 1 strict (bool): If True, checks if the series is continuously decreasing over the period. Default: False percent (float): Percent as an integer. Default: None asint (bool): Returns as binary. Default: True 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. """
.\pandas-ta\pandas_ta\trend\dpo.py
# -*- coding: utf-8 -*- # 从 pandas_ta.overlap 模块导入 sma 函数 from pandas_ta.overlap import sma # 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series def dpo(close, length=None, centered=True, offset=None, **kwargs): """Indicator: Detrend Price Oscillator (DPO)""" # 验证参数 # 将长度参数转换为整数，如果长度参数存在且大于0，则使用，否则默认为20 length = int(length) if length and length > 0 else 20 # 验证 close 是否为有效的时间序列，长度为指定的长度 close = verify_series(close, length) # 获取偏移量 offset = get_offset(offset) # 如果 kwargs 中的 "lookahead" 为 False，则将 centered 设置为 False if not kwargs.get("lookahead", True): centered = False # 如果 close 为空，则返回 None if close is None: return # 计算结果 # 计算 t，即 int(0.5 * length) + 1 t = int(0.5 * length) + 1 # 计算 close 的简单移动平均 ma = sma(close, length) # 计算 DPO，close 减去 ma 后向前位移 t 个周期 dpo = close - ma.shift(t) # 如果 centered 为 True，则再将 DPO 向后位移 t 个周期 if centered: dpo = (close.shift(t) - ma).shift(-t) # 偏移 if offset != 0: dpo = dpo.shift(offset) # 处理填充 # 如果 kwargs 中有 "fillna"，则使用该值填充 NaN if "fillna" in kwargs: dpo.fillna(kwargs["fillna"], inplace=True) # 如果 kwargs 中有 "fill_method"，则使用指定的填充方法 if "fill_method" in kwargs: dpo.fillna(method=kwargs["fill_method"], inplace=True) # 命名和分类 dpo.name = f"DPO_{length}" dpo.category = "trend" return dpo # 更新文档字符串 dpo.__doc__ = \ """Detrend Price Oscillator (DPO) Is an indicator designed to remove trend from price and make it easier to identify cycles. Sources: https://www.tradingview.com/scripts/detrendedpriceoscillator/ https://www.fidelity.com/learning-center/trading-investing/technical-analysis/technical-indicator-guide/dpo http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:detrended_price_osci Calculation: Default Inputs: length=20, centered=True SMA = Simple Moving Average t = int(0.5 * length) + 1 DPO = close.shift(t) - SMA(close, length) if centered: DPO = DPO.shift(-t) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 1 centered (bool): Shift the dpo back by int(0.5 * length) + 1. 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. """
.\pandas-ta\pandas_ta\trend\increasing.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 中导入所需的函数和模块 from pandas_ta.utils import get_drift, get_offset, is_percent, verify_series # 定义名为 increasing 的函数，用于计算序列是否递增 def increasing(close, length=None, strict=None, asint=None, percent=None, drift=None, offset=None, **kwargs): """Indicator: Increasing""" # 验证参数的有效性 length = int(length) if length and length > 0 else 1 # 将长度转换为整数，如果未提供或小于等于0，则设为1 strict = strict if isinstance(strict, bool) else False # 如果 strict 不是布尔值，则设为 False asint = asint if isinstance(asint, bool) else True # 如果 asint 不是布尔值，则设为 True close = verify_series(close, length) # 验证并处理输入的序列数据 drift = get_drift(drift) # 获取漂移值 offset = get_offset(offset) # 获取偏移值 percent = float(percent) if is_percent(percent) else False # 将百分比转换为浮点数，如果不是百分比，则设为 False if close is None: return # 如果序列为空，则返回空值 # 计算结果 close_ = (1 + 0.01 * percent) * close if percent else close # 如果有百分比参数，则对序列进行调整 if strict: # 返回值是否为 float64？必须转换为布尔值 increasing = close > close_.shift(drift) # 检查当前值是否大于移动后的值 for x in range(3, length + 1): increasing = increasing & (close.shift(x - (drift + 1)) > close_.shift(x - drift)) # 检查连续多个值是否递增 increasing.fillna(0, inplace=True) # 填充缺失值为0 increasing = increasing.astype(bool) # 将结果转换为布尔值 else: increasing = close_.diff(length) > 0 # 检查序列是否在给定周期内递增 if asint: increasing = increasing.astype(int) # 将结果转换为整数类型 # 偏移结果 if offset != 0: increasing = increasing.shift(offset) # 对结果进行偏移 # 处理填充值 if "fillna" in kwargs: increasing.fillna(kwargs["fillna"], inplace=True) # 使用指定值填充缺失值 if "fill_method" in kwargs: increasing.fillna(method=kwargs["fill_method"], inplace=True) # 使用指定方法填充缺失值 # 命名并分类结果 _percent = f"_{0.01 * percent}" if percent else '' # 根据是否存在百分比参数构建后缀 _props = f"{'S' if strict else ''}INC{'p' if percent else ''}" # 根据参数构建特性标识 increasing.name = f"{_props}_{length}{_percent}" # 构建结果的名称 increasing.category = "trend" # 将结果分类为趋势 return increasing # 返回计算结果 increasing.__doc__ = \ """Increasing Returns True if the series is increasing over a period, False otherwise. If the kwarg 'strict' is True, it returns True if it is continuously increasing over the period. When using the kwarg 'asint', then it returns 1 for True or 0 for False. Calculation: if strict: increasing = all(i < j for i, j in zip(close[-length:], close[1:])) else: increasing = close.diff(length) > 0 if asint: increasing = increasing.astype(int) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 1 strict (bool): If True, checks if the series is continuously increasing over the period. Default: False percent (float): Percent as an integer. Default: None asint (bool): Returns as binary. Default: True 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. """
.\pandas-ta\pandas_ta\trend\long_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 函数 def long_run(fast, slow, length=None, offset=None, **kwargs): """Indicator: Long Run""" # Validate Arguments # 将 length 转换为整数，如果 length 存在且大于 0；否则默认为 2 length = int(length) if length and length > 0 else 2 # 验证 fast 和 slow 是否为有效的序列，并将其长度限制为 length fast = verify_series(fast, length) slow = verify_series(slow, length) # 获取偏移量 offset = get_offset(offset) # 如果 fast 或 slow 为空，则返回空值 if fast is None or slow is None: return # Calculate Result # 计算可能的底部或底部的条件，即 fast 增长而 slow 减小 pb = increasing(fast, length) & decreasing(slow, length) # 计算 fast 和 slow 同时增长的条件 bi = increasing(fast, length) & increasing(slow, length) # 计算长期趋势的条件，可能的底部或底部，以及 fast 和 slow 同时增长的情况 long_run = pb | bi # Offset # 如果 offset 不为 0，则对长期趋势进行偏移 if offset != 0: long_run = long_run.shift(offset) # Handle fills # 处理填充值 if "fillna" in kwargs: long_run.fillna(kwargs["fillna"], inplace=True) # 使用指定的填充方法填充缺失值 if "fill_method" in kwargs: long_run.fillna(method=kwargs["fill_method"], inplace=True) # Name and Categorize it # 设置长期趋势指标的名称 long_run.name = f"LR_{length}" # 设置长期趋势指标的类别为 "trend" long_run.category = "trend" return long_run
PandasTA 源码解析（十三）（2）https://developer.aliyun.com/article/1506228
