.\pandas-ta\pandas_ta\overlap\wcp.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 # 定义函数 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. """
.\pandas-ta\pandas_ta\overlap\wma.py
# -*- 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. """
.\pandas-ta\pandas_ta\overlap\zlma.py
# -*- 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. """
.\pandas-ta\pandas_ta\overlap\__init__.py
# 导入 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
.\pandas-ta\pandas_ta\performance\drawdown.py
# 导入所需模块 # -*- 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 """
PandasTA 源码解析（十一）（2）https://developer.aliyun.com/article/1506213
