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 个交易日
  硬风险  决策系统昨夜给出卖出、回避或剔除信号；传导快照日与计划日不符；吸筹评分为高位派发

  判决    候选 = 三门槛全过、硬风险为空、确认成立
          关注 = 三门槛全过、硬风险为空，但确认线缺失（没有吸筹评分）或陈旧（明确吸筹但评分
                 日龄超过上限或日龄未知）——这是"系统无法判断"，交人裁决
          仅展示 = 其余，卡上标明未过项。只差券商覆盖、吸筹评分为潜在吸筹或其他非明确状态，
                 都归这一档（2026-09-03 台账 013：潜在吸筹市值中性后为负，不升格）

依据：方案 1.8d——"环节被指向、当日已涨 3% 以上、且明确吸筹"是纯可交易口径下唯一
合并样本为正的规则（五日 +2.18、十日 +1.94），但它是顺风策略、不免疫环境；两个旋钮
保持默认，等样本外复盘读数再定，不在历史样本上挑参数（2026-09-02 决定）。
关注定义的细化出自《主观量化系统方案_2026-09-03》第 3.2 节"无法判断"的三个出口：
候选自动进建仓方案，关注强制人工确认，仅展示不出提议。

## 因果论断证据线（2026-09-03）

证据字典可带 logic：数据基座因果论断视图给出的论断列表（方向、机制、时效、出处文档标题与
披露日、论断编号）。judge 把它整理成带出处的一行行文字放进输出的 logic 键，只展示、
不作门槛、不进判决——分析结论到选股的断裂点先接通，要不要当门槛等复盘案例再定。

## 边界

本模块不读库、不读配置，只吃调用方装配好的证据字典，输出判决字典。这样它能
离线单测（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,
          logic_stale_days: int = 90) -> 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   传导快照日与计划日不符
      logic          list   数据基座因果论断列表（每条是字典：direction / mechanism / horizon /
                            doc_title / disclosure_date / claim_id 等），只展示不进判决
    返回：verdict / reasons / missing / risk / gates / confirm / failed_gates / basis / logic。
      basis  一句话判决依据，说明落到这一档的原因（关注即"系统无法判断"，交人裁决）
      logic  因果论断整理成的带出处文字行，与判决无关
      logic_quality  论断的质量画像（条数、最新披露日、距今天数、几份来源、疑似误抽与
             分歧条数、按环节展开的条数），取数层在截断前算好带上来，与判决无关
      failed_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
    # 确认线的三种"不成立"要分开：缺失（没有评分）与陈旧（明确吸筹但日龄超限或未知）是
    # "系统无法判断"；评分为潜在吸筹、信号不明、无吸筹等非明确状态，是系统已经判断过、
    # 只是没有达到确认线，不升格（台账 013）。
    confirm_missing = not accum_state
    confirm_stale = accum_state.startswith("明确") and not accum_fresh
    confirm_negative = bool(accum_state) and not accum_state.startswith("明确")

    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("无券商覆盖（不升格关注，归仅展示）")
    elif not gates["covered"]:
        missing.append(f"预期空间 {upside:+.0%} 低于负容忍线" if upside is not None else "预期空间缺失")
    if confirm_missing:
        missing.append("无吸筹评分（未入池）")
    elif confirm_stale:
        missing.append(f"吸筹评分陈旧 {accum_age} 日" if isinstance(accum_age, int)
                       else "吸筹评分日龄未知")
    elif confirm_negative and "派发" not in accum_state:
        state_short = accum_state.split("·")[0].strip()
        missing.append(f"吸筹评分为「{state_short}」，未达确认线（不升格关注，归仅展示）")
    if not gates["pointed"]:
        missing.append("无传导")
    # 研报论断的质量提示。放进缺失项是因为它和"吸筹评分陈旧"是同一类事：证据还在，
    # 但已经旧到不该当新证据用。两者此前一个会过期一个不会，卡上看着一样新，
    # 这个不对称本身会误导人。这里只提示不作门槛——门槛仍是那三条可交易口径。
    missing.extend(logic_notes(ev.get("logic"), logic_stale_days))

    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())
    failed = [k for k, ok in gates.items() if not ok]
    if risk:
        verdict = VERDICT_SHOW
        basis = "硬风险否决：" + "；".join(risk)
    elif all_gates and confirm:
        verdict = VERDICT_CANDIDATE
        basis = "三门槛全过、无硬风险、明确吸筹且评分新鲜"
    elif all_gates and (confirm_missing or confirm_stale):
        verdict = VERDICT_WATCH
        basis = ("三门槛全过、无硬风险，但确认线"
                 + ("缺失（无吸筹评分）" if confirm_missing else "陈旧（明确吸筹但评分日龄超限或未知）")
                 + "——系统无法判断，交人裁决")
    elif all_gates:
        verdict = VERDICT_SHOW
        failed.append("confirm")
        basis = f"三门槛全过，但吸筹评分为「{accum_state.split('·')[0].strip()}」，未达确认线，不升格关注"
    else:
        verdict = VERDICT_SHOW
        basis = "门槛未过：" + "、".join(failed)

    return {"verdict": verdict, "reasons": reasons, "missing": missing, "risk": risk,
            "gates": gates, "confirm": confirm, "failed_gates": failed, "basis": basis,
            "logic": logic_lines(ev.get("logic")),
            "logic_quality": logic_quality(ev.get("logic"))}


def logic_lines(claims, limit: int = 3) -> list[str]:
    """把因果论断列表整理成带出处的文字行：方向、机制、时效，加"《文档标题》披露日·论断编号"。
    只展示，不进判决；缺字段的部分省略，不报错。"""
    out: list[str] = []
    for c in (claims or [])[:limit]:
        if not isinstance(c, dict):
            continue
        head = "：".join(x for x in (c.get("direction"), c.get("mechanism")) if x)
        if c.get("horizon"):
            head = f"{head}（{c['horizon']}）" if head else f"（{c['horizon']}）"
        if c.get("condition"):
            head = f"{head}，条件：{c['condition']}"
        src = "".join(x for x in (
            f"《{c['doc_title']}》" if c.get("doc_title") else "",
            f" {c['disclosure_date']}" if c.get("disclosure_date") else "",
            f" · {c['claim_id']}" if c.get("claim_id") else "") if x)
        line = head or "因果论断"
        if src:
            line = f"{line}——出处：{src.strip()}"
        out.append(line)
    return out


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))


def logic_quality(claims) -> dict | None:
    """把取数层算好的质量画像取出来。取数层在截断之前算，所以它统计的是这票的
    全部论断，而卡上只展示前几条——两个数不一样是对的。论断为空时返回 None，
    与"上游没给这个字段"同形（卡上都不显示这一节）。"""
    for c in (claims or []):
        if isinstance(c, dict) and isinstance(c.get("quality"), dict):
            return c["quality"]
    return None


def logic_notes(claims, stale_days: int = 90) -> list[str]:
    """论断质量的提示文字，只提示不作门槛。四条，都以实际读数为准，凑不齐就少写几条：

      陈旧：最新一条论断距数据日超过 stale_days 天。
      单一来源：这票有多条论断但全部出自同一份研报——那份研报的观点一旦过时或本身
                有偏，这票的整条研究证据一起失效，而分开看每一条都像是独立的证据。
      疑似误抽：逻辑评析读过之后标出来的，模型的判断不是确定性事实，所以只提示。
      未结分歧：同期方向相反、还没有结论的，表示市场没有共识。
    """
    q = logic_quality(claims)
    if not q or not q.get("n"):
        return []
    out: list[str] = []
    age = q.get("age_days")
    if isinstance(age, int) and age > int(stale_days):
        latest = q.get("latest_date")
        out.append(f"研报论断已过 {age} 天（最新一条 {latest}）" if latest
                   else f"研报论断已过 {age} 天")
    if q["n"] > 1 and q.get("n_docs") == 1:
        out.append(f"{q['n']} 条论断全部出自同一份研报")
    if q.get("n_flagged"):
        out.append(f"其中 {q['n_flagged']} 条被标为疑似误抽")
    if q.get("n_disputed"):
        out.append(f"其中 {q['n_disputed']} 条处于未结的方向分歧")
    return out