0.9.26 更新一批信号函数

czsc/signals/__init__.py

@@ -66,6 +66,7 @@
     vol_ti_suo_V221216,
     vol_gao_di_V221218,
     vol_window_V230731,
+    vol_window_V230801,
 )
 
 from czsc.signals.bar import (
@@ -202,6 +203,8 @@
     pos_bar_stop_V230524,
     pos_fix_exit_V230624,
     pos_profit_loss_V230624,
+    pos_status_V230808,
+    pos_holds_V230807,
 )
 
 

czsc/signals/pos.py

@@ -355,3 +355,106 @@ def pos_profit_loss_V230624(cat: CzscTrader, **kwargs) -> OrderedDict:
                 v1 = '空头止损'
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def pos_status_V230808(cat: CzscTrader, **kwargs) -> OrderedDict:
+    """Position策略的持仓状态
+
+    参数模板："{pos_name}_持仓状态_BS辅助V230808"
+
+    **信号逻辑：**
+
+    对指定的持仓策略，有三种状态：持多、持空、持币。
+
+    **信号列表：**
+
+    - Signal('日线三买多头N1_持仓状态_BS辅助V230808_持多_任意_任意_0')
+    - Signal('日线三买多头N1_持仓状态_BS辅助V230808_持空_任意_任意_0')
+    - Signal('日线三买多头N1_持仓状态_BS辅助V230808_持币_任意_任意_0')
+
+    :param cat: CzscTrader对象
+    :param kwargs: 参数字典
+        - pos_name: str，开仓信号的名称
+    :return:
+    """
+    pos_name = kwargs["pos_name"]
+    k1, k2, k3 = f"{pos_name}_持仓状态_BS辅助V230808".split("_")
+    # 如果没有持仓策略，则不产生信号
+    if not hasattr(cat, "positions"):
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1="其他")
+
+    pos = cat.get_position(pos_name)
+    v1 = '持币'
+    if len(pos.operates) == 0:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    op = pos.operates[-1]
+    if op['op'] == Operate.LO:
+        v1 = '持多'
+    if op['op'] == Operate.SO:
+        v1 = '持空'
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def pos_holds_V230807(cat: CzscTrader, **kwargs) -> OrderedDict:
+    """开仓后N根K线收益小于M%%，且当前收益大于T%%，平仓保本
+
+    参数模板："{pos_name}_{freq1}N{n}M{m}T{t}_BS辅助V230807"
+
+    **信号逻辑：**
+
+    1. 针对某个开仓点，如果 N 根K线之后，收益低于 M，则认为开仓失误；
+    2. 开仓的失误发生后，如果市场给了逃命的机会，不贪心，等待收益大于 T 时，平仓保本；
+    3. 保本有两种场景：开仓后先亏损后反弹；开仓后先盈利后回落。
+
+    **信号列表：**
+
+    - Signal('日线三买多头N1_60分钟N5M50T10_BS辅助V230807_多头保本_任意_任意_0')
+    - Signal('日线三买多头N1_60分钟N5M50T10_BS辅助V230807_空头保本_任意_任意_0')
+
+    :param cat: CzscTrader对象
+    :param kwargs: 参数字典
+
+        - pos_name: str，开仓信号的名称
+        - freq1: str，给定的K线周期
+        - n: int，最少持有K线数量，默认为 5，表示5根K线之后开始判断
+        - m: int，收益最小阈值，默认为 100，表示收益小于 100BP 时，需要开始判断保本单
+        - t: int，保本收益阈值，默认为 10，表示收益大于 10BP 时，可以平仓保本
+
+    :return:
+    """
+    pos_name = kwargs["pos_name"]
+    freq1 = kwargs["freq1"]
+    n = int(kwargs.get('n', 5))
+    m = int(kwargs.get('m', 50))
+    t = int(kwargs.get('t', 10))
+    assert m > t > 0, "参数 m 必须大于 t"
+    k1, k2, k3 = f"{pos_name}_{freq1}N{n}M{m}T{t}_BS辅助V230807".split("_")
+    v1 = '其他'
+    # 如果没有持仓策略，则不产生信号
+    if 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]
+    bars = [x for x in c.bars_raw[-100:] if x.dt > op['dt']]
+    if len(bars) < n:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    if op['op'] == Operate.LO:
+        zdf = (bars[-1].close - op['price']) / op['price'] * 10000
+        if t < zdf < m:
+            v1 = '多头保本'
+
+    if op['op'] == Operate.SO:
+        zdf = (op['price'] - bars[-1].close) / op['price'] * 10000
+        if t < zdf < m:
+            v1 = '空头保本'
+
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+

czsc/signals/vol.py

@@ -325,3 +325,39 @@ def vol_window_V230731(c: CZSC, **kwargs) -> OrderedDict:
 
     v1, v2 = f"高量N{max_vol_layer}", f"低量N{min_vol_layer}"
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
+
+
+def vol_window_V230801(c: CZSC, **kwargs) -> OrderedDict:
+    """指定窗口内成交量的特征
+
+    参数模板："{freq}_D{di}W{w}_窗口能量V230801"
+
+    **信号逻辑：**
+
+    观察一个固定窗口内的成交量特征，本信号以窗口内的最大成交量与最小成交量的先后顺序作为窗口成交量的特征。
+
+    **信号列表：**
+
+    - Signal('60分钟_D1W5_窗口能量V230801_先缩后放_任意_任意_0')
+    - Signal('60分钟_D1W5_窗口能量V230801_先放后缩_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 参数字典
+    
+        - :param di: 信号计算截止倒数第i根K线
+        - :param w: 观察的窗口大小。
+        
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    w = int(kwargs.get("w", 5))
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}W{w}_窗口能量V230801".split('_')
+    if len(c.bars_raw) < di + w:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1="其他")
+
+    vols = [x.vol for x in get_sub_elements(c.bars_raw, di=di, n=w)]
+    min_i, max_i = vols.index(min(vols)), vols.index(max(vols))
+    v1 = "先放后缩" if min_i > max_i else "先缩后放"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)

examples/signals_dev/pos_holds_V230807.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""
+author: zengbin93
+email: [email protected]
+create_dt: 2023/4/14 17:48
+describe: 
+"""
+import os
+import sys
+sys.path.insert(0, r'D:\ZB\git_repo\waditu\czsc')
+os.environ['czsc_verbose'] = '1'
+import pandas as pd
+from typing import List
+from loguru import logger
+from czsc import CzscStrategyBase, Position
+from czsc.connectors import research
+logger.enable('czsc.analyze')
+
+pd.set_option('expand_frame_repr', False)
+pd.set_option('display.max_rows', 1000)
+pd.set_option('display.max_columns', 1000)
+pd.set_option('display.width', 1000)
+
+
+class MyStrategy(CzscStrategyBase):
+
+    def create_pos(self, freq='60分钟', freq1='15分钟'):
+        _pos_name = f'{freq}通道突破'
+        _pos = {'symbol': self.symbol,
+                'name': _pos_name,
+                'opens': [{'operate': '开多',
+                           'factors': [
+                               {'name': f'{freq}看多',
+                                'signals_all': [f'{freq}_D1通道突破#5#30#30_BS辅助V230403_看多_任意_任意_0',
+                                                f'{_pos_name}_持仓状态_BS辅助V230808_持币_任意_任意_0']},
+                           ]},
+
+                          {'operate': '开空',
+                           'factors': [
+                               {'name': f'{freq}看空',
+                                'signals_all': [f'{freq}_D1通道突破#5#30#30_BS辅助V230403_看空_任意_任意_0',
+                                                f'{_pos_name}_持仓状态_BS辅助V230808_持币_任意_任意_0']},
+                           ]}
+                          ],
+                'exits': [
+                    {'operate': '平多',
+                     'factors': [
+                         {'name': f'{freq1}_{_pos_name}_止损V230414',
+                          'signals_all': [f'{_pos_name}_60分钟N5M50T10_BS辅助V230807_多头保本_任意_任意_0']},
+                     ]},
+
+                    {'operate': '平空',
+                     'factors': [
+                         {'name': f'{freq1}_{_pos_name}_止损V230414',
+                          'signals_all': [f'{_pos_name}_60分钟N5M50T10_BS辅助V230807_空头保本_任意_任意_0']},
+                     ]},
+                ],
+                'interval': 7200,
+                'timeout': 100,
+                'stop_loss': 500,
+                'T0': True}
+
+        return Position.load(_pos)
+
+    @property
+    def positions(self) -> List[Position]:
+        _pos_list = [self.create_pos(freq='日线', freq1='60分钟')]
+        return _pos_list
+
+
+def check():
+    from czsc.connectors import research
+
+    symbols = research.get_symbols('A股主要指数')
+    tactic = MyStrategy(symbol=symbols[0], signals_module_name='czsc.signals')
+    bars = research.get_raw_bars(symbols[0], tactic.base_freq, '20151101', '20210101', fq='前复权')
+
+    tactic.check(bars, res_path=r'C:\Users\zengb\.czsc\策略信号验证', refresh=True, sdt='20170101')
+
+
+if __name__ == '__main__':
+    check()
+
+
+

examples/signals_dev/pos_status_V230808.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""
+author: zengbin93
+email: [email protected]
+create_dt: 2023/4/14 17:48
+describe: 
+"""
+import os
+import sys
+sys.path.insert(0, r'D:\ZB\git_repo\waditu\czsc')
+os.environ['czsc_verbose'] = '1'
+import pandas as pd
+from typing import List
+from loguru import logger
+from czsc import CzscStrategyBase, Position
+from czsc.connectors import research
+logger.enable('czsc.analyze')
+
+pd.set_option('expand_frame_repr', False)
+pd.set_option('display.max_rows', 1000)
+pd.set_option('display.max_columns', 1000)
+pd.set_option('display.width', 1000)
+
+
+class MyStrategy(CzscStrategyBase):
+
+    def create_pos(self, freq='60分钟', freq1='15分钟'):
+        _pos_name = f'{freq}通道突破'
+        _pos = {'symbol': self.symbol,
+                'name': _pos_name,
+                'opens': [{'operate': '开多',
+                           'factors': [
+                               {'name': f'{freq}看多',
+                                'signals_all': [f'{freq}_D1通道突破#5#30#30_BS辅助V230403_看多_任意_任意_0',
+                                                f'{_pos_name}_持仓状态_BS辅助V230808_持币_任意_任意_0']},
+                           ]},
+
+                          {'operate': '开空',
+                           'factors': [
+                               {'name': f'{freq}看空',
+                                'signals_all': [f'{freq}_D1通道突破#5#30#30_BS辅助V230403_看空_任意_任意_0',
+                                                f'{_pos_name}_持仓状态_BS辅助V230808_持币_任意_任意_0']},
+                           ]}
+                          ],
+                'exits': [
+                    {'operate': '平多',
+                     'factors': [
+                         {'name': f'{freq1}_{_pos_name}_止损V230414',
+                          'signals_all': [f'{freq1}_{_pos_name}N1_止损V230414_多头止损_任意_任意_0']},
+                     ]},
+
+                    {'operate': '平空',
+                     'factors': [
+                         {'name': f'{freq1}_{_pos_name}_止损V230414',
+                          'signals_all': [f'{freq1}_{_pos_name}N1_止损V230414_空头止损_任意_任意_0']},
+                     ]},
+                ],
+                'interval': 7200,
+                'timeout': 100,
+                'stop_loss': 500,
+                'T0': True}
+
+        return Position.load(_pos)
+
+    @property
+    def positions(self) -> List[Position]:
+        _pos_list = [self.create_pos(freq='日线', freq1='60分钟')]
+        return _pos_list
+
+
+def check():
+    from czsc.connectors import research
+
+    symbols = research.get_symbols('A股主要指数')
+    tactic = MyStrategy(symbol=symbols[0], signals_module_name='czsc.signals')
+    bars = research.get_raw_bars(symbols[0], tactic.base_freq, '20151101', '20210101', fq='前复权')
+
+    tactic.check(bars, res_path=r'C:\Users\zengb\.czsc\策略信号验证', refresh=True)
+
+
+if __name__ == '__main__':
+    check()
+
+
+

examples/signals_dev/signal_match.py

@@ -44,7 +44,7 @@
     conf = sp.parse(signals_seq)
     parsed_name = {x['name'] for x in conf}
     print(f"total signal functions: {len(sp.sig_name_map)}; parsed: {len(parsed_name)}")
-    # total signal functions: 179; parsed: 179
+    # total signal functions: 182; parsed: 182
 
     # 测试信号配置生成信号
     from czsc import generate_czsc_signals, get_signals_freqs, get_signals_config