444 lines
24 KiB
Python
444 lines
24 KiB
Python
"""逻辑状态四态：判断支撑一只票的研究证据还在不在（纯函数，不连库、不读配置）。
|
||
|
||
## 它回答什么，不回答什么
|
||
|
||
它不判断"这条产业逻辑对不对"——那是研究员的活，系统做不了也不该做。
|
||
它只看支撑逻辑的证据还在不在、有没有在走弱：预测在不在下修、覆盖在不在收缩、
|
||
有没有出现方向相反的论断、产业研判的采信倾向有没有翻过来。
|
||
|
||
与候选卡的三态是正交的两维，不合并。候选卡回答"今天要不要买"，是当日无状态重算的
|
||
入场判决；这里回答"支撑它的研究证据还在不在"，是跨日的持有期跟踪。两者不能合并还有
|
||
一个硬理由：持仓票在候选筛选里被整行剔掉且剔在判决检查之前，所以候选卡三态对持仓
|
||
天然无效，而持仓恰恰是逻辑四态唯一真正有用的地方。
|
||
|
||
## 四路证据
|
||
|
||
甲路 研报论断 方向受控必填（利好、利空、中性三选一，缺方向的断言落库时直接丢弃），
|
||
行业级环节级公司级三种粒度都有，随研报入库即时。中速。
|
||
乙路 产业研判 主题级。判据用采信倾向这个四选一的受控枚举（偏多、偏空、中性、证据
|
||
不足）的迁移，不用那段自由文本——判"上修还是分化"会退化成中文匹配。慢速。
|
||
丙路 券商行动 个股级周度：同财年同预测期的每股收益预测中位数变化，与覆盖机构数变化。
|
||
看的是券商的行动不是言辞——券商极少明说不看好某个行业。中速。
|
||
丁路 公司事件 目前不可用：全库事件断言里九成五的方向字段为空，有值的也是动作词
|
||
（减持、增持、预增）不是利好利空的语义判断。这一路照样出一个"缺失"，
|
||
让卡上看得见缺的是它，而不是静悄悄地不存在。
|
||
|
||
## 每路先归一成一个信号
|
||
|
||
向好、平稳、转弱、缺失，另加一个分歧——分歧是"这一路自己内部方向相反"，设计里把它
|
||
单列为证据矛盾的一个来源，所以它不能被压进那四个里的任何一个。
|
||
|
||
一条硬要求：每路必须带自己的截止日。没有截止日的证据一律按缺失处理，因为"这条证据
|
||
是哪天的"决定了它还算不算数，不知道就不能采信。
|
||
|
||
## 合成的四条规则
|
||
|
||
一，每路各自归一，各带截止日与覆盖计数。
|
||
二，负向优先：任何一路转弱，整体最多到无法判断；只有两条路径允许整体落到逻辑存疑，
|
||
见 DOUBT_TRIGGERS。
|
||
三，缺失既不是负面也不是正面证据，从合成里剔除并记明缺哪一路。缺省态永远是
|
||
无法判断加证据不足——从来没有行业观点、券商也算不出来的票，不算逻辑成立。
|
||
四，慢信号定底色、中速定当前态、日度事件只作一次性复核触发不能单独定态。抖动不靠
|
||
平滑系数，靠三件事：等长窗口本身就是低通、状态迁移要求连续若干计划日同向、
|
||
退出负态比进入负态慢（settle）。
|
||
|
||
## 阈值
|
||
|
||
全部由调用方传入，一次定死，每个在台账里写明与哪个既有口径同源，绝不按复盘收益读数
|
||
回调——那是在历史样本上挑参数。本模块只给默认值，不读配置文件。
|
||
|
||
离线单测见 test_logic_state.py，不连库。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
|
||
# ---- 四态 ----
|
||
STATE_STRONG = "逻辑强化"
|
||
STATE_HOLD = "逻辑成立"
|
||
STATE_UNKNOWN = "无法判断"
|
||
STATE_DOUBT = "逻辑存疑"
|
||
|
||
# 无法判断的两个子因，动作不同：证据矛盾进人工复核清单（提示"需要裁决"并列出矛盾两路的
|
||
# 日期与出处）；证据不足不进清单，只在卡上标注缺哪一路（提示"需要补材料"）。
|
||
WHY_THIN = "证据不足"
|
||
WHY_CONFLICT = "证据矛盾"
|
||
|
||
# ---- 每路的信号 ----
|
||
SIG_UP = "向好"
|
||
SIG_FLAT = "平稳"
|
||
SIG_DOWN = "转弱"
|
||
SIG_NONE = "缺失"
|
||
SIG_SPLIT = "分歧"
|
||
|
||
# ---- 四路 ----
|
||
PATH_CLAIM = "研报论断"
|
||
PATH_JUDGE = "产业研判"
|
||
PATH_BROKER = "券商行动"
|
||
PATH_EVENT = "公司事件"
|
||
|
||
# 每路的节奏，合成规则四要用。慢信号不能单独定态。
|
||
SPEED = {PATH_JUDGE: "慢", PATH_CLAIM: "中", PATH_BROKER: "中", PATH_EVENT: "日"}
|
||
|
||
# 只有这两条路径的转弱允许整体落到逻辑存疑，别的转弱最多到无法判断。
|
||
DOUBT_TRIGGERS = {
|
||
PATH_BROKER: "每股收益预测下修且覆盖机构收缩，两个条件同时满足",
|
||
PATH_CLAIM: "论断方向跨期翻转（先前利好、最新一条转利空）",
|
||
PATH_JUDGE: "采信倾向从偏多迁移到偏空，且材料指纹变过",
|
||
}
|
||
|
||
# 采信倾向的受控四值（数据基座抽取端的提示词写死的，见 backend/workers/tasks.py）。
|
||
LEAN_BULL = "偏多"
|
||
LEAN_BEAR = "偏空"
|
||
LEAN_NEUTRAL = "中性"
|
||
LEAN_THIN = "证据不足"
|
||
|
||
|
||
def signal(path: str, sig: str, *, as_of: str | None = None, coverage: int = 0,
|
||
why: str = "", hard: bool = False, refs=None) -> dict:
|
||
"""一路证据归一后的形状。
|
||
|
||
hard 表示这一路的转弱达到了逻辑存疑的进入条件（见 DOUBT_TRIGGERS）；转弱而 hard
|
||
为假，整体最多到无法判断。没有截止日的证据一律降为缺失——不知道是哪天的证据，
|
||
就不知道它还算不算数。
|
||
"""
|
||
if sig != SIG_NONE and not as_of:
|
||
return {"path": path, "signal": SIG_NONE, "as_of": None, "coverage": 0,
|
||
"why": "这一路没有截止日，不能采信", "hard": False, "refs": []}
|
||
return {"path": path, "signal": sig, "as_of": as_of, "coverage": int(coverage or 0),
|
||
"why": why, "hard": bool(hard and sig == SIG_DOWN), "refs": list(refs or [])}
|
||
|
||
|
||
# ============================================================================
|
||
# 甲路：研报因果论断
|
||
# ============================================================================
|
||
|
||
def from_claims(claims, ds: str, *, fresh_days: int = 30, stale_days: int = 90) -> dict:
|
||
"""把一只票的因果论断归一成一路信号。
|
||
|
||
两个窗口，各管一件事：
|
||
|
||
fresh_days 新材料窗口。向好要求这个窗口里有新的利好论断——强化说的是"变好了"
|
||
而不是"现在是好的"，只有新材料才算变化。默认 30 天，与候选卡确认线的
|
||
三十个交易日在同一量级；这里取自然日，因为披露日本来就是自然日。
|
||
stale_days 还算数的上界。全部论断都超过它，这一路按缺失处理（陈旧等同于没有，
|
||
不是负面证据——规则三）。默认 90 天，与候选卡的论断陈旧线同源。
|
||
|
||
跨期方向翻转是近似判定：按披露日排序，最新一条转利空而在它之前的论断以利好为主，
|
||
就算翻转。真正的跨期改口判定在数据基座的融合层，那里能对上同一对主客体；这里只有
|
||
按票聚起来的论断，对不上主客体，所以这是个偏宽的近似——它会把"讲的不是同一件事的
|
||
两条论断"也算成翻转。之所以仍然用它：翻转只是把状态压到逻辑存疑，而逻辑存疑的动作
|
||
是出减持提议强制交人裁决，绝不自动卖，误判的代价是多一次人工复核。
|
||
"""
|
||
rows = [c for c in (claims or []) if isinstance(c, dict) and c.get("disclosure_date")]
|
||
if not rows:
|
||
return signal(PATH_CLAIM, SIG_NONE, why="没有研报论断")
|
||
rows.sort(key=lambda c: c["disclosure_date"])
|
||
latest = rows[-1]["disclosure_date"]
|
||
age = _gap(latest, ds)
|
||
if age is None:
|
||
return signal(PATH_CLAIM, SIG_NONE, why="论断的披露日认不出来")
|
||
if age > int(stale_days):
|
||
return signal(PATH_CLAIM, SIG_NONE, coverage=len(rows),
|
||
why=f"{len(rows)} 条论断最新一条已过 {age} 天，按缺失处理")
|
||
|
||
fresh = [c for c in rows if (_gap(c["disclosure_date"], ds) or 10 ** 6) <= int(fresh_days)]
|
||
up = [c for c in fresh if c.get("direction") == "利好"]
|
||
down = [c for c in fresh if c.get("direction") == "利空"]
|
||
refs = _refs(fresh or rows)
|
||
|
||
if up and down:
|
||
return signal(PATH_CLAIM, SIG_SPLIT, as_of=latest, coverage=len(fresh), refs=refs,
|
||
why=f"{fresh_days} 天内同时有 {len(up)} 条利好与 {len(down)} 条利空论断，方向分歧")
|
||
if down:
|
||
flipped = _flipped(rows)
|
||
why = (f"{fresh_days} 天内有 {len(down)} 条利空论断"
|
||
+ ("，且在此之前的论断以利好为主，方向跨期翻转" if flipped else ""))
|
||
return signal(PATH_CLAIM, SIG_DOWN, as_of=latest, coverage=len(fresh),
|
||
why=why, hard=flipped, refs=refs)
|
||
if up:
|
||
return signal(PATH_CLAIM, SIG_UP, as_of=latest, coverage=len(fresh), refs=refs,
|
||
why=f"{fresh_days} 天内有 {len(up)} 条新的利好论断")
|
||
return signal(PATH_CLAIM, SIG_FLAT, as_of=latest, coverage=len(rows), refs=_refs(rows),
|
||
why=f"{len(rows)} 条论断都还在 {stale_days} 天之内，但 {fresh_days} 天内没有新材料")
|
||
|
||
|
||
def _flipped(rows_sorted) -> bool:
|
||
"""最新一条转利空，而在它之前（严格早于它的披露日）的论断以利好为主。"""
|
||
last = rows_sorted[-1]
|
||
if last.get("direction") != "利空":
|
||
return False
|
||
before = [c for c in rows_sorted if c["disclosure_date"] < last["disclosure_date"]]
|
||
if not before:
|
||
return False
|
||
return sum(1 for c in before if c.get("direction") == "利好") > \
|
||
sum(1 for c in before if c.get("direction") == "利空")
|
||
|
||
|
||
def _refs(rows, limit: int = 3) -> list:
|
||
out = []
|
||
for c in rows[-limit:][::-1]:
|
||
out.append({"date": c.get("disclosure_date"), "doc": c.get("doc_title"),
|
||
"id": c.get("claim_id"), "direction": c.get("direction"),
|
||
"text": c.get("mechanism")})
|
||
return out
|
||
|
||
|
||
# ============================================================================
|
||
# 乙路：产业研判
|
||
# ============================================================================
|
||
|
||
def from_judgement(cur, prev=None, *, stale_days: int = 60) -> dict:
|
||
"""把一个主题的产业研判快照归一成一路信号。cur 与 prev 是 judgement.build_rows 的行。
|
||
|
||
判的是采信倾向这个受控枚举的迁移，不是它当前的取值。理由是强化说的是"变好了"：
|
||
一个一直偏多的主题每天都算向好的话，强化会变得到处都是，也就不再说明任何事。
|
||
所以稳定偏多归平稳——它是有效的非负面证据，够撑住逻辑成立，但撑不起逻辑强化。
|
||
|
||
只有材料指纹变过才算一次迁移。指纹没变而日子变老，只累加陈旧天数；超过 stale_days
|
||
按缺失处理，因为产业研判是人工点按钮触发的，长期不变说明没人重新看过它，
|
||
不是"看过了没变"。
|
||
"""
|
||
if not cur:
|
||
return signal(PATH_JUDGE, SIG_NONE, why="这个环节对不上任何产业研判主题")
|
||
as_of = _s(cur.get("review_date")) or _s(cur.get("plan_date"))
|
||
cov = int(cur.get("n_materials") or 0)
|
||
if cur.get("verified") == 0:
|
||
return signal(PATH_JUDGE, SIG_NONE, coverage=cov, why="产业研判的自我校验未通过")
|
||
lean = _s(cur.get("leaning"))
|
||
if lean == LEAN_THIN:
|
||
return signal(PATH_JUDGE, SIG_NONE, coverage=cov, why="产业研判自己给的是证据不足")
|
||
stale = int(cur.get("stale_days") or 0)
|
||
migrated = cur.get("migrated")
|
||
if migrated != 1 and stale > int(stale_days):
|
||
return signal(PATH_JUDGE, SIG_NONE, coverage=cov,
|
||
why=f"产业研判的材料已 {stale} 天没变过，按缺失处理")
|
||
|
||
prev_lean = _s(cur.get("leaning_prev")) or (_s(prev.get("leaning")) if prev else None)
|
||
refs = [{"theme": _s(cur.get("subject_name")) or _s(cur.get("cluster_key")),
|
||
"leaning": lean, "leaning_prev": prev_lean, "date": as_of,
|
||
"n_bull": cur.get("n_bull"), "n_bear": cur.get("n_bear")}]
|
||
if migrated == 1 and prev_lean and prev_lean != lean:
|
||
if prev_lean == LEAN_BULL and lean == LEAN_BEAR:
|
||
return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, hard=True, refs=refs,
|
||
why="产业研判的采信倾向从偏多迁移到偏空")
|
||
if lean == LEAN_BEAR:
|
||
return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, refs=refs,
|
||
why=f"产业研判的采信倾向从{prev_lean}迁移到偏空")
|
||
if lean == LEAN_BULL:
|
||
return signal(PATH_JUDGE, SIG_UP, as_of=as_of, coverage=cov, refs=refs,
|
||
why=f"产业研判的采信倾向从{prev_lean}迁移到偏多")
|
||
return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs,
|
||
why=f"产业研判的采信倾向从{prev_lean}迁移到{lean}，不是多空方向的变化")
|
||
if lean == LEAN_BEAR:
|
||
return signal(PATH_JUDGE, SIG_DOWN, as_of=as_of, coverage=cov, refs=refs,
|
||
why="产业研判的采信倾向是偏空")
|
||
return signal(PATH_JUDGE, SIG_FLAT, as_of=as_of, coverage=cov, refs=refs,
|
||
why=f"产业研判的采信倾向是{lean}，与上一版相同")
|
||
|
||
|
||
# ============================================================================
|
||
# 丙路：券商行动
|
||
# ============================================================================
|
||
|
||
def from_broker(now, prev, *, as_of: str | None = None, min_firms: int = 2,
|
||
eps_drop: float = 0.05, eps_rise: float = 0.05,
|
||
cover_drop: float = 0.30, cover_min_base: int = 5) -> dict:
|
||
"""把同一财年同一预测期的每股收益预测中位数与覆盖机构数，归一成一路信号。
|
||
|
||
now 与 prev 是两个等长窗口的读数，各是 {"eps": 中位数, "firms": 机构数}。
|
||
两个窗口必须等长——窗口不等长会让八成的票假显示覆盖收缩（实测：前 135 天对近 45 天
|
||
时，907 只票误报）。等长本身也是抗抖动的低通，见合成规则四。
|
||
|
||
每股收益从正转负或从负转正时，变化率没有意义，单独标记为转亏或扭亏而不算百分比。
|
||
转亏方向明确是坏消息，但它不满足逻辑存疑那个双条件，所以只是转弱不带 hard。
|
||
|
||
覆盖收缩要同时满足比例与基数两个条件才算数。基数门槛 cover_min_base 是 2026-09-04
|
||
全市场实测加进来的：不加它，六只判为逻辑存疑的票里有五只的触发是覆盖从三家减到两家，
|
||
也就是一家机构那两周没发报告就构成 33% 的收缩。机构数这么少时百分比不说明任何事。
|
||
默认 5 家的性质是——前窗口有五家时，走掉一家只有 20%、够不着 30% 这条线，必须走掉
|
||
两家才算，这正是"覆盖在收缩"该有的含义。基数不够时预测下修照样记转弱，只是不带
|
||
hard，于是整体落到无法判断进人工复核，而不是出减持提议。
|
||
"""
|
||
n0, n1 = int((prev or {}).get("firms") or 0), int((now or {}).get("firms") or 0)
|
||
if n0 < int(min_firms) or n1 < int(min_firms):
|
||
return signal(PATH_BROKER, SIG_NONE, coverage=min(n0, n1),
|
||
why=f"两个窗口里能算的机构不足 {min_firms} 家（前 {n0} 家、后 {n1} 家）")
|
||
e0, e1 = (prev or {}).get("eps"), (now or {}).get("eps")
|
||
if e0 is None or e1 is None:
|
||
return signal(PATH_BROKER, SIG_NONE, coverage=min(n0, n1), why="每股收益预测中位数缺失")
|
||
shrink = (n0 - n1) / n0 if n0 else 0.0
|
||
refs = [{"eps_prev": e0, "eps_now": e1, "firms_prev": n0, "firms_now": n1, "date": as_of}]
|
||
|
||
if (e0 > 0) != (e1 > 0):
|
||
turned = e1 <= 0
|
||
return signal(PATH_BROKER, SIG_DOWN if turned else SIG_UP, as_of=as_of,
|
||
coverage=n1, refs=refs,
|
||
why=("每股收益预测由盈转亏，变化率不适用" if turned
|
||
else "每股收益预测由亏转盈，变化率不适用"))
|
||
chg = (e1 - e0) / abs(e0) if e0 else 0.0
|
||
if chg <= -float(eps_drop):
|
||
shrunk = shrink > float(cover_drop)
|
||
hard = shrunk and n0 >= int(cover_min_base)
|
||
why = f"每股收益预测中位数下修 {abs(chg):.0%}"
|
||
if hard:
|
||
why += f"，同时覆盖机构从 {n0} 家收缩到 {n1} 家（减 {shrink:.0%}）"
|
||
elif shrunk:
|
||
why += (f"，覆盖机构从 {n0} 家减到 {n1} 家，但前窗口不足 {cover_min_base} 家，"
|
||
f"这个比例不作数")
|
||
return signal(PATH_BROKER, SIG_DOWN, as_of=as_of, coverage=n1, hard=hard,
|
||
why=why, refs=refs)
|
||
if chg >= float(eps_rise):
|
||
return signal(PATH_BROKER, SIG_UP, as_of=as_of, coverage=n1, refs=refs,
|
||
why=f"每股收益预测中位数上修 {chg:.0%}")
|
||
return signal(PATH_BROKER, SIG_FLAT, as_of=as_of, coverage=n1, refs=refs,
|
||
why=f"每股收益预测中位数变化 {chg:+.0%}，在阈值之内")
|
||
|
||
|
||
# ============================================================================
|
||
# 丁路：公司事件（目前不可用）
|
||
# ============================================================================
|
||
|
||
def from_events(*_a, **_k) -> dict:
|
||
"""这一路目前不可用，恒定返回缺失。
|
||
|
||
照样出一个缺失而不是干脆不实现，是为了让卡上看得见缺的是它：缺失要分得清
|
||
"无数据源"和"读不到"，静悄悄地不存在会让人以为系统判过了。
|
||
|
||
不可用有两个结构性原因，都不是接一下就能解决的。一是方向缺失：全库三十八万条
|
||
事件断言里三十六万五千条的方向字段为空，占九成五，有值的也是动作词（减持、增持、
|
||
预增、预减、回购、授予）不是利好利空的语义判断。二是粒度错位：十类事件闭集全部是
|
||
公司级的，没有一类是行业级催化，而事件谓词的客体类型受控集只有公司与实体、抽取时
|
||
主体也写死成公司类型，所以行业级催化在结构上不可能从这一路出来——它走甲路。
|
||
"""
|
||
return signal(PATH_EVENT, SIG_NONE, why="公司事件这一路目前无数据源（方向字段九成五为空）")
|
||
|
||
|
||
# ============================================================================
|
||
# 合成
|
||
# ============================================================================
|
||
|
||
def compose(paths, *, strong_needs=(PATH_CLAIM, PATH_JUDGE)) -> dict:
|
||
"""把各路信号合成一个状态。返回 state / why / paths / usable / missing / reasons。
|
||
|
||
strong_needs 是"逻辑强化必须包含其中至少一路"的名单，默认是甲路与乙路——
|
||
强化不能只靠券商预测上修单独成立，那只是卖方在调数字，不是产业逻辑变好了。
|
||
"""
|
||
rows = [p for p in (paths or []) if isinstance(p, dict) and p.get("path")]
|
||
usable = [p for p in rows if p["signal"] != SIG_NONE]
|
||
missing = [p for p in rows if p["signal"] == SIG_NONE]
|
||
miss_note = [f"{p['path']}：{p['why']}" for p in missing]
|
||
|
||
if not usable:
|
||
return _out(STATE_UNKNOWN, WHY_THIN, rows, usable, missing,
|
||
["四路证据一路都不可用"] + miss_note)
|
||
|
||
ups = [p for p in usable if p["signal"] == SIG_UP]
|
||
downs = [p for p in usable if p["signal"] == SIG_DOWN]
|
||
splits = [p for p in usable if p["signal"] == SIG_SPLIT]
|
||
|
||
# 规则二，负向优先。分歧与"一路向好另一路转弱"都是证据矛盾，动作是交人裁决。
|
||
if splits or (ups and downs):
|
||
rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in splits + ups + downs]
|
||
return _out(STATE_UNKNOWN, WHY_CONFLICT, rows, usable, missing,
|
||
["不同证据指向相反，需要裁决"] + rs + miss_note)
|
||
if downs:
|
||
hard = [p for p in downs if p.get("hard")]
|
||
rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in downs]
|
||
if hard:
|
||
return _out(STATE_DOUBT, None, rows, usable, missing,
|
||
[f"{p['path']}的转弱达到进入条件：{DOUBT_TRIGGERS.get(p['path'], '')}"
|
||
for p in hard] + rs + miss_note)
|
||
return _out(STATE_UNKNOWN, WHY_CONFLICT, rows, usable, missing,
|
||
["有证据在走弱，但没到逻辑存疑的进入条件"] + rs + miss_note)
|
||
|
||
# 规则四，慢信号不能单独定态。
|
||
if all(SPEED.get(p["path"]) == "慢" for p in usable):
|
||
return _out(STATE_UNKNOWN, WHY_THIN, rows, usable, missing,
|
||
["只剩产业研判这一路慢信号，不足以单独定态"] + miss_note)
|
||
|
||
if len(ups) >= 2 and any(p["path"] in strong_needs for p in ups):
|
||
rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in ups]
|
||
return _out(STATE_STRONG, None, rows, usable, missing, rs + miss_note)
|
||
|
||
rs = [f"{p['path']}（{p['as_of']}）：{p['why']}" for p in usable]
|
||
return _out(STATE_HOLD, None, rows, usable, missing, rs + miss_note)
|
||
|
||
|
||
def _out(state, why, rows, usable, missing, reasons) -> dict:
|
||
return {"state": state, "why": why, "paths": rows,
|
||
"usable": [p["path"] for p in usable], "missing": [p["path"] for p in missing],
|
||
"reasons": reasons,
|
||
"as_of": max([p["as_of"] for p in usable if p.get("as_of")], default=None)}
|
||
|
||
|
||
def settle(prev_state, raw_state, recent_raw, *, confirm_days: int = 2,
|
||
exit_doubt_days: int = 3) -> tuple:
|
||
"""状态迁移的抗抖动与进出不对称。返回（落定的状态，一句话说明）。
|
||
|
||
recent_raw 是含今天在内、最近若干个计划日的原始合成状态，最新的排在最后。
|
||
|
||
进入逻辑存疑即刻成立，退出要求连续 exit_doubt_days 个计划日都不再存疑——
|
||
这个不对称是有意的：漏判一次证据走弱的代价，比多留一次人工复核大得多。
|
||
其余方向的迁移要求连续 confirm_days 个计划日同向，避免一天一变。
|
||
"""
|
||
seq = [s for s in (recent_raw or []) if s]
|
||
if raw_state == STATE_DOUBT:
|
||
return STATE_DOUBT, "进入逻辑存疑即刻成立"
|
||
if prev_state == STATE_DOUBT:
|
||
tail = seq[-int(exit_doubt_days):]
|
||
if len(tail) >= int(exit_doubt_days) and all(s != STATE_DOUBT for s in tail):
|
||
return raw_state, f"连续 {exit_doubt_days} 个计划日不再存疑，退出逻辑存疑"
|
||
got = sum(1 for s in tail if s != STATE_DOUBT)
|
||
return STATE_DOUBT, f"退出逻辑存疑要连续 {exit_doubt_days} 个计划日不存疑，现在 {got} 天"
|
||
if prev_state is None or raw_state == prev_state:
|
||
return raw_state, "维持"
|
||
tail = seq[-int(confirm_days):]
|
||
if len(tail) >= int(confirm_days) and all(s == raw_state for s in tail):
|
||
return raw_state, f"连续 {confirm_days} 个计划日同向，迁移到{raw_state}"
|
||
return prev_state, f"迁移要连续 {confirm_days} 个计划日同向，今天先维持{prev_state}"
|
||
|
||
|
||
def apply_to_card(verdict: str, state: str) -> dict:
|
||
"""逻辑状态怎么作用于候选卡的判决。收敛规则是单调的，两条都不许越界：
|
||
|
||
逻辑强化只能把卡内序提前、在提议卡上多写一行，永远不能把判决往上升一档。
|
||
逻辑存疑只能改分流通道，不能把仅展示变成可执行。
|
||
|
||
真正需要新语义的只有一格：候选加逻辑存疑，含义是可交易口径成立而研究证据在走弱，
|
||
这正是杀逻辑与杀估值那条边界。处置是强制人工确认，复用现有的强制确认字段、与关注
|
||
判决走同一条队列，页面文案区分来源。
|
||
"""
|
||
out = {"verdict": verdict, "rank_bonus": 0, "force_confirm": False,
|
||
"pause_add": False, "note": ""}
|
||
if state == STATE_STRONG:
|
||
out["rank_bonus"] = 1
|
||
out["note"] = "研究证据在走强，卡内序提前；判决不变"
|
||
elif state == STATE_DOUBT:
|
||
out["pause_add"] = True
|
||
if verdict == "候选":
|
||
out["force_confirm"] = True
|
||
out["note"] = "可交易口径成立而研究证据在走弱，强制人工确认；停增持侧自主动作"
|
||
else:
|
||
out["note"] = "研究证据在走弱，停增持侧自主动作；判决不变"
|
||
elif state == STATE_UNKNOWN:
|
||
out["note"] = "研究证据不足以定论；判决不变"
|
||
return out
|
||
|
||
|
||
# ---- 小工具 ----
|
||
|
||
def _s(v):
|
||
if v is None:
|
||
return None
|
||
s = str(v).strip()
|
||
return s or None
|
||
|
||
|
||
def _gap(a: str, b: str):
|
||
"""a 到 b 的自然日数；认不出返回 None。"""
|
||
try:
|
||
return (dt.date.fromisoformat(str(b)[:10]) - dt.date.fromisoformat(str(a)[:10])).days
|
||
except (ValueError, TypeError):
|
||
return None
|