akg-factor-bridge/test_logic_state.py

"""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_judgement_history():
    print("乙路 硬触发的维持窗口（审查 2026-09-07 第 9 条）")
    # 迁移只在指纹变化那天记 migrated=1，次日归零；不带历史时硬触发只维持一个计划日。
    flip = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1, plan_date="2026-09-02")
    after1 = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date="2026-09-03")
    today = jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date="2026-09-04")
    r = ls.from_judgement(today)
    t("不带历史：迁移次日之后只是转弱、不硬", r["signal"] == ls.SIG_DOWN and not r["hard"])
    r = ls.from_judgement(today, history=[flip, after1])
    t("带历史：近几日翻过空且此后一直偏空 -> 硬触发维持", r["hard"] and "此后一直偏空" in r["why"])
    back = jrow(leaning=ls.LEAN_BULL, prev=ls.LEAN_BEAR, migrated=1, plan_date="2026-09-03")
    r = ls.from_judgement(today, history=[flip, back])
    t("翻空之后又回到偏多过一次 -> 硬触发作废", not r["hard"])
    old = [jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BULL, migrated=1, plan_date="2026-08-20")] \
        + [jrow(leaning=ls.LEAN_BEAR, prev=ls.LEAN_BEAR, migrated=0, plan_date=f"2026-08-2{i}")
           for i in range(1, 7)]
    r = ls.from_judgement(today, history=old, hold_days=5)
    t("翻空早于维持窗口 -> 不再撑着硬触发", not r["hard"])
    r = ls.from_judgement(today, history=[None, "坏行", flip])
    t("历史里的坏行被跳过、不抛错", r["hard"])


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": 1.8, "firms": 2}, {"eps": 2.4, "firms": 3}, as_of=DS)
    t("下修且比例上收缩，但前窗口只有三家 -> 转弱不带硬触发（一家没发报告就是 33%）",
      r["signal"] == ls.SIG_DOWN and not r["hard"] and "不作数" in r["why"])
    r = ls.from_broker({"eps": 1.8, "firms": 4}, {"eps": 2.4, "firms": 5}, as_of=DS)
    t("前窗口够五家、走掉一家只有 20% -> 够不着收缩线",
      r["signal"] == ls.SIG_DOWN and not r["hard"])
    r = ls.from_broker({"eps": 1.8, "firms": 3}, {"eps": 2.4, "firms": 5},
                       as_of=DS, cover_min_base=3)
    t("基数门槛旋钮生效", r["signal"] == ls.SIG_DOWN and r["hard"])
    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_judgement_history()
    test_broker()
    test_compose()
    test_settle()
    test_apply()
    print("ALL OK — 四路归一 / 乙路硬触发维持窗口 / 合成四条规则 / 四态入口 / 负向优先 / 进出不对称 / "
          "作用于候选卡的单调性 全部通过")


if __name__ == "__main__":
    main()