From 40fcf160a39e1c647cc322ceed9ffc9477302fab Mon Sep 17 00:00:00 2001 From: zlt Date: Thu, 27 Aug 2026 12:48:42 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=9B=9E=E6=B5=8B=E8=84=9A?= =?UTF-8?q?=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/backtest_churn.py | 395 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 395 insertions(+) create mode 100644 scripts/backtest_churn.py diff --git a/scripts/backtest_churn.py b/scripts/backtest_churn.py new file mode 100644 index 0000000..e2f93fd --- /dev/null +++ b/scripts/backtest_churn.py @@ -0,0 +1,395 @@ +# -*- coding: utf-8 -*- +""" +换手体检: 决策系统驱动的卖出, 是"避坑"还是"来回瞎折腾" (只读, 不写任何表) +========================================================================== +回答你担心的那件事: 决策系统偏动量、容易今天买明天卖 —— 这些快进快出的卖出, 扣掉 +手续费和滑点之后, 到底帮我们躲过了下跌(该卖), 还是把还会涨的票甩了、白交学费(瞎折腾)? +以及: 若加一道"刚建仓 N 个交易日内、非硬止损的决策系统卖出先不自动执行"的护栏, 净收益 +是变好还是变差? 用你自己的历史账本把这条曲线量出来, 别拍脑袋。 + +一句话方法: 全部基于评审账本 pms_action_ledger —— 它对每个买卖决策都记了当时现价 price_at, +天生就是反事实判分的锚。五段: + ① 样本盘点 账本里买入类/卖出类各多少, 其中"决策系统驱动"的卖出多少, 快速来回多少。 + ② 来回配对 同一只票"一次买紧跟一次卖"配成来回, 算持有天数 + 扣成本后的来回净收益。 + ③ 前向收益 每笔卖出: 卖出价 vs 该股此后 T+1/T+5/T+20 收盘。卖完还涨=卖飞(疑似瞎折腾), + 卖完就跌=避坑(该卖)。聚合看均值与"卖飞比例"—— 这段是"是不是瞎折腾"的正面回答。 + ④ 成本账 决策系统驱动的快速来回一共交了多少手续费+滑点(纯摩擦成本)。 + ⑤ 反事实 对 N=1/3/5/10 交易日扫一遍"最小持有期护栏": 刚建仓 N 日内、非硬止损的决策 + 系统卖出改成"不卖、持有到 T+H", 比较装护栏前后的净收益差。正=护栏有用。 + +成本口径(A股, 顶部常量可调): + 佣金双边各 COMMISSION_RATE(默认万2.5, 每边最低5元)、印花税卖出单边 STAMP_RATE(默认千0.5)、 + 过户费双边 TRANSFER_RATE(默认十万1)、滑点每边 SLIPPAGE_BPS(默认8bp)。 + +前向收盘价源: fetch_daily_closes() 已按你给的 DATA_MODEL 接到 gp_day_data (18.199 db_gp_cj, + 走 app 的 index 源), 并处理了它的两个坑(symbol 前缀式、close 是 VARCHAR)。库不可达时 + ③⑤两段自动跳过、只出①②④, 不报错。 + +运行(桥机 factorevaluation, 与你其它只读脚本一致): + docker compose run --rm pms-web python scripts/backtest_churn.py --days 120 +读数纪律: 样本少于 MIN_SAMPLE 的段落只列数、不下结论 —— 不等数据的判分是编故事。 +""" +from __future__ import annotations + +import argparse +import json +import os +import sys +from datetime import datetime, timedelta + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.db.session import fetch_all, DBUnavailable # noqa: E402 只读单表, 过单表守卫 + +# ------------------------------------------------------------------ 可调常量 +MIN_SAMPLE = 5 # 少于这个数只报数不下结论 + +COMMISSION_RATE = 0.00025 # 佣金费率(每边); 万2.5 +COMMISSION_MIN = 5.0 # 佣金每笔最低(元) +STAMP_RATE = 0.0005 # 印花税(仅卖出); 千0.5 +TRANSFER_RATE = 0.00001 # 过户费(双边); 十万1 +SLIPPAGE_BPS = 8.0 # 滑点(每边, 基点); 8bp = 0.08% + +# 账本 action 归类 (与 app/core/planner.py、signal_service.py 的口径对齐; 后续在真机再核一遍) +BUY_ACTIONS = {"OPEN", "FILL", "ADD", "DCA"} # 真正增仓的落地动作 +SELL_ACTIONS = {"EXIT", "TRIM"} # 真正减仓的落地动作 +# 决策系统驱动的判据: 卖出 + 来源是盘中/风控信号 (hard_numbers.source 或 reason 里含这些词) +SIGNAL_SOURCES = {"intraday", "risk_sell"} +SIGNAL_WORDS = ("决策系统", "风控", "SELL", "转弱", "派发") +# 近似"硬止损"(护栏放行的一档): 入场到卖出这段已经深亏, 视为真出事, 不该被护栏拦 +HARD_RISK_DROP = -0.08 + +FWD_HORIZONS = (1, 5, 20) # 前向收益看这几个交易日 +GUARD_DAYS = (1, 3, 5, 10) # 反事实扫的最小持有期 +GUARD_HOLD_TO = 5 # 护栏放行改为"持有到 T+这么多交易日"再看 + + +# ------------------------------------------------------------------ 小工具 +def _f(v, d=0.0): + try: + return float(v) + except (TypeError, ValueError): + return d + + +def _load_json(s): + if not s: + return {} + try: + return json.loads(s) if isinstance(s, str) else dict(s) + except (json.JSONDecodeError, TypeError, ValueError): + return {} + + +def _pct(entry, exit_): + e, x = _f(entry), _f(exit_) + return (x / e - 1.0) if e > 0 and x > 0 else None + + +def _fmt_pct(x): + return f"{x:+.2%}" if x is not None else "—" + + +# ------------------------------------------------------------------ 读账本 (单表, 过守卫) +def read_ledger(since_dt: datetime) -> list: + """按时间读评审账本。只读一张表, 满足 153 代理的严格单表访问。""" + rows = fetch_all( + "SELECT id, ts_code, decided_at, action, arbiter, verdict, price_at, " + "hard_numbers_json, reason, ref_id " + "FROM pms_action_ledger WHERE decided_at >= :since " + "ORDER BY ts_code, decided_at, id", + {"since": since_dt.strftime("%Y-%m-%d %H:%M:%S")}, source="proxy") + out = [] + for r in rows: + r = dict(r) + r["hn"] = _load_json(r.get("hard_numbers_json")) + out.append(r) + return out + + +def classify(row: dict) -> dict: + """给账本一行贴标签: side(buy/sell/other) 与 signal_driven(是否决策系统驱动的卖出)。""" + act = str(row.get("action") or "").upper() + verdict = str(row.get("verdict") or "").upper() + acted = verdict == "PASS" # 只认真正放行落地的决策 + if act in BUY_ACTIONS and acted: + side = "buy" + elif act in SELL_ACTIONS and acted: + side = "sell" + else: + side = "other" + src = str((row.get("hn") or {}).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} + + +# ------------------------------------------------------------------ 历史收盘价源 (gp_day_data @ 18.199) +_CLOSE_CACHE: dict = {} +_ANALYSIS_START = None # main 里设; 让 _fwd_closes_after 按整个分析区间取足够宽的每股窗口 +_ANALYSIS_END = None + + +def _to_prefix(ts_code: str) -> str: + """点式 600000.SH → 前缀式 SH600000 (gp_day_data.symbol 用前缀式, 见 DATA_MODEL 约定)。""" + s = str(ts_code or "").strip().upper() + if "." in s: + num, ex = s.split(".", 1) + return ex + num + return s + + +def fetch_daily_closes(ts_code: str, start_dt: datetime, end_dt: datetime) -> dict: + """返回 {date: 收盘价}, 只含有行情的交易日。 + + 源: gp_day_data (db_gp_cj @ 192.168.18.199, 走 app 的 index 数据源 = DB_MYSQL_URL)。 + PMS 本来就连这台取大盘 zs_day_data, 同库里 gp_day_data 就是个股原始日线。按 DATA_MODEL + 的两个坑处理: symbol 是**前缀式** SH600000(账本 ts_code 是点式, 这里转一下); close 是 + **VARCHAR**, 读出来转 float。用原始收盘(非前复权): ≤20 日窗口内除权少, 且与账本 price_at + 同为原始价, 口径一致。库不可达或该股无行情返回 {}, ③⑤两段自动跳过、不报错。 + """ + sym = _to_prefix(ts_code) + try: + rows = fetch_all( + "SELECT `timestamp` AS d, `close` AS c FROM gp_day_data " + "WHERE symbol = :s AND `timestamp` >= :a AND `timestamp` <= :b " + "ORDER BY `timestamp`", + {"s": sym, "a": start_dt.strftime("%Y-%m-%d 00:00:00"), + "b": end_dt.strftime("%Y-%m-%d 23:59:59")}, source="index") + except Exception: + return {} + out = {} + for r in rows: + d, c = r.get("d"), _f(r.get("c")) + if c <= 0 or d is None: + continue + dd = d.date() if isinstance(d, datetime) else datetime.fromisoformat(str(d)).date() + out[dd] = c + return out + + +def _fwd_closes_after(ts_code: str, sell_dt: datetime, n_max: int = 30) -> list: + """卖出日之后的交易日收盘序列 (按日期升序), 供取 T+1/T+5/T+20。 + 按股缓存**整个分析区间**的日线, 覆盖该股在窗口内的所有卖点 (同股多次卖不会漏窗)。""" + if ts_code not in _CLOSE_CACHE: + lo = (_ANALYSIS_START or sell_dt) - timedelta(days=10) + hi = (_ANALYSIS_END or sell_dt) + timedelta(days=60) + _CLOSE_CACHE[ts_code] = fetch_daily_closes(ts_code, lo, hi) + closes = _CLOSE_CACHE[ts_code] + if not closes: + return [] + sd = sell_dt.date() + after = sorted((d, c) for d, c in closes.items() if d > sd) + return [c for _, c in after] + + +def forward_returns(ts_code: str, sell_dt: datetime, sell_price: float) -> dict: + """卖出后 T+h 的涨跌 (相对卖出价)。正=卖飞(卖完还涨), 负=避坑(卖完就跌)。""" + seq = _fwd_closes_after(ts_code, sell_dt, n_max=max(FWD_HORIZONS)) + out = {} + for h in FWD_HORIZONS: + out[h] = _pct(sell_price, seq[h - 1]) if len(seq) >= h else None + return out + + +# ------------------------------------------------------------------ 成本 +def trade_cost(notional: float, is_sell: bool) -> float: + """单边交易成本(元): 佣金(最低5) + 卖出印花税 + 过户费 + 滑点。""" + n = abs(_f(notional)) + if n <= 0: + return 0.0 + commission = max(n * COMMISSION_RATE, COMMISSION_MIN) + stamp = n * STAMP_RATE if is_sell else 0.0 + transfer = n * TRANSFER_RATE + slip = n * SLIPPAGE_BPS / 10000.0 + return commission + stamp + transfer + slip + + +# ------------------------------------------------------------------ 来回配对 +def pair_round_trips(rows_by_code: dict) -> list: + """同一只票: 把"一次买"和其后"第一次卖"配成一个来回(粗配, 不做逐笔 FIFO)。 + 返回每个来回: 入场/出场时间价、持有交易日近似(自然日近似)、来回毛收益、是否决策系统驱动。 + 说明: 这是方向与量级的体检, 不是会计账; 精算逐笔在 pms_lot, 要精算另说。""" + trips = [] + for code, rows in rows_by_code.items(): + pending_buy = None + for r in rows: + tag = classify(r) + if tag["side"] == "buy": + if pending_buy is None: + pending_buy = r + elif tag["side"] == "sell" and pending_buy is not None: + b, s = pending_buy, r + bt, st = b["decided_at"], s["decided_at"] + bt = bt if isinstance(bt, datetime) else datetime.fromisoformat(str(bt)) + st = st if isinstance(st, datetime) else datetime.fromisoformat(str(st)) + hold_days = (st.date() - bt.date()).days + gross = _pct(b["price_at"], s["price_at"]) + trips.append({ + "ts_code": code, "buy_at": bt, "sell_at": st, + "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"), + }) + pending_buy = None + return trips + + +# ------------------------------------------------------------------ 各段输出 +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"]) + 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" 配成来回 {len(trips)} 组 · 决策系统驱动的快速来回(≤{max(GUARD_DAYS)}交易日) {len(quick)} 组") + return quick + + +def section_roundtrips(trips): + sig = [t for t in trips if t["signal_driven"]] + print("\n② 决策系统驱动的来回 (毛收益, 未扣成本)") + if len(sig) < MIN_SAMPLE: + print(f" 样本 {len(sig)} 组 (<{MIN_SAMPLE}), 只列数不下结论。") + rets = [t["gross_ret"] for t in sig if t["gross_ret"] is not None] + if rets: + avg = sum(rets) / len(rets) + win = sum(1 for x in rets if x > 0) / len(rets) + holds = [t["hold_days"] for t in sig if t["hold_days"] is not None] + avg_hold = sum(holds) / len(holds) if holds else None + print(f" 平均持有 {avg_hold:.1f} 天 · 平均毛收益 {_fmt_pct(avg)} · 盈利占比 {win:.0%}") + for t in sorted(sig, key=lambda x: (x["gross_ret"] is None, x["gross_ret"] or 0))[:8]: + print(f" {t['ts_code']} 持{t['hold_days']}天 毛{_fmt_pct(t['gross_ret'])} " + 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"]] + got = {h: [] for h in FWD_HORIZONS} + for r in sig_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]) + if not any(got.values()): + print(" (前向价源未接通 fetch_daily_closes, 本段跳过 —— 接上后重跑即出)") + return + for h in FWD_HORIZONS: + v = got[h] + if 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(" 读法: 卖后平均为正、卖飞比例高 → 这批卖出多在砍还会涨的票(瞎折腾); " + "为负 → 多在避坑(该卖)。") + + +def section_cost(trips): + print("\n④ 决策系统驱动的来回, 纯摩擦成本") + sig = [t for t in trips if t["signal_driven"] and t["gross_ret"] is not None] + if not sig: + print(" 无样本。") + return + # 无法拿到每笔真实金额时, 以"单位名义1"估相对成本率; 有 hard_numbers.amount 时用真实额 + total_rate = 0.0 + for t in sig: + rt_cost_rate = (COMMISSION_RATE * 2 + STAMP_RATE + TRANSFER_RATE * 2 + + SLIPPAGE_BPS / 10000.0 * 2) + total_rate += rt_cost_rate + print(f" 每个来回的往返摩擦约 {(_rt_cost_rate()):.3%} (佣金双边+印花+过户+滑点双边)") + print(f" {len(sig)} 组来回累计摩擦 ≈ 名义规模的 {total_rate:.2%} " + f"(即平均毛收益要先跑赢这条线才算真挣到)") + + +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 + for N in GUARD_DAYS: + deltas = [] + for t in affected0: + if t["hold_days"] is None or t["hold_days"] > N: + continue # 护栏只管"刚建仓 N 日内"的卖出 + if (t["gross_ret"] or 0) <= HARD_RISK_DROP: + continue # 硬止损放行, 不受护栏拦 + seq = _fwd_closes_after(t["ts_code"], t["sell_at"], n_max=GUARD_HOLD_TO) + if len(seq) < GUARD_HOLD_TO: + continue + # 实际(卖了): 拿到 gross_ret, 并付了一次卖出摩擦; 之后空仓 = 0 + actual = (t["gross_ret"] or 0) - _rt_cost_rate() / 2 + # 反事实(没卖, 持有到 T+H 再看): 用 T+H 相对入场的收益, 只付了买入侧摩擦 + held = _pct(t["buy_price"], seq[GUARD_HOLD_TO - 1]) + if held is None: + continue + counter = held - _rt_cost_rate() / 2 + deltas.append(counter - actual) + if 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)} · " + f"变好占比 {helped:.0%}") + print(" 读法: 某个 N 上平均变化持续为正且变好占比过半 → 这套信号是「噪音来回型」, " + "护栏该上、N 取那档; 若普遍为负 → 是「大波段型」, 别压, 快卖多数是对的。") + + +# ------------------------------------------------------------------ main +def main(): + ap = argparse.ArgumentParser(description="换手体检: 决策系统卖出是避坑还是瞎折腾 (只读)") + ap.add_argument("--days", type=int, default=120, help="回看多少自然日 (默认120)") + ap.add_argument("--since", type=str, default=None, help="或指定起始日 YYYY-MM-DD") + args = ap.parse_args() + + since = (datetime.strptime(args.since, "%Y-%m-%d") if args.since + else datetime.now() - timedelta(days=args.days)) + global _ANALYSIS_START, _ANALYSIS_END + _ANALYSIS_START, _ANALYSIS_END = since, datetime.now() + print(f"换手体检 · 账本自 {since:%Y-%m-%d} 起 · 成本口径 往返≈{_rt_cost_rate():.3%}") + + try: + rows = read_ledger(since) + except DBUnavailable as e: + print(f"[FAIL] 读账本失败(库不可达): {e}"); sys.exit(1) + if not rows: + print("账本在该窗口内为空 —— 换个更长的 --days 再看。"); return + + by_code = {} + for r in rows: + by_code.setdefault(r["ts_code"], []).append(r) + trips = pair_round_trips(by_code) + + section_inventory(rows, trips) + section_roundtrips(trips) + section_forward(rows) + section_cost(trips) + section_counterfactual(trips) + print("\n完成。前向两段若显示「未接通」, 是历史收盘价源(fetch_daily_closes)还没接 —— " + "确认接哪张表后重跑即全。") + + +if __name__ == "__main__": + main()