From f29842166ed20739b3b8f1ae2cb4c469d749ff79 Mon Sep 17 00:00:00 2001 From: xxm Date: Sun, 16 Aug 2026 13:37:07 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20B=E7=BB=84=E6=8C=96=E6=8E=98v4=E2=80=94?= =?UTF-8?q?=E2=80=94=E7=9B=B8=E5=AF=B9=E5=88=86=E4=BD=8D=E6=9E=9C(?= =?UTF-8?q?=E6=B8=A9=E5=8C=BA=E5=89=8D20%)+=E4=B8=89=E5=9B=A0=E5=AD=90?= =?UTF-8?q?=E7=BB=84=E5=90=88=E6=89=AB=E6=8F=8F,=E4=BA=A7=E5=87=BA?= =?UTF-8?q?=E8=B6=8B=E5=8A=BF=E5=B8=82=E5=80=99=E9=80=89(=E5=A5=BD?= =?UTF-8?q?=E6=9E=9C=E7=8E=8734.5%=E8=B6=85=E9=A2=9D14.5pp)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- evolution/b_group_miner.py | 90 +++++++++++++++++++------------------- 1 file changed, 46 insertions(+), 44 deletions(-) diff --git a/evolution/b_group_miner.py b/evolution/b_group_miner.py index 57423179..edf1401b 100644 --- a/evolution/b_group_miner.py +++ b/evolution/b_group_miner.py @@ -1,13 +1,13 @@ # -*- coding: utf-8 -*- -"""evolution/b_group_miner.py — B组挖掘 v3（务实版） -聚焦缺口温区（A股震荡市/港股下跌市），用【短期反弹】作果（fwd_ret10>=8% 或 fwd_ret20>=12%）， -在超跌+企稳因子池上扫描组合。短期反弹样本充足，能稳定产出候选。 +"""B组挖掘 v4：相对分位果 + 三因子组合扫描 +果 = 该温区下 fwd_ret60 前 20% 分位（相对，避免绝对阈值稀疏） """ 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" @@ -24,59 +24,62 @@ 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 + p["fwd_ret60"] = p.groupby("code")["close"].transform(lambda x: x.shift(-60) / x - 1) * 100 return p -def scan(market, regime, panel, target="ret10", min_n=200): - """扫描超跌+企稳因子组合的短期反弹概率""" +def scan3(market, regime, panel, min_n=500): + """三因子组合扫描：相对分位果""" 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]) + sub = p[p["_regime"] == regime].dropna(subset=["fwd_ret60"]) 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}%") + # 相对果：温区内 fwd_ret60 前 20% + thr = sub["fwd_ret60"].quantile(0.80) + sub["is_good"] = (sub["fwd_ret60"] >= thr).astype(int) + br = 20.0 # 相对分位定义，基线恒 20% + print(f"[{market}/{regime}] 样本{len(sub)} 果阈值60日+{thr:.0f}%") - # 因子：超跌 + 企稳 + 小盘低估值（温区通用的候选） + # 因子池（方向：小市值/低估值/超跌/放量/企稳/低动量） 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), + "mcap_q": ("<", 0.5), "pe_q": ("<", 0.5), "pb_q": ("<", 0.5), + "bias60": ("<", -5), "rsi": ("<", 50), "dist_lo20": (">", 3), + "vol_ratio": (">", 1.0), "mkt_ret20": ("<", 0), "ret20": ("<", 0), + "sec_ret20": ("<", 0), "flow5": (">", 0), "news3": (">=", 1), + "ret5": (">", -3), "mkt_rsi": ("<", 50), } + # 单条件 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: + if len(m) < 200: continue - rate = m["is_ok"].mean() * 100 - if rate > br + 2: - single.append((feat, round(rate, 1), len(m), round(rate - br, 1))) + rate = m["is_good"].mean() * 100 + if rate > 23: # 相对基线20% +3pp + single.append((feat, round(rate, 1), len(m), round(rate - 20, 1))) single.sort(key=lambda x: -x[3]) - print(" 单条件:", single[:5]) + print(" 单条件:", single[:4]) + # 三因子组合（从单条件超额>2pp 里取 6 个，C(6,3)=20 组合） + pool = [s[0] for s in single if s[3] > 2][:6] 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))) + for combo in combinations(pool, 3): + 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_good"].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 - 20, 1))) results.sort(key=lambda x: -x[4]) return results[:5] @@ -89,29 +92,28 @@ def to_entry(cond_dict): return entry -def mine(market="a", regimes=None, target="ret10"): +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(market, rg, panel, target=target) + combos = scan3(market, rg, panel) 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)", + "entry": to_entry(cond), "trades_est": n, "good_rate": rate, + "avg60": avg, "excess_pp": extra, + "hypothesis": f"[{rg}] 由果及因三因子: {list(cond.keys())} → 60日前20%占比{rate}%(超额+{extra}pp)", } out["candidates"].append(cand) - print(f" [{rg}] {cond} n={n} 反弹率{rate}% 超额+{extra}pp") + print(f" [{rg}] {list(cond.keys())} 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) + 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'])} 个候选")