feat: B组策略挖掘管线(b_group_miner)——由果及因按温区扫描因子组合,产出B组候选供AB对照

evolution/b_group_miner.py

@@ -0,0 +1,117 @@
+# -*- coding: utf-8 -*-
+"""evolution/b_group_miner.py — B组挖掘 v3（务实版）
+聚焦缺口温区（A股震荡市/港股下跌市），用【短期反弹】作果（fwd_ret10>=8% 或 fwd_ret20>=12%），
+在超跌+企稳因子池上扫描组合。短期反弹样本充足，能稳定产出候选。
+"""
+import json
+import sqlite3
+import numpy as np
+import pandas as pd
+from datetime import datetime
+
+DATA_DIR = "/home/hmo/MoFin/data"
+OUT_JSON = f"{DATA_DIR}/b_group_candidates.json"
+
+
+def load_regime_map(market="a"):
+    conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
+    rows = conn.execute("SELECT date, regime FROM market_regime WHERE market=?", (market,)).fetchall()
+    conn.close()
+    return {d: r for d, r in rows}
+
+
+def load_panel(market):
+    path = "/tmp/panel_12d_hk.pkl" if market == "hk" else "/tmp/panel_12d.pkl"
+    p = pd.read_pickle(path)
+    p = p.sort_values(["code", "date"]).reset_index(drop=True)
+    p["fwd_ret10"] = p.groupby("code")["close"].transform(lambda x: x.shift(-10) / x - 1) * 100
+    p["fwd_ret20"] = p.groupby("code")["close"].transform(lambda x: x.shift(-20) / x - 1) * 100
+    return p
+
+
+def scan(market, regime, panel, target="ret10", min_n=200):
+    """扫描超跌+企稳因子组合的短期反弹概率"""
+    rm = load_regime_map(market)
+    p = panel.copy()
+    p["_regime"] = p["date"].map(rm)
+    col = "fwd_ret10" if target == "ret10" else "fwd_ret20"
+    sub = p[p["_regime"] == regime].dropna(subset=[col])
+    if len(sub) < min_n:
+        return []
+    th = 8 if target == "ret10" else 12
+    sub["is_ok"] = (sub[col] >= th).astype(int)
+    br = sub["is_ok"].mean() * 100
+    print(f"[{market}/{regime}] 样本{len(sub)} 基线短期反弹率({th}%/{target}){br:.1f}%")
+
+    # 因子：超跌 + 企稳 + 小盘低估值（温区通用的候选）
+    factor_defs = {
+        "bias60": ("<", -5), "bias60_deep": ("<", -15), "rsi": ("<", 40),
+        "rsi_shallow": ("<", 55), "dist_lo20": (">", 3), "vol_ratio": (">", 1.0),
+        "ret5": (">", -3), "mcap_q": ("<", 0.3), "pe_q": ("<", 0.3),
+        "sec_ret20": ("<", 0), "mkt_rsi": ("<", 50), "mkt_ret20": ("<", -3),
+    }
+    single = []
+    for feat, (op, val) in factor_defs.items():
+        if feat not in sub.columns:
+            continue
+        cond = sub[feat] < val if op == "<" else sub[feat] > val
+        m = sub[cond]
+        if len(m) < 100:
+            continue
+        rate = m["is_ok"].mean() * 100
+        if rate > br + 2:
+            single.append((feat, round(rate, 1), len(m), round(rate - br, 1)))
+    single.sort(key=lambda x: -x[3])
+    print("  单条件:", single[:5])
+
+    results = []
+    strong = [s[0] for s in single[:6]]
+    for i in range(len(strong)):
+        for j in range(i+1, len(strong)):
+            f1, f2 = strong[i], strong[j]
+            cond = pd.Series(True, index=sub.index)
+            for feat, (op, val) in [(f1, factor_defs[f1]), (f2, factor_defs[f2])]:
+                cond &= (sub[feat] < val) if op == "<" else (sub[feat] > val)
+            m = sub[cond]
+            if len(m) >= 100:
+                rate = m["is_ok"].mean() * 100
+                results.append(({f1: factor_defs[f1], f2: factor_defs[f2]},
+                                len(m), round(rate, 1), round(m[col].mean(), 1), round(rate - br, 1)))
+    results.sort(key=lambda x: -x[4])
+    return results[:5]
+
+
+def to_entry(cond_dict):
+    entry = {}
+    for feat, (op, val) in cond_dict.items():
+        key = feat + ("_min" if op == ">" else "_max")
+        entry[key] = float(val)
+    return entry
+
+
+def mine(market="a", regimes=None, target="ret10"):
+    regimes = regimes or ["trend_up", "choppy", "trend_down"]
+    panel = load_panel(market)
+    out = {"market": market, "mined_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "candidates": []}
+    for rg in regimes:
+        combos = scan(market, rg, panel, target=target)
+        for cond, n, rate, avg, extra in combos[:3]:
+            cand = {
+                "regime": rg, "market": market, "group": "B", "status": "candidate",
+                "entry": to_entry(cond),
+                "trades_est": n, "ok_rate": rate, "avg_ret": avg, "excess_pp": extra,
+                "hypothesis": f"[{rg}] 由果及因: {cond} → 短期反弹率{rate}%(基线+{extra}pp)",
+            }
+            out["candidates"].append(cand)
+            print(f"  [{rg}] {cond} n={n} 反弹率{rate}% 超额+{extra}pp")
+    return out
+
+
+if __name__ == "__main__":
+    import sys
+    market = sys.argv[1] if len(sys.argv) > 1 else "hk"
+    target = sys.argv[2] if len(sys.argv) > 2 else "ret10"
+    res = mine(market, target=target)
+    with open(OUT_JSON, "w", encoding="utf-8") as f:
+        json.dump(res, f, ensure_ascii=False, indent=1)
+    print(f"写入 {OUT_JSON}: {len(res['candidates'])} 个候选")