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"}

# 昨夜信号的代号翻译成中文。这些代号是决策系统那边的枚举值，只在系统之间传递；
# 一旦要写进给人看的句子里，就必须先翻成人话——页面上出现一个 SELL，读的人得先
# 知道这是个枚举值才看得懂。翻不出来的原样带上，总比吞掉强。
SIGNAL_CN = {"SELL": "卖出", "AVOID": "回避", "DROPPED": "已剔除",
             "BUY": "买入", "WATCH": "观察", "HOLD": "持有"}


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, require_started: bool = True) -> 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")),
        # require_started 关掉时这一条恒过：涨跌多少都不拦，没有当日行情也不拦。
        # 这是整条门槛的开关，不是把阈值调成一个很小的负数——那样"没有行情"的票
        # 仍然会被拦下，而关掉门槛的本意是这一条根本不参与判断。
        "started": (not require_started) or (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"择时决策系统昨夜给出{SIGNAL_CN.get(y_signal, 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(upside_text(upside, neg_tol))
    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"] and require_started:
        reasons.append(f"当日已启动 {pct0:+.1f}%")
    # 启动线关着时**理由行不写当日涨幅**（2026-09-07 审查修）。09-04 曾写成
    # "当日涨幅 x%（启动线已关闭，不作门槛）"——理由行在 PMS 送研判的白名单里，择时层的
    # 模型能直接读到这句选股层的政策旁白，它回答"启动线关闭、价格处高位滞涨区"就是被
    # 牵着走的结果。当日涨幅仍在 card 子字典的 pct0 里给人看，只是不再当理由、也不送研判。
    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]
    # 判决依据是这张卡上唯一一句"为什么"，PMS 原样显示在页面上给人读。所以它必须是
    # 一句完整的中文，带上具体数字，不许出现门槛的变量名（started 这类）或信号的
    # 枚举值（SELL 这类）——那些是系统之间传递用的，不是给人看的。
    if risk:
        verdict = VERDICT_SHOW
        basis = "；".join(risk)
    elif all_gates and confirm:
        verdict = VERDICT_CANDIDATE
        basis = (f"三条门槛都过了，没有风险，且明确吸筹（{accum_age} 个交易日前评的）"
                 if isinstance(accum_age, int) else "三条门槛都过了，没有风险，且明确吸筹")
    elif all_gates and (confirm_missing or confirm_stale):
        verdict = VERDICT_WATCH
        why = ("这只票还没有吸筹评分" if confirm_missing
               else (f"吸筹评分是 {accum_age} 个交易日前的，太旧了"
                     if isinstance(accum_age, int) else "吸筹评分有多旧算不出来"))
        basis = f"三条门槛都过了，也没有风险，但{why}，系统判断不了，交人裁决"
    elif all_gates:
        verdict = VERDICT_SHOW
        failed.append("confirm")
        basis = (f"三条门槛都过了，但吸筹只到「{accum_state.split('·')[0].strip()}」，"
                 f"没到明确吸筹这条线")
    else:
        verdict = VERDICT_SHOW
        basis = "；".join(gate_reasons(failed, pct0=pct0, start_pct=start_pct,
                                       covered=bool(ev.get("covered")), upside=upside,
                                       neg_tol=neg_tol))

    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']}"
        # 出处只写研报名、披露日、距今多少天。论断编号是 36 位的内部标识，
        # 原先挂在行尾占掉半行、对读的人没有任何信息；追溯用的编号仍在论断原值里，
        # 计划的 card.logic_claims 那一支照旧带着，只是不进给人读的这一行。
        # 距今天数取取数层算好的质量画像——旧不旧交给人自己判，系统不替他判。
        age = (c.get("quality") or {}).get("age_days")
        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"（距今 {age} 天）" if isinstance(age, int) 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


def gate_reasons(failed, *, pct0=None, start_pct: float = 3.0, covered: bool = False,
                 upside=None, neg_tol: float = 0.0) -> list[str]:
    """把没过的门槛翻成一句句人话，带上当时的实际数字。

    门槛在代码里的名字是 pointed、started、covered，那是给程序读的。写进给人看的
    句子必须换成人话，并且带上数字——「当日跌 3.2%，没到 3% 的启动线」比「涨幅门槛
    未过」有用得多，因为前者能让人自己判断差多少。

    认不出的门槛名原样带上而不是吞掉：真出现了新门槛，宁可页面上难看一次，
    也好过一句话里少了一条原因。
    """
    out: list[str] = []
    for k in failed or []:
        if k == "pointed":
            out.append("所在环节当日没有被传导指向")
        elif k == "started":
            out.append(f"当日涨幅 {pct0:+.1f}%，没到 {start_pct:g}% 的启动线"
                       if pct0 is not None else "没有当日行情，算不出涨幅")
        elif k == "covered":
            if not covered:
                out.append("没有券商覆盖")
            elif upside is not None:
                out.append(upside_text(upside, neg_tol))
            else:
                out.append("算不出券商预期空间")
        elif k == "confirm":
            out.append("吸筹没到明确吸筹这条线")
        elif k == "clean_name":
            out.append("简称带 ST（含 *ST 与退市整理），这类票一律不买")
        else:
            # 兜底要让人看出"这是系统没写说明"，而不是让人去猜一个英文词是什么意思。
            # 内部代号退到括号里，句子主干仍是中文。
            out.append(f"还有一条买入条件没通过，暂时没写对应的说明（内部代号 {k}）")
    return out


def upside_text(upside, neg_tol: float = 0.0) -> str:
    """券商预期空间没过门槛时的那句话。两处用到，口径必须一样。

    不写成"预期空间 -5% 低于 +0% 的容忍线"，有两个原因。一是默认配置下容忍线就是 0，
    说"低于 0% 的容忍线"等于没说；二是两个数一个带正号一个带负号、方向相反，
    人要在脑子里翻一次才比得了大小。改成同向同单位：券商目标价比现价低多少，
    我们最多接受低多少。
    """
    if upside is None:
        return "算不出券商预期空间（券商没给目标价，或现价取不到）"
    gap = -float(upside)
    if float(neg_tol) > 0:
        return f"券商给的目标价比现价低 {gap:.0%}，最多只接受低 {float(neg_tol):.0%}"
    return f"券商给的目标价比现价低 {gap:.0%}，目标价低于现价的不买"


# ============================================================================
# 安全边际三情景的那一行（2026-09-07 下一阶段方案第四件）：只展示，不进判决
# ============================================================================

def _period_cn(q) -> str:
    """预测期写成人话：2026Q4 是 2026 年度，2026Q2 是 2026 年中期，认不出的原样。"""
    s = str(q or "").strip().upper()
    if len(s) == 6 and s[:4].isdigit() and s[4] == "Q":
        return f"{s[:4]} 年度" if s[5] == "4" else f"{s[:4]} 年{'一季' if s[5] == '1' else '中期' if s[5] == '2' else '三季'}"
    return s or "未知预测期"


def valuation_view(scn) -> str:
    """把三情景估值写成一句完整的话。数据缺席、四种不适用、算得出三种情形各有各的写法。

    赔率是中性上行对悲观下行，写成"0.87 比 1"——不到一比一就是说这个位置的赔率不吸引人。
    这一行回答的是"最坏情况下现价还有多少下跌空间"，是事前的安全边际；与建仓后的浮盈无关。
    """
    if not isinstance(scn, dict):
        return "安全边际：近三个月没有券商的盈利预测，算不出"
    if scn.get("na"):
        return f"安全边际算不出：{scn['na']}"
    head = f"安全边际（{_period_cn(scn.get('quarter'))}预测，{scn.get('firms')} 家）"
    body = (f"悲观 {scn['pess']:.2f} 元（{scn['down']:+.1%}）、中性 {scn['neut']:.2f} 元（{scn['up_neut']:+.1%}）、"
            f"乐观 {scn['opt']:.2f} 元（{scn['up_opt']:+.1%}）；隐含市盈率 {scn['implied_pe']:.1f} 倍")
    if scn.get("odds") is not None:
        tail = f"赔率 {scn['odds']:.2f} 比 1（中性上行对悲观下行）"
    else:
        tail = scn.get("note") or "赔率不成立"
    if scn.get("wide"):
        sp = scn.get("spread") or {}
        tail += (f"。机构分歧极大（每股收益最高是最低的 {sp.get('eps', 0):.1f} 倍、"
                 f"市盈率 {sp.get('pe', 0):.1f} 倍），两头的数字只当参考")
    return f"{head}：{body}；{tail}"


def valuation_short(scn) -> str:
    """表格里放得下的短写法：悲观下行 / 中性上行，赔率。"""
    if not isinstance(scn, dict):
        return "—"
    if scn.get("na"):
        return "算不出"
    odds = f"赔率 {scn['odds']:.2f}" if scn.get("odds") is not None else "赔率不成立"
    return f"{scn['down']:+.0%}/{scn['up_neut']:+.0%}，{odds}" + ("（分歧极大）" if scn.get("wide") else "")