akg-factor-bridge/plan_review.py

"""候选单复盘（只读）：这个系统唯一的评价方式——名单级观察收益，不是回测。

## 为什么要有它

方案（docs/主观选股改进方案_2026-09-02.md 第 2.4 节）把评价单位定为"候选单"而不是因子：
不建仓、不计成本、不定仓位、不算净值，只看几份名单在后五、十、二十个交易日相对
全池与主榜的超额和命中率，按环境与证据线分组，每周五出报告。它取代了原对比器
score_lab（两套读数工具两种口径的坑），也承接了 09-02 手工归因的三组读数：
第一步先"对账"——用 signal_close 口径复现方案 1.8b 的四档读数（强传导 −2.04、
观察档 +0.52），证明脚本口径与手工一致；第二步再切常规口径 next_close 出周报。

本脚本不提供权重参数，只提供分组与期限——防止复盘变成调参或模拟交易。

## 六份名单

  生产名单   PMS 实际拿到的候选：PMS 计划快照名册（pms_plan_snapshot.roster_json）里档位为
             强传导、按分数降序前 N（N 取当时 PMS_PLAN_TOP_N，历史值不可还原，用现值并注明）；
             快照缺失的日子退为桥按同规则重算并标注
  候选单     候选卡判决为"候选"（候选卡上线日之前的历史日，按当日快照或重算得出，
             "已启动"依赖已动成员视图，视图未建的日子候选为空并标注）
  关注单     判决为"关注"
  环节名单   当日被传导指向的环节的全体成员等权（把"定位对"与"选票对"分开）
  主榜等权   当日全部主榜
  全池等权   基座行情快照当日全部个股（基准）

## 另四份名单（2026-09-03 方案第 3.3 节"复盘四份名单与对照节"）

  机器通过名单   PMS 动作账本 pms_action_ledger 当日 action='OPEN'、arbiter='judge'、verdict='PASS' 的票
  人批名单       同表当日 arbiter='user'、verdict='PASS'
  人拒名单       同表当日 arbiter='user'、verdict='REJECT'
  择时看多名单   择时决策系统结论表 strategy_daily_results 当日 signal_type='BUY' 的票
  账本按 decided_at 的日历日筛，代码统一转前缀式；任一路读失败该名单为空并在逐日注记里说明。
  周报第七节"三套对照"把选股系统候选单、择时决策系统自评、PMS 账本四份名单摆在一张表里；
  第八节"台账对表"列出 docs/复盘决定台账.md 的条目，留"一致 / 不一致"两列给人填。

## 口径

  起点价  next_close（默认，实盘买得到）：T 日出计划，T+1 收盘买，收益 = Σ pct[T+2 .. T+1+h]
          signal_close（对账用）：收益 = Σ pct[T+1 .. T+h]，与方案 1.8 系列的手工口径一致
  期限    5 / 10 / 20 个交易日
  指标    均收益、相对全池超额、相对主榜超额、跑赢全池比例（命中率）
  分组    档位、判决、吸筹三态（基座行情快照当日的 accum 状态）、事后环境（全池后 h 日涨跌，
          只作解释）、事前标签（当日快照 regime 段，上线后才有）、市值三分位（成交额除以换手率）

## 跑法【桥机 155 · ~/project/akg-factor-bridge】

  docker compose exec -T akg-factor-bridge python plan_review.py --since 2026-07-29 --horizons 5,10,20
  docker compose exec -T akg-factor-bridge python plan_review.py --since 2026-07-29 --start-price signal_close --horizons 5   # 对账 1.8b

只读：因子表、基座视图与行情快照、PMS 计划快照全部 SELECT；只写 data/review/ 下的报告与明细。
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import re

import pandas as pd

import common
import config
import db
import plan

HORIZONS_DEFAULT = (5, 10, 20)
MAIN_MIN = 150.0
LEDGER_LISTS = ("机器通过名单", "人批名单", "人拒名单")
TIMING_LIST = "择时看多名单"
DECISION_LEDGER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "docs", "复盘决定台账.md")


# ============================================================================
# 数据
# ============================================================================
def plan_dates(since: str, until: str | None) -> list[str]:
    q = "SELECT DISTINCT trade_date FROM t_factor_akg_gate WHERE trade_date >= %s"
    args = [since]
    if until:
        q += " AND trade_date <= %s"
        args.append(until)
    df = db.read_mysql("factor", q + " ORDER BY trade_date", tuple(args))
    return [pd.Timestamp(x).date().isoformat() for x in df["trade_date"]]


def price_panel(since: str, days_after: int = 30) -> pd.DataFrame:
    """基座行情快照：trade_date × 前缀码 -> 日涨幅（百分数）。"""
    end = (dt.date.fromisoformat(since) + dt.timedelta(days=200)).isoformat()
    df = db.read_pg(
        "SELECT trade_date, code, (metrics->>'pct_change')::float AS pct, "
        "metrics->'accum'->>'state' AS accum FROM mkt_daily "
        "WHERE kind='stock' AND trade_date >= %s AND trade_date <= %s", (since, end))
    df["k"] = df["code"].map(common.to_prefix)
    df["trade_date"] = pd.to_datetime(df["trade_date"]).dt.date.astype(str)
    return df


def cap_bucket(day: str) -> dict[str, str]:
    """市值三分位（成交额除以换手率的近似流通市值），同日分桶。读不到返回空。"""
    try:
        df = db.read_mysql(
            "factor", "SELECT symbol, amount, turnoverrate FROM gp_day_data "
                      "WHERE DATE(`timestamp`) = %s AND turnoverrate > 0 AND amount > 0", (day,))
    except Exception as e:  # noqa: BLE001
        print(f"  （市值分桶读取失败 {day}: {e!r}）")
        return {}
    if df.empty:
        return {}
    df["k"] = df["symbol"].astype(str).str.strip().map(common.to_prefix)
    df["mv"] = df["amount"] / df["turnoverrate"]
    try:
        df["cap"] = pd.qcut(df["mv"], 3, labels=["小盘", "中盘", "大盘"])
    except ValueError:
        return {}
    return dict(zip(df["k"], df["cap"].astype(str)))


def pms_roster(day: str) -> tuple[list[str], str]:
    """PMS 当日拿到的生产名单：计划快照名册里档位强传导、按分数序前 N。返回 (代码, 注记)。"""
    try:
        df = db.read_mysql(
            "pms", "SELECT roster_json, fetched_at FROM pms_plan_snapshot "
                   "WHERE plan_date = %s ORDER BY id DESC LIMIT 1", (day,))
        n_df = db.read_mysql(
            "pms", "SELECT param_value FROM pms_runtime_param WHERE param_key='PMS_PLAN_TOP_N'")
    except Exception as e:  # noqa: BLE001
        # 异常原文会被印进周报的逐日注记。只进日志，不进报告。
        print(f"  （PMS 计划快照读取失败: {e!r}）")
        return [], "当天没能连上 PMS 数据库，取不到计划快照"
    if df.empty:
        return [], "PMS 无当日快照，生产名单退为桥重算"
    n = int(n_df.iloc[0, 0]) if not n_df.empty else 30
    roster = json.loads(df.iloc[0]["roster_json"] or "[]")
    rows = [r for r in roster if str(r.get("t") or "") == "强传导" and str(r.get("b") or "main") == "main"]
    rows.sort(key=lambda r: -(r.get("s") or 0))
    return [common.to_prefix(str(r["c"]).strip()) for r in rows[:n] if r.get("c")], \
        f"PMS 快照名册（N={n} 为现值，历史 N 不可还原）"


def _uniq_codes(values) -> list[str]:
    """代码列 -> 去重、去占位符（账本里宏观闸等行的 ts_code 是 "-"）、转前缀式，保持出现顺序。"""
    out, seen = [], set()
    for v in values:
        s = str(v or "").strip()
        if not s or s == "-" or s.lower() == "nan":
            continue
        k = common.to_prefix(s.upper())
        if k not in seen:
            seen.add(k)
            out.append(k)
    return out


def ledger_lists(day: str) -> tuple[dict[str, list[str]], str]:
    """PMS 动作账本当日新建仓评审的三份名单：机器通过（研判闸 judge 放行）、人批、人拒。
    decided_at 按日历日筛（day 零点到次日零点）；读失败三份都为空并返回原因。"""
    empty = {name: [] for name in LEDGER_LISTS}
    nxt = (dt.date.fromisoformat(day) + dt.timedelta(days=1)).isoformat()
    try:
        df = db.read_mysql(
            "pms", "SELECT ts_code, arbiter, verdict FROM pms_action_ledger "
                   "WHERE action = 'OPEN' AND decided_at >= %s AND decided_at < %s", (day, nxt))
    except Exception as e:  # noqa: BLE001
        return empty, f"PMS 账本读取失败（{e!r}），机器通过、人批、人拒三份名单为空"
    if df.empty:
        return empty, "PMS 账本当日无新建仓评审行"
    arb = df["arbiter"].astype(str).str.strip().str.lower()
    vd = df["verdict"].astype(str).str.strip().str.upper()
    return {
        "机器通过名单": _uniq_codes(df.loc[(arb == "judge") & (vd == "PASS"), "ts_code"]),
        "人批名单": _uniq_codes(df.loc[(arb == "user") & (vd == "PASS"), "ts_code"]),
        "人拒名单": _uniq_codes(df.loc[(arb == "user") & (vd == "REJECT"), "ts_code"]),
    }, f"PMS 账本当日评审行 {len(df)} 条"


def timing_bullish(day: str) -> tuple[list[str], str]:
    """择时决策系统结论表当日 signal_type='BUY' 的票（trade_date 是整数 YYYYMMDD）。读失败为空。"""
    try:
        df = db.read_mysql(
            "pms", "SELECT stock_code FROM strategy_daily_results "
                   "WHERE trade_date = %s AND signal_type = 'BUY'", (int(day.replace("-", "")),))
    except Exception as e:  # noqa: BLE001
        return [], f"择时决策系统结论表读取失败（{e!r}），择时看多名单为空"
    return _uniq_codes(df["stock_code"]) if not df.empty else [], ""


def timing_self_eval(since: str, until: str | None) -> str:
    """择时决策系统自评口径与本期读数：判分表 decision_outcome 里日终策略（ref_type='strategy'）
    五日方向命中率——命中按原始收益方向判，不是超额（方案第 1.3 节列为已知缺陷）。读不到写"未接入"。"""
    lo = int(since.replace("-", ""))
    hi = int((until or dt.date.today().isoformat()).replace("-", ""))
    try:
        df = db.read_mysql(
            "pms", "SELECT COUNT(*) AS n, SUM(dir_hit) AS hits FROM decision_outcome "
                   "WHERE ref_type = 'strategy' AND horizon = 5 AND dir_hit IS NOT NULL "
                   "AND base_date >= %s AND base_date <= %s", (lo, hi))
        n = int(df.iloc[0]["n"] or 0) if not df.empty else 0
        hits = int(df.iloc[0]["hits"] or 0) if not df.empty else 0
    except Exception as e:  # noqa: BLE001
        # 表名是 PMS 的库表名，不该出现在周报的读数列里。原因写进运行日志。
        print(f"  （择时决策系统的打分表读取失败: {e!r}）")
        return (f"读不到择时决策系统的打分数据，"
                f"{since} 至 {until or '今日'} 这段的五日方向命中率这期算不出来")
    if n == 0:
        # 与"读不到"分开写：这两种情况在表里要一眼看得出差别。
        return "择时决策系统这段时间没有打过分的日终策略记录，这期没有命中率"
    return (f"看 5 个交易日后涨跌方向判断对不对，{n} 条里对了 {hits} 条，"
            f"命中率 {hits / n * 100:.0f}%（{since} 至 {until or '今日'}）")


def decision_ledger_entries(path: str = DECISION_LEDGER_PATH) -> list[dict]:
    """解析 docs/复盘决定台账.md 的条目标题行 "## 0NN · 日期 · 标题" -> [{no, date, title}]。
    文件不存在或没有条目返回空列表，周报对表节据此写"台账文件缺失"。"""
    try:
        with open(path, "r", encoding="utf-8") as f:
            text = f.read()
    except OSError:
        return []
    pat = re.compile(r"^##\s+(\d{3})\s*·\s*(\d{4}-\d{2}-\d{2})\s*·\s*(.+?)\s*$", re.M)
    return [{"no": m.group(1), "date": m.group(2), "title": m.group(3)} for m in pat.finditer(text)]


# ============================================================================
# 收益
# ============================================================================
def forward(pivot: pd.DataFrame, days: list[str], day: str, h: int, start: str) -> pd.Series | None:
    """从 day 起按口径取后 h 日累计涨幅（每票）。数据不够返回 None。"""
    if day not in days:
        return None
    i = days.index(day)
    lo = i + (2 if start == "next_close" else 1)
    hi = lo + h                                  # 切片 [lo, hi)
    if hi > len(days):
        return None
    block = pivot.loc[days[lo:hi]]
    return block.sum(min_count=h)


# 表头的中文说法。周报是给人读的，列名不该是代码里的字段名。
# 有几处必须区分开，不能想当然地译：
#   excess_all 与 beat 都跟"全池"比，但一个是收益差、一个是只数比例，
#     译成同一个说法会在同一张表里紧挨着撞车；
#   h 不能译成"持有天数"——这个脚本开头反复写明不建仓、不计成本、不是回测，
#     它只是"往后看几个交易日"；
#   grade 只有两档（可读、方向），代码在 :387，导读段落此前写着"三档"是错的。
_COL_CN = {
    "days": "一共几个计划日", "n": "一共几只股票", "ret": "名单平均涨跌%",
    "excess_all": "比全池多涨几个点", "excess_main": "比主榜多涨几个点",
    "excess_cap": "比同市值股票多涨几个点",
    "beat": "跑赢全池的股票占比%", "share": "最多的一天占了多少%",
    "grade": "样本够不够看", "h": "往后看几个交易日",
    "list": "名单", "group": "分组", "date": "计划日",
    "regime_post": "这几天全池是涨是跌", "regime_pre": "下单当天是不是弱势日",
}


# 加权版列名到本体列名的对应。多数是直接去掉 _w 后缀就能对上，只有 excess_w 例外——
# 它算的是相对全池的超额（:433），本体列叫 excess_all，去掉后缀查不到。
_W_BASE = {"excess_w": "excess_all"}


def _col_cn(c) -> str:
    """列名翻成中文。带 _w 后缀的是同一个读数的样本加权版本，只用于对账。

    翻不出来的原样返回而不是硬编一个中文——列名对不上时宁可露出原名让人发现，
    也好过给它安一个错的说法。
    """
    name = str(c)
    if name.endswith("_w"):
        base = _COL_CN.get(_W_BASE.get(name) or name[:-2])
        return f"股票多的日子算得重·{base}" if base else name
    return _COL_CN.get(name, name)


def _md(df: pd.DataFrame) -> str:
    """自己拼 Markdown 表：桥镜像没装 tabulate，pandas.to_markdown 用不了。

    表头统一在这里翻成中文，每张表都受益，不用逐处改。"""
    if df is None or df.empty:
        return "（无数据）"
    cols = [_col_cn(c) for c in df.columns]
    lines = ["| " + " | ".join(cols) + " |", "|" + "---|" * len(cols)]
    for _, r in df.iterrows():
        lines.append("| " + " | ".join("" if (isinstance(v, float) and pd.isna(v)) else str(v)
                                        for v in r.tolist()) + " |")
    return "\n".join(lines)


def summarize(ret: pd.Series, codes: list[str], base_all: float, base_main: float,
              caps: dict | None = None, cap_base: dict | None = None) -> dict | None:
    """一份名单在一个计划日、一个期限上的读数。
    excess_all 相对全池等权；excess_main 相对主榜等权（同为有券商覆盖的篮子，剥掉大票对小票的贝塔）；
    excess_cap 相对同市值桶均值（每票减当日同桶全池均值再平均，方案 1.8c 的市值中性口径）。"""
    r = ret.reindex([c for c in codes if c in ret.index]).dropna()
    if r.empty:
        return None
    out = {"n": int(len(r)), "ret": round(float(r.mean()), 2),
           "excess_all": round(float(r.mean() - base_all), 2),
           "excess_main": round(float(r.mean() - base_main), 2) if base_main is not None else None,
           "excess_cap": None,
           "beat": round(float((r > base_all).mean() * 100), 1)}
    if caps and cap_base:
        adj = [float(v) - cap_base[caps[c]] for c, v in r.items() if caps.get(c) in cap_base]
        if adj:
            out["excess_cap"] = round(sum(adj) / len(adj), 2)
    return out


# ============================================================================
# 主流程
# ============================================================================
def run(since: str, until: str | None, horizons: tuple, start: str, out_dir: str,
        with_cards: bool = True) -> dict:
    days_plan = plan_dates(since, until)
    if not days_plan:
        raise SystemExit("区间内没有档位日。")
    px = price_panel(since)
    pivot = px.pivot_table(index="trade_date", columns="k", values="pct")
    days = sorted(pivot.index.tolist())
    accum_by_day = {d: dict(zip(g["k"], g["accum"])) for d, g in px.groupby("trade_date")}

    rows, notes = [], []
    for day in days_plan:
        try:
            data = plan.collect(day, top=5000, obs_top=5000, theme_cap=0)
        except Exception as e:  # noqa: BLE001
            notes.append(f"{day}: 计划重算失败 {e!r}")
            continue
        full = data.get("_full") or {}
        main_rows, obs_rows = full.get("main", []), full.get("observe", [])
        main_codes = [r["code"] for r in main_rows]
        obs_codes = [r["code"] for r in obs_rows]
        tier_of = {r["code"]: r.get("tier") for r in main_rows}
        verdict_of = {r["code"]: r.get("verdict") for r in main_rows + obs_rows}
        seg_codes = sorted({r["code"] for r in main_rows + obs_rows if r.get("evidence")})
        prod, prod_note = pms_roster(day)
        if not prod:
            # 无 PMS 快照的日子不再退化为桥重算（那不是 PMS 拿到的名单）：交付名单当日剔除并注记（台账 010）
            prod_note = f"{prod_note}；无快照，交付名单当日剔除"
        cands = [r["code"] for r in data.get("candidates") or []]
        watch = [r["code"] for r in data.get("watch") or []]
        ledger, ledger_note = ledger_lists(day)
        bullish, bullish_note = timing_bullish(day)
        caps = cap_bucket(day)
        acc = accum_by_day.get(day, {})
        reg = None
        try:
            import regime
            reg = regime.read_from_snapshot(day)
        except Exception:  # noqa: BLE001
            reg = None

        for h in horizons:
            ret = forward(pivot, days, day, h, start)
            if ret is None:
                continue
            base_all = float(ret.dropna().mean())
            main_ret = ret.reindex([c for c in main_codes if c in ret.index]).dropna()
            base_main = float(main_ret.mean()) if not main_ret.empty else None
            cap_base: dict = {}                 # 当日三个市值桶各自的全池均值，供 excess_cap 用
            if caps:
                for cap in ("小盘", "中盘", "大盘"):
                    v = ret.reindex([c for c, b in caps.items() if b == cap and c in ret.index]).dropna()
                    if not v.empty:
                        cap_base[cap] = float(v.mean())
            regime_post = "涨周" if base_all > 0 else "跌周"
            regime_pre = (("弱势日" if reg.get("weak_day") else "非弱势日")
                          if reg and reg.get("weak_day") is not None else "无标签")
            lists = {
                "强传导交付名单": prod, "候选单": cands, "关注单": watch, "环节名单": seg_codes,
                "主榜等权": main_codes, "观察档等权": obs_codes,
                "全池等权": list(ret.dropna().index),
                # 2026-09-03：PMS 账本三份与择时看多一份，与其余名单同口径算收益
                **ledger, TIMING_LIST: bullish,
            }
            for name, codes in lists.items():
                s = summarize(ret, codes, base_all, base_main, caps, cap_base)
                if s:
                    rows.append({"date": day, "h": h, "list": name, "group": "全部",
                                 "regime_post": regime_post, "regime_pre": regime_pre, **s})
            # 分组：档位、判决、吸筹三态、市值
            for tier in ("强传导", "弱传导", "无传导"):
                codes = [c for c, t in tier_of.items() if t == tier]
                s = summarize(ret, codes, base_all, base_main, caps, cap_base)
                if s:
                    rows.append({"date": day, "h": h, "list": "主榜", "group": f"档位={tier}",
                                 "regime_post": regime_post, "regime_pre": regime_pre, **s})
            for v in ("候选", "关注", "仅展示"):
                codes = [c for c, vv in verdict_of.items() if vv == v]
                s = summarize(ret, codes, base_all, base_main, caps, cap_base)
                if s:
                    rows.append({"date": day, "h": h, "list": "档位表", "group": f"判决={v}",
                                 "regime_post": regime_post, "regime_pre": regime_pre, **s})
            for st_label, pred in (("明确吸筹", lambda s: str(s).startswith("明确")),
                                   ("潜在吸筹", lambda s: str(s).startswith("潜在")),
                                   ("其他", lambda s: not (str(s).startswith("明确") or str(s).startswith("潜在")))):
                codes = [c for c, s in acc.items() if s and pred(s)]
                s = summarize(ret, codes, base_all, base_main, caps, cap_base)
                if s:
                    rows.append({"date": day, "h": h, "list": "全池", "group": f"吸筹={st_label}",
                                 "regime_post": regime_post, "regime_pre": regime_pre, **s})
            if caps:
                for cap in ("小盘", "中盘", "大盘"):
                    codes = [c for c in cands if caps.get(c) == cap]
                    s = summarize(ret, codes, base_all, base_main, caps, cap_base)
                    if s:
                        rows.append({"date": day, "h": h, "list": "候选单", "group": f"市值={cap}",
                                     "regime_post": regime_post, "regime_pre": regime_pre, **s})
        notes.append(f"{day}: 主榜 {len(main_codes)} 观察 {len(obs_rows)} 候选 {len(cands)} "
                     f"关注 {len(watch)} 生产 {len(prod)}（{prod_note}）；"
                     f"账本 机器通过 {len(ledger['机器通过名单'])} 人批 {len(ledger['人批名单'])} "
                     f"人拒 {len(ledger['人拒名单'])}（{ledger_note}）；择时看多 {len(bullish)}"
                     + (f"（{bullish_note}）" if bullish_note else ""))

    df = pd.DataFrame(rows)
    if df.empty:
        raise SystemExit("没有任何可算的期限（数据尾部不足）。")
    os.makedirs(out_dir, exist_ok=True)
    stamp = days_plan[-1]
    csv_path = os.path.join(out_dir, f"复盘明细_{stamp}_{start}.csv")
    df.to_csv(csv_path, index=False, encoding="utf-8-sig")

    # 汇总：按名单 × 期限（全部）；按分组 × 期限；按事后环境 × 名单（只作解释）
    def agg(g: pd.DataFrame) -> pd.Series:
        # 主口径按日等权：每个计划日先算名单均值，再跨日平均——一份名单一天一票；
        # 带 _w 的三列按样本加权（大名单的日子权重大），与方案 1.8b/1.8c 手工读数同口径，只作对账。
        n = g["n"].astype(float)
        w = n / n.sum() if n.sum() else n
        days, total = g["date"].nunique(), int(n.sum())
        # 样本纪律（方案 2.4）机器标注，只有两档：可读=计划日≥20 且样本≥100；其余只看方向
        grade = "够" if (days >= 20 and total >= 100) else "不够，只看方向"
        return pd.Series({"days": days, "n": total,
                          "ret": round(g["ret"].mean(), 2),
                          "excess_all": round(g["excess_all"].mean(), 2),
                          "excess_main": round(g["excess_main"].mean(), 2) if g["excess_main"].notna().any() else None,
                          "excess_cap": round(g["excess_cap"].mean(), 2) if g["excess_cap"].notna().any() else None,
                          "beat": round(g["beat"].mean(), 1),
                          "share": round(float(n.max() / n.sum() * 100), 0) if n.sum() else None,
                          "grade": grade,
                          "ret_w": round(float((g["ret"] * w).sum()), 2),
                          "excess_w": round(float((g["excess_all"] * w).sum()), 2),
                          "beat_w": round(float((g["beat"] * w).sum()), 1)})
    lists_tbl = df[df["group"] == "全部"].groupby(["list", "h"]).apply(agg).reset_index()
    groups_tbl = df[df["group"] != "全部"].groupby(["list", "group", "h"]).apply(agg).reset_index()
    regime_tbl = df[df["group"] == "全部"].groupby(["regime_post", "list", "h"]).apply(agg).reset_index()
    pre_tbl = df[(df["group"] == "全部") & (df["regime_pre"] != "无标签")] \
        .groupby(["regime_pre", "list", "h"]).apply(agg).reset_index()
    h0 = int(df["h"].min())
    daily_c = df[(df["list"] == "候选单") & (df["group"] == "全部") & (df["h"] == h0)] \
        [["date", "n", "ret", "excess_all", "excess_cap", "beat", "regime_post"]].sort_values("date")

    md = [f"# 候选单复盘 · {days_plan[0]} 至 {stamp}（起点价 {start}）", "",
          f"这份周报覆盖 {len(days_plan)} 个计划日。"
          f"一份名单要够二十个计划日、且总共够一百只股票，读数才算够看；"
          f"不够的只看方向，不看具体数值。", "",
          "## 一、六份名单 × 期限", "",
          "怎么读这张表。"
          "「比全池多涨几个点」是拿这份名单和当天全部股票的平均涨跌相比；"
          "「比主榜多涨几个点」是和主榜相比，这样能去掉大盘股和小盘股整体差异的影响；"
          "「比同市值股票多涨几个点」是只和市值相近的股票比。"
          "「跑赢全池的股票占比」和上面几列不是一回事：上面是收益差，这一列是只数比例。"
          "全池自己这一行的占比只有四成多，因为少数大涨的股票把平均值拉高了；"
          "所以读别的名单时要拿它和全池那一行相减，不能直接看绝对值。"
          "「最多的一天占了多少」超过三成，说明整个读数被一天主导，那就只看逐日那张表。"
          "「往后看几个交易日」不是持有天数——这个脚本不建仓、不计成本，不是回测。"
          "不带前缀的列按计划日等权，一天算一票，这是主口径；"
          "带「股票多的日子算得重」前缀的列是另一种算法，只用来和手工读数对账。", "",
          _md(lists_tbl), "",
          "## 二、分组读数", "", _md(groups_tbl), "",
          f"## 二之二、候选单逐日（期限 {h0} 日；第一节按日等权的读数就是这张表的平均，看集中度）", "",
          _md(daily_c), "",
          "## 三、按事后环境分组（未来 h 日全池涨跌，只作解释，不作交易前置）", "",
          _md(regime_tbl), ""]
    if not pre_tbl.empty:
        md += ["## 四、按事前线上标签分组（快照 regime 段，上线后才有）", "",
               _md(pre_tbl), ""]
    else:
        md += ["## 四、按事前线上标签分组", "", "（区间内没有带环境标签的快照，本节待环境标签上线后出现。）", ""]
    md += ["## 五、逐日注记", ""] + [f"- {n}" for n in notes] + ["",
           "## 六、待决定事项", "", "（只列读数与选项，不改任何东西——由每周五人工填写。）", ""]
    md += ["## 七、三套对照（选股系统、择时决策系统、PMS 账本各自的读数摆在一张表里，只对照不合并）", "",
           _md(compare_table(lists_tbl, since, until)), ""]
    md += ["## 八、台账对表（一致 / 不一致两列由人填；对表依据是方案第 4.1 节一致性检查表）", "",
           _md(ledger_table()), ""]
    md_path = os.path.join(out_dir, f"复盘_{stamp}_{start}.md")
    with open(md_path, "w", encoding="utf-8") as f:
        f.write("\n".join(md))
    print("\n".join(md[:8]))
    print(f"\n已写入 {md_path} 与 {csv_path}")
    return {"md": md_path, "csv": csv_path, "days": len(days_plan)}


def _list_reading(lists_tbl: pd.DataFrame, name: str) -> str:
    """第一节汇总表里一份名单各期限的按日等权读数，拼成一句；没有样本写明。"""
    if lists_tbl is None or lists_tbl.empty or "list" not in lists_tbl.columns:
        return "无样本（区间内该名单没有可算的期限：名单为空、读失败或数据尾部不足）"
    sub = lists_tbl[lists_tbl["list"] == name].sort_values("h")
    if sub.empty:
        return "无样本（区间内该名单没有可算的期限：名单为空、读失败或数据尾部不足）"
    parts = []
    for _, r in sub.iterrows():
        parts.append(f"{int(r['h'])} 日：计划日 {int(r['days'])}，样本 {int(r['n'])}，"
                     f"比全池多涨 {r['excess_all']:+.2f} 个点，跑赢全池的占 {r['beat']:.1f}%，样本{r['grade']}")
    return "；".join(parts)


def compare_table(lists_tbl: pd.DataFrame, since: str, until: str | None) -> pd.DataFrame:
    """第七节"三套对照"：三行——选股系统候选单按日等权读数；择时决策系统自评口径与本期方向命中率；
    PMS 账本四份名单读数。三套口径不同，只并列不合并。"""
    rows = [
        {"系统": "选股系统", "口径": "候选单按计划日等权，相对全池等权超额与跑赢比例（第一节主口径）",
         "读数": _list_reading(lists_tbl, "候选单")},
        {"系统": "择时决策系统",
         "口径": "择时决策系统自己打的分。每个交易日收盘后给出的策略，看 5 个交易日后涨跌"
                 "方向判断对不对，统计判对的比例。只看股票自己的涨跌，没有减去大盘涨跌，"
                 "与上一行候选单的超额口径不同，两个数字不能直接比大小。",
         "读数": timing_self_eval(since, until)},
        {"系统": "PMS 账本",
         "口径": "这一行是 PMS 当天的新开仓审批记录，拆成三份名单：系统自动放行的、"
                 "人工同意的、人工否决的。否决的也照样算收益，用来看当初拒得对不对。"
                 "另外单列一份择时决策系统当天看多的股票。这四份名单的买入时点、"
                 "往后看几天、跟谁比，都和候选单完全一样，可以直接跟候选单那一行对着看。",
         "读数": "；".join(f"{name}—{_list_reading(lists_tbl, name)}"
                           for name in (*LEDGER_LISTS, TIMING_LIST))},
    ]
    return pd.DataFrame(rows)


def ledger_table() -> pd.DataFrame:
    """第八节"台账对表"：台账条目编号、日期、标题，加"一致""不一致"两列空着给人填。"""
    entries = decision_ledger_entries()
    if not entries:
        return pd.DataFrame([{"编号": "—", "日期": "—", "标题": f"台账文件缺失或无条目（{DECISION_LEDGER_PATH}）",
                              "一致": "", "不一致": ""}])
    return pd.DataFrame([{"编号": e["no"], "日期": e["date"], "标题": e["title"], "一致": "", "不一致": ""}
                         for e in entries])


def main() -> int:
    ap = argparse.ArgumentParser(description="候选单复盘（只读，名单级观察收益，不是回测）")
    ap.add_argument("--since", default="2026-07-29")
    ap.add_argument("--until")
    ap.add_argument("--horizons", default="5,10,20")
    ap.add_argument("--start-price", choices=["next_close", "signal_close"], default="next_close",
                    help="next_close=次日收盘起算（默认，实盘口径）；signal_close=信号日收盘起算（对账方案 1.8 系列）")
    ap.add_argument("--out", default="data/review")
    a = ap.parse_args()
    hs = tuple(int(x) for x in a.horizons.split(",") if x.strip())
    run(a.since, a.until, hs, a.start_price, a.out)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())