feat: AB融合机制(merge_b_group)——B组verified候选注册为策略版本(A股入库/hk_strategies),融合≠自动上线,老莫可用性把关

evolution/merge_b_group.py

@@ -0,0 +1,131 @@
+# -*- 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 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", "")
+        out.append(r)
+    return {"merged": 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))