146 lines
6.7 KiB
Python
146 lines
6.7 KiB
Python
"""候选卡：把一只票的各条证据线装配成判决与理由（纯函数，零 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))
|