src/llm/agents/orchestrator.py

@@ -34,6 +34,11 @@ logger = logging.getLogger(__name__)
 _AGENT_PROVIDER_CACHE: dict[str, tuple[float, LLMProvider]] = {}
 _AGENT_PROVIDER_CACHE_TTL = 60.0
 
+# 当前预测的模型覆盖(调用方显式指定)。用模块级变量而非新增形参,因为
+# run_specialists 会被既有测试 mock,加形参会破坏那些测试的调用签名。
+# 由 predict_match_multi 在进入时 set、退出时 reset。
+_ACTIVE_MODEL_OVERRIDE: str | None = None
+
 
 # ── 5 个专家 agent 定义 ──
 # A=近期状态 B=攻防数据 C=主客因素 D=联赛排名 E=历史交锋
@@ -105,16 +110,19 @@ class MultiPredictResult:
     raw: dict | None = None
 
 
-async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider:
+async def _agent_provider(agent_id: str, *, tier: str, model_override: str | None = None) -> LLMProvider:
     """构造某 agent 专属 provider。
 
     覆盖优先级:
-      模型: AGENT_MODEL_{ID}(运行时) → 层级默认(LLM_SPECIALIST/AGGREGATOR_MODEL) → 全局 LLM_MODEL
+      模型: model_override(调用方显式指定) → AGENT_MODEL_{ID}(运行时) → 层级默认(LLM_SPECIALIST/AGGREGATOR_MODEL) → 全局 LLM_MODEL
       地址/密钥: AGENT_BASE_URL_{ID} / AGENT_API_KEY_{ID}(运行时) → 全局 LLM_BASE_URL / LLM_API_KEY
 
     P3-2: 结果缓存 60 秒,避免每次预测都多次查询运行时配置 DB。
+    注意:model_override 生效时跳过缓存读写 —— 否则带 override 的结果会泄漏给
+    不带 override 的调用(反之亦然),导致跨调用的模型串味。
     """
     cache_key = f"{agent_id}:{tier}"
+    if model_override is None:
         cached = _AGENT_PROVIDER_CACHE.get(cache_key)
         if cached is not None:
             ts, provider = cached
@@ -129,6 +137,9 @@ async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider:
     model = await get_runtime_value(f"{pfx}MODEL")
     if model:
         p.model = model
+    # 调用方显式传入的 model 优先级最高,高于 agent 级与层级默认
+    if model_override:
+        p.model = model_override
     base = await get_runtime_value(f"{pfx}BASE_URL")
     if base:
         p.base_url = base
@@ -136,6 +147,7 @@ async def _agent_provider(agent_id: str, *, tier: str) -> LLMProvider:
     if key:
         p.api_key = key
 
+    if model_override is None:
         _AGENT_PROVIDER_CACHE[cache_key] = (time.time(), p)
         # 简单淘汰:超过 20 条时清空(60s TTL 下不会累积太多)
         if len(_AGENT_PROVIDER_CACHE) > 20:
@@ -152,9 +164,14 @@ async def run_specialists(
     """并行执行 5 个专家 agent。fail-open: 单个失败不影响其他。
 
     before: 数据截止时间(回测防泄漏)。None 表示不限制。
+
+    模型覆盖通过 _ACTIVE_MODEL_OVERRIDE 传递,而不是新增形参:既有测试
+    会 mock 本函数(见 tests/test_agent_weights_persist.py),加形参会破坏
+    它们的调用签名。predict_match_multi 在调用前后 set/reset 该变量。
     """
+    model_override = _ACTIVE_MODEL_OVERRIDE
     tasks = [
-        _run_one(spec, header, await _agent_provider(spec.name, tier="specialist"), version=version, before=before)
+        _run_one(spec, header, await _agent_provider(spec.name, tier="specialist", model_override=model_override), version=version, before=before)
         for spec in SPECIALIST_SPECS
     ]
     results = await asyncio.gather(*tasks, return_exceptions=True)
@@ -222,11 +239,13 @@ async def predict_match_multi(
     version: str = "v1",
     backtest: bool = False,
     cutoff_at=None,
+    model: str | None = None,
 ) -> MultiPredictResult:
     """多 agent 端到端预测: 切片 → 并行专家 → 终裁 → 存库。
 
     backtest: 回测模式。True 时 cutoff 自动设为 match_dt - 1 天。
     cutoff_at: 显式截止时间(优先于 backtest 自动计算)。
+    model: 显式指定模型,优先于 agent 级/层级默认配置(single 模式语义一致)。
     """
     start = time.perf_counter()
 
@@ -247,7 +266,14 @@ async def predict_match_multi(
     prediction_cutoff_at = cutoff
 
     # 2. 并行专家(各自独立配置,使用统一 cutoff)
+    # 显式 model 覆盖通过模块级变量下传,避免改动 run_specialists 的签名
+    global _ACTIVE_MODEL_OVERRIDE
+    _prev_override = _ACTIVE_MODEL_OVERRIDE
+    _ACTIVE_MODEL_OVERRIDE = model
+    try:
         reports = await run_specialists(header, version=version, before=cutoff)
+    finally:
+        _ACTIVE_MODEL_OVERRIDE = _prev_override
 
     # 2.5 统计有效专家报告数量
     ok_reports = [r for r in reports if r.status == "ok"]
@@ -279,7 +305,7 @@ async def predict_match_multi(
         }
         agg_prompt_tokens = 0
         agg_completion_tokens = 0
-        aggregator_model = settings.LLM_MODEL  # 占位,无实际 LLM 调用
+        aggregator_model = model or settings.LLM_MODEL  # 占位,无实际 LLM 调用
 
     latency_ms = int((time.perf_counter() - start) * 1000)
 
@@ -347,7 +373,7 @@ async def predict_match_multi(
             "预测完成 match=%s mode=%s status=%s pred=%s:%s (%s) latency=%sms, experts=%d/%d, prediction_id=%s",
             match_id, "multi", pred_status,
             pred.pred_home_goals, pred.pred_away_goals, pred.pred_1x2,
-            latency_ms, ok_reports, len(reports), pred.id,
+            latency_ms, len(ok_reports), len(reports), pred.id,
         )
 
         return MultiPredictResult(

src/llm/backtest.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 import asyncio
 import logging
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timezone
 
 from sqlalchemy import select
 from sqlalchemy.orm import selectinload
@@ -79,6 +79,35 @@ class BacktestSummary:
     results: list[BacktestMatchResult] = field(default_factory=list)
 
 
+def _parse_date_bound(value, *, end_of_day: bool) -> datetime | None:
+    """把日期入参解析成可与 timestamptz 列比较的 aware datetime。
+
+    支持 "YYYY-MM-DD"、完整 ISO 串(可带偏移)以及 datetime 对象;None 原样返回。
+    裸日期按 UTC 锚定 —— Match.match_date 是 timestamptz,naive datetime 与之
+    比较会因时区不同而偏移;start 取当天 00:00,end 取当天 23:59:59.999999
+    (闭区间,否则最后一天会被静默排除)。
+
+    解析失败抛 ValueError(不静默吞掉):fromisoformat 对非法输入统一抛 ValueError,
+    这里包一层以带上原始值,便于定位是哪个参数写错了。
+    """
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        dt = value
+    else:
+        try:
+            dt = datetime.fromisoformat(str(value))
+        except ValueError as e:
+            raise ValueError(f"无法解析日期: {value!r}(应为 YYYY-MM-DD 或 ISO 格式)") from e
+
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    # 闭区间上界:日期串解析出来是 00:00,取当天末刻才能让最后一天参与回测
+    if end_of_day:
+        dt = dt.replace(hour=23, minute=59, second=59, microsecond=999999)
+    return dt
+
+
 async def _get_historical_matches(
     db,
     *,
@@ -106,10 +135,12 @@ async def _get_historical_matches(
     )
     if league_id is not None:
         stmt = stmt.where(Match.league_id == league_id)
-    if date_from:
+    dt_from = _parse_date_bound(date_from, end_of_day=False)
-        stmt = stmt.where(Match.match_date >= date_from)
+    if dt_from is not None:
-    if date_to:
+        stmt = stmt.where(Match.match_date >= dt_from)
-        stmt = stmt.where(Match.match_date <= date_to)
+    dt_to = _parse_date_bound(date_to, end_of_day=True)
+    if dt_to is not None:
+        stmt = stmt.where(Match.match_date <= dt_to)
 
     stmt = stmt.order_by(Match.match_date.desc()).limit(limit)
     result = await db.execute(stmt)

src/llm/predict.py

@@ -187,13 +187,14 @@ async def predict_match(
         )
     from src.llm.agents.orchestrator import predict_match_multi
 
-    # 回测参数完整传递到 multi-agent 路径
+    # 回测参数 + 模型覆盖完整传递到 multi-agent 路径
     return await predict_match_multi(
         match_id,
         provider=provider,
         version=(prompt_version or "v1").removeprefix("multi_"),
         backtest=backtest,
         cutoff_at=cutoff_at,
+        model=model,
     )
 
 

tests/test_review_required_fixes.py

@@ -0,0 +1,452 @@
+"""回归测试: 代码评审确认的 5 个缺陷修复(R1-R5)。
+
+R1 429 key 轮换路径调用不存在的 _km → NameError(且是凭证脱敏点)
+R2 ingest_bzzoiro_standings 被截断,永不写 standings 表
+R3 orchestrator 完成日志把 list 喂给 %d → logging TypeError
+R4 mode="multi" 静默丢弃调用方传入的 model
+R5 回测把字符串日期直接与 timestamptz 列比较
+
+R2 采用行为测试(假 db + monkeypatch 抓取函数),其余为单元/结构断言。
+"""
+from __future__ import annotations
+
+import inspect
+import logging
+import pathlib
+
+import pytest
+
+from src.data.key_ring import _mask
+from src.llm import backtest as bt_mod
+from src.llm.agents import orchestrator as orch_mod
+
+_REPO_ROOT = pathlib.Path(__file__).resolve().parents[1]
+_BZZOIRO_SRC = _REPO_ROOT / "src" / "data" / "bzzoiro.py"
+
+# 在导入期就抓取真实的 _agent_provider。
+# 原因: tests/test_multi_agent_cutoff.py:52/87/117 会直接
+#   orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test")
+# 且不做清理(既有测试,本次任务不允许改动),导致模块属性在整套测试跑完后
+# 被永久替换成同步 lambda。导入期快照可以规避这种跨测试污染。
+_REAL_AGENT_PROVIDER = orch_mod._agent_provider
+
+
+def _bzzoiro_source() -> str:
+    return _BZZOIRO_SRC.read_text(encoding="utf-8")
+
+
+# ============================================================
+# R1 — _km → _mask
+# ============================================================
+
+class TestR1KeyMasking:
+    def test_mask_is_importable_from_bzzoiro(self):
+        """修复点:bzzoiro 通过 key_ring 复用它,而不是本地重实现。"""
+        import src.data.bzzoiro as bz
+
+        assert bz._mask("abcd1234efgh5678") == "abcd...5678"
+
+    def test_mask_long_key_shows_head_and_tail(self):
+        assert _mask("abcd1234efgh5678") == "abcd...5678"
+
+    def test_mask_short_key_hides_middle(self):
+        assert _mask("short") == "sh***"
+
+    def test_no_km_call_remains_in_bzzoiro_source(self):
+        """源码守卫: _km 在整个代码库不存在,这行一旦执行必抛 NameError。
+
+        这是 NameError 类缺陷(静态即可判定),且位于凭证脱敏日志行上,
+        因此用源码文本守卫是恰当的,而不是只测运行时路径。
+        """
+        assert "_km(" not in _bzzoiro_source()
+
+
+# ============================================================
+# R2 — 积分榜 upsert(行为测试)
+# ============================================================
+
+class _FakeScalars:
+    def __init__(self, items):
+        self._items = list(items)
+
+    def all(self):
+        return list(self._items)
+
+    def __iter__(self):
+        return iter(self._items)
+
+    def scalar_one_or_none(self):
+        return self._items[0] if self._items else None
+
+
+class _FakeResult:
+    def __init__(self, items):
+        self._items = list(items)
+
+    def scalars(self):
+        return _FakeScalars(self._items)
+
+    def scalar_one_or_none(self):
+        return self._items[0] if self._items else None
+
+
+class _FakeDb:
+    """极简假 db: 记录 add/flush,按队列返回预置查询结果。
+
+    只实现 ingest_bzzoiro_standings 真正用到的部分:
+      - execute(...) → 依次弹出 _results 里的结果
+      - add(obj)     → 记录
+      - flush()      → 给尚无 id 的对象补一个自增 id(模拟 DB 回填主键)
+    """
+
+    def __init__(self, results=None):
+        self._results = list(results or [])
+        self.added: list = []
+        self.flush_count = 0
+        self._next_id = 1000
+
+    async def execute(self, _stmt):
+        if self._results:
+            return self._results.pop(0)
+        return _FakeResult([])
+
+    def add(self, obj):
+        self.added.append(obj)
+
+    async def flush(self):
+        self.flush_count += 1
+        for obj in self.added:
+            if getattr(obj, "id", None) is None:
+                self._next_id += 1
+                obj.id = self._next_id
+
+
+async def test_r2_standings_actually_upserts(monkeypatch):
+    """行为测试: 喂一份积分榜 payload,断言真的构造了 Standing 且计数 > 0。"""
+    import src.data.bzzoiro as bz
+    from src.db.models import League, Standing, Team
+
+    payload = {
+        "season": {"start_date": "2025-08-01", "end_date": "2026-05-31"},
+        "standings": [
+            {
+                "position": 1, "team_name": "Arsenal FC",
+                "played": 10, "won": 8, "drawn": 1, "lost": 1,
+                "gf": 22, "ga": 8, "gd": 14, "pts": 25,
+                "xgf": 18.5, "xga": 9.1, "form": "WWDLW",
+                "zone": {"key": "champions_league", "label": "Champions League"},
+            },
+            {
+                "position": 2, "team_name": "Chelsea FC",
+                "played": 10, "won": 6, "drawn": 2, "lost": 2,
+                "gf": 18, "ga": 12, "gd": 6, "pts": 20,
+            },
+        ],
+    }
+
+    async def _fake_fetch(league_code, season=None):
+        return payload
+
+    monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _fake_fetch, raising=True)
+
+    # 查询顺序: League 命中(避免建联赛) → Team 预载(空) → 每行 Standing(未命中)
+    league = League(code="EPL", name="Premier League", country="England")
+    league.id = 42
+    db = _FakeDb(results=[_FakeResult([league]), _FakeResult([])])
+
+    result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"])
+
+    assert result["errors"] == []
+    assert result["total_upserted"] == 2, result
+    assert result["leagues"]["EPL"]["rows"] == 2
+    assert result["leagues"]["EPL"]["upserted"] == 2
+    # 两支球队都是新建的
+    assert result["leagues"]["EPL"]["teams_created"] == 2
+
+    standings = [o for o in db.added if isinstance(o, Standing)]
+    assert len(standings) == 2, "应真的构造 Standing 行"
+    assert all(isinstance(o, (Standing, Team)) for o in db.added)
+
+    first = standings[0]
+    assert first.league_id == 42
+    assert first.season == "2025-2026"          # 8 月起 → 跨年标签
+    assert first.position == 1
+    assert first.points == 25
+    assert first.xg_for == 18.5
+    assert first.zone == "Champions League"     # 优先取 label
+
+
+async def test_r2_standings_upsert_updates_existing(monkeypatch):
+    """行为测试: 已存在同 (league, season, team) 时应就地更新而非新增。"""
+    import src.data.bzzoiro as bz
+    from src.db.models import League, Standing
+
+    payload = {
+        "season": {"start_date": "2025-08-01", "end_date": "2026-05-31"},
+        "standings": [{"position": 1, "team_name": "Arsenal FC", "pts": 30}],
+    }
+
+    async def _fake_fetch(league_code, season=None):
+        return payload
+
+    monkeypatch.setattr(bz, "fetch_bzzoiro_standings", _fake_fetch, raising=True)
+
+    league = League(code="EPL", name="Premier League", country="England")
+    league.id = 42
+    team = __import__("src.db.models", fromlist=["Team"]).Team(name="Arsenal FC", name_zh="阿森纳")
+    team.id = 7
+
+    existing = Standing(league_id=42, season="2025-2026", team_id=7, position=9)
+    existing.points = 1
+
+    # 查询顺序: League → Team 预载(命中) → Standing 查询(命中已有行)
+    db = _FakeDb(results=[_FakeResult([league]), _FakeResult([team]), _FakeResult([existing])])
+
+    result = await bz.ingest_bzzoiro_standings(db, leagues=["EPL"])
+
+    assert result["total_upserted"] == 1
+    assert existing.points == 30, "已有行应被就地更新"
+    assert result["leagues"]["EPL"]["teams_created"] == 0
+    # 不应新增 Standing(只有 league/team 层面的 add)
+    assert not [o for o in db.added if isinstance(o, Standing)]
+
+
+def test_r2_source_contains_real_upsert_loop():
+    """结构断言(行为测试之外的兜底): 确认函数未被截断。"""
+    import src.data.bzzoiro as bz
+
+    src = inspect.getsource(bz.ingest_bzzoiro_standings)
+    assert "total_upserted" in src
+    assert 'result["total_upserted"] +=' in src, "total_upserted 必须真的被累加"
+    assert "Standing(" in src, "必须真的构造 Standing"
+    assert "select(Standing)" in src, "必须查询已有快照以决定 insert/update"
+
+
+# ============================================================
+# R3 — logging 参数类型
+# ============================================================
+
+def test_r3_completion_log_record_formats_without_raising():
+    """%d 占位符拿到 list 时 logging 会抛 TypeError;修复后应为计数。"""
+    fmt = (
+        "预测完成 match=%s mode=%s status=%s pred=%s:%s (%s) latency=%sms, "
+        "experts=%d/%d, prediction_id=%s"
+    )
+    ok_reports = ["a", "b", "c"]
+    reports = ["a", "b", "c", "d", "e"]
+
+    record = logging.LogRecord(
+        "src.llm.agents.orchestrator", logging.INFO, __file__, 1, fmt,
+        (999, "multi", "success", 2, 1, "1", 100, len(ok_reports), len(reports), 7),
+        None,
+    )
+    msg = record.getMessage()  # 修复前此处抛 TypeError
+    assert "experts=3/5" in msg
+
+    # 反证: 原缺陷写法(直接传 list)确实会炸,确保这条测试真的有鉴别力
+    bad = logging.LogRecord(
+        "x", logging.INFO, __file__, 1, fmt,
+        (999, "multi", "success", 2, 1, "1", 100, ok_reports, len(reports), 7),
+        None,
+    )
+    with pytest.raises(TypeError):
+        bad.getMessage()
+
+
+# ============================================================
+# R4 — multi 模式透传 model
+# ============================================================
+
+_ORCH_SRC_PATH = _REPO_ROOT / "src" / "llm" / "agents" / "orchestrator.py"
+
+
+def _orchestrator_source() -> str:
+    """直接读源码,而不是 inspect.getsource(模块属性)。
+
+    既有测试(如 test_agent_weights_persist.py)会在运行期把
+    orch_mod._agent_provider 换成 lambda/MagicMock,导致 inspect.getsource
+    拿到的是 mock 的定义。本文件的断言针对真实源码,故从磁盘读取。
+    """
+    return _ORCH_SRC_PATH.read_text(encoding="utf-8")
+
+
+def test_r4_predict_match_multi_accepts_model():
+    """predict_match_multi 必须接受 model 且默认 None(向后兼容既有调用)。"""
+    src_fn = _orchestrator_source().split("async def predict_match_multi(", 1)[1]
+    header = src_fn.split(") -> MultiPredictResult:", 1)[0]
+    assert "model: str | None = None" in header, header
+
+
+def test_r4_agent_provider_accepts_model_override():
+    src_fn = _orchestrator_source().split("async def _agent_provider(", 1)[1]
+    header = src_fn.split(") -> LLMProvider:", 1)[0]
+    assert "model_override: str | None = None" in header, header
+
+
+async def test_r4_dispatch_forwards_model_to_multi(monkeypatch):
+    """行为测试: predict_match(mode=multi, model=...) 必须把 model 送到 multi 路径。"""
+    from src.llm import predict as predict_mod
+
+    captured: dict = {}
+
+    async def _fake_multi(match_id, **kwargs):
+        captured["match_id"] = match_id
+        captured.update(kwargs)
+        return "SENTINEL"
+
+    # predict.py:188 是函数内 `from ... import`,import 发生在调用时,
+    # 所以必须打在 orchestrator 模块的属性上。
+    monkeypatch.setattr(orch_mod, "predict_match_multi", _fake_multi, raising=True)
+
+    out = await predict_mod.predict_match(999, model="my-model-x", mode="multi")
+
+    assert out == "SENTINEL"
+    assert captured["model"] == "my-model-x"
+    assert captured["match_id"] == 999
+
+
+async def test_r4_specialist_provider_honors_model_override(monkeypatch):
+    """行为测试: model_override 应覆盖 agent 级/层级默认模型。"""
+    from src.llm.provider import LLMProvider
+    import src.llm.agents.orchestrator as real_orch
+
+    async def _fake_default():
+        return LLMProvider(api_key="k", base_url="http://x", model="default-model", timeout=1.0)
+
+    async def _fake_runtime(key):
+        if key.endswith("_MODEL"):
+            return "agent-level-model"
+        return None
+
+    monkeypatch.setattr(real_orch, "get_default_provider", _fake_default, raising=True)
+    monkeypatch.setattr(real_orch, "get_runtime_value", _fake_runtime, raising=True)
+    monkeypatch.setattr(real_orch, "settings", _settings_with_specialist_model(), raising=True)
+    monkeypatch.setattr(real_orch, "_AGENT_PROVIDER_CACHE", {}, raising=True)
+
+    overridden = await _REAL_AGENT_PROVIDER("form", tier="specialist", model_override="OVERRIDE")
+    assert overridden.model == "OVERRIDE"
+
+    # 不带 override 时仍走原优先级(agent 级运行时配置)
+    real_orch._AGENT_PROVIDER_CACHE.clear()
+    normal = await _REAL_AGENT_PROVIDER("form", tier="specialist")
+    assert normal.model == "agent-level-model"
+
+
+def _settings_with_specialist_model():
+    class _S:
+        LLM_SPECIALIST_MODEL = "tier-specialist-model"
+        LLM_AGGREGATOR_MODEL = "tier-aggregator-model"
+
+    return _S()
+
+
+def test_r4_override_does_not_pollute_cache():
+    """override 结果不得写入 60s provider 缓存(否则会串味给普通调用)。"""
+    src = _orchestrator_source().split("async def _agent_provider(", 1)[1]
+    src = src.split("async def run_specialists(", 1)[0]
+    assert "if model_override is None:" in src
+    assert "_AGENT_PROVIDER_CACHE[cache_key]" in src
+
+
+async def test_r4_dispatch_sets_override_for_specialists(monkeypatch):
+    """行为测试: predict_match_multi 应把 model 放进 _ACTIVE_MODEL_OVERRIDE,
+    并在 run_specialists 执行期间对 specialist 生效(退出后复位)。
+
+    参照 tests/test_multi_agent_degraded.py 的 stub 方式,避免触碰真实 DB。
+    """
+    from unittest.mock import MagicMock
+
+    from src.db.unit_of_work import get_uow
+
+    seen: dict = {}
+
+    async def _fake_header(match_id):
+        h = MagicMock()
+        h.match_id = match_id
+        h.match_dt = None
+        return h
+
+    async def _fake_specialists(header, *, version, before):
+        seen["override_during_run"] = orch_mod._ACTIVE_MODEL_OVERRIDE
+        return []
+
+    async def _fake_upsert(session, **kw):
+        p = MagicMock()
+        p.id = 1
+        p.provider = "test"
+        p.model = kw.get("model")
+        p.prompt_version = "v1"
+        p.pred_home_goals = None
+        p.pred_away_goals = None
+        p.pred_1x2 = None
+        p.alt_pred_home_goals = None
+        p.alt_pred_away_goals = None
+        p.subjective_confidence = None
+        p.reasoning = ""
+        p.agent_outputs = []
+        p.agent_weights = {}
+        p.prompt_tokens = 0
+        p.completion_tokens = 0
+        return p
+
+    class _FakeUow:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *a):
+            return False
+
+        async def get(self, cls, id):
+            return MagicMock()
+
+    monkeypatch.setattr(orch_mod, "load_match_header", _fake_header, raising=True)
+    monkeypatch.setattr(orch_mod, "run_specialists", _fake_specialists, raising=True)
+    monkeypatch.setattr(orch_mod, "_upsert_prediction", _fake_upsert, raising=True)
+    monkeypatch.setattr(orch_mod, "get_uow", _FakeUow, raising=True)
+    monkeypatch.setattr(orch_mod, "_ACTIVE_MODEL_OVERRIDE", None, raising=False)
+
+    assert get_uow is not None  # 确保 import 生效,session 未被真实打开
+
+    await orch_mod.predict_match_multi(999, model="OVERRIDE-X")
+
+    assert seen["override_during_run"] == "OVERRIDE-X"
+    assert orch_mod._ACTIVE_MODEL_OVERRIDE is None, "退出后必须复位"
+
+
+# ============================================================
+# R5 — 回测日期解析
+# ============================================================
+
+class TestR5DateBound:
+    def test_plain_date_start_is_start_of_day_utc(self):
+        dt = bt_mod._parse_date_bound("2026-01-01", end_of_day=False)
+        assert dt is not None
+        assert dt.tzinfo is not None
+        assert (dt.year, dt.month, dt.day) == (2026, 1, 1)
+        assert (dt.hour, dt.minute, dt.second) == (0, 0, 0)
+
+    def test_plain_date_end_is_inclusive_end_of_day(self):
+        """闭区间: 结束日必须取当天末刻,否则最后一天被静默排除。"""
+        dt = bt_mod._parse_date_bound("2026-01-01", end_of_day=True)
+        assert dt is not None
+        assert (dt.hour, dt.minute, dt.second) == (23, 59, 59)
+        assert dt.microsecond == 999999
+
+    def test_full_iso_string_is_parsed(self):
+        dt = bt_mod._parse_date_bound("2026-01-01T12:30:00+08:00", end_of_day=False)
+        assert dt is not None
+        assert dt.utcoffset() is not None
+
+    def test_none_returns_none(self):
+        assert bt_mod._parse_date_bound(None, end_of_day=False) is None
+        assert bt_mod._parse_date_bound(None, end_of_day=True) is None
+
+    def test_datetime_passthrough(self):
+        from datetime import datetime, timezone
+
+        given = datetime(2026, 5, 5, 6, 0, tzinfo=timezone.utc)
+        assert bt_mod._parse_date_bound(given, end_of_day=False) == given
+
+    def test_invalid_input_raises_value_error(self):
+        with pytest.raises(ValueError):
+            bt_mod._parse_date_bound("not-a-date", end_of_day=False)