首页 > 编程语言 >PandasTA 源码解析(六)

PandasTA 源码解析(六)

时间:2024-04-15 13:45:13浏览次数:21  
标签:kc squeeze PandasTA length 源码 kwargs fillna 解析 True

.\pandas-ta\pandas_ta\momentum\rsi.py

# -*- coding: utf-8 -*-
# 导入所需模块和函数
from pandas import DataFrame, concat
from pandas_ta import Imports
from pandas_ta.overlap import rma
from pandas_ta.utils import get_drift, get_offset, verify_series, signals


def rsi(close, length=None, scalar=None, talib=None, drift=None, offset=None, **kwargs):
    """Indicator: Relative Strength Index (RSI)"""
    # 验证参数
    # 如果指定了长度且长度大于0,则将长度转换为整数;否则默认为14
    length = int(length) if length and length > 0 else 14
    # 如果指定了标量,则将其转换为浮点数;否则默认为100
    scalar = float(scalar) if scalar else 100
    # 确保收盘价是有效的数据序列,且长度符合要求
    close = verify_series(close, length)
    # 获取漂移参数,默认为1
    drift = get_drift(drift)
    # 获取偏移参数,默认为0
    offset = get_offset(offset)
    # 如果指定了 talib 参数且为 True,则使用 talib 模式;否则默认为 True
    mode_tal = bool(talib) if isinstance(talib, bool) else True

    if close is None: return

    # 计算结果
    if Imports["talib"] and mode_tal:
        # 如果启用了 talib 并且处于 talib 模式,则使用 talib 库计算 RSI
        from talib import RSI
        rsi = RSI(close, length)
    else:
        # 否则,按照公式手动计算 RSI
        # 计算价格变化的正值和负值
        negative = close.diff(drift)
        positive = negative.copy()

        # 将正值序列中小于0的值置为0,以保证正值序列只包含正的价格变化
        positive[positive < 0] = 0  
        # 将负值序列中大于0的值置为0,以保证负值序列只包含负的价格变化
        negative[negative > 0] = 0  

        # 计算正值序列和负值序列的移动平均值
        positive_avg = rma(positive, length=length)
        negative_avg = rma(negative, length=length)

        # 计算 RSI
        rsi = scalar * positive_avg / (positive_avg + negative_avg.abs())

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

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

    # 设置名称和分类
    rsi.name = f"RSI_{length}"
    rsi.category = "momentum"

    # 如果指定了信号指示器参数为 True,则返回包含信号指示器的数据框;否则返回 RSI 序列
    signal_indicators = kwargs.pop("signal_indicators", False)
    if signal_indicators:
        # 如果启用了信号指示器,则生成包含信号指示器的数据框
        signalsdf = concat(
            [
                DataFrame({rsi.name: rsi}),
                signals(
                    indicator=rsi,
                    xa=kwargs.pop("xa", 80),
                    xb=kwargs.pop("xb", 20),
                    xserie=kwargs.pop("xserie", None),
                    xserie_a=kwargs.pop("xserie_a", None),
                    xserie_b=kwargs.pop("xserie_b", None),
                    cross_values=kwargs.pop("cross_values", False),
                    cross_series=kwargs.pop("cross_series", True),
                    offset=offset,
                ),
            ],
            axis=1,
        )

        return signalsdf
    else:
        # 否则,返回 RSI 序列
        return rsi


# 设置 RSI 函数的文档字符串
rsi.__doc__ = \
"""Relative Strength Index (RSI)

The Relative Strength Index is popular momentum oscillator used to measure the
velocity as well as the magnitude of directional price movements.

Sources:
    https://www.tradingview.com/wiki/Relative_Strength_Index_(RSI)

Calculation:
    Default Inputs:
        length=14, scalar=100, drift=1
    ABS = Absolute Value
    RMA = Rolling Moving Average

    diff = close.diff(drift)
    positive = diff if diff > 0 else 0
    negative = diff if diff < 0 else 0

    pos_avg = RMA(positive, length)
    neg_avg = ABS(RMA(negative, length))

    RSI = scalar * pos_avg / (pos_avg + neg_avg)

Args:

"""
    # close (pd.Series): 'close' 是一个 Pandas Series 对象,存储了收盘价数据
    # length (int): 计算指标所用的周期长度,默认为 14
    # scalar (float): 放大倍数,默认为 100
    # talib (bool): 如果 TA Lib 被安装并且 talib 参数为 True,则返回 TA Lib 的版本信息,默认为 True
    # drift (int): 差分的周期长度,默认为 1
    # offset (int): 结果向后偏移的周期数,默认为 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\momentum\rsx.py

# 设置文件编码为 UTF-8
# -*- coding: utf-8 -*-

# 导入 numpy 库中的 nan 常量,并将其重命名为 npNaN
from numpy import nan as npNaN

# 从 pandas 库中导入 concat、DataFrame 和 Series 类
from pandas import concat, DataFrame, Series

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

# 定义 RSX 指标函数,计算相对强度 Xtra(灵感来自于 Jurik RSX)
def rsx(close, length=None, drift=None, offset=None, **kwargs):
    # 验证参数
    # 如果 length 存在且大于 0,则转换为整数,否则设为默认值 14
    length = int(length) if length and length > 0 else 14
    # 验证收盘价序列,确保其长度为 length
    close = verify_series(close, length)
    # 获取漂移值
    drift = get_drift(drift)
    # 获取偏移值
    offset = get_offset(offset)

    # 如果收盘价序列为空,则返回空
    if close is None: return

    # 初始化变量
    vC, v1C = 0, 0
    v4, v8, v10, v14, v18, v20 = 0, 0, 0, 0, 0, 0
    f0, f8, f10, f18, f20, f28, f30, f38 = 0, 0, 0, 0, 0, 0, 0, 0
    f40, f48, f50, f58, f60, f68, f70, f78 = 0, 0, 0, 0, 0, 0, 0, 0
    f80, f88, f90 = 0, 0, 0

    # 计算结果
    m = close.size
    result = [npNaN for _ in range(0, length - 1)] + [0]
    for i in range(length, m):
        if f90 == 0:
            f90 = 1.0
            f0 = 0.0
            if length - 1.0 >= 5:
                f88 = length - 1.0
            else:
                f88 = 5.0
            f8 = 100.0 * close.iloc[i]
            f18 = 3.0 / (length + 2.0)
            f20 = 1.0 - f18
        else:
            if f88 <= f90:
                f90 = f88 + 1
            else:
                f90 = f90 + 1
            f10 = f8
            f8 = 100 * close.iloc[i]
            v8 = f8 - f10
            f28 = f20 * f28 + f18 * v8
            f30 = f18 * f28 + f20 * f30
            vC = 1.5 * f28 - 0.5 * f30
            f38 = f20 * f38 + f18 * vC
            f40 = f18 * f38 + f20 * f40
            v10 = 1.5 * f38 - 0.5 * f40
            f48 = f20 * f48 + f18 * v10
            f50 = f18 * f48 + f20 * f50
            v14 = 1.5 * f48 - 0.5 * f50
            f58 = f20 * f58 + f18 * abs(v8)
            f60 = f18 * f58 + f20 * f60
            v18 = 1.5 * f58 - 0.5 * f60
            f68 = f20 * f68 + f18 * v18
            f70 = f18 * f68 + f20 * f70
            v1C = 1.5 * f68 - 0.5 * f70
            f78 = f20 * f78 + f18 * v1C
            f80 = f18 * f78 + f20 * f80
            v20 = 1.5 * f78 - 0.5 * f80

            if f88 >= f90 and f8 != f10:
                f0 = 1.0
            if f88 == f90 and f0 == 0.0:
                f90 = 0.0

        if f88 < f90 and v20 > 0.0000000001:
            v4 = (v14 / v20 + 1.0) * 50.0
            if v4 > 100.0:
                v4 = 100.0
            if v4 < 0.0:
                v4 = 0.0
        else:
            v4 = 50.0
        result.append(v4)

    # 将结果转换为 Series,索引与收盘价相同
    rsx = Series(result, index=close.index)

    # 如果偏移不为 0,则将结果向前偏移指定数量的周期
    if offset != 0:
        rsx = rsx.shift(offset)

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

    # 设置指标名称和分类
    rsx.name = f"RSX_{length}"
    rsx.category = "momentum"

    # 推导出信号指示器
    signal_indicators = kwargs.pop("signal_indicators", False)
    # 如果信号指标存在
    if signal_indicators:
        # 将RSX指标转换为DataFrame格式,并与信号指标合并
        signalsdf = concat(
            [
                DataFrame({rsx.name: rsx}),
                # 调用signals函数生成信号指标
                signals(
                    indicator=rsx,
                    xa=kwargs.pop("xa", 80),
                    xb=kwargs.pop("xb", 20),
                    xserie=kwargs.pop("xserie", None),
                    xserie_a=kwargs.pop("xserie_a", None),
                    xserie_b=kwargs.pop("xserie_b", None),
                    cross_values=kwargs.pop("cross_values", False),
                    cross_series=kwargs.pop("cross_series", True),
                    offset=offset,
                ),
            ],
            axis=1
        )
        # 返回合并后的DataFrame
        return signalsdf
    else:
        # 如果信号指标不存在,直接返回RSX指标
        return rsx
# 设置 rsx 函数的文档字符串,说明了该函数的作用、来源、计算方法、参数和返回值
rsx.__doc__ = \
"""Relative Strength Xtra (rsx)

The Relative Strength Xtra is based on the popular RSI indicator and inspired
by the work Jurik Research. The code implemented is based on published code
found at 'prorealcode.com'. This enhanced version of the rsi reduces noise and
provides a clearer, only slightly delayed insight on momentum and velocity of
price movements.

Sources:
    http://www.jurikres.com/catalog1/ms_rsx.htm
    https://www.prorealcode.com/prorealtime-indicators/jurik-rsx/

Calculation:
    Refer to the sources above for information as well as code example.

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

.\pandas-ta\pandas_ta\momentum\rvgi.py

# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.overlap 模块中导入 swma 函数
from pandas_ta.overlap import swma
# 从 pandas_ta.utils 模块中导入 get_offset、non_zero_range、verify_series 函数
from pandas_ta.utils import get_offset, non_zero_range, verify_series


# 定义 RVGI 函数,计算相对活力指数
def rvgi(open_, high, low, close, length=None, swma_length=None, offset=None, **kwargs):
    """Indicator: Relative Vigor Index (RVGI)"""
    # 验证参数
    # 计算 high 和 low 之间的范围,不为零
    high_low_range = non_zero_range(high, low)
    # 计算 close 和 open 之间的范围,不为零
    close_open_range = non_zero_range(close, open_)
    # 将 length 转换为整数,如果 length 不存在或小于等于 0,则默认为 14
    length = int(length) if length and length > 0 else 14
    # 将 swma_length 转换为整数,如果 swma_length 不存在或小于等于 0,则默认为 4
    swma_length = int(swma_length) if swma_length and swma_length > 0 else 4
    # 计算最大长度,取 length 和 swma_length 中的最大值
    _length = max(length, swma_length)
    # 验证 open_、high、low、close,使其长度为 _length
    open_ = verify_series(open_, _length)
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    close = verify_series(close, _length)
    # 获取偏移量
    offset = get_offset(offset)

    # 如果 open_、high、low、close 中有任何一个为 None,则返回
    if open_ is None or high is None or low is None or close is None: return

    # 计算结果
    # 计算分子,为 close_open_range 的 swma,长度为 swma_length 的滚动和
    numerator = swma(close_open_range, length=swma_length).rolling(length).sum()
    # 计算分母,为 high_low_range 的 swma,长度为 swma_length 的滚动和
    denominator = swma(high_low_range, length=swma_length).rolling(length).sum()

    # 计算 RVGI,为分子除以分母
    rvgi = numerator / denominator
    # 计算信号线,为 RVGI 的 swma,长度为 swma_length
    signal = swma(rvgi, length=swma_length)

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

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

    # 名称和类别
    # 设置 RVGI 的名称为 "RVGI_length_swma_length",设置 signal 的名称为 "RVGIs_length_swma_length"
    rvgi.name = f"RVGI_{length}_{swma_length}"
    signal.name = f"RVGIs_{length}_{swma_length}"
    # 设置 RVGI 和 signal 的类别为 "momentum"
    rvgi.category = signal.category = "momentum"

    # 准备返回的 DataFrame
    # 创建 DataFrame,包含 RVGI 和 signal,列名为其名称
    df = DataFrame({rvgi.name: rvgi, signal.name: signal})
    # 设置 DataFrame 的名称为 "RVGI_length_swma_length"
    df.name = f"RVGI_{length}_{swma_length}"
    # 设置 DataFrame 的类别为 RVGI 的类别
    df.category = rvgi.category

    return df


# 设置 RVGI 函数的文档字符串
rvgi.__doc__ = \
"""Relative Vigor Index (RVGI)

The Relative Vigor Index attempts to measure the strength of a trend relative to
its closing price to its trading range.  It is based on the belief that it tends
to close higher than they open in uptrends or close lower than they open in
downtrends.

Sources:
    https://www.investopedia.com/terms/r/relative_vigor_index.asp

Calculation:
    Default Inputs:
        length=14, swma_length=4
    SWMA = Symmetrically Weighted Moving Average
    numerator = SUM(SWMA(close - open, swma_length), length)
    denominator = SUM(SWMA(high - low, swma_length), length)
    RVGI = numerator / denominator

Args:
    open_ (pd.Series): Series of 'open's
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 14
    swma_length (int): It's period. Default: 4
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    fillna (value, optional): pd.DataFrame.fillna(value)
"""
    # fill_method 是一个可选参数,用于指定填充方法的类型
# 返回一个 Pandas Series 对象,表示生成的新特征
Returns:
    pd.Series: New feature generated.

.\pandas-ta\pandas_ta\momentum\slope.py

# -*- coding: utf-8 -*-  # 指定文件编码格式为 UTF-8

# 从 numpy 库中导入 arctan 函数并起别名为 npAtan
from numpy import arctan as npAtan
# 从 numpy 库中导入 pi 常量并起别名为 npPi
from numpy import pi as npPi
# 从 pandas_ta 库中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series

# 定义函数 slope,用于计算数据序列的斜率
def slope(close, length=None, as_angle=None, to_degrees=None, vertical=None, offset=None, **kwargs):
    """Indicator: Slope"""  # 函数的说明文档字符串

    # 验证参数
    length = int(length) if length and length > 0 else 1  # 如果 length 存在且大于 0,则转换为整数,否则设为默认值 1
    as_angle = True if isinstance(as_angle, bool) else False  # 如果 as_angle 是布尔值,则设为 True,否则设为 False
    to_degrees = True if isinstance(to_degrees, bool) else False  # 如果 to_degrees 是布尔值,则设为 True,否则设为 False
    close = verify_series(close, length)  # 验证数据序列,并确保长度为 length
    offset = get_offset(offset)  # 获取偏移量

    if close is None: return  # 如果数据序列为空,则返回空值

    # 计算结果
    slope = close.diff(length) / length  # 计算斜率
    if as_angle:  # 如果需要将斜率转换为角度
        slope = slope.apply(npAtan)  # 将斜率应用 arctan 函数
        if to_degrees:  # 如果需要将角度转换为度
            slope *= 180 / npPi  # 将角度乘以 180/π

    # 偏移
    if offset != 0:  # 如果偏移量不为零
        slope = slope.shift(offset)  # 对结果斜率进行偏移操作

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

    # 给结果命名和分类
    slope.name = f"SLOPE_{length}" if not as_angle else f"ANGLE{'d' if to_degrees else 'r'}_{length}"  # 根据参数设定名称
    slope.category = "momentum"  # 将结果分类为动量指标

    return slope  # 返回结果斜率

# 设置函数 slope 的说明文档字符串
slope.__doc__ = \
"""Slope

Returns the slope of a series of length n. Can convert the slope to angle.
Default: slope.

Sources: Algebra I

Calculation:
    Default Inputs:
        length=1
    slope = close.diff(length) / length

    if as_angle:
        slope = slope.apply(atan)
        if to_degrees:
            slope *= 180 / PI

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period.  Default: 1
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    as_angle (value, optional): Converts slope to an angle. Default: False
    to_degrees (value, optional): Converts slope angle to degrees. Default: False
    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\momentum\smi.py

# -*- coding: utf-8 -*-                             # 指定编码格式为 UTF-8

# 导入必要的模块和函数
from pandas import DataFrame                       
from .tsi import tsi                                 # 从当前目录下的 tsi 模块中导入 tsi 函数
from pandas_ta.overlap import ema                    # 从 pandas_ta 库中的 overlap 模块导入 ema 函数
from pandas_ta.utils import get_offset, verify_series  # 从 pandas_ta 库中的 utils 模块导入 get_offset 和 verify_series 函数


def smi(close, fast=None, slow=None, signal=None, scalar=None, offset=None, **kwargs):
    """Indicator: SMI Ergodic Indicator (SMIIO)"""   # 定义函数 smi,计算 SMI 指标
    # 验证参数
    fast = int(fast) if fast and fast > 0 else 5     # 如果 fast 存在且大于 0,则将其转换为整数类型,否则设为默认值 5
    slow = int(slow) if slow and slow > 0 else 20    # 如果 slow 存在且大于 0,则将其转换为整数类型,否则设为默认值 20
    signal = int(signal) if signal and signal > 0 else 5  # 如果 signal 存在且大于 0,则将其转换为整数类型,否则设为默认值 5
    if slow < fast:                                  # 如果 slow 小于 fast,则交换两者的值
        fast, slow = slow, fast
    scalar = float(scalar) if scalar else 1          # 如果 scalar 存在,则转换为浮点数,否则设为默认值 1
    close = verify_series(close, max(fast, slow, signal))  # 验证 close 数据,保证长度足够
    offset = get_offset(offset)                      # 获取偏移量

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

    # 计算结果
    tsi_df = tsi(close, fast=fast, slow=slow, signal=signal, scalar=scalar)  # 调用 tsi 函数计算 TSI 指标
    smi = tsi_df.iloc[:, 0]                         # 获取 TSI 列
    signalma = tsi_df.iloc[:, 1]                    # 获取信号线列
    osc = smi - signalma                            # 计算 SMI 指标的震荡值

    # 偏移
    if offset != 0:                                 # 如果偏移量不为 0
        smi = smi.shift(offset)                     # 对 SMI 进行偏移
        signalma = signalma.shift(offset)           # 对信号线进行偏移
        osc = osc.shift(offset)                     # 对震荡值进行偏移

    # 填充缺失值
    if "fillna" in kwargs:                          # 如果参数中包含 "fillna"
        smi.fillna(kwargs["fillna"], inplace=True)  # 使用指定的填充值填充缺失值
        signalma.fillna(kwargs["fillna"], inplace=True)
        osc.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:                     # 如果参数中包含 "fill_method"
        smi.fillna(method=kwargs["fill_method"], inplace=True)  # 使用指定的填充方法填充缺失值
        signalma.fillna(method=kwargs["fill_method"], inplace=True)
        osc.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置指标名称和分类
    _scalar = f"_{scalar}" if scalar != 1 else ""   # 如果 scalar 不等于 1,则添加到名称中
    _props = f"_{fast}_{slow}_{signal}{_scalar}"   # 构建指标属性字符串
    smi.name = f"SMI{_props}"                       # 设置 SMI 指标名称
    signalma.name = f"SMIs{_props}"                 # 设置信号线指标名称
    osc.name = f"SMIo{_props}"                      # 设置震荡值指标名称
    smi.category = signalma.category = osc.category = "momentum"  # 设置指标分类为动量型

    # 准备要返回的 DataFrame
    data = {smi.name: smi, signalma.name: signalma, osc.name: osc}  # 构建包含指标数据的字典
    df = DataFrame(data)                            # 构建 DataFrame
    df.name = f"SMI{_props}"                        # 设置 DataFrame 名称
    df.category = smi.category                      # 设置 DataFrame 分类与指标相同

    return df                                       # 返回 DataFrame


smi.__doc__ = \
"""SMI Ergodic Indicator (SMI)

The SMI Ergodic Indicator is the same as the True Strength Index (TSI) developed
by William Blau, except the SMI includes a signal line. The SMI uses double
moving averages of price minus previous price over 2 time frames. The signal
line, which is an EMA of the SMI, is plotted to help trigger trading signals.
The trend is bullish when crossing above zero and bearish when crossing below
zero. This implementation includes both the SMI Ergodic Indicator and SMI
Ergodic Oscillator.

Sources:
    https://www.motivewave.com/studies/smi_ergodic_indicator.htm
    https://www.tradingview.com/script/Xh5Q0une-SMI-Ergodic-Oscillator/
    https://www.tradingview.com/script/cwrgy4fw-SMIIO/

Calculation:
    Default Inputs:
        fast=5, slow=20, signal=5
    TSI = True Strength Index
    EMA = Exponential Moving Average

    ERG = TSI(close, fast, slow)
    Signal = EMA(ERG, signal)
    OSC = ERG - Signal

Args:
    close (pd.Series): Series of 'close's

"""                                                 # 设置 smi 函数的文档字符串
    # 快速线的周期,默认为5
    fast (int): The short period. Default: 5
    # 慢速线的周期,默认为20
    slow (int): The long period. Default: 20
    # 信号线的周期,默认为5
    signal (int): The signal period. Default: 5
    # 放大倍数,默认为1
    scalar (float): How much to magnify. Default: 1
    # 结果的偏移周期数,默认为0
    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: smi, signal, oscillator columns.

.\pandas-ta\pandas_ta\momentum\squeeze.py

# -*- coding: utf-8 -*-  # 指定文件编码格式为 UTF-8
from numpy import nan as npNaN  # 导入 numpy 库中的 nan 并起别名为 npNaN
from pandas import DataFrame  # 导入 pandas 库中的 DataFrame 类
from pandas_ta.momentum import mom  # 导入 pandas_ta 库中的 momentum 模块,并导入其中的 mom 函数
from pandas_ta.overlap import ema, linreg, sma  # 导入 pandas_ta 库中的 overlap 模块,并导入其中的 ema、linreg、sma 函数
from pandas_ta.trend import decreasing, increasing  # 导入 pandas_ta 库中的 trend 模块,并导入其中的 decreasing、increasing 函数
from pandas_ta.volatility import bbands, kc  # 导入 pandas_ta 库中的 volatility 模块,并导入其中的 bbands、kc 函数
from pandas_ta.utils import get_offset  # 导入 pandas_ta 库中的 utils 模块,并导入其中的 get_offset 函数
from pandas_ta.utils import unsigned_differences, verify_series  # 导入 pandas_ta 库中的 utils 模块,并导入其中的 unsigned_differences、verify_series 函数

# 定义函数:Squeeze Momentum (SQZ)
def squeeze(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_scalar=None, mom_length=None, mom_smooth=None, use_tr=None, mamode=None, offset=None, **kwargs):
    """Indicator: Squeeze Momentum (SQZ)"""  # 函数文档字符串,指示 SQZ 指标的作用
    # 验证参数
    bb_length = int(bb_length) if bb_length and bb_length > 0 else 20  # 如果 bb_length 存在且大于 0,则转换为整数,否则设为默认值 20
    bb_std = float(bb_std) if bb_std and bb_std > 0 else 2.0  # 如果 bb_std 存在且大于 0,则转换为浮点数,否则设为默认值 2.0
    kc_length = int(kc_length) if kc_length and kc_length > 0 else 20  # 如果 kc_length 存在且大于 0,则转换为整数,否则设为默认值 20
    kc_scalar = float(kc_scalar) if kc_scalar and kc_scalar > 0 else 1.5  # 如果 kc_scalar 存在且大于 0,则转换为浮点数,否则设为默认值 1.5
    mom_length = int(mom_length) if mom_length and mom_length > 0 else 12  # 如果 mom_length 存在且大于 0,则转换为整数,否则设为默认值 12
    mom_smooth = int(mom_smooth) if mom_smooth and mom_smooth > 0 else 6  # 如果 mom_smooth 存在且大于 0,则转换为整数,否则设为默认值 6
    _length = max(bb_length, kc_length, mom_length, mom_smooth)  # 计算参数的最大长度
    high = verify_series(high, _length)  # 验证 high 序列的长度
    low = verify_series(low, _length)  # 验证 low 序列的长度
    close = verify_series(close, _length)  # 验证 close 序列的长度
    offset = get_offset(offset)  # 获取偏移量

    if high is None or low is None or close is None: return  # 如果 high、low、close 中有任何一个为 None,则返回

    use_tr = kwargs.setdefault("tr", True)  # 设置参数 tr,默认为 True
    asint = kwargs.pop("asint", True)  # 弹出参数 asint,默认为 True
    detailed = kwargs.pop("detailed", False)  # 弹出参数 detailed,默认为 False
    lazybear = kwargs.pop("lazybear", False)  # 弹出参数 lazybear,默认为 False
    mamode = mamode if isinstance(mamode, str) else "sma"  # 如果 mamode 是字符串类型,则保持不变,否则设为默认值 "sma"

    # 定义函数:简化列名
    def simplify_columns(df, n=3):
        df.columns = df.columns.str.lower()  # 将列名转换为小写
        return [c.split("_")[0][n - 1:n] for c in df.columns]  # 返回简化后的列名列表

    # 计算结果
    bbd = bbands(close, length=bb_length, std=bb_std, mamode=mamode)  # 计算布林带指标
    kch = kc(high, low, close, length=kc_length, scalar=kc_scalar, mamode=mamode, tr=use_tr)  # 计算 Keltner 通道指标

    # 简化 KC 和 BBAND 列名以便动态访问
    bbd.columns = simplify_columns(bbd)  # 简化布林带指标列名
    kch.columns = simplify_columns(kch)  # 简化 Keltner 通道指标列名

    if lazybear:  # 如果 lazybear 参数为真
        highest_high = high.rolling(kc_length).max()  # 计算最高高度
        lowest_low = low.rolling(kc_length).min()  # 计算最低低度
        avg_ = 0.25 * (highest_high + lowest_low) + 0.5 * kch.b  # 计算平均值

        squeeze = linreg(close - avg_, length=kc_length)  # 计算线性回归

    else:  # 如果 lazybear 参数为假
        momo = mom(close, length=mom_length)  # 计算动量
        if mamode.lower() == "ema":  # 如果 mamode 参数为 "ema"
            squeeze = ema(momo, length=mom_smooth)  # 计算指数移动平均
        else:  # 否则(mamode 参数为 "sma")
            squeeze = sma(momo, length=mom_smooth)  # 计算简单移动平均

    # 分类 Squeeze
    squeeze_on = (bbd.l > kch.l) & (bbd.u < kch.u)  # 计算 Squeeze on
    squeeze_off = (bbd.l < kch.l) & (bbd.u > kch.u) 
    # 如果参数中包含 "fillna",则使用指定值填充缺失值
    if "fillna" in kwargs:
        # 使用指定值填充缺失值,inplace=True 表示在原数据上进行修改
        squeeze.fillna(kwargs["fillna"], inplace=True)
        squeeze_on.fillna(kwargs["fillna"], inplace=True)
        squeeze_off.fillna(kwargs["fillna"], inplace=True)
        no_squeeze.fillna(kwargs["fillna"], inplace=True)
    
    # 如果参数中包含 "fill_method",则使用指定方法填充缺失值
    if "fill_method" in kwargs:
        # 使用指定方法填充缺失值,inplace=True 表示在原数据上进行修改
        squeeze.fillna(method=kwargs["fill_method"], inplace=True)
        squeeze_on.fillna(method=kwargs["fill_method"], inplace=True)
        squeeze_off.fillna(method=kwargs["fill_method"], inplace=True)
        no_squeeze.fillna(method=kwargs["fill_method"], inplace=True)

    # 设置名称和分类
    _props = "" if use_tr else "hlr"
    _props += f"_{bb_length}_{bb_std}_{kc_length}_{kc_scalar}"
    _props += "_LB" if lazybear else ""
    squeeze.name = f"SQZ{_props}"

    # 创建数据字典
    data = {
        squeeze.name: squeeze,
        f"SQZ_ON": squeeze_on.astype(int) if asint else squeeze_on,
        f"SQZ_OFF": squeeze_off.astype(int) if asint else squeeze_off,
        f"SQZ_NO": no_squeeze.astype(int) if asint else no_squeeze,
    }
    # 创建 DataFrame 对象
    df = DataFrame(data)
    df.name = squeeze.name
    df.category = squeeze.category = "momentum"

    # 如果需要详细信息
    if detailed:
        # 分别获取正数和负数的数据
        pos_squeeze = squeeze[squeeze >= 0]
        neg_squeeze = squeeze[squeeze < 0]

        # 计算正数和负数的增量和减量
        pos_inc, pos_dec = unsigned_differences(pos_squeeze, asint=True)
        neg_inc, neg_dec = unsigned_differences(neg_squeeze, asint=True)

        # 计算正数和负数的增量和减量乘以原数据
        pos_inc *= squeeze
        pos_dec *= squeeze
        neg_dec *= squeeze
        neg_inc *= squeeze

        # 将值为 0 的数据替换为 NaN
        pos_inc.replace(0, np.NaN, inplace=True)
        pos_dec.replace(0, np.NaN, inplace=True)
        neg_dec.replace(0, np.NaN, inplace=True)
        neg_inc.replace(0, np.NaN, inplace=True)

        # 计算正数和负数的增量和减量乘以原数据
        sqz_inc = squeeze * increasing(squeeze)
        sqz_dec = squeeze * decreasing(squeeze)
        sqz_inc.replace(0, np.NaN, inplace=True)
        sqz_dec.replace(0, np.NaN, inplace=True)

        # 处理填充值
        if "fillna" in kwargs:
            sqz_inc.fillna(kwargs["fillna"], inplace=True)
            sqz_dec.fillna(kwargs["fillna"], inplace=True)
            pos_inc.fillna(kwargs["fillna"], inplace=True)
            pos_dec.fillna(kwargs["fillna"], inplace=True)
            neg_dec.fillna(kwargs["fillna"], inplace=True)
            neg_inc.fillna(kwargs["fillna"], inplace=True)
        if "fill_method" in kwargs:
            sqz_inc.fillna(method=kwargs["fill_method"], inplace=True)
            sqz_dec.fillna(method=kwargs["fill_method"], inplace=True)
            pos_inc.fillna(method=kwargs["fill_method"], inplace=True)
            pos_dec.fillna(method=kwargs["fill_method"], inplace=True)
            neg_dec.fillna(method=kwargs["fill_method"], inplace=True)
            neg_inc.fillna(method=kwargs["fill_method"], inplace=True)

        # 添加详细信息列
        df[f"SQZ_INC"] = sqz_inc
        df[f"SQZ_DEC"] = sqz_dec
        df[f"SQZ_PINC"] = pos_inc
        df[f"SQZ_PDEC"] = pos_dec
        df[f"SQZ_NDEC"] = neg_dec
        df[f"SQZ_NINC"] = neg_inc

    # 返回 DataFrame 对象
    return df
# 设置 squeeze 函数的文档字符串,解释了 squeeze 指标的作用和计算方法
squeeze.__doc__ = \
"""Squeeze (SQZ)

The default is based on John Carter's "TTM Squeeze" indicator, as discussed
in his book "Mastering the Trade" (chapter 11). The Squeeze indicator attempts
to capture the relationship between two studies: Bollinger Bands® and Keltner's
Channels. When the volatility increases, so does the distance between the bands,
conversely, when the volatility declines, the distance also decreases. It finds
sections of the Bollinger Bands® study which fall inside the Keltner's Channels.

Sources:
    https://tradestation.tradingappstore.com/products/TTMSqueeze
    https://www.tradingview.com/scripts/lazybear/
    https://tlc.thinkorswim.com/center/reference/Tech-Indicators/studies-library/T-U/TTM-Squeeze

Calculation:
    Default Inputs:
        bb_length=20, bb_std=2, kc_length=20, kc_scalar=1.5, mom_length=12,
        mom_smooth=12, tr=True, lazybear=False,
    BB = Bollinger Bands
    KC = Keltner Channels
    MOM = Momentum
    SMA = Simple Moving Average
    EMA = Exponential Moving Average
    TR = True Range

    RANGE = TR(high, low, close) if using_tr else high - low
    BB_LOW, BB_MID, BB_HIGH = BB(close, bb_length, std=bb_std)
    KC_LOW, KC_MID, KC_HIGH = KC(high, low, close, kc_length, kc_scalar, TR)

    if lazybear:
        HH = high.rolling(kc_length).max()
        LL = low.rolling(kc_length).min()
        AVG  = 0.25 * (HH + LL) + 0.5 * KC_MID
        SQZ = linreg(close - AVG, kc_length)
    else:
        MOMO = MOM(close, mom_length)
        if mamode == "ema":
            SQZ = EMA(MOMO, mom_smooth)
        else:
            SQZ = EMA(momo, mom_smooth)

    SQZ_ON  = (BB_LOW > KC_LOW) and (BB_HIGH < KC_HIGH)
    SQZ_OFF = (BB_LOW < KC_LOW) and (BB_HIGH > KC_HIGH)
    NO_SQZ = !SQZ_ON and !SQZ_OFF

Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    bb_length (int): Bollinger Bands period. Default: 20
    bb_std (float): Bollinger Bands Std. Dev. Default: 2
    kc_length (int): Keltner Channel period. Default: 20
    kc_scalar (float): Keltner Channel scalar. Default: 1.5
    mom_length (int): Momentum Period. Default: 12
    mom_smooth (int): Smoothing Period of Momentum. Default: 6
    mamode (str): Only "ema" or "sma". Default: "sma"
    offset (int): How many periods to offset the result. Default: 0

Kwargs:
    tr (value, optional): Use True Range for Keltner Channels. Default: True
    asint (value, optional): Use integers instead of bool. Default: True
    mamode (value, optional): Which MA to use. Default: "sma"
    lazybear (value, optional): Use LazyBear's TradingView implementation.
        Default: False
    detailed (value, optional): Return additional variations of SQZ for
        visualization. Default: False
    fillna (value, optional): pd.DataFrame.fillna(value)
    fill_method (value, optional): Type of fill method

Returns:
"""
    # 创建一个 pandas DataFrame,默认包含 SQZ、SQZ_ON、SQZ_OFF、NO_SQZ 列。如果 'detailed' 参数为 True,则包含更详细的列。
"""
# 导入必要的模块:字符串 I/O 以及 zip 文件处理
from io import BytesIO
import zipfile

# 定义函数 unzip_data,接受文件名参数
def unzip_data(zip_file):
    # 打开 zip 文件并读取其内容
    with zipfile.ZipFile(zip_file, 'r') as zip_ref:
        # 解压缩文件到指定路径
        zip_ref.extractall('unzipped_data')

.\pandas-ta\pandas_ta\momentum\squeeze_pro.py

# -*- coding: utf-8 -*-
# 导入所需模块和对象
from numpy import NaN as npNaN
from pandas import DataFrame
from pandas_ta.momentum import mom
from pandas_ta.overlap import ema, sma
from pandas_ta.trend import decreasing, increasing
from pandas_ta.volatility import bbands, kc
from pandas_ta.utils import get_offset
from pandas_ta.utils import unsigned_differences, verify_series

# 定义函数:Squeeze Momentum (SQZ) PRO,计算挤压动量指标
def squeeze_pro(high, low, close, bb_length=None, bb_std=None, kc_length=None, kc_scalar_wide=None, kc_scalar_normal=None, kc_scalar_narrow=None, mom_length=None, mom_smooth=None, use_tr=None, mamode=None, offset=None, **kwargs):
    """Indicator: Squeeze Momentum (SQZ) PRO"""
    # 验证参数有效性,并设置默认值
    bb_length = int(bb_length) if bb_length and bb_length > 0 else 20
    bb_std = float(bb_std) if bb_std and bb_std > 0 else 2.0
    kc_length = int(kc_length) if kc_length and kc_length > 0 else 20
    kc_scalar_wide = float(kc_scalar_wide) if kc_scalar_wide and kc_scalar_wide > 0 else 2
    kc_scalar_normal = float(kc_scalar_normal) if kc_scalar_normal and kc_scalar_normal > 0 else 1.5
    kc_scalar_narrow = float(kc_scalar_narrow) if kc_scalar_narrow and kc_scalar_narrow > 0 else 1
    mom_length = int(mom_length) if mom_length and mom_length > 0 else 12
    mom_smooth = int(mom_smooth) if mom_smooth and mom_smooth > 0 else 6

    _length = max(bb_length, kc_length, mom_length, mom_smooth)
    # 确保输入数据的有效性
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    close = verify_series(close, _length)
    offset = get_offset(offset)

    # 验证 KC 系数的有效性
    valid_kc_scaler = kc_scalar_wide > kc_scalar_normal and kc_scalar_normal > kc_scalar_narrow
    if not valid_kc_scaler: return
    if high is None or low is None or close is None: return

    # 获取参数中的 True Range 数据,默认为真
    use_tr = kwargs.setdefault("tr", True)
    asint = kwargs.pop("asint", True)
    detailed = kwargs.pop("detailed", False)
    mamode = mamode if isinstance(mamode, str) else "sma"

    # 函数:简化数据框列名
    def simplify_columns(df, n=3):
        df.columns = df.columns.str.lower()
        return [c.split("_")[0][n - 1:n] for c in df.columns]

    # 计算结果
    bbd = bbands(close, length=bb_length, std=bb_std, mamode=mamode)
    kch_wide = kc(high, low, close, length=kc_length, scalar=kc_scalar_wide, mamode=mamode, tr=use_tr)
    kch_normal = kc(high, low, close, length=kc_length, scalar=kc_scalar_normal, mamode=mamode, tr=use_tr)
    kch_narrow = kc(high, low, close, length=kc_length, scalar=kc_scalar_narrow, mamode=mamode, tr=use_tr)

    # 简化 KC 和 BBAND 列名以便动态访问
    bbd.columns = simplify_columns(bbd)
    kch_wide.columns = simplify_columns(kch_wide)
    kch_normal.columns = simplify_columns(kch_normal)
    kch_narrow.columns = simplify_columns(kch_narrow)

    # 计算动量
    momo = mom(close, length=mom_length)
    # 根据参数选择使用 EMA 还是 SMA
    if mamode.lower() == "ema":
        squeeze = ema(momo, length=mom_smooth)
    else: # "sma"
        squeeze = sma(momo, length=mom_smooth)

    # 分类挤压状态
    squeeze_on_wide = (bbd.l > kch_wide.l) & (bbd.u < kch_wide.u)
    # 计算是否在正常宽度 Keltner 通道内挤压
    squeeze_on_normal = (bbd.l > kch_normal.l) & (bbd.u < kch_normal.u)
    # 计算是否在狭窄 Keltner 通道内挤压
    squeeze_on_narrow = (bbd.l > kch_narrow.l) & (bbd.u < kch_narrow.u)
    # 计算是否在宽幅 Keltner 通道外挤压
    squeeze_off_wide = (bbd.l < kch_wide.l) & (bbd.u > kch_wide.u)
    # 计算未挤压的情况
    no_squeeze = ~squeeze_on_wide & ~squeeze_off_wide

    # 偏移处理
    if offset != 0:
        # 将挤压标志位移
        squeeze = squeeze.shift(offset)
        squeeze_on_wide = squeeze_on_wide.shift(offset)
        squeeze_on_normal = squeeze_on_normal.shift(offset)
        squeeze_on_narrow = squeeze_on_narrow.shift(offset)
        squeeze_off_wide = squeeze_off_wide.shift(offset)
        no_squeeze = no_squeeze.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        # 使用指定值填充 NaN 值
        squeeze.fillna(kwargs["fillna"], inplace=True)
        squeeze_on_wide.fillna(kwargs["fillna"], inplace=True)
        squeeze_on_normal.fillna(kwargs["fillna"], inplace=True)
        squeeze_on_narrow.fillna(kwargs["fillna"], inplace=True)
        squeeze_off_wide.fillna(kwargs["fillna"], inplace=True)
        no_squeeze.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        # 使用指定的填充方法填充 NaN 值
        squeeze.fillna(method=kwargs["fill_method"], inplace=True)
        squeeze_on_wide.fillna(method=kwargs["fill_method"], inplace=True)
        squeeze_on_normal.fillna(method=kwargs["fill_method"], inplace=True)
        squeeze_on_narrow.fillna(method=kwargs["fill_method"], inplace=True)
        squeeze_off_wide.fillna(method=kwargs["fill_method"], inplace=True)
        no_squeeze.fillna(method=kwargs["fill_method"], inplace=True)

    # 命名和分类
    _props = "" if use_tr else "hlr"
    _props += f"_{bb_length}_{bb_std}_{kc_length}_{kc_scalar_wide}_{kc_scalar_normal}_{kc_scalar_narrow}"
    # 设置挤压系列的名称
    squeeze.name = f"SQZPRO{_props}"

    # 创建数据字典
    data = {
        squeeze.name: squeeze,
        # 基于不同条件的挤压情况创建数据列
        f"SQZPRO_ON_WIDE": squeeze_on_wide.astype(int) if asint else squeeze_on_wide,
        f"SQZPRO_ON_NORMAL": squeeze_on_normal.astype(int) if asint else squeeze_on_normal,
        f"SQZPRO_ON_NARROW": squeeze_on_narrow.astype(int) if asint else squeeze_on_narrow,
        f"SQZPRO_OFF": squeeze_off_wide.astype(int) if asint else squeeze_off_wide,
        f"SQZPRO_NO": no_squeeze.astype(int) if asint else no_squeeze,
    }
    # 创建 DataFrame 对象
    df = DataFrame(data)
    # 设置 DataFrame 和挤压系列的名称和分类
    df.name = squeeze.name
    df.category = squeeze.category = "momentum"

    # 详细的挤压系列
    # 如果 detailed 参数为 True,则执行以下逻辑
    if detailed:
        # 从 squeeze 中提取非负值
        pos_squeeze = squeeze[squeeze >= 0]
        # 从 squeeze 中提取负值
        neg_squeeze = squeeze[squeeze < 0]

        # 计算非负值序列的无符号差异,转换为整数
        pos_inc, pos_dec = unsigned_differences(pos_squeeze, asint=True)
        # 计算负值序列的无符号差异,转换为整数
        neg_inc, neg_dec = unsigned_differences(neg_squeeze, asint=True)

        # 将差异乘以 squeeze,得到调整后的差异值
        pos_inc *= squeeze
        pos_dec *= squeeze
        neg_dec *= squeeze
        neg_inc *= squeeze

        # 将调整后的差异中为零的值替换为 NaN
        pos_inc.replace(0, np.NaN, inplace=True)
        pos_dec.replace(0, np.NaN, inplace=True)
        neg_dec.replace(0, np.NaN, inplace=True)
        neg_inc.replace(0, np.NaN, inplace=True)

        # 计算 squeeze 乘以增加函数的结果,并将为零的值替换为 NaN
        sqz_inc = squeeze * increasing(squeeze)
        sqz_dec = squeeze * decreasing(squeeze)
        sqz_inc.replace(0, np.NaN, inplace=True)
        sqz_dec.replace(0, np.NaN, inplace=True)

        # 处理填充值
        if "fillna" in kwargs:
            sqz_inc.fillna(kwargs["fillna"], inplace=True)
            sqz_dec.fillna(kwargs["fillna"], inplace=True)
            pos_inc.fillna(kwargs["fillna"], inplace=True)
            pos_dec.fillna(kwargs["fillna"], inplace=True)
            neg_dec.fillna(kwargs["fillna"], inplace=True)
            neg_inc.fillna(kwargs["fillna"], inplace=True)
        if "fill_method" in kwargs:
            sqz_inc.fillna(method=kwargs["fill_method"], inplace=True)
            sqz_dec.fillna(method=kwargs["fill_method"], inplace=True)
            pos_inc.fillna(method=kwargs["fill_method"], inplace=True)
            pos_dec.fillna(method=kwargs["fill_method"], inplace=True)
            neg_dec.fillna(method=kwargs["fill_method"], inplace=True)
            neg_inc.fillna(method=kwargs["fill_method"], inplace=True)

        # 将调整后的差异值和 squeeze 乘以增加函数的结果添加到 DataFrame 中
        df[f"SQZPRO_INC"] = sqz_inc
        df[f"SQZPRO_DEC"] = sqz_dec
        df[f"SQZPRO_PINC"] = pos_inc
        df[f"SQZPRO_PDEC"] = pos_dec
        df[f"SQZPRO_NDEC"] = neg_dec
        df[f"SQZPRO_NINC"] = neg_inc

    # 返回处理后的 DataFrame
    return df
# 设置Squeeze PRO指标的文档字符串,描述了该指标的作用、来源以及计算方法等信息
squeeze_pro.__doc__ = \
"""Squeeze PRO(SQZPRO)

This indicator is an extended version of "TTM Squeeze" from John Carter.
The default is based on John Carter's "TTM Squeeze" indicator, as discussed
in his book "Mastering the Trade" (chapter 11). The Squeeze indicator attempts
to capture the relationship between two studies: Bollinger Bands® and Keltner's
Channels. When the volatility increases, so does the distance between the bands,
conversely, when the volatility declines, the distance also decreases. It finds
sections of the Bollinger Bands® study which fall inside the Keltner's Channels.

# 给出指标来源的链接
Sources:
    https://usethinkscript.com/threads/john-carters-squeeze-pro-indicator-for-thinkorswim-free.4021/
    https://www.tradingview.com/script/TAAt6eRX-Squeeze-PRO-Indicator-Makit0/

# 指标的计算方法,包括默认输入参数、所使用的指标、以及计算过程
Calculation:
    Default Inputs:
        bb_length=20, bb_std=2, kc_length=20, kc_scalar_wide=2,
        kc_scalar_normal=1.5, kc_scalar_narrow=1, mom_length=12,
        mom_smooth=6, tr=True,
    BB = Bollinger Bands
    KC = Keltner Channels
    MOM = Momentum
    SMA = Simple Moving Average
    EMA = Exponential Moving Average
    TR = True Range

# 计算所需的变量
    RANGE = TR(high, low, close) if using_tr else high - low
    BB_LOW, BB_MID, BB_HIGH = BB(close, bb_length, std=bb_std)
    KC_LOW_WIDE, KC_MID_WIDE, KC_HIGH_WIDE = KC(high, low, close, kc_length, kc_scalar_wide, TR)
    KC_LOW_NORMAL, KC_MID_NORMAL, KC_HIGH_NORMAL = KC(high, low, close, kc_length, kc_scalar_normal, TR)
    KC_LOW_NARROW, KC_MID_NARROW, KC_HIGH_NARROW = KC(high, low, close, kc_length, kc_scalar_narrow, TR)

# 计算动量指标
    MOMO = MOM(close, mom_length)
    if mamode == "ema":
        SQZPRO = EMA(MOMO, mom_smooth)
    else:
        SQZPRO = EMA(momo, mom_smooth)

# 判断是否处于Squeeze状态
    SQZPRO_ON_WIDE  = (BB_LOW > KC_LOW_WIDE) and (BB_HIGH < KC_HIGH_WIDE)
    SQZPRO_ON_NORMAL  = (BB_LOW > KC_LOW_NORMAL) and (BB_HIGH < KC_HIGH_NORMAL)
    SQZPRO_ON_NARROW  = (BB_LOW > KC_LOW_NARROW) and (BB_HIGH < KC_HIGH_NARROW)
    SQZPRO_OFF_WIDE = (BB_LOW < KC_LOW_WIDE) and (BB_HIGH > KC_HIGH_WIDE)
    SQZPRO_NO = !SQZ_ON_WIDE and !SQZ_OFF_WIDE

# 定义函数参数及参数说明
Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    bb_length (int): Bollinger Bands period. Default: 20
    bb_std (float): Bollinger Bands Std. Dev. Default: 2
    kc_length (int): Keltner Channel period. Default: 20
    kc_scalar_wide (float): Keltner Channel scalar for wider channel. Default: 2
    kc_scalar_normal (float): Keltner Channel scalar for normal channel. Default: 1.5
    kc_scalar_narrow (float): Keltner Channel scalar for narrow channel. Default: 1
    mom_length (int): Momentum Period. Default: 12
    mom_smooth (int): Smoothing Period of Momentum. Default: 6
    mamode (str): Only "ema" or "sma". Default: "sma"
    offset (int): How many periods to offset the result. Default: 0

# 定义可选关键字参数及其默认值
Kwargs:
    tr (value, optional): Use True Range for Keltner Channels. Default: True
"""
    # asint (value, optional): 是否使用整数而不是布尔值。默认值为True
    # mamode (value, optional): 使用哪种移动平均线。默认值为"sma"
    # detailed (value, optional): 返回用于可视化的SQZ的额外变体。默认值为False
    # fillna (value, optional): pd.DataFrame.fillna(value)的填充值
    # fill_method (value, optional): 填充方法的类型
# 返回一个 pandas DataFrame 对象,包含默认的 SQZPRO、SQZPRO_ON_WIDE、SQZPRO_ON_NORMAL、SQZPRO_ON_NARROW、SQZPRO_OFF_WIDE、SQZPRO_NO 列。如果 'detailed' 参数为 True,则包含更详细的列。

标签:kc,squeeze,PandasTA,length,源码,kwargs,fillna,解析,True
From: https://www.cnblogs.com/apachecn/p/18135772

相关文章

  • PandasTA 源码解析(五)
    .\pandas-ta\pandas_ta\momentum\kst.py#-*-coding:utf-8-*-#导入DataFrame类frompandasimportDataFrame#导入roc函数from.rocimportroc#导入验证序列函数、获取漂移和偏移的函数frompandas_ta.utilsimportget_drift,get_offset,verify_series#定义......
  • PandasTA 源码解析(八)
    .\pandas-ta\pandas_ta\momentum\__init__.py#设置文件编码为UTF-8#导入ao指标from.aoimportao#导入apo指标from.apoimportapo#导入bias指标from.biasimportbias#导入bop指标from.bopimportbop#导入brar指标from.brarimportbrar#导......
  • PandasTA 源码解析(七)
    .\pandas-ta\pandas_ta\momentum\stc.py#-*-coding:utf-8-*-从pandas库中导入DataFrame和Series类从pandas_ta.overlap模块中导入ema函数从pandas_ta.utils模块中导入get_offset、non_zero_range和verify_series函数#定义函数:SchaffTrendCycle(STC......
  • 【合合TextIn】智能文档处理系列—电子文档解析技术全格式解析
    一、引言在当今的数字化时代,电子文档已成为信息存储和交流的基石。从简单的文本文件到复杂的演示文档,各种格式的电子文档承载着丰富的知识与信息,支撑着教育、科研、商业和日常生活的各个方面。随着信息量的爆炸性增长,如何高效、准确地处理和分析这些电子文档,已经成为信息技术领......
  • 2、APIView执行流程以及request对象源码分析
    一、基于View编写5个接口1、创建模型表models.pyfromdjango.dbimportmodelsclassBook(models.Model):name=models.CharField(max_length=64)price=models.IntegerField()publish=models.CharField(max_length=32)2、视图函数views.pyfrom......
  • JDK 源码阅读:java.lang.Object 类
    记录下自己阅读过程的笔记,如有错误,欢迎指正!源码参考:https://github.com/kangjianwei/LearningJDK/tree/master1.基本介绍在Java中,Object类是类层次结构的根类几乎每个Java类都直接或间接继承自Object类,意味着每个类都继承了Object的方法类结构:2.源码分析2.......
  • AddHandler导致的解析漏洞
    https://www.freebuf.com/vuls/303745.html在etc/apache/site-enabled下设置配置文件添加AddHandlerapplication/x-httpd-php.php然后重启apache服务配置问题导致总结:1)如果在Apache的/etc/apache2/apache2.conf里有这样的配置<FilesMatch"YLion.jpg">​SetHandler......
  • populateBean方法解析
    populateBean方法实现的功能autowired解析Autowired实例,code如下:packagecom.gientech.populateBean.annotation;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Controller;@ControllerpublicclassBook......
  • AOP解析
    AOPbean准备1AOPsample1.1创建切面packagecom.gientech.aop.xml.util;importorg.aspectj.lang.JoinPoint;importorg.aspectj.lang.ProceedingJoinPoint;importorg.aspectj.lang.Signature;importjava.util.Arrays;publicclassLogUtil{publicvoid......
  • SRPCore GenerateHLSL解析及其扩展
    序言在之前的项目开发HDRP的时候,就觉得绑定RT的管理就像一坨屎一样难受,改了管线RT的绑定顺序,原来的Shader输出又会有问题(主要是SV_Target顺序问题),没办法只能够在管线原来的绑定上面,重新写了一份文件,让Shader的共用Pass引用,这样就能规范不同Shaderpass的输出了。但是这只是解......