201 lines
8.5 KiB
Python
201 lines
8.5 KiB
Python
# -*- 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 = []
|
||
trade_details = []
|
||
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
|
||
hold_days = maxh
|
||
for j, (_, fb) in enumerate(fut.iterrows()):
|
||
if fb["close"] <= ep * (1 - sl / 100):
|
||
res = -sl
|
||
hold_days = j + 1
|
||
break
|
||
if fb["close"] >= ep * (1 + tp / 100):
|
||
res = tp
|
||
hold_days = j + 1
|
||
break
|
||
if res is None:
|
||
res = (fut.iloc[-1]["close"] / ep - 1) * 100
|
||
hold_days = len(fut)
|
||
trades.append(res)
|
||
trade_details.append({"entry_date": str(g.loc[i, "date"]), "pnl_pct": round(res, 2),
|
||
"profit_pct": round(res, 2), "hold_days": hold_days,
|
||
"code": str(code)})
|
||
if not trades:
|
||
return None
|
||
wins = [x for x in trades if x > 0]
|
||
if not trades:
|
||
return None
|
||
return {"n": len(trades), "win_rate": len(wins) / len(trades) * 100,
|
||
"avg_pnl": sum(trades) / len(trades), "trades": trade_details}
|
||
|
||
|
||
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:
|
||
# 多参数模拟验证，取最优
|
||
# 先筛胜率≥50%的参数，再取其中收益最高（2026-08-16 修正：原取收益最高可能选中胜率<50%参数）
|
||
passed_params = []
|
||
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 r["win_rate"] >= 50 and r["avg_pnl"] > 0:
|
||
passed_params.append((tp, sl, mh, r))
|
||
if passed_params:
|
||
best = max(passed_params, key=lambda x: x[3]["avg_pnl"])
|
||
tp, sl, mh, rd = best
|
||
tn, twr, tavg = rd["n"], rd["win_rate"], rd["avg_pnl"]
|
||
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,
|
||
"trades": rd.get("trades", []),
|
||
"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'])} 个候选")
|