diff --git a/scripts/backtest_churn.py b/scripts/backtest_churn.py index e2f93fd..2fae1c5 100644 --- a/scripts/backtest_churn.py +++ b/scripts/backtest_churn.py @@ -56,6 +56,9 @@ SELL_ACTIONS = {"EXIT", "TRIM"} # 真正减仓的落地 # 决策系统驱动的判据: 卖出 + 来源是盘中/风控信号 (hard_numbers.source 或 reason 里含这些词) SIGNAL_SOURCES = {"intraday", "risk_sell"} SIGNAL_WORDS = ("决策系统", "风控", "SELL", "转弱", "派发") +# 卖出主导信号(hard_numbers.dominant_signal): "take_profit" 这一档是止盈/动量兑现, 其余(转弱、 +# 派发、大额流出等)归"风控止损"。你担心的"今天买明天卖"正是止盈/动量这拨, 单拎出来判它。 +TAKE_PROFIT_DOM = "take_profit" # 近似"硬止损"(护栏放行的一档): 入场到卖出这段已经深亏, 视为真出事, 不该被护栏拦 HARD_RISK_DROP = -0.08 @@ -108,7 +111,8 @@ def read_ledger(since_dt: datetime) -> list: def classify(row: dict) -> dict: - """给账本一行贴标签: side(buy/sell/other) 与 signal_driven(是否决策系统驱动的卖出)。""" + """给账本一行贴标签: side(buy/sell/other)、signal_driven(是否决策系统驱动的卖出)、 + 以及 cohort(止盈动量 / 风控止损, 按 hard_numbers.dominant_signal 分, 只对决策系统卖出有值)。""" act = str(row.get("action") or "").upper() verdict = str(row.get("verdict") or "").upper() acted = verdict == "PASS" # 只认真正放行落地的决策 @@ -118,11 +122,17 @@ def classify(row: dict) -> dict: side = "sell" else: side = "other" - src = str((row.get("hn") or {}).get("source") or "").lower() + hn = row.get("hn") or {} + src = str(hn.get("source") or "").lower() reason = str(row.get("reason") or "") signal_driven = (side == "sell") and ( src in SIGNAL_SOURCES or any(w in reason for w in SIGNAL_WORDS)) - return {"side": side, "signal_driven": signal_driven} + dom = str(hn.get("dominant_signal") or "").strip().lower() + cohort = None + if signal_driven: + cohort = "止盈动量" if dom == TAKE_PROFIT_DOM else "风控止损" + return {"side": side, "signal_driven": signal_driven, + "dominant_signal": dom, "cohort": cohort} # ------------------------------------------------------------------ 历史收盘价源 (gp_day_data @ 18.199) @@ -231,6 +241,7 @@ def pair_round_trips(rows_by_code: dict) -> list: "buy_price": _f(b["price_at"]), "sell_price": _f(s["price_at"]), "hold_days": hold_days, "gross_ret": gross, "signal_driven": tag["signal_driven"], "sell_reason": s.get("reason"), + "cohort": tag.get("cohort"), }) pending_buy = None return trips @@ -238,13 +249,18 @@ def pair_round_trips(rows_by_code: dict) -> list: # ------------------------------------------------------------------ 各段输出 def section_inventory(rows, trips): - buys = sum(1 for r in rows if classify(r)["side"] == "buy") - sells = sum(1 for r in rows if classify(r)["side"] == "sell") - sig_sells = sum(1 for r in rows if classify(r)["signal_driven"]) + tags = [classify(r) for r in rows] + buys = sum(1 for t in tags if t["side"] == "buy") + sells = sum(1 for t in tags if t["side"] == "sell") + sig_sells = sum(1 for t in tags if t["signal_driven"]) + tp_sells = sum(1 for t in tags if t["cohort"] == "止盈动量") + rk_sells = sum(1 for t in tags if t["cohort"] == "风控止损") quick = [t for t in trips if t["signal_driven"] and t["hold_days"] is not None and t["hold_days"] <= max(GUARD_DAYS)] print("\n① 样本盘点") print(f" 账本落地买入 {buys} 笔 · 落地卖出 {sells} 笔 · 其中决策系统驱动的卖出 {sig_sells} 笔") + print(f" └ 再拆两拨(按 hard_numbers.dominant_signal): 止盈/动量 {tp_sells} 笔 · " + f"风控止损 {rk_sells} 笔") print(f" 配成来回 {len(trips)} 组 · 决策系统驱动的快速来回(≤{max(GUARD_DAYS)}交易日) {len(quick)} 组") return quick @@ -266,31 +282,50 @@ def section_roundtrips(trips): f"| {str(t['sell_reason'] or '')[:40]}") -def section_forward(rows): - """③ 前向收益: 卖完之后股价怎么走 —— 是不是瞎折腾的正面回答。""" - print("\n③ 决策系统卖出之后的前向收益 (正=卖飞/疑瞎折腾, 负=避坑/该卖)") - sig_sells = [r for r in rows if classify(r)["signal_driven"]] +def _forward_stats(sells): + """给一批卖出行, 算各前向 horizon 的样本收益序列 {h: [卖后涨跌...]}。""" got = {h: [] for h in FWD_HORIZONS} - for r in sig_sells: + for r in sells: st = r["decided_at"] st = st if isinstance(st, datetime) else datetime.fromisoformat(str(st)) fr = forward_returns(r["ts_code"], st, _f(r["price_at"])) for h in FWD_HORIZONS: if fr[h] is not None: got[h].append(fr[h]) + return got + + +def _print_forward(label, got): + """打印一拨卖出的前向收益。至少有一个 horizon 有样本才打印, 返回是否打了。""" if not any(got.values()): - print(" (前向价源未接通 fetch_daily_closes, 本段跳过 —— 接上后重跑即出)") - return + return False + print(f" 【{label}】") for h in FWD_HORIZONS: v = got[h] if len(v) < MIN_SAMPLE: - print(f" T+{h}: 样本 {len(v)} (<{MIN_SAMPLE}), 只报数") + print(f" T+{h}: 样本 {len(v)} (<{MIN_SAMPLE}), 只报数") continue avg = sum(v) / len(v) flew = sum(1 for x in v if x > 0.01) / len(v) # 卖完还涨超1%算卖飞 - print(f" T+{h}: 样本 {len(v)} · 卖后平均 {_fmt_pct(avg)} · 卖飞比例 {flew:.0%}") + print(f" T+{h}: 样本 {len(v)} · 卖后平均 {_fmt_pct(avg)} · 卖飞比例 {flew:.0%}") + return True + + +def section_forward(rows): + """③ 前向收益: 卖完之后股价怎么走 —— 是不是瞎折腾的正面回答。 + 拆三拨: 全部决策系统卖出 / 止盈动量 / 风控止损, 分开看到底谁在瞎折腾。""" + print("\n③ 决策系统卖出之后的前向收益 (正=卖飞/疑瞎折腾, 负=避坑/该卖)") + tagged = [(r, classify(r)) for r in rows] + sig_sells = [r for r, t in tagged if t["signal_driven"]] + tp_sells = [r for r, t in tagged if t["cohort"] == "止盈动量"] + rk_sells = [r for r, t in tagged if t["cohort"] == "风控止损"] + if not _print_forward("全部决策系统卖出", _forward_stats(sig_sells)): + print(" (前向价源未接通 fetch_daily_closes, 本段跳过 —— 接上后重跑即出)") + return + _print_forward("止盈/动量 (你最担心的今买明卖这拨)", _forward_stats(tp_sells)) + _print_forward("风控止损", _forward_stats(rk_sells)) print(" 读法: 卖后平均为正、卖飞比例高 → 这批卖出多在砍还会涨的票(瞎折腾); " - "为负 → 多在避坑(该卖)。") + "为负 → 多在避坑(该卖)。重点看「止盈/动量」这拨是不是正的 —— 是, 才印证你的担心。") def section_cost(trips): @@ -314,23 +349,14 @@ def _rt_cost_rate(): return COMMISSION_RATE * 2 + STAMP_RATE + TRANSFER_RATE * 2 + SLIPPAGE_BPS / 10000.0 * 2 -def section_counterfactual(trips): - """⑤ 反事实: 最小持有期护栏 N 扫一遍。需要前向价源, 缺则跳过。""" - print("\n⑤ 反事实: 加「最小持有期护栏」后净收益怎么变 (正=护栏有用)") - affected0 = [t for t in trips if t["signal_driven"]] - # 探一下前向价源是否可用 - probe = None - for t in affected0: - seq = _fwd_closes_after(t["ts_code"], t["sell_at"], n_max=GUARD_HOLD_TO) - if seq: - probe = True - break - if not probe: - print(" (前向价源未接通, 本段跳过 —— 接上 fetch_daily_closes 后重跑即出)") - return +def _guard_sweep(label, affected): + """给一拨受影响的来回, 按 N 扫最小持有期护栏, 逐档打印净收益变化。 + 返回 True 表示至少有一档出了结论(样本够), 供上层判空。""" + print(f" 【{label}】") + printed = False for N in GUARD_DAYS: deltas = [] - for t in affected0: + for t in affected: if t["hold_days"] is None or t["hold_days"] > N: continue # 护栏只管"刚建仓 N 日内"的卖出 if (t["gross_ret"] or 0) <= HARD_RISK_DROP: @@ -347,14 +373,38 @@ def section_counterfactual(trips): counter = held - _rt_cost_rate() / 2 deltas.append(counter - actual) if len(deltas) < MIN_SAMPLE: - print(f" N={N}: 受影响样本 {len(deltas)} (<{MIN_SAMPLE}), 只报数") + print(f" N={N}: 受影响样本 {len(deltas)} (<{MIN_SAMPLE}), 只报数") continue avg = sum(deltas) / len(deltas) helped = sum(1 for d in deltas if d > 0) / len(deltas) - print(f" N={N} 交易日护栏: 受影响 {len(deltas)} 笔 · 平均净收益变化 {_fmt_pct(avg)} · " + printed = True + print(f" N={N} 交易日护栏: 受影响 {len(deltas)} 笔 · 平均净收益变化 {_fmt_pct(avg)} · " f"变好占比 {helped:.0%}") + return printed + + +def section_counterfactual(trips): + """⑤ 反事实: 最小持有期护栏 N 扫一遍。需要前向价源, 缺则跳过。 + 拆两拨: 先看全部决策系统卖出, 再单看「止盈/动量」这拨 —— 护栏本就是冲这拨设计的, + 真要装, 该由它这拨的账说话, 而不是被风控止损那拨稀释。""" + print("\n⑤ 反事实: 加「最小持有期护栏」后净收益怎么变 (正=护栏有用)") + affected0 = [t for t in trips if t["signal_driven"]] + # 探一下前向价源是否可用 + probe = None + for t in affected0: + seq = _fwd_closes_after(t["ts_code"], t["sell_at"], n_max=GUARD_HOLD_TO) + if seq: + probe = True + break + if not probe: + print(" (前向价源未接通, 本段跳过 —— 接上 fetch_daily_closes 后重跑即出)") + return + _guard_sweep("全部决策系统卖出", affected0) + _guard_sweep("止盈/动量 (护栏本就冲这拨设计)", + [t for t in trips if t.get("cohort") == "止盈动量"]) print(" 读法: 某个 N 上平均变化持续为正且变好占比过半 → 这套信号是「噪音来回型」, " - "护栏该上、N 取那档; 若普遍为负 → 是「大波段型」, 别压, 快卖多数是对的。") + "护栏该上、N 取那档; 若普遍为负 → 是「大波段型」, 别压, 快卖多数是对的。" + "判护栏该不该上, 以「止盈/动量」这拨的账为准。") # ------------------------------------------------------------------ main