MoFin/scripts/research/step35_param_scan.py

#!/usr/bin/env python3
"""step35_param_scan.py — 槽位×持有期×仓位 参数扫描（数据驱动）
在 step34 框架上，扫描：
- 槽位: 6/8/10/12
- 持有期: 30/40/50/60日
- 单票仓位: 8%/10%/12%/15%
找年化最优组合（同时看回撤）
"""
import numpy as np
import pandas as pd
import sqlite3
import sys
sys.path.insert(0, "/tmp")
from sr_calculator import SRCalculator

print("=== 加载 ===", flush=True)
panel = pd.read_pickle("/tmp/panel_12d.pkl")
panel = panel.sort_values(["code", "date"]).reset_index(drop=True)
panel["_key"] = panel["code"] + "_" + panel["date"]
pos_map = {k: i for i, k in enumerate(panel["_key"])}
dates = sorted(panel["date"].unique())

sig_cond = (
    (panel["mkt_rsi"] < 50) & (panel["mcap_q"] < 0.2) & (panel["pe_q"] < 0.2) &
    (panel["news3"] >= 1) & (panel["sec_ret20"] < 0) & (panel["bias60"] < -20)
)
cand = panel[sig_cond][["code", "date"]].copy()
cand = cand.sort_values(["code", "date"])
cand["prev"] = cand.groupby("code")["date"].shift(1)
cand["gap"] = (pd.to_datetime(cand["date"]) - pd.to_datetime(cand["prev"])).dt.days
cand = cand[(cand["prev"].isna()) | (cand["gap"] > 30)]
print("信号:", len(cand), flush=True)

conn = sqlite3.connect("file:/home/hmo/MoFin/data/mofin.db?mode=ro", uri=True)
codes = cand["code"].unique().tolist()
ph = ",".join("?" * len(codes))
df = pd.read_sql("SELECT code, date, close, high, low FROM stock_daily WHERE code IN ({}) ORDER BY code, date".format(ph), conn, params=codes)
df["date"] = df["date"].astype(str)
df["code"] = df["code"].astype(str).str.zfill(6)
df = df.sort_values(["code", "date"]).reset_index(drop=True)
df["_key"] = df["code"] + "_" + df["date"]
dpos = {k: i for i, k in enumerate(df["_key"])}
closes = df["close"].values
highs = df["high"].values
lows = df["low"].values

cand["kidx"] = (cand["code"] + "_" + cand["date"]).map(dpos)
cand = cand.dropna(subset=["kidx"]).copy()
cand["kidx"] = cand["kidx"].astype(int)

# 支撑压力（缓存）
sr = SRCalculator()
print("=== 计算支撑压力 ===", flush=True)
sig_list = []
for r in cand.itertuples():
    bars_code = sr.get_bars(r.code)
    d_idx = bars_code.index[bars_code["date"] == r.date]
    if len(d_idx) == 0:
        continue
    sr_full = sr.sr_full(r.code, d_idx[0])
    pv = sr_full["pivot"]
    chip = sr_full["chip"]
    if not pv:
        continue
    sig_close = closes[r.kidx]
    support = pv["s2"]
    if chip and chip["chip_ss"] < sig_close:
        support = max(pv["s2"], chip["chip_ss"])
    resist = pv["r2"]
    if chip and chip["chip_sr"] > sig_close:
        resist = min(pv["r2"], chip["chip_sr"])
    if support >= sig_close or resist <= sig_close:
        continue
    sig_list.append({"code": r.code, "date": r.date, "kidx": r.kidx,
                     "sig_close": sig_close, "support": support, "resist": resist})

sigdf = pd.DataFrame(sig_list)
print("有支撑压力:", len(sigdf), flush=True)

def run_sim(slots, hold, pos_frac, stop_buf=0.05):
    """运行一次模拟，返回 (cagr, max_dd, n_trades)"""
    sig_by_date = {}
    for r in sigdf.itertuples():
        sig_by_date.setdefault(r.date, []).append(r)
    INIT_CAP = 1_000_000
    positions = {}
    cash = INIT_CAP
    navs = []
    for di, d in enumerate(dates):
        for code in list(positions.keys()):
            pos = positions[code]
            k = dpos.get(code + "_" + d)
            if k is None:
                continue
            hi, lo, cl = highs[k], lows[k], closes[k]
            if hi >= pos["tp"]:
                sell_qty = pos["qty"] // 2
                if sell_qty > 0:
                    cash += sell_qty * pos["tp"]
                    pos["qty"] -= sell_qty
                if pos["qty"] <= 0:
                    del positions[code]
                    continue
            if lo <= pos["stop"]:
                cash += pos["qty"] * lo
                del positions[code]
                continue
            if di - pos["entry_di"] >= hold:
                cash += pos["qty"] * cl
                del positions[code]
                continue
        if d in sig_by_date:
            for r in sig_by_date[d]:
                if len(positions) >= slots:
                    break
                if r.code in positions:
                    continue
                cur_nav = cash
                for c, p in positions.items():
                    k = dpos.get(c + "_" + d)
                    px = closes[k] if k is not None else p["avg_cost"]
                    cur_nav += p["qty"] * px
                pos_val = cur_nav * pos_frac
                price = r.sig_close
                if price <= 0:
                    continue
                qty = int(pos_val / price)
                if qty <= 0 or qty * price > cash:
                    continue
                cash -= qty * price
                positions[r.code] = {
                    "entry_di": di, "qty": qty, "avg_cost": price,
                    "stop": r.support * (1 - stop_buf), "tp": r.resist,
                }
        nav = cash
        for c, p in positions.items():
            k = dpos.get(c + "_" + d)
            px = closes[k] if k is not None else p["avg_cost"]
            nav += p["qty"] * px
        navs.append(nav)
    nav_arr = np.array(navs)
    final = nav_arr[-1]
    years = len(navs) / 250
    cagr = ((final / INIT_CAP) ** (1 / years) - 1) * 100 if final > 0 else -100
    cummax = np.maximum.accumulate(nav_arr)
    dd = ((nav_arr - cummax) / cummax).min() * 100
    return cagr, dd

print("\n=== 参数扫描 ===", flush=True)
print("| 槽位 | 持有 | 仓位 | 年化 | 回撤 |", flush=True)
print("|---|---|---|---:|---:|", flush=True)
results = []
for slots in [6, 8, 10, 12]:
    for hold in [30, 40, 50]:
        for pos_frac in [0.08, 0.10, 0.12, 0.15]:
            cagr, dd = run_sim(slots, hold, pos_frac)
            results.append({"slots": slots, "hold": hold, "pos": pos_frac, "cagr": cagr, "dd": dd})
            print("| {} | {} | {}% | {:.2f}% | {:.1f}% |".format(
                slots, hold, int(pos_frac*100), cagr, dd), flush=True)

res = pd.DataFrame(results)
res = res.sort_values("cagr", ascending=False)
print("\n=== 最优组合 ===", flush=True)
print(res.head(10).to_string(index=False), flush=True)
res.to_csv("/tmp/step35_params.csv", index=False)
print("\n=== 完成 ===", flush=True)