PandasTA 源码解析（十三）

作者：我家小花儿 | 2024-04-16 06:35:02

踩

PandasTA 源码解析（十三）

`.\pandas-ta\pandas_ta\trend\decreasing.py`

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

# 定义一个名为 decreasing 的函数，用于计算序列是否递减
def decreasing(close, length=None, strict=None, asint=None, percent=None, drift=None, offset=None, **kwargs):
    """Indicator: Decreasing"""
    # 验证参数
    # 如果 length 参数存在且大于 0，则将其转换为整数，否则设为 1
    length = int(length) if length and length > 0 else 1
    # 如果 strict 参数是布尔类型，则保持不变，否则设为 False
    strict = strict if isinstance(strict, bool) else False
    # 如果 asint 参数是布尔类型，则保持不变，否则设为 True
    asint = asint if isinstance(asint, bool) else True
    # 对 close 序列进行验证，并设定长度为 length
    close = verify_series(close, length)
    # 获取 drift 和 offset 参数的值
    drift = get_drift(drift)
    offset = get_offset(offset)
    # 如果 percent 参数是百分比，则将其转换为浮点数，否则设为 False
    percent = float(percent) if is_percent(percent) else False

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

    # 计算结果
    # 如果 percent 存在，则对 close 序列进行缩放
    close_ = (1 - 0.01 * percent) * close if percent else close
    # 如果 strict 为 True，则进行严格递减的计算
    if strict:
        # 使用循环检查连续递减的情况
        decreasing = close < close_.shift(drift)
        for x in range(3, length + 1):
            decreasing = decreasing & (close.shift(x - (drift + 1)) < close_.shift(x - drift))

        # 填充缺失值为 0，并将结果转换为布尔类型
        decreasing.fillna(0, inplace=True)
        decreasing = decreasing.astype(bool)
    else:
        # 否则，使用简单的递减计算
        decreasing = close_.diff(length) < 0

    # 如果 asint 为 True，则将结果转换为整数
    if asint:
        decreasing = decreasing.astype(int)

    # 对结果进行偏移
    if offset != 0:
        decreasing = decreasing.shift(offset)

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

    # 设定结果的名称和类别
    _percent = f"_{0.01 * percent}" if percent else ''
    _props = f"{'S' if strict else ''}DEC{'p' if percent else ''}"
    decreasing.name = f"{_props}_{length}{_percent}"
    decreasing.category = "trend"

    return decreasing

# 为 decreasing 函数添加文档字符串
decreasing.__doc__ = \
"""Decreasing

Returns True if the series is decreasing over a period, False otherwise.
If the kwarg 'strict' is True, it returns True if it is continuously decreasing
over the period. When using the kwarg 'asint', then it returns 1 for True
or 0 for False.

Calculation:
    if strict:
        decreasing = all(i > j for i, j in zip(close[-length:], close[1:]))
    else:
        decreasing = close.diff(length) < 0

    if asint:
        decreasing = decreasing.astype(int)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 1
    strict (bool): If True, checks if the series is continuously decreasing over the period. Default: False
    percent (float): Percent as an integer. Default: None
    asint (bool): Returns as binary. Default: True
    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.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

`.\pandas-ta\pandas_ta\trend\dpo.py`

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


def dpo(close, length=None, centered=True, offset=None, **kwargs):
    """Indicator: Detrend Price Oscillator (DPO)"""
    # 验证参数
    # 将长度参数转换为整数，如果长度参数存在且大于0，则使用，否则默认为20
    length = int(length) if length and length > 0 else 20
    # 验证 close 是否为有效的时间序列，长度为指定的长度
    close = verify_series(close, length)
    # 获取偏移量
    offset = get_offset(offset)
    # 如果 kwargs 中的 "lookahead" 为 False，则将 centered 设置为 False
    if not kwargs.get("lookahead", True):
        centered = False

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

    # 计算结果
    # 计算 t，即 int(0.5 * length) + 1
    t = int(0.5 * length) + 1
    # 计算 close 的简单移动平均
    ma = sma(close, length)

    # 计算 DPO，close 减去 ma 后向前位移 t 个周期
    dpo = close - ma.shift(t)
    # 如果 centered 为 True，则再将 DPO 向后位移 t 个周期
    if centered:
        dpo = (close.shift(t) - ma).shift(-t)

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

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

    # 命名和分类
    dpo.name = f"DPO_{length}"
    dpo.category = "trend"

    return dpo


# 更新文档字符串
dpo.__doc__ = \
"""Detrend Price Oscillator (DPO)

Is an indicator designed to remove trend from price and make it easier to
identify cycles.

Sources:
    https://www.tradingview.com/scripts/detrendedpriceoscillator/
    https://www.fidelity.com/learning-center/trading-investing/technical-analysis/technical-indicator-guide/dpo
    http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:detrended_price_osci

Calculation:
    Default Inputs:
        length=20, centered=True
    SMA = Simple Moving Average
    t = int(0.5 * length) + 1

    DPO = close.shift(t) - SMA(close, length)
    if centered:
        DPO = DPO.shift(-t)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 1
    centered (bool): Shift the dpo back by int(0.5 * length) + 1. 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.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

`.\pandas-ta\pandas_ta\trend\increasing.py`

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

# 定义名为 increasing 的函数，用于计算序列是否递增
def increasing(close, length=None, strict=None, asint=None, percent=None, drift=None, offset=None, **kwargs):
    """Indicator: Increasing"""
    # 验证参数的有效性
    length = int(length) if length and length > 0 else 1  # 将长度转换为整数，如果未提供或小于等于0，则设为1
    strict = strict if isinstance(strict, bool) else False  # 如果 strict 不是布尔值，则设为 False
    asint = asint if isinstance(asint, bool) else True  # 如果 asint 不是布尔值，则设为 True
    close = verify_series(close, length)  # 验证并处理输入的序列数据
    drift = get_drift(drift)  # 获取漂移值
    offset = get_offset(offset)  # 获取偏移值
    percent = float(percent) if is_percent(percent) else False  # 将百分比转换为浮点数，如果不是百分比，则设为 False

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

    # 计算结果
    close_ = (1 + 0.01 * percent) * close if percent else close  # 如果有百分比参数，则对序列进行调整
    if strict:
        # 返回值是否为 float64？必须转换为布尔值
        increasing = close > close_.shift(drift)  # 检查当前值是否大于移动后的值
        for x in range(3, length + 1):
            increasing = increasing & (close.shift(x - (drift + 1)) > close_.shift(x - drift))  # 检查连续多个值是否递增

        increasing.fillna(0, inplace=True)  # 填充缺失值为0
        increasing = increasing.astype(bool)  # 将结果转换为布尔值
    else:
        increasing = close_.diff(length) > 0  # 检查序列是否在给定周期内递增

    if asint:
        increasing = increasing.astype(int)  # 将结果转换为整数类型

    # 偏移结果
    if offset != 0:
        increasing = increasing.shift(offset)  # 对结果进行偏移

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

    # 命名并分类结果
    _percent = f"_{0.01 * percent}" if percent else ''  # 根据是否存在百分比参数构建后缀
    _props = f"{'S' if strict else ''}INC{'p' if percent else ''}"  # 根据参数构建特性标识
    increasing.name = f"{_props}_{length}{_percent}"  # 构建结果的名称
    increasing.category = "trend"  # 将结果分类为趋势

    return increasing  # 返回计算结果


increasing.__doc__ = \
"""Increasing

Returns True if the series is increasing over a period, False otherwise.
If the kwarg 'strict' is True, it returns True if it is continuously increasing
over the period. When using the kwarg 'asint', then it returns 1 for True
or 0 for False.

Calculation:
    if strict:
        increasing = all(i < j for i, j in zip(close[-length:], close[1:]))
    else:
        increasing = close.diff(length) > 0

    if asint:
        increasing = increasing.astype(int)

Args:
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 1
    strict (bool): If True, checks if the series is continuously increasing over the period. Default: False
    percent (float): Percent as an integer. Default: None
    asint (bool): Returns as binary. Default: True
    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.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

`.\pandas-ta\pandas_ta\trend\long_run.py`

# -*- coding: utf-8 -*-
# 导入必要的模块和函数
from .decreasing import decreasing  # 从当前目录下的 decreasing 模块导入 decreasing 函数
from .increasing import increasing  # 从当前目录下的 increasing 模块导入 increasing 函数
from pandas_ta.utils import get_offset, verify_series  # 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数


def long_run(fast, slow, length=None, offset=None, **kwargs):
    """Indicator: Long Run"""
    # Validate Arguments
    # 将 length 转换为整数，如果 length 存在且大于 0；否则默认为 2
    length = int(length) if length and length > 0 else 2
    # 验证 fast 和 slow 是否为有效的序列，并将其长度限制为 length
    fast = verify_series(fast, length)
    slow = verify_series(slow, length)
    # 获取偏移量
    offset = get_offset(offset)

    # 如果 fast 或 slow 为空，则返回空值
    if fast is None or slow is None: return

    # Calculate Result
    # 计算可能的底部或底部的条件，即 fast 增长而 slow 减小
    pb = increasing(fast, length) & decreasing(slow, length)
    # 计算 fast 和 slow 同时增长的条件
    bi = increasing(fast, length) & increasing(slow, length)
    # 计算长期趋势的条件，可能的底部或底部，以及 fast 和 slow 同时增长的情况
    long_run = pb | bi

    # Offset
    # 如果 offset 不为 0，则对长期趋势进行偏移
    if offset != 0:
        long_run = long_run.shift(offset)

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

    # Name and Categorize it
    # 设置长期趋势指标的名称
    long_run.name = f"LR_{length}"
    # 设置长期趋势指标的类别为 "trend"
    long_run.category = "trend"

    return long_run
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

`.\pandas-ta\pandas_ta\trend\psar.py`

# -*- coding: utf-8 -*-
# 导入 numpy 库中的 nan 作为 npNaN
from numpy import nan as npNaN
# 导入 DataFrame 和 Series 类
from pandas import DataFrame, Series
# 导入 get_offset, verify_series, zero 函数
from pandas_ta.utils import get_offset, verify_series, zero

# 定义 PSAR 指标函数，参数包括 high, low, close, af0, af, max_af, offset
def psar(high, low, close=None, af0=None, af=None, max_af=None, offset=None, **kwargs):
    """Indicator: Parabolic Stop and Reverse (PSAR)"""
    # 验证参数
    high = verify_series(high)
    low = verify_series(low)
    af = float(af) if af and af > 0 else 0.02
    af0 = float(af0) if af0 and af0 > 0 else af
    max_af = float(max_af) if max_af and max_af > 0 else 0.2
    offset = get_offset(offset)

    # 定义 _falling 函数，用于返回最后一个 -DM 值
    def _falling(high, low, drift:int=1):
        """Returns the last -DM value"""
        # 不要与 ta.falling() 混淆
        up = high - high.shift(drift)
        dn = low.shift(drift) - low
        _dmn = (((dn > up) & (dn > 0)) * dn).apply(zero).iloc[-1]
        return _dmn > 0

    # 如果第一个 NaN -DM 为正，则为下降趋势
    falling = _falling(high.iloc[:2], low.iloc[:2])
    if falling:
        sar = high.iloc[0]
        ep = low.iloc[0]
    else:
        sar = low.iloc[0]
        ep = high.iloc[0]

    # 如果存在 close 参数，则使用 close 的值
    if close is not None:
        close = verify_series(close)
        sar = close.iloc[0]

    # 初始化 long, short, reversal, _af
    long = Series(npNaN, index=high.index)
    short = long.copy()
    reversal = Series(0, index=high.index)
    _af = long.copy()
    _af.iloc[0:2] = af0

    # 计算结果
    m = high.shape[0]
    for row in range(1, m):
        high_ = high.iloc[row]
        low_ = low.iloc[row]

        if falling:
            _sar = sar + af * (ep - sar)
            reverse = high_ > _sar

            if low_ < ep:
                ep = low_
                af = min(af + af0, max_af)

            _sar = max(high.iloc[row - 1], high.iloc[row - 2], _sar)
        else:
            _sar = sar + af * (ep - sar)
            reverse = low_ < _sar

            if high_ > ep:
                ep = high_
                af = min(af + af0, max_af)

            _sar = min(low.iloc[row - 1], low.iloc[row - 2], _sar)

        if reverse:
            _sar = ep
            af = af0
            falling = not falling
            ep = low_ if falling else high_

        sar = _sar

        if falling:
            short.iloc[row] = sar
        else:
            long.iloc[row] = sar

        _af.iloc[row] = af
        reversal.iloc[row] = int(reverse)

    # 偏移
    if offset != 0:
        _af = _af.shift(offset)
        long = long.shift(offset)
        short = short.shift(offset)
        reversal = reversal.shift(offset)

    # 处理填充
    if "fillna" in kwargs:
        _af.fillna(kwargs["fillna"], inplace=True)
        long.fillna(kwargs["fillna"], inplace=True)
        short.fillna(kwargs["fillna"], inplace=True)
        reversal.fillna(kwargs["fillna"], inplace=True)
    # 检查参数中是否包含'fill_method'关键字
    if "fill_method" in kwargs:
        # 使用指定的填充方法填充数据，直接修改原数据，无需返回值
        _af.fillna(method=kwargs["fill_method"], inplace=True)
        long.fillna(method=kwargs["fill_method"], inplace=True)
        short.fillna(method=kwargs["fill_method"], inplace=True)
        reversal.fillna(method=kwargs["fill_method"], inplace=True)

    # 准备要返回的 DataFrame
    # 根据参数构建用于命名列的字符串
    _params = f"_{af0}_{max_af}"
    # 构建包含 PSAR 数据的字典
    data = {
        f"PSARl{_params}": long,
        f"PSARs{_params}": short,
        f"PSARaf{_params}": _af,
        f"PSARr{_params}": reversal,
    }
    # 从字典创建 DataFrame
    psardf = DataFrame(data)
    # 设置 DataFrame 的名称
    psardf.name = f"PSAR{_params}"
    # 设置 DataFrame 及其列的类别属性为 'trend'
    psardf.category = long.category = short.category = "trend"

    # 返回 PSAR 数据的 DataFrame
    return psardf
# 设置 psar 函数的文档字符串，用于描述 Parabolic Stop and Reverse (PSAR) 指标的作用、计算方式和参数说明
psar.__doc__ = \
"""Parabolic Stop and Reverse (psar)

Parabolic Stop and Reverse (PSAR) was developed by J. Wells Wilder, that is used
to determine trend direction and it's potential reversals in price. PSAR uses a
trailing stop and reverse method called "SAR," or stop and reverse, to identify
possible entries and exits. It is also known as SAR.

PSAR indicator typically appears on a chart as a series of dots, either above or
below an asset's price, depending on the direction the price is moving. A dot is
placed below the price when it is trending upward, and above the price when it
is trending downward.

Sources:
    https://www.tradingview.com/pine-script-reference/#fun_sar
    https://www.sierrachart.com/index.php?page=doc/StudiesReference.php&ID=66&Name=Parabolic

Calculation:
    Default Inputs:
        af0=0.02, af=0.02, max_af=0.2

    See Source links

Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series, optional): Series of 'close's. Optional
    af0 (float): Initial Acceleration Factor. Default: 0.02
    af (float): Acceleration Factor. Default: 0.02
    max_af (float): Maximum Acceleration Factor. Default: 0.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.DataFrame: long, short, af, and reversal columns.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

`.\pandas-ta\pandas_ta\trend\qstick.py`

# -*- coding: utf-8 -*-
# 从 pandas_ta.overlap 模块导入 dema, ema, hma, rma, sma 函数
from pandas_ta.overlap import dema, ema, hma, rma, sma
# 从 pandas_ta.utils 模块导入 get_offset, non_zero_range, verify_series 函数
from pandas_ta.utils import get_offset, non_zero_range, verify_series

# 定义 Q Stick 指标函数
def qstick(open_, close, length=None, offset=None, **kwargs):
    """Indicator: Q Stick"""
    # 验证参数
    length = int(length) if length and length > 0 else 10
    ma = kwargs.pop("ma", "sma")
    open_ = verify_series(open_, length)
    close = verify_series(close, length)
    offset = get_offset(offset)

    if open_ is None or close is None: return

    # 计算结果
    diff = non_zero_range(close, open_)

    if ma == "dema":
        qstick = dema(diff, length=length, **kwargs)
    elif ma == "ema":
        qstick = ema(diff, length=length, **kwargs)
    elif ma == "hma":
        qstick = hma(diff, length=length)
    elif ma == "rma":
        qstick = rma(diff, length=length)
    else: # "sma"
        qstick = sma(diff, length=length)

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

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

    # 命名和分类
    qstick.name = f"QS_{length}"
    qstick.category = "trend"

    return qstick

# 设置 Q Stick 函数的文档字符串
qstick.__doc__ = \
"""Q Stick

The Q Stick indicator, developed by Tushar Chande, attempts to quantify and
identify trends in candlestick charts.

Sources:
    https://library.tradingtechnologies.com/trade/chrt-ti-qstick.html

Calculation:
    Default Inputs:
        length=10
    xMA is one of: sma (default), dema, ema, hma, rma
    qstick = xMA(close - open, length)

Args:
    open (pd.Series): Series of 'open's
    close (pd.Series): Series of 'close's
    length (int): It's period. Default: 1
    ma (str): The type of moving average to use. Default: None, which is 'sma'
    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.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

`.\pandas-ta\pandas_ta\trend\short_run.py`

# -*- coding: utf-8 -*-
# 导入必要的模块和函数
from .decreasing import decreasing  # 从当前目录下的 decreasing 模块中导入 decreasing 函数
from .increasing import increasing  # 从当前目录下的 increasing 模块中导入 increasing 函数
from pandas_ta.utils import get_offset, verify_series  # 从 pandas_ta 包中的 utils 模块导入 get_offset 和 verify_series 函数

# 定义一个名为 short_run 的函数，用于计算短期趋势
def short_run(fast, slow, length=None, offset=None, **kwargs):
    """Indicator: Short Run"""
    # 验证参数
    length = int(length) if length and length > 0 else 2  # 如果 length 存在且大于0，则将其转换为整数类型，否则设为默认值2
    fast = verify_series(fast, length)  # 验证 fast 参数，确保其为有效的序列，并可能调整长度
    slow = verify_series(slow, length)  # 验证 slow 参数，确保其为有效的序列，并可能调整长度
    offset = get_offset(offset)  # 获取偏移量

    if fast is None or slow is None: return  # 如果 fast 或 slow 为空，则返回空值

    # 计算结果
    pt = decreasing(fast, length) & increasing(slow, length)  # 潜在顶部或顶部的条件
    bd = decreasing(fast, length) & decreasing(slow, length)  # 快速和慢速都在下降的条件
    short_run = pt | bd  # 判断是否出现短期趋势

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

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

    # 命名和分类
    short_run.name = f"SR_{length}"  # 设置结果的名称
    short_run.category = "trend"  # 设置结果的分类为趋势

    return short_run  # 返回计算结果
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

`.\pandas-ta\pandas_ta\trend\tsignals.py`

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

# 从 pandas 库导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.utils 模块导入 get_drift, get_offset, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, verify_series

# 定义 tsignals 函数，用于计算趋势信号
def tsignals(trend, asbool=None, trend_reset=0, trade_offset=None, drift=None, offset=None, **kwargs):
    """Indicator: Trend Signals"""
    # 验证参数
    # 将 trend 参数转换为 pandas Series 对象
    trend = verify_series(trend)
    # 将 asbool 参数转换为布尔值，默认为 False
    asbool = bool(asbool) if isinstance(asbool, bool) else False
    # 将 trend_reset 参数转换为整数，默认为 0
    trend_reset = int(trend_reset) if trend_reset and isinstance(trend_reset, int) else 0
    # 如果 trade_offset 不为 0，则将其转换为整数，默认为 0
    if trade_offset != 0:
        trade_offset = int(trade_offset) if trade_offset and isinstance(trade_offset, int) else 0
    # 获取 drift 和 offset 参数的值
    drift = get_drift(drift)
    offset = get_offset(offset)

    # 计算结果
    # 将趋势值转换为整数类型
    trends = trend.astype(int)
    # 计算交易信号
    trades = trends.diff(drift).shift(trade_offset).fillna(0).astype(int)
    # 计算进入交易的信号
    entries = (trades > 0).astype(int)
    # 计算退出交易的信号
    exits = (trades < 0).abs().astype(int)

    # 如果 asbool 为 True，则将结果转换为布尔值
    if asbool:
        trends = trends.astype(bool)
        entries = entries.astype(bool)
        exits = exits.astype(bool)

    # 构建结果数据
    data = {
        f"TS_Trends": trends,
        f"TS_Trades": trades,
        f"TS_Entries": entries,
        f"TS_Exits": exits,
    }
    # 创建 DataFrame 对象
    df = DataFrame(data, index=trends.index)

    # 处理偏移
    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)

    # 设置名称和类别
    df.name = f"TS"
    df.category = "trend"

    return df

# 设置 tsignals 函数的文档字符串
tsignals.__doc__ = \
"""Trend Signals

Given a Trend, Trend Signals returns the Trend, Trades, Entries and Exits as
boolean integers. When 'asbool=True', it returns Trends, Entries and Exits as
boolean values which is helpful when combined with the vectorbt backtesting
package.

A Trend can be a simple as: 'close' > 'moving average' or something more complex
whose values are boolean or integers (0 or 1).

Examples:
ta.tsignals(close > ta.sma(close, 50), asbool=False)
ta.tsignals(ta.ema(close, 8) > ta.ema(close, 21), asbool=True)

Source: Kevin Johnson

Calculation:
    Default Inputs:
        asbool=False, trend_reset=0, trade_offset=0, drift=1

    trades = trends.diff().shift(trade_offset).fillna(0).astype(int)
    entries = (trades > 0).astype(int)
    exits = (trades < 0).abs().astype(int)

Args:
    trend (pd.Series): Series of 'trend's. The trend can be either a boolean or
        integer series of '0's and '1's
    asbool (bool): If True, it converts the Trends, Entries and Exits columns to
        booleans. When boolean, it is also useful for backtesting with
        vectorbt's Portfolio.from_signal(close, entries, exits) Default: False
    trend_reset (value): Value used to identify if a trend has ended. Default: 0
    trade_offset (value): Value used shift the trade entries/exits Use 1 for
        backtesting and 0 for live. Default: 0
    drift (int): The difference period. Default: 1

"""
    offset (int): How many periods to offset the result. Default: 0
# 函数参数说明部分，描述了函数的参数和返回值
Kwargs:
    # fillna 参数，用于填充缺失值，使用 pd.DataFrame.fillna 函数
    fillna (value, optional): pd.DataFrame.fillna(value)
    # fill_method 参数，填充方法的类型说明
    fill_method (value, optional): Type of fill method

# 返回值说明部分，描述了函数返回的 DataFrame 的列
Returns:
    # 返回一个 pandas DataFrame，包含以下列：
    pd.DataFrame with columns:
    # Trends 列，表示趋势，有趋势为 1，无趋势为 0
    Trends (trend: 1, no trend: 0),
    # Trades 列，表示交易，进入为 1，退出为 -1，其他情况为 0
    Trades (Enter: 1, Exit: -1, Otherwise: 0),
    # Entries 列，表示进入，进入为 1，无操作为 0
    Entries (entry: 1, nothing: 0),
    # Exits 列，表示退出，退出为 1，无操作为 0
    Exits (exit: 1, nothing: 0)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

`.\pandas-ta\pandas_ta\trend\ttm_trend.py`

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

# 从 pandas 库导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.overlap 模块导入 hl2 函数
from pandas_ta.overlap import hl2
# 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series


# 定义 ttm_trend 函数，用于计算 TTM Trend 指标
def ttm_trend(high, low, close, length=None, offset=None, **kwargs):
    """Indicator: TTM Trend (TTM_TRND)"""

    # 验证参数
    # 如果 length 存在且大于 0，则将其转换为整数，否则将其设置为默认值 6
    length = int(length) if length and length > 0 else 6
    # 验证 high、low、close 数据是否为 Series，并确保它们的长度为 length
    high = verify_series(high, length)
    low = verify_series(low, length)
    close = verify_series(close, length)
    # 获取 offset 的偏移量
    offset = get_offset(offset)

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

    # 计算结果
    # 计算中间价（hl2）的平均值
    trend_avg = hl2(high, low)
    for i in range(1, length):
        trend_avg = trend_avg + hl2(high.shift(i), low.shift(i))

    trend_avg = trend_avg / length

    # 计算 ttm_trend，如果 close 大于 trend_avg，则设置为 1，否则设置为 -1
    tm_trend = (close > trend_avg).astype(int)
    tm_trend.replace(0, -1, inplace=True)

    # 调整结果的偏移量
    if offset != 0:
        tm_trend = tm_trend.shift(offset)

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

    # 设置名称和类别
    tm_trend.name = f"TTM_TRND_{length}"
    tm_trend.category = "momentum"

    # 准备返回的 DataFrame
    data = {tm_trend.name: tm_trend}
    df = DataFrame(data)
    df.name = f"TTMTREND_{length}"
    df.category = tm_trend.category

    return df


# 设置 ttm_trend 函数的文档字符串
ttm_trend.__doc__ = \
"""TTM Trend (TTM_TRND)

This indicator is from John Carters book “Mastering the Trade” and plots the
bars green or red. It checks if the price is above or under the average price of
the previous 5 bars. The indicator should hep you stay in a trade until the
colors chance. Two bars of the opposite color is the signal to get in or out.

Sources:
    https://www.prorealcode.com/prorealtime-indicators/ttm-trend-price/

Calculation:
    Default Inputs:
        length=6
    averageprice = (((high[5]+low[5])/2)+((high[4]+low[4])/2)+((high[3]+low[3])/2)+
    ((high[2]+low[2])/2)+((high[1]+low[1])/2)+((high[6]+low[6])/2)) / 6

    if close > averageprice:
        drawcandle(open,high,low,close) coloured(0,255,0)

    if close < averageprice:
        drawcandle(open,high,low,close) coloured(255,0,0)

Args:
    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: 6
    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: ttm_trend.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

`.\pandas-ta\pandas_ta\trend\vhf.py`

# -*- coding: utf-8 -*-
# 导入 fabs 函数并重命名为 npFabs
from numpy import fabs as npFabs
# 从 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


# 定义垂直水平过滤器（VHF）指标函数
def vhf(close, length=None, drift=None, offset=None, **kwargs):
    """Indicator: Vertical Horizontal Filter (VHF)"""
    # 验证参数
    # 将长度转换为整数，如果长度大于0，则为指定的长度，否则为默认值28
    length = int(length) if length and length > 0 else 28
    # 验证收盘价序列，长度为指定的长度
    close = verify_series(close, length)
    # 获取漂移值
    drift = get_drift(drift)
    # 获取偏移值
    offset = get_offset(offset)

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

    # 计算结果
    # 最高收盘价
    hcp = close.rolling(length).max()
    # 最低收盘价
    lcp = close.rolling(length).min()
    # 收盘价变化的绝对值
    diff = npFabs(close.diff(drift))
    # 垂直水平过滤器值
    vhf = npFabs(non_zero_range(hcp, lcp)) / diff.rolling(length).sum()

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

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

    # 命名和分类
    vhf.name = f"VHF_{length}"
    vhf.category = "trend"

    return vhf


# 设置 VHF 函数的文档字符串
vhf.__doc__ = \
"""Vertical Horizontal Filter (VHF)

VHF was created by Adam White to identify trending and ranging markets.

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

Calculation:
    Default Inputs:
        length = 28
    HCP = Highest Close Price in Period
    LCP = Lowest Close Price in Period
    Change = abs(Ct - Ct-1)
    VHF = (HCP - LCP) / RollingSum[length] of Change

Args:
    source (pd.Series): Series of prices (usually close).
    length (int): The period length. Default: 28
    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.
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

`.\pandas-ta\pandas_ta\trend\vortex.py`

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

# 从 pandas 库导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.volatility 模块导入 true_range 函数
from pandas_ta.volatility import true_range
# 从 pandas_ta.utils 模块导入 get_drift, get_offset, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, verify_series

# 定义函数 vortex，用于计算 Vortex 指标
def vortex(high, low, close, length=None, drift=None, offset=None, **kwargs):
    """Indicator: Vortex"""
    # 验证参数
    length = length if length and length > 0 else 14
    min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length
    _length = max(length, min_periods)
    # 验证并处理输入的 high、low、close 数据
    high = verify_series(high, _length)
    low = verify_series(low, _length)
    close = verify_series(close, _length)
    drift = get_drift(drift)  # 获取 drift 参数
    offset = get_offset(offset)  # 获取 offset 参数

    # 若输入数据中有空值，则返回空
    if high is None or low is None or close is None: return

    # 计算结果
    tr = true_range(high=high, low=low, close=close)  # 计算真实范围
    tr_sum = tr.rolling(length, min_periods=min_periods).sum()  # 对真实范围进行滚动求和

    # 计算正向运动价格动量 (VMP) 和反向运动价格动量 (VMM)
    vmp = (high - low.shift(drift)).abs()
    vmm = (low - high.shift(drift)).abs()

    # 计算正向和反向运动指标 (VIP 和 VIM)
    vip = vmp.rolling(length, min_periods=min_periods).sum() / tr_sum
    vim = vmm.rolling(length, min_periods=min_periods).sum() / tr_sum

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

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

    # 命名并分类化指标
    vip.name = f"VTXP_{length}"
    vim.name = f"VTXM_{length}"
    vip.category = vim.category = "trend"

    # 准备返回的 DataFrame
    data = {vip.name: vip, vim.name: vim}
    vtxdf = DataFrame(data)
    vtxdf.name = f"VTX_{length}"
    vtxdf.category = "trend"

    return vtxdf


# 设置函数文档字符串
vortex.__doc__ = \
"""Vortex

Two oscillators that capture positive and negative trend movement.

Sources:
    https://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:vortex_indicator

Calculation:
    Default Inputs:
        length=14, drift=1
    TR = True Range
    SMA = Simple Moving Average
    tr = TR(high, low, close)
    tr_sum = tr.rolling(length).sum()

    vmp = (high - low.shift(drift)).abs()
    vmn = (low - high.shift(drift)).abs()

    VIP = vmp.rolling(length).sum() / tr_sum
    VIM = vmn.rolling(length).sum() / tr_sum

Args:
    high (pd.Series): Series of 'high's
    low (pd.Series): Series of 'low's
    close (pd.Series): Series of 'close's
    length (int): ROC 1 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.DataFrame: vip and vim columns
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/432834