hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 策略实验室多版本体系 — v1~v4.1迭代+12维上下文(大盘/趋势变化)+因子归因+版本对比列表

Browse Source
This commit is contained in:
hmo
2026-07-28 23:57:23 +08:00
parent 62305cb062
commit 3acbd6f4fd
3 changed files with 822 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strategy_lab.py
+628
View File
@@ -0,0 +1,628 @@
#!/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
# ══════════════════════════════════════════════════════
# 策略版本注册表
# ══════════════════════════════════════════════════════
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,
},
},
}
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 = prepare_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']
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'):
kelly = compute_kelly(total_score, (target-close)/close if close>0 else 0.1,
(close-stop)/close if close>0 else 0.07)
max_hold = exit_cfg.get('max_hold_days', 20)
future = bars[i+1:i+1+max_hold]
exit_price = exit_reason = None
hold_days = 0
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 fh >= target:
exit_price, exit_reason, hold_days = target, 'target', k+1
break
elif fl <= stop:
exit_price, exit_reason, hold_days = fc, 'stop', 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),
'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))