0.9.27 新增两个信号函数

waditu · Aug 15, 2023 · 34354ad · 34354ad
1 parent b772fa6
commit 34354ad
Show file tree

Hide file tree

Showing 6 changed files with 244 additions and 4 deletions.
diff --git a/czsc/analyze.py b/czsc/analyze.py
@@ -369,7 +369,8 @@ def ubi_fxs(self) -> List[FX]:
     @property
     def ubi(self):
         """Unfinished Bi，未完成的笔"""
-        if not self.bars_ubi or not self.bi_list:
+        ubi_fxs = self.ubi_fxs
+        if not self.bars_ubi or not self.bi_list or not ubi_fxs:
             return None
 
         bars_raw = [y for x in self.bars_ubi for y in x.raw_bars]
@@ -387,8 +388,8 @@ def ubi(self):
             "low_bar": low_bar,
             "bars": self.bars_ubi,
             "raw_bars": bars_raw,
-            "fxs": self.ubi_fxs,
-            "fx_a": self.ubi_fxs[0],
+            "fxs": ubi_fxs,
+            "fx_a": ubi_fxs[0],
         }
         return bi
 

diff --git a/czsc/signals/__init__.py b/czsc/signals/__init__.py
@@ -198,6 +198,8 @@
     cat_macd_V230518,
     cat_macd_V230520,
     tas_macd_bc_V230803,
+    tas_macd_bc_V230804,
+    tas_macd_bc_ubi_V230804,
 )
 
 from czsc.signals.pos import (

diff --git a/czsc/signals/tas.py b/czsc/signals/tas.py
@@ -19,7 +19,7 @@
 from collections import OrderedDict
 from deprecated import deprecated
 from czsc.analyze import CZSC
-from czsc.objects import Signal, Direction, BI, RawBar, FX, Mark
+from czsc.objects import Signal, Direction, BI, RawBar, FX, Mark, ZS
 from czsc.traders.base import CzscSignals
 from czsc.utils import get_sub_elements, fast_slow_cross, count_last_same, create_single_signal
 from czsc.utils.sig import cross_zero_axis, cal_cross_num, down_cross_count
@@ -3265,3 +3265,101 @@ def tas_macd_bc_V230803(c: CZSC, **kwargs) -> OrderedDict:
             v1 = '多头'
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def tas_macd_bc_V230804(c: CZSC, **kwargs) -> OrderedDict:
+    """MACD黄白线辅助背驰判断
+
+    参数模板："{freq}_D{di}MACD背驰_BS辅助V230804"
+
+    **信号逻辑：**
+
+    以向上笔为例，当前笔在中枢中轴上方，且MACD黄白线不是最高，认为是背驰，做空；反之，做多。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1MACD背驰_BS辅助V230804_空头_任意_任意_0')
+    - Signal('60分钟_D1MACD背驰_BS辅助V230804_多头_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get('di', 1))
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}MACD背驰_BS辅助V230804".split('_')
+    v1 = '其他'
+    cache_key = update_macd_cache(c)
+    if len(c.bi_list) < 7 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=7)
+    zs = ZS(bis=bis[-5:])
+    if not zs.is_valid:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    dd = min([bi.low for bi in bis])
+    gg = max([bi.high for bi in bis])
+    b1, b2, b3, b4, b5 = bis[-5:]
+    if b5.direction == Direction.Up and b5.high > (gg - (gg - dd) / 4):
+        b5_dif = max([x.cache[cache_key]['dif'] for x in b5.fx_b.raw_bars])
+        od_dif = max([x.cache[cache_key]['dif'] for x in b1.fx_b.raw_bars + b3.fx_b.raw_bars])
+        if 0 < b5_dif < od_dif:
+            v1 = '空头'
+
+    if b5.direction == Direction.Down and b5.low < (dd + (gg - dd) / 4):
+        b5_dif = min([x.cache[cache_key]['dif'] for x in b5.fx_b.raw_bars])
+        od_dif = min([x.cache[cache_key]['dif'] for x in b1.fx_b.raw_bars + b3.fx_b.raw_bars])
+        if 0 > b5_dif > od_dif:
+            v1 = '多头'
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def tas_macd_bc_ubi_V230804(c: CZSC, **kwargs) -> OrderedDict:
+    """未完成笔MACD黄白线辅助背驰判断
+
+    参数模板："{freq}_MACD背驰_BS辅助V230804"
+
+    **信号逻辑：**
+
+    以向上未完成笔为例，当前笔在中枢中轴上方，且MACD黄白线不是最高，认为是背驰，做空；反之，做多。
+
+    **信号列表：**
+
+    - Signal('60分钟_MACD背驰_UBI观察V230804_多头_任意_任意_0')
+    - Signal('60分钟_MACD背驰_UBI观察V230804_空头_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_MACD背驰_UBI观察V230804".split('_')
+    v1 = '其他'
+    cache_key = update_macd_cache(c)
+    ubi = c.ubi
+    if len(c.bi_list) < 7 or not ubi or len(ubi['raw_bars']) < 7:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bis = get_sub_elements(c.bi_list, di=1, n=6)
+    zs = ZS(bis=bis[-5:])
+    if not zs.is_valid:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    dd = min([bi.low for bi in bis])
+    gg = max([bi.high for bi in bis])
+    b1, b2, b3, b4, b5 = bis[-5:]
+    if ubi['direction'] == Direction.Up and ubi['high'] > (gg - (gg - dd) / 4):
+        b5_dif = max([x.cache[cache_key]['dif'] for x in ubi['raw_bars'][-5:]])
+        od_dif = max([x.cache[cache_key]['dif'] for x in b2.fx_b.raw_bars + b4.fx_b.raw_bars])
+        if 0 < b5_dif < od_dif:
+            v1 = '空头'
+
+    if ubi['direction'] == Direction.Down and ubi['low'] < (dd + (gg - dd) / 4):
+        b5_dif = min([x.cache[cache_key]['dif'] for x in ubi['raw_bars'][-5:]])
+        od_dif = min([x.cache[cache_key]['dif'] for x in b2.fx_b.raw_bars + b4.fx_b.raw_bars])
+        if 0 > b5_dif > od_dif:
+            v1 = '多头'
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
diff --git a/czsc/traders/base.py b/czsc/traders/base.py
@@ -249,6 +249,7 @@ def check_signals_acc(bars: List[RawBar], signals_config: List[dict], delta_days
     s_cols = [x for x in df.columns if len(x.split("_")) == 3]
     signals = []
     for col in s_cols:
+        print('=' * 100, "\n", df[col].value_counts())
         signals.extend([Signal(f"{col}_{v}") for v in df[col].unique() if "其他" not in v])
 
     print(f"signals: {'+' * 100}")

diff --git a/examples/signals_dev/tas_macd_bc_V230804.py b/examples/signals_dev/tas_macd_bc_V230804.py
@@ -0,0 +1,69 @@
+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 tas_macd_bc_V230804(c: CZSC, **kwargs) -> OrderedDict:
+    """MACD黄白线辅助背驰判断
+
+    参数模板："{freq}_D{di}MACD背驰_BS辅助V230804"
+
+    **信号逻辑：**
+
+    以向上笔为例，当前笔在中枢中轴上方，且MACD黄白线不是最高，认为是背驰，做空；反之，做多。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1MACD背驰_BS辅助V230804_空头_任意_任意_0')
+    - Signal('60分钟_D1MACD背驰_BS辅助V230804_多头_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get('di', 1))
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}MACD背驰_BS辅助V230804".split('_')
+    v1 = '其他'
+    cache_key = update_macd_cache(c)
+    if len(c.bi_list) < 7 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=7)
+    zs = ZS(bis=bis[-5:])
+    if not zs.is_valid:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    dd = min([bi.low for bi in bis])
+    gg = max([bi.high for bi in bis])
+    b1, b2, b3, b4, b5 = bis[-5:]
+    if b5.direction == Direction.Up and b5.high > (gg - (gg - dd) / 4):
+        b5_dif = max([x.cache[cache_key]['dif'] for x in b5.fx_b.raw_bars])
+        od_dif = max([x.cache[cache_key]['dif'] for x in b1.fx_b.raw_bars + b3.fx_b.raw_bars])
+        if 0 < b5_dif < od_dif:
+            v1 = '空头'
+
+    if b5.direction == Direction.Down and b5.low < (dd + (gg - dd) / 4):
+        b5_dif = min([x.cache[cache_key]['dif'] for x in b5.fx_b.raw_bars])
+        od_dif = min([x.cache[cache_key]['dif'] for x in b1.fx_b.raw_bars + b3.fx_b.raw_bars])
+        if 0 > b5_dif > od_dif:
+            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': tas_macd_bc_V230804, 'freq': "60分钟"}]
+    check_signals_acc(bars, signals_config=signals_config, height='780px', delta_days=5)  # type: ignore
+
+
+if __name__ == '__main__':
+    check()
diff --git a/examples/signals_dev/tas_macd_bc_ubi_V230804.py b/examples/signals_dev/tas_macd_bc_ubi_V230804.py
@@ -0,0 +1,69 @@
+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 tas_macd_bc_ubi_V230804(c: CZSC, **kwargs) -> OrderedDict:
+    """未完成笔MACD黄白线辅助背驰判断
+
+    参数模板："{freq}_MACD背驰_BS辅助V230804"
+
+    **信号逻辑：**
+
+    以向上未完成笔为例，当前笔在中枢中轴上方，且MACD黄白线不是最高，认为是背驰，做空；反之，做多。
+
+    **信号列表：**
+
+    - Signal('60分钟_MACD背驰_UBI观察V230804_多头_任意_任意_0')
+    - Signal('60分钟_MACD背驰_UBI观察V230804_空头_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 无
+    :return: 信号识别结果
+    """
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_MACD背驰_UBI观察V230804".split('_')
+    v1 = '其他'
+    cache_key = update_macd_cache(c)
+    ubi = c.ubi
+    if len(c.bi_list) < 7 or not ubi or len(ubi['raw_bars']) < 7:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bis = get_sub_elements(c.bi_list, di=1, n=6)
+    zs = ZS(bis=bis[-5:])
+    if not zs.is_valid:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    dd = min([bi.low for bi in bis])
+    gg = max([bi.high for bi in bis])
+    b1, b2, b3, b4, b5 = bis[-5:]
+    if ubi['direction'] == Direction.Up and ubi['high'] > (gg - (gg - dd) / 4):
+        b5_dif = max([x.cache[cache_key]['dif'] for x in ubi['raw_bars'][-5:]])
+        od_dif = max([x.cache[cache_key]['dif'] for x in b2.fx_b.raw_bars + b4.fx_b.raw_bars])
+        if 0 < b5_dif < od_dif:
+            v1 = '空头'
+
+    if ubi['direction'] == Direction.Down and ubi['low'] < (dd + (gg - dd) / 4):
+        b5_dif = min([x.cache[cache_key]['dif'] for x in ubi['raw_bars'][-5:]])
+        od_dif = min([x.cache[cache_key]['dif'] for x in b2.fx_b.raw_bars + b4.fx_b.raw_bars])
+        if 0 > b5_dif > od_dif:
+            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': tas_macd_bc_ubi_V230804, 'freq': "60分钟"}]
+    check_signals_acc(bars, signals_config=signals_config, height='780px', delta_days=5)  # type: ignore
+
+
+if __name__ == '__main__':
+    check()