diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml index a31c0639b..244050925 100644 --- a/.github/workflows/pythonpackage.yml +++ b/.github/workflows/pythonpackage.yml @@ -5,7 +5,7 @@ name: Python package on: push: - branches: [ master, V0.9.26 ] + branches: [ master, V0.9.27 ] pull_request: branches: [ master ] diff --git a/.vscode/settings.json b/.vscode/settings.json index 2e2231483..19186f932 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -17,4 +17,5 @@ "120" ], "typescript.locale": "zh-CN", + "python.analysis.typeCheckingMode": "basic", } \ No newline at end of file diff --git a/czsc/__init__.py b/czsc/__init__.py index b7b5c3d2b..186f8a94b 100644 --- a/czsc/__init__.py +++ b/czsc/__init__.py @@ -29,10 +29,10 @@ from czsc.utils.cache import home_path, get_dir_size, empty_cache_path -__version__ = "0.9.26" +__version__ = "0.9.27" __author__ = "zengbin93" __email__ = "zeng_bin8888@163.com" -__date__ = "20230726" +__date__ = "20230812" def welcome(): diff --git a/czsc/analyze.py b/czsc/analyze.py index 47bcad0ff..049bf7609 100644 --- 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 index 94d9ded73..41f10bcd5 100644 --- a/czsc/signals/__init__.py +++ b/czsc/signals/__init__.py @@ -41,6 +41,9 @@ cxt_eleven_bi_V230622, cxt_range_oscillation_V230620, cxt_intraday_V230701, + cxt_ubi_end_V230816, + cxt_bi_end_V230815, + cxt_bi_stop_V230815, ) @@ -198,6 +201,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/cxt.py b/czsc/signals/cxt.py index f2cf01476..aa6d52211 100644 --- a/czsc/signals/cxt.py +++ b/czsc/signals/cxt.py @@ -1863,7 +1863,7 @@ def cxt_intraday_V230701(cat: CzscSignals, **kwargs) -> OrderedDict: assert 21 > di > 0, "di必须为大于0小于21的整数，暂不支持当日走势分类" k1, k2, k3 = f"{freq1}#{freq2}_D{di}日_走势分类V230701".split('_') v1 = "其他" - if '30分钟' not in cat.kas.keys() or '日线' not in cat.kas.keys(): + if not cat.kas or freq1 not in cat.kas.keys() or freq2 not in cat.kas.keys(): return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1) c1, c2 = cat.kas[freq1], cat.kas[freq2] @@ -1889,11 +1889,11 @@ def cxt_intraday_V230701(cat: CzscSignals, **kwargs) -> OrderedDict: zs1, zs2 = zs_list[0], zs_list[-1] zs1_high, zs1_low = max([x.high for x in zs1]), min([x.low for x in zs1]) zs2_high, zs2_low = max([x.high for x in zs2]), min([x.low for x in zs2]) - if _dir == "上涨" and zs1_high < zs2_low: + if _dir == "上涨" and zs1_high < zs2_low: # type: ignore v1 = f"双中枢{_dir}" return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1) - if _dir == "下跌" and zs1_low > zs2_high: + if _dir == "下跌" and zs1_low > zs2_high: # type: ignore v1 = f"双中枢{_dir}" return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1) @@ -1908,3 +1908,139 @@ def cxt_intraday_V230701(cat: CzscSignals, **kwargs) -> OrderedDict: v1 = "转折平衡市" return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1) + + +def cxt_ubi_end_V230816(c: CZSC, **kwargs) -> OrderedDict: + """当前是未完成笔的第几次新低或新高，用于笔结束辅助 + + 参数模板："{freq}_UBI_BE辅助V230816" + + **信号逻辑：** + + 以向上未完成笔为例：取所有顶分型，计算创新高的底分型数量N，如果当前K线创新高，则新高次数为N+1 + + **信号列表：** + + - Signal('日线_UBI_BE辅助V230816_新低_第4次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第5次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第6次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第2次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第3次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第4次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第5次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第6次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第7次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第2次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第3次_任意_0') + + :param c: CZSC对象 + :param kwargs: + :return: 信号识别结果 + """ + freq = c.freq.value + k1, k2, k3 = f"{freq}_UBI_BE辅助V230816".split('_') + v1, v2 = '其他','其他' + ubi = c.ubi + if not ubi or len(ubi['fxs']) <= 2 or len(c.bars_ubi) <= 5: + return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2) + + fxs = ubi['fxs'] + if ubi['direction'] == Direction.Up: + fxs = [x for x in fxs if x.mark == Mark.G] + cnt = 1 + cur_hfx = fxs[0] + for fx in fxs[1:]: + if fx.high > cur_hfx.high: + cnt += 1 + cur_hfx = fx + + if ubi['raw_bars'][-1].high > cur_hfx.high: + v1 = '新高' + v2 = f"第{cnt + 1}次" + + if ubi['direction'] == Direction.Down: + fxs = [x for x in fxs if x.mark == Mark.D] + cnt = 1 + cur_lfx = fxs[0] + for fx in fxs[1:]: + if fx.low < cur_lfx.low: + cnt += 1 + cur_lfx = fx + + if ubi['raw_bars'][-1].low < cur_lfx.low: + v1 = '新低' + 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/pos.py b/czsc/signals/pos.py index bf3f9e8a3..5e1e786d4 100644 --- a/czsc/signals/pos.py +++ b/czsc/signals/pos.py @@ -432,7 +432,7 @@ def pos_holds_V230807(cat: CzscTrader, **kwargs) -> OrderedDict: k1, k2, k3 = f"{pos_name}_{freq1}N{n}M{m}T{t}_BS辅助V230807".split("_") v1 = '其他' # 如果没有持仓策略，则不产生信号 - if not hasattr(cat, "positions"): + 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] diff --git a/czsc/signals/tas.py b/czsc/signals/tas.py index b0925cc5a..6738e0412 100644 --- 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 @@ -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,18 +3209,21 @@ 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 + n or len(c.bars_ubi) >= 7: + 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) - bis = get_sub_elements(c.bi_list, di=di, n=n) + bis = get_sub_elements(c.bi_list, di=di, n=n*2+1) b1 = bis[-1] 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] + 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) @@ -3265,3 +3268,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/signals/vol.py b/czsc/signals/vol.py index dd1d2d16d..da2d9c6e7 100644 --- a/czsc/signals/vol.py +++ b/czsc/signals/vol.py @@ -273,7 +273,7 @@ def _check_gao_liang_zhu(bars: List[RawBar]): def vol_window_V230731(c: CZSC, **kwargs) -> OrderedDict: """指定窗口内成交量的特征 - 参数模板："{freq}_D{di}W{window}M{m}N{n}_窗口能量V230731" + 参数模板："{freq}_D{di}W{w}M{m}N{n}_窗口能量V230731" **信号逻辑：** diff --git a/czsc/traders/base.py b/czsc/traders/base.py index 7194fddc2..dbd787eaf 100644 --- a/czsc/traders/base.py +++ b/czsc/traders/base.py @@ -14,7 +14,7 @@ from datetime import datetime, timedelta from deprecated import deprecated from collections import OrderedDict -from typing import Callable, List, AnyStr, Union +from typing import Callable, List, AnyStr, Union, Optional from pyecharts.charts import Tab from pyecharts.components import Table from pyecharts.options import ComponentTitleOpts @@ -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}") @@ -314,7 +315,8 @@ def __init__(self, bg: BarGenerator = None, positions: List[Position] = None, vote - 投票表决，pos = 1 max - 取最大，pos = 1 - 对于传入回调函数的情况，输入是 self.positions + 对于传入回调函数的情况，函数的输入为 dict，key 为 position.name，value 为 position.pos, 样例输入： + {'多头策略A': 1, '多头策略B': 1, '空头策略A': -1} """ self.positions = positions if self.positions: @@ -403,7 +405,7 @@ def get_ensemble_pos(self, method: Union[AnyStr, Callable] = None) -> float: raise ValueError else: - pos = method(self.positions) + pos = method({x.name: x.pos for x in self.positions}) return pos @@ -472,3 +474,40 @@ def take_snapshot(self, file_html=None, width: str = "1400px", height: str = "58 else: return tab + def get_ensemble_weight(self, method: Optional[Union[AnyStr, Callable]] = None): + """获取 CzscTrader 中所有 positions 按照 method 方法集成之后的权重 + + :param method: str or callable + 集成方法，可选值包括：'mean', 'max', 'min', 'vote' + 也可以传入自定义的函数，函数的输入为 dict，key 为 position.name，value 为 position.pos, 样例输入： + {'多头策略A': 1, '多头策略B': 1, '空头策略A': -1} + :param kwargs: + :return: pd.DataFrame + columns = ['dt', 'symbol', 'weight', 'price'] + """ + from czsc.traders.weight_backtest import get_ensemble_weight + method = self.__ensemble_method if not method else method + return get_ensemble_weight(self, method) + + def weight_backtest(self, **kwargs): + """执行仓位集成权重的回测 + + :param kwargs: + + - method: str or callable，集成方法，参考 get_ensemble_weight 方法 + - digits: int，权重小数点后保留的位数，例如 2 表示保留两位小数 + - fee_rate: float，手续费率，例如 0.0002 表示万二 + - res_path: str，回测结果保存路径 + + :return: 回测结果 + """ + from czsc.traders.weight_backtest import WeightBacktest + + method = kwargs.get("method", self.__ensemble_method) + digits = kwargs.get("digits", 2) + fee_rate = kwargs.get("fee_rate", 0.0002) + res_path = kwargs.get("res_path", "./weight_backtest") + dfw = self.get_ensemble_weight(method) + wb = WeightBacktest(dfw, digits=digits, fee_rate=fee_rate, res_path=res_path) + _res = wb.backtest() + return _res diff --git a/czsc/traders/weight_backtest.py b/czsc/traders/weight_backtest.py index 1e103b45f..a99ee04b0 100644 --- a/czsc/traders/weight_backtest.py +++ b/czsc/traders/weight_backtest.py @@ -8,12 +8,12 @@ import numpy as np import pandas as pd import plotly.express as px -from czsc.traders.base import CzscTrader from loguru import logger from pathlib import Path from typing import Union, AnyStr, Callable -from czsc.utils.stats import daily_performance, evaluate_pairs +from czsc.traders.base import CzscTrader from czsc.utils.io import save_json +from czsc.utils.stats import daily_performance, evaluate_pairs def get_ensemble_weight(trader: CzscTrader, method: Union[AnyStr, Callable] = 'mean'): @@ -68,13 +68,13 @@ class WeightBacktest: def __init__(self, dfw, digits=2, **kwargs) -> None: """持仓权重回测 - :param dfw: pd.DataFrame, columns = ['dt', 'symbol', 'weight', 'price'], 持仓权重数据， - 其中 - dt 为结束时间， - symbol 为合约代码， - weight 为持仓权重， - price 为结束时间对应的交易价格，可以是当前K线的收盘价，或者下一根K线的开盘价，或者未来N根K线的TWAP、VWAP等 - + :param dfw: pd.DataFrame, columns = ['dt', 'symbol', 'weight', 'price'], 持仓权重数据，其中 + + dt 为K线结束时间， + symbol 为合约代码， + weight 为K线结束时间对应的持仓权重， + price 为结束时间对应的交易价格，可以是当前K线的收盘价，或者下一根K线的开盘价，或者未来N根K线的TWAP、VWAP等 + 数据样例如下： =================== ======== ======== ======= dt symbol weight price @@ -88,7 +88,10 @@ def __init__(self, dfw, digits=2, **kwargs) -> None: :param digits: int, 权重列保留小数位数 :param kwargs: + - fee_rate: float，单边交易成本，包括手续费与冲击成本, 默认为 0.0002 + - res_path: str，回测结果保存路径，默认为 "weight_backtest" + """ self.kwargs = kwargs self.dfw = dfw.copy() @@ -139,10 +142,7 @@ def get_symbol_daily(self, symbol): def get_symbol_pairs(self, symbol): """获取某个合约的开平交易记录""" dfs = self.dfw[self.dfw['symbol'] == symbol].copy() - dfs['direction'] = 0 - dfs.loc[dfs['weight'] > 0, 'direction'] = 1 - dfs.loc[dfs['weight'] < 0, 'direction'] = -1 - dfs['volume'] = (dfs['weight'] * pow(10, self.digits)).astype(int) * dfs['direction'] + dfs['volume'] = (dfs['weight'] * pow(10, self.digits)).astype(int) dfs['bar_id'] = list(range(1, len(dfs)+1)) # 根据权重变化生成开平仓记录 @@ -227,7 +227,9 @@ def backtest(self): dret = pd.concat([v['daily'] for v in res.values()], ignore_index=True) dret = pd.pivot_table(dret, index='date', columns='symbol', values='return').fillna(0) dret['total'] = dret[list(res.keys())].mean(axis=1) - logger.info(f"品种等权费后日收益率：{daily_performance(dret['total'])}") + stats = {"开始日期": dret.index.min().strftime("%Y%m%d"), "结束日期": dret.index.max().strftime("%Y%m%d")} + stats.update(daily_performance(dret['total'])) + logger.info(f"品种等权费后日收益率：{stats}") dret.to_excel(self.res_path.joinpath("daily_return.xlsx"), index=True) logger.info(f"品种等权费后日收益率已保存到 {self.res_path.joinpath('daily_return.xlsx')}") @@ -243,7 +245,8 @@ def backtest(self): pairs_stats = evaluate_pairs(dfp) pairs_stats = {k: v for k, v in pairs_stats.items() if k in ['单笔收益', '持仓K线数', '交易胜率', '持仓天数']} logger.info(f"所有开平交易记录的表现：{pairs_stats}") - save_json(pairs_stats, self.res_path.joinpath("pairs_stats.json")) - logger.info(f"所有开平交易记录的表现已保存到 {self.res_path.joinpath('pairs_stats.json')}") - + stats.update(pairs_stats) + logger.info(f"策略评价：{stats}") + save_json(stats, self.res_path.joinpath("stats.json")) + res['stats'] = stats return res diff --git a/czsc/utils/stats.py b/czsc/utils/stats.py index 1dcff36ce..4acd3b494 100644 --- a/czsc/utils/stats.py +++ b/czsc/utils/stats.py @@ -136,8 +136,8 @@ def cal_break_even_point(seq: List[float]) -> float: """ if sum(seq) < 0: return 1.0 - seq = np.cumsum(sorted(seq)) - return (np.sum(seq < 0) + 1) / len(seq) + seq = np.cumsum(sorted(seq)) # type: ignore + return (np.sum(seq < 0) + 1) / len(seq) # type: ignore def evaluate_pairs(pairs: pd.DataFrame, trade_dir: str = "多空") -> dict: @@ -161,14 +161,10 @@ def evaluate_pairs(pairs: pd.DataFrame, trade_dir: str = "多空") -> dict: from czsc.objects import cal_break_even_point pairs = pairs.copy() - if trade_dir in ["多头", "空头"]: - pairs = pairs[pairs["交易方向"] == trade_dir] - else: - assert trade_dir == "多空", "trade_dir 参数错误，可选值 ['多头', '空头', '多空']" p = { "交易方向": trade_dir, - "交易次数": len(pairs), + "交易次数": 0, "累计收益": 0, "单笔收益": 0, "盈利次数": 0, @@ -188,7 +184,16 @@ def evaluate_pairs(pairs: pd.DataFrame, trade_dir: str = "多空") -> dict: if len(pairs) == 0: return p + if trade_dir in ["多头", "空头"]: + pairs = pairs[pairs["交易方向"] == trade_dir] + else: + assert trade_dir == "多空", "trade_dir 参数错误，可选值 ['多头', '空头', '多空']" + + if len(pairs) == 0: + return p + pairs = pairs.to_dict(orient='records') + p['交易次数'] = len(pairs) p["盈亏平衡点"] = round(cal_break_even_point([x['盈亏比例'] for x in pairs]), 4) p["累计收益"] = round(sum([x["盈亏比例"] for x in pairs]), 2) p["单笔收益"] = round(p["累计收益"] / p["交易次数"], 2) diff --git a/examples/cal_ensemble_weight.py b/examples/cal_ensemble_weight.py new file mode 100644 index 000000000..9fabd25d0 --- /dev/null +++ b/examples/cal_ensemble_weight.py @@ -0,0 +1,22 @@ +import czsc +import pandas as pd + + +def run_by_weights(): + """从持仓权重样例数据中回测""" + dfw = pd.read_feather(r"C:\Users\zengb\Desktop\230814\weight_example.feather") + wb = czsc.WeightBacktest(dfw, digits=1, fee_rate=0.0002, res_path=r"C:\Users\zengb\Desktop\230814\weight_example") + res = wb.backtest() + + +def run_by_ensemble(): + """从单个 trader 中获取持仓权重，然后回测""" + trader = czsc.dill_load(r"C:\Users\zengb\Desktop\230814\DLi9001.trader") + + def __ensemble_method(x): + return x['A股日线SMA#5多头'] + 0.5 * x['A股日线SMA#5空头'] + + dfw = czsc.get_ensemble_weight(trader, method=__ensemble_method) + wb = czsc.WeightBacktest(dfw, digits=1, fee_rate=0.0002, res_path=r"C:\Users\zengb\Desktop\230814\DLi9001_ensemble") + res = wb.backtest() + diff --git a/examples/create_one_three.py b/examples/create_one_three.py new file mode 100644 index 000000000..3a756ddd1 --- /dev/null +++ b/examples/create_one_three.py @@ -0,0 +1,211 @@ +# 编写策略样例：https://s0cqcxuy3p.feishu.cn/wiki/D3VEwHnpjiA8cokFTWEcuqDunZc + +from typing import List +from czsc.objects import Event, Position +from czsc import CzscStrategyBase, Position + + +class Strategy(CzscStrategyBase): + + def create_pos_a(self, symbol, **kwargs): + """_summary_ + + https://czsc.readthedocs.io/en/latest/api/czsc.signals.cxt_third_buy_V230228.html + https://czsc.readthedocs.io/en/latest/api/czsc.signals.cxt_first_sell_V221126.html + + :param symbol: _description_ + :return: _description_ + """ + base_freq = kwargs.get("base_freq", "30分钟") + + opens = [ + { + "operate": "开多", + "signals_not": [], + "signals_all": [], + "factors": [{"name": "三买多头", "signals_all": ["日线_D1_三买辅助V230228_三买_任意_任意_0"]}], + } + ] + + exits = [ + { + "operate": "平多", + "signals_all": [], + "signals_not": [], + "factors": [ + { + "name": "平多", + "signals_all": ["30分钟_D1B_SELL1_一卖_任意_任意_0"], + "signals_any": [ + "30分钟_D1B_SELL1_一卖_9笔_任意_0", + "30分钟_D1B_SELL1_一卖_11笔_任意_0", + "30分钟_D1B_SELL1_一卖_13笔_任意_0", + ], + } + ], + } + ] + opens[0]["signals_all"].append(f"{base_freq}_D1_涨跌停V230331_任意_任意_任意_0") + pos_name = "日线三买多头A" + + T0 = kwargs.get("T0", False) + pos_name = f"{pos_name}T0" if T0 else f"{pos_name}" + + pos = Position( + name=pos_name, + symbol=symbol, + opens=[Event.load(x) for x in opens], + exits=[Event.load(x) for x in exits], + interval=kwargs.get("interval", 3600 * 2), + timeout=kwargs.get("timeout", 16 * 30), + stop_loss=kwargs.get("stop_loss", 300), + T0=T0, + ) + return pos + + def create_pos_b(self, symbol, **kwargs): + """_summary_ + + https://czsc.readthedocs.io/en/latest/api/czsc.signals.pos_status_V230808.html + https://czsc.readthedocs.io/en/latest/api/czsc.signals.cxt_bi_status_V230102.html + + :param symbol: _description_ + :return: _description_ + """ + base_freq = kwargs.get("base_freq", "30分钟") + last_pos_name = "日线三买多头A" + + opens = [ + { + "operate": "开空", + "signals_not": [], + "signals_all": [], + "factors": [ + { + "name": "第一次平多", + "signals_all": [ + "30分钟_D1_表里关系V230102_向上_顶分_任意_0", + f"{last_pos_name}_持仓状态_BS辅助V230808_持多_任意_任意_0", + ], + } + ], + } + ] + + exits = [ + { + "operate": "平空", + "signals_all": [], + "signals_not": [], + "factors": [ + { + "name": "平多", + "signals_all": [f"{last_pos_name}_持仓状态_BS辅助V230808_任意_任意_任意_0"], + "signals_any": [ + f"{last_pos_name}_持仓状态_BS辅助V230808_持空_任意_任意_0", + f"{last_pos_name}_持仓状态_BS辅助V230808_持币_任意_任意_0", + ], + } + ], + } + ] + opens[0]["signals_all"].append(f"{base_freq}_D1_涨跌停V230331_任意_任意_任意_0") + pos_name = "日线三买第一次平仓" + + T0 = kwargs.get("T0", False) + pos_name = f"{pos_name}T0" if T0 else f"{pos_name}" + + pos = Position( + name=pos_name, + symbol=symbol, + opens=[Event.load(x) for x in opens], + exits=[Event.load(x) for x in exits], + interval=kwargs.get("interval", 3600 * 2), + timeout=kwargs.get("timeout", 16 * 30), + stop_loss=kwargs.get("stop_loss", 1000), + T0=T0, + ) + return pos + + def create_pos_c(self, symbol, **kwargs): + """_summary_ + + https://czsc.readthedocs.io/en/latest/api/czsc.signals.pos_status_V230808.html + https://czsc.readthedocs.io/en/latest/api/czsc.signals.cxt_bi_status_V230102.html + + :param symbol: _description_ + :return: _description_ + """ + base_freq = kwargs.get("base_freq", "30分钟") + last_pos_a = "日线三买多头A" + last_pos_b = "日线三买第一次平仓" + + opens = [ + { + "operate": "开空", + "signals_not": [], + "signals_all": [], + "factors": [ + { + "name": "第二次平多", + "signals_all": [ + "日线_D1_表里关系V230102_向上_顶分_任意_0", + f"{last_pos_a}_持仓状态_BS辅助V230808_持多_任意_任意_0", + f"{last_pos_b}_持仓状态_BS辅助V230808_持空_任意_任意_0", + ], + } + ], + } + ] + + exits = [ + { + "operate": "平空", + "signals_all": [], + "signals_not": [], + "factors": [ + { + "name": "平多", + "signals_all": [f"{last_pos_a}_持仓状态_BS辅助V230808_任意_任意_任意_0"], + "signals_any": [ + f"{last_pos_a}_持仓状态_BS辅助V230808_持空_任意_任意_0", + f"{last_pos_a}_持仓状态_BS辅助V230808_持币_任意_任意_0", + ], + } + ], + } + ] + opens[0]["signals_all"].append(f"{base_freq}_D1_涨跌停V230331_任意_任意_任意_0") + pos_name = "日线三买第二次平仓" + + T0 = kwargs.get("T0", False) + pos_name = f"{pos_name}T0" if T0 else f"{pos_name}" + + pos = Position( + name=pos_name, + symbol=symbol, + opens=[Event.load(x) for x in opens], + exits=[Event.load(x) for x in exits], + interval=kwargs.get("interval", 3600 * 2), + timeout=kwargs.get("timeout", 16 * 30), + stop_loss=kwargs.get("stop_loss", 1000), + T0=T0, + ) + return pos + + @property + def positions(self) -> List[Position]: + _pos = [ + self.create_pos_a(symbol=self.symbol, base_freq="30分钟", T0=False), + self.create_pos_b(symbol=self.symbol, base_freq="30分钟", T0=False), + self.create_pos_c(symbol=self.symbol, base_freq="30分钟", T0=False), + ] + return _pos + + +if __name__ == "__main__": + from czsc.connectors.research import get_raw_bars + + tactic = Strategy(symbol="000001.SH") + bars = get_raw_bars("000001.SH", freq="30分钟", sdt="2015-01-01", edt="2022-07-01") + tactic.replay(bars, res_path=r"C:\Users\zengb\Desktop\230814\一开多平") diff --git a/examples/signals_dev/cxt_bi_end_V230815.py b/examples/signals_dev/cxt_bi_end_V230815.py new file mode 100644 index 000000000..65de92f44 --- /dev/null +++ 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 new file mode 100644 index 000000000..0be562ac2 --- /dev/null +++ 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/cxt_ubi_end_V230816.py b/examples/signals_dev/cxt_ubi_end_V230816.py new file mode 100644 index 000000000..3039d3ccb --- /dev/null +++ b/examples/signals_dev/cxt_ubi_end_V230816.py @@ -0,0 +1,85 @@ +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_ubi_end_V230816(c: CZSC, **kwargs) -> OrderedDict: + """当前是未完成笔的第几次新低或新高，用于笔结束辅助 + + 参数模板："{freq}_UBI_BE辅助V230816" + + **信号逻辑：** + + 以向上未完成笔为例：取所有顶分型，计算创新高的底分型数量N，如果当前K线创新高，则新高次数为N+1 + + **信号列表：** + + - Signal('日线_UBI_BE辅助V230816_新低_第4次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第5次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第6次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第2次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第3次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第4次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第5次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第6次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新高_第7次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第2次_任意_0') + - Signal('日线_UBI_BE辅助V230816_新低_第3次_任意_0') + + :param c: CZSC对象 + :param kwargs: + :return: 信号识别结果 + """ + freq = c.freq.value + k1, k2, k3 = f"{freq}_UBI_BE辅助V230816".split('_') + v1, v2 = '其他','其他' + ubi = c.ubi + if not ubi or len(ubi['fxs']) <= 2 or len(c.bars_ubi) <= 5: + return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2) + + fxs = ubi['fxs'] + if ubi['direction'] == Direction.Up: + fxs = [x for x in fxs if x.mark == Mark.G] + cnt = 1 + cur_hfx = fxs[0] + for fx in fxs[1:]: + if fx.high > cur_hfx.high: + cnt += 1 + cur_hfx = fx + + if ubi['raw_bars'][-1].high > cur_hfx.high: + v1 = '新高' + v2 = f"第{cnt + 1}次" + + if ubi['direction'] == Direction.Down: + fxs = [x for x in fxs if x.mark == Mark.D] + cnt = 1 + cur_lfx = fxs[0] + for fx in fxs[1:]: + if fx.low < cur_lfx.low: + cnt += 1 + cur_lfx = fx + + if ubi['raw_bars'][-1].low < cur_lfx.low: + v1 = '新低' + v2 = f"第{cnt + 1}次" + + 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_ubi_end_V230816, 'freq': '日线', 'di': 1, 'max_overlap': 1}] + 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 index fe0f45fc7..942a95884 100644 --- 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,18 +43,21 @@ 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 + n or len(c.bars_ubi) >= 7: + 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) - bis = get_sub_elements(c.bi_list, di=di, n=n) + bis = get_sub_elements(c.bi_list, di=di, n=n*2+1) b1 = bis[-1] 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] + 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/tas_macd_bc_V230804.py b/examples/signals_dev/tas_macd_bc_V230804.py new file mode 100644 index 000000000..a1861ccdb --- /dev/null +++ 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 new file mode 100644 index 000000000..a939047e7 --- /dev/null +++ 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() diff --git a/test/test_trader_base.py b/test/test_trader_base.py index a66b7fb88..b2be4ca5e 100644 --- a/test/test_trader_base.py +++ b/test/test_trader_base.py @@ -4,9 +4,10 @@ email: zeng_bin8888@163.com create_dt: 2021/11/7 21:07 """ +import os import pandas as pd from copy import deepcopy -from typing import List +from czsc.utils.cache import home_path from czsc.traders.base import CzscSignals, BarGenerator, CzscTrader from czsc.traders.sig_parse import get_signals_config, get_signals_freqs from czsc.objects import Signal, Factor, Event, Operate, Position @@ -245,13 +246,17 @@ def __create_sma20_pos(): assert [x.pos for x in ct.positions] == [0, 0, 0] # 测试自定义仓位集成 - def _weighted_ensemble(positions: List[Position]): - return 0.5 * positions[0].pos + 0.5 * positions[1].pos + def _weighted_ensemble(poss): + return 0.5 * poss['测试A'] + 0.5 * poss['测试B'] assert ct.get_ensemble_pos(_weighted_ensemble) == 0 assert ct.get_ensemble_pos('vote') == 0 assert ct.get_ensemble_pos('max') == 0 assert ct.get_ensemble_pos('mean') == 0 + dfw = ct.get_ensemble_weight(method='mean') + assert len(dfw) == len(bars_right) + + res = ct.weight_backtest(method='mean', res_path=os.path.join(home_path, "test_trader")) # 通过 on_bar 执行 ct1 = CzscTrader(deepcopy(bg), signals_config=signals_config,