From 2a400f68af3affedd8081f554cac3a5133d8a344 Mon Sep 17 00:00:00 2001 From: zlt Date: Fri, 4 Sep 2026 09:09:15 +0800 Subject: [PATCH] =?UTF-8?q?=E9=80=BB=E8=BE=91=E7=8A=B6=E6=80=81=E5=9B=9B?= =?UTF-8?q?=E6=80=81=EF=BC=9A=E5=90=88=E6=88=90=E6=A8=A1=E5=9D=97=E4=B8=8E?= =?UTF-8?q?=E7=A6=BB=E7=BA=BF=E5=8D=95=E6=B5=8B=EF=BC=88=E5=8F=AA=E5=86=99?= =?UTF-8?q?=E4=B8=8D=E6=8E=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 判断的不是"这条产业逻辑对不对"——那是研究员的活；只判断支撑它的证据还在不在。 四路证据各自归一成一个信号（向好、平稳、转弱、缺失，另加分歧表示这一路自己内部 方向相反）：甲路研报论断（方向受控必填，两个窗口——新材料窗口决定向好、陈旧线决定 还算不算数）、乙路产业研判（判采信倾向这个受控四值的迁移，不判自由文本）、丙路券商 行动（等长窗口的每股收益预测中位数与覆盖机构数）、丁路公司事件（恒定缺失，方向字段 九成五为空，照样出一个缺失让卡上看得见缺的是它）。 合成四条规则都落成了断言：每路必须带截止日（没有就降为缺失）、负向优先（转弱最多到 无法判断，只有两条路径允许落到逻辑存疑）、缺失既不算负面也不算正面（剔出合成但记名）、 慢信号不能单独定态。另有进出不对称：进入逻辑存疑即刻成立，退出要连续三个计划日不再 存疑；漏判一次证据走弱的代价比多留一次人工复核大得多。 作用于候选卡是单调的：逻辑强化只能提前卡内序、永远不升判决；逻辑存疑只能改分流通道、 不能把仅展示变成可执行。唯一的新语义是候选加逻辑存疑——可交易口径成立而研究证据在 走弱，正是杀逻辑与杀估值那条边界，处置是强制人工确认。 本次只写不接进计划装配：乙路依赖行业观点快照攒够几天才有迁移可判。 测试 59 例通过，不连库。 Co-Authored-By: Claude Opus 5 --- logic_state.py | 432 ++++++++++++++++++++++++++++++++++++++++++++ test_logic_state.py | 244 +++++++++++++++++++++++++ 2 files changed, 676 insertions(+) create mode 100644 logic_state.py create mode 100644 test_logic_state.py diff --git a/logic_state.py b/logic_state.py new file mode 100644 index 0000000..97071e8 --- /dev/null +++ b/logic_state.py @@ -0,0 +1,432 @@ +"""逻辑状态四态：判断支撑一只票的研究证据还在不在（纯函数，不连库、不读配置）。 + +## 它回答什么，不回答什么 + +它不判断"这条产业逻辑对不对"——那是研究员的活，系统做不了也不该做。 +它只看支撑逻辑的证据还在不在、有没有在走弱：预测在不在下修、覆盖在不在收缩、 +有没有出现方向相反的论断、产业研判的采信倾向有没有翻过来。 + +与候选卡的三态是正交的两维，不合并。候选卡回答"今天要不要买"，是当日无状态重算的 +入场判决；这里回答"支撑它的研究证据还在不在"，是跨日的持有期跟踪。两者不能合并还有 +一个硬理由：持仓票在候选筛选里被整行剔掉且剔在判决检查之前，所以候选卡三态对持仓 +天然无效，而持仓恰恰是逻辑四态唯一真正有用的地方。 + +## 四路证据 + + 甲路 研报论断 方向受控必填（利好、利空、中性三选一，缺方向的断言落库时直接丢弃）， + 行业级环节级公司级三种粒度都有，随研报入库即时。中速。 + 乙路 产业研判 主题级。判据用采信倾向这个四选一的受控枚举（偏多、偏空、中性、证据 + 不足）的迁移，不用那段自由文本——判"上修还是分化"会退化成中文匹配。慢速。 + 丙路 券商行动 个股级周度：同财年同预测期的每股收益预测中位数变化，与覆盖机构数变化。 + 看的是券商的行动不是言辞——券商极少明说不看好某个行业。中速。 + 丁路 公司事件 目前不可用：全库事件断言里九成五的方向字段为空，有值的也是动作词 + （减持、增持、预增）不是利好利空的语义判断。这一路照样出一个"缺失"， + 让卡上看得见缺的是它，而不是静悄悄地不存在。 + +## 每路先归一成一个信号 + +向好、平稳、转弱、缺失，另加一个分歧——分歧是"这一路自己内部方向相反"，设计里把它 +单列为证据矛盾的一个来源，所以它不能被压进那四个里的任何一个。 + +一条硬要求：每路必须带自己的截止日。没有截止日的证据一律按缺失处理，因为"这条证据 +是哪天的"决定了它还算不算数，不知道就不能采信。 + +## 合成的四条规则 + + 一，每路各自归一，各带截止日与覆盖计数。 + 二，负向优先：任何一路转弱，整体最多到无法判断；只有两条路径允许整体落到逻辑存疑， + 见 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) -> dict: + """把同一财年同一预测期的每股收益预测中位数与覆盖机构数，归一成一路信号。 + + now 与 prev 是两个等长窗口的读数，各是 {"eps": 中位数, "firms": 机构数}。 + 两个窗口必须等长——窗口不等长会让八成的票假显示覆盖收缩（实测：前 135 天对近 45 天 + 时，907 只票误报）。等长本身也是抗抖动的低通，见合成规则四。 + + 每股收益从正转负或从负转正时，变化率没有意义，单独标记为转亏或扭亏而不算百分比。 + 转亏方向明确是坏消息，但它不满足逻辑存疑那个双条件，所以只是转弱不带 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): + hard = shrink > float(cover_drop) + why = f"每股收益预测中位数下修 {abs(chg):.0%}" + if hard: + why += f"，同时覆盖机构从 {n0} 家收缩到 {n1} 家（减 {shrink:.0%}）" + 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 diff --git a/test_logic_state.py b/test_logic_state.py new file mode 100644 index 0000000..406d43b --- /dev/null +++ b/test_logic_state.py @@ -0,0 +1,244 @@ +"""logic_state 的离线单测（不连库、不读配置）。 + +覆盖：四路各自的归一（甲路两个窗口与跨期翻转、乙路迁移与三种缺失、丙路双条件与 +盈亏翻转、丁路恒缺失）、合成四条规则（截止日必填、负向优先、缺失不算证据、慢信号 +不能单独定态）、逻辑强化的两个限制（要两路且必须含甲或乙）、状态迁移的进出不对称、 +以及作用于候选卡时的单调性（强化不升判决、存疑不把仅展示变成可执行）。 + +跑法：python3 test_logic_state.py 或 pytest test_logic_state.py +""" +import logic_state as ls + +DS = "2026-09-03" + + +def t(name, cond): + assert cond, name + print(" ok", name) + + +def claim(date, direction="利好", **kw): + d = {"disclosure_date": date, "direction": direction, "mechanism": "机制", + "doc_title": "研报", "claim_id": "c" + date.replace("-", "")} + d.update(kw) + return d + + +def test_claims(): + print("甲路 研报论断") + t("没有论断 -> 缺失", ls.from_claims([], DS)["signal"] == ls.SIG_NONE) + t("论断没有披露日 -> 缺失", + ls.from_claims([{"direction": "利好"}], DS)["signal"] == ls.SIG_NONE) + + r = ls.from_claims([claim("2026-02-01")], DS) + t("全部超陈旧线 -> 缺失（陈旧等同于没有，不是负面证据）", + r["signal"] == ls.SIG_NONE and "已过" in r["why"]) + + r = ls.from_claims([claim("2026-08-25")], DS) + t("新材料窗口内的利好 -> 向好", r["signal"] == ls.SIG_UP and r["as_of"] == "2026-08-25") + t("向好带截止日与出处", r["coverage"] == 1 and r["refs"][0]["id"] == "c20260825") + + r = ls.from_claims([claim("2026-07-01")], DS) + t("还在陈旧线内但没有新材料 -> 平稳（强化说的是变好了，不是现在是好的）", + r["signal"] == ls.SIG_FLAT) + + r = ls.from_claims([claim("2026-08-25"), claim("2026-08-28", "利空")], DS) + t("新材料窗口内多空同现 -> 分歧", r["signal"] == ls.SIG_SPLIT) + + r = ls.from_claims([claim("2026-08-28", "利空")], DS) + t("窗口内只有利空、之前没有论断 -> 转弱但不到存疑门槛", + r["signal"] == ls.SIG_DOWN and not r["hard"]) + + r = ls.from_claims([claim("2026-07-10"), claim("2026-07-20"), claim("2026-08-28", "利空")], DS) + t("先前以利好为主、最新一条转利空 -> 转弱且达到存疑门槛", + r["signal"] == ls.SIG_DOWN and r["hard"] and "跨期翻转" in r["why"]) + + r = ls.from_claims([claim("2026-07-10", "利空"), claim("2026-08-28", "利空")], DS) + t("先前就是利空、最新还是利空 -> 转弱但不算翻转", r["signal"] == ls.SIG_DOWN and not r["hard"]) + + r = ls.from_claims([claim("2026-08-01")], DS, fresh_days=60) + t("新材料窗口旋钮生效", r["signal"] == ls.SIG_UP) + r = ls.from_claims([claim("2026-02-01")], DS, stale_days=365) + t("陈旧线旋钮生效", r["signal"] == ls.SIG_FLAT) + + +def jrow(leaning=ls.LEAN_BULL, prev=None, migrated=0, stale=3, verified=1, **kw): + d = {"leaning": leaning, "leaning_prev": prev, "migrated": migrated, "stale_days": stale, + "verified": verified, "review_date": "2026-08-19", "n_materials": 35, + "subject_name": "钨产业链", "n_bull": 3, "n_bear": 3} + d.update(kw) + return d + + +def test_judgement(): + print("乙路 产业研判") + t("对不上任何主题 -> 缺失", ls.from_judgement(None)["signal"] == ls.SIG_NONE) + r = ls.from_judgement(jrow(verified=0)) + t("自我校验未通过 -> 缺失", r["signal"] == ls.SIG_NONE and "自我校验" in r["why"]) + r = ls.from_judgement(jrow(leaning=ls.LEAN_THIN)) + t("研判自己给的就是证据不足 -> 缺失", r["signal"] == ls.SIG_NONE) + r = ls.from_judgement(jrow(stale=90)) + t("材料长期没变过 -> 缺失（人工触发，不变说明没人重看过）", + r["signal"] == ls.SIG_NONE and "没变过" in r["why"]) + r = ls.from_judgement(jrow(stale=90, migrated=1)) + t("指纹刚变过就不算陈旧", r["signal"] != ls.SIG_NONE) + + t("稳定偏多 -> 平稳，不是向好", + ls.from_judgement(jrow())["signal"] == ls.SIG_FLAT) + r = ls.from_judgement(jrow(leaning=ls.LEAN_BULL, prev=ls.LEAN_NEUTRAL, migrated=1)) + t("采信倾向迁移到偏多 -> 向好", r["signal"] == ls.SIG_UP) + r = ls.from_judgement(jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1)) + t("偏多迁移到偏空 -> 转弱且达到存疑门槛", r["signal"] == ls.SIG_DOWN and r["hard"]) + r = ls.from_judgement(jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_NEUTRAL, migrated=1)) + t("中性迁移到偏空 -> 转弱但不到存疑门槛", r["signal"] == ls.SIG_DOWN and not r["hard"]) + r = ls.from_judgement(jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=0)) + t("指纹没变就不算迁移，当前偏空仍记转弱但不是硬触发", + r["signal"] == ls.SIG_DOWN and not r["hard"]) + r = ls.from_judgement(jrow(leaning=ls.LEAN_NEUTRAL, prev=ls.LEAN_BULL, migrated=1)) + t("迁到中性 -> 平稳（不是多空方向的变化）", r["signal"] == ls.SIG_FLAT) + + +def test_broker(): + print("丙路 券商行动") + r = ls.from_broker({"eps": 2.3, "firms": 1}, {"eps": 2.4, "firms": 5}, as_of=DS) + t("任一窗口机构不足两家 -> 缺失", r["signal"] == ls.SIG_NONE) + r = ls.from_broker({"eps": 2.30, "firms": 5}, {"eps": 2.34, "firms": 5}, as_of=DS) + t("变化在阈值之内 -> 平稳", r["signal"] == ls.SIG_FLAT) + r = ls.from_broker({"eps": 2.6, "firms": 5}, {"eps": 2.0, "firms": 5}, as_of=DS) + t("只有预测上修 -> 向好", r["signal"] == ls.SIG_UP) + r = ls.from_broker({"eps": 1.8, "firms": 5}, {"eps": 2.4, "firms": 5}, as_of=DS) + t("只下修、覆盖没收缩 -> 转弱但不到存疑门槛", + r["signal"] == ls.SIG_DOWN and not r["hard"]) + r = ls.from_broker({"eps": 1.8, "firms": 3}, {"eps": 2.4, "firms": 9}, as_of=DS) + t("下修且覆盖收缩，两个条件同现 -> 转弱且达到存疑门槛", + r["signal"] == ls.SIG_DOWN and r["hard"] and "收缩" in r["why"]) + r = ls.from_broker({"eps": 2.4, "firms": 3}, {"eps": 2.4, "firms": 9}, as_of=DS) + t("只覆盖收缩而预测没动 -> 平稳（覆盖只作否决位，不单独定负）", + r["signal"] == ls.SIG_FLAT) + r = ls.from_broker({"eps": -0.3, "firms": 5}, {"eps": 0.8, "firms": 5}, as_of=DS) + t("由盈转亏 -> 转弱，不给百分比", r["signal"] == ls.SIG_DOWN and "不适用" in r["why"]) + r = ls.from_broker({"eps": 0.5, "firms": 5}, {"eps": -0.4, "firms": 5}, as_of=DS) + t("由亏转盈 -> 向好，不给百分比", r["signal"] == ls.SIG_UP and "不适用" in r["why"]) + t("丁路恒定缺失", ls.from_events()["signal"] == ls.SIG_NONE) + + +def sig(path, s, hard=False, as_of=DS): + return ls.signal(path, s, as_of=as_of, coverage=3, why="测试", hard=hard) + + +def test_compose(): + print("合成四条规则") + r = ls.compose([]) + t("一路都没有 -> 无法判断加证据不足（缺省态）", + r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_THIN) + r = ls.compose([ls.from_events()]) + t("只有一路不可用 -> 仍是证据不足，且记明缺哪一路", + r["state"] == ls.STATE_UNKNOWN and r["missing"] == [ls.PATH_EVENT]) + + # 规则一：没有截止日的证据不能采信 + bad = ls.signal(ls.PATH_BROKER, ls.SIG_UP, as_of=None, coverage=9, why="上修") + t("规则一：没有截止日一律降为缺失", bad["signal"] == ls.SIG_NONE) + + # 规则三：缺失既不是负面也不是正面 + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_BROKER, ls.SIG_UP), + ls.from_events()]) + t("规则三：缺失被剔出合成，不影响状态", r["state"] == ls.STATE_STRONG) + t("缺失仍记在名单里", ls.PATH_EVENT in r["missing"] and ls.PATH_EVENT not in r["usable"]) + + # 规则四：慢信号不能单独定态 + r = ls.compose([sig(ls.PATH_JUDGE, ls.SIG_FLAT)]) + t("规则四：只剩产业研判这一路慢信号 -> 无法判断加证据不足", + r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_THIN) + r = ls.compose([sig(ls.PATH_JUDGE, ls.SIG_FLAT), sig(ls.PATH_BROKER, ls.SIG_FLAT)]) + t("配上一路中速信号就能定态", r["state"] == ls.STATE_HOLD) + + print("四态各自的入口") + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_JUDGE, ls.SIG_UP)]) + t("两路向好且含甲乙 -> 逻辑强化", r["state"] == ls.STATE_STRONG) + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP)]) + t("只有一路向好 -> 逻辑成立而不是强化", r["state"] == ls.STATE_HOLD) + r = ls.compose([sig(ls.PATH_BROKER, ls.SIG_UP), sig(ls.PATH_EVENT, ls.SIG_UP)]) + t("两路向好但不含甲乙 -> 不给强化（券商上修不能单独撑起强化）", + r["state"] == ls.STATE_HOLD) + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_FLAT)]) + t("一路平稳无负面 -> 逻辑成立", r["state"] == ls.STATE_HOLD) + + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_SPLIT)]) + t("一路内部分歧 -> 无法判断加证据矛盾", + r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT) + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_BROKER, ls.SIG_DOWN)]) + t("两路方向相反 -> 无法判断加证据矛盾", + r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT) + + print("规则二 负向优先") + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_DOWN)]) + t("转弱但没到进入条件 -> 最多到无法判断，不落存疑", + r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT) + r = ls.compose([sig(ls.PATH_BROKER, ls.SIG_DOWN, hard=True)]) + t("转弱且达到进入条件 -> 逻辑存疑", r["state"] == ls.STATE_DOUBT) + t("存疑的理由写明是哪条路径触发的", + any(ls.DOUBT_TRIGGERS[ls.PATH_BROKER] in x for x in r["reasons"])) + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_FLAT), sig(ls.PATH_BROKER, ls.SIG_DOWN, hard=True)]) + t("一路平稳压不住另一路的硬触发", r["state"] == ls.STATE_DOUBT) + r = ls.compose([sig(ls.PATH_CLAIM, ls.SIG_UP), sig(ls.PATH_BROKER, ls.SIG_DOWN, hard=True)]) + t("一路向好与硬触发同现 -> 是矛盾交人裁决，不直接落存疑", + r["state"] == ls.STATE_UNKNOWN and r["why"] == ls.WHY_CONFLICT) + t("硬标记只对转弱有效", ls.signal(ls.PATH_CLAIM, ls.SIG_UP, as_of=DS, hard=True)["hard"] is False) + + +def test_settle(): + print("迁移的抗抖动与进出不对称") + s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_DOUBT, [ls.STATE_HOLD, ls.STATE_DOUBT]) + t("进入存疑即刻成立，不等确认", s == ls.STATE_DOUBT) + s, why = ls.settle(ls.STATE_DOUBT, ls.STATE_HOLD, [ls.STATE_DOUBT, ls.STATE_HOLD]) + t("退出存疑：只有一天不存疑，维持存疑", s == ls.STATE_DOUBT and "现在 1 天" in why) + s, _ = ls.settle(ls.STATE_DOUBT, ls.STATE_HOLD, + [ls.STATE_HOLD, ls.STATE_HOLD, ls.STATE_HOLD]) + t("退出存疑：连续三天不存疑才放行", s == ls.STATE_HOLD) + s, _ = ls.settle(ls.STATE_DOUBT, ls.STATE_HOLD, + [ls.STATE_HOLD, ls.STATE_DOUBT, ls.STATE_HOLD]) + t("三天里夹一天存疑就不放行", s == ls.STATE_DOUBT) + + s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_STRONG, [ls.STATE_HOLD, ls.STATE_STRONG]) + t("其余迁移：只有一天同向，先维持原状", s == ls.STATE_HOLD) + s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_STRONG, [ls.STATE_STRONG, ls.STATE_STRONG]) + t("连续两天同向才迁移", s == ls.STATE_STRONG) + s, _ = ls.settle(None, ls.STATE_HOLD, [ls.STATE_HOLD]) + t("第一次见到这只票，直接采用", s == ls.STATE_HOLD) + s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_HOLD, [ls.STATE_HOLD]) + t("状态没变时不需要确认", s == ls.STATE_HOLD) + s, _ = ls.settle(ls.STATE_HOLD, ls.STATE_STRONG, [ls.STATE_STRONG], confirm_days=1) + t("确认天数旋钮生效", s == ls.STATE_STRONG) + + +def test_apply(): + print("作用于候选卡的单调性") + r = ls.apply_to_card("关注", ls.STATE_STRONG) + t("强化不把关注升成候选", r["verdict"] == "关注" and r["rank_bonus"] == 1) + r = ls.apply_to_card("仅展示", ls.STATE_STRONG) + t("强化也不把仅展示升上来", r["verdict"] == "仅展示" and not r["force_confirm"]) + r = ls.apply_to_card("候选", ls.STATE_DOUBT) + t("候选加存疑 -> 强制人工确认（杀逻辑与杀估值那条边界）", + r["force_confirm"] and r["pause_add"] and r["verdict"] == "候选") + r = ls.apply_to_card("仅展示", ls.STATE_DOUBT) + t("存疑不把仅展示变成可执行", r["verdict"] == "仅展示" and not r["force_confirm"]) + r = ls.apply_to_card("候选", ls.STATE_UNKNOWN) + t("无法判断不改判决也不强制确认", + r["verdict"] == "候选" and not r["force_confirm"] and not r["pause_add"]) + r = ls.apply_to_card("候选", ls.STATE_HOLD) + t("逻辑成立什么都不做", r["rank_bonus"] == 0 and not r["force_confirm"] and not r["note"]) + + +def main(): + test_claims() + test_judgement() + test_broker() + test_compose() + test_settle() + test_apply() + print("ALL OK — 四路归一 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / " + "作用于候选卡的单调性 全部通过") + + +if __name__ == "__main__": + main()