#!/usr/bin/env python3 """交易详情图定稿模板v3(2026-08-29用户最终确认): 10行×4列=40笔 关键特性: - K线+成交量双段, 红涨绿跌 - 序号x轴(去非交易日空隙, K线连续) - x轴只在关键节点(T0/ta/tb/买入/卖出)显示年月日刻度, 文字标注不含日期 - 信号链三竖虚线+图内顶部白底汉字标签 - 粉红▲买入(次日开盘+价格), 紫▼卖出(原因+收益率) - 每格右下角强度0-100分档标注(可选, 读strength_baseline.csv) - dpi=200高分辨率 用户原话: "固定这个模板，以后就按这个来" """ 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'] = ['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的信号链 (t0_idx, ta_idx, tb_idx)""" 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['zt_vol_filter'] 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 if ztv > 0: ma5z = vol[max(0, t0 - 5):t0].mean() if t0 >= 5 else vol[:t0].mean() if ma5z > 0 and v0 >= ma5z * ztv: continue p0 = close[t0] ta = 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: ta = j break if ta is None: continue for k in range(ta + 1, min(t0 + bu + 1, n)): if close[k] > p0 * (1 + bs): 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): """绘制一笔交易: 连续K线(序号x轴)+成交量+信号链+买卖点+强度标注""" 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] key_ticks.append((bd_num, str(t['buy_date']))) key_ticks.append((sd_num, str(t['sell_date']))) 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': '超时'}[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)) # 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'])) # 强度标注(可选): 读strength_baseline.csv std_map = {} try: bl = pd.read_csv(f'{BASE}/strength_baseline.csv') bl['std'] = bl['raw_score'].rank(pct=True) * 100 std_map = dict(zip(zip(bl['code'], bl['buy_date'].astype(str)), bl['std'])) except Exception: pass 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']) # 选样本: 默认随机; 改sorted的key/reverse可改为收益最高/最低10笔 chosen = [] pool_dailies = dailies for t in sorted(trades, key=lambda x: x['ret_net'], reverse=True): # reverse=False=最低10笔 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笔", 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('四池理想策略交易详情(40笔): 强度分档标注(右下角=强度0-100)', fontsize=15, y=0.994) fig.subplots_adjust(top=0.968, bottom=0.05, left=0.05, right=0.985, hspace=0.34, wspace=0.22) out = f'{BASE}/backtest_zt_ideal_examples_v3.png' plt.savefig(out, dpi=200) print(f"交易详情图: {out}") if __name__ == '__main__': main()