.\pandas-ta\pandas_ta\momentum\cg.py

# -*- coding: utf-8 -*-

# 从 pandas_ta.utils 中导入 get_offset, verify_series, weights 函数
from pandas_ta.utils import get_offset, verify_series, weights

# 定义 Center of Gravity (CG) 指标函数
def cg(close, length=None, offset=None, **kwargs):
    """Indicator: Center of Gravity (CG)"""
    # 验证参数
    # 将长度转换为整数，如果长度存在且大于0，则取其值，否则默认为10
    length = int(length) if length and length > 0 else 10
    # 验证 close 数据，并将长度设为 length
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)

    # 如果 close 为空，则返回空
    if close is None: return

    # 计算结果
    # 计算系数列表，范围为 [length-0, length-1, ..., length-length]
    coefficients = [length - i for i in range(0, length)]
    # 计算分子，使用 close 的滚动窗口应用权重函数，raw=True 表示原始数据传递给权重函数
    numerator = -close.rolling(length).apply(weights(coefficients), raw=True)
    # 计算 Center of Gravity (CG)，即分子除以 close 滚动窗口的和
    cg = numerator / close.rolling(length).sum()

    # 偏移结果
    if offset != 0:
        # 对 CG 应用偏移
        cg = cg.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        # 使用指定值填充缺失值
        cg.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        # 使用指定方法填充缺失值
        cg.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置指标名称和分类
    cg.name = f"CG_{length}"
    cg.category = "momentum"

    return cg

# 设置 Center of Gravity (CG) 函数的文档字符串
cg.__doc__ = \
"""Center of Gravity (CG)

The Center of Gravity Indicator by John Ehlers attempts to identify turning
points while exhibiting zero lag and smoothing.

Sources:
    http://www.mesasoftware.com/papers/TheCGOscillator.pdf

Calculation:
    Default Inputs:
        length=10

Args:
    close (pd.Series): Series of 'close's
    length (int): The length of the period. Default: 10
    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\momentum\cmo.py

# -*- coding: utf-8 -*-

# 导入必要的模块和函数
from pandas_ta import Imports
from pandas_ta.overlap import rma
from pandas_ta.utils import get_drift, get_offset, verify_series

# 定义 Chande Momentum Oscillator (CMO) 指标函数
def cmo(close, length=None, scalar=None, talib=None, drift=None, offset=None, **kwargs):
    """Indicator: Chande Momentum Oscillator (CMO)"""
    # 验证参数
    length = int(length) if length and length > 0 else 14
    scalar = float(scalar) if scalar else 100
    close = verify_series(close, length)
    drift = get_drift(drift)
    offset = get_offset(offset)
    mode_tal = bool(talib) if isinstance(talib, bool) else True

    # 如果 close 为空，返回空值
    if close is None: return

    # 计算结果
    if Imports["talib"] and mode_tal:
        from talib import CMO
        # 使用 TA-Lib 计算 CMO
        cmo = CMO(close, length)
    else:
        # 计算动量
        mom = close.diff(drift)
        positive = mom.copy().clip(lower=0)
        negative = mom.copy().clip(upper=0).abs()

        if mode_tal:
            # 使用 RMA 函数计算动量的指数移动平均值
            pos_ = rma(positive, length)
            neg_ = rma(negative, length)
        else:
            # 使用滚动窗口计算动量的总和
            pos_ = positive.rolling(length).sum()
            neg_ = negative.rolling(length).sum()

        # 计算 CMO 指标值
        cmo = scalar * (pos_ - neg_) / (pos_ + neg_)

    # 对结果进行偏移处理
    if offset != 0:
        cmo = cmo.shift(offset)

    # 处理填充值
    if "fillna" in kwargs:
        cmo.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        cmo.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置指标名称和分类
    cmo.name = f"CMO_{length}"
    cmo.category = "momentum"

    return cmo

# 设置 CMO 函数的文档字符串
cmo.__doc__ = \
"""Chande Momentum Oscillator (CMO)

Attempts to capture the momentum of an asset with overbought at 50 and
oversold at -50.

Sources:
    https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/chande-momentum-oscillator-cmo/
    https://www.tradingview.com/script/hdrf0fXV-Variable-Index-Dynamic-Average-VIDYA/

Calculation:
    Default Inputs:
        drift=1, scalar=100

    # Same Calculation as RSI except for this step
    CMO = scalar * (PSUM - NSUM) / (PSUM + NSUM)

Args:
    close (pd.Series): Series of 'close's
    scalar (float): How much to magnify. Default: 100
    talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
        version. If TA Lib is not installed but talib is True, it runs the Python
        version TA Lib. Default: True
    drift (int): The short period. Default: 1
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    talib (bool): If True, uses TA-Libs implementation. Otherwise uses EMA version. Default: True
    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\momentum\coppock.py

# -*- coding: utf-8 -*-
# 从 roc 模块导入 roc 函数
from .roc import roc
# 从 pandas_ta.overlap 模块导入 wma 函数
from pandas_ta.overlap import wma
# 从 pandas_ta.utils 模块导入 get_offset, verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义 Coppock Curve (COPC) 指标函数
def coppock(close, length=None, fast=None, slow=None, offset=None, **kwargs):
    """Indicator: Coppock Curve (COPC)"""
    # 验证参数
    length = int(length) if length and length > 0 else 10
    fast = int(fast) if fast and fast > 0 else 11
    slow = int(slow) if slow and slow > 0 else 14
    close = verify_series(close, max(length, fast, slow))
    offset = get_offset(offset)

    if close is None: return

    # 计算结果
    total_roc = roc(close, fast) + roc(close, slow)
    coppock = wma(total_roc, length)

    # 偏移
    if offset != 0:
        coppock = coppock.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        coppock.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        coppock.fillna(method=kwargs["fill_method"], inplace=True)

    # 命名和分类
    coppock.name = f"COPC_{fast}_{slow}_{length}"
    coppock.category = "momentum"

    return coppock

# 设置 Coppock Curve (COPC) 函数的文档字符串
coppock.__doc__ = \
"""Coppock Curve (COPC)

Coppock Curve (originally called the "Trendex Model") is a momentum indicator
is designed for use on a monthly time scale.  Although designed for monthly
use, a daily calculation over the same period can be made, converting the
periods to 294-day and 231-day rate of changes, and a 210-day weighted
moving average.

Sources:
    https://en.wikipedia.org/wiki/Coppock_curve

Calculation:
    Default Inputs:
        length=10, fast=11, slow=14
    SMA = Simple Moving Average
    MAD = Mean Absolute Deviation
    tp = typical_price = hlc3 = (high + low + close) / 3
    mean_tp = SMA(tp, length)
    mad_tp = MAD(tp, length)
    CCI = (tp - mean_tp) / (c * mad_tp)

Args:
    close (pd.Series): Series of 'close's
    length (int): WMA period. Default: 10
    fast (int): Fast ROC period. Default: 11
    slow (int): Slow ROC period. Default: 14
    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\momentum\cti.py

# -*- coding: utf-8 -*-
# 从 pandas 库导入 Series 类
from pandas import Series
# 从 pandas_ta.overlap 模块导入 linreg 函数
from pandas_ta.overlap import linreg
# 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义函数 cti，计算 Correlation Trend Indicator（相关趋势指标）
def cti(close, length=None, offset=None, **kwargs) -> Series:
    """Indicator: Correlation Trend Indicator"""
    # 如果 length 存在且大于 0，则将其转换为整数类型；否则使用默认值 12
    length = int(length) if length and length > 0 else 12
    # 验证并确保 close 是一个 pd.Series 类型的数据，并且长度为 length
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)

    # 如果 close 为 None，则返回空值
    if close is None: return

    # 计算相关趋势指标
    cti = linreg(close, length=length, r=True)

    # 偏移结果
    if offset != 0:
        cti = cti.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        cti.fillna(method=kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        cti.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置指标名称
    cti.name = f"CTI_{length}"
    # 设置指标类别
    cti.category = "momentum"
    # 返回相关趋势指标的 Series
    return cti


# 设置 cti 函数的文档字符串
cti.__doc__ = \
"""Correlation Trend Indicator (CTI)

The Correlation Trend Indicator is an oscillator created by John Ehler in 2020.
It assigns a value depending on how close prices in that range are to following
a positively- or negatively-sloping straight line. Values range from -1 to 1.
This is a wrapper for ta.linreg(close, r=True).

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 12
    offset (int): How many periods to offset the result. Default: 0

Returns:
    pd.Series: Series of the CTI values for the given period.
"""

.\pandas-ta\pandas_ta\momentum\dm.py

# -*- coding: utf-8 -*-

# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta 库中导入 Imports 模块
from pandas_ta import Imports
# 从 pandas_ta.overlap 模块中导入 ma 函数
from pandas_ta.overlap import ma
# 从 pandas_ta.utils 模块中导入 get_offset, verify_series, get_drift, zero 函数
from pandas_ta.utils import get_offset, verify_series, get_drift, zero

# 定义函数 dm，计算 Directional Movement 指标
def dm(high, low, length=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
    """Indicator: DM"""
    # 验证参数
    length = int(length) if length and length > 0 else 14
    mamode = mamode.lower() if mamode and isinstance(mamode, str) else "rma"
    high = verify_series(high)
    low = verify_series(low)
    drift = get_drift(drift)
    offset = get_offset(offset)
    mode_tal = bool(talib) if isinstance(talib, bool) else True

    # 如果 high 或 low 为空，则返回空
    if high is None or low is None:
        return

    # 如果导入了 talib 并且 mode_tal 为 True，则使用 talib 库中的 PLUS_DM 和 MINUS_DM 函数
    if Imports["talib"] and mode_tal:
        from talib import MINUS_DM, PLUS_DM
        pos = PLUS_DM(high, low, length)
        neg = MINUS_DM(high, low, length)
    else:
        # 计算 up 和 dn
        up = high - high.shift(drift)
        dn = low.shift(drift) - low

        pos_ = ((up > dn) & (up > 0)) * up
        neg_ = ((dn > up) & (dn > 0)) * dn

        pos_ = pos_.apply(zero)
        neg_ = neg_.apply(zero)

        # 计算 pos 和 neg
        pos = ma(mamode, pos_, length=length)
        neg = ma(mamode, neg_, length=length)

    # 偏移
    if offset != 0:
        pos = pos.shift(offset)
        neg = neg.shift(offset)

    _params = f"_{length}"
    data = {
        f"DMP{_params}": pos,
        f"DMN{_params}": neg,
    }

    # 创建 DataFrame 对象
    dmdf = DataFrame(data)
    # 设置 DataFrame 的名称和类别
    dmdf.name = f"DM{_params}"
    dmdf.category = "trend"

    return dmdf

# 设置 dm 函数的文档字符串
dm.__doc__ = \
"""Directional Movement (DM)

The Directional Movement was developed by J. Welles Wilder in 1978 attempts to
determine which direction the price of an asset is moving. It compares prior
highs and lows to yield to two series +DM and -DM.

Sources:
    https://www.tradingview.com/pine-script-reference/#fun_dmi
    https://www.sierrachart.com/index.php?page=doc/StudiesReference.php&ID=24&Name=Directional_Movement_Index

Calculation:
    Default Inputs:
        length=14, mamode="rma", drift=1
            up = high - high.shift(drift)
        dn = low.shift(drift) - low

        pos_ = ((up > dn) & (up > 0)) * up
        neg_ = ((dn > up) & (dn > 0)) * dn

        pos_ = pos_.apply(zero)
        neg_ = neg_.apply(zero)

        # Not the same values as TA Lib's -+DM
        pos = ma(mamode, pos_, length=length)
        neg = ma(mamode, neg_, length=length)

Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    mamode (str): See ```help(ta.ma)```py.  Default: 'rma'
    talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
        version. Default: True
    drift (int): The difference period. Default: 1
    offset (int): How many periods to offset the result. Default: 0

Returns:
    pd.DataFrame: DMP (+DM) and DMN (-DM) columns.
"""

.\pandas-ta\pandas_ta\momentum\er.py

# -*- coding: utf-8 -*-

# 从 pandas 库中导入 DataFrame 和 concat 函数
from pandas import DataFrame, concat
# 从 pandas_ta.utils 模块中导入 get_drift, get_offset, verify_series, signals 函数
from pandas_ta.utils import get_drift, get_offset, verify_series, signals

# 定义效率比率（ER）指标函数
def er(close, length=None, drift=None, offset=None, **kwargs):
    """Indicator: Efficiency Ratio (ER)"""
    # 验证参数
    # 将长度参数转换为整数，并确保大于零
    length = int(length) if length and length > 0 else 10
    # 验证 close 参数，并且使用长度参数进行验证
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)
    # 获取漂移
    drift = get_drift(drift)

    # 如果 close 为 None，则返回 None
    if close is None: return

    # 计算结果
    # 计算价格变化的绝对值
    abs_diff = close.diff(length).abs()
    # 计算价格波动的绝对值
    abs_volatility = close.diff(drift).abs()

    # 计算效率比率
    er = abs_diff
    er /= abs_volatility.rolling(window=length).sum()

    # 偏移
    if offset != 0:
        er = er.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        er.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        er.fillna(method=kwargs["fill_method"], inplace=True)

    # 命名和分类
    er.name = f"ER_{length}"
    er.category = "momentum"

    # 处理信号指标
    signal_indicators = kwargs.pop("signal_indicators", False)
    if signal_indicators:
        # 将效率比率和信号指标拼接成一个 DataFrame
        signalsdf = concat(
            [
                DataFrame({er.name: er}),
                signals(
                    indicator=er,
                    xa=kwargs.pop("xa", 80),
                    xb=kwargs.pop("xb", 20),
                    xserie=kwargs.pop("xserie", None),
                    xserie_a=kwargs.pop("xserie_a", None),
                    xserie_b=kwargs.pop("xserie_b", None),
                    cross_values=kwargs.pop("cross_values", False),
                    cross_series=kwargs.pop("cross_series", True),
                    offset=offset,
                ),
            ],
            axis=1,
        )

        return signalsdf
    else:
        return er

# 设置函数文档字符串
er.__doc__ = \
"""Efficiency Ratio (ER)

The Efficiency Ratio was invented by Perry J. Kaufman and presented in his book "New Trading Systems and Methods". It is designed to account for market noise or volatility.

It is calculated by dividing the net change in price movement over N periods by the sum of the absolute net changes over the same N periods.

Sources:
    https://help.tc2000.com/m/69404/l/749623-kaufman-efficiency-ratio

Calculation:
    Default Inputs:
        length=10
    ABS = Absolute Value
    EMA = Exponential Moving Average

    abs_diff = ABS(close.diff(length))
    volatility = ABS(close.diff(1))
    ER = abs_diff / SUM(volatility, length)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's 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\momentum\eri.py

# -*- coding: utf-8 -*-

# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta 库中导入 ema 函数
from pandas_ta.overlap import ema
# 从 pandas_ta 库中导入 get_offset, verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义函数 eri，计算 Elder Ray Index (ERI)
def eri(high, low, close, length=None, offset=None, **kwargs):
    """Indicator: Elder Ray Index (ERI)"""
    # 验证参数
   length = int(length) if length and length > 0 else 13
    high = verify_series(high, length)
    low = verify_series(low, length)
    close = verify_series(close, length)
    offset = get_offset(offset)

    # 如果 high、low、close 有任何一个为 None，则返回空
    if high is None or low is None or close is None: return

    # 计算结果
    ema_ = ema(close, length)
    bull = high - ema_
    bear = low - ema_

    # 偏移结果
    if offset != 0:
        bull = bull.shift(offset)
        bear = bear.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        bull.fillna(kwargs["fillna"], inplace=True)
        bear.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        bull.fillna(method=kwargs["fill_method"], inplace=True)
        bear.fillna(method=kwargs["fill_method"], inplace=True)

    # 命名和分类
    bull.name = f"BULLP_{length}"
    bear.name = f"BEARP_{length}"
    bull.category = bear.category = "momentum"

    # 准备返回的 DataFrame
    data = {bull.name: bull, bear.name: bear}
    df = DataFrame(data)
    df.name = f"ERI_{length}"
    df.category = bull.category

    return df

# 设置函数 eri 的文档字符串
eri.__doc__ = \
"""Elder Ray Index (ERI)

Elder's Bulls Ray Index contains his Bull and Bear Powers. Which are useful ways
to look at the price and see the strength behind the market. Bull Power
measures the capability of buyers in the market, to lift prices above an average
consensus of value.

Bears Power measures the capability of sellers, to drag prices below an average
consensus of value. Using them in tandem with a measure of trend allows you to
identify favourable entry points. We hope you've found this to be a useful
discussion of the Bulls and Bears Power indicators.

Sources:
    https://admiralmarkets.com/education/articles/forex-indicators/bears-and-bulls-power-indicator

Calculation:
    Default Inputs:
        length=13
    EMA = Exponential Moving Average

    BULLPOWER = high - EMA(close, length)
    BEARPOWER = low - EMA(close, length)

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
    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: bull power and bear power columns.
"""

.\pandas-ta\pandas_ta\momentum\fisher.py

# -*- coding: utf-8 -*-
# 导入所需的库和函数
from numpy import log as nplog
from numpy import nan as npNaN
from pandas import DataFrame, Series
from pandas_ta.overlap import hl2
from pandas_ta.utils import get_offset, high_low_range, verify_series

# 定义 Fisher Transform (FISHT) 指标函数
def fisher(high, low, length=None, signal=None, offset=None, **kwargs):
    """Indicator: Fisher Transform (FISHT)"""
    # 验证参数
    length = int(length) if length and length > 0 else 9
    signal = int(signal) if signal and signal > 0 else 1
    _length = max(length, signal)
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    offset = get_offset(offset)

    if high is None or low is None: return

    # 计算结果
    hl2_ = hl2(high, low)
    highest_hl2 = hl2_.rolling(length).max()
    lowest_hl2 = hl2_.rolling(length).min()

    hlr = high_low_range(highest_hl2, lowest_hl2)
    hlr[hlr < 0.001] = 0.001

    position = ((hl2_ - lowest_hl2) / hlr) - 0.5

    v = 0
    m = high.size
    result = [npNaN for _ in range(0, length - 1)] + [0]
    for i in range(length, m):
        v = 0.66 * position.iloc[i] + 0.67 * v
        if v < -0.99: v = -0.999
        if v > 0.99: v = 0.999
        result.append(0.5 * (nplog((1 + v) / (1 - v)) + result[i - 1]))
    fisher = Series(result, index=high.index)
    signalma = fisher.shift(signal)

    # 调整偏移量
    if offset != 0:
        fisher = fisher.shift(offset)
        signalma = signalma.shift(offset)

    # 处理填充值
    if "fillna" in kwargs:
        fisher.fillna(kwargs["fillna"], inplace=True)
        signalma.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        fisher.fillna(method=kwargs["fill_method"], inplace=True)
        signalma.fillna(method=kwargs["fill_method"], inplace=True)

    # 命名和分类
    _props = f"_{length}_{signal}"
    fisher.name = f"FISHERT{_props}"
    signalma.name = f"FISHERTs{_props}"
    fisher.category = signalma.category = "momentum"

    # 准备返回的 DataFrame
    data = {fisher.name: fisher, signalma.name: signalma}
    df = DataFrame(data)
    df.name = f"FISHERT{_props}"
    df.category = fisher.category

    return df

# 设置 Fisher Transform (FISHT) 指标的计算说明
fisher.__doc__ = \
"""Fisher Transform (FISHT)

Attempts to identify significant price reversals by normalizing prices over a
user-specified number of periods. A reversal signal is suggested when the the
two lines cross.

Sources:
    TradingView (Correlation >99%)

Calculation:
    Default Inputs:
        length=9, signal=1
    HL2 = hl2(high, low)
    HHL2 = HL2.rolling(length).max()
    LHL2 = HL2.rolling(length).min()

    HLR = HHL2 - LHL2
    HLR[HLR < 0.001] = 0.001

    position = ((HL2 - LHL2) / HLR) - 0.5

    v = 0
    m = high.size
    FISHER = [npNaN for _ in range(0, length - 1)] + [0]
    for i in range(length, m):
        v = 0.66 * position[i] + 0.67 * v
        if v < -0.99: v = -0.999
        if v >  0.99: v =  0.999
        FISHER.append(0.5 * (nplog((1 + v) / (1 - v)) + FISHER[i - 1]))
"""
    # 使用 FISHER 的 shift() 方法来对信号进行移位处理，并将结果赋值给 SIGNAL
    SIGNAL = FISHER.shift(signal)
# 定义一个函数，用于计算 Fisher 变换后的特征
def fisher(high, low, length=9, signal=1, offset=0, **kwargs):
    # 在参数中传入的高价序列
    high
    # 在参数中传入的低价序列
    low
    # Fisher 变换的周期，默认为9
    length
    # Fisher 信号的周期，默认为1
    signal
    # 结果的偏移周期数，默认为0
    offset

    # 可选参数
    # 填充缺失值的数值或填充方法
    fillna = kwargs.get('fillna')
    fill_method = kwargs.get('fill_method')

    # 返回新生成的特征序列
    return pd.Series

# `.\pandas-ta\pandas_ta\momentum\inertia.py`

```py
# 设置文件编码为 utf-8
# 从 pandas_ta.overlap 模块导入 linreg 函数
# 从 pandas_ta.volatility 模块导入 rvi 函数
# 从 pandas_ta.utils 模块导入 get_drift, get_offset, verify_series 函数
def inertia(close=None, high=None, low=None, length=None, rvi_length=None, scalar=None, refined=None, thirds=None, mamode=None, drift=None, offset=None, **kwargs):
    """Indicator: Inertia (INERTIA)"""
    # 验证参数
   # 如果 length 存在且大于 0，则将其转换为整数，否则设为 20
    length = int(length) if length and length > 0 else 20
   # 如果 rvi_length 存在且大于 0，则将其转换为整数，否则设为 14
    rvi_length = int(rvi_length) if rvi_length and rvi_length > 0 else 14
   # 如果 scalar 存在且大于 0，则将其转换为浮点数，否则设为 100
    scalar = float(scalar) if scalar and scalar > 0 else 100
   # 如果 refined 为 None，则设为 False
    refined = False if refined is None else True
   # 如果 thirds 为 None，则设为 False
    thirds = False if thirds is None else True
   # 如果 mamode 不是字符串，则设为 "ema"
    mamode = mamode if isinstance(mamode, str) else "ema"
   # 选择 length 和 rvi_length 中的最大值
    _length = max(length, rvi_length)
   # 验证 close 序列
    close = verify_series(close, _length)
   # 获取 drift 值
    drift = get_drift(drift)
   # 获取 offset 值
    offset = get_offset(offset)

    # 如果 close 为 None，则返回
    if close is None: return

    # 如果 refined 或 thirds 为 True
    if refined or thirds:
        # 验证 high 和 low 序列
        high = verify_series(high, _length)
        low = verify_series(low, _length)
        # 如果 high 或 low 为 None，则返回
        if high is None or low is None: return

    # 计算结果
    if refined:
        # 使用 'r' 模式计算 rvi
        _mode, rvi_ = "r", rvi(close, high=high, low=low, length=rvi_length, scalar=scalar, refined=refined, mamode=mamode)
    elif thirds:
        # 使用 't' 模式计算 rvi
        _mode, rvi_ = "t", rvi(close, high=high, low=low, length=rvi_length, scalar=scalar, thirds=thirds, mamode=mamode)
    else:
        # 计算 rvi
        _mode, rvi_ = "",  rvi(close, length=rvi_length, scalar=scalar, mamode=mamode)

    # 计算惯性
    inertia = linreg(rvi_, length=length)

    # 偏移
    if offset != 0:
        inertia = inertia.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        inertia.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        inertia.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置名称和类别
    _props = f"_{length}_{rvi_length}"
    inertia.name = f"INERTIA{_mode}{_props}"
    inertia.category = "momentum"

    return inertia

# 设置 inertia 函数的文档字符串
inertia.__doc__ = \
"""Inertia (INERTIA)

Inertia was developed by Donald Dorsey and was introduced his article
in September, 1995. It is the Relative Vigor Index smoothed by the Least
Squares Moving Average. Postive Inertia when values are greater than 50,
Negative Inertia otherwise.

Sources:
    https://www.investopedia.com/terms/r/relative_vigor_index.asp

Calculation:
    Default Inputs:
        length=14, ma_length=20
    LSQRMA = Least Squares Moving Average

    INERTIA = LSQRMA(RVI(length), ma_length)

Args:
    open_ (pd.Series): Series of 'open's
    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: 20
    rvi_length (int): RVI period. Default: 14
    refined (bool): Use 'refined' calculation. Default: False
    thirds (bool): Use 'thirds' calculation. Default: False
    mamode (str): See ```help(ta.ma)```py. Default: 'ema'
    drift (int): The difference period. Default: 1

"""
    offset (int): How many periods to offset the result. Default: 0

这是一个函数参数的说明。`offset` 是一个整数类型的参数，表示结果要偏移的周期数，默认为0。
# 接收关键字参数
Kwargs:
    # fillna 参数用于填充缺失值，可选参数，传入的 value 将用于填充 DataFrame 中的缺失值
    fillna (value, optional): pd.DataFrame.fillna(value)
    # fill_method 参数指定填充的方法，可选参数
    fill_method (value, optional): Type of fill method

# 返回一个 Pandas Series 对象
Returns:
    # 返回一个由该函数生成的新特征
    pd.Series: New feature generated.

.\pandas-ta\pandas_ta\momentum\kdj.py

# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta 库中导入 rma 函数
from pandas_ta.overlap import rma
# 从 pandas_ta 库中导入 get_offset、non_zero_range、verify_series 函数
from pandas_ta.utils import get_offset, non_zero_range, verify_series


# 定义 KDJ 指标函数
def kdj(high=None, low=None, close=None, length=None, signal=None, offset=None, **kwargs):
    """Indicator: KDJ (KDJ)"""
    # 验证参数
    length = int(length) if length and length > 0 else 9
    signal = int(signal) if signal and signal > 0 else 3
    _length = max(length, signal)
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    close = verify_series(close, _length)
    offset = get_offset(offset)

    # 如果 high、low、close 任一参数为 None，则返回空值
    if high is None or low is None or close is None: return

    # 计算结果
    highest_high = high.rolling(length).max()
    lowest_low = low.rolling(length).min()

    fastk = 100 * (close - lowest_low) / non_zero_range(highest_high, lowest_low)

    k = rma(fastk, length=signal)
    d = rma(k, length=signal)
    j = 3 * k - 2 * d

    # 偏移
    if offset != 0:
        k = k.shift(offset)
        d = d.shift(offset)
        j = j.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        k.fillna(kwargs["fillna"], inplace=True)
        d.fillna(kwargs["fillna"], inplace=True)
        j.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        k.fillna(method=kwargs["fill_method"], inplace=True)
        d.fillna(method=kwargs["fill_method"], inplace=True)
        j.fillna(method=kwargs["fill_method"], inplace=True)

    # 命名和分类
    _params = f"_{length}_{signal}"
    k.name = f"K{_params}"
    d.name = f"D{_params}"
    j.name = f"J{_params}"
    k.category = d.category = j.category = "momentum"

    # 准备返回的 DataFrame
    kdjdf = DataFrame({k.name: k, d.name: d, j.name: j})
    kdjdf.name = f"KDJ{_params}"
    kdjdf.category = "momentum"

    return kdjdf


# 设置 KDJ 函数的文档字符串
kdj.__doc__ = \
"""KDJ (KDJ)

The KDJ indicator is actually a derived form of the Slow
Stochastic with the only difference being an extra line
called the J line. The J line represents the divergence
of the %D value from the %K. The value of J can go
beyond [0, 100] for %K and %D lines on the chart.

Sources:
    https://www.prorealcode.com/prorealtime-indicators/kdj/
    https://docs.anychart.com/Stock_Charts/Technical_Indicators/Mathematical_Description#kdj

Calculation:
    Default Inputs:
        length=9, signal=3
    LL = low for last 9 periods
    HH = high for last 9 periods

    FAST_K = 100 * (close - LL) / (HH - LL)

    K = RMA(FAST_K, signal)
    D = RMA(K, signal)
    J = 3K - 2D

Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    length (int): Default: 9
    signal (int): Default: 3
    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.
"""