tradingSystem/scripts/test_batch32_units.py

# -*- coding: utf-8 -*-
"""盘中确认包 (2026-09-14 方案第三节, 台账 012)。全部离线, 不连库。

给技术面这一票加盘中确认, 只用于入场。时段与量比与解锁重问同源 (台账 003)。
  A intraday_confirm 纯函数: 相位无关原样/收口突破要价与量同时满足/突破开关关不升/取不到按等待/
    开口确认要均价与量/时段外开口进观察。
  B 同源哨兵与映射带上轨: in_window 是 reask_rules 那一个; synthesize base 与 _compact 带 boll_upper。
  C 接线 (proposal_service._apply_intraday): 突破改立场重合议为放行、硬数字带 intraday;
    开口未确认路由改观察带 wait_confirm; 总闸关掉 _intraday_cfg 为 None (逐字回旧)。
  E 参数登记与范围。
"""
import os
import sys
import traceback
from datetime import datetime

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

from app.core import tech_rules as tr           # noqa: E402
from app.core import reask_rules as rr          # noqa: E402
from app.services import param_store as ps       # noqa: E402

RESULTS = []


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


def _st(phase, **kw):
    """一个技术面立场块 (synthesize 的紧凑态)。"""
    d = {"stance": "中性", "strength": None, "phase": phase, "sar_side": "多",
         "boll_upper": 10.0, "reason": "x"}
    d.update(kw)
    return d


NOW = datetime(2026, 9, 15, 10, 0)          # 在 0945-1430 时段内


# ================================================================ A intraday_confirm 纯函数
@case("A 相位无关 (趋势多) → 原样返回, kind 空, 不改")
def _():
    r = tr.intraday_confirm(_st("趋势多"), price=11, vwap=10, upper_prev=10,
                            vol_ratio=3, now=NOW)
    assert r["kind"] is None and r["changed"] is False


@case("A 收口等待·价站上昨日上轨且量足 → breakout (看多/强/盘中开口向上)")
def _():
    r = tr.intraday_confirm(_st("收口等待"), price=10.5, vwap=10.0, upper_prev=10.2,
                            vol_ratio=2.0, now=NOW, vol_min=1.5)
    assert r["kind"] == "breakout" and r["changed"] is True
    assert r["state"]["stance"] == "看多" and r["state"]["strength"] == "强"
    assert r["state"]["phase"] == "盘中开口向上"


@case("A 收口等待·价没过上轨 → 不升 (原样, kind 空)")
def _():
    r = tr.intraday_confirm(_st("收口等待"), price=10.1, vwap=10.0, upper_prev=10.2,
                            vol_ratio=2.0, now=NOW, vol_min=1.5)
    assert r["kind"] is None and r["changed"] is False


@case("A 收口等待·量不足 → 不升")
def _():
    r = tr.intraday_confirm(_st("收口等待"), price=10.5, vwap=10.0, upper_prev=10.2,
                            vol_ratio=1.2, now=NOW, vol_min=1.5)
    assert r["kind"] is None


@case("A 收口等待·突破开关关掉 → 不升 (breakout_on 假)")
def _():
    r = tr.intraday_confirm(_st("收口等待"), price=10.5, vwap=10.0, upper_prev=10.2,
                            vol_ratio=2.0, now=NOW, vol_min=1.5, breakout_on=False)
    assert r["kind"] is None and r["changed"] is False


@case("A 收口等待·昨日上轨取不到 → 不升 (取不到不放行)")
def _():
    r = tr.intraday_confirm(_st("收口等待"), price=10.5, vwap=10.0, upper_prev=None,
                            vol_ratio=2.0, now=NOW, vol_min=1.5)
    assert r["kind"] is None


@case("A 开口向上·价≥均价且量足 → opened_ok, 块不变")
def _():
    r = tr.intraday_confirm(_st("开口向上", stance="看多"), price=10.5, vwap=10.0,
                            upper_prev=10.2, vol_ratio=2.0, now=NOW, vol_min=1.5)
    assert r["kind"] == "opened_ok" and r["changed"] is False
    assert r["state"]["phase"] == "开口向上" and r["state"]["stance"] == "看多"


@case("A 开口向上·价低于均价 → opened_wait")
def _():
    r = tr.intraday_confirm(_st("开口向上", stance="看多"), price=9.8, vwap=10.0,
                            upper_prev=10.2, vol_ratio=2.0, now=NOW, vol_min=1.5)
    assert r["kind"] == "opened_wait" and r["changed"] is False


@case("A 开口向上·量不足 → opened_wait")
def _():
    r = tr.intraday_confirm(_st("开口向上", stance="看多"), price=10.5, vwap=10.0,
                            upper_prev=10.2, vol_ratio=1.0, now=NOW, vol_min=1.5)
    assert r["kind"] == "opened_wait"


@case("A 开口向上·读数取不到 (现价 None) → opened_wait (取不到按等待)")
def _():
    r = tr.intraday_confirm(_st("开口向上", stance="看多"), price=None, vwap=10.0,
                            upper_prev=10.2, vol_ratio=2.0, now=NOW, vol_min=1.5)
    assert r["kind"] == "opened_wait"


@case("A 时段外·开口向上 → outside_window (进观察)")
def _():
    early = datetime(2026, 9, 15, 9, 40)        # 早于 09:45
    r = tr.intraday_confirm(_st("开口向上", stance="看多"), price=10.5, vwap=10.0,
                            upper_prev=10.2, vol_ratio=2.0, now=early, window="0945-1430")
    assert r["kind"] == "outside_window" and r["changed"] is False


@case("A 时段外·收口等待 → 原样 (kind 空, 不动)")
def _():
    early = datetime(2026, 9, 15, 9, 40)
    r = tr.intraday_confirm(_st("收口等待"), price=10.5, vwap=10.0, upper_prev=10.2,
                            vol_ratio=2.0, now=early, window="0945-1430")
    assert r["kind"] is None and r["changed"] is False


# ================================================================ B 同源哨兵与映射带上轨
@case("B 同源哨兵·tech_rules 的 in_window 就是 reask_rules 那一个函数对象")
def _():
    assert tr.in_window is rr.in_window


@case("B 映射带上轨·synthesize base 与 _compact 都带 boll_upper")
def _():
    from app.services import tech_service as ts
    rows = [{"data_date": 20260914, "reanchored": 1, "quality": "OK", "sar_side": "多",
             "boll_pos": 0.6, "boll_upper": 12.34, "boll_squeeze": 1, "sar_flip_days": 5,
             "bbi_state": "多头区", "sar_value": 11.0, "boll_bw_pct": 3.0}]
    st = tr.synthesize(rows)
    assert st.get("boll_upper") == 12.34
    assert "boll_upper" in ts._compact(st) and ts._compact(st)["boll_upper"] == 12.34


# ================================================================ C 接线 (proposal_service._apply_intraday)
def _cfg(**kw):
    c = {"breakout_on": True, "vol_min": 1.5, "window": "0945-1430",
         "cparams": {"fund_required": True, "consensus_route": True, "weak_confirm": True}}
    c.update(kw)
    return c


@case("C 突破·改立场重合议为放行, 硬数字 tech 带 intraday")
def _():
    from app.services import proposal_service as pp
    blocks = {"fund": {"stance": "看多"}, "timing": {"stance": "中性"},
              "tech": {"phase": "收口等待", "boll_upper": 10.2, "stance": "中性"},
              "consensus": {"route": "观察", "direction": "中性", "route_reason": "等技术面开口"}}
    cache = {"600000.SH": (10.5, 10.0, 2.0)}    # 价站上上轨、量足
    kind = pp._apply_intraday("600000.SH", blocks, _cfg(), NOW, cache)
    assert kind == "breakout"
    assert blocks["tech"]["stance"] == "看多" and blocks["tech"]["phase"] == "盘中开口向上"
    assert blocks["tech"]["intraday"]["kind"] == "breakout"
    assert blocks["consensus"]["direction"] == "看多" and blocks["consensus"]["route"] == "放行"


@case("C 开口未确认·路由改观察带 wait_confirm, 硬数字带 intraday")
def _():
    from app.services import proposal_service as pp
    from app.core import action_engine as ae
    blocks = {"fund": {"stance": "看多"}, "timing": {"stance": "看多"},
              "tech": {"phase": "开口向上", "boll_upper": 10.2, "stance": "看多"},
              "consensus": {"route": "放行", "direction": "看多"}}
    cache = {"600000.SH": (9.5, 10.0, 2.0)}     # 价低于均价 → 未确认
    kind = pp._apply_intraday("600000.SH", blocks, _cfg(), NOW, cache)
    assert kind == "opened_wait"
    assert blocks["consensus"]["route"] == "观察"
    assert blocks["consensus"]["disp"] == ae.DISP_WAIT_CONFIRM
    assert blocks["tech"]["intraday"]["kind"] == "opened_wait"


@case("C 总闸关掉·_intraday_cfg 返回 None (逐字回旧)")
def _():
    from app.services import proposal_service as pp
    o = pp.param_store.get_bool
    pp.param_store.get_bool = lambda k, d=False: (False if k == "PMS_TECH_INTRADAY_CONFIRM"
                                                 else o(k, d))
    try:
        assert pp._intraday_cfg() is None
    finally:
        pp.param_store.get_bool = o


# ================================================================ E 参数
@case("E 参数·盘中确认四键登记, 默认值对")
def _():
    assert ps.RUNTIME_EXTRA["PMS_TECH_INTRADAY_CONFIRM"][0] is True
    assert ps.RUNTIME_EXTRA["PMS_TECH_INTRADAY_BREAKOUT"][0] is True
    assert abs(ps.RUNTIME_EXTRA["PMS_TECH_VOL_CONFIRM"][0] - 1.5) < 1e-9
    assert ps.RUNTIME_EXTRA["PMS_TECH_INTRADAY_WINDOW"][0] == "0945-1430"


@case("E 参数·量比阈值范围校验 0.5 到 10")
def _():
    assert ps._range_check("PMS_TECH_VOL_CONFIRM", 0.3)      # 低于下限 → 报错串
    assert ps._range_check("PMS_TECH_VOL_CONFIRM", 20)       # 高于上限 → 报错串
    assert ps._range_check("PMS_TECH_VOL_CONFIRM", 1.5) is None


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())