384 lines
16 KiB
Python
384 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
evolution/evolution_engine.py — 策略自我进化引擎（每周六 22:00，hermes cron）
|
||
|
||
设计依据：docs/decisions/2026-08-15-策略自我进化闭环重构.md（老莫已批准）
|
||
闭环：统计数据每日自动更新 → 本引擎每周检测退化 → 生成参数变体 → 回测验证 → 有价值才推送
|
||
|
||
流程：
|
||
1. 读激活策略集合（data/strategy_weights.json：A股 active + 港股 markets.hk.active）
|
||
2. 退化信号检测（宁缺毋滥，任一命中即触发研究）：
|
||
S1 健康度连续低：strategy_health 连续 5 天 health_score < 40（排除 50 中性=无实盘数据）
|
||
S2 温区表现衰减：激活策略在其适应温区（strategy_regime_perf）温区级组合年化 cagr_pct < 0
|
||
3. 有退化 → 生成参数变体：
|
||
- 只对 lab.STRATEGIES 里可回测的策略（v_oversold/v_weak 等标准回测体系）
|
||
- 参数空间从策略 config 实际数值字段出发（递归遍历，单变量 ±20%，一次只动一个）
|
||
- 港股走 hk_backtest（entry/exit 字段 ±20%）
|
||
4. 回测验证（统一资金约束）：
|
||
- A股：lab.run_backtest(save=False)，取 portfolio_full
|
||
- 港股：hk_backtest.gen_trades_defensive + lab.portfolio_sim(max_positions=8)
|
||
- 验收：温区级组合年化 cagr_pct ≥ 原策略 + 3pp 且 max_dd 不劣化超过 2pp
|
||
5. 达标变体 → 写 strategy_evolution（promoted=0）+ XMPP 推送老莫（附对比证据）
|
||
（永不自动 promote，老莫说"上线"才进路由）
|
||
6. 无退化或变体全灭 → 当周静默（不制造噪音）
|
||
|
||
单例守卫：fcntl.flock 防并发（deploy_guard / 手动重跑均安全）
|
||
"""
|
||
import sys, os, json, sqlite3, copy, io, traceback
|
||
from datetime import datetime, timedelta
|
||
|
||
sys.path.insert(0, "/home/hmo/MoFin")
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
|
||
DB = os.environ.get("MOFIN_DB", "/home/hmo/MoFin/data/mofin.db")
|
||
WEIGHTS_JSON = "/home/hmo/MoFin/data/strategy_weights.json"
|
||
|
||
# 退化信号参数
|
||
HEALTH_LOW = 40 # 健康度低于此值视为低
|
||
HEALTH_STREAK_DAYS = 5 # 连续天数
|
||
REGIME_CAGR_BAD = 0.0 # 温区级组合年化低于此值视为退化
|
||
|
||
# 变体生成参数
|
||
VAR_PCT = 0.20 # ±20% 网格
|
||
MAX_VARIANTS = 6 # 每策略最多生成变体数
|
||
MAX_VARIANTS_TEST = 1 # 最多回测验证的变体数（资源约束：单变体2y全市场回测6-8分钟/6-8GB，详见下方BT注释）
|
||
|
||
# 验证回测周期（2026-08-15：原5y全市场回测单变体8+分钟/5GB内存，改为2y控制资源；
|
||
# 验收对比用同周期原策略数据，相对改善仍有效）
|
||
BT_START = "2024-07-01"
|
||
BT_END = "2026-07-24"
|
||
BT_PERIOD_TAG = "2y"
|
||
|
||
# 验收门槛（2026-08-15 口径说明：变体与 parent 用同周期 strategy_research 2y 整体组合年化对比，
|
||
# 相对改善有效；原设计"温区级组合年化"需按温区分段重跑变体，资源过重，整体同口径更务实）
|
||
ACCEPT_CAGR_PP = 3.0 # 组合年化 ≥ 原 + 3pp
|
||
ACCEPT_DD_PP = 2.0 # max_dd 不劣化超过 2pp
|
||
|
||
|
||
def log(msg):
|
||
line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}"
|
||
print(line, flush=True)
|
||
|
||
|
||
# ── 单例守卫（fcntl，Windows 不可用则跳过）──
|
||
try:
|
||
import fcntl
|
||
_LOCK_FD = open("/tmp/evolution_engine.lock", "w")
|
||
try:
|
||
fcntl.flock(_LOCK_FD, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
except OSError:
|
||
log("已有 evolution_engine 实例在运行，退出")
|
||
sys.exit(0)
|
||
except ImportError:
|
||
pass
|
||
|
||
|
||
def get_conn():
|
||
conn = sqlite3.connect(DB)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
|
||
# ── 1. 激活策略集合 ──
|
||
def load_active_strategies():
|
||
"""返回 [(version, market, regime)]：A股 active + 港股 markets.hk.active"""
|
||
try:
|
||
d = json.load(io.open(WEIGHTS_JSON, encoding="utf-8"))
|
||
except Exception as e:
|
||
log(f"读 strategy_weights.json 失败: {e}")
|
||
return []
|
||
out = []
|
||
for v in (d.get("active") or []):
|
||
info = (d.get("weights") or {}).get(v, {})
|
||
out.append({"version": v, "market": "a",
|
||
"regime": info.get("best_regime") or info.get("regime") or d.get("state")})
|
||
hk = (d.get("markets") or {}).get("hk") or {}
|
||
for v in (hk.get("active") or []):
|
||
out.append({"version": v, "market": "hk", "regime": hk.get("state")})
|
||
return out
|
||
|
||
|
||
# ── 2. 退化信号检测 ──
|
||
def detect_degradation(conn, version, market):
|
||
"""返回退化原因列表（空=健康）。S1 健康度连续低；S2 温区组合年化<0"""
|
||
reasons = []
|
||
|
||
# S1：健康度连续 5 天 < 40（排除 50 中性=无实盘）
|
||
rows = conn.execute(
|
||
"SELECT date, health_score FROM strategy_health WHERE strategy_version=? ORDER BY date DESC LIMIT ?",
|
||
(version, HEALTH_STREAK_DAYS)).fetchall()
|
||
if len(rows) >= HEALTH_STREAK_DAYS:
|
||
scores = [r["health_score"] for r in rows]
|
||
# 排除"无实盘=50中性"污染：只要连续5天都 < 40 且不是 50 占位
|
||
if all(s is not None and s < HEALTH_LOW for s in scores) and any(s != 50 for s in scores):
|
||
reasons.append(f"S1 健康度连续{HEALTH_STREAK_DAYS}天<{HEALTH_LOW}（{scores}）")
|
||
|
||
# S2：适应温区温区级组合年化 < 0（strategy_regime_perf.cagr_pct）
|
||
r = conn.execute(
|
||
"SELECT regime, cagr_pct, trades FROM strategy_regime_perf WHERE strategy=? AND market=? ORDER BY trades DESC LIMIT 1",
|
||
(version, market)).fetchone()
|
||
if r and r["cagr_pct"] is not None and r["cagr_pct"] < REGIME_CAGR_BAD:
|
||
reasons.append(f"S2 适应温区[{r['regime']}]组合年化{r['cagr_pct']}%<0（{r['trades']}笔）")
|
||
|
||
return reasons
|
||
|
||
|
||
# ── 3. 变体生成（数据驱动，从实际 config 出发）──
|
||
_NUM_KEYS = ("tp_pct", "sl_pct", "sl_atr", "max_hold_days", "min_score", "min_momentum",
|
||
"adx_min", "atr_pct_min", "atr_pct_max", "roc_min", "roc_max",
|
||
"macd_hist_min", "macd_hist_max", "dist_ma20_min", "vol_ratio_min",
|
||
"vol_ratio_max", "ma20_slope_max", "mkt_slope_max", "mkt_adx_min",
|
||
"sector_slope_max", "bias_max", "rsi_max", "ret_max", "mom20_max",
|
||
"amount_max", "rsi_delta_min", "mkt_rsi_max", "mkt_dd60_max",
|
||
"mcap_q_max", "pe_q_max", "news3_min", "sec_ret20_max",
|
||
"pe_q_max", "mcap_q_max", "sec_ret20_min", "bias60_max",
|
||
"vol_ratio_min", "rsi_delta_min", "bias60_min")
|
||
_SKIP_KEYS = ("mode", "family", "launch", "version", "name", "summary", "hypothesis",
|
||
"mkt_mode", "mkt_above_ma20", "hh_only", "hl_only", "sector_above_ma20")
|
||
|
||
|
||
def iter_numeric_fields(node, path=()):
|
||
"""递归遍历 config，产出 (path_list, field_name, value) 数值字段"""
|
||
if isinstance(node, dict):
|
||
for k, v in node.items():
|
||
if k in _SKIP_KEYS:
|
||
continue
|
||
if isinstance(v, (int, float)) and not isinstance(v, bool) and k in _NUM_KEYS:
|
||
yield list(path) + [k], k, v
|
||
elif isinstance(v, dict):
|
||
yield from iter_numeric_fields(v, list(path) + [k])
|
||
|
||
|
||
def get_parent_cagr(conn, version, market, period_tag=BT_PERIOD_TAG):
|
||
"""原策略基准：优先同周期 strategy_research（period_tag=2y），无则温区级组合年化（全量）"""
|
||
if market == "a":
|
||
r = conn.execute(
|
||
"SELECT results_json FROM strategy_research WHERE version=? AND period_tag=? "
|
||
"AND market='a' ORDER BY id DESC LIMIT 1", (version, period_tag)).fetchone()
|
||
if r:
|
||
res = json.loads(r["results_json"] or "{}")
|
||
s = res.get("summary", {})
|
||
pf = s.get("portfolio_full", {})
|
||
cagr = pf.get("cagr_pct")
|
||
dd = pf.get("portfolio_max_dd_pct")
|
||
if cagr is not None:
|
||
return cagr, dd
|
||
# 回退：温区级组合年化（strategy_regime_perf，全量）——仅当同周期数据缺失时
|
||
r = conn.execute(
|
||
"SELECT cagr_pct, portfolio_max_dd_pct FROM strategy_regime_perf WHERE strategy=? AND market=? ORDER BY trades DESC LIMIT 1",
|
||
(version, market)).fetchone()
|
||
if r:
|
||
return r["cagr_pct"], r["portfolio_max_dd_pct"]
|
||
return None, None
|
||
|
||
|
||
def generate_variants(version, market, config):
|
||
"""生成变体参数建议：单变量 ±20%，最多 MAX_VARIANTS 个
|
||
返回 [{version, name, config, change_desc, field, delta}]"""
|
||
fields = list(iter_numeric_fields(config))
|
||
if not fields:
|
||
return []
|
||
variants = []
|
||
for path, fname, val in fields:
|
||
if val <= 0:
|
||
continue
|
||
for factor, tag in [(1 - VAR_PCT, "减20%"), (1 + VAR_PCT, "加20%")]:
|
||
new_val = round(val * factor, 4)
|
||
if new_val <= 0:
|
||
continue
|
||
# 克隆 config 并修改目标字段
|
||
new_cfg = copy.deepcopy(config)
|
||
node = new_cfg
|
||
for p in path[:-1]:
|
||
node = node[p]
|
||
node[path[-1]] = new_val
|
||
variants.append({
|
||
"version": f"evo_{version}_{fname}_{tag.replace('20%','')}{round(new_val, 2)}",
|
||
"name": f"自进化-{version}-{fname}{tag}",
|
||
"config": new_cfg,
|
||
"change_desc": f"{fname}: {val} → {new_val}（{tag}）",
|
||
"field": fname,
|
||
"delta": round(new_val - val, 4),
|
||
})
|
||
if len(variants) >= MAX_VARIANTS:
|
||
return variants
|
||
return variants
|
||
|
||
|
||
# ── 4. 回测验证 ──
|
||
def verify_variant_a(variant, parent_version):
|
||
"""A股变体验证：注册进 lab 跑回测（save=False），返回 summary 关键指标"""
|
||
import strategy_lab as lab
|
||
name = variant["version"]
|
||
base = lab.get_strategy(parent_version)
|
||
cfg = copy.deepcopy(base)
|
||
cfg["version"] = name
|
||
cfg["name"] = variant["name"]
|
||
# 用变体 config 覆盖（变体 config 从原 config 克隆并改了一个字段）
|
||
merged = copy.deepcopy(base["config"])
|
||
_deep_update(merged, variant["config"])
|
||
cfg["config"] = merged
|
||
lab.STRATEGIES[name] = cfg
|
||
try:
|
||
r = lab.run_backtest(name, BT_START, BT_END, 913000, save=False,
|
||
universe="a", period_tag=BT_PERIOD_TAG)
|
||
s = r.get("summary", {})
|
||
pf = s.get("portfolio_full", {})
|
||
return {
|
||
"trades": s.get("total_trades"),
|
||
"win_rate": s.get("win_rate"),
|
||
"cagr": pf.get("cagr_pct"),
|
||
"total_return": pf.get("total_return_pct"),
|
||
"max_dd": pf.get("portfolio_max_dd_pct"),
|
||
}
|
||
finally:
|
||
lab.STRATEGIES.pop(name, None)
|
||
|
||
|
||
def _deep_update(dst, src):
|
||
for k, v in src.items():
|
||
if isinstance(v, dict) and isinstance(dst.get(k), dict):
|
||
_deep_update(dst[k], v)
|
||
else:
|
||
dst[k] = v
|
||
|
||
|
||
def verify_variant_hk(variant, parent_version):
|
||
"""港股变体验证：hk_backtest 生成交易 + portfolio_sim 8槽"""
|
||
import pandas as pd
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
from hk_strategies import HK_STRATEGIES, get_hk_strategy
|
||
import hk_backtest
|
||
import strategy_lab as lab
|
||
|
||
base = get_hk_strategy(parent_version)
|
||
if not base:
|
||
return None
|
||
new_cfg = copy.deepcopy(base)
|
||
new_cfg["version"] = variant["version"]
|
||
new_cfg["name"] = variant["name"]
|
||
_deep_update(new_cfg, variant["config"])
|
||
panel = hk_backtest.load_panel()
|
||
# 2y 窗口过滤（与 A股验证周期一致，控制资源）
|
||
panel = panel[(panel["date"] >= BT_START) & (panel["date"] <= BT_END)].copy()
|
||
trades = hk_backtest.gen_trades_defensive(panel, new_cfg, strike=3, cooldown_days=15)
|
||
if not trades:
|
||
return {"trades": 0, "win_rate": None, "cagr": None, "total_return": None, "max_dd": None}
|
||
sim = lab.portfolio_sim(trades, 1000000, max_positions=8)
|
||
return {
|
||
"trades": len(trades),
|
||
"win_rate": round(100 * sum(1 for t in trades if t["profit_pct"] > 0) / len(trades), 1),
|
||
"cagr": sim.get("cagr_pct"),
|
||
"total_return": sim.get("total_return_pct"),
|
||
"max_dd": sim.get("portfolio_max_dd_pct"),
|
||
}
|
||
|
||
|
||
# ── 5. 记录 + 推送 ──
|
||
def record_and_notify(conn, parent_version, market, variant, result, parent_cagr, parent_dd):
|
||
"""写 strategy_evolution + XMPP 推送"""
|
||
conn.execute("""
|
||
INSERT INTO strategy_evolution (parent_version, child_version, change_description, backtest_result, promoted)
|
||
VALUES (?, ?, ?, ?, 0)
|
||
""", (parent_version, variant["version"], variant["change_desc"],
|
||
json.dumps(result, ensure_ascii=False)))
|
||
conn.commit()
|
||
|
||
msg = (f"🧬 策略进化建议 [{parent_version}]\n"
|
||
f"改动: {variant['change_desc']}\n"
|
||
f"回测: 年化 {parent_cagr}% → {result.get('cagr')}%"
|
||
f" (Δ{round((result.get('cagr') or 0) - (parent_cagr or 0), 1)}pp)"
|
||
f" | 回撤 {parent_dd}% → {result.get('max_dd')}%\n"
|
||
f"胜率 {result.get('win_rate')}% / {result.get('trades')}笔\n"
|
||
f"【验证达标，待你决定是否上线】")
|
||
try:
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
from alert_helper import notify, ACTION
|
||
notify("策略进化", msg, level=ACTION)
|
||
log(f"XMPP 推送: {parent_version} → {variant['version']}")
|
||
except Exception as e:
|
||
log(f"XMPP 推送失败: {e}")
|
||
return msg
|
||
|
||
|
||
# ── 主流程 ──
|
||
def run_evolution():
|
||
conn = get_conn()
|
||
actives = load_active_strategies()
|
||
log(f"激活策略: {[a['version'] for a in actives]}")
|
||
if not actives:
|
||
log("无激活策略，退出")
|
||
conn.close()
|
||
return
|
||
|
||
findings = [] # 退化发现
|
||
passed = [] # 达标变体
|
||
|
||
for act in actives:
|
||
v, mkt = act["version"], act["market"]
|
||
reasons = detect_degradation(conn, v, mkt)
|
||
if not reasons:
|
||
continue
|
||
log(f"退化信号: {v} [{mkt}] → {'; '.join(reasons)}")
|
||
findings.append((v, mkt, reasons))
|
||
|
||
# 生成变体（A股从 lab 读 config；港股从 HK_STRATEGIES）
|
||
if mkt == "hk":
|
||
sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
from hk_strategies import get_hk_strategy
|
||
base = get_hk_strategy(v)
|
||
if not base:
|
||
log(f" {v} 无港股策略定义，跳过")
|
||
continue
|
||
variants = generate_variants(v, mkt, base)
|
||
verify_fn = verify_variant_hk
|
||
else:
|
||
try:
|
||
import strategy_lab as lab
|
||
base = lab.get_strategy(v)
|
||
except ValueError:
|
||
log(f" {v} 不在标准回测体系（scanner 类策略），跳过变体研究")
|
||
continue
|
||
variants = generate_variants(v, mkt, base["config"])
|
||
verify_fn = verify_variant_a
|
||
if not variants:
|
||
log(f" {v} 无可用变体字段，跳过")
|
||
continue
|
||
|
||
parent_cagr, parent_dd = get_parent_cagr(conn, v, mkt)
|
||
log(f" {v} 原温区年化 {parent_cagr}% / 回撤 {parent_dd}% | 生成 {len(variants)} 个变体，验证前 {MAX_VARIANTS_TEST} 个")
|
||
for var in variants[:MAX_VARIANTS_TEST]:
|
||
try:
|
||
res = verify_fn(var, v)
|
||
except Exception as e:
|
||
log(f" {var['version']} 回测失败: {str(e)[:100]}")
|
||
continue
|
||
if not res or res.get("cagr") is None:
|
||
log(f" {var['version']} 无结果（0笔或空），跳过")
|
||
continue
|
||
ok_cagr = parent_cagr is None or res["cagr"] >= (parent_cagr or 0) + ACCEPT_CAGR_PP
|
||
ok_dd = parent_dd is None or res["max_dd"] <= (parent_dd or 0) + ACCEPT_DD_PP
|
||
status = "✅达标" if (ok_cagr and ok_dd) else "❌不达标"
|
||
log(f" {var['version']}: 年化 {parent_cagr}→{res['cagr']}% 回撤 {parent_dd}→{res['max_dd']}% [{status}]")
|
||
if ok_cagr and ok_dd:
|
||
record_and_notify(conn, v, mkt, var, res, parent_cagr, parent_dd)
|
||
passed.append((v, var, res))
|
||
|
||
conn.close()
|
||
|
||
# 汇总
|
||
if not findings:
|
||
log("── 无退化信号，当周静默 ──")
|
||
else:
|
||
log(f"── 检测 {len(findings)} 个退化策略，{len(passed)} 个达标变体已推送 ──")
|
||
return findings, passed
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
run_evolution()
|
||
except Exception as e:
|
||
log(f"evolution_engine 异常: {e}")
|
||
traceback.print_exc()
|
||
sys.exit(1)
|