tradingSystem/scripts/test_batch27_units.py

# -*- coding: utf-8 -*-
"""技术面接入 · 工作包一 (2026-09-11 方案第五节, 台账 008)。全部离线, 不连库不连网。

  A 相位合成 (core/tech_rules): 四种无读数、九个相位逐格、震荡市抑制翻向信号、数翻向。
  B 取数解析 (tech_service.parse_item): 代码归一 SH600000→600000.SH、字段映射、认不出返回 None。
  C 分页取数 (fetch_all_pages): 多页拼接、matched 终止、状态非 OK 抛错 (注入 fetch, 不连网)。
  D 映射 (build_map): 正常合成只覆盖有当天读数的相关票、读数陈旧整份按无读数 (桩掉库与网络)。
  E 参数与口径: PMS_TECH_* 登记进 RUNTIME_EXTRA 与 _RANGES、交易日差口径。
  F 落表 (pull): 接口状态非 OK 整轮不落表、正常落表并推进日期 (桩掉库)。
"""
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.services import param_store                                            # noqa: E402
from app.services import tech_service as ts                                     # 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 _row(dd, *, side="多", pos=0.6, squeeze=False, bw=5.0, bbi="中性区",
         flip=8, quality="OK", reanchor=1, sar=10.0):
    """造一行 pms_tech_daily 读数 (只填 synthesize 用得到的字段)。"""
    return {"data_date": dd, "ts_code": "600000.SH", "sar_side": side, "boll_pos": pos,
            "boll_squeeze": 1 if squeeze else 0, "boll_bw_pct": bw, "bbi_state": bbi,
            "sar_flip_days": flip, "quality": quality, "reanchored": reanchor, "sar_value": sar}


# ================================================================ A 相位合成
@case("A 无读数: 没有当天行 → 弃权, 不折成看空")
def test_noread_empty():
    r = tr.synthesize([])
    assert r["stance"] == "无读数" and r["no_read_why"]


@case("A 无读数: 未除权重锚")
def test_noread_reanchor():
    r = tr.synthesize([_row(20260910, reanchor=0)])
    assert r["stance"] == "无读数" and "除权" in r["reason"]


@case("A 无读数: 数据质量非 OK")
def test_noread_quality():
    r = tr.synthesize([_row(20260910, quality="STALE")])
    assert r["stance"] == "无读数" and "质量" in r["reason"]


@case("A 无读数: 关键读数缺失 (SAR 方向)")
def test_noread_missing():
    r = tr.synthesize([_row(20260910, side="")])
    assert r["stance"] == "无读数"


@case("A 收口等待: 今天收口一律中性")
def test_squeeze():
    r = tr.synthesize([_row(20260910, squeeze=True, side="多", bbi="多头区")])
    assert r["phase"] == "收口等待" and r["stance"] == "中性"


@case("A 趋势多: SAR 多 + 多空布林线多头区 = 看多强")
def test_trend_long_strong():
    r = tr.synthesize([_row(20260910, side="多", bbi="多头区", flip=8)])
    assert r["phase"] == "趋势多" and r["stance"] == "看多" and r["strength"] == "强"


@case("A 趋势多: SAR 多 + 中性区 = 看多弱")
def test_trend_long_weak():
    r = tr.synthesize([_row(20260910, side="多", bbi="中性区", flip=8)])
    assert r["phase"] == "趋势多" and r["stance"] == "看多" and r["strength"] == "弱"


@case("A 趋势空: SAR 空 + 空头区 = 看空强")
def test_trend_short_strong():
    r = tr.synthesize([_row(20260910, side="空", bbi="空头区", flip=8)])
    assert r["phase"] == "趋势空" and r["stance"] == "看空" and r["strength"] == "强"


@case("A 趋势空: SAR 空 + 中性区 = 看空弱")
def test_trend_short_weak():
    r = tr.synthesize([_row(20260910, side="空", bbi="中性区", flip=8)])
    assert r["phase"] == "趋势空" and r["stance"] == "看空" and r["strength"] == "弱"


@case("A 分歧: SAR 多 + 空头区 = 中性")
def test_diverge_long():
    r = tr.synthesize([_row(20260910, side="多", bbi="空头区", flip=8)])
    assert r["phase"] == "分歧" and r["stance"] == "中性"


@case("A 分歧: SAR 空 + 多头区 = 中性")
def test_diverge_short():
    r = tr.synthesize([_row(20260910, side="空", bbi="多头区", flip=8)])
    assert r["phase"] == "分歧" and r["stance"] == "中性"


@case("A 转多: SAR 刚翻多 + 多空布林线非空头 = 看多弱")
def test_flip_long():
    r = tr.synthesize([_row(20260910, side="多", bbi="中性区", flip=1)])
    assert r["phase"] == "转多" and r["stance"] == "看多" and r["strength"] == "弱"


@case("A 转空: 刚翻空 + 非空头非贴轨 = 看空弱 (未确认)")
def test_flip_short_unconfirmed():
    r = tr.synthesize([_row(20260910, side="空", bbi="中性区", pos=0.5, flip=1)])
    assert r["phase"] == "转空" and r["confirm"] is False and r["strength"] == "弱"


@case("A 转空: 刚翻空 + 空头区 = 看空强 (确认)")
def test_flip_short_confirmed():
    r = tr.synthesize([_row(20260910, side="空", bbi="空头区", flip=1)])
    assert r["phase"] == "转空" and r["confirm"] is True and r["strength"] == "强"


@case("A 开口向上: 收口后放开 + 上半部 + SAR 多 = 看多强")
def test_open_up():
    rows = [_row(20260909, squeeze=True, bw=5.0, side="多", bbi="中性区", flip=8),
            _row(20260910, squeeze=False, bw=6.2, pos=0.6, side="多", bbi="中性区", flip=8)]
    r = tr.synthesize(rows)
    assert r["phase"] == "开口向上" and r["stance"] == "看多" and r["strength"] == "强"


@case("A 开口向下: 收口后放开 + 下半部 + SAR 空 = 看空强")
def test_open_down():
    rows = [_row(20260909, squeeze=True, bw=5.0, side="空", bbi="中性区", flip=8),
            _row(20260910, squeeze=False, bw=6.2, pos=0.4, side="空", bbi="中性区", flip=8)]
    r = tr.synthesize(rows)
    assert r["phase"] == "开口向下" and r["stance"] == "看空" and r["strength"] == "强"


@case("A 带宽没扩够不算开口, 落回趋势")
def test_open_not_enough():
    rows = [_row(20260909, squeeze=True, bw=5.0, side="多", bbi="中性区", flip=8),
            _row(20260910, squeeze=False, bw=5.5, pos=0.6, side="多", bbi="中性区", flip=8)]
    r = tr.synthesize(rows)
    assert r["phase"] != "开口向上" and r["phase"] == "趋势多"


@case("A 震荡市里刚翻空不判转空 (SAR 翻向不当信号), 按趋势走")
def test_choppy_suppresses_flip():
    # 20 行交替方向 → 翻向 19 次, 远超 4 → 震荡市; 今日刚翻空 (flip=1)
    flippy = [_row(20260800 + i, side=("多" if i % 2 == 0 else "空"), bbi="中性区", flip=1)
              for i in range(20)]
    r = tr.synthesize(flippy)
    assert r["choppy"] is True
    assert r["phase"] != "转空", "震荡市里刚翻空不该判转空"
    assert r["phase"] == "趋势空" and r["stance"] == "看空"


@case("A count_sar_flips: 数相邻方向变化")
def test_count_flips():
    rows = [_row(i, side=("多" if i % 2 == 0 else "空")) for i in range(5)]
    assert tr.count_sar_flips(rows) == 4
    assert tr.count_sar_flips([_row(0, side="多"), _row(1, side="多")]) == 0


# ================================================================ B 取数解析
@case("B parse_item: 代码归一 + 字段映射 + 认不出返回 None")
def test_parse_item():
    it = {"stock_code": "SH600000", "stock_name": "浦发银行",
          "boll": {"upper": 9.4, "mid": 9.1, "lower": 8.9, "bandwidth_pct": 5.4,
                   "pos": 0.85, "state": "带内偏上", "squeeze": True},
          "bbiboll": {"bbi": 9.2, "state": "多头区", "dist_pct": 1.0},
          "sar": {"value": 9.1, "side": "多", "flip_days": 8, "dist_pct": 2.5},
          "reanchored": True, "in_pool": False, "quality": "OK",
          "bars_used": 120, "last_bar_date": 20260910, "bars_lag": 0}
    r = ts.parse_item(it, 20260910, "tech_view_v1")
    assert r["ts_code"] == "600000.SH" and r["stock_name"] == "浦发银行"
    assert r["boll_squeeze"] == 1 and r["reanchored"] == 1 and r["in_pool"] == 0
    assert r["sar_side"] == "多" and r["bbi_state"] == "多头区" and r["sar_flip_days"] == 8
    assert r["algo_version"] == "tech_view_v1" and r["data_date"] == 20260910
    assert ts.parse_item({"stock_code": ""}, 20260910) is None
    assert ts.parse_item("x", 20260910) is None


# ================================================================ C 分页取数
@case("C fetch_all_pages: 分页拼接 + matched 终止 + 状态非 OK 抛错")
def test_fetch_pages():
    pages = {0: [{"stock_code": "SH600000", "sar": {"side": "多"}, "reanchored": True,
                  "quality": "OK", "boll": {"pos": 0.5}},
                 {"stock_code": "SZ000001", "sar": {"side": "空"}, "reanchored": True,
                  "quality": "OK", "boll": {"pos": 0.4}}],
             2: [{"stock_code": "SH600004", "sar": {"side": "多"}, "reanchored": True,
                  "quality": "OK", "boll": {"pos": 0.6}}]}

    def f(url, params, timeout):
        return {"status": "OK", "data_date": 20260910, "matched": 3, "algo_version": "v1",
                "items": pages.get(params["offset"], [])}
    got = ts.fetch_all_pages(base="x", path="/y", timeout=5, page_size=2, fetch=f)
    assert got["total"] == 3 and got["pages"] == 2, got
    assert {r["ts_code"] for r in got["rows"]} == {"600000.SH", "000001.SZ", "600004.SH"}

    def bad(url, params, timeout):
        return {"status": "DEGRADED", "items": []}
    try:
        ts.fetch_all_pages(base="x", path="/y", timeout=5, fetch=bad)
        assert False, "状态非 OK 应抛 TechFeedError"
    except ts.TechFeedError:
        pass


# ================================================================ D 映射
@case("D build_map: 正常合成, 只覆盖有当天读数的相关票")
def test_build_map_ok():
    P = {"enabled": True, "stale_tdays": 2, "keep_days": 40, "base": "x", "path": "/y",
         "timeout": 20, "rules": dict(tr.DEFAULTS)}
    saved = {}
    with _Patch() as p:
        p(ts, "_params", lambda: P)
        p(ts.tech_repo, "latest_date", lambda: 20260910)
        p(ts, "_relevant_codes", lambda: ["600000.SH", "000001.SZ"])
        p(ts.tech_repo, "history_multi",
          lambda codes, since=0: {"600000.SH": [_row(20260910, side="多", bbi="多头区", flip=8)]})
        p(ts, "_save_map", lambda m: (saved.update(m), {"ok": True})[1])
        out = ts.build_map(now=datetime(2026, 9, 11, 8, 45))
    assert out["ok"] and out["data_date"] == 20260910 and out["codes"] == 2
    assert out["states"] == 1, "000001 今日无读数, 不进映射"
    assert saved["states"]["600000.SH"]["stance"] == "看多"


@case("D build_map: 读数陈旧整份按无读数, 映射置空带原因")
def test_build_map_stale():
    P = {"enabled": True, "stale_tdays": 2, "keep_days": 40, "base": "x", "path": "/y",
         "timeout": 20, "rules": dict(tr.DEFAULTS)}
    saved = {}
    with _Patch() as p:
        p(ts, "_params", lambda: P)
        p(ts.tech_repo, "latest_date", lambda: 20260901)      # 距 9/11 超过 2 个交易日
        p(ts, "_relevant_codes", lambda: ["600000.SH"])
        p(ts, "_save_map", lambda m: (saved.update(m), {"ok": True})[1])
        out = ts.build_map(now=datetime(2026, 9, 11, 8, 45))
    assert not out["ok"] and out["error"] == "读数陈旧"
    assert saved.get("stale") is True and saved["states"] == {}


# ================================================================ E 参数与口径
@case("E 参数登记: PMS_TECH_* 在 RUNTIME_EXTRA 与 _RANGES")
def test_params_registered():
    for k in ("PMS_TECH_ENABLED", "PMS_TECH_API_BASE", "PMS_TECH_API_PATH", "PMS_TECH_TIMEOUT",
              "PMS_TECH_KEEP_DAYS", "PMS_TECH_STALE_TDAYS", "PMS_TECH_SQUEEZE_LOOKBACK",
              "PMS_TECH_OPEN_BW_GROWTH", "PMS_TECH_CHOPPY_FLIPS", "PMS_TECH_FLIP_FRESH_DAYS",
              "PMS_TECH_STATE_MAP"):
        assert k in param_store.RUNTIME_EXTRA, k
    for k in ("PMS_TECH_TIMEOUT", "PMS_TECH_KEEP_DAYS", "PMS_TECH_STALE_TDAYS",
              "PMS_TECH_SQUEEZE_LOOKBACK", "PMS_TECH_OPEN_BW_GROWTH", "PMS_TECH_CHOPPY_FLIPS",
              "PMS_TECH_FLIP_FRESH_DAYS"):
        assert k in param_store._RANGES, k


@case("E _tdays_elapsed: 昨→今 1 个交易日, 同日 0")
def test_tdays():
    assert ts._tdays_elapsed(20260910, 20260911) == 1
    assert ts._tdays_elapsed(20260911, 20260911) == 0


# ================================================================ F 落表
@case("F pull: 接口状态非 OK 整轮不落表")
def test_pull_no_ok():
    P = {"enabled": True, "base": "x", "path": "/y", "timeout": 5, "keep_days": 40,
         "stale_tdays": 2, "rules": dict(tr.DEFAULTS)}
    called = {"upsert": 0}
    with _Patch() as p:
        p(ts, "_params", lambda: P)
        p(ts.tech_repo, "upsert_daily",
          lambda rows: (called.__setitem__("upsert", called["upsert"] + 1), len(rows))[1])
        out = ts.pull(fetch=lambda url, params, timeout: {"status": "ERR", "items": []})
    assert not out["ok"] and called["upsert"] == 0, "状态非 OK 绝不落表"


@case("F pull: 正常落表并推进日期")
def test_pull_ok():
    P = {"enabled": True, "base": "x", "path": "/y", "timeout": 5, "keep_days": 40,
         "stale_tdays": 2, "rules": dict(tr.DEFAULTS)}
    got = {"rows": None}

    def good(url, params, timeout):
        if params["offset"] == 0:
            return {"status": "OK", "data_date": 20260910, "matched": 1, "algo_version": "v1",
                    "items": [{"stock_code": "SH600000", "boll": {"squeeze": False, "pos": 0.6},
                               "bbiboll": {"state": "多头区"},
                               "sar": {"side": "多", "value": 9.1, "flip_days": 8},
                               "reanchored": True, "quality": "OK"}]}
        return {"status": "OK", "data_date": 20260910, "matched": 1, "items": []}
    with _Patch() as p:
        p(ts, "_params", lambda: P)
        p(ts.tech_repo, "upsert_daily", lambda rows: (got.__setitem__("rows", rows), len(rows))[1])
        p(ts.tech_repo, "prune", lambda before: 0)
        out = ts.pull(fetch=good)
    assert out["ok"] and out["data_date"] == 20260910 and out["rows"] == 1
    assert got["rows"][0]["ts_code"] == "600000.SH"


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