akg-factor-bridge/card.py

"""候选卡：把一只票的各条证据线装配成判决与理由（纯函数，零 IO）。

## 为什么要有它

桥原来的产物是一个分数：主榜分 = 200 + 传导档位×20 + 组内分。分数排在最前的
强传导档，实证里过去一个月是四档中最差的（五日超额 −2.04，见
docs/主观选股改进方案_2026-09-02.md 第 1.8b 节）。方案把桥从"打分排序器"改成
"候选卡装配器"：分数与档位一律不动，另出一份带理由的候选单。规则只有三条硬门槛、
一条确认、三条硬风险，全部是一个主观选股的人会用的判断，不算总分。

  门槛一  所在环节当日被传导指向（只认环节类目标）
  门槛二  成员当日已启动：数据日涨幅达到 CARD_START_PCT（默认 3%，与基座热点扫描同口径）
  门槛三  有券商覆盖、预期空间不低于负容忍线、非 ST 与退市族
  确认    决策系统吸筹评分为"明确吸筹"，且评分日龄不超过 CARD_ACCUM_MAX_AGE 个交易日
  硬风险  决策系统昨夜给出卖出、回避或剔除信号；传导快照日与计划日不符；吸筹评分为高位派发

  判决    候选 = 三门槛全过、硬风险为空、确认成立
          关注 = 硬风险为空，且（只差覆盖一项）或（门槛全过但无确认）
          仅展示 = 其余，卡上标明未过项

依据：方案 1.8d——"环节被指向、当日已涨 3% 以上、且明确吸筹"是纯可交易口径下唯一
合并样本为正的规则（五日 +2.18、十日 +1.94），但它是顺风策略、不免疫环境；两个旋钮
保持默认，等样本外复盘读数再拍，不在历史样本上挑参数（2026-09-02 拍板）。

## 边界

本模块不读库、不读配置，只吃调用方装配好的证据字典，输出判决字典。这样它能
离线单测（test_card.py），也保证同一段规则在计划装配、复盘脚本、对账工具里只有一份。
坏信号集合 BAD_SIGNALS 是"三处同源"纪律里桥的那一份（另两处在决策系统 pms_advisor.py
与 PMS rule_gate.py），改一处必须三处同改；pool.py 从这里引用，桥内只此一处。
"""
from __future__ import annotations

from typing import Any

# 决策系统夜间结论表 signal_type 里的坏信号（三处同源，见模块说明）
BAD_SIGNALS = {"SELL", "AVOID", "DROPPED"}

VERDICT_CANDIDATE = "候选"
VERDICT_WATCH = "关注"
VERDICT_SHOW = "仅展示"
_VERDICT_ORDER = {VERDICT_CANDIDATE: 0, VERDICT_WATCH: 1, VERDICT_SHOW: 2}


def _num(v) -> float | None:
    if v is None:
        return None
    try:
        x = float(v)
    except (TypeError, ValueError):
        return None
    if x != x:            # NaN
        return None
    return x


def judge(ev: dict[str, Any], *, start_pct: float = 3.0,
          accum_max_age: int = 30, neg_tol: float = 0.0) -> dict[str, Any]:
    """一只票的证据字典 -> 判决字典。

    ev 里认的键（缺键按缺失处理，不报错）：
      pointed        bool   所在环节当日被传导指向（只认 Segment 目标）
      theme          str    目标环节名；n_sources int 源数；chain_fit 链符（展示用）
      pct0           float  数据日涨幅（百分数，3.2 表示 +3.2%）
      covered        bool   有券商覆盖；upside float 预期空间（比例，0.25 表示 +25%）
      risk_name      bool   证券简称命中 ST / *ST / 退市族
      accum_state    str    吸筹状态文案（决策系统 fund_flow.state）；accum_score float；
      accum_age      int    评分日龄（交易日）
      y_signal       str    决策系统昨夜 signal_type
      stale_snapshot bool   传导快照日与计划日不符
    返回：verdict / reasons / missing / risk / gates / confirm。
    """
    pct0 = _num(ev.get("pct0"))
    upside = _num(ev.get("upside"))
    accum_score = _num(ev.get("accum_score"))
    accum_age = ev.get("accum_age")
    accum_state = str(ev.get("accum_state") or "")
    y_signal = str(ev.get("y_signal") or "").strip().upper()

    gates = {
        "pointed": bool(ev.get("pointed")),
        "started": pct0 is not None and pct0 >= float(start_pct),
        "covered": bool(ev.get("covered")) and upside is not None
                   and upside >= -float(neg_tol or 0.0),
        "clean_name": not bool(ev.get("risk_name")),
    }
    accum_fresh = (isinstance(accum_age, int) and 0 <= accum_age <= int(accum_max_age))
    confirm = accum_state.startswith("明确") and accum_fresh

    risk: list[str] = []
    if y_signal in BAD_SIGNALS:
        risk.append(f"决策系统昨夜信号 {y_signal}")
    if ev.get("stale_snapshot"):
        risk.append("传导快照日与计划日不符")
    if "派发" in accum_state:
        risk.append("高位派发")

    missing: list[str] = []
    if pct0 is None:
        missing.append("无行情")
    if not ev.get("covered"):
        missing.append("无券商覆盖")
    if not accum_state:
        missing.append("无吸筹评分（未入池）")
    elif accum_state.startswith("明确") and not accum_fresh:
        missing.append(f"吸筹评分陈旧 {accum_age} 日" if isinstance(accum_age, int)
                       else "吸筹评分日龄未知")
    if not gates["pointed"]:
        missing.append("无传导")

    reasons: list[str] = []
    if gates["pointed"]:
        theme = ev.get("theme") or "环节"
        n = ev.get("n_sources")
        reasons.append(f"所在环节「{theme}」被{n}路指向" if n else f"所在环节「{theme}」被指向")
    if gates["started"]:
        reasons.append(f"当日已启动 {pct0:+.1f}%")
    if confirm:
        reasons.append(f"明确吸筹（评分 {accum_score:.0f}，{accum_age} 日前）"
                       if accum_score is not None else f"明确吸筹（{accum_age} 日前）")
    if gates["covered"]:
        reasons.append(f"券商覆盖，预期空间 {upside:+.0%}")

    all_gates = all(gates.values())
    only_missing_coverage = (gates["pointed"] and gates["started"] and gates["clean_name"]
                             and not gates["covered"])
    if risk:
        verdict = VERDICT_SHOW
    elif all_gates and confirm:
        verdict = VERDICT_CANDIDATE
    elif only_missing_coverage or (all_gates and not confirm):
        verdict = VERDICT_WATCH
    else:
        verdict = VERDICT_SHOW

    failed = [k for k, ok in gates.items() if not ok]
    return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk,
            "gates": gates, "confirm": confirm, "failed_gates": failed}


def sort_key(row: dict[str, Any]) -> tuple:
    """卡内序：先判决（候选 < 关注 < 仅展示），再按数据日涨幅降序。不算总分。"""
    v = _VERDICT_ORDER.get(row.get("verdict"), 9)
    pct0 = _num(row.get("pct0"))
    return (v, -(pct0 if pct0 is not None else -1e9))