tradingSystem/app/core/action_engine.py

# -*- coding: utf-8 -*-
"""
动作引擎 · 自主提议扫描 (纯逻辑, 无外部依赖, 可单测)
=====================================================
设计 POSITION_MGMT_DESIGN.md §6 的四类自主动作。每条规则的触发口径与约束都照抄设计表:

| 引擎 | 触发 | 关键约束 |
|---|---|---|
| FILL 回踩补足 | 建仓期(≤10 交易日)内回踩支撑带不破, 浮亏 < 3% | 每票 1 次; 补后 ≤ 目标仓位 |
| ADD 盈利加仓 | 安全垫 ≥ +3% 且创 5 日新高或站上压力位 | 距上次 ≥2 交易日; ≤ 单股上限; 距 MA5 <+6% |
| DCA 补仓 | 浮亏触及 −8%/−15% 评估档 (各评估一次, 执行终身一次) | ≤ 底仓 50%; −15% 及更深永远需用户确认 |
| TRIM 保垫减仓 | 安全垫峰值 ≥6% 且回吐过半 → 减 1/3 锁盈 | 纯规则自动执行 (减持方向不设确认门槛) |
| EXIT 目标价到价 | 现价 ≥ 用户「设定某股目标价」命令里的价 | 全部持仓; **必定交人拍板, 绝不自动卖** |
| OPEN 新建仓 | 上游候选池里的新票, 且还有持仓名额与可投金额 | 名额与金额边走边扣; 只提底仓批 |

本模块只回答「该不该动、动多少、为什么」, 不查库不下发:
  * 交易日相关的输入 (建仓天数、距上次加仓天数) 由 services 层用交易日历算好传进来,
    避免把日历依赖塞进纯逻辑。
  * 上限/一手/冻结等硬约束**不在这里重复判**, 统一由规则闸终检 (职责单一, 口径唯一)。
    这里只做「引擎自身的触发条件」与「批次额度」计算。
    **唯一的例外是新建仓**: 它一轮能产出多条候选, 而规则闸每条拿到的都是同一份本轮开始时
    的组合快照, 于是"每条单独看都不超上限、加起来超了"这种情况它拦不住。所以那条路的上限
    校验必须在这里就滚动算一遍 —— 用的仍然是规则闸那个 check_all_caps, 口径没有第二份。

输出候选统一结构, 供 proposal_service 走 规则闸 → 研判闸 → 按自主档位分流。

扫描入口有两个, 输入不同, 互不影响:
  scan()       输入是**已有持仓**, 产出 FILL/ADD/DCA/TRIM 与 EXIT (目标价到价, 2026-09-03 新增)
  scan_open()  输入是**上游候选池**, 产出 OPEN (2026-08-06 新增)
候选池取不到时 scan() 照常跑, 反之亦然 —— 一条外部接口的故障不该让整轮扫描停摆。
"""
from __future__ import annotations

from app.core.cushion import dca_stage, trim_trigger
from app.core import copy
from app.core.sizer import LOT, lot_of, lot_qty, split_batches
# 新建仓要在这里滚动校验上限, 用的必须是规则闸那一份 check_all_caps, 不能另写一套。
# _new_name_ctx / _ctx_after 是命令驱动建仓 (planner.plan_increase_exposure) 滚动更新组合
# 快照用的同两个函数, 一起借过来 —— 为的是让「自主建仓」与「命令建仓」的上限口径逐字一致。
# 带下划线的名字跨模块引用不好看, 但比复制一份口径出来强: 口径有两份, 迟早会分叉。
from app.core.planner import check_all_caps, _ctx_after, _new_name_ctx

A_FILL, A_ADD, A_DCA, A_TRIM = "FILL", "ADD", "DCA", "TRIM"
A_OPEN = "OPEN"                # 新建仓 (与 planner.A_OPEN、executor.BUY_ACTIONS 同名同义)
# 清仓 (与 planner.A_EXIT、executor.SELL_ACTIONS、页面动作词表同名同义)。
# 目标价到价产出的退出提议用的就是它 —— 不新造动作名, 是为了让「采纳提议 → 落指令」那一步
# (web/main.py 按动作反推买卖方向) 与执行器的清仓口径 (允许零股一次性卖出) 直接复用现成的路。
A_EXIT = "EXIT"
BUY, SELL = "buy", "sell"

# 同一轮里出现多条减持时留哪一条 (2026-09-03): 数字小的优先。用户自己设的目标价到价排在
# 系统按规则算出来的保垫减仓前面 —— 人已经说了到价就清, 这一轮就不该再自作主张先卖一部分。
# 只在同一只票的同一轮里比较, 不影响不同票, 也不影响下一轮。
_SELL_PRIORITY = {A_EXIT: 0, A_TRIM: 1}


def _sell_priority(c: dict) -> int:
    """同一轮多条减持留哪一条 (数字小的优先; 2026-09-07 第三件起按动作加来源):
    到价清仓 (人的意思) → 研究证据走弱的减持 (必交人) → 保垫减仓 (规则自动)。"""
    if c.get("action") == A_EXIT:
        return 0
    if c.get("source") == SRC_RESEARCH_WEAK:
        return 1
    return 2

# 候选来源 (2026-09-03)。每条候选都带一个来源, 提议分流那一步拿它决定「这条减持能不能不等人」。
#
# 为什么需要这一项: 分流从前写的是「卖出方向一律自动执行」, 只看方向、不看来源。那条规矩
# 的本意只有一条 —— 规则算出来的保垫减仓不该等人 —— 但写成只看方向之后, 将来任何一条卖出
# 候选都会被同一条捷径放过去, 包括本该交人裁决的那些。来源把「哪种减持可以不等人」写明白,
# 不再让方向替来源做这个决定。
#
# 要说清的边界: 决策系统高置信风控卖出的自动清仓**不走这条路** (signal_service 直接落卖出
# 指令), 用户命令驱动的清仓也不走 (命令服务 → 方案生成器 → 执行器)。所以来源这一项管的
# 只有动作引擎自己产出的候选, 以及将来接进提议分流的新来源。
SRC_ENGINE = "engine"                 # 动作引擎按规则算出来的 (FILL / ADD / DCA / TRIM / OPEN)
SRC_RESEARCH_WEAK = "research_weak"   # 研究证据走弱触发的减持 —— 必定交人裁决, 绝不自动卖出
SRC_TARGET_PRICE = "target_price"     # 用户设的目标价到价 —— 必定交人裁决, 绝不自动卖出

WHY_RESEARCH_WEAK_CONFIRM = "研究证据走弱触发的减持，必须交人裁决，不自动卖出"
WHY_TARGET_PRICE_CONFIRM = "你设的目标价到价，卖不卖由你决定，系统不自动卖出"

# 必定强制入人工队列的来源: {来源: 交给人的原因}。
# 写成一份名单而不是一个 if, 是为了将来再出现「不许自动执行」的来源时加一行就够,
# 提议分流那一段一个字都不用改。
FORCE_QUEUE_SOURCES = {SRC_RESEARCH_WEAK: WHY_RESEARCH_WEAK_CONFIRM,
                       SRC_TARGET_PRICE: WHY_TARGET_PRICE_CONFIRM}


def source_confirm_why(source):
    """这个来源要不要强制交人裁决: 要则回一句原因, 不要回 None。

    认不出的来源回 None (走原有的方向与档位判定) —— 与判决值那条「认不出一律交人」相反,
    是因为两者问的不是同一件事: 判决是上游对这只票的结论, 词表变了说明我们看不懂它的意思;
    而来源是本系统内部给候选打的标记, 没打标记的就是动作引擎自己那几类老动作, 行为必须
    与从前逐字一致。新来源要不要强制交人, 由加它的人写进上面那份名单。
    """
    return FORCE_QUEUE_SOURCES.get(str(source or "").strip()) or None

# 上游选股系统给每张候选卡的判决 (2026-09-02 起随计划接口每行下发; 旧版计划没有这个字段)。
#   候选   门槛全过、无硬风险、确认线在 —— 走现有的自动流程 (规则闸 → 研判闸 → 档位分流)
#   关注   门槛全过、无硬风险、但确认线缺失或陈旧 —— 这是上游说「无法判断」, 强制交人裁决
#   仅展示 只差覆盖或潜在吸筹 —— 不出提议, 只在候选处置里记一句原因
# 三个值是中文原文, 与上游接口逐字对应, 这里不做任何映射。
VERDICT_CANDIDATE, VERDICT_WATCH, VERDICT_DISPLAY = "候选", "关注", "仅展示"
WHY_WATCH_CONFIRM = "选股系统判为关注（无法判断），交人裁决"
WHY_DISPLAY_ONLY = "选股系统判为仅展示"

# 逻辑状态四态里唯一会触发动作的值 (2026-09-07 第三件)。四态是选股系统按「支撑这只票的研究证据
# 还在不在」算出来的跨日跟踪, 与候选卡判决是正交的两维; 只有「逻辑存疑」会改分流通道 —— 持仓停
# 增持侧自主动作、按开关出必交人的减持; 判决候选而逻辑存疑的新建仓强制人工确认。其余三个值
# (逻辑强化 / 逻辑成立 / 无法判断) 不产生任何动作。中文原文与上游逐字对应, 不做映射。
LOGIC_DOUBT = "逻辑存疑"
WHY_LOGIC_DOUBT_CONFIRM = "可交易口径成立，但研究证据在走弱（逻辑存疑），交人裁决"
WHY_LOGIC_DOUBT_PAUSE = "研究证据在走弱（逻辑存疑），停掉增持侧的自主动作"


def is_logic_doubt(logic_state) -> bool:
    return isinstance(logic_state, dict) and str(logic_state.get("state") or "").strip() == LOGIC_DOUBT


def logic_doubt_text(logic_state, head: str) -> str:
    """把存疑的出处接在一句话后面: 第一条说明与证据截止日, 让人看得出是哪一路在走弱、证据是哪天的。"""
    st = logic_state if isinstance(logic_state, dict) else {}
    parts = [head]
    reasons = [str(x) for x in (st.get("reasons") or []) if x]
    if reasons:
        parts.append(reasons[0][:120])
    if st.get("as_of"):
        parts.append(f"证据截止 {st.get('as_of')}")
    return "；".join(parts)


def logic_confirm_why(logic_state):
    """按逻辑状态要不要强制人工确认: 逻辑存疑回原因 (带出处), 其余一律 None。
    与 verdict_confirm_why 并列: 判决管「今天要不要买」, 逻辑状态管「研究证据还在不在」。"""
    if not is_logic_doubt(logic_state):
        return None
    return logic_doubt_text(logic_state, WHY_LOGIC_DOUBT_CONFIRM)


# 冻结的三种来路，对交易员是三件不同的事：自己下的命令、自己手设的、系统刹车期间。
# 原先一句「该股 COMMAND_HALT, 禁增持」三种都不分，还把枚举值印在了页面上。
_FROZEN_WHY = {
    "COMMAND_HALT": "你下过命令把这只票设成只卖不买",
    "MANUAL": "你手动把这只票设成只卖不买",
    "BRAKE": "组合刹车期间这只票只卖不买",
}


def frozen_why(reason) -> str:
    """这只票为什么不能加仓。认不出的来路说一句实话，不硬猜。"""
    r = str(reason or "").strip()
    return _FROZEN_WHY.get(r, "这只票现在只卖不买") + "，所以不评估加仓"


def display_only_why(row) -> str:
    """这一栏回答的是「为什么今天没买这只票」。所以它只写原因, 不写系统内部的判决名。

    2026-09-04 改过一次。改之前写的是「选股系统判为仅展示, 不出提议: 门槛未过: started」,
    三处毛病: 「仅展示」「不出提议」是内部说法, 读的人不知道是什么意思; started 是门槛在
    代码里的变量名, 不该出现在页面上; 整句话绕了两层还是没说清到底为什么不买。

    现在直接用判决依据 —— 那是选股系统判决时当场写下的一句完整中文, 例如
    「当日涨幅 -3.2%, 没到 3% 的启动线」或「择时决策系统昨夜给出卖出信号」。
    旧版计划没有这个字段, 那时退回一句大白话, 不留空。
    """
    basis = (str(row.get("basis") or "").strip() if isinstance(row, dict) else "")
    return basis or "选股系统今天不建议买入"


def verdict_confirm_why(verdict):
    """按判决要不要强制人工确认: 要则回原因, 不要回 None。

    候选与没有判决 (旧版计划) 都回 None —— 走档位。关注回固定文案。仅展示不该走到这里
    (scan_open 在更前面就跳过了), 真走到也按关注处理。判决值认不出来 (上游改了词表)
    一律强制交人: 宁可多问一次人, 也不拿一个看不懂的判决自动建仓。
    """
    v = (str(verdict).strip() if verdict is not None else "")
    if not v or v == VERDICT_CANDIDATE:
        return None
    if v == VERDICT_WATCH:
        return WHY_WATCH_CONFIRM
    if v == VERDICT_DISPLAY:
        return WHY_DISPLAY_ONLY + "，交人裁决"
    return f"选股系统判决值无法识别（{v}），交人裁决"

# **有资格**送研判闸的动作 (设计 §7: 自主提议的补足/加仓/补仓/调仓; 2026-08-06 加入新建仓)。
# 注意只是"有资格"——真正送不送由页面参数 PMS_JUDGE_ACTIONS 决定 (judge.request 第一行就按它
# 过滤)。两道门分开是有用的: 决策系统那侧的 OPEN 判据万一要退回去, 页面上摘掉一个词就行,
# 不改码、不部署、不重启。
JUDGE_ACTIONS = {A_FILL, A_ADD, A_DCA, A_OPEN}


def _f(v, d=0.0):
    try:
        return float(v)
    except (TypeError, ValueError):
        return d


def _cand(p, action, side, qty, reason, hard, *, confirm=False, source=SRC_ENGINE):
    return {"ts_code": p["ts_code"], "action": action, "side": side, "qty": int(qty),
            "reason": reason, "hard_numbers": hard, "needs_user_confirm": bool(confirm),
            "judge_required": action in JUDGE_ACTIONS, "source": source}


def reduce_on_weak_research(ts_code, *, qty, reason, hard_numbers=None, action=A_TRIM):
    """研究证据走弱触发的减持候选 —— **预留通道, 目前还没有上游在产出它** (2026-09-03)。

    为什么空着也要先建: 减持这条路从前是「方向是卖就自动执行」, 研究证据走弱这类减持哪天
    接上来, 默认结果就是自动卖出; 而设计里这类减持必须交人裁决。通道与「必定入人工队列」
    的判定先钉在这里并由单测守住, 将来接上游时不必再想一遍, 也不会有人凭印象接错。

    产出的候选与四类自主动作同结构, 可以直接送 proposal_service._route_one。它带三样东西:
    来源标记 (分流据此强制入队)、needs_user_confirm (与深档补仓同一个字段)、交人的原因。
    """
    c = _cand({"ts_code": ts_code}, action, SELL, qty, reason, dict(hard_numbers or {}),
              confirm=True, source=SRC_RESEARCH_WEAK)
    c["confirm_why"] = WHY_RESEARCH_WEAK_CONFIRM
    return c


def batch_amount(p: dict, params: dict, idx: int) -> float:
    """第 idx 批 (0=底仓 1=回踩补足 2=盈利加仓) 的金额额度 = 目标仓位 × 该批比例。"""
    scale = _f(params.get("scale"))
    target_pct = _f(p.get("target_pct")) or _f(params.get("stock_target_default"), 0.06)
    split = params.get("batch_split") or (0.5, 0.25, 0.25)
    ratio = split[idx] if idx < len(split) else 0.0
    return scale * target_pct * ratio


def room_to_target(p: dict, params: dict) -> float:
    """距目标仓位还差多少钱 (已达标返回 0)。"""
    scale = _f(params.get("scale"))
    target_pct = _f(p.get("target_pct")) or _f(params.get("stock_target_default"), 0.06)
    return max(0.0, target_pct * scale - _f(p.get("market_value")))


# ================================================================ 四类动作
def eval_fill(p: dict, params: dict, mkt: dict):
    """回踩补足: 建仓期内回踩到支撑带但没破, 浮亏还浅 —— 把底仓补到位。"""
    if int(p.get("fill_count") or 0) > 0:
        return None                                   # 每票 1 次
    tdays = mkt.get("tdays_since_open")
    if tdays is None or tdays > int(params.get("build_window_tdays", 10)):
        return None                                   # 只在建仓期内
    cushion = p.get("cushion_pct")
    if cushion is None:
        return None
    floor = _f(params.get("fill_max_loss"), -0.03)
    if not (floor < _f(cushion) <= 0):
        return None                                   # 浮亏 <3% 且尚未转盈
    support = _f(p.get("support_ref"))
    price = _f(p.get("price"))
    if support <= 0 or price <= 0:
        return None                                   # 没有支撑参考位就不做这件事
    if price < support:
        return None                                   # 支撑已破, 不补
    room = room_to_target(p, params)
    amt = min(batch_amount(p, params, 1), room)
    qty = lot_qty(amt, price)
    if qty < lot_of(p.get("ts_code")):      # 科创板买入 200 股起
        return None
    return _cand(p, A_FILL, BUY, qty,
                 f"回踩补足: 建仓第 {tdays} 个交易日, 浮亏 {_f(cushion):.2%} 未破支撑 "
                 f"{support}, 补 {qty} 股至目标仓位",
                 {"cushion_pct": cushion, "support": support, "price": price,
                  "tdays_since_open": tdays, "room_to_target": round(room, 2)})


def eval_add(p: dict, params: dict, mkt: dict):
    """盈利加仓: 垫子够厚 + 走出新高或站上压力位 —— 顺势加。"""
    cushion = p.get("cushion_pct")
    solid = _f(params.get("cushion_solid"), 0.03)
    if cushion is None or _f(cushion) < solid:
        return None                                   # 垫子不厚不加
    price = _f(p.get("price"))
    if price <= 0:
        return None
    high5, pressure = _f(mkt.get("high5")), _f(p.get("pressure_ref"))
    breakout = (high5 > 0 and price >= high5) or (pressure > 0 and price >= pressure)
    if not breakout:
        return None
    since_add = mkt.get("tdays_since_last_add")
    if since_add is not None and since_add < 2:
        return None                                   # 距上次加仓 ≥2 交易日
    ma5 = _f(mkt.get("ma5"))
    no_chase = _f(params.get("no_chase_ma5"), 0.06)
    if ma5 > 0 and price / ma5 - 1 > no_chase:
        return None                                   # 不追高 (规则闸还会再拦一次)
    room = room_to_target(p, params)
    amt = min(batch_amount(p, params, 2), room) if room > 0 else batch_amount(p, params, 2)
    qty = lot_qty(amt, price)
    if qty < lot_of(p.get("ts_code")):      # 科创板买入 200 股起
        return None
    why = "创 5 日新高" if (high5 > 0 and price >= high5) else "站上压力位"
    return _cand(p, A_ADD, BUY, qty,
                 f"盈利加仓: 安全垫 {_f(cushion):.2%} ≥ {solid:.0%} 且{why} "
                 f"({price} vs 高点 {high5 or '—'}/压力 {pressure or '—'}), 加 {qty} 股",
                 {"cushion_pct": cushion, "price": price, "high5": high5,
                  "pressure": pressure, "ma5": ma5, "tdays_since_last_add": since_add})


def eval_dca(p: dict, params: dict, mkt: dict):
    """补仓: 浮亏触及评估档才评估, 各档评估一次、执行终身一次, 深档永远要用户点头。"""
    cushion = p.get("cushion_pct")
    if cushion is None or _f(cushion) >= 0:
        return None
    if int(p.get("dca_qty") or 0) > 0:
        return None                                   # 终身一次, 已执行过
    triggers = params.get("dca_triggers") or (-0.08, -0.15)
    stage = dca_stage(_f(cushion), triggers)
    if stage <= 0:
        return None
    evaluated = int(p.get("dca_count") or 0)
    if stage <= evaluated:
        return None                                   # 该档已评估过, 不重复提
    price = _f(p.get("price"))
    base_qty = int(p.get("base_qty") or 0)
    if price <= 0 or base_qty <= 0:
        return None
    max_ratio = _f(params.get("dca_max_ratio"), 0.5)
    qty = int(base_qty * max_ratio // LOT) * LOT
    if qty < lot_of(p.get("ts_code")):      # 科创板买入 200 股起
        return None
    deep = _f(params.get("dca_deep_confirm"), -0.15)
    is_deep = _f(cushion) <= deep + 1e-12
    depth = sorted(triggers, reverse=True)[stage - 1]
    return _cand(p, A_DCA, BUY, qty,
                 f"补仓评估: 浮亏 {_f(cushion):.2%} 触及 {depth:.0%} 档 (第 {stage} 档), "
                 f"拟补 {qty} 股 (≤底仓 {max_ratio:.0%})"
                 + ("; **深档: 必须用户确认**, 研判须回答杀逻辑还是杀情绪" if is_deep else ""),
                 {"cushion_pct": cushion, "stage": stage, "trigger": depth,
                  "base_qty": base_qty, "price": price, "stop_ref": p.get("stop_ref")},
                 confirm=is_deep)


def eval_trim(p: dict, params: dict, mkt: dict = None):
    """保垫减仓: 垫子冲高后回吐过半 —— 减 1/3 把利润锁住。减持不设确认门槛, 规则自动执行。"""
    peak = _f(p.get("cushion_peak"))
    now = p.get("cushion_pct")
    if now is None:
        return None
    if not trim_trigger(peak, _f(now), _f(params.get("trim_peak"), 0.06),
                        _f(params.get("trim_giveback"), 0.5)):
        return None
    total = int(p.get("total_qty") or 0)
    qty = int(total / 3 // LOT) * LOT
    if qty < lot_of(p.get("ts_code")):      # 科创板部分卖 200 股起 (不足就不自动减)
        return None
    return _cand(p, A_TRIM, SELL, qty,
                 f"保垫减仓: 安全垫峰值 {peak:.2%} 回吐至 {_f(now):.2%} (过半), "
                 f"减 {qty} 股锁盈",
                 {"cushion_peak": peak, "cushion_pct": now, "total_qty": total,
                  "price": p.get("price")})


def eval_target(p: dict, params: dict = None, mkt: dict = None):
    """目标价到价: 现价摸到你设的目标价 —— 提议把这只票清掉, 交你拍板, 系统不自动卖。

    这是「设定某股目标价」这条命令的落地处 (2026-09-03 接通)。在此之前这条命令只把价格
    存进命令表, 全库没有任何地方读它: 页面点得下去、状态会变成生效中、列表里看得见,
    而到价那天什么都不会发生 —— 设计里写的「触发即生成止盈提议」一直是空的。

    三处口径是照抄现成的, 没有新阈值:
      触发     现价 ≥ 目标价。目标价是你自己填的数, 不是系统算的, 所以不设缓冲、不做平滑,
               也**不随收益读数回调** —— 系统在这件事上没有可调的旋钮。
      数量     全部持仓 (零股一并), 与「清仓某股」命令的 planner.plan_exit_stock 同一口径;
               动作名也用同一个 EXIT, 于是执行器的清仓择时 (允许零股一次性卖出) 直接复用。
      落点     与保垫减仓同层 —— 同样是动作引擎产出候选, 走 规则闸 → 提议分流。区别只有
               一条: 保垫减仓是纯规则算的、可以自动执行, 而目标价是**你的意思**, 到价只
               意味着「该问你了」, 不意味着「替你卖」。所以带上强制交人的来源标记, 分流
               那一步无论档位是不是 full 都必定落进等你拍板的队列。

    没有现价 (停牌、行情断了) 时不产出: 扫描入口已经把 price_ok 为假的票整只挡掉了,
    这里再看一眼 price 本身, 是因为本函数也可能被单独调用。「拿不到价」绝不能折成「到价了」。
    """
    target = _f(p.get("target_price"))
    if target <= 0:
        return None                                   # 没设目标价, 这条动作根本不存在
    if p.get("price_ok") is False:
        return None                                   # 价是拿摊薄成本顶的, 不是行情
    price = _f(p.get("price"))
    if price <= 0:
        return None                                   # 取不到现价, 宁可不动
    if price < target:
        return None                                   # 还没到价
    total = int(p.get("total_qty") or 0)
    if total <= 0:
        return None
    # 按 T+1 可卖量报数, 不按全部持仓 (2026-09-07 审查修)。到价当天若有买入成交
    # (自动补仓、加仓、做 T 都会), 可卖量小于总持仓; 原先按总持仓报, 规则闸一句
    # 「卖出 > T+1 可卖」整条拒掉, 再被当日去重挡住, 人当天根本看不到这条到价提议,
    # 明天价格回落就错过。执行器那条路对卖出本来就按可卖量夹紧, 这里与它对齐。
    # 可卖量为零 (今天刚建的仓) 不产出: 明天可卖量重置后若仍到价, 自然再来。
    qty = clamp_sell_qty(total, p)
    if qty <= 0:
        return None
    note = "" if qty == total else f" (总持仓 {total} 股, 今天只能卖 T+1 可卖的 {qty} 股)"
    c = _cand(p, A_EXIT, SELL, qty,
              f"目标价到价: 现价 {price} 已到你设的目标价 {target}, "
              f"拟清仓 {qty} 股 (含零股){note}; 卖不卖由你拍板, 系统不自动卖",
              {"target_price": target, "price": price, "total_qty": total,
               "avail_qty": int(p.get("avail_qty") or 0),
               "cushion_pct": p.get("cushion_pct"), "avg_cost": p.get("avg_cost")},
              confirm=True, source=SRC_TARGET_PRICE)
    c["confirm_why"] = WHY_TARGET_PRICE_CONFIRM
    return c


def eval_weak_research(p: dict, params: dict, mkt: dict = None):
    """研究证据走弱的减持 (2026-09-07 第三件): 持仓行上的逻辑状态是「逻辑存疑」时, 提议减掉一部分。

    这是 reduce_on_weak_research 那条预留通道的第一个上游。三条口径一次定死:
      开关   params["logic_doubt_trim"] (PMS_LOGIC_DOUBT_TRIM), **默认关** —— 先跑停增持一个月看误伤;
      比例   params["logic_doubt_trim_ratio"] (PMS_LOGIC_DOUBT_TRIM_RATIO), 默认与保垫减仓相同的三分之一;
      数量   按 T+1 可卖量夹紧, 不足最小申报数量不产出 (与保垫减仓、到价清仓同口径)。
    产出的候选带 research_weak 来源与强制确认标记, 分流那一步一票否决入人工队列, 绝不自动卖。
    没有逻辑状态 (早上没查到) 不产出 —— 「没有读数」不是「证据走弱」。
    """
    if not params.get("logic_doubt_trim"):
        return None
    st = p.get("logic_state")
    if not is_logic_doubt(st):
        return None
    total = int(p.get("total_qty") or 0)
    ratio = _f(params.get("logic_doubt_trim_ratio"), 1.0 / 3)
    if ratio <= 0 or ratio > 1:
        ratio = 1.0 / 3
    qty = clamp_sell_qty(int(total * ratio // LOT) * LOT, p)
    if qty < lot_of(p.get("ts_code")):
        return None
    why = logic_doubt_text(st, f"研究证据走弱: 拟减 {qty} 股 (持仓 {total} 股的 {ratio:.0%}), 卖不卖由你拍板")
    return reduce_on_weak_research(
        p["ts_code"], qty=qty, reason=why,
        hard_numbers={"logic_state": st, "price": p.get("price"),
                      "cushion_pct": p.get("cushion_pct"), "total_qty": total,
                      "avail_qty": p.get("avail_qty"), "trim_ratio": round(ratio, 4)})


def clamp_sell_qty(qty: int, pos: dict) -> int:
    """卖出数量按 T+1 可卖量夹紧。可卖量缺失 (None) 时按总持仓, 不当成零。

    页面采纳一条清仓提议时也走这里: 提议记的是扫描那一刻的股数, 等人拍板时持仓
    可能已经被别的路卖少了, 照原数落单会被卖出前的检查整条驳回, 人点了「清掉」
    结果一股没卖。按拍板那一刻的可卖量重算, 才能保证「人说卖, 就真的卖得出去」。
    """
    q = int(qty or 0)
    if q <= 0:
        return 0
    avail = pos.get("avail_qty")
    if avail is None:
        return q
    return max(0, min(q, int(avail or 0)))


# 研究走弱的减持排在保垫减仓前面 (2026-09-07 第三件): 两者动作名都是 TRIM, 同轮同现时按
# _sell_priority 留研究走弱那条 (它必交人), 保垫减仓让路 —— 证据在走弱的票, 不该由规则自动先卖一刀。
EVALUATORS = ((A_TRIM, eval_weak_research), (A_TRIM, eval_trim), (A_EXIT, eval_target),
              (A_ADD, eval_add), (A_FILL, eval_fill), (A_DCA, eval_dca))

# 减持方向的动作 —— 冻结只禁增持, 这几类照评 (与规则闸 _check_sell 同一口径)。
SELL_SIDE_ACTIONS = (A_TRIM, A_EXIT)


# ================================================================ 跳过原因
SKIP_WHY_FALLBACK = "已有在途提议或指令, 或今日已被闸门拒过"


def skip_why(skip, key):
    """这个 (代码, 动作) 为什么被跳过。不在跳过集合里返回 None。

    skip 可以是 `{(代码, 动作): 原因}` 字典, 也可以是只有键的集合 (旧口径)。
    集合形式退回一句笼统的兜底话 —— 这是 2026-08-06 之前的样子, 留着是为了兼容
    既有单测与临时调用, 生产路径由 proposal_service 传字典进来。

    为什么值得专门改这一处: 那句兜底话把四种处境揉成了一句 —— 有在途提议、有在途指令、
    今天被规则闸拒过、今天被研判闸驳回过。前两种明天照样会被挡, 后两种日切就重新评估,
    而它们在页面上长得一模一样, 看的人判断不了这只票明天还会不会再被评估。
    「明着跳过了」不该等于「说不清为什么跳过」。
    """
    if not skip or key not in skip:
        return None
    if isinstance(skip, dict):
        return skip.get(key) or SKIP_WHY_FALLBACK
    return SKIP_WHY_FALLBACK


# ================================================================ 扫描入口
def scan(*, positions: list, params: dict, market: dict, skip=None,
         strategy_codes=None, stock_params=None) -> dict:
    """扫描全部持仓, 产出候选动作。

    positions:    portfolio.positions_view()["held"] 的口径
    market:       {ts_code: {ma5, high5, tdays_since_open, tdays_since_last_add}}
    skip:         不再评估的 (代码, 动作) —— 字典时值是跳过原因, 集合时退回兜底话
    stock_params: 个股参数命令的当前值 {ts_code: {target_price, stop_price, black, ...}},
                  由 command_service.effective_stock_params() 给出。**事实源是命令表**,
                  不是持仓行上的投影 —— 投影每天盘前会被参考位取数覆盖, 拿它做判断会
                  在第二天早上悄悄失效。不传时目标价这条动作不评估, 其余四类行为不变。
    返回 {"candidates": [...], "skipped": [...]}
    """
    skip = skip or {}
    strategy_codes = strategy_codes or set()
    stock_params = stock_params or {}
    out, skipped = [], []
    for p in positions or []:
        code = p.get("ts_code")
        if not code or int(p.get("total_qty") or 0) <= 0:
            continue
        # 挂了 ACTIVE 交易方案(策略)的票由策略层接管, 四类自主动作不评 (设计 §四)。
        # **但目标价那条照评** (2026-09-07 审查修): 目标价是用户下的命令, 优先级高于自动
        # 挂上的方案; 原先整只跳过, 结果是命令表显示生效中、到价那天什么都不发生,
        # 且策略层没有任何地方读目标价 —— 一条界面上点了会成功、实际永不触发的命令。
        on_strategy = code in strategy_codes
        if on_strategy:
            skipped.append({"ts_code": code, "action": "*",
                            "why": "挂了交易方案(策略), 由策略层接管, 补足/加仓/补仓/保垫减仓不自动提议; "
                                   "你设的目标价仍照常看"})
        # 取不到现价的票整只跳过, **并且留痕**。上游 (portfolio.positions_view) 在拿不到
        # 行情时会用摊薄成本顶住 price 让市值还能算, 但那个价不是行情 —— 拿它评动作会得出
        # 「安全垫恰好 0」「现价恰好等于成本」这类看着正常、实则凭空的结论。
        # 四个 evaluator 目前都会因为 cushion_pct is None 而自然返回 None, 但那是**碰巧**
        # 兜住了: eval_fill 还会拿这个假价去比支撑位。所以在入口显式挡掉, 并写进 skipped ——
        # 「什么都没发生」和「明着跳过了」在页面上必须是两回事。
        if p.get("price_ok") is False:
            skipped.append({"ts_code": code, "action": "*",
                            "why": "取不到现价 (price 是拿摊薄成本顶的), 本轮不评估该票"})
            continue
        mkt = (market or {}).get(code) or {}
        # 个股参数命令的当前值并进这一票的快照, 供目标价那条动作读。只加不覆盖既有字段,
        # 且是**这一轮的副本**, 不动调用方传进来的持仓行。
        sp = stock_params.get(code) or {}
        if sp:
            p = {**p, "target_price": sp.get("target_price"), "stop_price": sp.get("stop_price")}
        frozen = (p.get("frozen_reason") or "NONE") != "NONE"
        # 逻辑存疑停增持侧 (2026-09-07 第三件): 持仓行上挂着早上查回的逻辑状态 (logic_state_service.attach),
        # 存疑的票补足 / 加仓 / 补仓一律不评, 减持侧照评。没挂这个键 = 没有读数, 不拦。
        doubt = bool(params.get("logic_state_route")) and is_logic_doubt(p.get("logic_state"))
        # 跨轮的减持让路 (2026-09-07 审查修)。同轮只发一条减持那条规矩 (见下) 只管一次扫描;
        # 下一分钟再扫时, 等人拍板的到价清仓已成在途、按 (代码, EXIT) 被跳过, 而保垫减仓
        # 单独产出、不需确认、卖出方向又不走强制入队 —— 系统刚说「卖不卖由你定」, 一分钟后
        # 自己卖了三分之一, 人随后采纳的清仓单永远等不到可卖量。
        # 修法: 这只票只要有任何一条减持在跳过集合里 (在途提议、在途指令、今日被拒过),
        # 本轮其余减持一律让路。宁可少卖一次, 不能抢在人前面卖。
        inflight_sell = next((a for a in SELL_SIDE_ACTIONS if skip_why(skip, (code, a))), None)
        cands_this = []
        for action, fn in EVALUATORS:
            why = skip_why(skip, (code, action))
            if why:
                skipped.append({"ts_code": code, "action": action, "why": why})
                continue
            if on_strategy and action != A_EXIT:
                continue                                  # 策略票只看目标价, 原因已在上面记过
            if inflight_sell and action in SELL_SIDE_ACTIONS:
                skipped.append({"ts_code": code, "action": action,
                                "why": f"这只票已有一条减持 ({inflight_sell}) 在处理中, "
                                       f"处理完之前不再另发减持, 免得抢在人前面卖"})
                continue
            # 冻结只禁增持, 减仓照评 (与规则闸同一口径, 这里先剪枝少算一遍)
            if frozen and action not in SELL_SIDE_ACTIONS:
                skipped.append({"ts_code": code, "action": action,
                                "why": frozen_why(p.get("frozen_reason"))})
                continue
            if doubt and action not in SELL_SIDE_ACTIONS:
                skipped.append({"ts_code": code, "action": action,
                                "why": logic_doubt_text(p.get("logic_state"), WHY_LOGIC_DOUBT_PAUSE)})
                continue
            try:
                c = fn(p, params, mkt)
            except Exception as e:      # 单票异常不能拖垮整轮扫描
                skipped.append({"ts_code": code, "action": action,
                                "why": f"评估异常 {type(e).__name__}: {e}"})
                continue
            if c:
                cands_this.append(c)
        # 同一只票同轮买卖互斥 (2026-08-28 审查修): 保垫减仓的峰值是**全时段**只增不减,
        # 加仓判据看的是近 5 日窗口, 两套时间基准可以同时成立 —— 一轮里对同一只票
        # 一边 TRIM 锁盈一边 ADD 加仓, 自动对倒空耗手续费。触发减持时买入侧让路
        # (先落袋为安, 保守方向优先), 买入条件真成立的话下一轮不触发减持时自然会来。
        # 2026-09-03: 判据从「有没有 TRIM」放宽成「有没有减持」—— 目标价到价那条也是减持,
        # 一边提议清仓一边提议加仓同样是自相矛盾, 不该只有保垫减仓享受这条互斥。
        sells_this = [c["action"] for c in cands_this if c["side"] == SELL]
        if sells_this:
            for c in cands_this:
                if c["side"] == BUY:
                    skipped.append({"ts_code": code, "action": c["action"],
                                    "why": f"同轮已触发减持({'/'.join(sells_this)}), "
                                           f"买卖互斥, 买入侧让路"})
            cands_this = [c for c in cands_this if c["side"] != BUY]
        # 同一轮里只留一条减持 (2026-09-03 修)。两条减持同时留下会让人的操作落空:
        # 保垫减仓不需要确认、当场就卖掉一部分, 而到价清仓要等人拍板, 它记的股数是扫描
        # 那一刻的持仓; 等人第二天点采纳, 持仓已经被前一条卖少了, 落指令时数量超过实际
        # 持仓, 卖出前的检查会整条驳回 —— 人点了「清掉」, 结果一股没卖, 只在评审账本上
        # 留下一条拒绝。留哪一条按「人的意思优先」: 用户自己设的目标价到价排在系统按规则
        # 算出来的减仓前面。被让路的那条记进跳过原因, 下一轮条件仍成立时自然会再来。
        sells_now = [c for c in cands_this if c["side"] == SELL]
        if len(sells_now) > 1:
            keep = min(sells_now, key=_sell_priority)
            for c in sells_now:
                if c is not keep:
                    skipped.append({"ts_code": code, "action": c["action"],
                                    "why": f"同轮已有优先级更高的减持({keep['action']}), "
                                           f"一轮只发一条减持, 本条让路"})
            cands_this = [c for c in cands_this if c["side"] != SELL or c is keep]
        out.extend(cands_this)
    return {"candidates": out, "skipped": skipped}


# ================================================================ 新建仓 (OPEN)
def eval_open(c: dict, params: dict, caps: dict, room_amt: float):
    """一条候选票能不能从零建仓、建多少股。

    返回 `(候选, None)` 或 `(None, 跳过原因)` —— 与上面四个求值器只回 None 不同,
    这里**必须给得出原因**: 候选池里明明有这只票却没被提, 页面上要看得出是名额满了、
    钱不够、买不足一手, 还是被上限拦了。只是"没出现"等于什么都没说。

    c:        {ts_code, price, score, sector, theme, tier, upside, heat, rank, bucket, src}
              price 必须是**实时价**——规划用的昨收不能拿来下单 (见 market.plan_price 的注释)。
    caps:     滚动中的组合上下文 (portfolio.caps_ctx 的产出, 由 scan_open 边走边更新)
    room_amt: 本轮还剩多少钱可投 (已取过「仓位口径」与「真实可用资金」的小者)
    """
    code = c.get("ts_code")
    price = _f(c.get("price"))
    if price <= 0:
        return None, "取不到实时价, 不建仓 (规划用的昨收不能拿来下单)"
    scale = _f(params.get("scale"))
    if scale <= 0:
        return None, "总规模未设置"
    target_pct = _f(params.get("stock_target_default"), 0.06)
    want = target_pct * scale
    if want <= 0:
        return None, "单股目标仓位为 0"
    # **钱不够一整只就不开。** 这一条与命令驱动建仓有意不同: 那边是
    # `want = min(单股目标, 命令剩余额度)`, 允许最后一只按剩下的钱缩水买 —— 那是用户
    # 明确下了一条「投这么多」的命令, 缩水的那只是命令的收尾。
    # 自主建仓没有这层意思: 开一只新仓要占掉一个持仓名额, 拿一个名额去换一只 0.5% 的
    # 零头仓位是亏的 —— 它永远补不到目标, 却挡住了后面真正建得起来的票。
    if want > room_amt:
        return None, (f"剩余可投金额 {room_amt:,.0f} 元不足一只目标仓位 "
                      f"{want:,.0f} 元 ({target_pct:.0%}), 不开半截新仓")
    sp = split_batches(want, price, splits=params.get("batch_split"),
                       merge=bool(params.get("min_lot_merge", True)),
                       min_lot=lot_of(code))
    if not sp["ok"]:
        return None, sp["reason"]

    # 上限校验按**这只票的整只目标金额**算, 不是只按这次要买的底仓批。
    # 决定「要不要开这只新仓」的时候就该把它将来要占的位置留出来 —— 只按底仓批算的话,
    # 一跳能开出一堆将来永远补不满的半仓。这也与命令驱动建仓 (plan_increase_exposure
    # 里的 `actual = sum(b["qty"] * price ...)`) 是同一个口径。
    # 一句要说破的话: 这份预留只在**本轮**有效, 下一跳的组合快照是按实际市值重新取的。
    full_amt = sum(b["qty"] * price for b in sp["batches"])
    bad = check_all_caps(ts_code=code, add_amount=full_amt, ctx=_new_name_ctx(caps, c))
    if bad:
        # 这一串是纯显示用途 —— 它只被塞进 by_code[].why 给页面看, 没有任何地方拿它
        # 归类或做键 (做键的是 executor 那条路, 用的是另一份带前缀的原串)。所以在这里
        # 剥掉大写判据码, 页面上就不会出现「PORTFOLIO_CAP: …」这种东西。
        return None, copy.join_reasons(bad)

    base = sp["batches"][0]
    qty = int(base["qty"])
    if qty < lot_of(code):
        return None, f"底仓批 {qty} 股不足最小申报数量 ({lot_of(code)} 股)"

    hard = {
        # ---- 定性材料: 这只票凭什么被选出来。研判闸要看的就是这几项 ----
        "price": price, "score": c.get("score"), "theme": c.get("theme"),
        "tier": c.get("tier"), "upside": c.get("upside"), "heat": c.get("heat"),
        "plan_rank": c.get("rank"), "plan_bucket": c.get("bucket"),
        "plan_src": c.get("src"), "sector": c.get("sector"),
        # 决策系统今天盘中判过这只票转多没有。这是**定性材料**不是仓位数字, 所以送研判时
        # 也带过去 (见 judge.OPEN_JUDGE_KEYS) —— 等于当面告诉它「你自己今天判过这只票转多」,
        # 它可以拿自己的结论去对照; 没有信号时这两项是 None, 不误导。
        "intraday_buy_signal": bool(c.get("sig_buy")),
        "intraday_buy_reason": (c.get("sig_buy") or {}).get("reason"),
        # 上游候选卡的判决与理由 (2026-09-02 起下发; 旧版计划这五项都是 None)。同样是定性
        # 材料: 送研判时一并带过去 (judge.OPEN_JUDGE_KEYS), 让决策系统逐条核对「这几条理由
        # 到今天还成不成立」; 进评审账本与提议卡, 让人看得见系统为什么挑了它。
        "verdict": c.get("verdict"), "reasons": c.get("reasons"),
        "missing": c.get("missing"), "risk": c.get("risk"), "card_rank": c.get("card_rank"),
        # 研究理由两键 (2026-09-03): 判决依据是上游对自己这条判决的一句话解释, 因果论断是
        # 研报里说这家公司好在哪、出处是哪份研报哪一天。两者只进评审账本与提议卡给人看,
        # 不参与任何判定, 也不送研判 —— 产业逻辑归数据基座与选股系统, 不归择时决策系统。
        "basis": c.get("basis"), "logic": c.get("logic"),
        # 逻辑状态四态 (2026-09-07 第三件): 入场那一刻「研究证据还在不在」的读数, 进账本与提议卡,
        # 持仓视图的「当初为什么买」从这里回溯; 不送研判 (judge.OPEN_JUDGE_KEYS 不收它)。
        "logic_state": c.get("logic_state"),
        # 安全边际三情景的整句 (2026-09-07 第四件): 只给人看, 不送研判。
        "valuation_text": c.get("valuation_text"),
        # 催化事件与定价状态 (2026-09-08 量价研判链 3.5): 原值进账本与提议卡; 两句整句另送研判
        # (它们是事件与量价, 不是产业逻辑, 见 judge.OPEN_JUDGE_KEYS)。不参与任何判定。
        "events": c.get("events"), "events_text": c.get("events_text"),
        "pricing_state": c.get("pricing_state"), "pricing_text": c.get("pricing_text"),
        # ---- 仓位口径: 只进评审账本做判分锚。judge.py 送研判时会把这几项过滤掉,
        #      理由见那边的 OPEN_JUDGE_KEYS —— 决策系统本来就不管仓位, 别送过去带偏它。
        "target_pct": target_pct, "target_amount": round(full_amt, 2),
        "base_amount": round(qty * price, 2),
        "batch_scheme": ",".join(str(x) for x in (sp.get("scheme") or ())),
        "names_before": caps.get("names_count"), "max_names": caps.get("max_names"),
        "room_amt_before": round(_f(room_amt), 2),
    }
    reason = (f"新建仓: 候选池第 {c.get('rank') or '—'} 名 (分数 {c.get('score') or '—'}"
              f"{', 主题 ' + str(c.get('theme')) if c.get('theme') else ''})"
              + ("; **决策系统今天盘中判过该股转多**" if c.get("sig_buy") else "")
              + f", 现价 {price}, 目标仓位 {target_pct:.0%} 约 {full_amt:,.0f} 元, "
                f"先建底仓 {qty} 股 (约 {qty * price:,.0f} 元)")
    cand = _cand(c, A_OPEN, BUY, qty, reason, hard)
    # 这两项是 OPEN 独有的, 供 proposal_service 用:
    #   price  —— 新票在账本里没有行, _pos_of 拿不到现价, 取它会是 0 而被规则闸判 PRICE_MISSING
    #   sector —— caps_ctx 对新票带不出行业名, 不显式传的话行业集中度那道硬拦截会静默跳过
    cand["price"] = price
    cand["sector"] = c.get("sector")
    cand["target_amount"] = round(full_amt, 2)
    return cand, None


def scan_open(*, candidates: list, params: dict, caps: dict, room_amt: float,
              slots: int, skip=None) -> dict:
    """从上游候选池挑新票建仓。名额与金额**边走边扣**, 所以产出的这一组候选彼此不冲突。

    为什么滚动必须在这里做, 而不是交给规则闸: `_route_one` 给每条候选调 `caps_ctx` 时,
    拿的是**本轮开始时**的那一份组合快照, 不会因为前面几条已经落了指令而更新。于是一跳产出
    五条新建仓, 每条单独看都不超上限、五条加起来超了, 规则闸一条也拦不住 —— 它每次看到的
    都是同一个旧快照。命令驱动那条路没这个问题, 因为 planner 在循环里用 `_ctx_after` 滚动。
    自主这条路照做。

    **刻意没有每日开仓上限。** 上界由三层自己收敛: 这里的名额与金额边走边扣;
    后面规则闸的资金与上限终检; 再后面择时的买入区间 (现价不在区间内一律等待、不追)。

    slots:    还能开几只新仓 = 最大持仓数 − 当前持仓数 (由调用方算好)
    room_amt: 还有多少钱可投, 已取「总仓上限×总规模 − 组合市值」与「真实可用资金」的小者
    skip:     不再评估的 (代码, 动作) —— 字典时值是跳过原因 (在途提议 / 在途指令 /
              今日被规则闸拒过 / 今日被研判闸驳回过, 这四种处境完全不同, 必须分得开)

    按上游判决分流 (2026-09-03, params["open_route_by_verdict"] 为真时生效):
      仅展示 → 不产出候选, 写一条跳过原因 (正常路径上 plan_feed 已在候选阶段剔掉, 这里是
               兜底 —— 直接调本函数的人也得到同一口径);
      关注   → 照常产出候选, 但打上 needs_user_confirm 并带上原因 (confirm_why), 由
               proposal_service._route_one 强制入人工队列;
      候选 / 没有判决 → 一个字不改。
    落点选在这里而不是 eval_open: 判决决定的是「这条候选往哪条路走」, 与「建不建、建多少」
    是两个问题; 放在产出候选的循环里, 仅展示的行连名额与金额都不扣, 而且开关关着时
    这段代码一行都不执行 —— 旧行为逐字保留。
    """
    skip = skip or {}
    out, skipped = [], []
    slots = int(slots or 0)
    left = _f(room_amt)
    route = bool(params.get("open_route_by_verdict"))
    if slots <= 0:
        return {"candidates": [], "skipped": [
            {"ts_code": "*", "action": A_OPEN,
             "why": f"持仓数已达上限 {caps.get('max_names')}, 没有新仓名额"}]}
    if left <= 0:
        return {"candidates": [], "skipped": [
            {"ts_code": "*", "action": A_OPEN,
             "why": "组合已到总仓上限, 或可用资金为零 —— 没有可投金额"}]}

    ctx = dict(caps)
    sector_on = bool(ctx.get("sector_source_ready", True))
    # 排序: 今天被决策系统判过「盘中转多」的票排最前, 其余按分数降序、同分按榜内名次
    # (与 plan_feed.select_candidates 的次序一致)。
    #
    # 为什么让信号插队而不是加分: 分数是昨夜算的静态排名, 「此刻转多」是盘中才有的新信息,
    # 两者不同量纲, 折算成分数就得凭空定一个系数。而名额是稀缺的 —— 只剩两个名额时,
    # 候选池第 25 名按分数永远轮不上, 哪怕决策系统刚刚判它转多。插队直接表达了这件事。
    # **只影响先后, 不影响资格**: 不在候选池里的票根本不会走到这里 (调用方传进来的就是
    # 候选池的产物), 候选层那一整套过滤一道都不绕。
    # PMS_OPEN_SIGNAL_PRIORITY 关掉即退回纯分数排序, 页面可改, 不用改码。
    sig_first = bool(params.get("open_signal_priority", True))
    for c in sorted(candidates or [],
                    key=lambda x: (0 if (sig_first and x.get("sig_buy")) else 1,
                                   -_f(x.get("score")), _f(x.get("rank"), 10 ** 9),
                                   str(x.get("ts_code") or ""))):
        code = c.get("ts_code")
        if not code:
            continue
        if route and c.get("verdict") == VERDICT_DISPLAY:
            # 仅展示的行不占名额、不占金额, 也不进后面任何一道闸; 放在名额判断之前,
            # 是为了让它的跳过原因永远是「仅展示」而不是碰巧的「名额用完」。
            skipped.append({"ts_code": code, "action": A_OPEN,
                            "why": display_only_why(c)})
            continue
        if slots <= 0:
            skipped.append({"ts_code": code, "action": A_OPEN,
                            "why": "本轮新仓名额已用完 (下一跳按最新持仓数重算)"})
            continue
        if left <= 0:
            skipped.append({"ts_code": code, "action": A_OPEN,
                            "why": "本轮可投金额已用完 (下一跳按最新组合市值重算)"})
            continue
        why_skip = skip_why(skip, (code, A_OPEN))
        if why_skip:
            skipped.append({"ts_code": code, "action": A_OPEN, "why": why_skip})
            continue
        try:
            cand, why = eval_open(c, params, ctx, left)
        except Exception as e:              # 单票异常不能拖垮整轮扫描 (与 scan() 同口径)
            skipped.append({"ts_code": code, "action": A_OPEN,
                            "why": f"评估异常 {type(e).__name__}: {e}"})
            continue
        if not cand:
            skipped.append({"ts_code": code, "action": A_OPEN, "why": why or "未产出候选"})
            continue
        if route:
            confirm_why = verdict_confirm_why(c.get("verdict"))
            if confirm_why:
                # 关注 (或认不出的判决): 候选照常产出、照常占名额、照常过规则闸与研判闸,
                # 只是最后一步不许自动落指令 —— 与深档补仓强制确认走的是同一个字段。
                cand["needs_user_confirm"] = True
                cand["confirm_why"] = confirm_why
        # 判决候选而逻辑存疑 (2026-09-07 第三件): 可交易口径成立而研究证据在走弱, 这正是杀逻辑与
        # 杀估值那条边界, 强制交人; 复用同一个强制确认字段, 与关注判决走同一条队列。已经要确认的
        # (关注 / 认不出的判决) 保留原因, 不覆盖。开关 PMS_LOGIC_STATE_ROUTE 关着时一行都不执行。
        if params.get("open_route_by_logic") and not cand.get("needs_user_confirm"):
            lw = logic_confirm_why(c.get("logic_state"))
            if lw:
                cand["needs_user_confirm"] = True
                cand["confirm_why"] = lw
        out.append(cand)
        used = _f(cand.get("target_amount"))
        left -= used
        slots -= 1
        ctx = _ctx_after(ctx, used, is_new_name=True,
                         sector=(c.get("sector") if sector_on else None))
    return {"candidates": out, "skipped": skipped}