0.8.28 更新信号案例 (waditu#65)

czsc/signals/__init__.py

@@ -11,4 +11,6 @@
 from . import vol
 from . import cat
 from . import pos
+from . import example
+
 

czsc/signals/example.py

@@ -0,0 +1,169 @@
+# -*- coding: utf-8 -*-
+"""
+author: zengbin93
+email: [email protected]
+create_dt: 2022/7/3 19:07
+describe: 信号计算案例汇总
+"""
+
+import numpy as np
+import talib as ta
+from collections import OrderedDict
+from typing import List, Union, Tuple, Dict
+from czsc.objects import Freq, Signal, RawBar, NewBar
+from czsc.traders.advanced import CzscAdvancedTrader
+from czsc.signals.utils import check_cross_info
+
+
+def update_sma_cache(cat: CzscAdvancedTrader, freq: str,
+                     sma_params: Tuple = (5, 13, 21, 34, 55, 89, 144, 233)):
+    """更新某个级别的均线缓存
+
+    :param cat: 交易对象
+    :param freq: 指定周期
+    :param sma_params: 均线参数
+    :return:
+    """
+    assert freq in cat.freqs, f"{freq} 不在 {cat.freqs} 中"
+    cache_key = f"{freq}均线"
+    sma_cache = cat.cache.get(cache_key, {})
+    sma_cache['update_dt'] = cat.end_dt
+    close = np.array([x.close for x in cat.kas[freq].bars_raw])
+    sma_cache['close'] = close
+    for t in sma_params:
+        sma_cache[f"SMA{t}"] = ta.SMA(close, timeperiod=t)
+    cat.cache[cache_key] = sma_cache
+
+
+def update_macd_cache(cat: CzscAdvancedTrader, freq: str):
+    """更新某个级别的均线缓存
+
+    :param cat: 交易对象
+    :param freq: 指定周期
+    :return:
+    """
+    assert freq in cat.freqs, f"{freq} 不在 {cat.freqs} 中"
+    cache_key = f"{freq}MACD"
+    cache = cat.cache.get(cache_key, {})
+    cache['update_dt'] = cat.end_dt
+    close = np.array([x.close for x in cat.kas[freq].bars_raw])
+    cache['close'] = close
+
+    dif, dea, macd = ta.MACD(close, fastperiod=12, slowperiod=26, signalperiod=9)
+    cross = check_cross_info(dif, dea)
+    cache.update({"dif": dif, 'dea': dea, 'macd': macd, 'cross': cross})
+    cat.cache[cache_key] = cache
+
+
+def update_boll_cache(cat: CzscAdvancedTrader, freq: str):
+    """更新某个级别的均线缓存
+
+    :param cat: 交易对象
+    :param freq: 指定周期
+    :return:
+    """
+    assert freq in cat.freqs, f"{freq} 不在 {cat.freqs} 中"
+    cache_key = f"{freq}BOLL"
+    cache = cat.cache.get(cache_key, {})
+    cache['update_dt'] = cat.end_dt
+    close = np.array([x.close for x in cat.kas[freq].bars_raw])
+    cache['close'] = close
+
+    u1, m, l1 = ta.BBANDS(close, timeperiod=20, nbdevup=1.382, nbdevdn=1.382, matype=ta.MA_Type.SMA)
+    u2, m, l2 = ta.BBANDS(close, timeperiod=20, nbdevup=2, nbdevdn=2, matype=ta.MA_Type.SMA)
+    u3, m, l3 = ta.BBANDS(close, timeperiod=20, nbdevup=2.764, nbdevdn=2.764, matype=ta.MA_Type.SMA)
+
+    cache.update({"上轨3": u3, "上轨2": u2, "上轨1": u1, "中线": m, "下轨1": l1, "下轨2": l2, "下轨3": l3})
+    cat.cache[cache_key] = cache
+
+
+def single_sma(cat: CzscAdvancedTrader, freq: str, t_seq=(5, 13, 21)) -> OrderedDict:
+    """单均线相关信号
+
+    完全分类：
+        Signal('日线_倒1K_SMA5_多头_向上_任意_0'),
+        Signal('日线_倒1K_SMA5_空头_向下_任意_0'),
+        Signal('日线_倒1K_SMA5_多头_向下_任意_0'),
+        Signal('日线_倒1K_SMA5_空头_向上_任意_0'),
+
+        Signal('日线_倒1K_SMA13_空头_向下_任意_0'),
+        Signal('日线_倒1K_SMA13_空头_向上_任意_0'),
+        Signal('日线_倒1K_SMA13_多头_向上_任意_0'),
+        Signal('日线_倒1K_SMA13_多头_向下_任意_0'),
+
+        Signal('日线_倒1K_SMA21_空头_向下_任意_0'),
+        Signal('日线_倒1K_SMA21_多头_向下_任意_0'),
+        Signal('日线_倒1K_SMA21_多头_向上_任意_0'),
+        Signal('日线_倒1K_SMA21_空头_向上_任意_0')
+    :param cat:
+    :param freq:
+    :param t_seq:
+    :return:
+    """
+    assert freq in cat.freqs, f"{freq} 不在 {cat.freqs} 中"
+    cache_key = f"{freq}均线"
+    sma_cache = cat.cache[cache_key]
+    assert sma_cache and sma_cache['update_dt'] == cat.end_dt
+
+    s = OrderedDict()
+    k1 = freq
+    k2 = "倒1K"
+    close = sma_cache['close']
+    for t in t_seq:
+        sma = sma_cache[f"SMA{t}"]
+        if len(sma) == 0:
+            v1, v2 = '其他', '其他'
+        else:
+            v1 = "多头" if close[-1] >= sma[-1] else "空头"
+            v2 = "向上" if sma[-1] >= sma[-2] else "向下"
+
+        x1 = Signal(k1=k1, k2=k2, k3=f"SMA{t}", v1=v1, v2=v2)
+        s[x1.key] = x1.value
+    return s
+
+
+def macd_base(cat: CzscAdvancedTrader, freq: str):
+    """MACD柱子信号
+
+    完全分类：
+        Signal('日线_倒1K_MACD_空头_向下_任意_0'),
+        Signal('日线_倒1K_MACD_多头_向下_任意_0'),
+        Signal('日线_倒1K_MACD_空头_向上_任意_0'),
+        Signal('日线_倒1K_MACD_多头_向上_任意_0'),
+
+        Signal('日线_倒1K_MACD强弱_强势_任意_任意_0'),
+        Signal('日线_倒1K_MACD强弱_超弱_任意_任意_0'),
+        Signal('日线_倒1K_MACD强弱_超强_任意_任意_0'),
+        Signal('日线_倒1K_MACD强弱_弱势_任意_任意_0')
+    :return:
+    """
+    s = OrderedDict()
+    cache_key = f"{freq}MACD"
+    cache = cat.cache[cache_key]
+    assert cache and cache['update_dt'] == cat.end_dt
+    dif, dea, macd = cache['dif'], cache['dea'], cache['macd']
+
+    v1 = "多头" if macd[-1] >= 0 else "空头"
+    v2 = "向上" if macd[-1] >= macd[-2] else "向下"
+    signal = Signal(k1=freq, k2="倒1K", k3="MACD", v1=v1, v2=v2)
+    s[signal.key] = signal.value
+
+    # MACD强弱
+    if dif[-1] >= dea[-1] >= 0:
+        v1 = "超强"
+    elif dif[-1] - dea[-1] > 0:
+        v1 = "强势"
+    elif dif[-1] <= dea[-1] <= 0:
+        v1 = "超弱"
+    elif dif[-1] - dea[-1] < 0:
+        v1 = "弱势"
+    else:
+        v1 = "其他"
+
+    signal = Signal(k1=freq, k2="倒1K", k3="MACD强弱", v1=v1)
+    s[signal.key] = signal.value
+    return s
+
+
+
+

czsc/signals/utils.py

@@ -102,9 +102,13 @@ def check_cross_info(fast: [List, np.array], slow: [List, np.array]):
     cross_info = []
     last_i = -1
     last_v = 0
+    temp_fast = []
+    temp_slow = []
     for i, v in enumerate(delta):
         last_i += 1
         last_v += abs(v)
+        temp_fast.append(fast[i])
+        temp_slow.append(slow[i])
 
         if i >= 2 and delta[i-1] <= 0 < delta[i]:
             kind = "金叉"
@@ -114,9 +118,16 @@ def check_cross_info(fast: [List, np.array], slow: [List, np.array]):
             continue
 
         cross_info.append({'位置': i, "类型": kind, "快线": fast[i], "慢线": slow[i],
-                           "距离": last_i, "面积": last_v, '价差': v, '距今': length - i})
+                           "距离": last_i, '距今': length - i,
+                           "面积": round(last_v, 4), '价差': round(v, 4),
+                           "快线高点": max(temp_fast), "快线低点": min(temp_fast),
+                           "慢线高点": max(temp_slow), "慢线低点": min(temp_slow),
+                           })
         last_i = 0
         last_v = 0
+        temp_fast = []
+        temp_slow = []
+
     return cross_info
 
 

examples/gm_check_point.py

@@ -5,11 +5,30 @@
 create_dt: 2021/11/17 22:26
 describe: 使用掘金数据验证买卖点
 """
+# from czsc.gm_utils import trader_tactic_snapshot
+# from czsc.strategies import trader_strategy_a as strategy
+#
+# if __name__ == '__main__':
+#     _symbol = "SZSE.300669"
+#     ct = trader_tactic_snapshot(_symbol, end_dt="2022-03-18 13:15", strategy=strategy)
+
+import numpy as np
 from czsc.gm_utils import trader_tactic_snapshot
 from czsc.strategies import trader_strategy_a as strategy
+from czsc.signals.utils import check_cross_info
+try:
+    import talib as ta
+except:
+    from czsc.utils import ta
+
 
 if __name__ == '__main__':
-    _symbol = "SZSE.300669"
-    ct = trader_tactic_snapshot(_symbol, end_dt="2022-03-18 13:15", strategy=strategy)
+    _symbol = "SHSE.000001"
+    ct = trader_tactic_snapshot(_symbol, end_dt="2022-07-03 15:15", strategy=strategy)
+    close = np.array([x.close for x in ct.kas['60分钟'].bars_raw])
+    fast = ta.SMA(close, timeperiod=5)
+    slow = ta.SMA(close, timeperiod=13)
+    cross = check_cross_info(fast.round(2), slow.round(2))
+
 
 

examples/ts_check_signal_acc.py

@@ -21,22 +21,27 @@
 dc = TsDataCache(data_path, sdt='2010-01-01', edt='20211209')
 
 symbol = '000001.SZ'
-bars = dc.pro_bar_minutes(ts_code=symbol, asset='E', freq='5min',
+bars = dc.pro_bar_minutes(ts_code=symbol, asset='E', freq='15min',
                           sdt='20181101', edt='20210101', adj='qfq', raw_bar=True)
 
 def get_signals(cat: CzscAdvancedTrader) -> OrderedDict:
     s = OrderedDict({"symbol": cat.symbol, "dt": cat.end_dt, "close": cat.latest_price})
-    for _, c in cat.kas.items():
-        if c.freq == Freq.F5:
-            s.update(signals.bxt.get_s_like_bs(c, di=1))
+
+    # 使用缓存来更新信号的方法
+    signals.example.update_macd_cache(cat, '日线')
+    s.update(signals.example.macd_base(cat, '日线'))
+
+    # for _, c in cat.kas.items():
+    #     if c.freq == Freq.F5:
+    #         s.update(signals.bxt.get_s_like_bs(c, di=1))
     return s
 
 
 def trader_strategy_base(symbol):
     tactic = {
         "symbol": symbol,
-        "base_freq": '5分钟',
-        "freqs": ['15分钟', '30分钟', '60分钟', '日线'],
+        "base_freq": '15分钟',
+        "freqs": ['30分钟', '60分钟', '日线'],
         "get_signals": get_signals,
         "signals_n": 0,
     }