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

# -*- coding: utf-8 -*-
# 从 numpy 导入 sqrt 函数，并将其命名为 npsqrt
from numpy import sqrt as npsqrt
# 从 variance 模块导入 variance 函数
from .variance import variance
# 从 pandas_ta 模块导入 Imports 类
from pandas_ta import Imports
# 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义 stdev 函数，用于计算标准差
def stdev(close, length=None, ddof=None, talib=None, offset=None, **kwargs):
    """Indicator: Standard Deviation"""
    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则设为 30
    length = int(length) if length and length > 0 else 30
    # 如果 ddof 是整数且大于等于 0 且小于 length，则将其转换为整数，否则设为 1
    ddof = int(ddof) if isinstance(ddof, int) and ddof >= 0 and ddof < length else 1
    # 验证 close 参数是否为有效的 Series，并根据 length 进行截断
    close = verify_series(close, length)
    # 获取 offset 参数
    offset = get_offset(offset)
    # 如果 talib 存在且为布尔值，则使用 talib 参数值，否则默认为 True
    mode_tal = bool(talib) if isinstance(talib, bool) else True
    # 如果 close 为 None，则返回
    if close is None: return
    # 计算结果
    # 如果 Imports 中有 "talib" 并且 mode_tal 为 True
    if Imports["talib"] and mode_tal:
        # 从 talib 中导入 STDDEV 函数，并计算标准差
        from talib import STDDEV
        stdev = STDDEV(close, length)
    else:
        # 否则使用自定义的 variance 函数计算方差，然后对结果应用平方根
        stdev = variance(close=close, length=length, ddof=ddof).apply(npsqrt)
    # 偏移结果
    if offset != 0:
        stdev = stdev.shift(offset)
    # 处理填充
    if "fillna" in kwargs:
        stdev.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        stdev.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置名称和类别
    stdev.name = f"STDEV_{length}"
    stdev.category = "statistics"
    # 返回结果
    return stdev
# 设置 stdev 函数的文档字符串
stdev.__doc__ = \
"""Rolling Standard Deviation
Sources:
Calculation:
    Default Inputs:
        length=30
    VAR = Variance
    STDEV = variance(close, length).apply(np.sqrt)
Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    ddof (int): Delta Degrees of Freedom.
                The divisor used in calculations is N - ddof,
                where N represents the number of elements. Default: 1
    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 -*-
# 从 numpy 库中导入 array 别名为 npArray
from numpy import array as npArray
# 从 numpy 库中导入 arange 别名为 npArange
from numpy import arange as npArange
# 从 numpy 库中导入 polyfit 别名为 npPolyfit
from numpy import polyfit as npPolyfit
# 从 numpy 库中导入 std 别名为 npStd
from numpy import std as npStd
# 从 pandas 库中导入 DataFrame、DatetimeIndex、Series
from pandas import DataFrame, DatetimeIndex, Series
# 从 .stdev 模块中导入 stdev 别名为 stdev
from .stdev import stdev as stdev
# 从 pandas_ta.utils 模块中导入 get_offset、verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义函数 tos_stdevall，计算 Think or Swim 标准偏差
def tos_stdevall(close, length=None, stds=None, ddof=None, offset=None, **kwargs):
    """Indicator: TD Ameritrade's Think or Swim Standard Deviation All"""
    # 验证参数
    # 如果 stds 是非空列表，则使用 stds，否则默认为 [1, 2, 3]
    stds = stds if isinstance(stds, list) and len(stds) > 0 else [1, 2, 3]
    # 如果 stds 中有小于等于 0 的数，则返回空
    if min(stds) <= 0: return
    # 如果 stds 中存在逆序排列，则将其反转为升序排列
    if not all(i < j for i, j in zip(stds, stds[1:])):
        stds = stds[::-1]
    # 将 ddof 转换为整数，确保在合理范围内，默认为 1
    ddof = int(ddof) if ddof and ddof >= 0 and ddof < length else 1
    # 获取偏移量
    offset = get_offset(offset)
    # 属性名称
    _props = f"TOS_STDEVALL"
    # 如果 length 为 None，则使用全部数据；否则，使用指定长度的数据
    if length is None:
        length = close.size
    else:
        # 将 length 转换为整数，确保大于 2，默认为 30
        length = int(length) if isinstance(length, int) and length > 2 else 30
        # 仅保留最近 length 个数据
        close = close.iloc[-length:]
        _props = f"{_props}_{length}"
    # 确保 close 是一个 Series，并且长度为 length
    close = verify_series(close, length)
    # 如果 close 为空，则返回空
    if close is None: return
    # 计算结果
    X = src_index = close.index
    # 如果 close 的索引是 DatetimeIndex 类型，则创建等差数组 X，并将 close 转换为数组
    if isinstance(close.index, DatetimeIndex):
        X = npArange(length)
        close = npArray(close)
    # 使用线性回归拟合得到斜率 m 和截距 b
    m, b = npPolyfit(X, close, 1)
    # 计算线性回归线 lr，索引与 close 保持一致
    lr = Series(m * X + b, index=src_index)
    # 计算标准差 stdev
    stdev = npStd(close, ddof=ddof)
    # 组装结果 DataFrame
    df = DataFrame({f"{_props}_LR": lr}, index=src_index)
    # 对于每个标准偏差值，计算上下界，并设置名称和分类
    for i in stds:
        df[f"{_props}_L_{i}"] = lr - i * stdev
        df[f"{_props}_U_{i}"] = lr + i * stdev
        df[f"{_props}_L_{i}"].name = df[f"{_props}_U_{i}"].name = f"{_props}"
        df[f"{_props}_L_{i}"].category = df[f"{_props}_U_{i}"].category = "statistics"
    # 对结果进行偏移
    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)
    # 准备返回的 DataFrame
    df.name = f"{_props}"
    df.category = "statistics"
    return df
# 设定函数的文档字符串
tos_stdevall.__doc__ = \
"""TD Ameritrade's Think or Swim Standard Deviation All (TOS_STDEV)
A port of TD Ameritrade's Think or Swim Standard Deviation All indicator which
returns the standard deviation of data for the entire plot or for the interval
of the last bars defined by the length parameter.
Sources:
    https://tlc.thinkorswim.com/center/reference/thinkScript/Functions/Statistical/StDevAll
Calculation:
    Default Inputs:
        length=None (All), stds=[1, 2, 3], ddof=1
    LR = Linear Regression
    STDEV = Standard Deviation
    LR = LR(close, length)
    STDEV = STDEV(close, length, ddof)
    for level in stds:
        LOWER = LR - level * STDEV
        UPPER = LR + level * STDEV
Args:
    close (pd.Series): Series of 'close's
    length (int): Bars from current bar. Default: None
"""
    stds (list): 存储标准偏差的列表，按照从中心线性回归线开始增加的顺序排列。默认值为 [1,2,3]
    ddof (int): Delta 自由度。在计算中使用的除数是 N - ddof，其中 N 表示元素的数量。默认值为 1
    offset (int): 结果的偏移周期数。默认值为 0
# 函数参数说明，接受关键字参数
Kwargs:
    # fillna 参数，用于填充缺失值的数值
    fillna (value, optional): pd.DataFrame.fillna(value)
    # fill_method 参数，指定填充方法的类型
    fill_method (value, optional): Type of fill method
# 返回值说明，返回一个 pandas DataFrame 对象
Returns:
    # 返回一个 pandas DataFrame 对象，包含中心 LR 和基于标准差倍数的上下 LR 线对
    pd.DataFrame: Central LR, Pairs of Lower and Upper LR Lines based on
        mulitples of the standard deviation. Default: returns 7 columns.

# 设置文件编码为 UTF-8
# -*- 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
def variance(close, length=None, ddof=None, talib=None, offset=None, **kwargs):
    """Indicator: Variance"""
    # 验证参数
    # 将长度转换为整数，如果长度存在且大于1，则为长度，否则为30
    length = int(length) if length and length > 1 else 30
    # 将 ddof 转换为整数，如果 ddof 是整数且大于等于0且小于长度，则为 ddof，否则为1
    ddof = int(ddof) if isinstance(ddof, int) and ddof >= 0 and ddof < length else 1
    # 如果kwargs中存在"min_periods"，则将其转换为整数，否则为长度
    min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
    # 验证close是否为有效Series，长度为最大值（长度，min_periods）
    close = verify_series(close, max(length, min_periods))
    # 获取偏移量
    offset = get_offset(offset)
    # 确定是否使用 talib
    mode_tal = bool(talib) if isinstance(talib, bool) else True
    # 如果close为None，则返回
    if close is None: return
    # 计算结果
    # 如果 pandas_ta 已导入 talib 并且 mode_tal 为真，则使用 talib 中的 VAR 函数计算方差
    if Imports["talib"] and mode_tal:
        from talib import VAR
        variance = VAR(close, length)
    # 否则，使用 rolling 方法计算滚动方差
    else:
        variance = close.rolling(length, min_periods=min_periods).var(ddof)
    # 偏移结果
    if offset != 0:
        variance = variance.shift(offset)
    # 处理填充
    # 如果kwargs中存在"fillna"，则填充NaN值
    if "fillna" in kwargs:
        variance.fillna(kwargs["fillna"], inplace=True)
    # 如果kwargs中存在"fill_method"，则使用指定的填充方法
    if "fill_method" in kwargs:
        variance.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置名称和类别
    variance.name = f"VAR_{length}"
    variance.category = "statistics"
    return variance
# 设置 variance 函数的文档字符串
variance.__doc__ = \
"""Rolling Variance
Sources:
Calculation:
    Default Inputs:
        length=30
    VARIANCE = close.rolling(length).var()
Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    ddof (int): Delta Degrees of Freedom.
                The divisor used in calculations is N - ddof,
                where N represents the number of elements. Default: 0
    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 -*- 
# 从pandas_ta.overlap模块中导入sma函数
from pandas_ta.overlap import sma
# 从本地的stdev模块中导入stdev函数
from .stdev import stdev
# 从pandas_ta.utils模块中导入get_offset和verify_series函数
from pandas_ta.utils import get_offset, verify_series
# 定义函数zscore，用于计算Z分数指标
def zscore(close, length=None, std=None, offset=None, **kwargs):
    """Indicator: Z Score"""
    # 验证参数
    # 将length转换为整数，如果length存在且大于1，则取其值，否则默认为30
    length = int(length) if length and length > 1 else 30
    # 将std转换为浮点数，如果std存在且大于1，则取其值，否则默认为1
    std = float(std) if std and std > 1 else 1
    # 验证close是否为有效的Series，长度为length
    close = verify_series(close, length)
    # 获取offset值
    offset = get_offset(offset)
    # 如果close为空，则返回空
    if close is None: return
    # 计算结果
    # 将std乘以stdev函数计算的标准差值
    std *= stdev(close=close, length=length, **kwargs)
    # 计算均值，使用sma函数计算移动平均值
    mean = sma(close=close, length=length, **kwargs)
    # 计算Z分数
    zscore = (close - mean) / std
    # 调整偏移量
    if offset != 0:
        zscore = zscore.shift(offset)
    # 处理填充
    if "fillna" in kwargs:
        zscore.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        zscore.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置指标名称和类别
    zscore.name = f"ZS_{length}"
    zscore.category = "statistics"
    return zscore
# 设置zscore函数的文档字符串
zscore.__doc__ = \
"""Rolling Z Score
Sources:
Calculation:
    Default Inputs:
        length=30, std=1
    SMA = Simple Moving Average
    STDEV = Standard Deviation
    std = std * STDEV(close, length)
    mean = SMA(close, length)
    ZSCORE = (close - mean) / std
Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 30
    std (float): It's period. Default: 1
    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.
"""

# 设置文件编码为 UTF-8，以支持中文等非 ASCII 字符
# 导入自定义模块中的函数，用于计算数据的不同统计量
from .entropy import entropy  # 导入 entropy 函数，用于计算数据的熵
from .kurtosis import kurtosis  # 导入 kurtosis 函数，用于计算数据的峰度
from .mad import mad  # 导入 mad 函数，用于计算数据的绝对中位差
from .median import median  # 导入 median 函数，用于计算数据的中位数
from .quantile import quantile  # 导入 quantile 函数，用于计算数据的分位数
from .skew import skew  # 导入 skew 函数，用于计算数据的偏度
from .stdev import stdev  # 导入 stdev 函数，用于计算数据的标准差
from .tos_stdevall import tos_stdevall  # 导入 tos_stdevall 函数，用于计算数据的时间序列标准差
from .variance import variance  # 导入 variance 函数，用于计算数据的方差
from .zscore import zscore  # 导入 zscore 函数，用于计算数据的 Z 分数