0.9.27 新增几个信号函数

waditu · Aug 17, 2023 · 186e474 · 186e474
1 parent 687feef
commit 186e474
Show file tree

Hide file tree

Showing 6 changed files with 206 additions and 12 deletions.
diff --git a/czsc/signals/__init__.py b/czsc/signals/__init__.py
@@ -42,6 +42,8 @@
     cxt_range_oscillation_V230620,
     cxt_intraday_V230701,
     cxt_ubi_end_V230816,
+    cxt_bi_end_V230815,
+    cxt_bi_stop_V230815,
 )
 
 

diff --git a/czsc/signals/cxt.py b/czsc/signals/cxt.py
@@ -1972,3 +1972,75 @@ def cxt_ubi_end_V230816(c: CZSC, **kwargs) -> OrderedDict:
             v2 = f"第{cnt + 1}次"
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
+
+
+def cxt_bi_end_V230815(c: CZSC, **kwargs) -> OrderedDict:
+    """一两根K线快速突破反向笔
+
+    参数模板："{freq}_快速突破_BE辅助V230815"
+
+    **信号逻辑：**
+
+    以向上笔为例：右侧分型完成后第一根K线的最低价低于该笔的最低价，认为向上笔结束，反向向下笔开始。
+
+    **信号列表：**
+
+    - Signal('15分钟_快速突破_BE辅助V230815_向下_任意_任意_0')
+    - Signal('15分钟_快速突破_BE辅助V230815_向上_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+    :return: 信号识别结果
+    """
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_快速突破_BE辅助V230815".split('_')
+    v1 = '其他'
+    if len(c.bi_list) < 5 or len(c.bars_ubi) >= 5:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bi, last_bar = c.bi_list[-1], c.bars_ubi[-1]
+    if bi.direction == Direction.Up and last_bar.low < bi.low:
+        v1 = '向下'
+    if bi.direction == Direction.Down and last_bar.high > bi.high:
+        v1 = '向上'
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def cxt_bi_stop_V230815(c: CZSC, **kwargs) -> OrderedDict:
+    """定位笔的止损距离大小
+
+    参数模板："{freq}_距离{th}BP_止损V230815"
+
+    **信号逻辑：**
+
+    以向上笔为例：如果当前K线的收盘价高于该笔的最高价的1 - 0.5%，则认为在止损阈值内，否则认为在止损阈值外。
+
+    **信号列表：**
+
+    - Signal('15分钟_距离50BP_止损V230815_向下_阈值外_任意_0')
+    - Signal('15分钟_距离50BP_止损V230815_向上_阈值内_任意_0')
+    - Signal('15分钟_距离50BP_止损V230815_向下_阈值内_任意_0')
+    - Signal('15分钟_距离50BP_止损V230815_向上_阈值外_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 
+
+        - th: 止损距离阈值，单位为BP, 默认为50BP, 即0.5%
+
+    :return: 信号识别结果
+    """
+    th = int(kwargs.get('th', 50))
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_距离{th}BP_止损V230815".split('_')
+    v1, v2 = '其他', '其他'
+    if len(c.bi_list) < 5:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bi, last_bar = c.bi_list[-1], c.bars_ubi[-1]
+    if bi.direction == Direction.Up:
+        v1 = '向下'
+        v2 = "阈值内" if last_bar.close > bi.high * (1 - th / 10000) else "阈值外"
+    if bi.direction == Direction.Down:
+        v1 = '向上'
+        v2 = "阈值内" if last_bar.close < bi.low * (1 + th / 10000) else "阈值外"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
diff --git a/czsc/signals/tas.py b/czsc/signals/tas.py
@@ -3187,17 +3187,17 @@ def cat_macd_V230520(cat: CzscSignals, **kwargs) -> OrderedDict:
 def tas_angle_V230802(c: CZSC, **kwargs) -> OrderedDict:
     """笔的角度比较 贡献者：谌意勇
 
-    参数模板："{freq}_D{di}N{n}_笔角度V230802"
+    参数模板："{freq}_D{di}N{n}T{th}_笔角度V230802"
 
     **信号逻辑：**
 
     笔的角度，走过的笔的空间最高价和最低价的空间与走过的时间（原始K的数量）形成比值。
-    如果当前笔的角度小于前面N笔的平均角度，当前笔向上认为是空头笔，否则是多头笔。
+    如果当前笔的角度小于前面9笔的平均角度的50%，当前笔向上认为是空头笔，否则是多头笔。
 
     **信号列表：**
 
-    - Signal('60分钟_D1N9_笔角度V230802_多头_任意_任意_0')
-    - Signal('60分钟_D1N9_笔角度V230802_空头_任意_任意_0')
+    - Signal('60分钟_D1N9T50_笔角度V230802_空头_任意_任意_0')
+    - Signal('60分钟_D1N9T50_笔角度V230802_多头_任意_任意_0')
 
     :param c: CZSC对象
     :param kwargs:
@@ -3209,8 +3209,11 @@ def tas_angle_V230802(c: CZSC, **kwargs) -> OrderedDict:
     """
     di = int(kwargs.get('di', 1))
     n = int(kwargs.get('n', 9))
+    th = int(kwargs.get('th', 50))
+    assert 300 > th > 30, "th 取值范围为 30 ~ 300"
+
     freq = c.freq.value
-    k1, k2, k3 = f"{freq}_D{di}N{n}_笔角度V230802".split('_')
+    k1, k2, k3 = f"{freq}_D{di}N{n}T{th}_笔角度V230802".split('_')
     v1 = '其他'
     if len(c.bi_list) < di + 2 * n + 2 or len(c.bars_ubi) >= 7:
         return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
@@ -3220,7 +3223,7 @@ def tas_angle_V230802(c: CZSC, **kwargs) -> OrderedDict:
     b1_angle = b1.power_price / b1.length
     same_dir_ang = [bi.power_price / bi.length for bi in bis[:-1] if bi.direction == b1.direction][-n:]
 
-    if b1_angle < np.mean(same_dir_ang):
+    if b1_angle < np.mean(same_dir_ang) * th / 100:
         v1 = '空头' if b1.direction == Direction.Up else '多头'
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
 

diff --git a/examples/signals_dev/cxt_bi_end_V230815.py b/examples/signals_dev/cxt_bi_end_V230815.py
@@ -0,0 +1,53 @@
+import talib as ta
+import numpy as np
+from czsc import CZSC, Direction
+from collections import OrderedDict
+from czsc.objects import Mark
+from czsc.utils import create_single_signal, get_sub_elements
+
+
+def cxt_bi_end_V230815(c: CZSC, **kwargs) -> OrderedDict:
+    """一两根K线快速突破反向笔
+
+    参数模板："{freq}_快速突破_BE辅助V230815"
+
+    **信号逻辑：**
+
+    以向上笔为例：右侧分型完成后第一根K线的最低价低于该笔的最低价，认为向上笔结束，反向向下笔开始。
+
+    **信号列表：**
+
+    - Signal('15分钟_快速突破_BE辅助V230815_向下_任意_任意_0')
+    - Signal('15分钟_快速突破_BE辅助V230815_向上_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+    :return: 信号识别结果
+    """
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_快速突破_BE辅助V230815".split('_')
+    v1 = '其他'
+    if len(c.bi_list) < 5 or len(c.bars_ubi) >= 5:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bi, last_bar = c.bi_list[-1], c.bars_ubi[-1]
+    if bi.direction == Direction.Up and last_bar.low < bi.low:
+        v1 = '向下'
+    if bi.direction == Direction.Down and last_bar.high > bi.high:
+        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': cxt_bi_end_V230815, 'freq': '15分钟'}]
+    check_signals_acc(bars, signals_config=signals_config, height='780px') # type: ignore
+
+
+if __name__ == '__main__':
+    check()
diff --git a/examples/signals_dev/cxt_bi_stop_V230815.py b/examples/signals_dev/cxt_bi_stop_V230815.py
@@ -0,0 +1,61 @@
+import talib as ta
+import numpy as np
+from czsc import CZSC, Direction
+from collections import OrderedDict
+from czsc.objects import Mark
+from czsc.utils import create_single_signal, get_sub_elements
+
+
+def cxt_bi_stop_V230815(c: CZSC, **kwargs) -> OrderedDict:
+    """定位笔的止损距离大小
+
+    参数模板："{freq}_距离{th}BP_止损V230815"
+
+    **信号逻辑：**
+
+    以向上笔为例：如果当前K线的收盘价高于该笔的最高价的1 - 0.5%，则认为在止损阈值内，否则认为在止损阈值外。
+
+    **信号列表：**
+
+    - Signal('15分钟_距离50BP_止损V230815_向下_阈值外_任意_0')
+    - Signal('15分钟_距离50BP_止损V230815_向上_阈值内_任意_0')
+    - Signal('15分钟_距离50BP_止损V230815_向下_阈值内_任意_0')
+    - Signal('15分钟_距离50BP_止损V230815_向上_阈值外_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 
+
+        - th: 止损距离阈值，单位为BP, 默认为50BP, 即0.5%
+
+    :return: 信号识别结果
+    """
+    th = int(kwargs.get('th', 50))
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_距离{th}BP_止损V230815".split('_')
+    v1, v2 = '其他', '其他'
+    if len(c.bi_list) < 5:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bi, last_bar = c.bi_list[-1], c.bars_ubi[-1]
+    if bi.direction == Direction.Up:
+        v1 = '向下'
+        v2 = "阈值内" if last_bar.close > bi.high * (1 - th / 10000) else "阈值外"
+    if bi.direction == Direction.Down:
+        v1 = '向上'
+        v2 = "阈值内" if last_bar.close < bi.low * (1 + th / 10000) else "阈值外"
+    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': cxt_bi_stop_V230815, 'freq': '15分钟'}]
+    check_signals_acc(bars, signals_config=signals_config, height='780px') # type: ignore
+
+
+if __name__ == '__main__':
+    check()
diff --git a/examples/signals_dev/merged/tas_angle_V230802.py b/examples/signals_dev/merged/tas_angle_V230802.py
@@ -21,17 +21,17 @@
 def tas_angle_V230802(c: CZSC, **kwargs) -> OrderedDict:
     """笔的角度比较 贡献者：谌意勇
 
-    参数模板："{freq}_D{di}N{n}_笔角度V230802"
+    参数模板："{freq}_D{di}N{n}T{th}_笔角度V230802"
 
     **信号逻辑：**
 
     笔的角度，走过的笔的空间最高价和最低价的空间与走过的时间（原始K的数量）形成比值。
-    如果当前笔的角度小于前面N笔的平均角度，当前笔向上认为是空头笔，否则是多头笔。
+    如果当前笔的角度小于前面9笔的平均角度的50%，当前笔向上认为是空头笔，否则是多头笔。
 
     **信号列表：**
 
-    - Signal('60分钟_D1N9_笔角度V230802_多头_任意_任意_0')
-    - Signal('60分钟_D1N9_笔角度V230802_空头_任意_任意_0')
+    - Signal('60分钟_D1N9T50_笔角度V230802_空头_任意_任意_0')
+    - Signal('60分钟_D1N9T50_笔角度V230802_多头_任意_任意_0')
 
     :param c: CZSC对象
     :param kwargs:
@@ -43,8 +43,11 @@ def tas_angle_V230802(c: CZSC, **kwargs) -> OrderedDict:
     """
     di = int(kwargs.get('di', 1))
     n = int(kwargs.get('n', 9))
+    th = int(kwargs.get('th', 50))
+    assert 300 > th > 30, "th 取值范围为 30 ~ 300"
+
     freq = c.freq.value
-    k1, k2, k3 = f"{freq}_D{di}N{n}_笔角度V230802".split('_')
+    k1, k2, k3 = f"{freq}_D{di}N{n}T{th}_笔角度V230802".split('_')
     v1 = '其他'
     if len(c.bi_list) < di + 2 * n + 2 or len(c.bars_ubi) >= 7:
         return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
@@ -54,7 +57,7 @@ def tas_angle_V230802(c: CZSC, **kwargs) -> OrderedDict:
     b1_angle = b1.power_price / b1.length
     same_dir_ang = [bi.power_price / bi.length for bi in bis[:-1] if bi.direction == b1.direction][-n:]
 
-    if b1_angle < np.mean(same_dir_ang):
+    if b1_angle < np.mean(same_dir_ang) * th / 100:
         v1 = '空头' if b1.direction == Direction.Up else '多头'
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)