#!/usr/bin/env python3 """MoFin 策略实验室 v2 — 多版本策略回测 + 12维上下文 + 因子归因 维度: 个股技术(水平+趋势变化) / 大盘状态 / 行业强度 每个策略版本 = 命名配置 + 元数据(名称/假设/父版本)""" import sqlite3, json, math, os from datetime import datetime, timedelta DB_PATH = "/home/hmo/MoFin/data/mofin.db" import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from backtest_framework import prepare_bars, compute_single_score, compute_kelly # bars 缓存：批量跑多版本时共享 TA 计算 _BARS_CACHE = {} def _bars(code, start_date, end_date): key = (code, start_date, end_date) if key not in _BARS_CACHE: _BARS_CACHE[key] = prepare_bars(code, start_date, end_date) return _BARS_CACHE[key] # ══════════════════════════════════════════════════════ # 策略版本注册表 # ══════════════════════════════════════════════════════ STRATEGIES = { "v1.0": { "version": "v1.0", "name": "多因子基线", "summary": "五因子评分≥45 + 动量≥8，10%止盈 / 2×ATR止损，半Kelly", "hypothesis": "基线版本：验证多因子评分体系的基础有效性", "parent": None, "created": "2026-07-28", "config": { "entry": {"min_score": 45, "min_momentum": 8, "filters": {}}, "exit": {"tp_pct": 0.10, "sl_atr": 2.0, "max_hold_days": 20}, "sizing": {"kelly": True, "kelly_fraction": 0.5}, "eval_step": 5, }, }, "v2.0": { "version": "v2.0", "name": "趋势动能过滤", "summary": "v1 + MACD柱>0 + ROC>2 + ADX≥20 + ATR%≥2.8 过滤弱势入场", "hypothesis": "v1归因：MACD>0.69胜率54%vs33%，ROC>11胜率57%vs35%，ADX>43胜率50%，ATR%>4.3胜率49%vs31%。过滤无趋势/无动能/死鱼股", "parent": "v1.0", "created": "2026-07-28", "config": { "entry": {"min_score": 45, "min_momentum": 8, "filters": {"adx_min": 20, "atr_pct_min": 2.8, "roc_min": 2, "macd_hist_min": 0}}, "exit": {"tp_pct": 0.10, "sl_atr": 2.0, "max_hold_days": 20}, "sizing": {"kelly": True, "kelly_fraction": 0.5}, "eval_step": 5, }, }, "v3.0": { "version": "v3.0", "name": "强动量+优盈亏比", "summary": "v2 + ROC≥8 + 距MA20≥4% + 量比1.0~1.8；止盈15%/止损1.5×ATR（RR→2.2:1）", "hypothesis": "v2归因：ROC>17.5胜率58.5%，距MA20>12.9胜率56.5%，量比1.12~1.45胜率55.1%。且v2平均亏损-9.14%≈止盈10%，RR仅1.1:1是盈亏比恶化主因→收紧止损放大止盈", "parent": "v2.0", "created": "2026-07-28", "config": { "entry": {"min_score": 45, "min_momentum": 8, "filters": {"adx_min": 20, "atr_pct_min": 2.8, "roc_min": 8, "macd_hist_min": 0, "dist_ma20_min": 4, "vol_ratio_min": 1.0, "vol_ratio_max": 1.8}}, "exit": {"tp_pct": 0.15, "sl_atr": 1.5, "max_hold_days": 20}, "sizing": {"kelly": True, "kelly_fraction": 0.5}, "eval_step": 5, }, }, "v4.0": { "version": "v4.0", "name": "大盘回调+趋势结构", "summary": "v3 + 大盘须在MA20上且MA20斜率<-0.05(上升中回调) + 个股更高高点结构 + ROC 10~25 + MACD柱<1.3(避追高)", "hypothesis": "v3归因：大盘MA20斜率-1.76~-0.56时胜率58.8%vs平坡20.6%(差38pp最强信号)；大盘在MA20上胜率42%vs34%；hh结构+15pp；ROC甜区12.9~16.2胜率61%；MACD柱>1.33胜率仅28%(追高必死)；个股MA20斜率<1.5胜率56%vs≥1.5约35%(强势回调买)", "parent": "v3.0", "created": "2026-07-28", "config": { "entry": {"min_score": 45, "min_momentum": 8, "filters": {"adx_min": 20, "atr_pct_min": 3.5, "atr_pct_max": 5.5, "roc_min": 10, "roc_max": 25, "macd_hist_min": 0.25, "macd_hist_max": 1.3, "dist_ma20_min": 4, "vol_ratio_min": 1.2, "vol_ratio_max": 1.5, "ma20_slope_max": 1.5, "mkt_above_ma20": True, "mkt_slope_max": -0.05, "hh_only": True}}, "exit": {"tp_pct": 0.15, "sl_atr": 1.5, "max_hold_days": 20}, "sizing": {"kelly": True, "kelly_fraction": 0.5}, "eval_step": 5, }, }, "v4.1": { "version": "v4.1", "name": "大盘回调·宽松带", "summary": "v4.0放宽：ROC 8~25 + MACD柱0~1.5 + ATR 2.8~6.0 + 量比1.0~1.8；保留大盘MA20上+斜率<0 + hh结构", "hypothesis": "v4.0仅5笔交易=过滤器叠加过拟合(分桶样本仅36笔/桶)。保留归因最强的市场状态+趋势结构信号，放宽窄幅过滤器换取统计样本量", "parent": "v4.0", "created": "2026-07-28", "config": { "entry": {"min_score": 45, "min_momentum": 8, "filters": {"adx_min": 20, "atr_pct_min": 2.8, "atr_pct_max": 6.0, "roc_min": 8, "roc_max": 25, "macd_hist_min": 0, "macd_hist_max": 1.5, "dist_ma20_min": 4, "vol_ratio_min": 1.0, "vol_ratio_max": 1.8, "ma20_slope_max": 1.5, "mkt_above_ma20": True, "mkt_slope_max": 0, "hh_only": True}}, "exit": {"tp_pct": 0.15, "sl_atr": 1.5, "max_hold_days": 20}, "sizing": {"kelly": True, "kelly_fraction": 0.5}, "eval_step": 5, }, }, } # ══════════════════════════════════════════════════════ # v4.0 分支家族：消融实验（每次只动一个维度） # ══════════════════════════════════════════════════════ _V40_BASE = STRATEGIES["v4.0"]["config"] def _v40_branch(version, name, summary, hypothesis, entry_overrides=None, exit_overrides=None): import copy cfg = copy.deepcopy(_V40_BASE) for k, v in (entry_overrides or {}).items(): if k in ("min_score", "min_momentum"): cfg["entry"][k] = v else: cfg["entry"]["filters"][k] = v for k, v in (exit_overrides or {}).items(): cfg["exit"][k] = v return { "version": version, "name": name, "summary": summary, "hypothesis": hypothesis, "parent": "v4.0", "created": "2026-07-28", "config": cfg, } STRATEGIES.update({ # A组: 出场优化（严格入场不变） "v4.0a": _v40_branch("v4.0a", "移动止盈路径", "v4.0入场不变；出场改移动止盈(跟踪1.5×ATR)，无固定目标，让利润奔跑", "v4.0五笔4赢且均赢+13%，固定15%目标可能截断大趋势；跟踪止损可锁定利润同时保留上行空间", exit_overrides={"tp_pct": None, "trail_atr": 1.5, "sl_atr": 1.5, "max_hold_days": 25}), "v4.0b": _v40_branch("v4.0b", "延长持仓路径", "v4.0入场不变；持仓期20→25天，给趋势更多兑现时间", "v3归因显示15天+持仓胜率45.6%为各档最高，强信号可能需要更长时间兑现", exit_overrides={"max_hold_days": 25}), # B组: 单维度放宽（消融，找瓶颈） "v4.0c": _v40_branch("v4.0c", "单放ROC", "v4.0只放宽ROC: 10~25 → 8~30，其余全保持", "消融实验：ROC带是否是交易数瓶颈？放宽后若胜率不降则ROC带可永久放宽", entry_overrides={"roc_min": 8, "roc_max": 30}), "v4.0d": _v40_branch("v4.0d", "单放ATR", "v4.0只放宽ATR%: 3.5~5.5 → 2.8~6.5，其余全保持", "消融实验：ATR带是否过窄排除了高波动赢家？", entry_overrides={"atr_pct_min": 2.8, "atr_pct_max": 6.5}), "v4.0e": _v40_branch("v4.0e", "单放MACD", "v4.0只放宽MACD柱: 0.25~1.3 → 0~2.0，其余全保持", "消融实验：MACD柱0~0.25区间(v3:36%胜率)和>1.3区间(28%)是否真该排除？", entry_overrides={"macd_hist_min": 0, "macd_hist_max": 2.0}), "v4.0f": _v40_branch("v4.0f", "单放量比", "v4.0只放宽量比: 1.2~1.5 → 0.9~2.0，其余全保持", "消融实验：量比带的贡献度几何？", entry_overrides={"vol_ratio_min": 0.9, "vol_ratio_max": 2.0}), "v4.0g": _v40_branch("v4.0g", "单放大盘斜率", "v4.0只放宽大盘MA20斜率: ≤-0.05 → ≤0.5，其余全保持", "消融实验：大盘斜率是最强信号(38pp)但也是最大限制——放到0.5还能保住边缘吗？", entry_overrides={"mkt_slope_max": 0.5}), # C组: 结构替代 "v4.0h": _v40_branch("v4.0h", "去高点结构", "v4.0去掉hh_only(更高高点结构)要求，其余全保持", "hh结构+15pp但样本仅4笔False组——这个过滤器可能既限数量又未必真实有效", entry_overrides={"hh_only": False}), # D组: 融合胜出路径 "v5.0": _v40_branch("v5.0", "量比+ATR融合", "v4.0 + 量比0.9~2.0 + ATR 2.8~6.5（消融胜出的双放宽融合）", "消融结果：单放量比+10笔保73%胜率/盈亏比2.92，单放ATR+4笔保78%胜率——两者是唯一不稀释质量的放宽，融合期望20+笔且保住70%胜率", entry_overrides={"vol_ratio_min": 0.9, "vol_ratio_max": 2.0, "atr_pct_min": 2.8, "atr_pct_max": 6.5}), }) def get_strategy(version): if version not in STRATEGIES: raise ValueError(f"未知策略版本: {version}，可用: {list(STRATEGIES.keys())}") return STRATEGIES[version] # ══════════════════════════════════════════════════════ # 大盘 / 行业上下文 # ══════════════════════════════════════════════════════ _MKT_CTX = {} _SECTOR_CTX = {} _STOCK_SECTOR = {} def prepare_market_context(start_date, end_date): """大盘指数(sh000001)每日状态: 是否在MA20上、MA20斜率、ROC""" global _MKT_CTX _MKT_CTX = {} bars = prepare_bars('sh000001', start_date, end_date) if not bars: return for i, b in enumerate(bars): slope = None if i >= 5: m0, m1 = bars[i-5].get('ma20'), b.get('ma20') if m0 and m1: slope = round((m1 - m0) / m0 * 100, 3) ma20 = b.get('ma20') or 0 _MKT_CTX[b['date']] = { 'above_ma20': (b.get('close') or 0) > ma20 if ma20 > 0 else None, 'ma20_slope': slope, 'roc': b.get('roc'), } def prepare_sector_context(start_date, end_date): """每日行业强度: 平均涨幅/净流入/当日排名 + 个股→行业映射""" global _SECTOR_CTX, _STOCK_SECTOR _SECTOR_CTX, _STOCK_SECTOR = {}, {} conn = sqlite3.connect(DB_PATH) try: rows = conn.execute(""" SELECT substr(m.timestamp,1,10) as d, s.name, AVG(s.change_pct), SUM(s.net_inflow) FROM sector_snapshots s JOIN market_snapshots m ON s.snapshot_id = m.id WHERE m.timestamp >= ? AND m.timestamp <= ? GROUP BY d, s.name """, (start_date, end_date + ' 23:59')).fetchall() # 优先 THS 源命名（与 sector_snapshots 同体系），证监会分类作兜底 _STOCK_SECTOR = {} for code, sec, src in conn.execute( "SELECT code, sector_name, source FROM stock_sectors").fetchall(): if src == 'ths' or code not in _STOCK_SECTOR: _STOCK_SECTOR[code] = sec finally: conn.close() for d, name, chg, inflow in rows: _SECTOR_CTX.setdefault(d, {})[name] = {'change': round(chg or 0, 2), 'inflow': round(inflow or 0, 1)} for d in _SECTOR_CTX: ranked = sorted(_SECTOR_CTX[d].items(), key=lambda x: -(x[1]['change'])) total = len(ranked) for rank, (name, v) in enumerate(ranked): v['rank_pct'] = round(rank / total, 3) if total else None # 0=最强 def mkt_ctx(date): return _MKT_CTX.get(date, {}) def sector_ctx(code, date): sec = _STOCK_SECTOR.get(code) if not sec: return {} return _SECTOR_CTX.get(date, {}).get(sec, {}) # ══════════════════════════════════════════════════════ # 入场过滤器 # ══════════════════════════════════════════════════════ def pass_filters(factors, filters): if not filters: return True def chk(key, vmin=None, vmax=None): v = factors.get(key) if vmin is not None and (v is None or v < vmin): return False if vmax is not None and v is not None and v > vmax: return False return True if not chk('rsi', filters.get('rsi_min'), filters.get('rsi_max')): return False if not chk('adx', filters.get('adx_min'), filters.get('adx_max')): return False if not chk('dist_ma20', filters.get('dist_ma20_min'), filters.get('dist_ma20_max')): return False if not chk('vol_ratio', filters.get('vol_ratio_min'), filters.get('vol_ratio_max')): return False if not chk('roc', filters.get('roc_min'), filters.get('roc_max')): return False if not chk('atr_pct', filters.get('atr_pct_min'), filters.get('atr_pct_max')): return False if not chk('macd_hist', filters.get('macd_hist_min'), filters.get('macd_hist_max')): return False # 趋势变化 if not chk('ma20_slope', filters.get('ma20_slope_min'), filters.get('ma20_slope_max')): return False if not chk('macd_hist_delta', filters.get('macd_hist_delta_min'), filters.get('macd_hist_delta_max')): return False if filters.get('adx_rising') and not factors.get('adx_rising'): return False if filters.get('trend_only') and not factors.get('trend_aligned'): return False if filters.get('hh_only') and not factors.get('hh_structure'): return False if filters.get('no_new_high') and factors.get('near_high_20d'): return False # 大盘 if filters.get('mkt_above_ma20') and factors.get('mkt_above_ma20') is not True: return False if not chk('mkt_slope', filters.get('mkt_slope_min'), filters.get('mkt_slope_max')): return False # 行业 if not chk('sector_change', filters.get('sector_change_min'), filters.get('sector_change_max')): return False if not chk('sector_rank_pct', None, filters.get('sector_rank_pct_max')): return False return True def calc_factors(bars, idx): """个股因子: 水平值 + 趋势变化""" b = bars[idx] prev5 = bars[max(0, idx-5)] close = b.get('close') or 0 ma20 = b.get('ma20') or 0 atr = b.get('atr') or 0 vol = b.get('volume') or 0 pvol = prev5.get('volume') or 0 window = bars[max(0, idx-19):idx+1] high20 = max((x.get('high') or 0) for x in window) if window else 0 ma5, ma10 = b.get('ma5') or 0, b.get('ma10') or 0 f = { 'rsi': b.get('rsi'), 'adx': b.get('adx'), 'macd_hist': b.get('macd_hist'), 'roc': b.get('roc'), 'atr_pct': round(atr / close * 100, 2) if close > 0 and atr else None, 'dist_ma20': round((close - ma20) / ma20 * 100, 2) if ma20 > 0 else None, 'vol_ratio': round(vol / pvol, 2) if pvol > 0 else None, 'obv_delta': (b.get('obv') or 0) - (prev5.get('obv') or 0), 'trend_aligned': ma5 > ma10 > ma20 > 0, 'near_high_20d': close >= high20 * 0.98 if high20 > 0 else False, } # 趋势变化因子（不能孤立看点值，要看方向和变化） if idx >= 5: b5 = bars[idx-5] m0, m1 = b5.get('ma20'), b.get('ma20') f['ma20_slope'] = round((m1 - m0) / m0 * 100, 3) if m0 and m1 else None h0, h1 = b5.get('macd_hist'), b.get('macd_hist') f['macd_hist_delta'] = round(h1 - h0, 3) if h0 is not None and h1 is not None else None a0, a1 = b5.get('adx'), b.get('adx') f['adx_rising'] = (a1 > a0) if a0 is not None and a1 is not None else None r0, r1 = b5.get('rsi'), b.get('rsi') f['rsi_delta'] = round(r1 - r0, 2) if r0 is not None and r1 is not None else None if idx >= 10: h5 = max(x.get('high') or 0 for x in bars[idx-4:idx+1]) h10 = max(x.get('high') or 0 for x in bars[idx-9:idx-4]) l5 = min(x.get('low') or 1e9 for x in bars[idx-4:idx+1]) l10 = min(x.get('low') or 1e9 for x in bars[idx-9:idx-4]) f['hh_structure'] = h5 > h10 # 更高的高点 = 上升结构 f['hl_structure'] = l5 > l10 # 更高的低点 = 上升结构 return f # ══════════════════════════════════════════════════════ # 回测引擎（配置驱动 + 12维上下文记录） # ══════════════════════════════════════════════════════ def run_backtest(strategy_version, start_date, end_date, capital=1000000, save=True): strat = get_strategy(strategy_version) cfg = strat['config'] entry_cfg, exit_cfg = cfg['entry'], cfg['exit'] filters = entry_cfg.get('filters', {}) step = cfg.get('eval_step', 5) prepare_market_context(start_date, end_date) prepare_sector_context(start_date, end_date) conn = sqlite3.connect(DB_PATH) stocks = conn.execute(""" SELECT DISTINCT sd.code, COALESCE(s.name, sd.code) as name FROM stock_daily sd LEFT JOIN stocks s ON sd.code = s.code WHERE sd.date>=? AND sd.date<=? """, (start_date, end_date)).fetchall() conn.close() trades = [] screened = scored_n = 0 for code, name in stocks: screened += 1 bars = _bars(code, start_date, end_date) if not bars or len(bars) < 25: continue i = 20 while i < len(bars): window = bars[:i+1] sc = compute_single_score(window) if sc is None: i += step continue total_score, comp = sc scored_n += 1 last = bars[i] close = last.get('close') or 0 if total_score >= entry_cfg['min_score'] and comp['momentum'] >= entry_cfg['min_momentum']: factors = calc_factors(bars, i) # 附加大盘/行业上下文 date = last.get('date') mk = mkt_ctx(date) sc_ctx = sector_ctx(code, date) factors['mkt_above_ma20'] = mk.get('above_ma20') factors['mkt_slope'] = mk.get('ma20_slope') factors['mkt_roc'] = mk.get('roc') factors['sector_change'] = sc_ctx.get('change') factors['sector_rank_pct'] = sc_ctx.get('rank_pct') factors['sector_inflow'] = sc_ctx.get('inflow') if pass_filters(factors, filters): atr_val = last.get('atr') or 0 if exit_cfg.get('tp_pct'): target = close * (1 + exit_cfg['tp_pct']) elif exit_cfg.get('tp_atr') and atr_val > 0: target = close + atr_val * exit_cfg['tp_atr'] elif exit_cfg.get('trail_atr'): target = None # 移动止盈模式无固定目标 else: target = close * 1.10 if exit_cfg.get('sl_atr') and atr_val > 0: stop = close - atr_val * exit_cfg['sl_atr'] elif exit_cfg.get('sl_pct'): stop = close * (1 - exit_cfg['sl_pct']) else: stop = close * 0.93 kelly = 0 if cfg['sizing'].get('kelly'): rr_est = ((target - close) / close) if target else (2 * (close - stop) / close) kelly = compute_kelly(total_score, rr_est if close > 0 else 0.1, (close - stop) / close if close > 0 else 0.07) max_hold = exit_cfg.get('max_hold_days', 20) trail_atr = exit_cfg.get('trail_atr') future = bars[i+1:i+1+max_hold] exit_price = exit_reason = None hold_days = 0 highest_close = close for k, fb in enumerate(future): fh, fl, fc = fb.get('high') or 0, fb.get('low') or 0, fb.get('close') or 0 if target and fh >= target: exit_price, exit_reason, hold_days = target, 'target', k+1 break # 移动止损线：随最高收盘价上移，从不下移 eff_stop = stop if trail_atr and atr_val > 0: highest_close = max(highest_close, fc) eff_stop = max(stop, highest_close - atr_val * trail_atr) if fl <= eff_stop: if trail_atr and eff_stop > stop: exit_price, exit_reason = eff_stop, 'trail' else: exit_price, exit_reason = (eff_stop if trail_atr else fc), 'stop' hold_days = k + 1 break if exit_price is None: exit_price = future[-1].get('close') if future else close exit_reason, hold_days = 'keep', len(future) pnl = (exit_price - close) / close * 100 if close > 0 else 0 trades.append({ 'code': code, 'name': name, 'entry_date': date, 'entry_price': round(close, 2), 'exit_price': round(exit_price, 2), 'profit_pct': round(pnl, 2), 'exit_reason': exit_reason, 'hold_days': hold_days, 'score': total_score, 'score_comp': comp, 'kelly': round(kelly, 3), 'stop_loss': round(stop, 2), 'target': round(target, 2) if target else None, 'factors': {k: (round(v, 3) if isinstance(v, float) else v) for k, v in factors.items()}, }) i += step summary = calc_summary(trades, capital) result = { 'strategy': strat['version'], 'strategy_name': strat['name'], 'period': f"{start_date} ~ {end_date}", 'capital': capital, 'total_stocks_screened': screened, 'scored_events': scored_n, 'trades': trades, 'summary': summary, } if save: save_result(strat, result) return result def calc_summary(trades, capital): if not trades: return {} profits = [t['profit_pct'] for t in trades] wins = [t for t in trades if t['profit_pct'] > 0] losses = [t for t in trades if t['profit_pct'] <= 0] win_rate = len(wins) / len(trades) * 100 avg_p = sum(profits) / len(profits) avg_w = sum(t['profit_pct'] for t in wins) / len(wins) if wins else 0 avg_l = sum(t['profit_pct'] for t in losses) / len(losses) if losses else 0 mean_r = avg_p / 100 std_r = math.sqrt(sum((p/100 - mean_r)**2 for p in profits) / (len(profits)-1)) if len(profits) > 1 else 0 sharpe = mean_r / std_r * math.sqrt(252) if std_r > 0 else 0 curve = [capital] for t in trades: curve.append(curve[-1] * (1 + t['profit_pct']/100)) peak = capital max_dd = 0 for c in curve: peak = max(peak, c) max_dd = max(max_dd, (peak - c) / peak * 100) return { 'total_trades': len(trades), 'win_rate': round(win_rate, 1), 'avg_profit_pct': round(avg_p, 2), 'avg_win_pct': round(avg_w, 2), 'avg_loss_pct': round(avg_l, 2), 'sharpe_ratio': round(sharpe, 2), 'max_drawdown_pct': round(max_dd, 2), 'profit_factor': round(abs(avg_w/avg_l), 2) if avg_l != 0 else None, 'wins': len(wins), 'losses': len(losses), 'capital_end': round(curve[-1], 2), } # ══════════════════════════════════════════════════════ # 因子归因分析（连续分桶 + 布尔分组） # ══════════════════════════════════════════════════════ ANALYZE_FACTORS = ['rsi', 'adx', 'macd_hist', 'roc', 'atr_pct', 'dist_ma20', 'vol_ratio', 'ma20_slope', 'macd_hist_delta', 'rsi_delta', 'mkt_slope', 'mkt_roc', 'sector_change', 'sector_rank_pct', 'score'] BOOL_FACTORS = ['trend_aligned', 'hh_structure', 'hl_structure', 'adx_rising', 'mkt_above_ma20', 'near_high_20d'] def analyze_trades(strategy_version): conn = sqlite3.connect(DB_PATH) row = conn.execute( "SELECT results_json FROM strategy_research WHERE version=? ORDER BY id DESC LIMIT 1", (strategy_version,)).fetchone() conn.close() if not row: return {'error': f'无 {strategy_version} 的回测结果，请先运行回测'} result = json.loads(row[0]) return analyze_trade_list(result.get('trades', []), strategy_version) def analyze_trade_list(trades, label=''): if not trades: return {'error': '无交易数据'} wins = [t for t in trades if t['profit_pct'] > 0] report = { 'label': label, 'total': len(trades), 'wins': len(wins), 'losses': len(trades) - len(wins), 'factors': {}, 'bool_factors': {}, 'exit_reasons': {}, 'hold_analysis': {}, 'insights': [], } # 连续因子: 五分桶胜率 for f in ANALYZE_FACTORS: pairs = [(t['factors'].get(f), t['profit_pct'] > 0) for t in trades if t.get('factors', {}).get(f) is not None] if len(pairs) < 30: continue vals = sorted(pairs, key=lambda x: x[0]) w_vals = [v for v, w in pairs if w] l_vals = [v for v, w in pairs if not w] buckets = [] n = len(vals) for bi in range(5): seg = vals[int(n*bi/5):int(n*(bi+1)/5)] if seg: wr = sum(1 for _, w in seg if w) / len(seg) * 100 buckets.append({'range': f"{seg[0][0]:.2f}~{seg[-1][0]:.2f}", 'win_rate': round(wr, 1), 'count': len(seg)}) report['factors'][f] = { 'winner_mean': round(sum(w_vals)/len(w_vals), 3) if w_vals else None, 'loser_mean': round(sum(l_vals)/len(l_vals), 3) if l_vals else None, 'buckets': buckets, } # 布尔因子: True/False 分组胜率 for f in BOOL_FACTORS: pairs = [(t['factors'].get(f), t['profit_pct'] > 0) for t in trades if t.get('factors', {}).get(f) is not None] if len(pairs) < 30: continue t_grp = [w for v, w in pairs if v] f_grp = [w for v, w in pairs if not v] if t_grp and f_grp: report['bool_factors'][f] = { 'true_win_rate': round(sum(t_grp)/len(t_grp)*100, 1), 'true_count': len(t_grp), 'false_win_rate': round(sum(f_grp)/len(f_grp)*100, 1), 'false_count': len(f_grp), } # 出场方式 for t in trades: r = t['exit_reason'] report['exit_reasons'].setdefault(r, {'count': 0, 'total_pnl': 0, 'avg_hold': 0}) d = report['exit_reasons'][r] d['count'] += 1 d['total_pnl'] += t['profit_pct'] d['avg_hold'] += t['hold_days'] for r, d in report['exit_reasons'].items(): d['avg_pnl'] = round(d['total_pnl'] / d['count'], 2) d['avg_hold'] = round(d['avg_hold'] / d['count'], 1) d['total_pnl'] = round(d['total_pnl'], 1) # 持仓天数 hold_buckets = {} for t in trades: hb = '1-3天' if t['hold_days'] <= 3 else ('4-7天' if t['hold_days'] <= 7 else ('8-14天' if t['hold_days'] <= 14 else '15天+')) hold_buckets.setdefault(hb, {'count': 0, 'wins': 0}) hold_buckets[hb]['count'] += 1 if t['profit_pct'] > 0: hold_buckets[hb]['wins'] += 1 for hb, d in hold_buckets.items(): d['win_rate'] = round(d['wins'] / d['count'] * 100, 1) report['hold_analysis'] = hold_buckets # 自动洞察 ins = [] for f, d in report['factors'].items(): if len(d['buckets']) >= 4: wrs = [b['win_rate'] for b in d['buckets']] spread = max(wrs) - min(wrs) if spread >= 12: best = d['buckets'][wrs.index(max(wrs))] worst = d['buckets'][wrs.index(min(wrs))] ins.append(f"📌 {f} 区分度{spread:.0f}pp: [{best['range']}]胜率{best['win_rate']}% vs [{worst['range']}]胜率{worst['win_rate']}%") for f, d in report['bool_factors'].items(): diff = d['true_win_rate'] - d['false_win_rate'] if abs(diff) >= 8: arrow = '✅' if diff > 0 else '❌' ins.append(f"{arrow} {f}=True 胜率{d['true_win_rate']}% vs False {d['false_win_rate']}% (差{abs(diff):.0f}pp)") er = report['exit_reasons'] if 'stop' in er and er['stop']['count'] > er.get('target', {}).get('count', 0) * 2: ins.append(f"⚠️ 止损({er['stop']['count']})远多于止盈({er.get('target',{}).get('count',0)}): 入场追高或止损过紧") report['insights'] = ins return report # ══════════════════════════════════════════════════════ # 持久化 # ══════════════════════════════════════════════════════ def init_table(): conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS strategy_research ( id INTEGER PRIMARY KEY AUTOINCREMENT, version TEXT, name TEXT, summary TEXT, hypothesis TEXT, parent TEXT, config_json TEXT, results_json TEXT, analysis_json TEXT, period TEXT, created_at TEXT ) """) conn.commit() conn.close() def save_result(strat, result): init_table() conn = sqlite3.connect(DB_PATH) conn.execute(""" INSERT INTO strategy_research (version, name, summary, hypothesis, parent, config_json, results_json, period, created_at) VALUES (?,?,?,?,?,?,?,?,?) """, (strat['version'], strat['name'], strat['summary'], strat['hypothesis'], strat.get('parent'), json.dumps(strat['config'], ensure_ascii=False), json.dumps(result, ensure_ascii=False), result['period'], datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) conn.commit() conn.close() def save_analysis(version, analysis): init_table() conn = sqlite3.connect(DB_PATH) conn.execute(""" UPDATE strategy_research SET analysis_json=? WHERE id = (SELECT id FROM strategy_research WHERE version=? ORDER BY id DESC LIMIT 1) """, (json.dumps(analysis, ensure_ascii=False), version)) conn.commit() conn.close() def list_strategies(): init_table() conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT sr.* FROM strategy_research sr INNER JOIN (SELECT version, MAX(id) as max_id FROM strategy_research GROUP BY version) latest ON sr.id = latest.max_id ORDER BY sr.version """).fetchall() conn.close() out = [] for r in rows: d = dict(r) res = json.loads(d['results_json']) if d.get('results_json') else {} ana = json.loads(d['analysis_json']) if d.get('analysis_json') else None d['summary_stats'] = res.get('summary', {}) d['insights'] = (ana or {}).get('insights', []) d['trades_count'] = len(res.get('trades', [])) del d['results_json'] del d['analysis_json'] out.append(d) existing = {d['version'] for d in out} for v, s in STRATEGIES.items(): if v not in existing: out.append({ 'version': v, 'name': s['name'], 'summary': s['summary'], 'hypothesis': s['hypothesis'], 'parent': s.get('parent'), 'config_json': json.dumps(s['config'], ensure_ascii=False), 'summary_stats': {}, 'insights': [], 'created_at': s.get('created'), }) out.sort(key=lambda x: x['version']) return out if __name__ == '__main__': import sys ver = sys.argv[1] if len(sys.argv) > 1 else 'v3.0' end = '2026-07-24' start = '2026-01-21' r = run_backtest(ver, start, end) print(json.dumps(r['summary'], indent=2, ensure_ascii=False)) a = analyze_trade_list(r['trades'], ver) save_analysis(ver, a) print(json.dumps(a.get('insights', []), indent=2, ensure_ascii=False))