0.9.49 新增两个信号函数

czsc/signals/__init__.py

@@ -105,6 +105,8 @@
     bar_trend_V240209,
     bar_plr_V240427,
     bar_accelerate_V240428,
+    bar_polyfit_V240428,
+    bar_break_V240428,
 )
 
 from czsc.signals.jcc import (

czsc/signals/bar.py

@@ -1846,3 +1846,114 @@ def bar_plr_V240427(c: CZSC, **kwargs) -> OrderedDict:
         v1 = "满足" if plr > t / 10 else "不满足"
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def bar_polyfit_V240428(c: CZSC, **kwargs) -> OrderedDict:
+    """一阶、二阶多项式拟合
+
+    参考资料：
+    1. [基于低阶多项式拟合的日内趋势策略](https://zhuanlan.zhihu.com/p/391605615)
+    2. 罗军，广发证券，2011，《基于低阶多项式拟合的股指期货趋势交易(LPTT)策略》
+
+    参数模板："{freq}_D{di}W{w}_分类V240428"
+
+    **信号逻辑：**
+
+    若对一阶线性函数求一阶导数，也就是平常所说的斜率，若导数dy/dt>0，说明价格正处于上升趋势；若导数dy/dt<0，则为下跌趋势。
+    若对二阶线性函数求二阶导数，若二阶导数d2y/dt2>0，价格曲线为凹（开口向上）；若二阶导数d2y/dt2<0，价格曲线为凸（开口向下）。
+
+    做个类比，价格曲线就像汽车行走的距离轨迹，对距离（位移）求一阶导数就是速度，速度大于0说明朝正方向开，小于0说明朝反方向开。
+    求二阶导数就是加速度，假设此时汽车为正向行驶，加速度大于0说明在汽车速度还在增加的状态当中，在不断加速，反之则是处在减速的过程当中。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1W20_分类V240428_加速上涨_任意_任意_0')
+    - Signal('60分钟_D1W20_分类V240428_减速上涨_任意_任意_0')
+    - Signal('60分钟_D1W20_分类V240428_加速下跌_任意_任意_0')
+    - Signal('60分钟_D1W20_分类V240428_减速下跌_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+
+        - di: int, default 1, 周期偏移量
+        - w: int, default 60, 计算多项式拟合的K线数量
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    w = int(kwargs.get("w", 20))
+
+    assert di > 0, "参数 di 必须大于 0"
+    assert w >= 10, "参数 w 必须大于等于 10"
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}W{w}_分类V240428".split("_")
+    v1 = "其他"
+    if len(c.bars_raw) < 7 + w:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bars = get_sub_elements(c.bars_raw, di=di, n=w)
+    close = np.array([x.close for x in bars])
+    p1 = np.polyfit(np.arange(len(close)), close, 1)[0]
+    p2 = np.polyfit(np.arange(len(close)), close, 2)[0]
+
+    if p1 > 0 and p2 > 0:
+        v1 = "加速上涨"
+    elif p1 < 0 and p2 < 0:
+        v1 = "加速下跌"
+    elif p1 > 0 > p2:
+        v1 = "减速上涨"
+    elif p1 < 0 < p2:
+        v1 = "减速下跌"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def bar_break_V240428(c: CZSC, **kwargs) -> OrderedDict:
+    """极值突破
+
+    参考资料：
+    1. [后浪“拍死”前浪的日内波动极值策略](https://zhuanlan.zhihu.com/p/390025811)
+    2. 罗军，广发证券，2011，《基于日内波动极值的股指期货趋势跟随系统》
+
+    参数模板："{freq}_D{di}W{w}_事件V240428"
+
+    **信号逻辑：**
+
+    以60分钟级别为例，计算过程如下：
+    1. 获取w根K线的最高价和最低价，分别记为H和L；
+    2. 当前K线的收盘价，记为C；大于H时，触发收盘新高信号；小于L时，触发收盘新低信号。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1W20_事件V240428_收盘新低_任意_任意_0')
+    - Signal('60分钟_D1W20_事件V240428_收盘新高_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+
+        - di: int, default 1, 周期偏移量
+        - w: int, default 60, 计算多项式拟合的K线数量
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    w = int(kwargs.get("w", 20))
+
+    assert di > 0, "参数 di 必须大于 0"
+    assert w >= 10, "参数 w 必须大于等于 10"
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}W{w}_事件V240428".split("_")
+    v1 = "其他"
+    if len(c.bars_raw) < 7 + w:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bars = get_sub_elements(c.bars_raw, di=di, n=w)
+
+    if bars[-1].close > max([x.high for x in bars[:-1]]):
+        v1 = "收盘新高"
+    elif bars[-1].close < min([x.low for x in bars[:-1]]):
+        v1 = "收盘新低"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)

examples/signals_dev/bar_break_V240428.py

@@ -0,0 +1,74 @@
+import numpy as np
+from collections import OrderedDict
+from czsc.analyze import CZSC
+from czsc.objects import Direction, ZS
+from czsc.signals.tas import update_macd_cache
+from czsc.utils import create_single_signal, get_sub_elements
+
+
+def bar_break_V240428(c: CZSC, **kwargs) -> OrderedDict:
+    """极值突破
+
+    参考资料：
+    1. [后浪“拍死”前浪的日内波动极值策略](https://zhuanlan.zhihu.com/p/390025811)
+    2. 罗军，广发证券，2011，《基于日内波动极值的股指期货趋势跟随系统》
+
+    参数模板："{freq}_D{di}W{w}_事件V240428"
+
+    **信号逻辑：**
+
+    以60分钟级别为例，计算过程如下：
+    1. 获取w根K线的最高价和最低价，分别记为H和L；
+    2. 当前K线的收盘价，记为C；大于H时，触发收盘新高信号；小于L时，触发收盘新低信号。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1W20_事件V240428_收盘新低_任意_任意_0')
+    - Signal('60分钟_D1W20_事件V240428_收盘新高_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+
+        - di: int, default 1, 周期偏移量
+        - w: int, default 60, 计算多项式拟合的K线数量
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    w = int(kwargs.get("w", 20))
+
+    assert di > 0, "参数 di 必须大于 0"
+    assert w >= 10, "参数 w 必须大于等于 10"
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}W{w}_事件V240428".split("_")
+    v1 = "其他"
+    if len(c.bars_raw) < 7 + w:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bars = get_sub_elements(c.bars_raw, di=di, n=w)
+
+    if bars[-1].close > max([x.high for x in bars[:-1]]):
+        v1 = "收盘新高"
+    elif bars[-1].close < min([x.low for x in bars[:-1]]):
+        v1 = "收盘新低"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def check():
+    from czsc.connectors import research
+    from czsc.traders.base import check_signals_acc
+
+    symbols = research.get_symbols("A股主要指数")
+    bars = research.get_raw_bars(symbols[0], "15分钟", "20181101", "20210101", fq="前复权")
+
+    signals_config = [
+        {"name": bar_break_V240428, "freq": "60分钟"},
+        # {"name": bar_plr_V240427, "freq": "60分钟", "m": "空头"},
+    ]
+    check_signals_acc(bars, signals_config=signals_config, height="780px", delta_days=5)  # type: ignore
+
+
+if __name__ == "__main__":
+    check()

examples/signals_dev/bar_polyfit_V240428.py

@@ -0,0 +1,85 @@
+import numpy as np
+from collections import OrderedDict
+from czsc.analyze import CZSC
+from czsc.objects import Direction, ZS
+from czsc.signals.tas import update_macd_cache
+from czsc.utils import create_single_signal, get_sub_elements
+
+
+def bar_polyfit_V240428(c: CZSC, **kwargs) -> OrderedDict:
+    """一阶、二阶多项式拟合
+
+    参考资料：
+    1. [基于低阶多项式拟合的日内趋势策略](https://zhuanlan.zhihu.com/p/391605615)
+    2. 罗军，广发证券，2011，《基于低阶多项式拟合的股指期货趋势交易(LPTT)策略》
+
+    参数模板："{freq}_D{di}W{w}_分类V240428"
+
+    **信号逻辑：**
+
+    若对一阶线性函数求一阶导数，也就是平常所说的斜率，若导数dy/dt>0，说明价格正处于上升趋势；若导数dy/dt<0，则为下跌趋势。
+    若对二阶线性函数求二阶导数，若二阶导数d2y/dt2>0，价格曲线为凹（开口向上）；若二阶导数d2y/dt2<0，价格曲线为凸（开口向下）。
+
+    做个类比，价格曲线就像汽车行走的距离轨迹，对距离（位移）求一阶导数就是速度，速度大于0说明朝正方向开，小于0说明朝反方向开。
+    求二阶导数就是加速度，假设此时汽车为正向行驶，加速度大于0说明在汽车速度还在增加的状态当中，在不断加速，反之则是处在减速的过程当中。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1W20_分类V240428_加速上涨_任意_任意_0')
+    - Signal('60分钟_D1W20_分类V240428_减速上涨_任意_任意_0')
+    - Signal('60分钟_D1W20_分类V240428_加速下跌_任意_任意_0')
+    - Signal('60分钟_D1W20_分类V240428_减速下跌_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+
+        - di: int, default 1, 周期偏移量
+        - w: int, default 60, 计算多项式拟合的K线数量
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    w = int(kwargs.get("w", 20))
+
+    assert di > 0, "参数 di 必须大于 0"
+    assert w >= 10, "参数 w 必须大于等于 10"
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}W{w}_分类V240428".split("_")
+    v1 = "其他"
+    if len(c.bars_raw) < 7 + w:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bars = get_sub_elements(c.bars_raw, di=di, n=w)
+    close = np.array([x.close for x in bars])
+    p1 = np.polyfit(np.arange(len(close)), close, 1)[0]
+    p2 = np.polyfit(np.arange(len(close)), close, 2)[0]
+
+    if p1 > 0 and p2 > 0:
+        v1 = "加速上涨"
+    elif p1 < 0 and p2 < 0:
+        v1 = "加速下跌"
+    elif p1 > 0 and p2 < 0:
+        v1 = "减速上涨"
+    elif p1 < 0 and p2 > 0:
+        v1 = "减速下跌"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def check():
+    from czsc.connectors import research
+    from czsc.traders.base import check_signals_acc
+
+    symbols = research.get_symbols("A股主要指数")
+    bars = research.get_raw_bars(symbols[0], "15分钟", "20181101", "20210101", fq="前复权")
+
+    signals_config = [
+        {"name": bar_polyfit_V240428, "freq": "60分钟"},
+        # {"name": bar_plr_V240427, "freq": "60分钟", "m": "空头"},
+    ]
+    check_signals_acc(bars, signals_config=signals_config, height="780px", delta_days=5)  # type: ignore
+
+
+if __name__ == "__main__":
+    check()

examples/signals_dev/signal_match.py

@@ -45,7 +45,7 @@
     conf = sp.parse(signals_seq)
     parsed_name = {x["name"] for x in conf}
     print(f"total signal functions: {len(sp.sig_name_map)}; parsed: {len(parsed_name)}")
-    # total signal functions: 211; parsed: 211
+    # total signal functions: 215; parsed: 215
 
     # 测试信号配置生成信号
     from czsc import generate_czsc_signals, get_signals_freqs, get_signals_config