.\pandas-ta\pandas_ta\overlap\sma.py

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

# 从 pandas_ta 库中导入 Imports 对象
from pandas_ta import Imports
# 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义简单移动平均（Simple Moving Average，SMA）指标函数
def sma(close, length=None, talib=None, offset=None, **kwargs):
    """Indicator: Simple Moving Average (SMA)"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则默认为 10
    length = int(length) if length and length > 0 else 10
    # 如果 kwargs 中存在 "min_periods" 并且不为 None，则将其转换为整数，否则默认为 length
    min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
    # 验证 close 序列，确保其长度至少为 length 或 min_periods
    close = verify_series(close, max(length, min_periods))
    # 获取偏移量
    offset = get_offset(offset)
    # 如果 talib 存在且为布尔类型，则将其转换为布尔值，否则默认为 True
    mode_tal = bool(talib) if isinstance(talib, bool) else True

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

    # 计算结果
    # 如果 Imports["talib"] 为 True 且 mode_tal 为 True，则使用 TA Lib 库中的 SMA 函数计算移动平均值
    if Imports["talib"] and mode_tal:
        from talib import SMA
        sma = SMA(close, length)
    else:
        # 否则，使用 pandas 库中的 rolling 和 mean 方法计算移动平均值
        sma = close.rolling(length, min_periods=min_periods).mean()

    # 偏移
    # 如果偏移量不为 0，则对移动平均值进行偏移
    if offset != 0:
        sma = sma.shift(offset)

    # 处理填充
    # 如果 kwargs 中存在 "fillna"，则使用指定的值填充 NaN 值
    if "fillna" in kwargs:
        sma.fillna(kwargs["fillna"], inplace=True)
    # 如果 kwargs 中存在 "fill_method"，则使用指定的填充方法填充 NaN 值
    if "fill_method" in kwargs:
        sma.fillna(method=kwargs["fill_method"], inplace=True)

    # 名称和类别
    # 设置移动平均值序列的名称为 "SMA_长度"
    sma.name = f"SMA_{length}"
    # 设置移动平均值序列的类别为 "overlap"
    sma.category = "overlap"

    # 返回移动平均值序列
    return sma

# 设置 sma 函数的文档字符串
sma.__doc__ = \
"""Simple Moving Average (SMA)

The Simple Moving Average is the classic moving average that is the equally
weighted average over n periods.

Sources:
    https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/simple-moving-average-sma/

Calculation:
    Default Inputs:
        length=10
    SMA = SUM(close, length) / length

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
        version. Default: True
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    adjust (bool): Default: True
    presma (bool, optional): If True, uses SMA for initial value.
    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\overlap\ssf.py

# -*- coding: utf-8 -*-
# 导入 numpy 库，并将其中的 cos、exp、pi、sqrt 函数别名为 npCos、npExp、npPi、npSqrt
from numpy import cos as npCos
from numpy import exp as npExp
from numpy import pi as npPi
from numpy import sqrt as npSqrt
# 从 pandas_ta.utils 模块中导入 get_offset、verify_series 函数
from pandas_ta.utils import get_offset, verify_series


# 定义函数 ssf，实现 Ehler 的超平滑滤波器（SSF）
def ssf(close, length=None, poles=None, offset=None, **kwargs):
    """Indicator: Ehler's Super Smoother Filter (SSF)"""
    # 验证参数
    # 如果 length 不为空且大于 0，则将其转换为整数，否则设为 10
    length = int(length) if length and length > 0 else 10
    # 如果 poles 不为空且在 [2, 3] 中，则将其转换为整数，否则设为 2
    poles = int(poles) if poles in [2, 3] else 2
    # 验证 close 是否为有效的 Series，长度为 length
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    m = close.size
    # 复制 close 到 ssf
    ssf = close.copy()

    # 根据 poles 的值进行不同的计算
    if poles == 3:
        # 计算参数
        x = npPi / length # x = PI / n
        a0 = npExp(-x) # e^(-x)
        b0 = 2 * a0 * npCos(npSqrt(3) * x) # 2e^(-x)*cos(3^(.5) * x)
        c0 = a0 * a0 # e^(-2x)

        c4 = c0 * c0 # e^(-4x)
        c3 = -c0 * (1 + b0) # -e^(-2x) * (1 + 2e^(-x)*cos(3^(.5) * x))
        c2 = c0 + b0 # e^(-2x) + 2e^(-x)*cos(3^(.5) * x)
        c1 = 1 - c2 - c3 - c4

        # 循环计算 ssf
        for i in range(0, m):
            ssf.iloc[i] = c1 * close.iloc[i] + c2 * ssf.iloc[i - 1] + c3 * ssf.iloc[i - 2] + c4 * ssf.iloc[i - 3]

    else: # poles == 2
        # 计算参数
        x = npPi * npSqrt(2) / length # x = PI * 2^(.5) / n
        a0 = npExp(-x) # e^(-x)
        a1 = -a0 * a0 # -e^(-2x)
        b1 = 2 * a0 * npCos(x) # 2e^(-x)*cos(x)
        c1 = 1 - a1 - b1 # e^(-2x) - 2e^(-x)*cos(x) + 1

        # 循环计算 ssf
        for i in range(0, m):
            ssf.iloc[i] = c1 * close.iloc[i] + b1 * ssf.iloc[i - 1] + a1 * ssf.iloc[i - 2]

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

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

    # 设置名称和类别
    ssf.name = f"SSF_{length}_{poles}"
    ssf.category = "overlap"

    return ssf


# 设置函数 ssf 的文档字符串
ssf.__doc__ = \
"""Ehler's Super Smoother Filter (SSF) © 2013

John F. Ehlers's solution to reduce lag and remove aliasing noise with his
research in aerospace analog filter design. This indicator comes with two
versions determined by the keyword poles. By default, it uses two poles but
there is an option for three poles. Since SSF is a (Resursive) Digital Filter,
the number of poles determine how many prior recursive SSF bars to include in
the design of the filter. So two poles uses two prior SSF bars and three poles
uses three prior SSF bars for their filter calculations.

Sources:
    http://www.stockspotter.com/files/PredictiveIndicators.pdf
    https://www.tradingview.com/script/VdJy0yBJ-Ehlers-Super-Smoother-Filter/
    https://www.mql5.com/en/code/588
    https://www.mql5.com/en/code/589

Calculation:
    Default Inputs:
        length=10, poles=[2, 3]

    See the source code or Sources listed above.

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    poles (int): The number of poles to use, either 2 or 3. Default: 2

"""
    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

# 返回值说明
Returns:
    # 返回一个 Pandas Series 对象，表示生成的新特征
    pd.Series: New feature generated.

.\pandas-ta\pandas_ta\overlap\supertrend.py

# -*- coding: utf-8 -*-
# 导入 numpy 库中的 nan 函数并重命名为 npNaN
from numpy import nan as npNaN
# 导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta 库中导入 overlap 模块中的 hl2 函数
from pandas_ta.overlap import hl2
# 从 pandas_ta 库中导入 volatility 模块中的 atr 函数
from pandas_ta.volatility import atr
# 从 pandas_ta 库中导入 utils 模块中的 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义函数 supertrend，计算 Supertrend 指标
def supertrend(high, low, close, length=None, multiplier=None, offset=None, **kwargs):
    """Indicator: Supertrend"""
    # 验证参数
    length = int(length) if length and length > 0 else 7
    multiplier = float(multiplier) if multiplier and multiplier > 0 else 3.0
    high = verify_series(high, length)
    low = verify_series(low, length)
    close = verify_series(close, length)
    offset = get_offset(offset)

    if high is None or low is None or close is None: return

    # 计算结果
    m = close.size
    dir_, trend = [1] * m, [0] * m
    long, short = [npNaN] * m, [npNaN] * m

    hl2_ = hl2(high, low)
    matr = multiplier * atr(high, low, close, length)
    upperband = hl2_ + matr
    lowerband = hl2_ - matr

    for i in range(1, m):
        if close.iloc[i] > upperband.iloc[i - 1]:
            dir_[i] = 1
        elif close.iloc[i] < lowerband.iloc[i - 1]:
            dir_[i] = -1
        else:
            dir_[i] = dir_[i - 1]
            if dir_[i] > 0 and lowerband.iloc[i] < lowerband.iloc[i - 1]:
                lowerband.iloc[i] = lowerband.iloc[i - 1]
            if dir_[i] < 0 and upperband.iloc[i] > upperband.iloc[i - 1]:
                upperband.iloc[i] = upperband.iloc[i - 1]

        if dir_[i] > 0:
            trend[i] = long[i] = lowerband.iloc[i]
        else:
            trend[i] = short[i] = upperband.iloc[i]

    # 准备要返回的 DataFrame
    _props = f"_{length}_{multiplier}"
    df = DataFrame({
            f"SUPERT{_props}": trend,
            f"SUPERTd{_props}": dir_,
            f"SUPERTl{_props}": long,
            f"SUPERTs{_props}": short,
        }, index=close.index)

    df.name = f"SUPERT{_props}"
    df.category = "overlap"

    # 如果需要，应用偏移量
    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)

    return df

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

Supertrend is an overlap indicator. It is used to help identify trend
direction, setting stop loss, identify support and resistance, and/or
generate buy & sell signals.

Sources:
    http://www.freebsensetips.com/blog/detail/7/What-is-supertrend-indicator-its-calculation

Calculation:
    Default Inputs:
        length=7, multiplier=3.0
    Default Direction:
    Set to +1 or bullish trend at start

    MID = multiplier * ATR
    LOWERBAND = HL2 - MID
    UPPERBAND = HL2 + MID

    if UPPERBAND[i] < FINAL_UPPERBAND[i-1] and close[i-1] > FINAL_UPPERBAND[i-1]:
        FINAL_UPPERBAND[i] = UPPERBAND[i]
    else:
        FINAL_UPPERBAND[i] = FINAL_UPPERBAND[i-1])
"""
    # 如果当前下轨大于前一天的最终下轨，并且前一天的收盘价小于前一天的最终下轨
    if LOWERBAND[i] > FINAL_LOWERBAND[i-1] and close[i-1] < FINAL_LOWERBAND[i-1]:
        # 将当前下轨作为最终下轨
        FINAL_LOWERBAND[i] = LOWERBAND[i]
    else:
        # 否则将前一天的最终下轨作为最终下轨
        FINAL_LOWERBAND[i] = FINAL_LOWERBAND[i-1]

    # 如果当前收盘价小于等于最终上轨
    if close[i] <= FINAL_UPPERBAND[i]:
        # 将最终上轨作为超级趋势值
        SUPERTREND[i] = FINAL_UPPERBAND[i]
    else:
        # 否则将最终下轨作为超级趋势值
        SUPERTREND[i] = FINAL_LOWERBAND[i]
# 定义函数参数
Args:
    high (pd.Series): 'high' 数据序列
    low (pd.Series): 'low' 数据序列
    close (pd.Series): 'close' 数据序列
    length (int) : ATR 计算的长度。默认值为 7
    multiplier (float): 上下轨距离中间范围的系数。默认值为 3.0
    offset (int): 结果的偏移周期数。默认值为 0

Kwargs:
    fillna (value, optional): pd.DataFrame.fillna(value) 的填充值
    fill_method (value, optional): 填充方法的类型

Returns:
    pd.DataFrame: 包含 SUPERT (趋势), SUPERTd (方向), SUPERTl (长), SUPERTs (短) 列的数据框

.\pandas-ta\pandas_ta\overlap\swma.py

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

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

# 定义 Symmetric Weighted Moving Average (SWMA) 函数
def swma(close, length=None, asc=None, offset=None, **kwargs):
    """Indicator: Symmetric Weighted Moving Average (SWMA)"""
    
    # 验证参数
    length = int(length) if length and length > 0 else 10
    asc = asc if asc else True
    close = verify_series(close, length)
    offset = get_offset(offset)

    if close is None: return

    # 计算结果
    triangle = symmetric_triangle(length, weighted=True)
    swma = close.rolling(length, min_periods=length).apply(weights(triangle), raw=True)

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

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

    # 设置名称和类别
    swma.name = f"SWMA_{length}"
    swma.category = "overlap"

    return swma

# 设置 SWMA 函数的文档字符串
swma.__doc__ = \
"""Symmetric Weighted Moving Average (SWMA)

Symmetric Weighted Moving Average where weights are based on a symmetric
triangle.  For example: n=3 -> [1, 2, 1], n=4 -> [1, 2, 2, 1], etc...
This moving average has variable length in contrast to TradingView's fixed
length of 4.

Source:
    https://www.tradingview.com/study-script-reference/#fun_swma

Calculation:
    Default Inputs:
        length=10

    def weights(w):
        def _compute(x):
            return np.dot(w * x)
        return _compute

    triangle = utils.symmetric_triangle(length - 1)
    SWMA = close.rolling(length)_.apply(weights(triangle), raw=True)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    asc (bool): Recent values weigh more. 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\overlap\t3.py

# -*- coding: utf-8 -*-
# 导入 ema 函数
from .ema import ema
# 导入 Imports 模块
from pandas_ta import Imports
# 导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义 T3 函数，计算 T3 指标
def t3(close, length=None, a=None, talib=None, offset=None, **kwargs):
    """Indicator: T3"""
    # 验证参数
   # 将 length 转换为整数，如果 length 大于 0 则使用 length，否则使用默认值 10
    length = int(length) if length and length > 0 else 10
    # 将 a 转换为浮点数，如果 a 在 0 和 1 之间则使用 a，否则使用默认值 0.7
    a = float(a) if a and a > 0 and a < 1 else 0.7
    # 验证 close 数据类型为 Series，长度为 length
    close = verify_series(close, length)
    # 获取 offset
    offset = get_offset(offset)
    # 判断是否使用 talib，默认为 True
    mode_tal = bool(talib) if isinstance(talib, bool) else True

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

    # 计算结果
    if Imports["talib"] and mode_tal:
        # 如果导入了 talib 并且 mode_tal 为 True，则使用 talib 中的 T3 函数
        from talib import T3
        t3 = T3(close, length, a)
    else:
        # 计算 T3 指标的系数
        c1 = -a * a**2
        c2 = 3 * a**2 + 3 * a**3
        c3 = -6 * a**2 - 3 * a - 3 * a**3
        c4 = a**3 + 3 * a**2 + 3 * a + 1

        # 计算 ema1 到 ema6
        e1 = ema(close=close, length=length, **kwargs)
        e2 = ema(close=e1, length=length, **kwargs)
        e3 = ema(close=e2, length=length, **kwargs)
        e4 = ema(close=e3, length=length, **kwargs)
        e5 = ema(close=e4, length=length, **kwargs)
        e6 = ema(close=e5, length=length, **kwargs)
        # 计算 T3
        t3 = c1 * e6 + c2 * e5 + c3 * e4 + c4 * e3

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

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

    # 设置指标名称和类别
    t3.name = f"T3_{length}_{a}"
    t3.category = "overlap"

    return t3

# 设置 T3 函数的文档字符串
t3.__doc__ = """Tim Tillson's T3 Moving Average (T3)

Tim Tillson's T3 Moving Average is considered a smoother and more responsive
moving average relative to other moving averages.

Sources:
    http://www.binarytribune.com/forex-trading-indicators/t3-moving-average-indicator/

Calculation:
    Default Inputs:
        length=10, a=0.7
    c1 = -a^3
    c2 = 3a^2 + 3a^3 = 3a^2 * (1 + a)
    c3 = -6a^2 - 3a - 3a^3
    c4 = a^3 + 3a^2 + 3a + 1

    ema1 = EMA(close, length)
    ema2 = EMA(ema1, length)
    ema3 = EMA(ema2, length)
    ema4 = EMA(ema3, length)
    ema5 = EMA(ema4, length)
    ema6 = EMA(ema5, length)
    T3 = c1 * ema6 + c2 * ema5 + c3 * ema4 + c4 * ema3

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    a (float): 0 < a < 1. Default: 0.7
    talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
        version. Default: True
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    adjust (bool): Default: True
    presma (bool, optional): If True, uses SMA for initial value.
    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\overlap\tema.py

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

# 从 ema 模块中导入 ema 函数
from .ema import ema
# 从 pandas_ta 模块中导入 Imports 类
from pandas_ta import Imports
# 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义 Triple Exponential Moving Average (TEMA) 指标函数
def tema(close, length=None, talib=None, offset=None, **kwargs):
    """Indicator: Triple Exponential Moving Average (TEMA)"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 10
    length = int(length) if length and length > 0 else 10
    # 验证 close 数据，并设定长度
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)
    # 判断是否使用 talib 模式
    mode_tal = bool(talib) if isinstance(talib, bool) else True

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

    # 计算结果
    if Imports["talib"] and mode_tal:
        # 如果导入了 talib 模块并且使用 talib 模式，则调用 talib 中的 TEMA 函数
        from talib import TEMA
        tema = TEMA(close, length)
    else:
        # 否则，使用 ema 函数计算三次指数移动平均
        ema1 = ema(close=close, length=length, **kwargs)
        ema2 = ema(close=ema1, length=length, **kwargs)
        ema3 = ema(close=ema2, length=length, **kwargs)
        tema = 3 * (ema1 - ema2) + ema3

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

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

    # 设置指标名称和类别
    tema.name = f"TEMA_{length}"
    tema.category = "overlap"

    return tema

# 设置 TEMA 函数的文档字符串
tema.__doc__ = \
"""Triple Exponential Moving Average (TEMA)

A less laggy Exponential Moving Average.

Sources:
    https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/triple-exponential-moving-average-tema/

Calculation:
    Default Inputs:
        length=10
    EMA = Exponential Moving Average
    ema1 = EMA(close, length)
    ema2 = EMA(ema1, length)
    ema3 = EMA(ema2, length)
    TEMA = 3 * (ema1 - ema2) + ema3

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
        version. Default: True
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    adjust (bool): Default: True
    presma (bool, optional): If True, uses SMA for initial value.
    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\overlap\trima.py

# -*- coding: utf-8 -*-
# 从sma模块导入sma函数
from .sma import sma
# 从pandas_ta模块导入Imports
from pandas_ta import Imports
# 从pandas_ta.utils模块导入get_offset和verify_series函数
from pandas_ta.utils import get_offset, verify_series

# 定义Triangular Moving Average (TRIMA)指标函数
def trima(close, length=None, talib=None, offset=None, **kwargs):
    """Indicator: Triangular Moving Average (TRIMA)"""
    # 验证参数
    # 如果length存在且大于0，则将其转换为整数，否则设为10
    length = int(length) if length and length > 0 else 10
    # 验证close序列，确保长度为length
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)
    # 判断是否使用talib，默认为True
    mode_tal = bool(talib) if isinstance(talib, bool) else True

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

    # 计算结果
    if Imports["talib"] and mode_tal:
        # 如果导入了talib并且mode_tal为True，则使用talib库计算TRIMA
        from talib import TRIMA
        trima = TRIMA(close, length)
    else:
        # 否则，计算TRIMA的一半长度
        half_length = round(0.5 * (length + 1))
        # 计算SMA1
        sma1 = sma(close, length=half_length)
        # 计算TRIMA
        trima = sma(sma1, length=half_length)

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

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

    # 设置名称和类别
    trima.name = f"TRIMA_{length}"
    trima.category = "overlap"

    return trima


# 设置TRIMA函数的文档字符串
trima.__doc__ = \
"""Triangular Moving Average (TRIMA)

A weighted moving average where the shape of the weights are triangular and the
greatest weight is in the middle of the period.

Sources:
    https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/triangular-moving-average-trima/
    tma = sma(sma(src, ceil(length / 2)), floor(length / 2) + 1)  # Tradingview
    trima = sma(sma(x, n), n)  # Tradingview

Calculation:
    Default Inputs:
        length=10
    SMA = Simple Moving Average
    half_length = round(0.5 * (length + 1))
    SMA1 = SMA(close, half_length)
    TRIMA = SMA(SMA1, half_length)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
        version. Default: True
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    adjust (bool): 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\overlap\vidya.py

# -*- coding: utf-8 -*-
# 导入 numpy 库中的 nan 并重命名为 npNaN
from numpy import nan as npNaN
# 从 pandas 库中导入 Series 类
from pandas import Series
# 从 pandas_ta.utils 模块中导入 get_drift, get_offset, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, verify_series

# 定义变量指数动态平均值（VIDYA）指标函数
def vidya(close, length=None, drift=None, offset=None, **kwargs):
    """Indicator: Variable Index Dynamic Average (VIDYA)"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 14
    length = int(length) if length and length > 0 else 14
    # 验证 close 是否为 Series 类型，并且长度符合要求
    close = verify_series(close, length)
    # 获取漂移值
    drift = get_drift(drift)
    # 获取偏移值
    offset = get_offset(offset)

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

    # 定义 Chande 动量振荡器（CMO）函数
    def _cmo(source: Series, n:int , d: int):
        """Chande Momentum Oscillator (CMO) Patch
        For some reason: from pandas_ta.momentum import cmo causes
        pandas_ta.momentum.coppock to not be able to import it's
        wma like from pandas_ta.overlap import wma?
        Weird Circular TypeError!?!
        """
        # 计算动量
        mom = source.diff(d)
        # 获取正值
        positive = mom.copy().clip(lower=0)
        # 获取负值
        negative = mom.copy().clip(upper=0).abs()
        # 计算正值的滚动和
        pos_sum = positive.rolling(n).sum()
        # 计算负值的滚动和
        neg_sum = negative.rolling(n).sum()
        # 返回 CMO 值
        return (pos_sum - neg_sum) / (pos_sum + neg_sum)

    # 计算结果
    m = close.size
    alpha = 2 / (length + 1)
    abs_cmo = _cmo(close, length, drift).abs()
    vidya = Series(0, index=close.index)
    for i in range(length, m):
        vidya.iloc[i] = alpha * abs_cmo.iloc[i] * close.iloc[i] + vidya.iloc[i - 1] * (1 - alpha * abs_cmo.iloc[i])
    vidya.replace({0: npNaN}, inplace=True)

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

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

    # 设置名称和类别
    vidya.name = f"VIDYA_{length}"
    vidya.category = "overlap"

    return vidya


# 设置函数文档字符串
vidya.__doc__ = \
"""Variable Index Dynamic Average (VIDYA)

Variable Index Dynamic Average (VIDYA) was developed by Tushar Chande. It is
similar to an Exponential Moving Average but it has a dynamically adjusted
lookback period dependent on relative price volatility as measured by Chande
Momentum Oscillator (CMO). When volatility is high, VIDYA reacts faster to
price changes. It is often used as moving average or trend identifier.

Sources:
    https://www.tradingview.com/script/hdrf0fXV-Variable-Index-Dynamic-Average-VIDYA/
    https://www.perfecttrendsystem.com/blog_mt4_2/en/vidya-indicator-for-mt4

Calculation:
    Default Inputs:
        length=10, adjust=False, sma=True
    if sma:
        sma_nth = close[0:length].sum() / length
        close[:length - 1] = np.NaN
        close.iloc[length - 1] = sma_nth
    EMA = close.ewm(span=length, adjust=adjust).mean()

Args:
    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:
    adjust (bool, optional): Use adjust option for EMA calculation. Default: False

"""
    # sma (bool, optional): 如果为 True，使用简单移动平均（SMA）作为 EMA 计算的初始值。默认为 True
    sma (bool, optional): If True, uses SMA for initial value for EMA calculation. Default: True
    
    # talib (bool): 如果为 True，则使用 TA-Lib 的实现来计算 CMO。否则使用 EMA 版本。默认为 True
    talib (bool): If True, uses TA-Libs implementation for CMO. Otherwise uses EMA version. Default: True
    
    # fillna (value, optional): pd.DataFrame.fillna(value) 的参数，用于指定填充缺失值的值
    fillna (value, optional): Parameter for pd.DataFrame.fillna(value) to specify the value to fill missing values with.
    
    # fill_method (value, optional): 填充缺失值的方法类型
    fill_method (value, optional): Type of fill method.
# 返回类型声明：pd.Series，表示生成的新特征是一个 Pandas 的 Series 对象
pd.Series: New feature generated.

.\pandas-ta\pandas_ta\overlap\vwap.py

# -*- coding: utf-8 -*-
# 导入依赖库中的hlc3函数
from .hlc3 import hlc3
# 导入辅助函数
from pandas_ta.utils import get_offset, is_datetime_ordered, verify_series

# VWAP函数定义
def vwap(high, low, close, volume, anchor=None, offset=None, **kwargs):
    """Indicator: Volume Weighted Average Price (VWAP)"""
    # 验证参数
    high = verify_series(high)  # 验证high序列
    low = verify_series(low)  # 验证low序列
    close = verify_series(close)  # 验证close序列
    volume = verify_series(volume)  # 验证volume序列
    anchor = anchor.upper() if anchor and isinstance(anchor, str) and len(anchor) >= 1 else "D"  # 将anchor转换为大写字母，如果为空则默认为"D"
    offset = get_offset(offset)  # 获取offset值

    # 计算typical_price
    typical_price = hlc3(high=high, low=low, close=close)
    # 检查volume序列是否按时间排序
    if not is_datetime_ordered(volume):
        print(f"[!] VWAP volume series is not datetime ordered. Results may not be as expected.")
    # 检查typical_price序列是否按时间排序
    if not is_datetime_ordered(typical_price):
        print(f"[!] VWAP price series is not datetime ordered. Results may not be as expected.")

    # 计算结果
    wp = typical_price * volume  # 计算加权价格
    vwap  = wp.groupby(wp.index.to_period(anchor)).cumsum()  # 计算累积加权价格
    vwap /= volume.groupby(volume.index.to_period(anchor)).cumsum()  # 计算累积成交量

    # 偏移
    if offset != 0:
        vwap = vwap.shift(offset)  # 偏移vwap序列

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

    # 设置名称和类别
    vwap.name = f"VWAP_{anchor}"  # 设置VWAP的名称
    vwap.category = "overlap"  # 设置VWAP的类别为overlap（重叠型指标）

    return vwap  # 返回VWAP序列


# VWAP文档字符串
vwap.__doc__ = \
"""Volume Weighted Average Price (VWAP)

The Volume Weighted Average Price that measures the average typical price
by volume.  It is typically used with intraday charts to identify general
direction.

Sources:
    https://www.tradingview.com/wiki/Volume_Weighted_Average_Price_(VWAP)
    https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/volume-weighted-average-price-vwap/
    https://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:vwap_intraday

Calculation:
    tp = typical_price = hlc3(high, low, close)
    tpv = tp * volume
    VWAP = tpv.cumsum() / volume.cumsum()

Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    volume (pd.Series): Series of 'volume's
    anchor (str): How to anchor VWAP. Depending on the index values, it will
        implement various Timeseries Offset Aliases as listed here:
        https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#timeseries-offset-aliases
        Default: "D".
    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\overlap\vwma.py

# -*- coding: utf-8 -*-
# 从sma模块中导入sma函数
from .sma import sma
# 从pandas_ta.utils中导入get_offset和verify_series函数
from pandas_ta.utils import get_offset, verify_series

# 定义一个名为vwma的函数，计算Volume Weighted Moving Average（VWMA）
def vwma(close, volume, length=None, offset=None, **kwargs):
    """Indicator: Volume Weighted Moving Average (VWMA)"""
    # 验证参数
    # 将length转换为整数，如果length存在且大于0，否则设为10
    length = int(length) if length and length > 0 else 10
    # 验证close序列，长度为length
    close = verify_series(close, length)
    # 验证volume序列，长度为length
    volume = verify_series(volume, length)
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    # 计算pv（close * volume）
    pv = close * volume
    # 计算vwma（pv的SMA / volume的SMA）
    vwma = sma(close=pv, length=length) / sma(close=volume, length=length)

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

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

    # 名称和类别
    # 设置vwma的名称为"VWMA_长度"
    vwma.name = f"VWMA_{length}"
    # 设置vwma的类别为"overlap"
    vwma.category = "overlap"

    return vwma

# 设置vwma函数的文档字符串
vwma.__doc__ = \
"""Volume Weighted Moving Average (VWMA)

Volume Weighted Moving Average.

Sources:
    https://www.motivewave.com/studies/volume_weighted_moving_average.htm

Calculation:
    Default Inputs:
        length=10
    SMA = Simple Moving Average
    pv = close * volume
    VWMA = SMA(pv, length) / SMA(volume, length)

Args:
    close (pd.Series): Series of 'close's
    volume (pd.Series): Series of 'volume's
    length (int): It's 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.
"""