hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0298aa7660b7c2ee9ea08343e9013a92207fad30
MoFin/strategy_lab.py
T

830 lines
39 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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": 1,
},
},
"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}),
# E组: 资金面/板块增强
"v6.0": _v40_branch("v6.0", "资金流过滤",
"v4.0f + 资金5日净占比>-2.5(排除持续流出)",
"资金流归因：flow_5d<-2.7的持续流出组胜率仅37.5%，其余各桶55-78%。排除主力持续出逃的标的；flow_delta>4虽77.8%但会过度砍样本暂不启用",
entry_overrides={"vol_ratio_min": 0.9, "vol_ratio_max": 2.0,
"flow_5d_min": -2.5}),
"v6.1": _v40_branch("v6.1", "板块不追高",
"v4.0f + 板块MA20斜率≤1.0(排除已强涨板块)",
"板块归因：sector_slope>1.04的强涨板块入场胜率仅27.3%，而斜率≤0.08的回调/横盘板块胜率80%。与大盘/个股层面的'回调买'规律三层同构",
entry_overrides={"vol_ratio_min": 0.9, "vol_ratio_max": 2.0,
"sector_slope_max": 1.0}),
"v6.2": _v40_branch("v6.2", "资金+板块双滤",
"v4.0f + 资金5日净占比>-2.5 + 板块MA20斜率≤1.0",
"资金流(37.5%→55-78%)与板块(27.3%→80%)两个独立维度的负向排除叠加，期望在v4.0f基础上再提升胜率且不显著减样本",
entry_overrides={"vol_ratio_min": 0.9, "vol_ratio_max": 2.0,
"flow_5d_min": -2.5, "sector_slope_max": 1.0}),
})
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):
"""行业上下文: sector_index_daily(全历史) 提供板块趋势; sector_snapshots(近期) 补充净流入"""
global _SECTOR_CTX, _STOCK_SECTOR
_SECTOR_CTX, _STOCK_SECTOR = {}, {}
conn = sqlite3.connect(DB_PATH)
try:
# 1. 板块指数历史（全周期）
try:
idx_rows = conn.execute("""
SELECT sector, date, close, change_pct FROM sector_index_daily
WHERE date >= ? AND date <= ? ORDER BY sector, date
""", (start_date, end_date)).fetchall()
except sqlite3.OperationalError:
idx_rows = []
# 每板块计算 MA20 和斜率
from collections import defaultdict
by_sector = defaultdict(list)
for sec, d, close, chg in idx_rows:
by_sector[sec].append((d, close, chg))
for sec, series in by_sector.items():
closes = [c for _, c, _ in series]
for i, (d, close, chg) in enumerate(series):
above = slope = None
if i >= 19:
ma20 = sum(closes[i-19:i+1]) / 20
above = close > ma20
if i >= 24:
ma20_5 = sum(closes[i-24:i-4]) / 20
if ma20_5 > 0:
slope = round((ma20 - ma20_5) / ma20_5 * 100, 3)
_SECTOR_CTX.setdefault(d, {})[sec] = {
'change': chg, 'above_ma20': above, 'slope': slope,
}
# 2. sector_snapshots 补充净流入和涨幅（近期，THS命名）
snap_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()
for d, name, chg, inflow in snap_rows:
e = _SECTOR_CTX.setdefault(d, {}).setdefault(name, {})
e['inflow'] = round(inflow or 0, 1)
if 'change' not in e or e.get('change') is None:
e['change'] = round(chg or 0, 2)
# 3. 个股→行业映射：优先 EM 体系（覆盖全，与 sector_index_daily 对齐），THS 兜底
try:
for code, sec in conn.execute("SELECT code, sector FROM stock_sectors_em").fetchall():
_STOCK_SECTOR[code] = sec
except sqlite3.OperationalError:
pass
for code, sec, src in conn.execute(
"SELECT code, sector_name, source FROM stock_sectors").fetchall():
if code not in _STOCK_SECTOR and src == 'ths':
_STOCK_SECTOR[code] = sec
finally:
conn.close()
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, {})
# ══════════════════════════════════════════════════════
# 资金面上下文（stock_capital_flow 表）
# ══════════════════════════════════════════════════════
_FLOW_CTX = {} # code -> {date: main_pct}
_FLOW_SORTED = {} # code -> sorted list of (date, main_pct)
def prepare_flow_context(start_date, end_date):
"""加载个股主力资金净流入净占比历史"""
global _FLOW_CTX, _FLOW_SORTED
_FLOW_CTX, _FLOW_SORTED = {}, {}
conn = sqlite3.connect(DB_PATH)
try:
rows = conn.execute("""
SELECT code, date, main_pct FROM stock_capital_flow
WHERE date >= ? AND date <= ?
""", (start_date, end_date)).fetchall()
except sqlite3.OperationalError:
rows = [] # 表不存在时容忍
finally:
conn.close()
for code, d, pct in rows:
_FLOW_CTX.setdefault(code, {})[d] = pct
for code, dmap in _FLOW_CTX.items():
_FLOW_SORTED[code] = sorted(dmap.items())
def flow_ctx(code, date, bars_dates, idx):
"""资金流因子: 当日净占比 / 5日均值 / 5日趋势"""
series = _FLOW_SORTED.get(code)
if not series:
return {}
dmap = _FLOW_CTX[code]
# 找 date 之前(含)最近5个资金流数据点
dates = [d for d, _ in series if d <= date]
if not dates:
return {}
recent = dates[-5:]
prior = dates[-10:-5]
f = {'flow_pct': dmap.get(dates[-1])}
if recent:
vals = [dmap[d] for d in recent if dmap.get(d) is not None]
f['flow_5d'] = round(sum(vals) / len(vals), 2) if vals else None
if recent and prior:
v5 = [dmap[d] for d in recent if dmap.get(d) is not None]
p5 = [dmap[d] for d in prior if dmap.get(d) is not None]
if v5 and p5:
f['flow_delta'] = round(sum(v5)/len(v5) - sum(p5)/len(p5), 2)
return f
# ══════════════════════════════════════════════════════
# 入场过滤器
# ══════════════════════════════════════════════════════
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
if not chk('sector_slope', filters.get('sector_slope_min'), filters.get('sector_slope_max')): return False
if filters.get('sector_above_ma20') and factors.get('sector_above_ma20') is not True: return False
# 资金面
if not chk('flow_pct', filters.get('flow_pct_min'), filters.get('flow_pct_max')): return False
if not chk('flow_5d', filters.get('flow_5d_min'), filters.get('flow_5d_max')): return False
if not chk('flow_delta', filters.get('flow_delta_min'), filters.get('flow_delta_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)
# 指标预热：取数往前多取120天，保证窗口首日 MA60/RSI/ATR 等已收敛
fetch_start = (datetime.strptime(start_date, '%Y-%m-%d') - timedelta(days=120)).strftime('%Y-%m-%d')
prepare_market_context(fetch_start, end_date)
prepare_sector_context(start_date, end_date)
prepare_flow_context(fetch_start, 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, fetch_start, end_date)
if not bars or len(bars) < 25:
continue
i = 20
while i < len(bars):
# 只在考察窗口内产生交易，预热期 bars 仅供指标计算
if bars[i].get('date', '') < start_date:
i += 1
continue
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')
factors['sector_above_ma20'] = sc_ctx.get('above_ma20')
factors['sector_slope'] = sc_ctx.get('slope')
# 资金面因子
fl = flow_ctx(code, date, None, i)
factors.update(fl)
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', 'sector_slope', 'flow_pct', 'flow_5d', 'flow_delta', 'score']
BOOL_FACTORS = ['trend_aligned', 'hh_structure', 'hl_structure', 'adx_rising',
'mkt_above_ma20', 'near_high_20d', 'sector_above_ma20']
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))
Reference in New Issue View Git Blame Copy Permalink