tradingSystem/scripts/test_batch25_units.py

# -*- coding: utf-8 -*-
"""解锁重问 (2026-09-09 方案第四之一节, 台账 055)。全部离线, 不连库不连 Redis 不调模型。

  A 纯逻辑: 时段占比的四个边界、量比的单位换算与两条哨兵、窗口解析。
  B 四条护栏: 总开关、二十分钟间隔、当日一次、时段窗口, 各自的 blocked_by。
  C 三条解锁条件: U1 的先后与发送方、U2 的定性翻转与漂移边界、U3 的三条并列与 fail-closed。
  D 请求体: force 为假时与今天逐字节相同; 为真时只多两个键; 白名单收了那一句。
  E 名额与落账: 名额只在真发请求时用掉、试算不落账、规则闸拦下不消耗、重问后仍驳回加前缀。
  F 分流: 重问放行强制交人, 哪怕档位 full。
  G 参数与页面: 九个键登记齐、页面认得新动作名。
"""
import io
import os
import sys
import traceback
from datetime import datetime, timedelta

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.core import reask_rules as rr                                          # noqa: E402
from app.services import param_store                                            # noqa: E402
from app.services import reask_service as rs                                    # noqa: E402

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


class _Patch:
    def __init__(self): self._saved = []
    def __enter__(self): return self
    def __call__(self, obj, name, value):
        self._saved.append((obj, name, getattr(obj, name))); setattr(obj, name, value)
    def __exit__(self, *a):
        for obj, name, v in reversed(self._saved):
            setattr(obj, name, v)


def _t(h, m):
    return datetime(2026, 9, 9, h, m, 0)


P = {"enabled": True, "min_gap_min": 20, "max_per_day": 1, "window": "0945-1430",
     "ref_drift": 0.03, "up_min": 0.02, "vol_min": 1.5, "vol_mult": 1.3}


def _base(**kw):
    b = {"at": "2026-09-09T10:12:00", "price": 43.15, "chg_prev": 0.076, "chg_open": 0.041,
         "vol_ratio": 1.10, "judge_conf": 72, "y_signal": "WATCH",
         "support": 40.00, "pressure": 42.00}
    b.update(kw)
    return b


def _now(**kw):
    n = {"price": 43.15, "vol_ratio": 1.10, "y_signal": "WATCH", "support": 40.00,
         "pressure": 42.00, "bionic_buy_at": ""}
    n.update(kw)
    return n


@case("A 时段占比: 盘前 0、上午末 0.5、午休不走、收盘 1")
def test_frac():
    assert rr.elapsed_trading_frac(_t(9, 0)) == 0.0
    assert rr.elapsed_trading_frac(_t(9, 30)) == 0.0
    assert abs(rr.elapsed_trading_frac(_t(10, 30)) - 60 / 240.0) < 1e-9
    assert abs(rr.elapsed_trading_frac(_t(11, 30)) - 0.5) < 1e-9
    assert abs(rr.elapsed_trading_frac(_t(12, 0)) - 0.5) < 1e-9, "午休不该往前走"
    assert abs(rr.elapsed_trading_frac(_t(13, 0)) - 0.5) < 1e-9
    assert rr.elapsed_trading_frac(_t(15, 0)) == 1.0
    assert rr.elapsed_trading_frac(_t(16, 0)) == 1.0


@case("A 量比: 手与股差 100 倍的换算、两条哨兵、覆盖太短不判")
def test_vol_ratio():
    # 1736 万股 = 17.36 万手; 五日均量 18.5 万手; 覆盖 0.829 → 1.128 (09-09 四方股份实测值)
    assert rr.vol_ratio(17361626, 185554.7, 0.8292) == 1.128, rr.vol_ratio(17361626, 185554.7, 0.8292)
    assert rr.vol_ratio(100 * 185554.7, 185554.7, 1.0) == 1.0, "整整一份日均量、走完全天 = 1.0"
    assert rr.vol_ratio(17361626, 185554.7, 0.05) is None, "覆盖不足 8% 不判"
    assert rr.vol_ratio(100 * 185554.7 * 31, 185554.7, 1.0) is None, "原始比超 30 判数据异常"
    for bad in ((None, 1.0, 1.0), (1.0, None, 1.0), (1.0, 1.0, None), (1.0, 0, 1.0), (-1, 1.0, 1.0)):
        assert rr.vol_ratio(*bad) is None, bad


@case("A 窗口解析: 边界含端点; 格式写坏了不堵路")
def test_window():
    assert rr.in_window(_t(9, 45), "0945-1430") and rr.in_window(_t(14, 30), "0945-1430")
    assert not rr.in_window(_t(9, 44), "0945-1430") and not rr.in_window(_t(14, 31), "0945-1430")
    assert rr.in_window(_t(9, 0), "写坏了"), "格式错不该把整条路堵死"


@case("B 四条护栏各自挡住并说清是哪一条")
def test_gates():
    off = dict(P, enabled=False)
    r = rr.evaluate(base=_base(), now_read=_now(), params=off, used_today=0, now=_t(11, 0))
    assert r["blocked_by"] == "G0" and not r["unlock"]
    r = rr.evaluate(base=_base(), now_read=_now(), params=P, used_today=0, now=_t(10, 25))
    assert r["blocked_by"] == "G1" and not r["unlock"], r
    r = rr.evaluate(base=_base(), now_read=_now(), params=P, used_today=1, now=_t(11, 0))
    assert r["blocked_by"] == "G2" and not r["unlock"], r
    r = rr.evaluate(base=_base(), now_read=_now(), params=P, used_today=0, now=_t(14, 45))
    assert r["blocked_by"] == "G3" and not r["unlock"], r
    # 20 分钟整点即可
    r = rr.evaluate(base=_base(), now_read=_now(), params=P, used_today=0, now=_t(10, 32))
    assert r["blocked_by"] is None, r


@case("C U1: 晚于驳回才算; 早于驳回是循环论证; 没有留痕不成立")
def test_u1():
    late = rr.evaluate(base=_base(), now_read=_now(bionic_buy_at="2026-09-09T10:40:00"),
                       params=P, used_today=0, now=_t(11, 0))
    assert late["unlock"] and late["hits"] == ["U1"], late
    early = rr.evaluate(base=_base(), now_read=_now(bionic_buy_at="2026-09-09T10:04:00"),
                        params=P, used_today=0, now=_t(11, 0))
    assert not early["unlock"] and "循环论证" not in early["why"], early
    none = rr.evaluate(base=_base(), now_read=_now(), params=P, used_today=0, now=_t(11, 0))
    assert not none["unlock"]
    # 驳回时刻读不到时判不了先后 → 不成立 (fail-closed)
    nb = rr.evaluate(base=_base(at=""), now_read=_now(bionic_buy_at="2026-09-09T10:40:00"),
                     params=P, used_today=0, now=_t(11, 0))
    assert not nb["unlock"] and nb["notes"], nb


@case("C U2: 定性翻成看多、支撑压力漂过 3%; 2.9% 不算")
def test_u2():
    up = rr.evaluate(base=_base(), now_read=_now(y_signal="BUY"), params=P, used_today=0, now=_t(11, 0))
    assert up["unlock"] and up["hits"] == ["U2"], up
    assert not rr.evaluate(base=_base(y_signal="BUY"), now_read=_now(y_signal="BUY"),
                           params=P, used_today=0, now=_t(11, 0))["unlock"], "本来就是 BUY 不算改写"
    d29 = rr.evaluate(base=_base(), now_read=_now(support=40.00 * 1.029), params=P, used_today=0, now=_t(11, 0))
    assert not d29["unlock"], d29
    d31 = rr.evaluate(base=_base(), now_read=_now(support=40.00 * 1.031), params=P, used_today=0, now=_t(11, 0))
    assert d31["unlock"] and d31["hits"] == ["U2"], d31
    dn = rr.evaluate(base=_base(), now_read=_now(pressure=42.00 * 0.96), params=P, used_today=0, now=_t(11, 0))
    assert dn["unlock"], "往下漂同样是底牌变了"


@case("C U3: 三条并列才算; 少一条就不解锁; 读不到一律不成立")
def test_u3():
    ok = rr.evaluate(base=_base(), now_read=_now(price=44.10, vol_ratio=1.60),
                     params=P, used_today=0, now=_t(11, 0))
    assert ok["unlock"] and ok["hits"] == ["U3"], ok
    # 只涨不放量
    assert not rr.evaluate(base=_base(), now_read=_now(price=44.10, vol_ratio=1.20),
                           params=P, used_today=0, now=_t(11, 0))["unlock"]
    # 放量但没涨够 (09-09 四方股份下午就是这个形态的反面: 量比 1.128 且价格还跌了)
    assert not rr.evaluate(base=_base(), now_read=_now(price=43.50, vol_ratio=1.60),
                           params=P, used_today=0, now=_t(11, 0))["unlock"]
    # 量比够 1.5 但没到驳回时的 1.3 倍
    assert not rr.evaluate(base=_base(vol_ratio=1.40), now_read=_now(price=44.10, vol_ratio=1.55),
                           params=P, used_today=0, now=_t(11, 0))["unlock"]
    # fail-closed: 量比读不到、驳回时没记量比、价格读不到, 三种都不成立
    for nr, bs in ((_now(price=44.10, vol_ratio=None), _base()),
                   (_now(price=44.10, vol_ratio=1.60), _base(vol_ratio=None)),
                   (_now(price=None, vol_ratio=1.60), _base())):
        r = rr.evaluate(base=bs, now_read=nr, params=P, used_today=0, now=_t(11, 0))
        assert not r["unlock"], (r, nr, bs)


@case("C 多条同时成立时都记下来 (复盘要知道是哪几条一起促成的)")
def test_multi_hit():
    r = rr.evaluate(base=_base(), now_read=_now(price=44.10, vol_ratio=1.60, y_signal="BUY",
                                                bionic_buy_at="2026-09-09T10:40:00"),
                    params=P, used_today=0, now=_t(11, 0))
    assert r["unlock"] and set(r["hits"]) == {"U1", "U2", "U3"}, r


@case("D 请求体: force 为假时与今天逐字节相同; 为真时只多两个键")
def test_payload():
    from app.services import judge as jd
    cand = {"ts_code": "601126.SH", "action": "OPEN", "qty": 100, "reason": "x",
            "hard_numbers": {"price": 43.15}}
    seen = {}

    class _Resp:
        status_code = 200
        def raise_for_status(self): pass
        def json(self): return {"verdict": "PASS", "reason": "ok", "confidence": 80}

    import types
    fake_requests = types.SimpleNamespace(post=lambda url, json=None, timeout=None:
                                          (seen.setdefault(len(seen), json), _Resp())[1])
    with _Patch() as p:
        p(jd, "available", lambda: True)
        p(jd, "base_url", lambda: "http://x")
        sys.modules["requests"] = fake_requests
        jd.request(dict(cand))
        jd.request(dict(cand), force=True, reask_text="曾驳回于 43.15，现价 44.10")
    plain, forced = seen[0], seen[1]
    assert "force_refresh" not in plain and "reask_text" not in plain, plain
    assert forced["force_refresh"] is True and forced["reask_text"].startswith("曾驳回")
    assert set(forced) - set(plain) == {"force_refresh", "reask_text"}, set(forced) - set(plain)
    for k in plain:
        assert plain[k] == forced[k] or k == "hard_numbers", k
    assert "reask_text" in jd.OPEN_JUDGE_KEYS


@case("E 名额: 真发请求才落账; 落账失败就不算重问; 动作名不与建新仓混")
def test_commit():
    calls = []
    with _Patch() as p:
        p(rs.pms_repo, "insert_ledger", lambda **kw: calls.append(kw) or 1)
        assert rs.commit("601126.SH", {"hits": ["U3"], "why": "价升量增"}, 44.1) is True
    assert calls[0]["action"] == "OPEN_REASK" != "OPEN", calls[0]
    assert calls[0]["verdict"] == "NOTE" and calls[0]["arbiter"] == "rule"
    assert "U3" in calls[0]["reason"] and calls[0]["hard_numbers"]["reask"]["hits"] == ["U3"]

    def boom(**kw): raise RuntimeError("表写不进")
    with _Patch() as p:
        p(rs.pms_repo, "insert_ledger", boom)
        assert rs.commit("601126.SH", {}, 1.0) is False, "落账失败必须回 False, 让调用方别重问"


@case("E 判定零副作用: evaluate_all 一行都不写")
def test_readonly():
    wrote = []
    with _Patch() as p:
        p(rs.pms_repo, "insert_ledger", lambda **kw: wrote.append(kw))
        p(rs.pms_repo, "judge_rejected_open_rows",
          lambda since: {"601126.SH": {"at": "2026-09-09 10:12:00", "price": 43.15,
                                       "hard": {"reask_base": _base()}}})
        p(rs.pms_repo, "reask_count_today", lambda since: {})
        p(rs.pms_repo, "buy_signals_today", lambda since: {})
        p(rs, "_now_read", lambda code, at: _now(price=44.10, vol_ratio=1.60))
        p(rs, "params", lambda: P)
        out = rs.evaluate_all({("601126.SH", "OPEN"): "驳回"}, now=_t(11, 0))
    assert ("601126.SH", "OPEN") in out and out[("601126.SH", "OPEN")]["hits"] == ["U3"], out
    assert not wrote, "解锁判定每分钟都在跑, 一行都不许写"


@case("E 读账本失败一律不解锁 (宁可漏, 不可误)")
def test_repo_fail():
    def boom(since): raise RuntimeError("库抖")
    with _Patch() as p:
        p(rs.pms_repo, "judge_rejected_open_rows", boom)
        p(rs, "params", lambda: P)
        assert rs.evaluate_all({("601126.SH", "OPEN"): "驳回"}, now=_t(11, 0)) == {}


@case("E 只认决策系统发的转多留痕 (别的来源不算 U1)")
def test_u1_sender():
    with _Patch() as p:
        p(rs.pms_repo, "judge_rejected_open_rows",
          lambda since: {"601126.SH": {"at": "2026-09-09 10:12:00", "price": 43.15,
                                       "hard": {"reask_base": _base()}}})
        p(rs.pms_repo, "reask_count_today", lambda since: {})
        p(rs.pms_repo, "buy_signals_today",
          lambda since: {"601126.SH": {"at": "2026-09-09 10:40:00", "reason": "别处发来的看多"}})
        p(rs, "_now_read", lambda code, at: _now(bionic_buy_at=at))
        p(rs, "params", lambda: P)
        assert rs.evaluate_all({("601126.SH", "OPEN"): "驳回"}, now=_t(11, 0)) == {}, "发送方不对不算"


@case("F 重问放行强制交人; 重问仍驳回加前缀且带次序")
def test_route():
    src = io.open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                               "app/services/proposal_service.py"), encoding="utf-8").read()
    # 2026-09-10 改了断言的写法, 意图不变。原先写的是 `'or bool(reask_why))' in src`,
    # 它把「reask_why 必须是那一行的最后一项」也一起钉住了 —— 而那只是当时的排版,
    # 不是这条规矩的内容。同日往一票否决里加「上游标了硬风险的候选也交人」之后,
    # 那一行结尾从 `bool(reask_why))` 变成 `bool(risk_why))`, 这条断言就红了,
    # 而被它保护的行为一个字都没变。现在改成取出那一行、只看 reask_why 在不在里面。
    _fq = src[src.index("force_queue = ("):src.index("auto_exec = (not force_queue)")]
    assert "bool(reask_why)" in _fq, "重问必须进 force_queue 一票否决"
    # 2026-09-10 往这条链里插了「上游标了硬风险」一档, 排在重问之后 —— 重问仍是最前一项。
    assert src.index("why = (reask_why") > 0, "重问的原因要排在最前, 人最需要知道它"
    assert '"重问后仍驳回：" + _rsn' in src and '_hn["reask_seq"] = 1' in src
    assert 'reask_service.commit(code, _rk, price)' in src
    i = src.index("_rk = c.get(\"reask\") if not dry_run else None")
    j = src.index("verdict = judge.request(c,")
    assert "dry_run" in src[i:i + 60], "试算一律不落账不带强制标记"
    assert j > i, "名额必须在发请求之前落"
    # 规则闸在研判闸前面, 且被规则闸拦下的直接 return —— 不会走到落名额那一步
    assert src.index('if not gate["passed"]:') < i


@case("G 九个参数登记齐、有范围、页面认得新动作名")
def test_params_and_page():
    keys = ["PMS_OPEN_REASK_ENABLED", "PMS_OPEN_REASK_MIN_GAP_MIN", "PMS_OPEN_REASK_MAX_PER_DAY",
            "PMS_OPEN_REASK_WINDOW", "PMS_OPEN_REASK_REF_DRIFT", "PMS_OPEN_REASK_UP",
            "PMS_OPEN_REASK_VOL_MIN", "PMS_OPEN_REASK_VOL_MULT", "PMS_JUDGE_REJECT_CONF_MIN"]
    for k in keys:
        assert k in param_store.RUNTIME_EXTRA, k
        assert param_store.RUNTIME_EXTRA[k][2], k + " 没有中文说明"
    for k in keys:
        if param_store.RUNTIME_EXTRA[k][1] in (int, float):
            assert k in param_store._RANGES, k + " 没有范围"
    p = rs.params()
    assert set(p) == {"enabled", "min_gap_min", "max_per_day", "window", "ref_drift",
                      "up_min", "vol_min", "vol_mult"}, p
    from app.web import main as web
    html = io.open(os.path.join(web.STATIC_DIR, "index.html"), encoding="utf-8").read()
    assert "OPEN_REASK:'重问研判'" in html, "页面不认这个动作名就会把英文原码印给人看"


def main():
    ok = 0
    for name, fn in RESULTS:
        try:
            fn()
            ok += 1
            print("  ok  " + name)
        except Exception:
            print("  FAIL " + name)
            traceback.print_exc()
    print("-" * 60)
    if ok == len(RESULTS):
        print("ALL PASS (%d cases)" % ok)
        return 0
    print("FAILED %d/%d" % (len(RESULTS) - ok, len(RESULTS)))
    return 1


if __name__ == "__main__":
    sys.exit(main())