PandasTA 源码解析（十一）（1）

# -*- 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
# 定义函数 wcp，计算加权收盘价（WCP）
def wcp(high, low, close, talib=None, offset=None, **kwargs):
    """Indicator: Weighted Closing Price (WCP)"""
    # 验证参数
   high = verify_series(high)
   low = verify_series(low)
   close = verify_series(close)
    offset = get_offset(offset)
    mode_tal = bool(talib) if isinstance(talib, bool) else True
    # 计算结果
   if Imports["talib"] and mode_tal:
        from talib import WCLPRICE
        wcp = WCLPRICE(high, low, close)
    else:
        wcp = (high + low + 2 * close) / 4
    # 偏移
   if offset != 0:
        wcp = wcp.shift(offset)
    # 处理填充
   if "fillna" in kwargs:
        wcp.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        wcp.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置名称和类别
    wcp.name = "WCP"
    wcp.category = "overlap"
    return wcp
# 设置函数 wcp 的文档字符串
wcp.__doc__ = \
"""Weighted Closing Price (WCP)
Weighted Closing Price is the weighted price given: high, low
and double the close.
Sources:
    https://www.fmlabs.com/reference/default.htm?url=WeightedCloses.htm
Calculation:
    WCP = (2 * close + high + low) / 4
Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    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:
    fillna (value, optional): pd.DataFrame.fillna(value)
    fill_method (value, optional): Type of fill method
Returns:
    pd.Series: New feature generated.
"""

# -*- coding: utf-8 -*-
# 导入需要的模块和函数
from pandas import Series
from pandas_ta import Imports
from pandas_ta.utils import get_offset, verify_series
def wma(close, length=None, asc=None, talib=None, offset=None, **kwargs):
    """Indicator: Weighted Moving Average (WMA)"""
    # 验证参数
    length = int(length) if length and length > 0 else 10  # 确定长度为正整数，默认为10
    asc = asc if asc else True  # 默认为升序
    close = verify_series(close, length)  # 确保close为Series，长度为length
    offset = get_offset(offset)  # 获取偏移量
    mode_tal = bool(talib) if isinstance(talib, bool) else True  # 是否使用TA Lib，默认为True
    if close is None: return  # 如果close为空，则返回
    # 计算结果
    if Imports["talib"] and mode_tal:  # 如果安装了TA Lib且使用TA Lib模式
        from talib import WMA
        wma = WMA(close, length)  # 使用TA Lib中的WMA函数计算WMA
    else:
        from numpy import arange as npArange
        from numpy import dot as npDot
        total_weight = 0.5 * length * (length + 1)  # 计算总权重
        weights_ = Series(npArange(1, length + 1))  # 创建1到length的Series
        weights = weights_ if asc else weights_[::-1]  # 如果升序，则不变；否则倒序
        def linear(w):
            def _compute(x):
                return npDot(x, w) / total_weight  # 线性权重计算WMA
            return _compute
        close_ = close.rolling(length, min_periods=length)  # 创建长度为length的rolling对象
        wma = close_.apply(linear(weights), raw=True)  # 应用线性权重计算WMA
    # 偏移
    if offset != 0:
        wma = wma.shift(offset)  # 偏移结果
    # 处理填充
    if "fillna" in kwargs:
        wma.fillna(kwargs["fillna"], inplace=True)  # 使用指定值填充空值
    if "fill_method" in kwargs:
        wma.fillna(method=kwargs["fill_method"], inplace=True)  # 使用指定填充方法填充空值
    # 名称和类别
    wma.name = f"WMA_{length}"  # 设置名称
    wma.category = "overlap"  # 设置类别
    return wma  # 返回WMA
wma.__doc__ = \
"""Weighted Moving Average (WMA)
The Weighted Moving Average where the weights are linearly increasing and
the most recent data has the heaviest weight.
Sources:
    https://en.wikipedia.org/wiki/Moving_average#Weighted_moving_average
Calculation:
    Default Inputs:
        length=10, asc=True
    total_weight = 0.5 * length * (length + 1)
    weights_ = [1, 2, ..., length + 1]  # Ascending
    weights = weights if asc else weights[::-1]
    def linear_weights(w):
        def _compute(x):
            return (w * x).sum() / total_weight
        return _compute
    WMA = close.rolling(length)_.apply(linear_weights(weights), 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
    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:
    fillna (value, optional): pd.DataFrame.fillna(value)
    fill_method (value, optional): Type of fill method
Returns:
    pd.Series: New feature generated.
"""

# -*- coding: utf-8 -*-
# 导入所需的函数库
from . import (
    dema, ema, hma, linreg, rma, sma, swma, t3, tema, trima, vidya, wma
)
# 导入辅助函数
from pandas_ta.utils import get_offset, verify_series
# 定义 Zero Lag Moving Average (ZLMA) 函数
def zlma(close, length=None, mamode=None, offset=None, **kwargs):
    """Indicator: Zero Lag Moving Average (ZLMA)"""
    # 验证参数
    length = int(length) if length and length > 0 else 10
    mamode = mamode.lower() if isinstance(mamode, str) else "ema"
    close = verify_series(close, length)
    offset = get_offset(offset)
    if close is None: return
    # 计算结果
    lag = int(0.5 * (length - 1))
    close_ = 2 * close - close.shift(lag)
    # 根据不同的 mamode 选择不同的移动平均方法
    if   mamode == "dema":   zlma = dema(close_, length=length, **kwargs)
    elif mamode == "hma":    zlma = hma(close_, length=length, **kwargs)
    elif mamode == "linreg": zlma = linreg(close_, length=length, **kwargs)
    elif mamode == "rma":    zlma = rma(close_, length=length, **kwargs)
    elif mamode == "sma":    zlma = sma(close_, length=length, **kwargs)
    elif mamode == "swma":   zlma = swma(close_, length=length, **kwargs)
    elif mamode == "t3":     zlma = t3(close_, length=length, **kwargs)
    elif mamode == "tema":   zlma = tema(close_, length=length, **kwargs)
    elif mamode == "trima":  zlma = trima(close_, length=length, **kwargs)
    elif mamode == "vidya":  zlma = vidya(close_, length=length, **kwargs)
    elif mamode == "wma":    zlma = wma(close_, length=length, **kwargs)
    else:                    zlma = ema(close_, length=length, **kwargs) # "ema"
    # 偏移结果
    if offset != 0:
        zlma = zlma.shift(offset)
    # 处理填充
    if "fillna" in kwargs:
        zlma.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        zlma.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置名称和类别
    zlma.name = f"ZL_{zlma.name}"
    zlma.category = "overlap"
    return zlma
# 设置 ZLMA 函数的文档字符串
zlma.__doc__ = \
"""Zero Lag Moving Average (ZLMA)
The Zero Lag Moving Average attempts to eliminate the lag associated
with moving averages.  This is an adaption created by John Ehler and Ric Way.
Sources:
    https://en.wikipedia.org/wiki/Zero_lag_exponential_moving_average
Calculation:
    Default Inputs:
        length=10, mamode=EMA
    EMA = Exponential Moving Average
    lag = int(0.5 * (length - 1))
    SOURCE = 2 * close - close.shift(lag)
    ZLMA = MA(kind=mamode, SOURCE, length)
Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    mamode (str): Options: 'ema', 'hma', 'sma', 'wma'. Default: 'ema'
    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.
"""

# 导入 alma 指标计算模块
from .alma import alma
# 导入 dema 指标计算模块
from .dema import dema
# 导入 ema 指标计算模块
from .ema import ema
# 导入 fwma 指标计算模块
from .fwma import fwma
# 导入 hilo 指标计算模块
from .hilo import hilo
# 导入 hl2 指标计算模块
from .hl2 import hl2
# 导入 hlc3 指标计算模块
from .hlc3 import hlc3
# 导入 hma 指标计算模块
from .hma import hma
# 导入 hwma 指标计算模块
from .hwma import hwma
# 导入 ichimoku 指标计算模块
from .ichimoku import ichimoku
# 导入 jma 指标计算模块
from .jma import jma
# 导入 kama 指标计算模块
from .kama import kama
# 导入 linreg 指标计算模块
from .linreg import linreg
# 导入 ma 指标计算模块
from .ma import ma
# 导入 mcgd 指标计算模块
from .mcgd import mcgd
# 导入 midpoint 指标计算模块
from .midpoint import midpoint
# 导入 midprice 指标计算模块
from .midprice import midprice
# 导入 ohlc4 指标计算模块
from .ohlc4 import ohlc4
# 导入 pwma 指标计算模块
from .pwma import pwma
# 导入 rma 指标计算模块
from .rma import rma
# 导入 sinwma 指标计算模块
from .sinwma import sinwma
# 导入 sma 指标计算模块
from .sma import sma
# 导入 ssf 指标计算模块
from .ssf import ssf
# 导入 supertrend 指标计算模块
from .supertrend import supertrend
# 导入 swma 指标计算模块
from .swma import swma
# 导入 t3 指标计算模块
from .t3 import t3
# 导入 tema 指标计算模块
from .tema import tema
# 导入 trima 指标计算模块
from .trima import trima
# 导入 vidya 指标计算模块
from .vidya import vidya
# 导入 vwap 指标计算模块
from .vwap import vwap
# 导入 vwma 指标计算模块
from .vwma import vwma
# 导入 wcp 指标计算模块
from .wcp import wcp
# 导入 wma 指标计算模块
from .wma import wma
# 导入 zlma 指标计算模块
from .zlma import zlma

# 导入所需模块
# -*- coding: utf-8 -*-
# 从 numpy 模块中导入 log 函数并重命名为 nplog
from numpy import log as nplog
# 从 numpy 模块中导入 seterr 函数
from numpy import seterr
# 从 pandas 模块中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义函数 drawdown，用于计算资产或投资组合的回撤情况
def drawdown(close, offset=None, **kwargs) -> DataFrame:
    """Indicator: Drawdown (DD)"""
    # 验证参数合法性，确保 close 是一个 Series 对象
    close = verify_series(close)
    # 获取偏移量
    offset = get_offset(offset)
    # 计算结果
    # 计算历史最高收盘价
    max_close = close.cummax()
    # 计算回撤
    dd = max_close - close
    # 计算回撤百分比
    dd_pct = 1 - (close / max_close)
    # 临时忽略 numpy 的警告
    _np_err = seterr()
    seterr(divide="ignore", invalid="ignore")
    # 计算回撤的对数
    dd_log = nplog(max_close) - nplog(close)
    # 恢复 numpy 的警告设置
    seterr(divide=_np_err["divide"], invalid=_np_err["invalid"])
    # 调整偏移量
    if offset != 0:
        dd = dd.shift(offset)
        dd_pct = dd_pct.shift(offset)
        dd_log = dd_log.shift(offset)
    # 处理填充值
    if "fillna" in kwargs:
        dd.fillna(kwargs["fillna"], inplace=True)
        dd_pct.fillna(kwargs["fillna"], inplace=True)
        dd_log.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        dd.fillna(method=kwargs["fill_method"], inplace=True)
        dd_pct.fillna(method=kwargs["fill_method"], inplace=True)
        dd_log.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置列名和分类
    dd.name = "DD"
    dd_pct.name = f"{dd.name}_PCT"
    dd_log.name = f"{dd.name}_LOG"
    dd.category = dd_pct.category = dd_log.category = "performance"
    # 准备返回的 DataFrame
    data = {dd.name: dd, dd_pct.name: dd_pct, dd_log.name: dd_log}
    df = DataFrame(data)
    df.name = dd.name
    df.category = dd.category
    return df
# 设置函数文档字符串
drawdown.__doc__ = \
"""Drawdown (DD)
Drawdown is a peak-to-trough decline during a specific period for an investment,
trading account, or fund. It is usually quoted as the percentage between the
peak and the subsequent trough.
Sources:
    https://www.investopedia.com/terms/d/drawdown.asp
Calculation:
    PEAKDD = close.cummax()
    DD = PEAKDD - close
    DD% = 1 - (close / PEAKDD)
    DDlog = log(PEAKDD / close)
Args:
    close (pd.Series): Series of 'close's.
    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: drawdown, drawdown percent, drawdown log columns
"""