From 6aaa4398d6287d377efc170bb177bae08454088a Mon Sep 17 00:00:00 2001 From: zlt Date: Thu, 27 Aug 2026 13:26:24 +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_entry.py | 492 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 492 insertions(+) create mode 100644 scripts/backtest_entry.py diff --git a/scripts/backtest_entry.py b/scripts/backtest_entry.py new file mode 100644 index 0000000..3e54a1b --- /dev/null +++ b/scripts/backtest_entry.py @@ -0,0 +1,492 @@ +# -*- coding: utf-8 -*- +""" +入场体检: 决策系统的买入, 有多少是"买完就被风控清掉"的坏入场, 该不该在入场端加一道过滤 (只读) +============================================================================== +承接换手体检 backtest_churn.py 的结论: 决策系统的快速卖出这一批全是风控止损, 而且基本都砍对了 +(卖完股价还在跌), 加"卖出端最小持有期护栏"只会更差。于是问题被顶到了另一头 —— 不是卖得太快, +是买得不对: 系统反复买那种一两天内就触发风控、只能亏着清掉的票。这个脚本把这批"坏入场"拎出来, +量三件事: + 一, 它们买入时的价格情形有没有共性 (是不是买在冲高、买在已经下跌的票上); + 二, 买入之后股价怎么走 (入场时点本身好不好, 跟全体入场比); + 三, 若在入场端加一道过滤, 是净赚 (挡掉的多是亏的), 还是误伤好票 (挡掉的里不少是赚的)。 + +一句话方法: 同样只读评审账本 pms_action_ledger, 把"一次买紧跟一次卖"配成来回。来回里"卖出是 +决策系统驱动、且持有不超过 N 个交易日"的那批, 就是坏入场。买入时点的价格情形用 gp_day_data 的 +当日 OHLC 重建 (它有 open/high/low/close/volume, 见 DATA_MODEL §1.1)。 + +五段: + ① 样本盘点 全部买入多少、配成来回多少、其中"买完就被风控快速清掉"的坏入场多少。 + ② 坏入场画像 坏入场 vs 全体入场, 比买入日的三个价格特征: 追高度、当日涨跌、前5日动量。 + ③ 入场后前向 每笔买入之后 T+1/5/20 相对买价的涨跌, 坏入场 vs 全体, 看入场时点本身好不好。 + ④ 成本账 这批坏入场的来回一共交了多少纯摩擦。 + ⑤ 反事实过滤 扫几条入场过滤规则(追高/逆势/放量阴线): 各挡掉多少来回、其中坏入场多少(召回)、 + 挡掉的平均实现净收益、误伤率、以及组合净效果。挡掉的多是亏的且净效果为正=该过滤 + 有用; 误伤率高(挡掉的里不少是赚的)=别上。 + +成本口径与前向价源同 backtest_churn.py (往返≈0.262%; gp_day_data @ 18.199 走 app 的 index 源)。 +读数纪律: 样本少于 MIN_SAMPLE 的段落只列数、不下结论 —— 不等数据的判分是编故事。 + +一处口径说明: 判"坏入场"用的是卖出的 dominant_signal 与来源, 与换手体检同口径; 只认"决策系统 +驱动的卖出"配成的来回。到价止盈那类计划位平仓不在决策系统卖出信号里(它走执行层的目标价), +所以不进这里的坏入场, 也不该进 —— 这段量的就是"被风控快速清掉"的那类入场。 + +运行(桥机 factorevaluation, 项目根目录): + docker compose run --rm pms-web python scripts/backtest_entry.py --days 120 +""" +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 归类 (与 backtest_churn.py 同口径) +BUY_ACTIONS = {"OPEN", "FILL", "ADD", "DCA"} +SELL_ACTIONS = {"EXIT", "TRIM"} +SIGNAL_SOURCES = {"intraday", "risk_sell"} +SIGNAL_WORDS = ("决策系统", "风控", "SELL", "转弱", "派发") +TAKE_PROFIT_DOM = "take_profit" # 卖出 dominant_signal 是这个 = 止盈离场, 其余 = 风控止损 + +# "坏入场"判据: 买入配成的来回, 卖出是决策系统驱动、且持有不超过这么多交易日(自然日近似) +QUICK_DAYS = 10 +FWD_HORIZONS = (1, 5, 20) # 入场后前向看这几个交易日 + +# 入场过滤规则的阈值 (⑤ 反事实扫的候选闸门, 都只用买入日及之前的信息, 无未来函数) +CHASE_TOP_FRAC = 0.70 # 追高: 买价落在当日振幅的顶部这个比例以上 (1.0=买在最高) +DOWNTREND_5D = -0.03 # 逆势: 买入日相对前5个交易日收盘已跌超这个幅度 +REDVOL_VOL_MULT = 1.5 # 放量阴线: 买入日是阴线且量能≥前5日均量的这个倍数 + + +# ------------------------------------------------------------------ 小工具 +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 _avg(xs): + xs = [x for x in xs if x is not None] + return (sum(xs) / len(xs)) if xs else None + + +def _rt_cost_rate(): + return COMMISSION_RATE * 2 + STAMP_RATE + TRANSFER_RATE * 2 + SLIPPAGE_BPS / 10000.0 * 2 + + +# ------------------------------------------------------------------ 读账本 (单表, 过守卫) +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(是否决策系统驱动的卖出)、 + 以及 cohort(止盈动量 / 风控止损, 按 hard_numbers.dominant_signal 分, 只对决策系统卖出有值)。""" + 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" + 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)) + 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) +_BAR_CACHE: dict = {} +_ANALYSIS_START = None # main 里设; 让按整个分析区间取足够宽的每股窗口 +_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_bars(ts_code: str, start_dt: datetime, end_dt: datetime) -> dict: + """返回 {date: {open,high,low,close,vol,pre_close,pct}}, 只含有行情的交易日。 + + 源: gp_day_data (db_gp_cj @ 192.168.18.199, 走 app 的 index 数据源)。按 DATA_MODEL §1.1: + symbol 是**前缀式**(这里转一下); open/high/low/close 是 **VARCHAR**(读出转 float); + percent、pre_close 是 DECIMAL。用原始价(非前复权), 与账本 price_at 同口径。 + 库不可达或该股无行情返回 {}, ②③⑤ 相关段落自动跳过、不报错。 + """ + sym = _to_prefix(ts_code) + try: + rows = fetch_all( + "SELECT `timestamp` AS d, `open` AS o, `high` AS h, `low` AS l, `close` AS c, " + "`volume` AS v, `pre_close` AS pc, `percent` AS pct 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 = r.get("d") + if d is None: + continue + dd = d.date() if isinstance(d, datetime) else datetime.fromisoformat(str(d)).date() + close = _f(r.get("c")) + if close <= 0: + continue + out[dd] = {"open": _f(r.get("o")), "high": _f(r.get("h")), "low": _f(r.get("l")), + "close": close, "vol": _f(r.get("v")), + "pre_close": _f(r.get("pc")), "pct": _f(r.get("pct"))} + return out + + +def _bars(ts_code: str, anchor_dt: datetime) -> dict: + """按股缓存**整个分析区间**的日线(带前后余量), 覆盖该股所有买卖点。""" + if ts_code not in _BAR_CACHE: + lo = (_ANALYSIS_START or anchor_dt) - timedelta(days=30) # 前多留些, 好算前5日动量 + hi = (_ANALYSIS_END or anchor_dt) + timedelta(days=60) + _BAR_CACHE[ts_code] = fetch_daily_bars(ts_code, lo, hi) + return _BAR_CACHE[ts_code] + + +def _closes_after(ts_code: str, dt: datetime) -> list: + """某日之后的交易日收盘序列(升序), 供取前向 T+1/T+5/T+20。""" + bars = _bars(ts_code, dt) + if not bars: + return [] + d0 = dt.date() + after = sorted((d, b["close"]) for d, b in bars.items() if d > d0) + return [c for _, c in after] + + +def forward_returns(ts_code: str, dt: datetime, base_price: float) -> dict: + """dt 之后 T+h 相对 base_price 的涨跌。""" + seq = _closes_after(ts_code, dt) + return {h: (_pct(base_price, seq[h - 1]) if len(seq) >= h else None) for h in FWD_HORIZONS} + + +def entry_context(ts_code: str, buy_dt: datetime, buy_price: float) -> dict: + """重建买入时点的价格情形。返回 {chase, day_chg, mom5, red_vol}, 缺数据的项为 None。 + chase 追高度 = (买价-当日最低)/(当日最高-当日最低), 1.0=买在最高, 越高越追。 + day_chg 买入日涨跌 = 当日 percent(优先) 或 收盘/前收-1, 正=买在红盘。 + mom5 前5日动量 = 买入日收盘/前5个交易日收盘-1, 负=买在已经下跌的票上(逆势)。 + red_vol 放量阴线 = 买入日收盘<开盘 且 量能≥前5日均量×倍数 (True/False/None)。 + """ + bars = _bars(ts_code, buy_dt) + ctx = {"chase": None, "day_chg": None, "mom5": None, "red_vol": None} + if not bars: + return ctx + bd = buy_dt.date() + bar = bars.get(bd) + if bar is None: + # 账本时间戳那天没有行情(极少见), 用其后第一根近似 + later = sorted(d for d in bars if d >= bd) + if not later: + return ctx + bd = later[0] + bar = bars[bd] + hi, lo = bar["high"], bar["low"] + if buy_price > 0 and hi > lo: + ctx["chase"] = max(0.0, min(1.0, (buy_price - lo) / (hi - lo))) + if bar["pct"]: + ctx["day_chg"] = bar["pct"] / 100.0 + elif bar["pre_close"] > 0: + ctx["day_chg"] = bar["close"] / bar["pre_close"] - 1.0 + prior = sorted((d, b) for d, b in bars.items() if d < bd) + if len(prior) >= 5: + c5 = prior[-5][1]["close"] + if c5 > 0: + ctx["mom5"] = bar["close"] / c5 - 1.0 + vols = [b["vol"] for _, b in prior[-5:] if b["vol"] > 0] + if vols and bar["vol"] > 0: + avgv = sum(vols) / len(vols) + ctx["red_vol"] = (bar["close"] < bar["open"]) and (bar["vol"] >= REDVOL_VOL_MULT * avgv) + return ctx + + +# ------------------------------------------------------------------ 成本 +def trade_cost(notional: float, is_sell: bool) -> float: + 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)。 + 说明: 多次买(加仓)只认第一笔为入场; 入场时点体检看的是首次进场那一下。""" + 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"], "cohort": tag.get("cohort"), + "sell_reason": s.get("reason"), + }) + pending_buy = None + return trips + + +def is_bad_entry(t: dict) -> bool: + """坏入场: 决策系统驱动的卖出、且持有不超过 QUICK_DAYS —— 买完就被风控快速清掉的那批。""" + return bool(t["signal_driven"]) and t["hold_days"] is not None and t["hold_days"] <= QUICK_DAYS + + +# ------------------------------------------------------------------ 各段输出 +def section_inventory(rows, trips): + 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") + bad = [t for t in trips if is_bad_entry(t)] + print("\n① 样本盘点") + print(f" 账本落地买入 {buys} 笔 · 落地卖出 {sells} 笔 · 配成来回 {len(trips)} 组") + print(f" 其中「买完就被风控快速清掉」的坏入场(卖出为决策系统驱动、持有≤{QUICK_DAYS}交易日) " + f"{len(bad)} 组") + return bad + + +def _profile(trips): + """算一批来回的入场画像三特征均值。返回 (chase, day_chg, mom5, n_ctx)。""" + ch, dc, mo = [], [], [] + for t in trips: + c = entry_context(t["ts_code"], t["buy_at"], t["buy_price"]) + ch.append(c["chase"]); dc.append(c["day_chg"]); mo.append(c["mom5"]) + n_ctx = sum(1 for x in ch if x is not None) + return _avg(ch), _avg(dc), _avg(mo), n_ctx + + +def section_profile(trips): + """② 坏入场画像 vs 全体入场: 三个买入日价格特征。""" + print("\n② 坏入场画像 vs 全体入场 (买入日的价格情形)") + all_paired = [t for t in trips if t["buy_price"] > 0] + bad = [t for t in all_paired if is_bad_entry(t)] + ac, ad, am, an = _profile(all_paired) + bc, bd, bm, bn = _profile(bad) + if an == 0: + print(" (前向/日线价源未接通 fetch_daily_bars, 本段跳过 —— 接上后重跑即出)") + return + print(f" 全体入场 {len(all_paired)} 组(有行情 {an} 组): " + f"追高度均值 {ac:.2f} · 买入日涨跌 {_fmt_pct(ad)} · 前5日动量 {_fmt_pct(am)}") + if bn < MIN_SAMPLE: + print(f" 坏入场 {len(bad)} 组(有行情 {bn} 组, <{MIN_SAMPLE}): 只列数不下结论。") + print(f" 坏入场 {len(bad)} 组(有行情 {bn} 组): " + f"追高度均值 {bc if bc is None else round(bc,2)} · 买入日涨跌 {_fmt_pct(bd)} · " + f"前5日动量 {_fmt_pct(bm)}") + print(" 读法: 追高度=买价在当日最高最低之间的位置(1=买在最高)。坏入场若追高度明显更高、" + "或前5日动量明显更负, 说明它们多买在冲高或买在已下跌的票上 —— 那正是可在入场端拦的把手。") + + +def _fwd_profile(trips): + got = {h: [] for h in FWD_HORIZONS} + for t in trips: + fr = forward_returns(t["ts_code"], t["buy_at"], t["buy_price"]) + for h in FWD_HORIZONS: + if fr[h] is not None: + got[h].append(fr[h]) + return got + + +def _print_fwd(label, got): + if not any(got.values()): + 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}), 只报数") + continue + avg = sum(v) / len(v) + down = sum(1 for x in v if x < 0) / len(v) + print(f" T+{h}: 样本 {len(v)} · 买后平均 {_fmt_pct(avg)} · 买后就跌比例 {down:.0%}") + return True + + +def section_forward(trips): + """③ 入场后前向收益: 买入时点本身好不好 (坏入场 vs 全体)。""" + print("\n③ 买入之后的前向收益 (相对买价; 负=买在了下跌前, 入场时点差)") + all_paired = [t for t in trips if t["buy_price"] > 0] + bad = [t for t in all_paired if is_bad_entry(t)] + if not _print_fwd("全体入场", _fwd_profile(all_paired)): + print(" (前向价源未接通 fetch_daily_bars, 本段跳过 —— 接上后重跑即出)") + return + _print_fwd("坏入场 (买完就被风控清掉这批)", _fwd_profile(bad)) + print(" 读法: 坏入场买后前向明显比全体更负=这批入场时点确实差(不是卖错是买错); " + "若全体入场买后也普遍为负, 说明入场时点是系统性问题, 不止这十几笔。") + + +def section_cost(trips): + print("\n④ 坏入场的来回, 纯摩擦成本") + bad = [t for t in trips if is_bad_entry(t) and t["gross_ret"] is not None] + if not bad: + print(" 无样本。") + return + total_rate = len(bad) * _rt_cost_rate() + print(f" 每个来回往返摩擦约 {_rt_cost_rate():.3%} · {len(bad)} 组坏入场累计摩擦 ≈ " + f"名义规模的 {total_rate:.2%}") + print(" 读法: 这是「买错就得原路平掉」白交的过路费, 少买一笔坏入场就省一份。") + + +# ---- ⑤ 反事实入场过滤 ---- +def _filters(): + """候选入场闸门: 每条是 (名称, 说明, 判定函数(trip,ctx)->bool 命中即"该拦")。 + 只用买入日及之前的信息, 无未来函数。""" + return [ + ("追高", f"买价落在当日振幅顶部{(1-CHASE_TOP_FRAC):.0%}以内(追高度≥{CHASE_TOP_FRAC})", + lambda t, c: c["chase"] is not None and c["chase"] >= CHASE_TOP_FRAC), + ("逆势买", f"买入日已较前5日跌超{abs(DOWNTREND_5D):.0%}(前5日动量≤{DOWNTREND_5D:.0%})", + lambda t, c: c["mom5"] is not None and c["mom5"] <= DOWNTREND_5D), + ("放量阴线", f"买入日是阴线且量能≥前5日均量×{REDVOL_VOL_MULT}", + lambda t, c: c["red_vol"] is True), + ] + + +def section_counterfactual(trips): + """⑤ 反事实: 在入场端加一道过滤, 净赚还是误伤好票。""" + print("\n⑤ 反事实: 入场端加一道过滤, 净赚还是误伤好票") + uni = [t for t in trips if t["buy_price"] > 0 and t["gross_ret"] is not None] + # 预取每笔的入场情形, 顺带探价源 + ctxs = {id(t): entry_context(t["ts_code"], t["buy_at"], t["buy_price"]) for t in uni} + if not any(c["chase"] is not None or c["mom5"] is not None for c in ctxs.values()): + print(" (前向/日线价源未接通, 本段跳过 —— 接上 fetch_daily_bars 后重跑即出)") + return + n_bad = sum(1 for t in uni if is_bad_entry(t)) + print(f" 口径: 全体可判来回 {len(uni)} 组(其中坏入场 {n_bad} 组)。" + f"每条过滤=不买命中的那些票, 省掉其整个来回的实现净收益(毛收益减往返摩擦{_rt_cost_rate():.2%})。") + for name, desc, fn in _filters(): + blocked = [t for t in uni if fn(t, ctxs[id(t)])] + if not blocked: + print(f" · {name}({desc}): 一笔没命中, 跳过。") + continue + nets = [t["gross_ret"] - _rt_cost_rate() for t in blocked] + caught_bad = sum(1 for t in blocked if is_bad_entry(t)) + winners = sum(1 for x in nets if x > 0) + avg_net = sum(nets) / len(nets) + portfolio = -sum(nets) # 不买这些 → 组合少了它们的净收益; 正=少亏(有用) + recall = (caught_bad / n_bad) if n_bad else None + if len(blocked) < MIN_SAMPLE: + print(f" · {name}({desc}): 挡掉 {len(blocked)} 组(<{MIN_SAMPLE}, 只报数) · " + f"含坏入场 {caught_bad} 组 · 挡掉里赚钱 {winners} 组") + continue + print(f" · {name}: 挡掉 {len(blocked)} 组 · 含坏入场 {caught_bad} 组" + f"{'' if recall is None else f'(召回坏入场 {recall:.0%})'} · " + f"挡掉平均实现净收益 {_fmt_pct(avg_net)} · 误伤率 {winners/len(blocked):.0%} · " + f"组合净效果 {_fmt_pct(portfolio)}") + print(f" ({desc})") + print(" 读法: 某条过滤「挡掉平均实现净收益」为负(挡掉的多是亏钱来回)、误伤率低、组合净效果为正," + " 才值得上; 若误伤率高(挡掉里不少是赚的)或净效果为负, 说明它连好票一起误杀, 别上。" + " 召回=这条能拦住多少比例的坏入场。") + + +# ------------------------------------------------------------------ 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%} · " + f"坏入场=决策系统卖出且持有≤{QUICK_DAYS}交易日") + + 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_profile(trips) + section_forward(trips) + section_cost(trips) + section_counterfactual(trips) + print("\n完成。②③⑤ 若显示「未接通」, 是历史日线源(fetch_daily_bars)还没接 —— " + "确认接哪张表后重跑即全。") + + +if __name__ == "__main__": + main()