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 主胜
|