hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: B组挖掘v4——相对分位果(温区前20%)+三因子组合扫描,产出趋势市候选(好果率34.5%超额14.5pp)

Browse Source
This commit is contained in:
xxm
2026-08-16 13:37:07 +08:00
parent e82a3a7846
commit f29842166e
1 changed files with 46 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
evolution/b_group_miner.py
+46 -44
View File
@@ -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'])} 个候选")