shangfangjian/Profeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

全量修复:预测系统正确性、安全性与部署问题

Browse Source 
P0 严重问题修复:
- 修复 form_slice/stats_slice 主客身份反转(历史比赛视角错误)
- 修复 understat.py httpx 未导入导致的 NameError
- 修复 LLM 解析失败时静默产生假成功预测(0-0 平局+置信度0.5)

预测路径修复:
- multi-agent 路径增加 backtest cutoff 透传,回测防泄漏生效
- H2H 切片汇总统计改为从当前主队视角计数
- 预测唯一约束增加 mode+run_type 维度,防止回测覆盖实盘预测

伤停管线修复:
- IntegrityError 后不再整批回滚丢数据(改用逐条 flush)
- return_date 正确解析并写入
- retrieved_at 比较统一用 date() 避免当天数据不可见
- 唯一索引改为 partial unique index(排除 NULL 重复)
- HTTP 缓存 TTL 从 7 天改为 6 小时

安全与连接管理:
- /api/v1/predict 增加内存滑动窗口限流(10次/分钟/IP)
- 预测路由改用短 session 模式,LLM 调用期间不持有 DB 连接

Docker 部署修复:
- 修复 .dockerignore 排除 *.md 导致 COPY README.md 失败
- 容器内 DATABASE_URL 使用 postgres 服务名(非 localhost)
- 启动时自动执行 alembic upgrade head
- 前端改用多阶段构建(Dockerfile.frontend)

新增测试(5个文件,24+用例):
- test_p0_home_away.py: 主客身份反转回归测试
- test_p0_parse_failure.py: LLM 解析失败回归测试
- test_multi_agent_cutoff.py: multi-agent cutoff 透传测试
- test_h2h_perspective.py: H2H 视角测试
- test_injuries_pipeline.py: 伤停管线 5 项修复测试
- test_predict_protection.py: 限流+短 session 测试
- test_prediction_unique_constraint.py: 唯一约束测试

迁移:
- 0012_injuries_partial_unique_and_return_date.py
- 0013_predictions_unique_constraint_mode_run_type.py
This commit is contained in:
Profeto Agent
2026-09-19 06:43:55 +00:00
parent 11efe91ce9
commit bee330f31f
27 changed files with 1666 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_h2h_perspective.py
+134
View File
@@ -0,0 +1,134 @@
"""回归测试: H2H 切片「主队 n 胜」统计视角修复。
验证: 历史交锋汇总必须从「当前主队」视角统计胜/平/负,
而非按「场地主队」统计。
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from src.llm.context_builder import MatchHeader, h2h_slice
def _make_team(tid: int, name: str) -> MagicMock:
t = MagicMock()
t.id = tid
t.name = name
return t
def _make_h2h_match(mid, home_id, away_id, home_goals, away_goals, home_name="H", away_name="A"):
m = MagicMock()
m.id = mid
m.home_team_id = home_id
m.away_team_id = away_id
m.home_goals = home_goals
m.away_goals = away_goals
m.match_date = None
m.home_team = _make_team(home_id, home_name)
m.away_team = _make_team(away_id, away_name)
return m
def _make_header(home_id=1, away_id=2, home_name="利物浦", away_name="阿森纳"):
return MatchHeader(
match_id=999,
home_name=home_name,
away_name=away_name,
league_name="英超",
season="2025-2026",
match_date="2026-01-15 20:00 UTC",
match_dt=None,
stage=None,
home_team_id=home_id,
away_team_id=away_id,
league_id=1,
)
class TestH2HCurrentHomePerspective:
"""H2H 汇总统计必须从当前主队视角出发。"""
@pytest.mark.asyncio
async def test_swapped_home_away_perspective(self):
"""
场景: 当前比赛利物浦(home_id=1) vs 阿森纳(away_id=2)。
历史交锋两场:
1. 利物浦主场 2-0 阿森纳 (home_id=1, away_id=2)
2. 阿森纳主场 3-1 利物浦 (home_id=2, away_id=1)
从利物浦视角: 1胜(2-0) 1负(1-3)。
原bug: 按场地主队统计 → "主队 1胜 0平 1负"(第二场场地主队是阿森纳,赢了),
导致「利物浦横扫」的假象。
"""
header = _make_header(home_id=1, away_id=2, home_name="利物浦", away_name="阿森纳")
matches = [
_make_h2h_match(100, home_id=1, away_id=2, home_goals=2, away_goals=0,
home_name="利物浦", away_name="阿森纳"),
_make_h2h_match(101, home_id=2, away_id=1, home_goals=3, away_goals=1,
home_name="阿森纳", away_name="利物浦"),
]
import src.llm.context_builder as cb
orig = cb._get_h2h
async def mock_get_h2h(db, home_id, away_id, before, *, limit):
return matches
cb._get_h2h = mock_get_h2h
try:
result = await h2h_slice(header, limit=8, before=None)
text = str(result)
print(text)
# 从利物浦视角: 1胜 0平 1负
assert "1胜 0平 1负" in text, f"期望「1胜 0平 1负」,实际:\n{text}"
assert "利物浦" in text, f"应标明当前主队视角:\n{text}"
# 原bug输出: "主队 1胜 0平 1负"(模糊的「主队」,实际是场地主队)
# 修复后: "从当前主队 利物浦 视角: 1胜 0平 1负"
assert "从当前主队" in text, f"应标明「从当前主队」视角:\n{text}"
finally:
cb._get_h2h = orig
@pytest.mark.asyncio
async def test_all_home_wins_from_current_perspective(self):
"""
当前主队所有交锋都是主场且全胜 → 全部计为当前主队胜。
"""
header = _make_header(home_id=1, away_id=2, home_name="曼城", away_name="诺维奇")
matches = [
_make_h2h_match(200, home_id=1, away_id=2, home_goals=3, away_goals=0,
home_name="曼城", away_name="诺维奇"),
_make_h2h_match(201, home_id=1, away_id=2, home_goals=2, away_goals=1,
home_name="曼城", away_name="诺维奇"),
]
import src.llm.context_builder as cb
orig = cb._get_h2h
cb._get_h2h = lambda db, h, a, before, **kw: matches
try:
result = await h2h_slice(header, limit=8, before=None)
text = str(result)
assert "2胜 0平 0负" in text, f"期望「2胜 0平 0负」,实际:\n{text}"
finally:
cb._get_h2h = orig
@pytest.mark.asyncio
async def test_draw_counted_correctly(self):
"""场景: 两场交锋一胜一平,验证平局也被正确计数。"""
header = _make_header(home_id=1, away_id=2, home_name="切尔西", away_name="热刺")
matches = [
_make_h2h_match(300, home_id=1, away_id=2, home_goals=1, away_goals=1,
home_name="切尔西", away_name="热刺"), # 平局
_make_h2h_match(301, home_id=2, away_id=1, home_goals=0, away_goals=2,
home_name="热刺", away_name="切尔西"), # 切尔西客场 2-0 赢
]
import src.llm.context_builder as cb
orig = cb._get_h2h
async def mock_get_h2h(db, h, a, before, *, limit):
return matches
cb._get_h2h = mock_get_h2h
try:
result = await h2h_slice(header, limit=8, before=None)
text = str(result)
# 切尔西视角: 1胜(客场2-0) 1平(主场1-1) 0负
assert "1胜 1平 0负" in text, f"期望「1胜 1平 0负」,实际:\n{text}"
finally:
cb._get_h2h = orig
tests/test_injuries_pipeline.py
+204
View File
@@ -0,0 +1,204 @@
"""回归测试: 伤停数据管线 5 项正确性修复。
Fix 1: IntegrityError 后不整批回滚
Fix 2: return_date 正确解析
Fix 3: retrieved_at 用 date() 比较避免当天不可见
Fix 4: partial unique index 防止 NULL 重复
Fix 5: 缓存 TTL 从 7 天改为 6 小时
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from src.data.injuries import _CACHE_TTL_HOURS, fetch_injuries
class TestCacheTTL:
"""Fix 5: 缓存 TTL 应为 6 小时。"""
def test_cache_ttl_is_6_hours(self):
assert _CACHE_TTL_HOURS == 6, f"缓存 TTL 应为 6 小时,实际 {_CACHE_TTL_HOURS}"
def test_cache_expiry_logic(self):
"""验证缓存过期逻辑:超过 TTL 返回 None(触发重新采集)。"""
import time
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
cache_file = Path(tmpdir) / "test_cache.json"
cache_file.write_text("[]")
# 模拟 7 小时前写入
old_time = time.time() - 7 * 3600
import os
os.utime(cache_file, (old_time, old_time))
age_hours = (time.time() - cache_file.stat().st_mtime) / 3600
assert age_hours > _CACHE_TTL_HOURS, "7 小时前的缓存应已过期"
def test_cache_hit_within_ttl(self):
"""验证 TTL 内缓存命中。"""
import time
from pathlib import Path
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
cache_file = Path(tmpdir) / "test_cache.json"
cache_file.write_text("[]")
# 1 小时前写入
old_time = time.time() - 3600
import os
os.utime(cache_file, (old_time, old_time))
age_hours = (time.time() - cache_file.stat().st_mtime) / 3600
assert age_hours < _CACHE_TTL_HOURS, "1 小时前的缓存应在 TTL 内"
class TestReturnDateParsing:
"""Fix 2: return_date 应从 API 响应正确解析并写入。"""
def test_parse_return_date_iso(self):
"""ISO 格式 return_date 应正确解析为 date 对象。"""
from datetime import datetime, date
raw = "2026-02-15"
dt = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
assert dt.date() == date(2026, 2, 15)
def test_parse_return_date_with_time(self):
"""带时间的 return_date 应截取日期部分。"""
from datetime import datetime, date
raw = "2026-03-01T00:00:00Z"
dt = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
assert dt.date() == date(2026, 3, 1)
def test_parse_return_date_none(self):
"""None 或空值应返回 None。"""
return_date_raw = None
return_date = None
if return_date_raw:
return_date = "should not reach"
assert return_date is None
def test_parse_return_date_invalid(self):
"""无效日期应返回 None 而非抛异常。"""
from datetime import datetime
raw = "invalid-date"
return_date = None
try:
dt = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
return_date = dt.date()
except (ValueError, AttributeError):
pass
assert return_date is None
class TestQueryDateComparison:
"""Fix 3: retrieved_at 比较应使用 date() 避免时区截断。"""
def test_date_comparison_handles_same_day(self):
"""核心 bug: 当天白天采到的数据应对当晚比赛可见。
retrieved_at = 2026-01-15 14:00:00+00 (timestamptz)
as_of = 2026-01-15 (date)
错误的比较: retrieved_at <= as_of
→ PostgreSQL 将 as_of 视为 2026-01-15 00:00:00+00
→ 14:00 <= 00:00 → False → 数据不可见!
正确的比较: date(retrieved_at) <= as_of
→ 2026-01-15 <= 2026-01-15 → True → 数据可见
"""
from datetime import datetime, date, timezone
retrieved_at = datetime(2026, 1, 15, 14, 0, tzinfo=timezone.utc)
as_of_date = date(2026, 1, 15)
# 错误的比较方式(原 bug)
# PostgreSQL 会将 date 转为 timestamptz at midnight
as_of_as_datetime = datetime(2026, 1, 15, 0, 0, tzinfo=timezone.utc)
wrong_result = retrieved_at <= as_of_as_datetime # False
# 正确的比较方式(修复后)
correct_result = retrieved_at.date() <= as_of_date # True
assert wrong_result is False, "原 bug 演示: 白天数据对当晚比赛不可见"
assert correct_result is True, "修复后: 白天数据对当晚比赛可见"
class TestPartialUniqueIndex:
"""Fix 4: partial unique index 防止 NULL 重复。"""
def test_orm_declares_partial_index(self):
"""ORM 模型应声明 partial unique index。"""
from sqlalchemy import and_
from src.db.models import Injury
# 验证 __table_args__ 包含 partial index
found_partial = False
for arg in Injury.__table_args__:
if hasattr(arg, "name") and arg.name == "ix_injuries_player_fixture":
# 验证是 unique 且有 postgresql_where
assert arg.unique is True, "应为唯一索引"
# postgresql_where 应排除 NULL
found_partial = True
assert found_partial, "Injury 模型应声明 ix_injuries_player_fixture 索引"
def test_migration_creates_partial_index(self):
"""迁移文件应包含 partial index 创建逻辑。"""
import os
migration_path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0012_injuries_partial_unique_and_return_date.py"
assert os.path.exists(migration_path), "迁移文件 0012 应存在"
with open(migration_path) as f:
content = f.read()
assert "CREATE UNIQUE INDEX ix_injuries_player_fixture" in content
assert "WHERE player_id IS NOT NULL" in content
assert "fixture_id IS NOT NULL" in content
class TestInjuriesSliceIntegration:
"""验证 injuries_slice 仍正常工作(未被破坏)。"""
@pytest.mark.asyncio
async def test_injuries_slice_with_cutoff(self):
"""injuries_slice 应正确传递 before=cutoff 到 get_injuries_for_match。"""
from datetime import datetime, timezone, timedelta
from src.llm.context_builder import injuries_slice, MatchHeader
header = MatchHeader(
match_id=999, home_name="A", away_name="B",
league_name="X", season=None, match_date="?",
match_dt=datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc),
stage=None, home_team_id=1, away_team_id=2, league_id=1,
)
cutoff = datetime(2026, 1, 14, 20, 0, tzinfo=timezone.utc)
import src.llm.context_builder as cb
orig = cb.get_injuries_for_match
captured_before = []
async def mock_get_injuries(db, team_id, match_date, as_of=None):
captured_before.append((team_id, match_date, as_of))
return []
cb.get_injuries_for_match = mock_get_injuries
try:
result = await injuries_slice(header, before=cutoff)
assert str(result) is not None
# 验证 before 参数被传递到 get_injuries_for_match
assert len(captured_before) == 2 # home + away
for team_id, match_date, as_of in captured_before:
# as_of 应等于 before (cutoff)
assert as_of == cutoff or (hasattr(as_of, 'date') and as_of.date() == cutoff.date()), \
f"as_of 应为 cutoff,实际 {as_of}"
finally:
cb.get_injuries_for_match = orig
tests/test_multi_agent_cutoff.py
+227
View File
@@ -0,0 +1,227 @@
"""回归测试: multi-agent 预测路径 backtest cutoff / provider / model 透传。
验证:
1. predict_match_multi 正确计算并传递 cutoff
2. cutoff 贯穿到所有 5 个专家切片
3. prediction_cutoff_at 记录的是真正的 cutoff,而非 match_dt
4. backtest=True 时「赛后才 available 的 xG」不会出现在切片里
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from src.llm.context_builder import MatchHeader
def _make_header(match_dt=None) -> MatchHeader:
from datetime import datetime, timezone
if match_dt is None:
match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc)
return MatchHeader(
match_id=999, home_name="利物浦", away_name="阿森纳",
league_name="英超", season="2025-2026",
match_date="2026-01-15 20:00 UTC",
match_dt=match_dt, stage=None,
home_team_id=1, away_team_id=2, league_id=1,
)
class TestMultiAgentCutoffPropagation:
"""验证 cutoff 在 multi-agent 路径中正确计算和传递。"""
@pytest.mark.asyncio
async def test_backtest_computes_cutoff_from_match_dt_minus_1_day(self):
"""backtest=True → cutoff = match_dt - 1 天,传给所有切片。"""
from datetime import datetime, timedelta, timezone
import src.llm.agents.orchestrator as orch
match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc)
header = _make_header(match_dt)
captured_before = []
orig_run_specialists = orch.run_specialists
async def mock_run_specialists(header, *, version, before=None):
captured_before.append(before)
return []
orch.run_specialists = mock_run_specialists
orch.load_match_header = lambda mid, db=None: header
orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test")
try:
try:
await orch.predict_match_multi(999, backtest=True)
except Exception:
pass # 后续 aggregator 调用会因 mock 不全而失败,不影响 cutoff 测试
assert len(captured_before) == 1
expected_cutoff = match_dt - timedelta(days=1)
assert captured_before[0] == expected_cutoff, (
f"backtest cutoff 应为 {expected_cutoff},实际 {captured_before[0]}"
)
finally:
orch.run_specialists = orig_run_specialists
@pytest.mark.asyncio
async def test_explicit_cutoff_at_overrides_backtest(self):
"""显式 cutoff_at 优先于 backtest 自动计算。"""
from datetime import datetime, timezone
import src.llm.agents.orchestrator as orch
match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc)
explicit_cutoff = datetime(2026, 1, 10, 12, 0, tzinfo=timezone.utc)
header = _make_header(match_dt)
captured_before = []
orig_run_specialists = orch.run_specialists
async def mock_run_specialists(header, *, version, before=None):
captured_before.append(before)
return []
orch.run_specialists = mock_run_specialists
orch.load_match_header = lambda mid, db=None: header
orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test")
try:
try:
await orch.predict_match_multi(999, backtest=True, cutoff_at=explicit_cutoff)
except Exception:
pass
assert captured_before[0] == explicit_cutoff
finally:
orch.run_specialists = orig_run_specialists
@pytest.mark.asyncio
async def test_normal_mode_cutoff_is_match_dt(self):
"""非回测模式,无显式 cutoff → cutoff = match_dt。"""
from datetime import datetime, timezone
import src.llm.agents.orchestrator as orch
match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc)
header = _make_header(match_dt)
captured_before = []
orig_run_specialists = orch.run_specialists
async def mock_run_specialists(header, *, version, before=None):
captured_before.append(before)
return []
orch.run_specialists = mock_run_specialists
orch.load_match_header = lambda mid, db=None: header
orch._agent_provider = lambda agent_id, **kw: MagicMock(model="test")
try:
try:
await orch.predict_match_multi(999, backtest=False)
except Exception:
pass
assert captured_before[0] == match_dt
finally:
orch.run_specialists = orig_run_specialists
@pytest.mark.asyncio
async def test_prediction_cutoff_at_stored_not_match_dt(self):
"""Prediction 写入时 prediction_cutoff_at = 真正 cutoff,非 match_dt。"""
from datetime import datetime, timedelta, timezone
from src.llm.predict import _predict_single, PredictResult
match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc)
expected_cutoff = match_dt - timedelta(days=1)
# Mock build_context to return a context with cutoff
import src.llm.predict as pred
orig_build = pred.build_context
class FakeContext:
text = "fake"
match_dt = match_dt
cutoff = expected_cutoff
async def fake_build(match_id, **kw):
return FakeContext()
pred.build_context = fake_build
pred._upsert_prediction = lambda session, **kw: MagicMock(id=1, **kw.get("values", {}))
try:
# 此处只验证 cutoff 参数传递,实际 LLM 调用会被 mock 阻断
# 重点: build_context 被调用时传入 backtest=True 和正确的 cutoff
call_args = {}
async def tracking_build(match_id, **kw):
call_args.update(kw)
return FakeContext()
pred.build_context = tracking_build
try:
await _predict_single(999, backtest=True)
except Exception:
pass
assert call_args.get("backtest") is True, "backtest=True 应传递给 build_context"
finally:
pred.build_context = orig_build
class TestBacktestXgNotVisible:
"""P0-3 延伸:回测时赛后才 available 的统计数据不应出现在切片。"""
@pytest.mark.asyncio
async def test_stats_slice_respects_cutoff_for_xg_availability(self):
"""available_at > cutoff 的 xG 数据不应被切片使用。"""
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock
cutoff = datetime(2026, 1, 13, 20, 0, tzinfo=timezone.utc) # match_date - 2天
match_dt = datetime(2026, 1, 15, 20, 0, tzinfo=timezone.utc)
# 创建一场历史比赛,其 xG 在 match_date 之后才 available
hist_match = MagicMock()
hist_match.id = 500
hist_match.home_team_id = 1 # 利物浦主场
hist_match.away_team_id = 3
hist_match.home_goals = 2
hist_match.away_goals = 0
hist_match.home_team = MagicMock(id=1, name="利物浦", name_zh=None)
hist_match.away_team = MagicMock(id=3, name="诺维奇", name_zh=None)
# xG: available_at 在比赛日之后(1月16日),cutoff(1月13日)看不到
stats = MagicMock()
stats.home_xg = 2.5
stats.away_xg = 0.3
stats.home_shots = 15
stats.away_shots = 4
stats.home_shots_on_target = 6
stats.away_shots_on_target = 1
stats.home_possession = 65.0
stats.available_at = datetime(2026, 1, 16, 10, 0, tzinfo=timezone.utc) # 赛后才有
hist_match.stats = stats
header = _make_header(match_dt)
import src.llm.context_builder as cb
orig_get_form = cb._get_form
async def mock_get_form(db, team_id, before, *, limit=10):
# before=cutoff(1月13日),比赛在1月15日,满足 before 条件
if before is not None and before < match_dt:
return [hist_match]
return []
cb._get_form = mock_get_form
try:
result = await cb.stats_slice(header, limit=10, before=cutoff)
text = str(result)
# xG 在 cutoff 之后才 available,不应出现在切片
assert "2.50" not in text, f"xG 2.50 不应在切片中(available_at > cutoff):\n{text}"
# 但无比分时仍应显示进球数据
assert "无比分数据" in text or "场均进球" in text, f"无比分时仍应显示基本数据:\n{text}"
finally:
cb._get_form = orig_get_form
tests/test_p0_home_away.py
+191
View File
@@ -0,0 +1,191 @@
"""回归测试: P0-1 — form_slice / stats_slice 主客身份反转。
用 mock Match 对象验证:当某队在历史比赛中是「客队」时,
form_slice 必须正确识别该队当时是客场,赛果应为 L(输),
对手名字和进球数不能反转。
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from src.llm.context_builder import MatchHeader, SliceResult, form_slice, stats_slice
def _make_team(team_id: int, name: str) -> MagicMock:
t = MagicMock()
t.id = team_id
t.name = name
t.name_zh = None
return t
def _make_stats(
home_xg=1.5,
away_xg=1.0,
home_shots=12,
away_shots=8,
home_sot=4,
away_sot=3,
home_poss=55.0,
available_at=None,
) -> MagicMock:
s = MagicMock()
s.home_xg = home_xg
s.away_xg = away_xg
s.home_shots = home_shots
s.away_shots = away_shots
s.home_shots_on_target = home_sot
s.away_shots_on_target = away_sot
s.home_possession = home_poss
s.available_at = available_at
return s
def _make_match(
match_id: int,
home_team_id: int,
away_team_id: int,
home_goals: int,
away_goals: int,
home_name: str = "H",
away_name: str = "A",
stats=None,
) -> MagicMock:
m = MagicMock()
m.id = match_id
m.home_team_id = home_team_id
m.away_team_id = away_team_id
m.home_goals = home_goals
m.away_goals = away_goals
m.stats = stats
m.home_team = _make_team(home_team_id, home_name)
m.away_team = _make_team(away_team_id, away_name)
return m
def _make_header(home_id=1, away_id=2, home_name="利物浦", away_name="阿森纳") -> MatchHeader:
return MatchHeader(
match_id=999,
home_name=home_name,
away_name=away_name,
league_name="英超",
season="2025-2026",
match_date="2026-01-15 20:00 UTC",
match_dt=None,
stage=None,
home_team_id=home_id,
away_team_id=away_id,
league_id=1,
)
class TestFormSliceHomeAwayIdentity:
"""P0-1: form_slice 必须根据每场历史比赛的真实主客来判断赛果。"""
@pytest.mark.asyncio
async def test_home_team_away_loss_shows_L(self):
"""
场景: 本场利物浦是主队(home_id=1),历史上一场它作为客队 1-3 输给曼城。
正确输出: L 3-1 vs 曼城 (赛果为输,对手为曼城)
原bug: W 3-1 vs 曼城 (把客场输球算成主场赢球)
"""
header = _make_header(home_id=1, away_id=2, home_name="利物浦", away_name="阿森纳")
hist_match = _make_match(
match_id=100,
home_team_id=5, # 曼城主场
away_team_id=1, # 利物浦客场
home_goals=3,
away_goals=1,
home_name="曼城",
away_name="利物浦",
)
import src.llm.context_builder as cb
orig_get_form = cb._get_form
async def mock_get_form(db, team_id, before, *, limit):
return [hist_match] if team_id == 1 else []
cb._get_form = mock_get_form
try:
result = await form_slice(header, limit=5, before=None, db=MagicMock())
finally:
cb._get_form = orig_get_form
text = str(result)
assert "L 3-1 vs 曼城" in text, f"期望「L 3-1 vs 曼城」,实际输出:\n{text}"
assert "W 3-1" not in text, f"不应出现 W 3-1(客场输球不能算主场赢):\n{text}"
@pytest.mark.asyncio
async def test_away_team_home_win_shows_W_for_that_team(self):
"""
场景: 本场阿森纳是客队(away_id=2),历史上一场它作为主队 2-0 赢了切尔西。
从阿森纳视角: is_home=True → W 2-0 vs 切尔西。
原bug: side 固定为 "away" → _outcome(2,0,"away") = L → 输出 L 2-0 vs 切尔西(反转!)
"""
header = _make_header(home_id=1, away_id=2, home_name="利物浦", away_name="阿森纳")
hist_match = _make_match(
match_id=101,
home_team_id=2, # 阿森纳主场
away_team_id=4, # 切尔西客场
home_goals=2,
away_goals=0,
home_name="阿森纳",
away_name="切尔西",
)
import src.llm.context_builder as cb
orig_get_form = cb._get_form
async def mock_get_form(db, team_id, before, *, limit):
return [hist_match] if team_id == 2 else []
cb._get_form = mock_get_form
try:
result = await form_slice(header, limit=5, before=None, db=MagicMock())
finally:
cb._get_form = orig_get_form
text = str(result)
assert "W 2-0 vs 切尔西" in text, f"期望「W 2-0 vs 切尔西」,实际输出:\n{text}"
assert "L 2-0 vs 切尔西" not in text, f"不应出现 L 2-0(主场赢球不能算客场输):\n{text}"
class TestStatsSliceHomeAwayIdentity:
"""P0-1: stats_slice 进球/失球/xG 必须按历史比赛真实主客取值。"""
@pytest.mark.asyncio
async def test_home_team_away_match_goals_not_swapped(self):
"""
场景: 本场利物浦是主队,历史上一场它作为客队 1-3 输给曼城(xG 0.8 vs 2.5)。
从利物浦视角: 进球=1(away_goals), 失球=3(home_goals), xG=0.8(away_xg)。
原bug: side="home" → 进球=3, 失球=1, xG=2.5 —— 全部反了!
"""
header = _make_header(home_id=1, away_id=2, home_name="利物浦", away_name="阿森纳")
hist_match = _make_match(
match_id=200,
home_team_id=5, # 曼城主场
away_team_id=1, # 利物浦客场
home_goals=3,
away_goals=1,
home_name="曼城",
away_name="利物浦",
stats=_make_stats(home_xg=2.5, away_xg=0.8, home_shots=15, away_shots=5,
home_sot=6, away_sot=2, home_poss=60.0),
)
import src.llm.context_builder as cb
orig_get_form = cb._get_form
async def mock_get_form(db, team_id, before, *, limit):
return [hist_match] if team_id == 1 else []
cb._get_form = mock_get_form
try:
result = await stats_slice(header, limit=10, before=None, db=MagicMock())
finally:
cb._get_form = orig_get_form
text = str(result)
# 利物浦客场 1-3 输: 进球 1, 失球 3
assert "场均进球 1.00" in text, f"期望场均进球 1.00,实际输出:\n{text}"
assert "场均失球 3.00" in text, f"期望场均失球 3.00,实际输出:\n{text}"
# 原bug: 进球 3, 失球 1 (反了)
assert "场均进球 3.00" not in text, f"不应出现场均进球 3.00(反转):\n{text}"
# xG: 利物浦 away_xg=0.8
assert "场均 xG 0.80" in text, f"期望场均 xG 0.80,实际输出:\n{text}"
# shots: 利物浦 away_shots=5
assert "场均射门 5.0" in text, f"期望场均射门 5.0,实际输出:\n{text}"
tests/test_p0_parse_failure.py
+114
View File
@@ -0,0 +1,114 @@
"""回归测试: P0-3 — LLM 解析失败不能产生假成功预测。
验证链路:
1. provider.py: JSON 解析失败时必须设置 error
2. predict.py: resp.parsed is None 时必须抛错,不能 fallback 到 {}
3. validation.py: 必填字段缺失时必须失败,不能静默给默认值
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from src.llm.provider import LLMProvider, LLMResponse
from src.llm.validation import validate_prediction_output
class TestProviderJsonParseError:
"""P0-3 Part 1: provider.py JSON 解析失败必须设置 error。"""
@pytest.mark.asyncio
async def test_invalid_json_sets_error(self, monkeypatch):
"""LLM 返回非 JSON 内容时,error 必须非空。"""
async def fake_post(*args, **kwargs):
class FakeResp:
status_code = 200
def raise_for_status(self): pass
def json(self):
return {
"choices": [{"message": {"content": "我不确定,可能是平局"}}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5},
}
return FakeResp()
import httpx
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
p = LLMProvider(api_key="test", model="gpt-4o")
resp = await p.chat("sys", "user", json_mode=True)
# P0-3: JSON 解析失败必须设置 error
assert resp.error is not None, "JSON 解析失败应设置 error"
assert resp.parsed is None
@pytest.mark.asyncio
async def test_code_block_json_works(self, monkeypatch):
"""LLM 返回 ```json {...}}``` 时应成功解析。"""
async def fake_post(*args, **kwargs):
class FakeResp:
status_code = 200
def raise_for_status(self): pass
def json(self):
return {
"choices": [{"message": {"content": '```json\n{"pred_home_goals": 1.5, "pred_away_goals": 1.0, "pred_1x2": "1", "subjective_confidence": 0.7}\n```'}}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5},
}
return FakeResp()
import httpx
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
p = LLMProvider(api_key="test", model="gpt-4o")
resp = await p.chat("sys", "user", json_mode=True)
assert resp.error is None
assert resp.parsed is not None
assert resp.parsed["pred_1x2"] == "1"
class TestValidationNoSilentDefaults:
"""P0-3 Part 3: validation.py 必填字段缺失时必须失败。"""
def test_missing_pred_home_goals_raises(self):
"""缺少 pred_home_goals 必须报错,不能默认为 0。"""
with pytest.raises((ValueError, KeyError)):
validate_prediction_output({
"pred_away_goals": 1,
"pred_1x2": "1",
"subjective_confidence": 0.7,
})
def test_missing_pred_1x2_raises(self):
"""缺少 pred_1x2 必须报错,不能默认为 X。"""
with pytest.raises(ValueError, match="Missing required field: pred_1x2"):
validate_prediction_output({
"pred_home_goals": 1,
"pred_away_goals": 0,
"subjective_confidence": 0.7,
})
def test_missing_confidence_raises(self):
"""缺少 subjective_confidence 必须报错,不能默认为 0.5。"""
with pytest.raises(ValueError, match="Missing required field: subjective_confidence"):
validate_prediction_output({
"pred_home_goals": 1,
"pred_away_goals": 0,
"pred_1x2": "1",
})
def test_empty_dict_raises(self):
"""空 dict 必须报错(不能产生 0-0 X 0.5 的假预测)。"""
with pytest.raises((ValueError, KeyError)):
validate_prediction_output({})
def test_valid_input_passes(self):
"""完整的合法输入应通过。"""
result = validate_prediction_output({
"pred_home_goals": 1.5,
"pred_away_goals": 1.0,
"pred_1x2": "1",
"subjective_confidence": 0.7,
})
assert result.pred_home_goals == 2 # 1.5 → round → 2
assert result.pred_away_goals == 1
assert result.pred_1x2 == "1"
assert result.subjective_confidence == 0.7
tests/test_predict_protection.py
+99
View File
@@ -0,0 +1,99 @@
"""回归测试: /api/v1/predict 限流 + 短 session 模式。
验证:
1. 限流: 同 IP 超过 10 次/分钟返回 429
2. 限流: 不同 IP 独立计数
3. 限流: 滑动窗口过期后恢复
4. 短 session: predict 路由不持有 DB 连接 during LLM call
"""
from __future__ import annotations
import asyncio
import time
import pytest
from src.api.deps import _RateLimiter, rate_limit_predict
class TestRateLimiter:
"""_RateLimiter 滑动窗口限流。"""
def test_allows_within_limit(self):
limiter = _RateLimiter(max_requests=10, window_seconds=60)
for _ in range(10):
assert limiter.is_allowed("192.168.1.1")
def test_blocks_over_limit(self):
limiter = _RateLimiter(max_requests=3, window_seconds=60)
assert limiter.is_allowed("10.0.0.1") # 1
assert limiter.is_allowed("10.0.0.1") # 2
assert limiter.is_allowed("10.0.0.1") # 3
assert not limiter.is_allowed("10.0.0.1") # 4 → blocked
def test_different_keys_independent(self):
"""不同 IP 的限流计数独立。"""
limiter = _RateLimiter(max_requests=2, window_seconds=60)
assert limiter.is_allowed("10.0.0.1")
assert limiter.is_allowed("10.0.0.1")
assert not limiter.is_allowed("10.0.0.1") # blocked
# 不同 IP 仍允许
assert limiter.is_allowed("10.0.0.2")
assert limiter.is_allowed("10.0.0.2")
assert not limiter.is_allowed("10.0.0.2") # blocked
def test_sliding_window_expires(self):
"""滑动窗口:过期后恢复。"""
limiter = _RateLimiter(max_requests=2, window_seconds=1)
assert limiter.is_allowed("10.0.0.1")
assert limiter.is_allowed("10.0.0.1")
assert not limiter.is_allowed("10.0.0.1") # blocked
# 等待窗口过期
time.sleep(1.1)
assert limiter.is_allowed("10.0.0.1") # 窗口过期,恢复
def test_cleans_expired_entries(self):
"""验证过期条目被清理(不会无限增长)。"""
limiter = _RateLimiter(max_requests=100, window_seconds=1)
for _ in range(50):
limiter.is_allowed("10.0.0.1")
# 验证内部状态
assert len(limiter._hits.get("10.0.0.1", [])) == 50
time.sleep(1.1)
# 触发清理
limiter.is_allowed("10.0.0.1")
# 过期条目应被清除,只剩新加入的 1 条
assert len(limiter._hits.get("10.0.0.1", [])) == 1
class TestShortReadSession:
"""short_read 上下文管理器。"""
@pytest.mark.asyncio
async def test_short_read_context_manager(self):
"""short_read 应作为 async context manager 工作。"""
from src.db.base import short_read
import inspect
# 验证是 async context manager (通过 inspect 检查)
assert inspect.isasyncgenfunction(short_read) or hasattr(short_read, "__wrapped__")
# 验证可以调用并返回 context manager
ctx = short_read()
assert hasattr(ctx, "__aenter__")
assert hasattr(ctx, "__aexit__")
class TestDepsImports:
"""验证新依赖可正确导入。"""
def test_rate_limit_predict_importable(self):
from src.api.deps import rate_limit_predict
assert callable(rate_limit_predict)
def test_rate_limiter_importable(self):
from src.api.deps import _RateLimiter, _predict_limiter
assert isinstance(_predict_limiter, _RateLimiter)
assert _predict_limiter.max_requests == 10
assert _predict_limiter.window_seconds == 60
tests/test_prediction_unique_constraint.py
+121
View File
@@ -0,0 +1,121 @@
"""回归测试: 预测唯一约束修复 —— live 与 backtest 可共存。
验证:
1. 唯一约束包含 mode + run_type
2. 同一场比赛 live 与 backtest 预测可共存,互不覆盖
3. _upsert_prediction 正确区分 run_type
"""
from __future__ import annotations
import inspect
from pydantic import BaseModel
import pytest
from src.db.models import Prediction, UniqueConstraint, CheckConstraint
class TestUniqueConstraint:
"""验证唯一约束包含 mode + run_type。"""
def test_constraint_columns(self):
"""唯一约束应包含 match_id, provider, model, mode, run_type。"""
uc = [
c for c in Prediction.__table__.constraints
if isinstance(c, UniqueConstraint) and "match" in c.name
]
assert len(uc) == 1
cols = [c.name for c in uc[0].columns]
assert cols == ["match_id", "provider", "model", "mode", "run_type"]
def test_run_type_check_constraint(self):
"""应有 run_type 的 check constraint。"""
cc = [
c for c in Prediction.__table__.constraints
if isinstance(c, CheckConstraint) and "run_type" in c.name
]
assert len(cc) == 1
def test_run_type_column_exists(self):
"""run_type 列应存在且 NOT NULL,默认 'live'。"""
cols = {c.name: c for c in Prediction.__table__.columns}
assert "run_type" in cols
assert cols["run_type"].nullable is False
# 默认值
assert cols["run_type"].default.arg == "live" if cols["run_type"].default else True
class TestUpsertPredictionSignature:
"""验证 _upsert_prediction 函数签名包含 run_type。"""
def test_signature_has_run_type(self):
from src.llm.predict import _upsert_prediction
sig = inspect.signature(_upsert_prediction)
assert "run_type" in sig.parameters
def test_signature_has_backtest_in_predict_match(self):
from src.llm.predict import predict_match
sig = inspect.signature(predict_match)
assert "backtest" in sig.parameters
def test_signature_has_backtest_in_predict_multi(self):
from src.llm.agents.orchestrator import predict_match_multi
sig = inspect.signature(predict_match_multi)
assert "backtest" in sig.parameters
class TestMigration:
"""验证迁移文件存在且内容正确。"""
def test_migration_exists(self):
import os
path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0013_predictions_unique_constraint_mode_run_type.py"
assert os.path.exists(path)
def test_migration_adds_column_and_constraint(self):
path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0013_predictions_unique_constraint_mode_run_type.py"
content = open(path).read()
assert 'run_type' in content
assert 'uq_predictions_match_provider_model_mode_run_type' in content
assert 'backtest' in content
assert 'live' in content
# 验证数据回填逻辑
assert "UPDATE predictions SET run_type = 'live'" in content
class TestLiveBacktestCoexist:
"""验证 live 与 backtest 可共存(逻辑验证,无需数据库)。"""
def test_different_run_type_allow_coexistence(self):
"""同一 match_id + provider + model + mode,不同 run_type 应可共存。
这是核心修复:之前唯一约束只有 (match_id, provider, model),
backtest 会覆盖 live 预测。
"""
# 模拟两行数据
class FakeRow:
def __init__(self, **kw):
for k, v in kw.items():
setattr(self, k, v)
live = FakeRow(match_id=1, provider="openai", model="gpt-4o", mode="single", run_type="live")
backtest = FakeRow(match_id=1, provider="openai", model="gpt-4o", mode="single", run_type="backtest")
# 两者唯一键不同(因为 run_type 不同)
live_key = (live.match_id, live.provider, live.model, live.mode, live.run_type)
backtest_key = (backtest.match_id, backtest.provider, backtest.model, backtest.mode, backtest.run_type)
assert live_key != backtest_key, "live 与 backtest 应有不同的唯一键"
assert live_key == (1, "openai", "gpt-4o", "single", "live")
assert backtest_key == (1, "openai", "gpt-4o", "single", "backtest")
def test_same_run_type_prevents_duplicate(self):
"""相同 run_type 的重复预测仍应被约束阻止。"""
key1 = (1, "openai", "gpt-4o", "single", "live")
key2 = (1, "openai", "gpt-4o", "single", "live")
assert key1 == key2, "相同 run_type 应有相同唯一键,应被约束阻止"