tradingSystem/app/services/ledger_service.py

# -*- coding: utf-8 -*-
"""
账本服务: 成交回放 · 对账 · 除权 · 日终结算 (设计 §4「生命线」)
================================================================
纯逻辑在 app/core/recon.py, 本模块只负责取数、落库与状态推进。

摊薄成本口径 (与 core/cushion.PositionCost 等价, 但数据来自批次表):
    cum_buy  = Σ (剩余数量 + 已核销数量) × 开仓价
    cum_sell = Σ 已核销数量 × 核销均价
    avg_cost = max(0, (cum_buy − cum_sell) / 当前持股数)
  做T利润通过 T0 批次的买卖流水自然摊入上式, 故 **不再另行扣减** realized_t_profit
  (该列仅作展示统计, 重复扣减会把成本做低两次)。

铁律: 对账以下游为准; 修正一律走 RECON 批次留痕; 连续 N 日不一致升级 ERROR。
"""
from __future__ import annotations

import logging
from datetime import datetime, timedelta

from app.core import cushion as cu
from app.core import recon as rc
from app.core import tradedays as td
from app.repo import downstream_repo, pms_repo, qmt_repo
from app.services import market, param_store, portfolio

logger = logging.getLogger("pms.ledger")

CF_FEE, CF_CALIBRATE = "FEE", "CALIBRATE"   # pms_cash_flow.kind
CURSOR_KEY = "PMS_REPLAY_CURSOR"
CURSOR_ALL = "ALL"          # 游标设成这个值 = 显式要求从头全量回放 (见 _seed_cursor)
STREAK_KEY = "PMS_RECON_STREAK"
LIVE_INSTR = ("DISPATCHED", "JUDGE_PASSED", "RULE_PASSED")


# ================================================================ 回放
def _seed_cursor(out: dict) -> dict:
    """首次回放: 把游标对齐到当前最新成交, **不追认历史**。

    `trading_order` 里躺着旧系统多年的成交记录。PMS 刚上线时一条在途指令都没有, 若从头
    回放, 每一条历史成交都会被判成「外部成交」并入 BASE 批次 —— 账本上凭空长出一堆早就
    清掉的持仓, 摊薄成本与安全垫全错, 日志还刷几百条 EXTERNAL_FILL 告警。而安全垫是补仓、
    盈利加仓、保垫减仓共同的判断依据, 它一错整条纪律链跟着错。**账本必须从干净的起点开始。**

    要补历史有两条路 (页面「参数设置」改 PMS_REPLAY_CURSOR):
      * 设成某个 order_id  → 从它之后开始回放
      * 设成 ALL           → 从头全量回放
    """
    try:
        anchor = downstream_repo.latest_filled_order_id()
    except Exception as e:
        out.update({"ok": False, "errors": [f"读下游最新成交失败: {type(e).__name__}: {e}"]})
        return out
    if not anchor:
        out["note"] = "下游暂无已成交单, 游标待下次再对齐"
        return out
    pms_repo.set_param(CURSOR_KEY, anchor, updated_by="system")
    out.update({"cursor": anchor, "seeded": True})
    out["note"] = (f"首次回放: 游标已对齐到当前最新成交 {anchor}, **不追认历史** —— "
                   f"账本从现在起跟踪。需要补历史请在页面把 PMS_REPLAY_CURSOR 改成某个 "
                   f"order_id (从它之后开始) 或 {CURSOR_ALL} (从头全量)")
    logger.warning("[replay_fills] %s", out["note"])
    return out


def consume_ws_trades(*, limit: int = 500) -> dict:
    """消费 ws 通道落在 pms_qmt_inbox 的逐笔成交 → 批次入账 + 费用流水 (协议 §5.5)。

    与下面的 trading_order 回放是**两条互补的路**, 都挂在 replay_fills 里:
      本函数        ws 通道推来的成交 —— 自带 instruction_id, **精确认领**
      replay_fills  trading_order 增量 —— 只剩外部/人工成交, FIFO 认领退化为兜底

    幂等靠 inbox 的 processed 标记 (0 待入账 → 1 已入账), 而不是靠返回行数 —— 那个在
    CLIENT_FOUND_ROWS 下不可信, 见 qmt_repo.inbox_put 的注释。
    """
    out = {"ok": True, "trades": 0, "actions": 0, "fees": 0, "alerts": [], "errors": []}
    try:
        rows = qmt_repo.inbox_pending(limit=limit)
    except Exception as e:
        # ws 三表没建 (影子期正常) 或库不可用 —— 不该让整个回放任务失败
        out["skipped"] = f"inbox 不可读: {type(e).__name__}"
        return out
    trades = [r for r in rows if r.get("msg_type") == "trade"]
    if not trades:
        return out
    out["trades"] = len(trades)

    # 取每条成交所属父指令的 action, 用来定买入的批次类型
    parents = {rc.parent_instruction_id((r.get("payload") or {}).get("instruction_id") or "")
               for r in trades}
    parent_actions = {}
    for pid in parents:
        if not pid:
            continue
        try:
            ins = pms_repo.get_instruction(pid)
            if ins:
                parent_actions[pid] = ins.get("action")
        except Exception as e:
            out["errors"].append(f"读指令 {pid} 失败: {type(e).__name__}: {e}")

    mapped = rc.map_trades_to_book(trades, parent_actions=parent_actions)
    done = []
    for act in mapped["actions"]:
        try:
            _apply_action(act)
            out["actions"] += 1
        except Exception as e:
            logger.exception("ws 成交入账失败 %s", act)
            out["errors"].append(f"{act.get('ts_code')} 入账失败: {type(e).__name__}: {e}")
    ymd = td.ymd()
    for f in mapped["fees"]:
        try:
            pms_repo.insert_cash_flow(ymd=ymd, kind=CF_FEE, amount=f["amount"],
                                      ts_code=f["ts_code"], estimated=f["estimated"],
                                      trade_no=f["trade_no"],
                                      instruction_id=f["instruction_id"], note=f["note"])
            out["fees"] += 1
        except Exception as e:
            out["errors"].append(f"费用入账失败 {f.get('trade_no')}: {type(e).__name__}: {e}")

    for code in {a["ts_code"] for a in mapped["actions"]}:
        try:
            recompute_position(code)
        except Exception as e:
            out["errors"].append(f"{code} 成本重算失败: {e}")

    # 只有全程无错才标已入账 —— 有错就留在 processed=0, 下一跳重试。
    # 重试是安全的: _apply_action 幂等由 trade_no 兜着 (inbox 那层已按 trade_no 去过重)。
    if not out["errors"]:
        done = [s for s in mapped["seqs"] if s is not None]
        if done:
            try:
                qmt_repo.inbox_mark(done, processed=1, note="ws 成交已入账")
            except Exception as e:
                out["errors"].append(f"inbox 标记失败: {type(e).__name__}: {e}")
    out["alerts"] = mapped["alerts"]
    for a in out["alerts"]:
        logger.warning("[ws 入账告警] %s", a.get("message"))
    out["ok"] = not out["errors"]
    return out


def calibrate_fees(*, ymd=None, actual_fee=None) -> dict:
    """日终用资金快照反推当日真实费用, 写一条 CALIBRATE 平掉估算误差 (协议 §5.5)。

    对端明说逐笔 fee 是按费率估的。协议给的校准式子是
        当日实际费用 = 总资产变动 − 成交净额
    资金快照来自 ws 的 funds_update / snapshot(kind=funds), 落在 pms_qmt_inbox。**影子期
    没有这条数据**, 那就只登记一句"待校准", 不硬凑 —— 估算值本来就只影响现金账, 不影响
    成本与安全垫, 晚校准几天没有任何风险。这也是当初把费用挡在成本之外的意义。

    actual_fee 可显式传入 (人工按对账单校准时用), 传了就不去读快照。
    """
    ymd = int(ymd or td.ymd())
    out = {"ymd": ymd, "estimated": 0.0, "actual": None, "adjusted": 0.0, "note": ""}
    try:
        out["estimated"] = round(pms_repo.sum_cash_flow(ymd, CF_FEE), 2)
    except Exception as e:
        out["note"] = f"读当日费用流水失败: {type(e).__name__}: {e}"
        return out
    if actual_fee is None:
        out["note"] = ("当日估算费用已入现金账; 资金快照未接通 (ws 通道未启用), "
                       "暂不校准 —— 费用不进成本, 晚校准无风险")
        return out
    actual = -abs(float(actual_fee))
    diff = round(actual - out["estimated"], 2)
    out["actual"] = actual
    if abs(diff) < 0.01:
        out["note"] = "估算与实际一致, 无需校准"
        return out
    pms_repo.insert_cash_flow(
        ymd=ymd, kind=CF_CALIBRATE, amount=diff, estimated=0, trade_no=None,
        note=f"日终校准: 估算 {out['estimated']:.2f} → 实际 {actual:.2f}, 差额 {diff:+.2f}")
    out["adjusted"] = diff
    out["note"] = f"已按资金快照校准, 差额 {diff:+.2f} 元 (只调现金账, 不回溯改成本)"
    logger.info("[费用校准] %s", out["note"])
    return out


def replay_fills(*, limit: int = 500) -> dict:
    """成交回放一跳 = ws 逐笔入账 + trading_order 增量回放 (幂等)。

    两条路互补: ws 通道的成交自带 instruction_id 精确入账; trading_order 这条在 ws 接管后
    退化为**只兜外部/人工成交** (你在 QMT 手工下的单、别的系统下的单)。影子期只有后者。
    """
    out = {"ok": True, "fills": 0, "actions": 0, "alerts": [], "errors": [], "cursor": None}
    out["ws"] = consume_ws_trades(limit=limit)
    if out["ws"].get("errors"):
        out["errors"].extend(out["ws"]["errors"])
    cursor = pms_repo.get_param(CURSOR_KEY)
    if not cursor:                       # 从未设过 (或被清空) —— 冷启动, 只对齐游标不入账
        return _seed_cursor(out)
    if str(cursor).strip().upper() == CURSOR_ALL:
        cursor = None                    # 显式要求从头全量回放
    try:
        fills = downstream_repo.fetch_filled_orders(since_id=cursor, limit=limit)
    except Exception as e:
        out.update({"ok": False, "errors": [f"读 trading_order 失败: {type(e).__name__}: {e}"]})
        return out
    out["fills"] = len(fills)
    if not fills:
        out["cursor"] = cursor
        return out

    try:
        instrs = _open_instructions()
    except Exception as e:
        instrs = []
        out["errors"].append(f"读在途指令失败(按外部成交处理): {e}")

    mapped = rc.map_fills_to_book(fills, instrs)
    for act in mapped["actions"]:
        try:
            _apply_action(act)
            out["actions"] += 1
        except Exception as e:
            logger.exception("入账失败 %s", act)
            out["errors"].append(f"{act.get('ts_code')} 入账失败: {type(e).__name__}: {e}")
    out["alerts"].extend(mapped["alerts"])   # 与 ws 那批告警合并, 日报一处看全

    for code in {a["ts_code"] for a in mapped["actions"]}:
        try:
            recompute_position(code)
        except Exception as e:
            out["errors"].append(f"{code} 成本重算失败: {e}")

    try:
        nc = rc.next_cursor(fills, cursor)
        pms_repo.set_param(CURSOR_KEY, nc, "system")
        out["cursor"] = nc
    except Exception as e:
        out["errors"].append(f"游标推进失败: {e}")
    out["ok"] = not out["errors"]
    for a in out["alerts"]:
        logger.warning("[回放告警] %s", a.get("message"))
    return out


def _open_instructions() -> list:
    """在途指令 (回放认领的候选池)。

    dispatched_at 取「下发时点 → 建单时点」, **不用 updated_at**:
    updated_at 每次部分成交回写都会往前跳, 拿它当下发时点会让同一条指令的后续成交
    被时间守卫挡在门外, 误判成外部成交。建单时点是天然的下界, 宁松勿紧。
    """
    rows = pms_repo.list_instructions(statuses=list(LIVE_INSTR), limit=500)
    out = []
    for r in rows:
        prog = r.get("progress") or {}
        out.append({"instruction_id": r["instruction_id"], "ts_code": r["ts_code"],
                    "side": r.get("side"), "qty": int(r.get("qty") or 0),
                    "exec_qty": int(r.get("exec_qty") or 0), "action": r.get("action"),
                    "dispatched_at": str(prog.get("dispatched_at")
                                         or r.get("created_at") or "")})
    return out


def _apply_action(act: dict):
    code, qty, px = act["ts_code"], int(act["qty"]), float(act["price"] or 0)
    if qty <= 0:
        return
    pms_repo.ensure_position(code)
    if act["kind"] == "BUY":
        pms_repo.insert_lot(ts_code=code, lot_type=act.get("lot_type") or "BASE", qty=qty,
                            open_price=px, open_date=datetime.now().date(),
                            instruction_id=act.get("instruction_id"),
                            note="外部成交并入 BASE" if not act.get("instruction_id") else None)
        pms_repo.bump_position_qty(code, total_delta=qty, avail_delta=0)   # T+1: 当日买入不可卖
    else:
        lots = pms_repo.list_lots(code, status="OPEN")
        res = rc.apply_sell_to_lots(
            [{"lot_id": l["id"], "lot_type": l["lot_type"], "qty": int(l["qty"]),
              "open_date": _date_key(l["open_date"])} for l in lots], qty)
        by_id = {l["id"]: l for l in lots}
        t_profit = 0.0
        for a in res["alloc"]:
            lot = by_id[a["lot_id"]]
            pnl = (px - float(lot["open_price"] or 0)) * a["qty"]
            pms_repo.close_lot_qty(a["lot_id"], qty=a["qty"], close_price=px, realized_pnl=pnl)
            if lot["lot_type"] == "T0":
                t_profit += pnl
        pms_repo.bump_position_qty(code, total_delta=-qty, avail_delta=-qty)
        if t_profit:
            pos = pms_repo.get_position(code) or {}
            pms_repo.update_position(
                code, realized_t_profit=float(pos.get("realized_t_profit") or 0) + t_profit)
        for al in res["alerts"]:
            logger.warning("[卖出核销] %s %s", code, al.get("message"))
    if act.get("instruction_id"):
        try:
            pms_repo.add_instruction_exec(act["instruction_id"], qty)
            _settle_instruction(act["instruction_id"])
        except Exception as e:
            logger.warning("指令进度更新失败 %s: %s", act["instruction_id"], e)


def _settle_instruction(instruction_id: str):
    """成交量达到指令数量 → 置 CONFIRMED (部分成交保持在途, 由窗口/过期规则收口)。"""
    ins = pms_repo.get_instruction(instruction_id)
    if ins and int(ins.get("exec_qty") or 0) >= int(ins.get("qty") or 0) > 0:
        pms_repo.update_instruction(instruction_id, status="CONFIRMED")


def _date_key(d):
    try:
        return int(str(d).replace("-", "")[:8])
    except (TypeError, ValueError):
        return 0


def recompute_position(ts_code: str) -> dict:
    """由批次表重算持仓数量/摊薄成本/安全垫/垫子峰值。"""
    lots = pms_repo.list_lots(ts_code, status=None, limit=2000)
    qty = sum(int(l["qty"] or 0) for l in lots)
    cum_buy = sum((int(l["qty"] or 0) + int(l["closed_qty"] or 0)) * float(l["open_price"] or 0)
                  for l in lots)
    cum_sell = sum(int(l["closed_qty"] or 0) * float(l["close_avg_price"] or 0) for l in lots)
    by_type = {}
    for l in lots:
        if int(l["qty"] or 0) > 0:
            by_type[l["lot_type"]] = by_type.get(l["lot_type"], 0) + int(l["qty"])

    avg_cost = max(0.0, (cum_buy - cum_sell) / qty) if qty > 0 else None
    px = market.get_price(ts_code) or avg_cost or 0
    cp = (px / avg_cost - 1.0) if (avg_cost and avg_cost > 0 and px) else None
    solid = param_store.get_float("PMS_CUSHION_SOLID", 0.03)
    pos = pms_repo.get_position(ts_code) or {}
    peak = max(float(pos.get("cushion_peak") or 0), cp or 0)
    scale = param_store.get_float("PMS_TOTAL_SCALE", 0)

    fields = {
        "total_qty": qty, "base_qty": by_type.get("BASE", 0) + by_type.get("RECON", 0),
        "fill_qty": by_type.get("FILL", 0), "add_qty": by_type.get("ADD", 0),
        "dca_qty": by_type.get("DCA", 0), "t0_qty": by_type.get("T0", 0),
        "avg_cost": round(avg_cost, 3) if avg_cost else None,
        "cushion_pct": round(cp, 4) if cp is not None else None,
        "cushion_state": cu.cushion_state(cp, solid), "cushion_peak": round(peak, 4),
        "pct_of_scale": round(qty * px / scale, 4) if scale > 0 else None,
        "status": "CLOSED" if qty <= 0 else (pos.get("status") or "HOLDING"),
    }
    if qty > 0 and (pos.get("status") in (None, "PLANNED", "CLOSED")):
        fields["status"] = "HOLDING"
    if qty > 0 and not pos.get("opened_date"):
        fields["opened_date"] = datetime.now().date()
    pms_repo.update_position(ts_code, **fields)
    return fields


# ================================================================ 对账
def reconcile(*, apply_fix: bool = True) -> dict:
    """账本 vs 下游持仓, 以下游为准修正并留痕。"""
    out = {"ok": True, "diffs": [], "fixes": [], "errors": [], "columns": {}, "severity": rc.SEV_OK}
    try:
        ds = downstream_repo.fetch_positions()
    except Exception as e:
        out.update({"ok": False, "errors": [f"读 trading_position 失败: {type(e).__name__}: {e}"]})
        return out
    out["columns"] = ds["columns"]
    if ds["rows"] and ds["columns"].get("qty") is None:
        out.update({"ok": False, "errors": [
            "下游持仓表未识别出数量列 —— 请按 QMT_INTERFACE_REQUIREMENTS A1/D1 取得 DDL 后, "
            "把列名补进 downstream_repo.QTY_CANDIDATES"]})
        return out

    book = [{"ts_code": r["ts_code"], "total_qty": int(r.get("total_qty") or 0)}
            for r in pms_repo.list_positions()]
    diffs = rc.diff_positions(book, [{"ts_code": r["ts_code"], "qty": r["qty"]}
                                     for r in ds["rows"]])
    out["diffs"] = diffs

    streak = param_store.get_int(STREAK_KEY, 0)
    streak = streak + 1 if diffs else 0
    # 必须走 ParamStore 写入: 直接写库不会失效缓存, 会导致连续天数一直读到旧值
    param_store.set_param(STREAK_KEY, streak, "system")
    out["severity"] = rc.recon_severity(streak if diffs else 0,
                                        param_store.get_int("PMS_RECON_ALARM_DAYS", 3))

    if not diffs or not apply_fix:
        return out

    codes = [d["ts_code"] for d in diffs]
    prices = market.get_prices(codes)
    # 下游的成本价 —— 补仓位时的开仓价优先取它, 现价只兜底 (见 rc.build_recon_fixes 注释:
    # 拿现价当成本会让安全垫齐刷刷归零, 整条纪律链跟着失灵)
    costs = {r["ts_code"]: r.get("cost") for r in (ds.get("rows") or [])}
    lots_map = {c: [{"lot_id": l["id"], "lot_type": l["lot_type"], "qty": int(l["qty"]),
                     "open_date": _date_key(l["open_date"])}
                    for l in pms_repo.list_lots(c, status="OPEN")] for c in codes}
    fixes = rc.build_recon_fixes(diffs, price_map=prices, lots_map=lots_map, cost_map=costs)
    for f in fixes:
        try:
            _apply_fix(f)
            recompute_position(f["ts_code"])
        except Exception as e:
            logger.exception("对账修正失败 %s", f)
            out["errors"].append(f"{f['ts_code']} 修正失败: {type(e).__name__}: {e}")
    out["fixes"] = fixes
    out["ok"] = not out["errors"]
    if out["severity"] == rc.SEV_ERROR:
        logger.error("[对账] 连续 %s 日不一致, 升级 ERROR 待人工: %s 项差异", streak, len(diffs))
    return out


def _apply_fix(f: dict):
    code = f["ts_code"]
    pms_repo.ensure_position(code)
    if f["op"] == "ADD_RECON_LOT":
        px = float(f.get("price") or 0)
        pms_repo.insert_lot(ts_code=code, lot_type="RECON", qty=int(f["qty"]),
                            open_price=px, open_date=datetime.now().date(),
                            note=f["note"] + ("" if px > 0 else " [缺现价, 成本待人工核]"))
    else:
        lots = {l["id"]: l for l in pms_repo.list_lots(code, status="OPEN")}
        px = float(f.get("price") or 0)
        for a in f.get("alloc") or []:
            lot = lots.get(a["lot_id"])
            pnl = (px - float(lot["open_price"] or 0)) * a["qty"] if (lot and px) else 0.0
            pms_repo.close_lot_qty(a["lot_id"], qty=a["qty"], close_price=px or
                                   float(lot["open_price"] or 0), realized_pnl=pnl)
    pms_repo.insert_ledger(ts_code=code, action="RECON", arbiter="rule", verdict="PASS",
                           price_at=float(f.get("price") or 0),
                           hard_numbers={"op": f["op"], "qty": f["qty"]},
                           reason=f["note"])


# ================================================================ 除权
def detect_and_apply_ex_right() -> dict:
    """用昨日结算快照与今日持仓/价格比对, 识别送转股并按比例调整批次。"""
    out = {"checked": 0, "ex_rights": [], "mismatches": [], "errors": []}
    prev = _prev_snapshot()
    if not prev:
        out["errors"].append("无昨日结算快照, 本次跳过除权检测 (次日起生效)")
        return out
    for pos in pms_repo.list_positions(only_open=True):
        code = pos["ts_code"]
        old = prev.get(code)
        if not old:
            continue
        out["checked"] += 1
        px = market.get_price(code)
        r = rc.detect_ex_right(int(old.get("qty") or 0), int(pos.get("total_qty") or 0),
                               float(old.get("price") or 0), float(px or 0))
        if not r:
            continue
        if r["kind"] == "EX_RIGHT":
            try:
                lots = pms_repo.list_lots(code, status="OPEN")
                for l in rc.apply_ex_right(lots, r["ratio"]):
                    pms_repo.update_lot(l["id"], qty=l["qty"], open_price=l["open_price"],
                                        note=l["note"])
                recompute_position(code)
                pms_repo.insert_ledger(ts_code=code, action="RECON", arbiter="rule",
                                       verdict="PASS", price_at=px or 0, hard_numbers=r,
                                       reason=f"除权调整 ×{r['ratio']}")
                out["ex_rights"].append({"ts_code": code, **r})
            except Exception as e:
                out["errors"].append(f"{code} 除权调整失败: {e}")
        else:
            out["mismatches"].append({"ts_code": code, **r})
            logger.error("[除权] %s 比例不吻合, 待人工: %s", code, r.get("reason"))
    return out


def _prev_snapshot() -> dict:
    """取最近一份日终快照 {code: {qty, price}} (存在 pms_daily_report 里, 不新增表)。"""
    r = pms_repo.latest_report()
    if not r:
        return {}
    snap = (r.get("report") or {}).get("snapshot") or {}
    return snap if isinstance(snap, dict) else {}


# ================================================================ 盘前 / 日终
def premarket() -> dict:
    """盘前准备 (08:50): T+1 可卖重置、参考位取数、刹车结算。"""
    out = {"ok": True, "avail_reset": 0, "refs": 0, "brake": None, "errors": []}
    try:
        out["avail_reset"] = pms_repo.reset_avail_all()
    except Exception as e:
        out["errors"].append(f"可卖量重置失败: {e}")
    for pos in pms_repo.list_positions(only_open=True):
        code = pos["ts_code"]
        try:
            refs = market.get_refs(code, base_cost=pos.get("avg_cost"))
            pms_repo.update_position(code, support_ref=refs.get("support"),
                                     pressure_ref=refs.get("pressure"),
                                     stop_ref=refs.get("stop"),
                                     ref_source=refs.get("source"))
            out["refs"] += 1
        except Exception as e:
            out["errors"].append(f"{code} 参考位取数失败: {e}")
    try:
        out["brake"] = _settle_brake()
    except Exception as e:
        out["errors"].append(f"刹车结算失败: {e}")
    out["ok"] = not out["errors"]
    return out


def _settle_brake() -> dict:
    """组合刹车: 自高水位回撤 ≥ 阈值 → 自主增持停 N 个交易日 (命令类不受限)。"""
    v = portfolio.positions_view()
    mv = v["totals"]["portfolio_mv"]
    hw = param_store.get_float("PMS_HIGH_WATER", 0.0)
    dd_limit = param_store.get_float("PMS_BRAKE_DRAWDOWN", 0.05)
    days = param_store.get_int("PMS_BRAKE_DAYS", 3)
    until = param_store.get_int("PMS_BRAKE_UNTIL", 0)
    today = td.ymd()
    if mv > hw:
        param_store.set_param("PMS_HIGH_WATER", mv, "system")
        hw = mv
    drawdown = (1 - mv / hw) if hw > 0 else 0.0
    if hw > 0 and drawdown >= dd_limit and today >= until:
        until = td.ymd(td.next_trade_day(datetime.now().date(), days))
        param_store.set_param("PMS_BRAKE_UNTIL", until, "system")
        logger.warning("[刹车] 自高水位回撤 %.1f%% ≥ %.0f%%, 自主增持暂停至 %s",
                       drawdown * 100, dd_limit * 100, until)
    return {"high_water": hw, "portfolio_mv": mv, "drawdown": round(drawdown, 4),
            "brake_until": until, "active": today < until}


def daily_settle() -> dict:
    """日终结算 (15:10): 除权检测 → 全量对账 → 垫子峰值/连负天数 → 命令进度日结 → 快照留存。"""
    from app.services import command_service
    out = {"ok": True, "steps": {}, "errors": []}
    try:
        out["steps"]["ex_right"] = detect_and_apply_ex_right()
    except Exception as e:
        out["errors"].append(f"除权检测失败: {e}")
    try:
        out["steps"]["recon"] = reconcile()
    except Exception as e:
        out["errors"].append(f"对账失败: {e}")
    try:
        out["steps"]["cushion"] = _settle_cushion()
    except Exception as e:
        out["errors"].append(f"安全垫结算失败: {e}")
    try:
        out["steps"]["fee_calibrate"] = calibrate_fees()
    except Exception as e:
        out["errors"].append(f"费用校准失败: {e}")
    try:
        out["steps"]["commands"] = command_service.refresh_progress()
    except Exception as e:
        out["errors"].append(f"命令进度结算失败: {e}")
    try:
        pms_repo.expire_proposals()
    except Exception as e:
        out["errors"].append(f"提议过期处理失败: {e}")
    out["ok"] = not out["errors"]
    return out


def _settle_cushion() -> dict:
    """更新垫子峰值与「安全垫连续为负天数」(清弱票判定所需)。"""
    v = portfolio.positions_view()
    streak = portfolio.neg_streak_map()
    updated = 0
    for x in v["held"]:
        code, cp = x["ts_code"], x["cushion_pct"]
        streak[code] = (int(streak.get(code, 0)) + 1) if (cp is not None and cp < 0) else 0
        peak = max(float(x["cushion_peak"] or 0), cp or 0)
        pms_repo.update_position(code, cushion_pct=cp, cushion_peak=round(peak, 4),
                                 cushion_state=x["cushion_state"])
        updated += 1
    held = {x["ts_code"] for x in v["held"]}
    portfolio.save_neg_streak({k: v2 for k, v2 in streak.items() if k in held})
    return {"updated": updated,
            "neg_streak": {k: v2 for k, v2 in streak.items() if v2 > 0 and k in held}}


# ================================================================ 日报
def build_daily_report(ymd: int = None) -> dict:
    """运营日报 (15:30): 关注区 + 全量统计 + 当日快照 (快照供次日除权检测)。"""
    from app.services import command_service
    ymd = int(ymd or td.ymd())
    v = portfolio.positions_view()
    t = v["totals"]
    try:
        recon_state = {"streak": param_store.get_int(STREAK_KEY, 0)}
    except Exception:
        recon_state = {}

    cmds = pms_repo.list_commands(statuses=["EXECUTING", "PARTIAL", "PENDING", "PLANNING"],
                                  limit=100)
    proposals = pms_repo.list_proposals(statuses=("WAIT_USER",), limit=100)
    live_ins = pms_repo.list_instructions(statuses=list(LIVE_INSTR), limit=200)
    attention = []
    for c in cmds:
        prog = c.get("progress") or {}
        attention.append({"type": "命令进度", "command_id": c["command_id"],
                          "cmd_type": c["cmd_type"], "status": c["status"],
                          "done": prog.get("done_amount"), "target": prog.get("target_amount"),
                          "deadline": prog.get("deadline")})
    if proposals:
        attention.append({"type": "待确认提议", "count": len(proposals)})
    if recon_state.get("streak"):
        attention.append({"type": "对账差异", "streak": recon_state["streak"],
                          "severity": rc.recon_severity(
                              recon_state["streak"],
                              param_store.get_int("PMS_RECON_ALARM_DAYS", 3))})
    brake_until = param_store.get_int("PMS_BRAKE_UNTIL", 0)
    if brake_until > ymd:
        attention.append({"type": "组合刹车", "until": brake_until})
    if v["price_missing"]:
        attention.append({"type": "行情缺失", "codes": v["price_missing"]})
    if not v["sector_ready"]:
        attention.append({"type": "行业约束停用", "hint": "PMS_SECTOR_SOURCE 未配置"})
    if td.calendar_degraded():
        attention.append({"type": "交易日历降级", "hint": "未安装 chinesecalendar, 节假日不可辨"})

    report = {
        "ymd": ymd, "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "totals": t, "attention": attention,
        "commands": [{"command_id": c["command_id"], "cmd_type": c["cmd_type"],
                      "status": c["status"], "progress": c.get("progress")} for c in cmds],
        "proposals": len(proposals), "live_instructions": len(live_ins),
        "positions": [{"ts_code": x["ts_code"], "qty": x["total_qty"], "price": x["price"],
                       "avg_cost": x["avg_cost"], "cushion_pct": x["cushion_pct"],
                       "cushion_state": x["cushion_state"], "mv": x["market_value"],
                       "pct_of_scale": x["pct_of_scale"]} for x in v["held"]],
        # 次日除权检测用的快照 (数量 + 收盘价)
        "snapshot": {x["ts_code"]: {"qty": x["total_qty"], "price": x["price"]}
                     for x in v["held"]},
        "recon": recon_state,
    }
    try:
        pms_repo.upsert_report(ymd, report)
    except Exception as e:
        logger.error("日报落表失败: %s", e)
        report["save_error"] = str(e)
    return report


def expire_stale_instructions() -> int:
    """下发后长时间未被接受的指令置过期 (不自动重发 —— 设计 §13)。"""
    mins = param_store.get_int("PMS_DISPATCH_EXPIRE_MIN", 30)
    cut = datetime.now() - timedelta(minutes=mins)
    n = 0
    for r in pms_repo.list_instructions(statuses=["DISPATCHED"], limit=500):
        try:
            if str(r.get("updated_at") or "") and str(r["updated_at"]) < str(cut):
                pms_repo.update_instruction(r["instruction_id"], status="EXPIRED")
                n += 1
        except Exception:
            continue
    return n