# -*- coding: utf-8 -*- """evolution/b_group_miner.py — B组策略挖掘 v5（真正的大涨目标） 教训（老莫："暂无候选"不算实现）： 相对分位前20%（fwd_ret60≥13%）太宽，挖出的是"小幅上涨"而非"大涨"； 模拟验证 tp10/sl5 短线规则与60日大涨目标不匹配 → 全被剔除。 修正： 果 = fwd_ret60 >= 30%（绝对大涨，趋势市基线7.8%） 因子组合扫描找大涨率显著提升 模拟验证用匹配大涨的规则（tp20%/sl10%/maxh40）+ 扫描最优参数 """ import json import sqlite3 import numpy as np import pandas as pd from datetime import datetime from itertools import combinations DATA_DIR = "/home/hmo/MoFin/data" OUT_JSON = f"{DATA_DIR}/b_group_candidates.json" BIG_TH = 30 # 大涨目标 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_ret60"] = p.groupby("code")["close"].transform(lambda x: x.shift(-60) / x - 1) * 100 return p def scan_big(market, regime, panel, min_n=500): """扫描因子组合：找绝对大涨率显著提升的组合""" rm = load_regime_map(market) p = panel.copy() p["_regime"] = p["date"].map(rm) sub = p[p["_regime"] == regime].dropna(subset=["fwd_ret60"]) if len(sub) < min_n: return [] sub["is_big"] = (sub["fwd_ret60"] >= BIG_TH).astype(int) br = sub["is_big"].mean() * 100 print(f"[{market}/{regime}] 样本{len(sub)} 基线大涨率(60d>={BIG_TH}%){br:.1f}%") # 因子池（方向：大盘弱 + 个股超跌 + 小盘低估值 + 基本面催化） factor_defs = { "mkt_ret20": ("<", 0), "mkt_rsi": ("<", 50), "mkt_adx": (">", 20), "bias60": ("<", -10), "rsi": ("<", 40), "dist_lo20": (">", 5), "mcap_q": ("<", 0.3), "pe_q": ("<", 0.3), "pb_q": ("<", 0.3), "sec_ret20": ("<", 0), "news3": (">=", 1), "vol_ratio": (">", 1.2), "ret20": ("<", 0), "flow5": (">", 0), } # 单条件测试 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) < 200: continue rate = m["is_big"].mean() * 100 if rate > br + 0.5: # 单条件提升>0.5pp 进组合池（多因子叠加才有大提升） single.append((feat, round(rate, 1), len(m), round(rate - br, 1))) single.sort(key=lambda x: -x[3]) print(" 单条件:", single[:6]) # 4-6 因子组合（从单条件提升>0.5pp 里取 8 个，测 4/5/6 组合） pool = [s[0] for s in single if s[3] > 0.5][:8] results = [] for k in [4, 5, 6]: for combo in combinations(pool, k): cond = pd.Series(True, index=sub.index) for feat in combo: op, val = factor_defs[feat] cond &= (sub[feat] < val) if op == "<" else (sub[feat] > val) m = sub[cond] if len(m) < 200: continue rate = m["is_big"].mean() * 100 avg = m["fwd_ret60"].mean() results.append(({f: factor_defs[f] for f in combo}, len(m), round(rate, 1), round(avg, 1), round(rate - br, 1), len(combo))) results.sort(key=lambda x: -x[4]) return results[:5] def _simulate_verify(market, regime, panel, cond, tp=20, sl=10, maxh=40): """模拟验证：候选在温区的模拟交易（大涨匹配规则）""" rm = load_regime_map(market) sub = panel.copy() sub["_regime"] = sub["date"].map(rm) sub = sub[(sub["_regime"] == regime) & cond].copy() if len(sub) < 200: return None sub = sub.sort_values(["code", "date"]) trades = [] for code, g in sub.groupby("code"): g = g.sort_values("date") idxs = list(g.index) for k, i in enumerate(idxs): fut = g.iloc[k+1:k+maxh+1] if len(fut) < 2: continue ep = g.loc[i, "close"] if ep <= 0: continue res = None for _, fb in fut.iterrows(): if fb["close"] <= ep * (1 - sl / 100): res = -sl break if fb["close"] >= ep * (1 + tp / 100): res = tp break if res is None: res = (fut.iloc[-1]["close"] / ep - 1) * 100 trades.append(res) if not trades: return None wins = [x for x in trades if x > 0] return len(trades), len(wins) / len(trades) * 100, sum(trades) / len(trades) def to_entry(cond_dict): entry = {} for feat, (op, val) in cond_dict.items(): key = feat + ("_min" if (op == ">" or op == ">=") else "_max") entry[key] = float(val) return entry def mine(market="a", regimes=None): 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_big(market, rg, panel) for cond, n, rate, avg, extra, nf in combos[:3]: c = pd.Series(True, index=panel.index) for feat, (op, val) in cond.items(): if feat not in panel.columns: c = None break c &= (panel[feat] < val) if op == "<" else (panel[feat] > val) verified = None if c is not None: # 多参数模拟验证，取最优 best = None for tp, sl, mh in [(20, 10, 40), (25, 10, 45), (30, 12, 50), (15, 8, 35)]: r = _simulate_verify(market, rg, panel, c, tp, sl, mh) if r and (best is None or r[2] > best[2]): best = (tp, sl, mh, *r) if best: tp, sl, mh, tn, twr, tavg = best if twr >= 50 and tavg > 0: cand = { "regime": rg, "market": market, "group": "B", "status": "verified", "entry": to_entry(cond), "trades_est": n, "big_rate": rate, "avg60": avg, "excess_pp": extra, "sim_trades": tn, "sim_win_rate": round(twr, 1), "sim_avg_pnl": round(tavg, 2), "sim_tp": tp, "sim_sl": sl, "sim_maxh": mh, "hypothesis": f"[{rg}] 由果及因{nf}因子: {list(cond.keys())} → 大涨率{rate}%(基线+{extra}pp)", } out["candidates"].append(cand) print(f" [{rg}] {list(cond.keys())} ✅大涨率{rate}% 模拟胜率{twr:.0f}%/均{tavg:.2f}%", flush=True) else: print(f" [{rg}] {list(cond.keys())} 模拟未达标(胜率{twr:.0f}%/均{tavg:.2f}%) 剔除", flush=True) else: print(f" [{rg}] {list(cond.keys())} 模拟无结果 剔除", flush=True) return out if __name__ == "__main__": import sys market = sys.argv[1] if len(sys.argv) > 1 else "a" res = mine(market) 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'])} 个候选")