218 lines
9.7 KiB
Python
218 lines
9.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""evolution/merge_b_group.py — AB融合机制（2026-08-16 方向二闭环）
|
||
|
||
老莫：B组候选与A组对照后，融合/合并成为最终实施组（新的A组）。
|
||
|
||
流程：
|
||
1. 读 B 组 verified 候选（data/b_group_candidates.json, status='verified'）
|
||
2. 老莫选择要融合的候选 → 注册为正式策略版本：
|
||
- A股：写入 strategy_research（results_json 用回测验证的 trades）
|
||
- 港股：注册进 hk_strategies.py（entry 条件）
|
||
3. 加入候选池（strategy_weights 路由可识别）
|
||
4. 手动可用性把关（老莫决定是否启用）——融合≠自动上线
|
||
|
||
安全：不自动 promote，不自动启用；融合只是把候选变成"可用的新策略版本"。
|
||
"""
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
import sqlite3
|
||
from datetime import datetime
|
||
|
||
DATA_DIR = "/home/hmo/MoFin/data"
|
||
CAND_JSON = f"{DATA_DIR}/b_group_candidates.json"
|
||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||
|
||
|
||
def load_candidates():
|
||
try:
|
||
d = json.load(open(CAND_JSON, encoding="utf-8"))
|
||
return d.get("candidates", [])
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def get_verified():
|
||
return [c for c in load_candidates() if c.get("status") == "verified"]
|
||
|
||
|
||
def strategy_name(cand):
|
||
"""生成策略版本名：b{regime缩写}{序号}"""
|
||
rg_map = {"trend_up": "tu", "choppy": "ch", "trend_down": "td"}
|
||
rg = rg_map.get(cand.get("regime"), "x")
|
||
idx = cand.get("_idx", 1)
|
||
return f"b_{rg}{idx}"
|
||
|
||
|
||
def register_a_share(cand):
|
||
"""A股候选注册：写入 strategy_research（B组候选，供研究Tab/回测）
|
||
实际回测验证由进化引擎跑，这里先注册占位 + 候选条件记录
|
||
"""
|
||
conn = sqlite3.connect(DB, timeout=10)
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
name = strategy_name(cand)
|
||
# 检查是否已注册
|
||
exist = conn.execute("SELECT 1 FROM strategy_research WHERE version=? LIMIT 1", (name,)).fetchone()
|
||
if exist:
|
||
conn.close()
|
||
return {"status": "exists", "version": name}
|
||
conn.execute("""
|
||
INSERT INTO strategy_research (version, name, summary, hypothesis, parent, config_json,
|
||
results_json, analysis_json, period, created_at, market, period_tag, deprecated)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||
""", (name, f"B组-{cand.get('regime','')}", cand.get("hypothesis", ""),
|
||
"B组融合候选（由果及因挖掘）", "B组", json.dumps(cand.get("entry", {})),
|
||
json.dumps({"summary": {"total_trades": cand.get("trades_est"),
|
||
"win_rate": cand.get("sim_win_rate"),
|
||
"avg_profit_pct": cand.get("sim_avg_pnl")}}),
|
||
None, None, now, "a", "2y", None))
|
||
conn.commit()
|
||
conn.close()
|
||
return {"status": "registered", "version": name}
|
||
|
||
|
||
def register_hk(cand):
|
||
"""港股候选注册：追加到 hk_strategies.py"""
|
||
name = strategy_name(cand)
|
||
entry = cand.get("entry", {})
|
||
# 追加到 hk_strategies.py（先读再写）
|
||
path = "/home/hmo/MoFin/deploy/profile-scripts/hk_strategies.py"
|
||
src = open(path, encoding="utf-8").read()
|
||
if f'"{name}"' in src:
|
||
return {"status": "exists", "version": name}
|
||
new_block = f'''
|
||
"{name}": {{
|
||
"version": "{name}",
|
||
"name": "B组-{cand.get('regime','')}（由果及因融合）",
|
||
"regime": "{cand.get('regime','all')}",
|
||
"summary": "{cand.get('hypothesis','B组候选')[:80]}",
|
||
"entry": {json.dumps(entry, ensure_ascii=False)},
|
||
"exit": {{"tp_pct": 0.10, "sl_pct": 0.05, "max_hold_days": 20}},
|
||
}},
|
||
}}'''
|
||
# 在 HK_STRATEGIES 的收尾 "}" 前插入（精确：找最后一个顶层 dict 的收尾）
|
||
# HK_STRATEGIES 结构：{ "k1": {...}, ..., "kn": {...}, } 然后空行 + get_hk_strategy
|
||
marker = "\n\n\ndef get_hk_strategy"
|
||
idx = src.rfind(marker)
|
||
if idx == -1:
|
||
return {"status": "error", "version": name, "error": "hk_strategies 结构异常"}
|
||
insert_at = src.rfind("}", 0, idx)
|
||
# 去掉 new_block 末尾多余的 }}
|
||
clean_block = new_block.rstrip()
|
||
if clean_block.endswith("}}"):
|
||
clean_block = clean_block[:-1]
|
||
src = src[:insert_at] + clean_block + src[insert_at:]
|
||
open(path, "w", encoding="utf-8").write(src)
|
||
return {"status": "registered", "version": name}
|
||
|
||
|
||
def merge(version=None):
|
||
"""融合：把 verified 候选注册为策略版本。version 指定要融合的候选，None=全部"""
|
||
verified = get_verified()
|
||
if not verified:
|
||
return {"error": "无 verified B组候选（需先通过模拟验证门槛）", "verified": 0}
|
||
out = []
|
||
for i, cand in enumerate(verified):
|
||
if version and cand.get("version_name") != version:
|
||
continue
|
||
cand["_idx"] = i + 1
|
||
if cand.get("market") == "hk":
|
||
r = register_hk(cand)
|
||
else:
|
||
r = register_a_share(cand)
|
||
r["candidate"] = cand.get("hypothesis", "")
|
||
# 融合链路：多周期trades + 温区预计算 + 资格评估 + 可用性初始化
|
||
if r.get("status") in ("registered", "exists") and r.get("version"):
|
||
try:
|
||
link = _post_merge_chain(r["version"], cand)
|
||
r["chain"] = link
|
||
except Exception as e:
|
||
r["chain"] = {"error": str(e)}
|
||
out.append(r)
|
||
return {"merged": out}
|
||
|
||
|
||
def _post_merge_chain(version, cand):
|
||
"""融合后链路：按period_tag生成窗口trades → 温区预计算 → 资格评估 → 可用性
|
||
返回 {period_trades: {...}, regime_records: n, qualification: {...}, availability: {...}}"""
|
||
import subprocess, json as _json
|
||
out = {}
|
||
# 1) 生成各周期窗口trades 写入 strategy_research（每个 period_tag 记录独立 results_json）
|
||
# （B组候选的 trades 来自模拟验证，按 entry_date 过滤窗口）
|
||
try:
|
||
import sys as _sys
|
||
_sys.path.insert(0, "/home/hmo/MoFin")
|
||
_sys.path.insert(0, "/home/hmo/MoFin/evolution")
|
||
import sqlite3 as _sq
|
||
import pandas as _pd
|
||
from datetime import datetime as _dt, timedelta as _td
|
||
from b_group_miner import _simulate_verify
|
||
market = cand.get("market", "a")
|
||
regime = cand.get("regime", "trend_down")
|
||
entry = cand.get("entry", {})
|
||
panel_path = "/tmp/panel_12d_hk.pkl" if market == "hk" else "/tmp/panel_12d.pkl"
|
||
panel = _pd.read_pickle(panel_path)
|
||
panel = panel.sort_values(["code", "date"]).reset_index(drop=True)
|
||
panel["fwd_ret60"] = panel.groupby("code")["close"].transform(lambda x: x.shift(-60) / x - 1) * 100
|
||
cond = _pd.Series(True, index=panel.index)
|
||
for feat, val in entry.items():
|
||
if "_min" in feat:
|
||
cond &= panel[feat.replace("_min", "")] >= val
|
||
elif "_max" in feat:
|
||
cond &= panel[feat.replace("_max", "")] < val
|
||
elif feat in panel.columns:
|
||
cond &= panel[feat] == val
|
||
tp = int(cand.get("sim_tp", 15)); sl = int(cand.get("sim_sl", 8)); mh = int(cand.get("sim_maxh", 35))
|
||
r = _simulate_verify(market, regime, panel, cond, tp=tp, sl=sl, maxh=mh)
|
||
if not r:
|
||
out["period_trades"] = {"error": "模拟验证无结果"}
|
||
else:
|
||
all_trades = r["trades"]
|
||
latest_dt = _dt.strptime(max(t["entry_date"] for t in all_trades), "%Y-%m-%d")
|
||
conn = _sq.connect("/home/hmo/MoFin/data/mofin.db", timeout=30)
|
||
for pt, yrs in [("1y", 1), ("2y", 2), ("5y", 5), ("10y", 10)]:
|
||
cutoff = (latest_dt - _td(days=365 * yrs)).strftime("%Y-%m-%d")
|
||
wt = [t for t in all_trades if t["entry_date"] >= cutoff]
|
||
n = len(wt)
|
||
wins = [t for t in wt if t.get("profit_pct", 0) > 0]
|
||
wr = round(len(wins) / n * 100, 1) if n else 0
|
||
avg = round(sum(t.get("profit_pct", 0) for t in wt) / n, 2) if n else 0
|
||
results = {"summary": {"total_trades": n, "win_rate": wr, "avg_profit_pct": avg},
|
||
"trades": wt[:5000],
|
||
"sim_params": {"tp": tp, "sl": sl, "maxh": mh},
|
||
"window": {"cutoff": cutoff, "latest": max(t["entry_date"] for t in all_trades)}}
|
||
conn.execute("UPDATE strategy_research SET results_json=? WHERE version=? AND period_tag=?",
|
||
(_json.dumps(results, ensure_ascii=False), version, pt))
|
||
out.setdefault("period_trades", {})[pt] = {"n": n, "win_rate": wr}
|
||
conn.commit(); conn.close()
|
||
except Exception as e:
|
||
out["period_trades"] = {"error": str(e)}
|
||
# 2) 温区预计算
|
||
try:
|
||
mkt_flag = "--market=hk" if market == "hk" else "--market=a"
|
||
p = subprocess.run(["/home/hmo/MoFin/venv/bin/python",
|
||
"/home/hmo/MoFin/deploy/profile-scripts/regime_perf_by_period.py",
|
||
mkt_flag, "--periods=1y 2y 5y 10y"],
|
||
capture_output=True, text=True, timeout=900)
|
||
out["regime_run"] = {"rc": p.returncode, "tail": (p.stdout or "").strip().splitlines()[-1:]}
|
||
except Exception as e:
|
||
out["regime_run"] = {"error": str(e)}
|
||
# 3) 资格评估 + 可用性
|
||
try:
|
||
_sys.path.insert(0, "/home/hmo/MoFin/deploy/profile-scripts")
|
||
import strategy_qualify as sq
|
||
out["qualification"] = sq.evaluate_all_regimes(version, market=market)
|
||
sq.auto_init_availability([version])
|
||
av = sq.load_availability().get(version)
|
||
out["availability"] = av
|
||
except Exception as e:
|
||
out["qualification"] = {"error": str(e)}
|
||
return out
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
v = sys.argv[1] if len(sys.argv) > 1 else None
|
||
res = merge(v)
|
||
print(json.dumps(res, ensure_ascii=False, indent=1))
|