0.9.53 新增一批信号函数

czsc/signals/__init__.py

@@ -47,6 +47,8 @@
     cxt_bs_V240527,
     cxt_overlap_V240612,
     cxt_decision_V240612,
+    cxt_decision_V240614,
+    cxt_decision_V240613,
 )
 
 
@@ -116,6 +118,7 @@
     bar_break_V240428,
     bar_classify_V240606,
     bar_classify_V240607,
+    bar_decision_V240608,
 )
 
 from czsc.signals.jcc import (
@@ -227,6 +230,7 @@
     pos_take_V240428,
     pos_stop_V240331,
     pos_stop_V240608,
+    pos_stop_V240614,
 )
 
 

czsc/signals/bar.py

@@ -10,6 +10,7 @@
 from datetime import datetime
 from typing import List
 from loguru import logger
+from copy import deepcopy
 from deprecated import deprecated
 from collections import OrderedDict
 from czsc import envs, CZSC, Signal
@@ -2052,3 +2053,59 @@ def bar_classify_V240607(c: CZSC, **kwargs) -> OrderedDict:
         v1 = "中性"
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def bar_decision_V240608(c: CZSC, **kwargs) -> OrderedDict:
+    """W窗口内最近N根K线放量后进行决策，如果是多头放量，开空，如果是空头放量，开多
+
+    参数模板："{freq}_W{w}N{n}Q{q}放量_决策区域V240608"
+
+    **信号逻辑：**
+
+    1. 看多：第二根K线收盘价在第一根K线的最高价上方
+    2. 看空：第二根K线收盘价在第一根K线的最低价下方
+    3. 中性：其他情况
+
+    **信号列表：**
+
+    - Signal('60分钟_W300N20Q70放量_决策区域V240608_看空_任意_任意_0')
+    - Signal('60分钟_W300N20Q70放量_决策区域V240608_放量_任意_任意_0')
+    - Signal('60分钟_W300N20Q70放量_决策区域V240608_看多_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+
+        - w: int, default 300, 窗口大小
+        - n: int, default 20, 最近N根K线
+        - q: int, default 80, 分位数，取值范围 0-100，表示取最近 w 根K线的成交量的 q 分位数
+
+    :return: 信号识别结果
+    """
+    w = int(kwargs.get("w", 300))
+    n = int(kwargs.get("n", 10))
+    q = int(kwargs.get("q", 80))  # 分位数，取值范围 0-100，表示取最近 w 根K线的成交量的 q 分位数
+    assert w > n > 3, "参数 w 必须大于 n，且 n 必须大于 0"
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_W{w}N{n}Q{q}放量_决策区域V240608".split("_")
+    v1 = "其他"
+    if len(c.bars_raw) < w + n:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    w_bars = get_sub_elements(c.bars_raw, di=1, n=w)
+    n_bars = get_sub_elements(c.bars_raw, di=1, n=n)
+    n_bars = deepcopy(n_bars)
+    n_diff = n_bars[-1].close - n_bars[0].open
+
+    # 找出 n_bars 中成交量最大的3根K线
+    n_bars.sort(key=lambda x: x.vol, reverse=True)
+    n_bars = n_bars[:3]
+
+    # 计算 w_bars 中成交量的 q 分位数
+    qth = np.quantile([x.vol for x in w_bars], q / 100)
+    vol_match = all([x.vol > qth for x in n_bars])
+    if vol_match and n_diff > 0:
+        v1 = "看空"
+    if vol_match and n_diff < 0:
+        v1 = "看多"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)

czsc/signals/cxt.py

@@ -2316,37 +2316,38 @@ def cxt_second_bs_V240524(c: CZSC, **kwargs) -> OrderedDict:
     :return: 信号识别结果
     """
     di = int(kwargs.get("di", 1))
-    w = int(kwargs.get("w", 15))  # 中枢窗口
+    w = int(kwargs.get("w", 9))  # 中枢窗口
     t = int(kwargs.get("t", 2))  # 重合次数
     assert w > 5, "参数 w 必须大于5"
     assert t >= 2, "参数 t 必须大于等于2"
 
     freq = c.freq.value
     k1, k2, k3 = f"{freq}_D{di}W{w}T{t}_第二买卖点V240524".split("_")
     v1 = "其他"
-    if len(c.bi_list) < w + di + 5:
+    if len(c.bi_list) < w + di + 5 or len(c.bars_ubi) > 7:
         return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
 
     bis = get_sub_elements(c.bi_list, di=di, n=w)
     last: BI = bis[-1]
     last_fx_high = last.fx_b.high
     last_fx_low = last.fx_b.low
+    fxs = [x.fx_b for x in bis[:-1] if x.length >= 7]
 
     if last.direction.value == "向下" and last.length >= 7:
         zs_count = 0
-        for bi in bis[:-1]:
+        for fx in fxs:
             # bi 的结束分型区间与 last 的分型区间有重合
-            if bi.length >= 7 and max(bi.fx_b.low, last_fx_low) < min(bi.fx_b.high, last_fx_high):
+            if max(fx.low, last_fx_low) < min(fx.high, last_fx_high):
                 zs_count += 1
 
         if zs_count >= t:
             v1 = "二买"
 
     if last.direction.value == "向上" and last.length >= 7:
         zs_count = 0
-        for bi in bis[:-1]:
+        for fx in fxs:
             # bi 的结束分型区间与 last 的分型区间有重合
-            if bi.length >= 7 and max(bi.fx_b.low, last_fx_low) < min(bi.fx_b.high, last_fx_high):
+            if max(fx.low, last_fx_low) < min(fx.high, last_fx_high):
                 zs_count += 1
 
         if zs_count >= t:
@@ -2694,3 +2695,85 @@ def cxt_decision_V240612(c: CZSC, **kwargs) -> OrderedDict:
         v1 = "开空"
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def cxt_decision_V240613(c: CZSC, **kwargs) -> OrderedDict:
+    """取最近N笔，如果当前笔向下未新低，且累计成交量最大，开多；如果当前笔向上未新高，且累计成交量最大，开空
+
+    参数模板："{freq}_放量笔N{n}BS2_决策区域V240613"
+
+    **信号逻辑：**
+
+    取最近N笔，如果当前笔向下未新低，且累计成交量最大，开多；如果当前笔向上未新高，且累计成交量最大，开空
+
+    相比于 cxt_decision_V240614，区别就是有没有新高新低的判断
+
+    **信号列表：**
+
+    - Signal('15分钟_放量笔N4BS2_决策区域V240613_开多_任意_任意_0')
+    - Signal('15分钟_放量笔N4BS2_决策区域V240613_开空_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    n = int(kwargs.get("n", 4))
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_放量笔N{n}BS2_决策区域V240613".split("_")
+    v1 = "其他"
+    if len(c.bi_list) < n + 2 or len(c.bars_ubi) > 7:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bis = get_sub_elements(c.bi_list, di=1, n=n)
+    bis_vol = [x.power_volume for x in bis]
+    if bis[-1].power_volume != max(bis_vol):
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    if bis[-1].direction == Direction.Down and bis[-1].low != min([x.low for x in bis]):
+        v1 = "开多"
+
+    if bis[-1].direction == Direction.Up and bis[-1].high != max([x.high for x in bis]):
+        v1 = "开空"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def cxt_decision_V240614(c: CZSC, **kwargs) -> OrderedDict:
+    """取最近N笔，如果当前笔向下创出新低，且累计成交量最大，开多；如果当前笔向上创出新高，且累计成交量最大，开空
+
+    参数模板："{freq}_放量笔N{n}_决策区域V240614"
+
+    **信号逻辑：**
+
+    取最近N笔，如果当前笔向下创出新低，且累计成交量最大，开多；如果当前笔向上创出新高，且累计成交量最大，开空
+
+    **信号列表：**
+
+    - Signal('15分钟_放量笔N4_决策区域V240614_开多_任意_任意_0')
+    - Signal('15分钟_放量笔N4_决策区域V240614_开空_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    n = int(kwargs.get("n", 4))
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_放量笔N{n}_决策区域V240614".split("_")
+    v1 = "其他"
+    if len(c.bi_list) < n + 2 or len(c.bars_ubi) > 7:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bis = get_sub_elements(c.bi_list, di=1, n=n)
+    bis_vol = [x.power_volume for x in bis]
+    if bis[-1].power_volume != max(bis_vol):
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    if bis[-1].direction == Direction.Down and bis[-1].low == min([x.low for x in bis]):
+        v1 = "开多"
+
+    if bis[-1].direction == Direction.Up and bis[-1].high == max([x.high for x in bis]):
+        v1 = "开空"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)

czsc/signals/pos.py

@@ -882,3 +882,62 @@ def pos_stop_V240608(cat: CzscTrader, **kwargs) -> OrderedDict:
             v1 = "空头止损"
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def pos_stop_V240614(cat: CzscTrader, **kwargs) -> OrderedDict:
+    """止损：多头开仓后，有超过N根K线的最低价在成本价下方，提示止损；空头反之。
+
+    参数模板："{pos_name}_{freq1}N{n}_止损V240614"
+
+    **信号逻辑：**
+
+    以多头止损为例，计算过程如下：
+
+    1. 从多头开仓点开始，在给定的K线周期 freq1 上获取开仓后的所有K线，记为 bars；
+    2. 计算 bars 中的最低价小于开仓价的数量，记为 C；
+    3. 如果 C >= N，则提示多头止损信号。
+
+    空头止损逻辑同理。
+
+    **信号列表：**
+
+    - Signal('SMA5多头_15分钟N10_止损V240614_多头止损_任意_任意_0')
+    - Signal('SMA5空头_15分钟N10_止损V240614_空头止损_任意_任意_0')
+
+    :param cat: CzscTrader对象
+    :param kwargs: 参数字典
+
+        - pos_name: str，开仓信号的名称
+        - freq1: str，给定的K线周期
+        - n: int，最低价下方N个价位，默认为 10
+
+    :return: OrderedDict
+    """
+    pos_name = kwargs["pos_name"]
+    freq1 = kwargs["freq1"]
+    n = int(kwargs.get("n", 10))  # N根K线
+
+    k1, k2, k3 = f"{pos_name}_{freq1}N{n}_止损V240614".split("_")
+    v1 = "其他"
+
+    # 如果没有持仓策略，则不产生信号
+    if not cat.kas or not hasattr(cat, "positions"):
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    pos = [x for x in cat.positions if x.name == pos_name][0]
+    if len(pos.operates) == 0 or pos.operates[-1]["op"] in [Operate.SE, Operate.LE]:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    c = cat.kas[freq1]
+    op = pos.operates[-1]
+
+    # 开仓后的K线
+    a_bars = [x for x in c.bars_raw if x.dt >= op["dt"]]
+
+    if op["op"] == Operate.LO and len([x for x in a_bars if x.low < op["price"]]) >= n:
+        v1 = "多头止损"
+
+    if op["op"] == Operate.SO and len([x for x in a_bars if x.high > op["price"]]) >= n:
+        v1 = "空头止损"
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)