#!/usr/bin/env python3 """四池理想策略收益最高10笔交易详情(共40笔), 定稿模板画法""" import pandas as pd import numpy as np import json, sys, random import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import matplotlib.dates as mdates plt.rcParams['font.sans-serif'] = ['STHeiti', 'Noto Sans CJK SC', 'WenQuanYi Zen Hei', 'SimHei', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False sys.path.insert(0, '/tmp') import grid_or as g import ideal_engine as ie BASE = g.BASE PN = {'hs300': '沪深300', 'zz500': '中证500', 'zz1000': '中证1000', 'zz2000': '中证2000'} COMBO = { 'hs300': {'bp': {'vol_base': 'maN', 'ma_n': 4, 'vol_shrink': 0.7, 'shrink_win': 12, 'break_up': 15, 'break_strength': 0.0, 'zt_vol_filter': 0.0}, 'profit': 0.50, 'dd': 0.99, 'stop': 0.35, 'hold': 15}, 'zz500': {'bp': {'vol_base': 'maN', 'ma_n': 3, 'vol_shrink': 0.55, 'shrink_win': 10, 'break_up': 12, 'break_strength': 0.02, 'zt_vol_filter': 4.0}, 'profit': 0.45, 'dd': 0.25, 'stop': 0.20, 'hold': 15}, 'zz1000': {'bp': {'vol_base': 'zt', 'ma_n': 3, 'vol_shrink': 0.3, 'shrink_win': 12, 'break_up': 10, 'break_strength': 0.0, 'zt_vol_filter': 3.0}, 'profit': 0.20, 'dd': 0.25, 'stop': None, 'hold': 15}, 'zz2000': {'bp': {'vol_base': 'maN', 'ma_n': 4, 'vol_shrink': 0.15, 'shrink_win': 10, 'break_up': 15, 'break_strength': 0.025, 'zt_vol_filter': 4.0}, 'profit': 0.40, 'dd': 0.15, 'stop': 0.30, 'hold': 50}, } def chain_for_buy(d, code, bp, buy_date): """返回匹配buy_date的信号链 — 与引擎signal_detect一致(ta选最低+量能参数)""" d = d.sort_values('trade_date').reset_index(drop=True) pct = d['pct_chg'].astype(float).values close = d['close'].astype(float).values vol = d['vol'].astype(float).values dates = d['trade_date'].values if code.startswith(('60', '00')): bt = 0.099 elif code.startswith(('30', '68')): bt = 0.199 else: bt = 0.299 vb, vs, bs = bp['vol_base'], bp['vol_shrink'], bp['break_strength'] sw, bu, mn = int(bp['shrink_win']), int(bp['break_up']), int(bp['ma_n']) ztv = bp.get('t0_vol_max', bp.get('zt_vol_filter', 0)) t0min = bp.get('t0_vol_min', 0.0) taddmax = bp.get('ta_dd_max', 0.0) t0openmax = bp.get('t0_open_max', 0.0) t0prev = bp.get('t0_prev_limit', 0) talowfloor = bp.get('ta_low_floor', 0.0) tbt0_min = bp.get('tb_t0_min', 0.0) tbt0_max = bp.get('tb_t0_max', 0.0) tbmin = bp.get('tb_vol_min', 0.0) tbmax = bp.get('tb_vol_max', 0.0) n = len(d) for t0 in np.where(pct >= bt - 0.05)[0]: if t0 + 2 >= n: continue v0 = vol[t0] if v0 <= 0: continue ma10z = vol[max(0, t0 - 10):t0].mean() if t0 >= 10 else vol[:t0].mean() if t0min > 0: if ma10z <= 0 or v0 < ma10z * t0min: continue if ztv > 0: if ma10z > 0 and v0 >= ma10z * ztv: continue p0 = close[t0] p0_floor = p0 * (1 - taddmax) if taddmax > 0 else None # ta: 窗口内所有合格缩量日(且深度不破位), 选收盘价最低(与引擎一致) ta = None ta_min_close = None for j in range(t0 + 1, min(t0 + sw + 1, n)): shrink_ok = False if vb in ('zt', 'or') and vol[j] < v0 * vs: shrink_ok = True if vb in ('maN', 'or'): m = vol[max(0, j - mn):j].mean() if j >= mn else vol[:j].mean() if m > 0 and vol[j] < m * vs: shrink_ok = True if shrink_ok and close[j] < p0: if p0_floor is not None and close[j] < p0_floor: continue if talowfloor > 0 and open_[t0] > 0 and close[j] < open_[t0] * talowfloor: continue # 跌破铁底(T0开盘): 放弃该日 if ta_min_close is None or close[j] < ta_min_close: ta_min_close = close[j] ta = j if ta is None: continue for k in range(ta + 1, min(t0 + bu + 1, n)): if close[k] > p0 * (1 + bs): ok = True if tbt0_min > 0 and v0 > 0 and vol[k] < v0 * tbt0_min: ok = False if ok and tbt0_max > 0 and v0 > 0 and vol[k] > v0 * tbt0_max: ok = False if ok and (tbmin > 0 or tbmax > 0): cb = vol[t0 + 1:k].mean() if k > t0 + 1 else vol[k] if cb > 0: if tbmin > 0 and vol[k] < cb * tbmin: ok = False if ok and tbmax > 0 and vol[k] > cb * tbmax: ok = False else: ok = False if not ok: continue if k + 1 < n and dates[k + 1] == buy_date: return t0, ta, k break return None, None, None def draw_trade(gs, row, col, t, d, name, ind, code, std=None): t0i, tai, tbi = chain_for_buy(d, code, bp_save, t['buy_date']) if t0i is None: return tier = '强' if (std or 0) >= 70 else ('中' if (std or 0) >= 30 else '弱') tier_c = {'强': '#E91E63', '中': '#FF8F00', '弱': '#7f8c8d'}[tier] gss = gs[row, col].subgridspec(2, 1, height_ratios=[2.6, 1], hspace=0.06) ax = fig.add_subplot(gss[0]) axv = fig.add_subplot(gss[1], sharex=ax) bd = pd.Timestamp(t['buy_date']); sd = pd.Timestamp(t['sell_date']) t0d = pd.Timestamp(d['trade_date'].iloc[t0i]) win_start = t0d - pd.Timedelta(days=4) mask = (pd.to_datetime(d['trade_date']) >= win_start) & (pd.to_datetime(d['trade_date']) <= sd + pd.Timedelta(days=3)) sub = d[mask].reset_index(drop=True) # 用序号作x轴(去除非交易日空隙, K线连续) x = np.arange(len(sub)) xmax = x[-1] if len(x) else 0 for xi, row2 in sub.iterrows(): o, h, l, c_ = row2['open'], row2['high'], row2['low'], row2['close'] up = c_ >= o color = '#e74c3c' if up else '#2ecc71' ax.plot([x[xi], x[xi]], [l, h], color=color, linewidth=0.8) ax.bar(x[xi], abs(c_ - o), bottom=min(o, c_), width=0.6, color=color, alpha=0.9, edgecolor=color) vv = row2['vol'] axv.bar(x[xi], vv, width=0.6, color=color, alpha=0.8) ymax = ax.get_ylim()[1] ymin = ax.get_ylim()[0] key_ticks = [] for dt, lc, txt in [(t0d, '#e74c3c', 'T0涨停'), (pd.Timestamp(d['trade_date'].iloc[tai]), '#1565C0', 'ta缩量'), (pd.Timestamp(d['trade_date'].iloc[tbi]), '#FF8F00', 'tb突破')]: dt_s = str(pd.Timestamp(dt).date()).replace('-', '') pos = np.where(sub['trade_date'].values == dt_s)[0] if len(pos): ax.axvline(pos[0], color=lc, linewidth=1.2, linestyle='--', alpha=0.9) ax.text(pos[0], ymin + (ymax - ymin) * 0.97, txt, fontsize=8, color=lc, ha='center', fontweight='bold', bbox=dict(facecolor='white', alpha=0.7, edgecolor='none', pad=0.6)) key_ticks.append((pos[0], dt_s)) bd_num = np.where(sub['trade_date'].values == t['buy_date'])[0][0] sd_num = np.where(sub['trade_date'].values == t['sell_date'])[0][0] bd_s = str(t['buy_date']); sd_s = str(t['sell_date']) key_ticks.append((bd_num, bd_s)) key_ticks.append((sd_num, sd_s)) ax.scatter(bd_num, t['buy_price'], marker='^', color='#E91E63', s=70, zorder=6, edgecolors='white', linewidths=0.8) ax.annotate(f"买{t['buy_price']:.1f}", (bd_num, t['buy_price']), textcoords='offset points', xytext=(0, 8), fontsize=8, ha='center', color='#E91E63', fontweight='bold') ax.scatter(sd_num, t['sell_price'], marker='v', color='#7B1FA2', s=70, zorder=6, edgecolors='white', linewidths=0.8) rr = {'profit': '止盈', 'dd': '回撤', 'stop': '止损', 'timeout': '超时', 'trail': '跟踪'}[t['reason']] ax.annotate(f"卖{t['sell_price']:.1f} {rr} {t['ret_net']*100:+.0f}%", (sd_num, t['sell_price']), textcoords='offset points', xytext=(0, -9), fontsize=8, ha='center', color='#7B1FA2', fontweight='bold') ax.set_title(f"{name}·{ind}\n{t['code'][:9]} {PN[code_pool]} 持{t['hold_days']}天 {t['ret_net']*100:+.0f}%", fontsize=8.5, fontweight='bold') if std is not None: ax.text(0.98, 0.02, f"强度{std:.0f}·{tier}", transform=ax.transAxes, fontsize=7.5, fontweight='bold', ha='right', va='bottom', color=tier_c, bbox=dict(facecolor='white', alpha=0.85, edgecolor=tier_c, linewidth=0.8, pad=1.5)) ax.xaxis.set_major_formatter(mdates.DateFormatter('%y-%m-%d')) # x轴只显示关键节点(年月日), 文字标注不再重复日期 ticks = sorted(set(p for p, _ in key_ticks)) labs = dict(key_ticks) ax.set_xticks(ticks) axv.set_xticks(ticks) axv.set_xticklabels([f"{labs[p][2:4]}-{labs[p][4:6]}-{labs[p][6:]}" for p in ticks], fontsize=6, rotation=45) ax.tick_params(labelbottom=False) axv.tick_params(labelbottom=True) ax.tick_params(labelsize=5) axv.tick_params(labelsize=4) axv.set_xlabel('') plt.setp(ax.get_xticklabels(), visible=False) ax.grid(True, alpha=0.12) axv.grid(True, alpha=0.12) if not pool_label_done: ax.text(-0.18, 0.5, PN[code_pool], transform=ax.transAxes, rotation=90, va='center', ha='right', fontsize=13, fontweight='bold') bp_save = None fig = None code_pool = '' pool_label_done = False def main(): global bp_save, fig, code_pool, pool_label_done random.seed(7) sb = pd.read_pickle(f'{BASE}/stock_basic.pkl') name_map = dict(zip(sb['ts_code'], sb['name'])) ind_map = dict(zip(sb['ts_code'], sb['industry'])) # 基准池强度 bl = pd.read_csv(f'{BASE}/strength_baseline.csv') # 分池基准v3: 池内rank得0-100标准强度 bl['std'] = bl.groupby('pool')['raw_score'].rank(pct=True) * 100 std_map = dict(zip(zip(bl['code'], bl['buy_date'].astype(str)), bl['std'])) bl['tier'] = np.where(bl['std'] >= 70, '强', np.where(bl['std'] >= 30, '中', '弱')) stats_txt = [] for t in ['强', '中', '弱']: sub = bl[bl['tier'] == t] r = sub['ret'] stats_txt.append(f"{t}({sub['std'].min():.0f}-{sub['std'].max():.0f}): n={len(sub)} 均{r.mean()*100:+.1f}% 最大{r.max()*100:+.1f}% 最小{r.min()*100:+.1f}% 胜{(r>0).mean()*100:.0f}%") cache = {} samples = {} for pool, c in COMBO.items(): if pool not in cache: cache[pool] = g.load_pool(pool) members, dailies = cache[pool] bp = c['bp'] sig = g.build_sig(dailies, bp) mode = 'merge' if c['stop'] is None else ('A' if c['dd'] >= 0.99 else 'C') trades = ie.ideal_backtest(pool, members, dailies, sig, c['profit'], c['dd'], mode=mode, stop=c['stop'], hold=c['hold']) # 收益最高10笔(信号链可匹配) chosen = [] pool_dailies = dailies for t in sorted(trades, key=lambda x: x['ret_net'], reverse=True): d = pool_dailies[t['code']].sort_values('trade_date').reset_index(drop=True) t0i, _, _ = chain_for_buy(d, t['code'], bp, t['buy_date']) if t0i is not None: chosen.append(t) if len(chosen) >= 10: break samples[pool] = chosen print(f"{PN[pool]}: {len(trades)}笔中取收益最高10笔 top1={chosen[0]['ret_net']*100:+.1f}% top10={chosen[-1]['ret_net']*100:+.1f}%", flush=True) fig = plt.figure(figsize=(26, 31)) gs = fig.add_gridspec(10, 4, hspace=0.34, wspace=0.22) for pi, (pool, c) in enumerate(COMBO.items()): code_pool = pool for j, t in enumerate(samples[pool]): pool_label_done = (j > 0) bp_save = c['bp'] d = cache[pool][1][t['code']].sort_values('trade_date').reset_index(drop=True) name = name_map.get(t['code'], '?') ind = ind_map.get(t['code'], '?') std = std_map.get((t['code'], str(t['buy_date'])), None) if j < 8: row, col = pi * 2 + j // 4, j % 4 else: row, col = 8 + (j - 8), pi draw_trade(gs, row, col, t, d, name, ind, t['code'], std) from matplotlib.lines import Line2D leg = [Line2D([0], [0], marker='^', color='w', markerfacecolor='#E91E63', markersize=9, label='买入(次日开盘)'), Line2D([0], [0], marker='v', color='w', markerfacecolor='#7B1FA2', markersize=9, label='卖出(止盈/回撤/止损/超时)'), Line2D([0], [0], color='#e74c3c', linestyle='--', label='T0涨停日'), Line2D([0], [0], color='#1565C0', linestyle='--', label='ta缩量回调日'), Line2D([0], [0], color='#FF8F00', linestyle='--', label='tb突破日')] fig.legend(handles=leg, loc='lower center', ncol=5, fontsize=12, framealpha=0.9, bbox_to_anchor=(0.5, 0.012)) fig.suptitle('四池理想策略收益最高10笔交易详情(共40笔): 强度分档标注(右上角=强度0-100)', fontsize=15, y=0.994) # 强度统计框 fig.text(0.5, 0.973, f"强度分档统计(804笔): {' | '.join(stats_txt)}", fontsize=11.5, ha='center', bbox=dict(facecolor='#f8f8f8', edgecolor='#999', linewidth=0.8, boxstyle='round,pad=0.4')) fig.subplots_adjust(top=0.955, bottom=0.05, left=0.05, right=0.985, hspace=0.34, wspace=0.22) out = f'{BASE}/backtest_zt_ideal_top10.png' plt.savefig(out, dpi=200) print(f"交易详情图: {out}") if __name__ == '__main__': main()