#!/usr/bin/env python3 """ 高速系统性算法搜索 v5.3 — 内存回测引擎 核心优化：预加载所有数据到内存，避免回测时反复查询数据库。 - 旧版: ~22s/次 (DB查询) → 新版: ~0.2s/次 (内存读取) - 3000+ 种组合约需 10-15 分钟（而非 47 小时） 两阶段搜索： Phase 1: 用活跃季度(2025-Q3)快速筛选出Top 60 Phase 2: 用全期间(2025-01~2026-02)验证Top 60 """ import sys import os import time import itertools from datetime import date, datetime sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from backtest_recommend import get_db_conn, run_backtest, preload_all_data CAPITAL = 200_000 # ─── 搜索空间 ───────────────── SEARCH_SPACE = { 'tp_sl': [ (6, 3), (8, 4), (8, 6), (10, 5), (10, 8), (12, 6), (12, 8), (15, 8), (15, 10), (20, 10), (20, 12), ], 'min_triggered': [0, 1, 2, 3], 'sell_mode': ['normal', 'ignore', 'delay1', 'delay2', 'delay3'], 'trailing': [ None, (6, 2), (8, 3), (10, 3), (10, 5), (12, 4), (12, 5), (15, 5), ], 'position_pct': [3, 5, 8, 10, 15], 'signal_weight': [False, True], 'max_hold_days': [0, 20, 30, 60], } # Phase 1 筛选期（选一个有代表性的活跃季度） SCREEN_START = date(2025, 7, 1) SCREEN_END = date(2025, 9, 30) # Phase 2 全量验证期 FULL_START = date(2025, 1, 2) FULL_END = date(2026, 2, 25) TOP_N_SCREEN = 60 # Phase 1 筛出前60进入Phase 2 TOP_N_FINAL = 30 # Phase 2 展示前30 def build_params(tp_sl, min_triggered, sell_mode, trailing, position_pct, signal_weight, max_hold_days): """将组合参数转为 run_backtest 的 kwargs""" params = { 'total_capital': CAPITAL, 'position_pct': position_pct, 'signal_weight': signal_weight, 'use_5min_prices': True, # v7: 启用5分钟实时价格 'buy_time': '09:35', # v7: 最优买入时间 'sell_time': '13:40', # v7: 最优卖出时间 'verbose': False, } tp, sl = tp_sl params['stop_loss_pct'] = sl if trailing is not None: # 跟踪止盈模式：不用固定止盈，自动忽略卖出信号 params['take_profit_pct'] = None params['trailing_start_pct'] = trailing[0] params['trailing_gap_pct'] = trailing[1] params['ignore_sell_signal'] = True else: params['take_profit_pct'] = tp if min_triggered > 0: params['min_buy_triggered'] = min_triggered if sell_mode == 'ignore': params['ignore_sell_signal'] = True elif sell_mode.startswith('delay'): days = int(sell_mode.replace('delay', '')) params['sell_confirm_days'] = days if max_hold_days > 0: params['max_hold_days'] = max_hold_days return params def make_name(tp_sl, min_triggered, sell_mode, trailing, position_pct, signal_weight, max_hold_days): """生成人类可读的策略名""" parts = [] tp, sl = tp_sl if trailing: parts.append(f"T{trailing[0]}/{trailing[1]}") else: parts.append(f"TP{tp}") parts.append(f"SL{sl}") if min_triggered > 0: parts.append(f"trig≥{min_triggered}") if sell_mode == 'ignore': parts.append("ign") elif sell_mode.startswith('delay'): parts.append(sell_mode) if max_hold_days > 0: parts.append(f"h≤{max_hold_days}") parts.append(f"{position_pct}%") if signal_weight: parts.append("SW") return "|".join(parts) def is_redundant(sell_mode, trailing): """剪枝：跟踪止盈模式下，delay/normal卖出不生效""" if trailing is not None and sell_mode in ('delay1', 'delay2', 'delay3'): return True if trailing is not None and sell_mode == 'normal': return True return False def main(): conn = get_db_conn() # 生成所有组合并剪枝 all_combos = list(itertools.product( SEARCH_SPACE['tp_sl'], SEARCH_SPACE['min_triggered'], SEARCH_SPACE['sell_mode'], SEARCH_SPACE['trailing'], SEARCH_SPACE['position_pct'], SEARCH_SPACE['signal_weight'], SEARCH_SPACE['max_hold_days'], )) combos = [(tp_sl, mt, sm, tr, pp, sw, mh) for tp_sl, mt, sm, tr, pp, sw, mh in all_combos if not is_redundant(sm, tr)] print(f"{'='*100}") print(f" 🚀 高速系统性算法搜索 v7.0 (内存回测引擎 + 最优时点09:35/13:40)") print(f"{'='*100}") print(f" 本金: ¥{CAPITAL:,}") print(f" 组合总数: {len(all_combos):,} → 剪枝后: {len(combos):,}") print(f" Phase 1: 快速筛选 ({SCREEN_START} ~ {SCREEN_END})") print(f" Phase 2: 全量验证 Top {TOP_N_SCREEN} ({FULL_START} ~ {FULL_END})") print(f"{'='*100}\n") # ════════════════ 预加载数据 ════════════════ print(" 📦 预加载回测数据...") # 加载全量数据（覆盖Phase 1和Phase 2的完整范围，含5分钟K线） data_all = preload_all_data(conn, FULL_START, FULL_END, use_5min=True) print() # ════════════════ Phase 1: 快速筛选 ════════════════ print(f" ▶ Phase 1: 快速筛选 {len(combos):,} 种组合...") phase1_results = [] t_start = time.time() for i, (tp_sl, mt, sm, tr, pp, sw, mh) in enumerate(combos): name = make_name(tp_sl, mt, sm, tr, pp, sw, mh) params = build_params(tp_sl, mt, sm, tr, pp, sw, mh) if (i + 1) % 200 == 0 or i == 0: elapsed = time.time() - t_start speed = (i + 1) / elapsed if elapsed > 0 else 0 eta = (len(combos) - i - 1) / speed if speed > 0 else 0 best_name = phase1_results[0]['name'] if phase1_results else 'N/A' best_profit = phase1_results[0]['profit'] if phase1_results else 0 print(f" [{i+1:>5}/{len(combos)}] {elapsed:.0f}s ({speed:.1f}次/秒) " f"ETA:{eta:.0f}s Top1: ¥{best_profit:+,.0f} ({best_name[:35]})", flush=True) # 使用预加载数据进行内存回测 result = run_backtest( conn, start_date=SCREEN_START, end_date=SCREEN_END, preloaded=data_all, **params ) if result and result.get('stats'): s = result['stats'] phase1_results.append({ 'name': name, 'combo': (tp_sl, mt, sm, tr, pp, sw, mh), 'profit': s['profit'], 'capital_pct': s['capital_pct'], 'capital_ann_pct': s.get('capital_ann_pct', 0), 'win_rate': s['win_rate'], 'profit_factor': s['profit_factor'], 'trade_count': s['trade_count'], 'max_drawdown_pct': s.get('max_drawdown_pct', 0), }) phase1_results.sort(key=lambda x: x['profit'], reverse=True) p1_time = time.time() - t_start p1_speed = len(combos) / p1_time if p1_time > 0 else 0 print(f"\n ✅ Phase 1 完成! {p1_time:.0f}秒 ({p1_speed:.1f}次/秒), 有效结果: {len(phase1_results)}") print(f" Phase 1 Top 10:") for i, r in enumerate(phase1_results[:10], 1): print(f" {i:>2}. ¥{r['profit']:>+10,.0f} 收益{r['capital_pct']:>+6.1f}% " f"胜率{r['win_rate']:>5.1f}% PF{r['profit_factor']:>5.2f} 回撤{r['max_drawdown_pct']:>5.1f}% {r['name']}") # ════════════════ Phase 2: 全量验证 ════════════════ top_candidates = phase1_results[:TOP_N_SCREEN] print(f"\n ▶ Phase 2: 全期间验证 Top {len(top_candidates)} ...") phase2_results = [] t2_start = time.time() for i, cand in enumerate(top_candidates): tp_sl, mt, sm, tr, pp, sw, mh = cand['combo'] name = cand['name'] params = build_params(tp_sl, mt, sm, tr, pp, sw, mh) if (i + 1) % 10 == 0 or i == 0: elapsed = time.time() - t2_start speed = (i + 1) / elapsed if elapsed > 0 else 0 eta = (len(top_candidates) - i - 1) / speed if speed > 0 else 0 print(f" [{i+1}/{len(top_candidates)}] {elapsed:.0f}s ETA:{eta:.0f}s", flush=True) # 使用预加载数据进行全量回测 result = run_backtest( conn, start_date=FULL_START, end_date=FULL_END, preloaded=data_all, **params ) if result and result.get('stats'): s = result['stats'] phase2_results.append({ 'name': name, 'combo': cand['combo'], 'screen_profit': cand['profit'], 'profit': s['profit'], 'capital_pct': s['capital_pct'], 'capital_ann_pct': s.get('capital_ann_pct', 0), 'win_rate': s['win_rate'], 'profit_factor': s['profit_factor'], 'max_drawdown_pct': s.get('max_drawdown_pct', 0), 'trade_count': s['trade_count'], 'max_capital': s.get('max_capital', 0), 'avg_hold_days': s.get('avg_hold_days', 0), 'closed_count': s.get('closed_count', 0), 'wins': s.get('wins', 0), 'losses': s.get('losses', 0), 'capital_ann_method': s.get('capital_ann_method', ''), }) phase2_results.sort(key=lambda x: x['profit'], reverse=True) p2_time = time.time() - t2_start total_time = time.time() - t_start conn.close() # ═══════════════ 控制台输出 ═══════════════ print(f"\n{'='*130}") print(f" 🏆 搜索完成! {len(combos):,}种组合 | " f"Phase1:{p1_time:.0f}s Phase2:{p2_time:.0f}s | " f"总计:{total_time:.0f}s ({total_time/60:.1f}分钟)") print(f"{'='*130}") print(f"\n 📊 全期间 Top {min(TOP_N_FINAL, len(phase2_results))} 算法:\n") header = (f"{'排名':>4} {'全期盈利':>12} {'Q3盈利':>10} {'真实收益':>8} {'年化':>8} " f"{'胜率':>6} {'盈亏比':>6} {'回撤':>6} {'交易':>5} {'持仓天':>6} | {'策略'}") print(f" {header}") print(" " + "-" * 130) for rank, r in enumerate(phase2_results[:TOP_N_FINAL], 1): medal = "🏆" if rank == 1 else ("🥈" if rank == 2 else ("🥉" if rank == 3 else f" {rank:>2}")) print(f" {medal} {r['profit']:>+11,.0f} {r['screen_profit']:>+9,.0f} " f"{r['capital_pct']:>+7.1f}% {r['capital_ann_pct']:>+7.1f}% " f"{r['win_rate']:>5.1f}% {r['profit_factor']:>6.2f} {r['max_drawdown_pct']:>5.1f}% " f"{r['trade_count']:>5} {r['avg_hold_days']:>5.0f}d | {r['name']}") # ─── 维度分析 ───────────────────── if phase2_results: print(f"\n{'='*130}") print(f" 📊 维度影响分析") print(f"{'='*130}") # 止盈方式 print(f"\n 📈 止盈方式 (跟踪 vs 固定):") tr_sub = [r for r in phase2_results if r['combo'][3] is not None] fx_sub = [r for r in phase2_results if r['combo'][3] is None] if tr_sub: avg = sum(r['profit'] for r in tr_sub) / len(tr_sub) best = max(tr_sub, key=lambda x: x['profit']) print(f" 跟踪止盈: n={len(tr_sub):>3} 平均 ¥{avg:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f} ({best['name'][:40]})") if fx_sub: avg = sum(r['profit'] for r in fx_sub) / len(fx_sub) best = max(fx_sub, key=lambda x: x['profit']) print(f" 固定止盈: n={len(fx_sub):>3} 平均 ¥{avg:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f} ({best['name'][:40]})") # 仓位比例 print(f"\n 💰 仓位比例:") for pct in sorted(set(r['combo'][4] for r in phase2_results)): subset = [r for r in phase2_results if r['combo'][4] == pct] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) print(f" {pct:>2}%: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") # 信号加权 print(f"\n 📶 信号加权:") for sw in [False, True]: subset = [r for r in phase2_results if r['combo'][5] == sw] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) print(f" {'加权' if sw else '等权':>4}: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") # 卖出策略 print(f"\n 🛒 卖出策略:") for sm_label in ['normal', 'ignore', 'delay1', 'delay2', 'delay3']: subset = [r for r in phase2_results if r['combo'][2] == sm_label] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) print(f" {sm_label:>8}: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") # 止盈/止损参数 print(f"\n 🎯 止盈线 (固定止盈组合):") for tp in sorted(set(r['combo'][0][0] for r in fx_sub)) if fx_sub else []: subset = [r for r in fx_sub if r['combo'][0][0] == tp] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) print(f" TP={tp:>2}: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") print(f"\n 🛡️ 止损线:") for sl in sorted(set(r['combo'][0][1] for r in phase2_results)): subset = [r for r in phase2_results if r['combo'][0][1] == sl] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) print(f" SL={sl:>2}: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") # 最大持仓天数 print(f"\n ⏰ 最大持仓天数:") for mh in sorted(set(r['combo'][6] for r in phase2_results)): subset = [r for r in phase2_results if r['combo'][6] == mh] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) label = "不限" if mh == 0 else f"{mh}天" print(f" {label:>4}: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") # 买入信号触发数 print(f"\n 🔔 买入信号触发数:") for mt in sorted(set(r['combo'][1] for r in phase2_results)): subset = [r for r in phase2_results if r['combo'][1] == mt] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) label = "不限" if mt == 0 else f"≥{mt}" print(f" {label:>4}: n={len(subset):>3} 平均 ¥{avg_p:>+9,.0f} 最优 ¥{best['profit']:>+9,.0f}") # ─── 与之前冠军对比 ───────────────────── print(f"\n{'='*130}") print(f" 📊 与之前最优算法对比") print(f"{'='*130}") prev_best = {'profit': 56375, 'capital_pct': 21.5, 'capital_ann_pct': 18.5, 'win_rate': 61.2, 'profit_factor': 2.30, 'name': 'v5.2|忽略卖出+TP10+SL8+信号加权'} new_best = phase2_results[0] if phase2_results else None if new_best: print(f" 之前冠军: ¥{prev_best['profit']:>+10,.0f} 收益{prev_best['capital_pct']:>+6.1f}% " f"年化{prev_best['capital_ann_pct']:>+6.1f}% 胜率{prev_best['win_rate']:>5.1f}% " f"PF{prev_best['profit_factor']:>5.2f} {prev_best['name']}") print(f" 新冠军: ¥{new_best['profit']:>+10,.0f} 收益{new_best['capital_pct']:>+6.1f}% " f"年化{new_best['capital_ann_pct']:>+6.1f}% 胜率{new_best['win_rate']:>5.1f}% " f"PF{new_best['profit_factor']:>5.2f} {new_best['name']}") diff = new_best['profit'] - prev_best['profit'] print(f" 差异: ¥{diff:>+10,.0f} {'🎉 新纪录!' if diff > 0 else '❌ 未超越'}") # ─── Markdown 输出 ───────────────────── out_path = os.path.join(os.path.dirname(__file__), "docs", "algo_search_results.md") os.makedirs(os.path.dirname(out_path), exist_ok=True) with open(out_path, "w", encoding="utf-8") as f: f.write("# 🔍 系统性算法搜索结果 (v5.3 内存回测引擎)\n\n") f.write(f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n") f.write("## 搜索配置\n\n") f.write(f"| 项目 | 值 |\n") f.write(f"|------|----|\n") f.write(f"| 本金 | ¥{CAPITAL:,} |\n") f.write(f"| Phase 1 筛选期 | {SCREEN_START} ~ {SCREEN_END} |\n") f.write(f"| Phase 2 验证期 | {FULL_START} ~ {FULL_END} |\n") f.write(f"| 组合总数 | {len(all_combos):,} → 剪枝后 {len(combos):,} |\n") f.write(f"| Phase 1 耗时 | {p1_time:.0f}s ({p1_speed:.1f}次/秒) |\n") f.write(f"| Phase 2 耗时 | {p2_time:.0f}s |\n") f.write(f"| 总耗时 | {total_time:.0f}s ({total_time/60:.1f}分钟) |\n\n") f.write("## 🏆 全期间 Top 30\n\n") f.write("| 排名 | 全期盈利 | Q3盈利 | 真实收益 | 年化 | 胜率 | 盈亏比 | 回撤 | 交易 | 持仓天 | 策略 |\n") f.write("|------|---------|-------|---------|------|------|--------|------|------|--------|------|\n") for rank, r in enumerate(phase2_results[:TOP_N_FINAL], 1): medal = "🏆" if rank == 1 else ("🥈" if rank == 2 else ("🥉" if rank == 3 else f"#{rank}")) f.write(f"| {medal} | ¥{r['profit']:+,.0f} | ¥{r['screen_profit']:+,.0f} | " f"{r['capital_pct']:+.1f}% | {r['capital_ann_pct']:+.1f}% | " f"{r['win_rate']:.1f}% | {r['profit_factor']:.2f} | " f"{r['max_drawdown_pct']:.1f}% | {r['trade_count']} | " f"{r['avg_hold_days']:.0f}d | `{r['name']}` |\n") f.write("\n") # 冠军对比 if new_best: f.write("## 新冠军 vs 之前冠军\n\n") f.write("| 指标 | 之前冠军 | 新冠军 |\n") f.write("|------|---------|-------|\n") f.write(f"| 策略 | `{prev_best['name']}` | `{new_best['name']}` |\n") f.write(f"| 全期盈利 | ¥{prev_best['profit']:+,} | ¥{new_best['profit']:+,} |\n") f.write(f"| 真实收益 | {prev_best['capital_pct']:+.1f}% | {new_best['capital_pct']:+.1f}% |\n") f.write(f"| 年化 | {prev_best['capital_ann_pct']:+.1f}% | {new_best['capital_ann_pct']:+.1f}% |\n") f.write(f"| 胜率 | {prev_best['win_rate']:.1f}% | {new_best['win_rate']:.1f}% |\n") f.write(f"| 盈亏比 | {prev_best['profit_factor']:.2f} | {new_best['profit_factor']:.2f} |\n") f.write(f"| 回撤 | - | {new_best['max_drawdown_pct']:.1f}% |\n") diff = new_best['profit'] - prev_best['profit'] f.write(f"\n{'🎉 **新纪录！**' if diff > 0 else '❌ 未超越之前冠军'}\n\n") # 维度分析 f.write("## 维度影响分析\n\n") # 仓位比例 f.write("### 仓位比例\n\n") f.write("| 仓位 | 数量 | 平均盈利 | 最优盈利 |\n") f.write("|------|------|---------|--------|\n") for pct in sorted(set(r['combo'][4] for r in phase2_results)): subset = [r for r in phase2_results if r['combo'][4] == pct] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) f.write(f"| {pct}% | {len(subset)} | ¥{avg_p:+,.0f} | ¥{best['profit']:+,.0f} |\n") f.write("\n") # 信号加权 f.write("### 信号加权\n\n") f.write("| 模式 | 数量 | 平均盈利 | 最优盈利 |\n") f.write("|------|------|---------|--------|\n") for sw in [False, True]: subset = [r for r in phase2_results if r['combo'][5] == sw] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) f.write(f"| {'加权' if sw else '等权'} | {len(subset)} | ¥{avg_p:+,.0f} | ¥{best['profit']:+,.0f} |\n") f.write("\n") # 止损线 f.write("### 止损线\n\n") f.write("| 止损 | 数量 | 平均盈利 | 最优盈利 |\n") f.write("|------|------|---------|--------|\n") for sl in sorted(set(r['combo'][0][1] for r in phase2_results)): subset = [r for r in phase2_results if r['combo'][0][1] == sl] if subset: avg_p = sum(r['profit'] for r in subset) / len(subset) best = max(subset, key=lambda x: x['profit']) f.write(f"| {sl}% | {len(subset)} | ¥{avg_p:+,.0f} | ¥{best['profit']:+,.0f} |\n") f.write("\n") print(f"\n📝 完整结果已写入 {out_path}") if __name__ == "__main__": main()