tradingSystem/app/core/planner.py

# -*- coding: utf-8 -*-
"""
方案生成器 (纯逻辑, 无外部依赖, 可单测)
========================================
把一条任务命令展开成分股行动清单 (设计 POSITION_MGMT_DESIGN.md §3.2 / §5 / §6)。
只算数, 不落库、不下发 —— 调用方 (services/command_service.py) 负责写 pms_plan。

核心是「降仓凑额」四档优先级 (设计 §3.2):
    ① 撤销在途买入类指令 (先停止继续投入, 不计入释放额)
    ② 清弱票  —— 浮亏且安全垫为负、持续 ≥N 日 → 整票清仓
    ③ 收利润  —— 浮盈票卖出加仓批/补足批 (保留底仓)
    ④ 等比微减 —— 仍不足则各票按市值等比例微减 (此时才动底仓)

约定与口径:
  * 金额单位元, 数量单位股; A股一手 = 100 股。
  * 部分减持数量一律向下取整到一手; **整票清仓允许卖零股** (A股零股可一次性卖出),
    所以 EXIT 用 total_qty 原值, 不做取整。
  * 冻结 (frozen_reason != NONE) 只禁增持不禁减持 —— 降仓类方案照常纳入该票。
  * 同一票在同一份方案中只出一条减持动作 (跨档累计用 _sold 台账防超卖)。
  * 卖出数量受 T+1 可卖量限制的部分不在此处扣减: 方案给的是**目标数量**,
    分日出手由择时执行器按 avail_qty 处理 (设计 §8)。方案层只保证不超过 total_qty。
"""
from __future__ import annotations

from app.core.sizer import LOT, lot_qty, split_batches, check_caps
from app.core.cushion import sell_allocation  # noqa: F401  (供调用方做核销预览)

# 方案动作词表 (与 ddl_pms_v1.sql pms_plan.action 注释一致)
A_OPEN, A_FILL, A_ADD, A_DCA = "OPEN", "FILL", "ADD", "DCA"
A_TRIM, A_EXIT, A_HALT, A_T0 = "TRIM", "EXIT", "HALT", "T0_ROUND"

SIDE_BUY, SIDE_SELL, SIDE_NONE = "buy", "sell", "none"

# 优先级 (越小越先执行; 设计 §3.2 的四档次序)
P_HALT, P_WEAK, P_HARVEST, P_PRORATA = 10, 20, 30, 40
P_OPEN_BASE, P_OPEN_FILL, P_OPEN_ADD = 10, 20, 30


# ---------------------------------------------------------------- 小工具
def floor_lot(qty) -> int:
    return int(max(0, int(qty)) // LOT) * LOT


def ceil_lot(qty) -> int:
    q = max(0, int(qty))
    return ((q + LOT - 1) // LOT) * LOT


def mv_of(p: dict) -> float:
    """持仓市值 = 数量 × 现价 (缺价按 0, 由调用方剔除或告警)。"""
    return float(p.get("total_qty") or 0) * float(p.get("price") or 0)


def _item(ts_code, action, side, qty, amount, priority, reason, tier="", **kw):
    d = {"ts_code": ts_code, "action": action, "side": side, "qty": int(qty),
         "amount": round(float(amount), 2), "priority": priority, "reason": reason,
         "tier": tier}
    d.update(kw)
    return d


def _sortable(p, key, reverse=False):
    """稳定排序键: 主键 + ts_code (保证同值时结果确定, 便于单测与复现)。"""
    v = p.get(key)
    v = 0.0 if v is None else float(v)
    return (-v if reverse else v, p.get("ts_code") or "")


def _usable(positions, exclude_codes):
    """可纳入减持方案的持仓: 有数量、有现价、不在排除名单 (已有在途方案的票)。"""
    ex = set(exclude_codes or ())
    out = []
    for p in positions or []:
        if p.get("ts_code") in ex:
            continue
        if int(p.get("total_qty") or 0) <= 0 or float(p.get("price") or 0) <= 0:
            continue
        out.append(p)
    return out


# ================================================================
# 一、降仓凑额 (设计 §3.2 全流程示例的第 1 步)
# ================================================================
def plan_reduce_exposure(*, release_amount: float, positions: list, pending_buys=None,
                         params=None, exclude_codes=()) -> dict:
    """降仓 X% 的方案生成。

    release_amount  需释放金额 (元) = 规模 × pct
    positions       持仓快照列表, 每项至少含:
                    ts_code / price / total_qty / base_qty / cushion_pct /
                    neg_cushion_days (安全垫连续为负天数)
    pending_buys    在途买入指令 [{instruction_id, ts_code, qty, amount}], 全部撤销
    params          {"weak_neg_days":5}
    返回 {"ok","target_amount","planned_amount","gap","items","notes"}
    """
    params = params or {}
    weak_days = int(params.get("weak_neg_days", 5))
    target = float(release_amount or 0)
    items, notes = [], []

    # ---- ① 撤销在途买入 (不计入释放额, 但必须先做: 停止继续投入) ----
    for b in (pending_buys or []):
        items.append(_item(b.get("ts_code"), A_HALT, SIDE_NONE, b.get("qty") or 0,
                           b.get("amount") or 0, P_HALT,
                           "降仓命令: 撤销在途买入指令", tier="1_停新买",
                           cancel_instruction_id=b.get("instruction_id")))
    if pending_buys:
        notes.append(f"撤销在途买入指令 {len(pending_buys)} 条 (不计入释放额)")

    if target <= 0:
        return {"ok": False, "target_amount": 0.0, "planned_amount": 0.0, "gap": 0.0,
                "items": items, "notes": notes + ["释放金额为 0, 无减持动作"]}

    pool = _usable(positions, exclude_codes)
    if exclude_codes:
        notes.append(f"跳过已有在途方案的票: {sorted(set(exclude_codes))}")
    sold = {}   # ts_code -> 已计划卖出股数 (跨档防超卖)
    acc = 0.0

    def remain_qty(p):
        return int(p.get("total_qty") or 0) - sold.get(p["ts_code"], 0)

    # ---- ② 清弱票: 浮亏且安全垫为负、持续 ≥N 日 → 整票清仓 (最差的先清) ----
    weak = [p for p in pool
            if p.get("cushion_pct") is not None and float(p["cushion_pct"]) < 0
            and int(p.get("neg_cushion_days") or 0) >= weak_days]
    for p in sorted(weak, key=lambda x: _sortable(x, "cushion_pct")):
        if acc >= target:
            break
        qty = remain_qty(p)                    # 清仓卖全量, 零股一并卖出 (不取整)
        if qty <= 0:
            continue
        amt = qty * float(p["price"])
        sold[p["ts_code"]] = sold.get(p["ts_code"], 0) + qty
        acc += amt
        items.append(_item(p["ts_code"], A_EXIT, SIDE_SELL, qty, amt, P_WEAK,
                           f"清弱票: 安全垫 {float(p['cushion_pct']):.1%} 连续为负 "
                           f"{int(p.get('neg_cushion_days') or 0)} 日", tier="2_清弱票"))
    if acc > target and weak:
        notes.append(f"清弱票为整票清仓, 释放额 {acc:,.0f} 元已超目标 {target:,.0f} 元 "
                     f"(超出 {acc - target:,.0f} 元, 弱票不做拆卖)")

    # ---- ③ 收利润: 浮盈票卖加仓批+补足批, 保留底仓 (垫子厚的先收) ----
    if acc < target:
        profit = [p for p in pool
                  if p.get("cushion_pct") is not None and float(p["cushion_pct"]) > 0
                  and remain_qty(p) > 0]
        for p in sorted(profit, key=lambda x: _sortable(x, "cushion_pct", reverse=True)):
            if acc >= target:
                break
            keep_base = int(p.get("base_qty") or 0)
            sellable = floor_lot(min(remain_qty(p), max(0, int(p.get("total_qty") or 0) - keep_base)))
            if sellable <= 0:
                continue
            need = target - acc
            want = min(sellable, ceil_lot(need / float(p["price"])))
            if want <= 0:
                continue
            amt = want * float(p["price"])
            sold[p["ts_code"]] = sold.get(p["ts_code"], 0) + want
            acc += amt
            items.append(_item(p["ts_code"], A_TRIM, SIDE_SELL, want, amt, P_HARVEST,
                               f"收利润: 安全垫 {float(p['cushion_pct']):.1%}, "
                               f"卖出加仓/补足批 {want} 股, 保留底仓 {keep_base} 股",
                               tier="3_收利润"))

    # ---- ④ 等比微减: 仍不足 → 各票按市值等比例微减 (此时可动底仓) ----
    if acc < target:
        gap = target - acc
        cand = [p for p in pool if remain_qty(p) >= LOT]
        total_mv = sum(remain_qty(p) * float(p["price"]) for p in cand)
        if total_mv <= 0:
            notes.append("等比微减: 无可减持仓")
        else:
            plan_qty = {}
            for p in cand:
                w = remain_qty(p) * float(p["price"]) / total_mv
                q = floor_lot(min(remain_qty(p), lot_qty(gap * w, float(p["price"]))))
                if q > 0:
                    plan_qty[p["ts_code"]] = q
            # 取整造成的缺口: 按剩余市值从大到小逐手补齐
            planned = sum(q * float(next(x for x in cand if x["ts_code"] == c)["price"])
                          for c, q in plan_qty.items())
            order = sorted(cand, key=lambda x: (-(remain_qty(x) * float(x["price"])),
                                                x["ts_code"]))
            guard = 0
            while planned < gap - 1e-6 and guard < 10000:
                guard += 1
                progressed = False
                for p in order:
                    cur = plan_qty.get(p["ts_code"], 0)
                    if cur + LOT <= remain_qty(p):
                        plan_qty[p["ts_code"]] = cur + LOT
                        planned += LOT * float(p["price"])
                        progressed = True
                        if planned >= gap - 1e-6:
                            break
                if not progressed:
                    break
            for p in order:
                q = plan_qty.get(p["ts_code"], 0)
                if q <= 0:
                    continue
                amt = q * float(p["price"])
                sold[p["ts_code"]] = sold.get(p["ts_code"], 0) + q
                acc += amt
                items.append(_item(p["ts_code"], A_TRIM, SIDE_SELL, q, amt, P_PRORATA,
                                   f"等比微减: 按市值权重摊派 {q} 股", tier="4_等比微减"))

    gap = max(0.0, target - acc)
    if gap > 0:
        notes.append(f"可减持仓不足, 缺口 {gap:,.0f} 元 —— 命令将置部分完成并告警")
    return {"ok": gap <= 0, "target_amount": round(target, 2), "planned_amount": round(acc, 2),
            "gap": round(gap, 2), "items": items, "notes": notes}


# ================================================================
# 二、升仓 (设计 §3.1 B: 既有持仓补到目标 + 候选池新票建仓)
# ================================================================
def plan_increase_exposure(*, add_amount: float, positions: list, candidates=None,
                           ctx: dict = None, params=None) -> dict:
    """升仓 X%。既有持仓中「垫厚(SOLID)且未达目标」的票先补足, 再从候选池建新仓。

    candidates: [{ts_code, price, score, sector}] —— 上游计划池 ∪ 白名单, 由调用方备好并
                已剔除黑名单/已持有/冻结票。每只都要过组合约束 (累计口径)。
    ctx:        check_caps 所需上下文 (见 sizer.check_caps), 规划过程中滚动更新。
    """
    params = params or {}
    ctx = dict(ctx or {})
    target = float(add_amount or 0)
    items, notes, rejects = [], [], []
    acc = 0.0
    scale = float(ctx.get("scale") or 0)
    default_target = float(params.get("stock_target_default", 0.06))
    batch_split = params.get("batch_split") or (0.5, 0.25, 0.25)
    merge = bool(params.get("min_lot_merge", True))

    if target <= 0 or scale <= 0:
        return {"ok": False, "target_amount": target, "planned_amount": 0.0,
                "gap": target, "items": [], "notes": ["升仓金额或总规模为 0"], "rejects": []}

    # ---- ① 既有持仓补到目标 (只补垫厚票, 呼应设计"垫厚且决策系统看多") ----
    for p in sorted(positions or [], key=lambda x: _sortable(x, "cushion_pct", reverse=True)):
        if acc >= target:
            break
        if float(p.get("price") or 0) <= 0:
            continue
        if (p.get("cushion_state") or "") != "SOLID":
            continue
        if (p.get("frozen_reason") or "NONE") != "NONE":
            continue
        tgt_pct = float(p.get("target_pct") or default_target)
        room = tgt_pct * scale - mv_of(p)
        if room <= 0:
            continue
        want_amt = min(room, target - acc)
        q = lot_qty(want_amt, float(p["price"]))
        if q < LOT:
            continue
        amt = q * float(p["price"])
        bad = check_all_caps(ts_code=p["ts_code"], add_amount=amt, ctx=_stock_ctx(ctx, p, False))
        if bad:
            rejects.append({"ts_code": p["ts_code"], "reasons": bad})
            continue
        items.append(_item(p["ts_code"], A_ADD, SIDE_BUY, q, amt, P_OPEN_BASE,
                           f"升仓: 垫厚({float(p.get('cushion_pct') or 0):.1%})补至目标 "
                           f"{tgt_pct:.1%}", tier="1_补既有"))
        acc += amt
        ctx = _ctx_after(ctx, amt, is_new_name=False, sector=p.get("sector"))

    # ---- ② 候选池新票建仓 ----
    for c in sorted(candidates or [], key=lambda x: _sortable(x, "score", reverse=True)):
        if acc >= target:
            break
        price = float(c.get("price") or 0)
        if price <= 0:
            continue
        want_amt = min(default_target * scale, target - acc)
        sp = split_batches(want_amt, price, splits=batch_split, merge=merge)
        if not sp["ok"]:
            rejects.append({"ts_code": c.get("ts_code"), "reasons": [sp["reason"]]})
            continue
        base = sp["batches"][0]
        bad = check_all_caps(ts_code=c.get("ts_code"), add_amount=want_amt,
                             ctx=_new_name_ctx(ctx, c))
        if bad:
            rejects.append({"ts_code": c.get("ts_code"), "reasons": bad})
            continue
        items.extend(_open_batch_items(c.get("ts_code"), sp, price,
                                       reason_prefix="升仓建新仓"))
        acc += want_amt
        ctx = _ctx_after(ctx, want_amt, is_new_name=True, sector=c.get("sector"))
        notes.append(f"{c.get('ts_code')}: 目标 {want_amt:,.0f} 元, 首批 {base['qty']} 股")

    gap = max(0.0, target - acc)
    if gap > 0:
        notes.append(f"候选与补仓空间不足, 缺口 {gap:,.0f} 元")
    return {"ok": gap <= 0, "target_amount": round(target, 2), "planned_amount": round(acc, 2),
            "gap": round(gap, 2), "items": items, "notes": notes, "rejects": rejects}


# ================================================================
# 三、个股级方案
# ================================================================
def plan_open_target(*, ts_code: str, target_pct: float, price: float, ctx: dict,
                     params=None) -> dict:
    """建仓某股至目标% —— 50/25/25 分批 (含一手合并), 组合约束硬校验。

    BASE 批立即可执行; FILL/ADD 批以 gated=True 落方案, 由动作引擎按条件解锁 (设计 §6)。
    """
    params = params or {}
    batch_split = params.get("batch_split") or (0.5, 0.25, 0.25)
    merge = bool(params.get("min_lot_merge", True))
    scale = float(ctx.get("scale") or 0)
    stock_mv = float(ctx.get("stock_mv") or 0)
    add_amount = max(0.0, float(target_pct) * scale - stock_mv)

    if scale <= 0:
        return {"ok": False, "items": [], "notes": [], "rejects": [
            {"ts_code": ts_code, "reasons": ["SCALE_INVALID: 总规模未设置"]}]}
    if add_amount <= 0:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"{ts_code} 现有仓位已达/超过目标 {target_pct:.1%}, 无需建仓"],
                "rejects": []}

    bad = check_all_caps(ts_code=ts_code, add_amount=add_amount, ctx=ctx)
    if bad:
        return {"ok": False, "items": [], "target_amount": round(add_amount, 2),
                "planned_amount": 0.0, "gap": round(add_amount, 2), "notes": [],
                "rejects": [{"ts_code": ts_code, "reasons": bad}]}

    sp = split_batches(add_amount, float(price), splits=batch_split, merge=merge)
    if not sp["ok"]:
        return {"ok": False, "items": [], "target_amount": round(add_amount, 2),
                "planned_amount": 0.0, "gap": round(add_amount, 2), "notes": [],
                "rejects": [{"ts_code": ts_code, "reasons": [sp["reason"]]}]}

    items = _open_batch_items(ts_code, sp, float(price), reason_prefix="建仓命令")
    planned = sum(i["amount"] for i in items)
    notes = []
    if sp["scheme"] != tuple(batch_split):
        notes.append(f"一手检查: 批次自动合并为 {sp['scheme']} (原 {tuple(batch_split)})")
    return {"ok": True, "items": items, "target_amount": round(add_amount, 2),
            "planned_amount": round(planned, 2), "gap": 0.0, "notes": notes, "rejects": []}


def plan_exit_stock(*, ts_code: str, position: dict, reason: str = "清仓命令") -> dict:
    """清仓某股 (整票, 零股一并卖出)。"""
    qty = int((position or {}).get("total_qty") or 0)
    price = float((position or {}).get("price") or 0)
    if qty <= 0:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"{ts_code} 无持仓, 命令直接完成"], "rejects": []}
    amt = qty * price
    return {"ok": True, "target_amount": round(amt, 2), "planned_amount": round(amt, 2),
            "gap": 0.0, "notes": [], "rejects": [],
            "items": [_item(ts_code, A_EXIT, SIDE_SELL, qty, amt, P_WEAK, reason,
                            tier="个股清仓")]}


def plan_reduce_stock(*, ts_code: str, target_pct: float, position: dict, scale: float) -> dict:
    """减至 X%: 卖出超出目标仓位的部分 (向下取整到一手; 目标 0 等价清仓)。"""
    qty_hold = int((position or {}).get("total_qty") or 0)
    price = float((position or {}).get("price") or 0)
    if qty_hold <= 0 or price <= 0:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"{ts_code} 无持仓或无现价"], "rejects": []}
    if float(target_pct) <= 0:
        return plan_exit_stock(ts_code=ts_code, position=position, reason="减至 0% (等价清仓)")

    over = qty_hold * price - float(target_pct) * float(scale)
    if over <= 0:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"{ts_code} 当前仓位未超过目标 {target_pct:.1%}"],
                "rejects": []}
    qty = min(qty_hold, floor_lot(lot_qty(over, price)))
    if qty <= 0:
        return {"ok": False, "items": [], "target_amount": round(over, 2), "planned_amount": 0.0,
                "gap": round(over, 2), "rejects": [],
                "notes": [f"{ts_code} 超出额 {over:,.0f} 元不足一手, 不减持"]}
    amt = qty * price
    return {"ok": True, "target_amount": round(over, 2), "planned_amount": round(amt, 2),
            "gap": round(max(0.0, over - amt), 2), "notes": [], "rejects": [],
            "items": [_item(ts_code, A_TRIM, SIDE_SELL, qty, amt, P_HARVEST,
                            f"减至 {target_pct:.1%}: 卖出超出部分 {qty} 股", tier="个股减仓")]}


def plan_liquidate_all(*, positions: list, pending_buys=None) -> dict:
    """一键清仓 (紧急): 撤在途买入 + 全部持仓清仓, 市值大的先卖。"""
    items = []
    for b in (pending_buys or []):
        items.append(_item(b.get("ts_code"), A_HALT, SIDE_NONE, b.get("qty") or 0,
                           b.get("amount") or 0, P_HALT, "一键清仓: 撤销在途买入",
                           tier="1_停新买", cancel_instruction_id=b.get("instruction_id")))
    acc = 0.0
    for p in sorted(_usable(positions, ()), key=lambda x: (-mv_of(x), x["ts_code"])):
        qty = int(p.get("total_qty") or 0)
        amt = qty * float(p["price"])
        acc += amt
        items.append(_item(p["ts_code"], A_EXIT, SIDE_SELL, qty, amt, P_WEAK,
                           "一键清仓 (紧急, 不做择时优化)", tier="全部清仓"))
    return {"ok": True, "target_amount": round(acc, 2), "planned_amount": round(acc, 2),
            "gap": 0.0, "items": items, "rejects": [],
            "notes": [f"全部 {len([i for i in items if i['action'] == A_EXIT])} 只持仓清仓"]}


def plan_sector_exit(*, sector: str, positions: list) -> dict:
    """清仓某行业 (行业名由 IndustryClassifier 提供, positions 需带 sector 字段)。"""
    hit = [p for p in _usable(positions, ()) if (p.get("sector") or "") == sector]
    if not hit:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"行业[{sector}]当前无持仓"], "rejects": []}
    items, acc = [], 0.0
    for p in sorted(hit, key=lambda x: (-mv_of(x), x["ts_code"])):
        qty = int(p["total_qty"])
        amt = qty * float(p["price"])
        acc += amt
        items.append(_item(p["ts_code"], A_EXIT, SIDE_SELL, qty, amt, P_WEAK,
                           f"清仓行业[{sector}]", tier="行业清仓"))
    return {"ok": True, "target_amount": round(acc, 2), "planned_amount": round(acc, 2),
            "gap": 0.0, "items": items, "notes": [], "rejects": []}


def plan_sector_cap(*, sector: str, cap: float, positions: list) -> dict:
    """限制某行业上限: 超出部分在该行业内按市值等比例减持 (不足一手的票跳过)。"""
    pool = _usable(positions, ())
    port_mv = sum(mv_of(p) for p in pool)
    hit = [p for p in pool if (p.get("sector") or "") == sector]
    sec_mv = sum(mv_of(p) for p in hit)
    if port_mv <= 0 or sec_mv <= 0:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"行业[{sector}]无持仓, 仅写入上限参数"], "rejects": []}
    over = sec_mv - float(cap) * port_mv
    if over <= 0:
        return {"ok": True, "items": [], "target_amount": 0.0, "planned_amount": 0.0,
                "gap": 0.0, "notes": [f"行业[{sector}]占比 {sec_mv / port_mv:.1%} 未超上限 "
                                      f"{float(cap):.0%}, 仅写入参数"], "rejects": []}
    items, acc = [], 0.0
    for p in sorted(hit, key=lambda x: (-mv_of(x), x["ts_code"])):
        w = mv_of(p) / sec_mv
        q = min(int(p["total_qty"]), floor_lot(lot_qty(over * w, float(p["price"]))))
        if q <= 0:
            continue
        amt = q * float(p["price"])
        acc += amt
        items.append(_item(p["ts_code"], A_TRIM, SIDE_SELL, q, amt, P_PRORATA,
                           f"行业[{sector}]超限, 等比减持 {q} 股", tier="行业限额"))
    return {"ok": acc > 0, "target_amount": round(over, 2), "planned_amount": round(acc, 2),
            "gap": round(max(0.0, over - acc), 2), "items": items, "rejects": [],
            "notes": [f"行业[{sector}]占比 {sec_mv / port_mv:.1%} → 需减 {over:,.0f} 元"]}


def plan_halt_buy(*, pending_buys: list) -> dict:
    """全局暂停买入: 撤销全部在途买入指令 (卖出与止损不受影响)。"""
    items = [_item(b.get("ts_code"), A_HALT, SIDE_NONE, b.get("qty") or 0,
                   b.get("amount") or 0, P_HALT, "全局暂停买入: 撤销在途买入指令",
                   tier="停新买", cancel_instruction_id=b.get("instruction_id"))
             for b in (pending_buys or [])]
    return {"ok": True, "items": items, "target_amount": 0.0, "planned_amount": 0.0,
            "gap": 0.0, "rejects": [],
            "notes": [f"撤销在途买入指令 {len(items)} 条"] if items else ["无在途买入指令"]}


def plan_halt_all(*, pending_instructions: list) -> dict:
    """全局暂停执行 (休假模式): 撤销全部在途指令, 仅保留对账与日报。"""
    items = [_item(b.get("ts_code"), A_HALT, SIDE_NONE, b.get("qty") or 0,
                   b.get("amount") or 0, P_HALT,
                   f"休假模式: 撤销在途{'买入' if b.get('side') == SIDE_BUY else '卖出'}指令",
                   tier="停执行", cancel_instruction_id=b.get("instruction_id"))
             for b in (pending_instructions or [])]
    return {"ok": True, "items": items, "target_amount": 0.0, "planned_amount": 0.0,
            "gap": 0.0, "rejects": [],
            "notes": [f"撤销在途指令 {len(items)} 条"] if items else ["无在途指令"]}


# ================================================================
# 四、约束校验 (规则闸与方案生成器共用同一份口径)
# ================================================================
def check_all_caps(*, ts_code: str, add_amount: float, ctx: dict) -> list:
    """组合约束 (sizer.check_caps) + 预留现金约束。返回未通过项列表, 空 = 全过。

    预留现金 (设计 §3.1 A): 总规模中永不动用的部分, 与总仓上限双重约束 ——
    加仓后组合市值不得超过 规模 × (1 − 预留比例)。
    """
    v = list(check_caps(ts_code=ts_code, add_amount=add_amount, ctx=ctx))
    reserve = float(ctx.get("cash_reserve") or 0)
    scale = float(ctx.get("scale") or 0)
    if reserve > 0 and scale > 0:
        usable = scale * (1 - reserve)
        after = float(ctx.get("portfolio_mv") or 0) + float(add_amount)
        if after > usable + 1e-9:
            v.append(f"CASH_RESERVE: 加后市值 {after:,.0f} > 可用额度 {usable:,.0f} "
                     f"(预留现金 {reserve:.0%})")
    return v


def _stock_ctx(ctx: dict, p: dict, is_new: bool) -> dict:
    d = dict(ctx)
    sector = p.get("sector") if ctx.get("sector_source_ready", True) else None
    d["stock_mv"] = mv_of(p)
    d["is_new_name"] = is_new
    d["sector"] = sector
    d["sector_names"] = int((ctx.get("sector_names_map") or {}).get(sector, 0))
    d["sector_mv"] = float((ctx.get("sector_mv_map") or {}).get(sector, 0.0))
    return d


def _new_name_ctx(ctx: dict, c: dict) -> dict:
    d = dict(ctx)
    d["stock_mv"] = 0.0
    d["is_new_name"] = True
    d["sector"] = c.get("sector") if ctx.get("sector_source_ready", True) else None
    d["sector_names"] = int((ctx.get("sector_names_map") or {}).get(c.get("sector"), 0))
    d["sector_mv"] = float((ctx.get("sector_mv_map") or {}).get(c.get("sector"), 0.0))
    return d


def _ctx_after(ctx: dict, add_amount: float, is_new_name: bool, sector=None) -> dict:
    """规划过程中滚动更新组合快照, 保证多笔累计口径下的上限校验正确。"""
    d = dict(ctx)
    d["portfolio_mv"] = float(ctx.get("portfolio_mv") or 0) + float(add_amount)
    if is_new_name:
        d["names_count"] = int(ctx.get("names_count") or 0) + 1
    if sector:
        nm = dict(ctx.get("sector_names_map") or {})
        mm = dict(ctx.get("sector_mv_map") or {})
        if is_new_name:
            nm[sector] = int(nm.get(sector, 0)) + 1
        mm[sector] = float(mm.get(sector, 0.0)) + float(add_amount)
        d["sector_names_map"], d["sector_mv_map"] = nm, mm
    return d


def _open_batch_items(ts_code, sp, price, reason_prefix="建仓"):
    """把 split_batches 的结果转成方案条目: BASE 立即执行, FILL/ADD 挂 gated 待引擎解锁。"""
    names = [b["name"] for b in sp["batches"]]
    prio = {"BASE": P_OPEN_BASE, "FILL": P_OPEN_FILL, "ADD": P_OPEN_ADD}
    act = {"BASE": A_OPEN, "FILL": A_FILL, "ADD": A_ADD}
    gate_reason = {
        "BASE": "首批底仓, 规则闸通过即入择时队列",
        "FILL": "回踩补足批: 建仓期内回踩支撑不破且浮亏 <3% 时解锁",
        "ADD": "盈利加仓批: 安全垫 ≥3% 且创 5 日新高/站上压力位时解锁",
    }
    out = []
    for i, b in enumerate(sp["batches"]):
        nm = b["name"] if b["name"] in act else names[i]
        out.append(_item(ts_code, act.get(nm, A_OPEN), SIDE_BUY, b["qty"], b["qty"] * price,
                         prio.get(nm, P_OPEN_BASE),
                         f"{reason_prefix}: {nm} 批 {b['qty']} 股 —— {gate_reason.get(nm, '')}",
                         tier=f"批次_{nm}", gated=(nm != "BASE")))
    return out