0.9.54 新增 cci_decision_V240620 信号

czsc/signals/__init__.py

@@ -215,6 +215,7 @@
     tas_dma_bs_V240608,
     tas_dif_zero_V240612,
     tas_dif_zero_V240614,
+    cci_decision_V240620,
 )
 
 from czsc.signals.pos import (

czsc/signals/tas.py

@@ -1991,6 +1991,49 @@ def tas_cci_base_V230402(c: CZSC, **kwargs) -> OrderedDict:
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
 
 
+def cci_decision_V240620(c: CZSC, **kwargs) -> OrderedDict:
+    """根据CCI指标逆势用法，判断买卖决策区域
+
+    参数模板："{freq}_N{n}CCI_决策区域V240620"
+
+    **信号逻辑：**
+
+    取最近N根K线，如果最小的CCI值小于 -100，开多；如果最大的CCI值大于 100，开空。
+
+    **信号列表：**
+
+    - Signal('15分钟_N4CCI_决策区域V240620_开多_2次_任意_0')
+    - Signal('15分钟_N4CCI_决策区域V240620_开多_1次_任意_0')
+    - Signal('15分钟_N4CCI_决策区域V240620_开空_1次_任意_0')
+    - Signal('15分钟_N4CCI_决策区域V240620_开空_2次_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    n = int(kwargs.get("n", 2))
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_N{n}CCI_决策区域V240620".split("_")
+    v1 = "其他"
+    cache_key = update_cci_cache(c, timeperiod=14)
+    if len(c.bars_raw) < 100:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    cci_seq = [x.cache[cache_key] for x in c.bars_raw[-n:]]
+    short_cci = [x for x in cci_seq if x > 100]
+    long_cci = [x for x in cci_seq if x < -100]
+
+    v2 = "任意"
+    if min(cci_seq) < -100:
+        v1 = "开多"
+        v2 = f"{len(long_cci)}次"
+    if max(cci_seq) > 100:
+        v1 = "开空"
+        v2 = f"{len(short_cci)}次"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
+
+
 def tas_kdj_evc_V230401(c: CZSC, **kwargs) -> OrderedDict:
     """KDJ极值计数信号, evc 是 extreme value counts 的首字母缩写
 

examples/signals_dev/cci_decision_V240620.py

@@ -0,0 +1,64 @@
+import numpy as np
+from collections import OrderedDict
+from czsc.analyze import CZSC, BI, Direction
+from czsc.signals.tas import update_cci_cache
+from czsc.utils import create_single_signal, get_sub_elements
+from loguru import logger
+
+
+def cci_decision_V240620(c: CZSC, **kwargs) -> OrderedDict:
+    """根据CCI指标逆势用法，判断买卖决策区域
+
+    参数模板："{freq}_N{n}CCI_决策区域V240620"
+
+    **信号逻辑：**
+
+    取最近N根K线，如果最小的CCI值小于 -100，开多；如果最大的CCI值大于 100，开空。
+
+    **信号列表：**
+
+    - Signal('15分钟_N4CCI_决策区域V240620_开多_2次_任意_0')
+    - Signal('15分钟_N4CCI_决策区域V240620_开多_1次_任意_0')
+    - Signal('15分钟_N4CCI_决策区域V240620_开空_1次_任意_0')
+    - Signal('15分钟_N4CCI_决策区域V240620_开空_2次_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    n = int(kwargs.get("n", 2))
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_N{n}CCI_决策区域V240620".split("_")
+    v1 = "其他"
+    cache_key = update_cci_cache(c, timeperiod=14)
+    if len(c.bars_raw) < 100:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    cci_seq = [x.cache[cache_key] for x in c.bars_raw[-n:]]
+    short_cci = [x for x in cci_seq if x > 100]
+    long_cci = [x for x in cci_seq if x < -100]
+
+    v2 = "任意"
+    if min(cci_seq) < -100:
+        v1 = "开多"
+        v2 = f"{len(long_cci)}次"
+    if max(cci_seq) > 100:
+        v1 = "开空"
+        v2 = f"{len(short_cci)}次"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
+
+
+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": cci_decision_V240620, "freq": "15分钟", "n": 4}]
+    check_signals_acc(bars, signals_config=signals_config, height="780px", delta_days=5)  # type: ignore
+
+
+if __name__ == "__main__":
+    check()