tradingSystem/scripts/test_batch33_units.py

# -*- coding: utf-8 -*-
"""页面收尾包 · 单票研究面接口 (2026-09-14)。后端离线单测, 不连库。

研究面接口补齐六块 (公司/基本面/技术面/择时/合议/来源), 页面抽屉四芯片、候选栏芯片、
管理视图研究面列消费它。前端三处由两道页面守卫 (wiring/enum) 与真机视觉判收兜, 这里测后端:
  A 研究面来源解析 _research_st: 持仓票 / 候选票 / 都没有 三情形。
  B 研究面接口 api_research: 六块齐全且来源逐路标明; 无读数票 company 为 None、来源标不在持仓也不在候选。
"""
import os
import sys
import traceback

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

from app.web import main                                  # noqa: E402
from app.services import (consensus_service, tech_service,  # noqa: E402
                          logic_state_service, plan_feed)

RESULTS = []


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


class _patch:
    """按 (模块, 属性名, 替身) 列表打桩, with 结束自动还原。"""

    def __init__(self, *specs):
        self.specs = specs
        self._saved = []

    def __enter__(self):
        for mod, attr, fn in self.specs:
            self._saved.append((mod, attr, getattr(mod, attr)))
            setattr(mod, attr, fn)
        return self

    def __exit__(self, *a):
        for mod, attr, orig in reversed(self._saved):
            setattr(mod, attr, orig)
        return False


# ================================================================ A 来源解析
@case("A _research_st·持仓票 → held (逻辑状态映射有买方评析)")
def _():
    with _patch((logic_state_service, "state_map",
                 lambda: {"600000.SH": {"company_review": {"overall": "好"},
                                        "company_review_text": "x"}})):
        st, src = main._research_st("600000.SH")
    assert src == "held" and (st.get("company_review") or {}).get("overall") == "好"


@case("A _research_st·候选票 → candidate (不在持仓, 当日计划主榜有)")
def _():
    with _patch((logic_state_service, "state_map", lambda: {}),
                (plan_feed, "get_plan",
                 lambda **k: {"main": [{"ts_code": "600001.SH",
                                        "company_review": {"overall": "中"},
                                        "company_review_text": "y"}], "observe": []})):
        st, src = main._research_st("600001.SH")
    assert src == "candidate" and (st.get("company_review") or {}).get("overall") == "中"


@case("A _research_st·都没有 → none")
def _():
    with _patch((logic_state_service, "state_map", lambda: {}),
                (plan_feed, "get_plan", lambda **k: {"main": [], "observe": []})):
        st, src = main._research_st("600002.SH")
    assert st is None and src == "none"


# ================================================================ B 研究面接口
def _mock_assemble(bull=True):
    if bull:
        return lambda row, **k: {"fund": {"stance": "看多", "fact": "质地好"},
                                 "tech": {"stance": "看多", "phase": "趋势多"},
                                 "timing": {"stance": "看多", "nightly": "BUY", "intraday_flip_at": None},
                                 "consensus": {"direction": "看多", "route": "放行", "reason": "三方看多"}}
    return lambda row, **k: {"fund": {"stance": "无读数", "why": "没有买方评析"},
                             "tech": {"stance": "无读数"},
                             "timing": {"stance": "无读数", "no_read_why": "没有昨夜结论"},
                             "consensus": {"direction": "中性", "route": "跳过"}}


@case("B api_research·六块齐全, 来源逐路标 ok")
def _():
    with _patch(
        (logic_state_service, "state_map",
         lambda: {"600000.SH": {"company_review": {"overall": "好"}, "company_review_text": "x"}}),
        (logic_state_service, "company_view",
         lambda st: {"overall": "好", "line": "质地好（…）", "report_url": None}),
        (consensus_service, "nightly_map", lambda codes: {"600000.SH": {"verdict": "BUY", "trade_date": 20260911}}),
        (consensus_service, "state_map", lambda: {}),
        (consensus_service, "assemble", _mock_assemble(True)),
        (tech_service, "research_feed", lambda code: {"tech": {"stance": "看多", "phase": "趋势多", "latest": {}}}),
    ):
        r = main.api_research("600000.SH")
    d = r["data"]
    for k in ("company", "fund", "tech", "timing", "consensus", "sources"):
        assert k in d, k
    assert d["consensus"]["direction"] == "看多"
    assert d["timing"]["trade_date"] == 20260911
    assert d["sources"]["company"] == "ok" and d["sources"]["consensus"] == "ok"
    assert d["sources"]["fund"] == "ok" and d["sources"]["tech"] == "ok"


@case("B api_research·无读数票 → company 为 None, 来源标不在持仓也不在候选")
def _():
    with _patch(
        (logic_state_service, "state_map", lambda: {}),
        (plan_feed, "get_plan", lambda **k: {"main": [], "observe": []}),
        (consensus_service, "nightly_map", lambda codes: {}),
        (consensus_service, "state_map", lambda: {}),
        (consensus_service, "assemble", _mock_assemble(False)),
        (tech_service, "research_feed", lambda code: {"tech": None, "note": "这只票没有技术面读数"}),
    ):
        d = main.api_research("600002.SH")["data"]
    assert d["company"] is None
    assert d["sources"]["company"] == "不在持仓也不在候选"
    assert d["sources"]["fund"] != "ok" and d["sources"]["tech"] != "ok"


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