PandasTA 源码解析（十）（3）

# -*- 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.
"""

# -*- 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.

# -*- 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.
"""

# -*- 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.
"""