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

# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 .true_range 模块中导入 true_range 函数
from .true_range import true_range
# 从 pandas_ta.overlap 模块中导入 ma 函数
from pandas_ta.overlap import ma
# 从 pandas_ta.utils 模块中导入 get_offset, high_low_range, verify_series 函数
from pandas_ta.utils import get_offset, high_low_range, verify_series
# 定义函数 kc，用于计算 Keltner 通道（KC）指标
def kc(high, low, close, length=None, scalar=None, mamode=None, offset=None, **kwargs):
    """Indicator: Keltner Channels (KC)"""
    # 验证参数
    # 如果 length 存在且大于 0，则转换为整数，否则设置为默认值 20
    length = int(length) if length and length > 0 else 20
    # 如果 scalar 存在且大于 0，则转换为浮点数，否则设置为默认值 2
    scalar = float(scalar) if scalar and scalar > 0 else 2
    # 如果 mamode 是字符串类型，则保持不变，否则设置为默认值 "ema"
    mamode = mamode if isinstance(mamode, str) else "ema"
    # 验证 high、low、close 是否为有效的 Series，长度为 length
    high = verify_series(high, length)
    low = verify_series(low, length)
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)
    # 如果 high、low、close 存在空值，则返回空值
    if high is None or low is None or close is None: return
    # 计算结果
    # 判断是否使用 True Range（TR），默认为 True
    use_tr = kwargs.pop("tr", True)
    if use_tr:
        range_ = true_range(high, low, close)
    else:
        range_ = high_low_range(high, low)
    # 计算基准线和波动范围
    basis = ma(mamode, close, length=length)
    band = ma(mamode, range_, length=length)
    lower = basis - scalar * band
    upper = basis + scalar * band
    # 处理偏移量
    if offset != 0:
        lower = lower.shift(offset)
        basis = basis.shift(offset)
        upper = upper.shift(offset)
    # 处理填充值
    if "fillna" in kwargs:
        lower.fillna(kwargs["fillna"], inplace=True)
        basis.fillna(kwargs["fillna"], inplace=True)
        upper.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        lower.fillna(method=kwargs["fill_method"], inplace=True)
        basis.fillna(method=kwargs["fill_method"], inplace=True)
        upper.fillna(method=kwargs["fill_method"], inplace=True)
    # 命名并分类化
    _props = f"{mamode.lower()[0] if len(mamode) else ''}_{length}_{scalar}"
    lower.name = f"KCL{_props}"
    basis.name = f"KCB{_props}"
    upper.name = f"KCU{_props}"
    basis.category = upper.category = lower.category = "volatility"
    # 准备返回的 DataFrame
    data = {lower.name: lower, basis.name: basis, upper.name: upper}
    kcdf = DataFrame(data)
    kcdf.name = f"KC{_props}"
    kcdf.category = basis.category
    return kcdf
# 设置 kc 函数的文档字符串
kc.__doc__ = \
"""Keltner Channels (KC)
A popular volatility indicator similar to Bollinger Bands and
Donchian Channels.
Sources:
    https://www.tradingview.com/wiki/Keltner_Channels_(KC)
Calculation:
    Default Inputs:
        length=20, scalar=2, mamode=None, tr=True
    TR = True Range
    SMA = Simple Moving Average
    EMA = Exponential Moving Average
    if tr:
        RANGE = TR(high, low, close)
    else:
        RANGE = high - low
    if mamode == "ema":
        BASIS = sma(close, length)
        BAND = sma(RANGE, length)
    elif mamode == "sma":
        BASIS = sma(close, length)
        BAND = sma(RANGE, length)
    LOWER = BASIS - scalar * BAND
    UPPER = BASIS + scalar * BAND
Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
"""
    length (int): The short period.  Default: 20
    scalar (float): A positive float to scale the bands. Default: 2
    mamode (str): See ```help(ta.ma)```py. Default: 'ema'
    offset (int): How many periods to offset the result. Default: 0
# 函数参数：
#   tr (bool): 如果为 True，则使用 True Range 进行计算；如果为 False，则使用高 - 低作为范围计算。默认值为 True
#   fillna (value, optional): pd.DataFrame.fillna(value) 的可选参数，用于指定填充缺失值的值
#   fill_method (value, optional): 填充方法的类型
# 返回值：
#   返回一个 pandas DataFrame，包含 lower、basis、upper 列。

# -*- coding: utf-8 -*-
# 从 pandas_ta.overlap 模块导入 ema 函数
from pandas_ta.overlap import ema
# 从 pandas_ta.utils 模块导入 get_offset、non_zero_range、verify_series 函数
from pandas_ta.utils import get_offset, non_zero_range, verify_series
# 定义一个函数 massi，用于计算 Mass Index（MASSI）
def massi(high, low, fast=None, slow=None, offset=None, **kwargs):
    """Indicator: Mass Index (MASSI)"""
    # 验证参数的有效性
    # 如果 fast 有值且大于 0，则将其转换为整数，否则默认为 9
    fast = int(fast) if fast and fast > 0 else 9
    # 如果 slow 有值且大于 0，则将其转换为整数，否则默认为 25
    slow = int(slow) if slow and slow > 0 else 25
    # 如果 slow 小于 fast，则交换它们的值
    if slow < fast:
        fast, slow = slow, fast
    # 计算参数的最大值
    _length = max(fast, slow)
    # 验证 high 和 low 是否为有效序列
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    # 获取偏移量
    offset = get_offset(offset)
    # 移除 kwargs 中的 "length" 键
    if "length" in kwargs: kwargs.pop("length")
    # 如果 high 或 low 为 None，则返回
    if high is None or low is None: return
    # 计算结果
    # 计算高低价范围
    high_low_range = non_zero_range(high, low)
    # 计算高低价范围的 EMA
    hl_ema1 = ema(close=high_low_range, length=fast, **kwargs)
    # 计算高低价范围的 EMA 的 EMA
    hl_ema2 = ema(close=hl_ema1, length=fast, **kwargs)
    # 计算 hl_ratio
    hl_ratio = hl_ema1 / hl_ema2
    # 计算 MASSI
    massi = hl_ratio.rolling(slow, min_periods=slow).sum()
    # 调整偏移量
    if offset != 0:
        massi = massi.shift(offset)
    # 处理填充
    if "fillna" in kwargs:
        massi.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        massi.fillna(method=kwargs["fill_method"], inplace=True)
    # 给结果命名并归类
    massi.name = f"MASSI_{fast}_{slow}"
    massi.category = "volatility"
    return massi
# 将 massi 函数的文档字符串重新赋值，用于说明该函数的功能、计算方法以及参数等信息
massi.__doc__ = \
"""Mass Index (MASSI)
The Mass Index is a non-directional volatility indicator that utilitizes the
High-Low Range to identify trend reversals based on range expansions.
Sources:
    https://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:mass_index
    mi = sum(ema(high - low, 9) / ema(ema(high - low, 9), 9), length)
Calculation:
    Default Inputs:
        fast: 9, slow: 25
    EMA = Exponential Moving Average
    hl = high - low
    hl_ema1 = EMA(hl, fast)
    hl_ema2 = EMA(hl_ema1, fast)
    hl_ratio = hl_ema1 / hl_ema2
    MASSI = SUM(hl_ratio, slow)
Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    fast (int): The short period. Default: 9
    slow (int): The long period. Default: 25
    offset (int): How many periods to offset the result. Default: 0
Kwargs:
    fillna (value, optional): pd.DataFrame.fillna(value)
    fill_method (value, optional): Type of fill method
Returns:
    pd.Series: New feature generated.
"""

# -*- coding: utf-8 -*-
# 导入必要的库和模块
from .atr import atr
from pandas_ta import Imports
from pandas_ta.utils import get_drift, get_offset, verify_series
# 定义函数，计算标准化的平均真实范围（NATR）
def natr(high, low, close, length=None, scalar=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
    """Indicator: Normalized Average True Range (NATR)"""
    # 验证参数
    length = int(length) if length and length > 0 else 14
    mamode = mamode if isinstance(mamode, str) else "ema"
    scalar = float(scalar) if scalar else 100
    high = verify_series(high, length)
    low = verify_series(low, length)
    close = verify_series(close, length)
    drift = get_drift(drift)
    offset = get_offset(offset)
    mode_tal = bool(talib) if isinstance(talib, bool) else True
    # 如果输入的数据有缺失值，则返回空
    if high is None or low is None or close is None: return
    # 计算结果
    if Imports["talib"] and mode_tal:
        from talib import NATR
        natr = NATR(high, low, close, length)
    else:
        natr = scalar / close
        natr *= atr(high=high, low=low, close=close, length=length, mamode=mamode, drift=drift, offset=offset, **kwargs)
    # 偏移结果
    if offset != 0:
        natr = natr.shift(offset)
    # 处理填充值
    if "fillna" in kwargs:
        natr.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        natr.fillna(method=kwargs["fill_method"], inplace=True)
    # 设置名称和分类
    natr.name = f"NATR_{length}"
    natr.category = "volatility"
    return natr
# 设置函数文档字符串
natr.__doc__ = \
"""Normalized Average True Range (NATR)
Normalized Average True Range attempt to normalize the average true range.
Sources:
    https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/normalized-average-true-range-natr/
Calculation:
    Default Inputs:
        length=20
    ATR = Average True Range
    NATR = (100 / close) * ATR(high, low, close)
Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    length (int): The short period. Default: 20
    scalar (float): How much to magnify. Default: 100
    mamode (str): See ```help(ta.ma)```py. Default: 'ema'
    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
"""

# -*- coding: utf-8 -*-
# 从 pandas_ta.utils 模块导入 get_drift, get_offset, non_zero_range, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
# 定义函数 pdist，用于计算价格距离（PDIST）
def pdist(open_, high, low, close, drift=None, offset=None, **kwargs):
    """Indicator: Price Distance (PDIST)"""
    # 验证参数的有效性，确保它们都是 pd.Series 类型
    open_ = verify_series(open_)
    high = verify_series(high)
    low = verify_series(low)
    close = verify_series(close)
    # 获取漂移和偏移值，如果未提供，则使用默认值
    drift = get_drift(drift)
    offset = get_offset(offset)
    # 计算结果
    # PDIST = 2 * (high - low) - |close - open| + |open - close[drift]|
    pdist = 2 * non_zero_range(high, low)
    pdist += non_zero_range(open_, close.shift(drift)).abs()
    pdist -= non_zero_range(close, open_).abs()
    # 对结果进行偏移处理
    if offset != 0:
        pdist = pdist.shift(offset)
    # 处理填充值
    if "fillna" in kwargs:
        pdist.fillna(kwargs["fillna"], inplace=True)
    if "fill_method" in kwargs:
        pdist.fillna(method=kwargs["fill_method"], inplace=True)
    # 指定结果的名称和分类
    pdist.name = "PDIST"
    pdist.category = "volatility"
    return pdist
# 为 pdist 函数添加文档字符串
pdist.__doc__ = \
"""Price Distance (PDIST)
Measures the "distance" covered by price movements.
Sources:
    https://www.prorealcode.com/prorealtime-indicators/pricedistance/
Calculation:
    Default Inputs:
        drift=1
    PDIST = 2(high - low) - ABS(close - open) + ABS(open - close[drift])
Args:
    open_ (pd.Series): Series of 'opens's
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    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.
"""