#!/usr/bin/env python3 """step39_drawdown_fix.py — 回撤优化验证（数据驱动） 针对根因（信号扎堆满仓→全止损）测试： 方案A: 单日买入限流（每日最多N个新买入） 方案B: 连亏降仓（连续止损>=M次后仓位降到X%，恢复后再加回） 方案C: A+B 组合 对比: 年化/回撤/资金利用率 """ 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()) idx_df = pd.read_sql("SELECT date, close, high FROM stock_daily WHERE code='sh000001' ORDER BY date", sqlite3.connect("file:/home/hmo/MoFin/data/mofin.db?mode=ro", uri=True)) idx_df["date"] = idx_df["date"].astype(str) idx_df["hi60"] = idx_df["high"].rolling(60).max() idx_df["mkt_dd60"] = (idx_df["close"] / idx_df["hi60"] - 1) * 100 dd_map = dict(zip(idx_df["date"], idx_df["mkt_dd60"])) panel["mkt_dd60"] = panel["date"].map(dd_map) 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) & (panel["mkt_dd60"] <= -5) ) 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) sig_by_date = {} for r in sigdf.itertuples(): sig_by_date.setdefault(r.date, []).append(r) def run_sim(slots=10, hold=40, pos_frac=0.15, stop_buf=0.05, max_daily=99, loss_reduce=0, loss_threshold=99, reduce_frac=0.5): """max_daily: 单日最大买入数; loss_reduce: 连亏>=N次降仓""" INIT_CAP = 1_000_000 positions = {} cash = INIT_CAP navs = [] streak_loss = 0 cur_pos_frac = pos_frac 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] exited = False 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] exited = True streak_loss = 0 # 盈利离场，重置连亏 if not exited and lo <= pos["stop"]: cash += pos["qty"] * lo del positions[code] streak_loss += 1 # 止损，连亏+1 if loss_reduce > 0 and streak_loss >= loss_threshold: cur_pos_frac = pos_frac * reduce_frac # 降仓 exited = True if not exited and di - pos["entry_di"] >= hold: cash += pos["qty"] * cl del positions[code] streak_loss = 0 # 买入（限流） if d in sig_by_date: bought = 0 for r in sig_by_date[d]: if bought >= max_daily: break 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 * cur_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, } bought += 1 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 cash_ratio = np.mean([1 - nav_arr[i]/nav_arr[i] for i in range(len(navs))]) # placeholder return cagr, dd, len(navs) # 方案A: 单日限流 print("\n=== 方案A: 单日买入限流 ===", flush=True) for maxd in [1, 2, 3, 5, 99]: cagr, dd, _ = run_sim(max_daily=maxd) print("单日限{}个: 年化={:.2f}% 回撤={:.1f}%".format(maxd, cagr, dd), flush=True) # 方案B: 连亏降仓 print("\n=== 方案B: 连亏降仓 ===", flush=True) for thr, frac in [(2, 0.5), (3, 0.5), (3, 0.7), (4, 0.5)]: cagr, dd, _ = run_sim(loss_reduce=1, loss_threshold=thr, reduce_frac=frac) print("连亏{}次降{}%仓位: 年化={:.2f}% 回撤={:.1f}%".format(thr, int(frac*100), cagr, dd), flush=True) # 方案C: 组合 print("\n=== 方案C: 组合 ===", flush=True) for maxd, thr, frac in [(3, 3, 0.5), (2, 3, 0.5), (3, 2, 0.5)]: cagr, dd, _ = run_sim(max_daily=maxd, loss_reduce=1, loss_threshold=thr, reduce_frac=frac) print("限{}个+连亏{}降{}%: 年化={:.2f}% 回撤={:.1f}%".format(maxd, thr, int(frac*100), cagr, dd), flush=True) print("\n=== 完成 ===", flush=True)