245 lines
13 KiB
Python
245 lines
13 KiB
Python
"""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()
|