WorkBuddy
5c7fdce0a3
- PredictResult 扩展可选字段 mode/agent_outputs/agent_weights/prompt_tokens/completion_tokens - MultiPredictResult 变为 PredictResult 别名(保留 R4 守卫标记与 re-export) - baseline 改返回 PredictResult(修复 backtest 对 baseline AttributeError 的潜伏 bug) - 预测路由单一属性映射,删除全部 isinstance(result, dict) 分支 - _persist_baseline 属性化,baseline upsert 语义不变(prompt_version/token/latency 同前) - TDD: 5 新测试 + test_baseline.py 属性化;全量 257 passed
131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
"""测试极简基线预测:不调用 LLM,基于主客场场均进球估计,写入 prediction 表。
|
|
|
|
运行(需先 pip-sync requirements-dev.txt):
|
|
pytest tests/test_baseline.py -v
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from src.llm.baseline import _avg_goals, predict_baseline
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_avg_goals_no_data_returns_zero():
|
|
"""无历史数据时场均进球为 0(不抛异常)。"""
|
|
class FakeRow:
|
|
avg_goals = None
|
|
cnt = 0
|
|
|
|
class FakeResult:
|
|
def one(self):
|
|
return FakeRow()
|
|
|
|
class FakeSession:
|
|
async def execute(self, stmt):
|
|
return FakeResult()
|
|
|
|
avg = await _avg_goals(FakeSession(), team_id=1, side="home", league_id=1, before=None)
|
|
assert avg == 0.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_avg_goals_with_data():
|
|
"""有数据时返回正确均值。"""
|
|
class FakeRow:
|
|
avg_goals = 1.5
|
|
cnt = 10
|
|
|
|
class FakeResult:
|
|
def one(self):
|
|
return FakeRow()
|
|
|
|
class FakeSession:
|
|
async def execute(self, stmt):
|
|
return FakeResult()
|
|
|
|
avg = await _avg_goals(FakeSession(), team_id=1, side="home", league_id=1, before=None)
|
|
assert avg == 1.5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_predict_baseline_no_llm():
|
|
"""基线预测不调用 LLM(provider=model=baseline),latency_ms=0。"""
|
|
captured = {}
|
|
|
|
async def fake_avg(db, *, team_id, side, league_id, before):
|
|
captured[f"{side}_{team_id}"] = True
|
|
return 2.4 if side == "home" else 1.6
|
|
|
|
class FakeMatch:
|
|
id = 1
|
|
match_id = 1
|
|
home_team_id = 10
|
|
away_team_id = 20
|
|
league_id = 1
|
|
match_status = "scheduled"
|
|
|
|
with patch("src.llm.baseline._avg_goals", fake_avg), \
|
|
patch("src.llm.baseline.AsyncSessionLocal") as SLC:
|
|
class FakeSession:
|
|
async def get(self, cls, mid):
|
|
return FakeMatch()
|
|
class FakeCM:
|
|
async def __aenter__(self):
|
|
return FakeSession()
|
|
async def __aexit__(self, *a):
|
|
return None
|
|
SLC.return_value = FakeCM()
|
|
|
|
result = await predict_baseline(1)
|
|
|
|
assert result.provider == "baseline"
|
|
assert result.model == "baseline"
|
|
assert result.mode == "baseline"
|
|
assert result.latency_ms == 0
|
|
assert result.prompt_tokens == 0
|
|
assert result.completion_tokens == 0
|
|
# 2.4 → round = 2, 1.6 → round = 2 → 平局 X
|
|
assert result.pred_home_goals == 2.0
|
|
assert result.pred_away_goals == 2.0
|
|
assert result.pred_1x2 == "X"
|
|
assert result.subjective_confidence == 0.5
|
|
assert "非投注建议" in result.reasoning
|
|
# 确认未调用任何 LLM 相关模块
|
|
assert "home_10" in captured and "away_20" in captured
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_predict_baseline_clamps_to_range():
|
|
"""预测进球数裁剪到 [0, 10]。"""
|
|
async def fake_avg(db, *, team_id, side, league_id, before):
|
|
return 15.0 if side == "home" else -3.0
|
|
|
|
class FakeMatch:
|
|
id = 2
|
|
match_id = 2
|
|
home_team_id = 10
|
|
away_team_id = 20
|
|
league_id = 1
|
|
match_status = "scheduled"
|
|
|
|
with patch("src.llm.baseline._avg_goals", fake_avg), \
|
|
patch("src.llm.baseline.AsyncSessionLocal") as SLC:
|
|
class FakeSession:
|
|
async def get(self, cls, mid):
|
|
return FakeMatch()
|
|
class FakeCM:
|
|
async def __aenter__(self):
|
|
return FakeSession()
|
|
async def __aexit__(self, *a):
|
|
return None
|
|
SLC.return_value = FakeCM()
|
|
|
|
result = await predict_baseline(2)
|
|
|
|
assert result.pred_home_goals == 10.0 # clamped
|
|
assert result.pred_away_goals == 0.0 # clamped
|
|
assert result.pred_1x2 == "1" # 10:0 主胜
|