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

.\pandas-ta\pandas_ta\performance\log_return.py

# -*- coding: utf-8 -*-
# 从 numpy 中导入 log 函数并重命名为 nplog
from numpy import log as nplog
# 从 pandas_ta.utils 中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series


def log_return(close, length=None, cumulative=None, offset=None, **kwargs):
    """Indicator: Log Return"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为 1
    length = int(length) if length and length > 0 else 1
    # 如果 cumulative 存在且为真，则设为 True，否则设为 False
    cumulative = bool(cumulative) if cumulative is not None and cumulative else False
    # 验证 close 是否为有效的 Series，并设置长度
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    if cumulative:
        # 计算累积对数收益率
        log_return = nplog(close / close.iloc[0])
    else:
        # 计算对数收益率
        log_return = nplog(close / close.shift(length))

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

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

    # 设置名称和类别
    log_return.name = f"{'CUM' if cumulative else ''}LOGRET_{length}"
    log_return.category = "performance"

    return log_return


# 设置 log_return 函数的文档字符串
log_return.__doc__ = \
"""Log Return

Calculates the logarithmic return of a Series.
See also: help(df.ta.log_return) for additional **kwargs a valid 'df'.

Sources:
    https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe

Calculation:
    Default Inputs:
        length=1, cumulative=False
    LOGRET = log( close.diff(periods=length) )
    CUMLOGRET = LOGRET.cumsum() if cumulative

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 20
    cumulative (bool): If True, returns the cumulative returns. Default: False
    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\performance\percent_return.py

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

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

# 定义一个名为 percent_return 的函数，用于计算百分比收益率
def percent_return(close, length=None, cumulative=None, offset=None, **kwargs):
    """Indicator: Percent Return"""
    
    # 验证参数
    length = int(length) if length and length > 0 else 1
    cumulative = bool(cumulative) if cumulative is not None and cumulative else False
    close = verify_series(close, length)
    offset = get_offset(offset)

    if close is None: return

    # 计算结果
    if cumulative:
        pct_return = (close / close.iloc[0]) - 1
    else:
        pct_return = close.pct_change(length) # (close / close.shift(length)) - 1

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

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

    # 设置名称和类别
    pct_return.name = f"{'CUM' if cumulative else ''}PCTRET_{length}"
    pct_return.category = "performance"

    return pct_return

# 设置 percent_return 函数的文档字符串
percent_return.__doc__ = \
"""Percent Return

Calculates the percent return of a Series.
See also: help(df.ta.percent_return) for additional **kwargs a valid 'df'.

Sources:
    https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe

Calculation:
    Default Inputs:
        length=1, cumulative=False
    PCTRET = close.pct_change(length)
    CUMPCTRET = PCTRET.cumsum() if cumulative

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 20
    cumulative (bool): If True, returns the cumulative returns. Default: False
    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\performance\__init__.py

# 设置文件编码为 UTF-8
# 导入 drawdown 模块中的 drawdown 函数
# 导入 log_return 模块中的 log_return 函数
# 导入 percent_return 模块中的 percent_return 函数

.\pandas-ta\pandas_ta\statistics\entropy.py

# -*- coding: utf-8 -*-
# 导入 log 函数并将其命名为 npLog
from numpy import log as npLog
# 导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series


# 定义熵指标函数，接受收盘价、周期、对数的基数、偏移量和其他参数
def entropy(close, length=None, base=None, offset=None, **kwargs):
    """Indicator: Entropy (ENTP)"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 10
    length = int(length) if length and length > 0 else 10
    # 如果 base 存在且大于 0，则将其转换为浮点数，否则设为默认值 2.0
    base = float(base) if base and base > 0 else 2.0
    # 验证收盘价是否是一个有效的 Series，如果不是则返回 None
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    # 计算每个价格占总和的比例
    p = close / close.rolling(length).sum()
    # 计算熵
    entropy = (-p * npLog(p) / npLog(base)).rolling(length).sum()

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

    # 处理填充
    # 如果参数中包含 "fillna"，则使用指定值填充空值
    if "fillna" in kwargs:
        entropy.fillna(kwargs["fillna"], inplace=True)
    # 如果参数中包含 "fill_method"，则使用指定的填充方法填充空值
    if "fill_method" in kwargs:
        entropy.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置名称和类别
    entropy.name = f"ENTP_{length}"
    entropy.category = "statistics"

    return entropy


# 设置熵指标函数的文档字符串
entropy.__doc__ = \
"""Entropy (ENTP)

Introduced by Claude Shannon in 1948, entropy measures the unpredictability
of the data, or equivalently, of its average information. A die has higher
entropy (p=1/6) versus a coin (p=1/2).

Sources:
    https://en.wikipedia.org/wiki/Entropy_(information_theory)

Calculation:
    Default Inputs:
        length=10, base=2

    P = close / SUM(close, length)
    E = SUM(-P * npLog(P) / npLog(base), length)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 10
    base (float): Logarithmic Base. Default: 2
    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\statistics\kurtosis.py

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

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

# 定义函数 kurtosis，计算某个时间段内的峰度
def kurtosis(close, length=None, offset=None, **kwargs):
    """Indicator: Kurtosis"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数；否则将 length 设为默认值 30
    length = int(length) if length and length > 0 else 30
    # 如果 kwargs 中有 "min_periods" 参数且不为 None，则将其转换为整数；否则将 min_periods 设为 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)

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

    # 计算结果
    # 计算 close 的滚动窗口长度为 length 的峰度，并使用 min_periods 参数指定最小期数
    kurtosis = close.rolling(length, min_periods=min_periods).kurt()

    # 处理偏移
    if offset != 0:
        # 对计算的峰度结果进行偏移
        kurtosis = kurtosis.shift(offset)

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

    # 设定指标的名称和类别
    kurtosis.name = f"KURT_{length}"
    kurtosis.category = "statistics"

    # 返回计算结果
    return kurtosis


# 为函数 kurtosis 添加文档字符串
kurtosis.__doc__ = \
"""Rolling Kurtosis

Sources:

Calculation:
    Default Inputs:
        length=30
    KURTOSIS = close.rolling(length).kurt()

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    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\statistics\mad.py

# -*- coding: utf-8 -*-
# 从 numpy 中导入 fabs 函数并重命名为 npfabs
from numpy import fabs as npfabs
# 从 pandas_ta.utils 中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series


# 定义函数：均值绝对偏差
def mad(close, length=None, offset=None, **kwargs):
    """Indicator: Mean Absolute Deviation"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 30
    length = int(length) if length and length > 0 else 30
    # 如果 kwargs 中存在 "min_periods"，则将其转换为整数，否则使用 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)

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

    # 计算结果
    def mad_(series):
        """Mean Absolute Deviation"""
        # 计算序列与其均值的绝对差值的均值
        return npfabs(series - series.mean()).mean()

    # 使用 rolling 函数计算滚动均值绝对偏差
    mad = close.rolling(length, min_periods=min_periods).apply(mad_, raw=True)

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

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

    # 设定指标的名称和类别
    mad.name = f"MAD_{length}"
    mad.category = "statistics"

    return mad


# 设置函数文档字符串
mad.__doc__ = \
"""Rolling Mean Absolute Deviation

Sources:

Calculation:
    Default Inputs:
        length=30
    mad = close.rolling(length).mad()

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    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\statistics\median.py

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

# 定义一个名为 median 的函数，用于计算中位数指标
def median(close, length=None, offset=None, **kwargs):
    """Indicator: Median"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 30
    length = int(length) if length and length > 0 else 30
    # 如果 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 是否为有效的 Series，并设定最小长度为 length 和 min_periods 中的较大值
    close = verify_series(close, max(length, min_periods))
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    # 计算 close 的滚动中位数，窗口长度为 length，最小观测数为 min_periods
    median = close.rolling(length, min_periods=min_periods).median()

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

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

    # 设置指标名称和类别
    median.name = f"MEDIAN_{length}"
    median.category = "statistics"

    return median

# 设置 median 函数的文档字符串
median.__doc__ = \
"""Rolling Median

Rolling Median of over 'n' periods. Sibling of a Simple Moving Average.

Sources:
    https://www.incrediblecharts.com/indicators/median_price.php

Calculation:
    Default Inputs:
        length=30
    MEDIAN = close.rolling(length).median()

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    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\statistics\quantile.py

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

# 定义一个名为 quantile 的函数，用于计算滚动分位数
def quantile(close, length=None, q=None, offset=None, **kwargs):
    """Indicator: Quantile"""  # 函数文档字符串，指示 quantile 函数的作用
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 30
    length = int(length) if length and length > 0 else 30
    # 如果 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
    # 如果 q 存在且大于 0 且小于 1，则将其转换为浮点数，否则设为默认值 0.5
    q = float(q) if q and q > 0 and q < 1 else 0.5
    # 验证 close 序列，确保其长度不小于 length 和 min_periods 中的较大值
    close = verify_series(close, max(length, min_periods))
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    # 使用 close 序列进行滚动窗口计算分位数，窗口长度为 length，最小周期数为 min_periods
    quantile = close.rolling(length, min_periods=min_periods).quantile(q)

    # 偏移结果
    # 如果偏移量不为 0，则对 quantile 序列进行偏移
    if offset != 0:
        quantile = quantile.shift(offset)

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

    # 序列命名和分类
    # 设置 quantile 序列的名称为 "QTL_{length}_{q}"
    quantile.name = f"QTL_{length}_{q}"
    # 设置 quantile 序列的分类为 "statistics"
    quantile.category = "statistics"

    return quantile  # 返回 quantile 序列


# 为 quantile 函数添加文档字符串，说明其作用、源以及参数等信息
quantile.__doc__ = \
"""Rolling Quantile

Sources:

Calculation:
    Default Inputs:
        length=30, q=0.5
    QUANTILE = close.rolling(length).quantile(q)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    q (float): The quantile. Default: 0.5
    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\statistics\skew.py

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


def skew(close, length=None, offset=None, **kwargs):
    """Indicator: Skew"""
    # 验证参数
    # 如果 length 存在且大于0，则将其转换为整数，否则默认为30
    length = int(length) if length and length > 0 else 30
    # 如果 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 是否为 pd.Series 类型，并保证其长度不小于 length 和 min_periods 中的较大值
    close = verify_series(close, max(length, min_periods))
    # 获取偏移量
    offset = get_offset(offset)

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

    # 计算结果
    # 计算 close 的滚动 skewness（偏度）
    skew = close.rolling(length, min_periods=min_periods).skew()

    # 偏移结果
    # 如果偏移量不为零，则对 skew 进行偏移
    if offset != 0:
        skew = skew.shift(offset)

    # 处理填充值
    # 如果 kwargs 中包含 "fillna"，则使用指定值填充缺失值
    if "fillna" in kwargs:
        skew.fillna(kwargs["fillna"], inplace=True)
    # 如果 kwargs 中包含 "fill_method"，则使用指定的填充方法填充缺失值
    if "fill_method" in kwargs:
        skew.fillna(method=kwargs["fill_method"], inplace=True)

    # 名称和类别
    # 设置 skew 的名称为 "SKEW_长度"
    skew.name = f"SKEW_{length}"
    # 设置 skew 的类别为 "statistics"
    skew.category = "statistics"

    # 返回结果
    return skew


# 更新 skew 函数的文档字符串
skew.__doc__ = \
"""Rolling Skew

Sources:

Calculation:
    Default Inputs:
        length=30
    SKEW = close.rolling(length).skew()

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