93 lines
4.6 KiB
Python
93 lines
4.6 KiB
Python
"""计划装配接入逻辑状态四态的离线单测（不连库）。
|
||
|
||
钉住三件事：
|
||
一，四态随每张卡一起产出，并按约定的形状发给下游（每路带截止日与一句话说明）。
|
||
二，四态**不改判决**。候选卡的三门槛判决与逻辑四态是正交的两维，收敛规则是单调的：
|
||
逻辑强化只能提前卡内序，永远不能把判决往上升一档；逻辑存疑只能改分流通道，
|
||
不能把仅展示变成可执行。
|
||
三，取数任一路读不到都不让计划断产，那一路记缺失。
|
||
|
||
跑法：python3 test_plan_logic_state.py 或 pytest test_plan_logic_state.py
|
||
"""
|
||
import logic_state as ls
|
||
import plan
|
||
|
||
DS = "2026-09-03"
|
||
|
||
|
||
def t(name, cond):
|
||
assert cond, name
|
||
print(" ok", name)
|
||
|
||
|
||
def claim(date, direction="利好"):
|
||
return {"disclosure_date": date, "direction": direction, "mechanism": "机制",
|
||
"doc_title": "研报", "claim_id": "c" + date.replace("-", "")}
|
||
|
||
|
||
def jrow(leaning="偏多", verified=1, migrated=0, stale=3, seg="固态电解质"):
|
||
return {"leaning": leaning, "leaning_prev": None, "migrated": migrated,
|
||
"stale_days": stale, "verified": verified, "review_date": "2026-09-03",
|
||
"n_materials": 43, "subject_name": seg, "segment_name": seg,
|
||
"n_bull": 3, "n_bear": 3}
|
||
|
||
|
||
def main():
|
||
print("四态随卡产出")
|
||
evd = {"logic": [claim("2026-08-25")]}
|
||
seg_of = {"SH600000": ["固态电解质"]}
|
||
seg_view = {"固态电解质": jrow()}
|
||
broker = {"SH600000": ls.signal(ls.PATH_BROKER, ls.SIG_FLAT, as_of=DS,
|
||
coverage=6, why="预测中位数变化 +1%，在阈值之内")}
|
||
st = plan._logic_state_of("SH600000", evd, seg_of, seg_view, broker, DS) # noqa: SLF001
|
||
t("三路都在且无负面 -> 逻辑成立", st["state"] == ls.STATE_HOLD)
|
||
t("四路都记了名，缺的那一路写明是公司事件",
|
||
len(st["paths"]) == 4 and ls.PATH_EVENT in st["missing"])
|
||
|
||
print("取数缺席时不断产")
|
||
st = plan._logic_state_of("SH600000", {}, {}, {}, {}, DS) # noqa: SLF001
|
||
t("四路全缺 -> 无法判断加证据不足，不抛异常",
|
||
st["state"] == ls.STATE_UNKNOWN and st["why"] == ls.WHY_THIN)
|
||
t("缺失名单写全了四路", len(st["missing"]) == 4)
|
||
|
||
st = plan._logic_state_of("SH999999", evd, seg_of, seg_view, broker, DS) # noqa: SLF001
|
||
t("这只票不在任何被指向环节上 -> 产业研判缺席，其余照算",
|
||
ls.PATH_JUDGE in st["missing"] and ls.PATH_CLAIM in st["usable"])
|
||
|
||
print("一票挂多个环节")
|
||
st = plan._logic_state_of( # noqa: SLF001
|
||
"SH600000", evd, {"SH600000": ["没评过的环节", "固态电解质"]}, seg_view, broker, DS)
|
||
t("取第一个有行业观点的那个环节", ls.PATH_JUDGE in st["usable"])
|
||
|
||
print("发给下游的形状")
|
||
out = plan._state_out(st) # noqa: SLF001
|
||
t("状态、子因、截止日都在", set(out) >= {"state", "why", "as_of", "usable", "missing",
|
||
"reasons", "paths"})
|
||
t("每一路都带自己的截止日与一句话说明",
|
||
all(set(p) == {"path", "signal", "as_of", "why"} for p in out["paths"]))
|
||
t("不发内部中间量（硬触发标记、出处原值不外泄）",
|
||
all("hard" not in p and "refs" not in p for p in out["paths"]))
|
||
t("没有状态时发 None，不硬拼一个空壳", plan._state_out(None) is None) # noqa: SLF001
|
||
|
||
print("四态不改判决（收敛规则单调）")
|
||
for state in (ls.STATE_STRONG, ls.STATE_HOLD, ls.STATE_UNKNOWN, ls.STATE_DOUBT):
|
||
for verdict in ("候选", "关注", "仅展示"):
|
||
r = ls.apply_to_card(verdict, state)
|
||
assert r["verdict"] == verdict, (verdict, state, r)
|
||
t("十二种组合逐个扫过，判决一次都没被四态改动", True)
|
||
t("逻辑强化最多提前卡内序", ls.apply_to_card("关注", ls.STATE_STRONG)["rank_bonus"] == 1)
|
||
t("逻辑存疑不把仅展示变成可执行",
|
||
not ls.apply_to_card("仅展示", ls.STATE_DOUBT)["force_confirm"])
|
||
t("候选加逻辑存疑是唯一需要新语义的一格：强制人工确认",
|
||
ls.apply_to_card("候选", ls.STATE_DOUBT)["force_confirm"])
|
||
|
||
print("计划日与数据日差一天")
|
||
t("次日推算正确", plan._next_day("2026-09-03") == "2026-09-04") # noqa: SLF001
|
||
t("认不出的日期原样返回，不抛异常", plan._next_day("不是日期") == "不是日期") # noqa: SLF001
|
||
|
||
print("ALL OK — 四态随卡产出 / 缺席不断产 / 下游形状 / 判决不被改动 / 计划日推算 全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|