diff --git a/src/api/app.py b/src/api/app.py index dbd33d0..1a1e3f3 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -15,7 +15,7 @@ from src.core.config import settings async def lifespan(app: FastAPI) -> AsyncIterator[None]: from src.db.base import init_db from src.core.http_client import close_client - await init_db() + await init_db() # 验证连接,不建表 yield await close_client() @@ -53,6 +53,18 @@ def create_app() -> FastAPI: async def health(): return {"status": "healthy", "service": "profeto"} + + @app.get("/health/ready") + async def health_ready(): + """就绪检查: 验证数据库连接。""" + from src.db.base import engine + try: + async with engine.begin() as conn: + await conn.run_sync(lambda conn: None) + return {"status": "ready"} + except Exception: + return {"status": "not_ready"} + return app diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 6e7bb3a..7528de8 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -146,6 +146,43 @@ class BzzoiroSource: db.add(league) await db.flush() + # === 批量优化: 预加载球队和已有比赛到内存 === + team_name_to_id: dict[str, int] = {} + existing_match_keys: set[tuple[int, int, str]] = set() + if raw_events: + # 预加载所有涉及的球队名 + all_team_names = set() + for raw in raw_events: + nm = normalize_bzzoiro(raw, code) + if nm: + all_team_names.add(nm.home_team) + all_team_names.add(nm.away_team) + + if all_team_names: + from sqlalchemy import select + from src.db.models import Team + stmt = select(Team).where(Team.name.in_(all_team_names)) + teams = (await db.execute(stmt)).scalars().all() + team_name_to_id = {t.name: t.id for t in teams} + + # 预加载已有比赛 (league_id + home_id + away_id + date) + # 需要先获取球队 ID,所以分批处理 + date_strs = set() + for raw in raw_events: + nm = normalize_bzzoiro(raw, code) + if nm and nm.date: + date_strs.add(nm.date.date().isoformat() if hasattr(nm.date, "date") else str(nm.date)) + + if date_strs: + from sqlalchemy import func + stmt = ( + select(Match.home_team_id, Match.away_team_id, func.date(Match.match_date).label("d")) + .where(Match.league_id == league.id) + ) + rows = (await db.execute(stmt)).all() + for row in rows: + existing_match_keys.add((row.home_team_id, row.away_team_id, str(row.d))) + for raw in raw_events: try: nm = normalize_bzzoiro(raw, code) @@ -157,19 +194,34 @@ class BzzoiroSource: league_r["errors"].append(f"normalize: {e}") continue - # 球队 - home_team = await get_or_create_team(db, nm.home_team) - away_team = await get_or_create_team(db, nm.away_team) + # 球队: 内存查找 + 按需创建 + home_team_id = team_name_to_id.get(nm.home_team) + if home_team_id is None: + home = Team(name=nm.home_team) + db.add(home) + await db.flush() + home_team_id = home.id + team_name_to_id[nm.home_team] = home_team_id - # 查找已有比赛(天级匹配) - existing = await find_existing_match(db, league.id, nm.home_team, nm.away_team, nm.date) + away_team_id = team_name_to_id.get(nm.away_team) + if away_team_id is None: + away = Team(name=nm.away_team) + db.add(away) + await db.flush() + away_team_id = away.id + team_name_to_id[nm.away_team] = away_team_id + + # 查找已有比赛: 内存查找 + date_key = nm.date.date().isoformat() if hasattr(nm.date, "date") else str(nm.date) + match_key = (home_team_id, away_team_id, date_key) + existing = None if match_key not in existing_match_keys else "exists" if existing is None: m = Match( league_id=league.id, season=nm.season_label or None, - home_team_id=home_team.id, - away_team_id=away_team.id, + home_team_id=home_team_id, + away_team_id=away_team_id, match_date=nm.date, match_date_date=nm.date.date() if hasattr(nm.date, "date") else nm.date, match_status=nm.match_status, @@ -181,6 +233,7 @@ class BzzoiroSource: ) db.add(m) await db.flush() + existing_match_keys.add(match_key) # 防止同批重复 if nm.home_xg is not None or nm.away_xg is not None: stats = MatchStats( match_id=m.id, @@ -201,35 +254,45 @@ class BzzoiroSource: db.add(stats) league_r["inserted"] += 1 else: - # 更新(只补空 / 状态升级) + # 已有比赛: 需要查询对象来更新 + # 注意: 这里为了简化仍查询一次,但只在"已有"时触发 + from sqlalchemy import func + stmt = ( + select(Match) + .where(Match.league_id == league.id) + .where(Match.home_team_id == home_team_id) + .where(Match.away_team_id == away_team_id) + .where(func.date(Match.match_date) == date_key) + ) + existing_match = (await db.execute(stmt)).scalar_one() + changed = False - if existing.match_status != nm.match_status and nm.match_status == "finished": - existing.match_status = nm.match_status + if existing_match.match_status != nm.match_status and nm.match_status == "finished": + existing_match.match_status = nm.match_status changed = True - if existing.home_goals is None and nm.home_goals is not None: - existing.home_goals = nm.home_goals - existing.away_goals = nm.away_goals - existing.home_ht_goals = nm.home_ht_goals - existing.away_ht_goals = nm.away_ht_goals + if existing_match.home_goals is None and nm.home_goals is not None: + existing_match.home_goals = nm.home_goals + existing_match.away_goals = nm.away_goals + existing_match.home_ht_goals = nm.home_ht_goals + existing_match.away_ht_goals = nm.away_ht_goals changed = True - if existing.match_stage is None and nm.match_stage: - existing.match_stage = nm.match_stage + if existing_match.match_stage is None and nm.match_stage: + existing_match.match_stage = nm.match_stage changed = True - # stats 只补空 - if existing.stats is None and (nm.home_xg is not None or nm.away_xg is not None): - existing.stats = MatchStats(match_id=existing.id) - db.add(existing.stats) + if existing_match.stats is None and (nm.home_xg is not None or nm.away_xg is not None): + existing_match.stats = MatchStats(match_id=existing_match.id) + db.add(existing_match.stats) await db.flush() - if existing.stats is not None: + if existing_match.stats is not None: for fld in ("home_xg", "away_xg", "home_shots", "away_shots", "home_shots_on_target", "away_shots_on_target", "home_corners", "away_corners", "home_possession", "home_yellow_cards", "away_yellow_cards", "home_red_cards", "away_red_cards"): - if getattr(existing.stats, fld, None) is None: + if getattr(existing_match.stats, fld, None) is None: v = getattr(nm, fld, None) if v is not None: - setattr(existing.stats, fld, v) + setattr(existing_match.stats, fld, v) changed = True if changed: league_r["updated"] += 1 diff --git a/src/data/injuries.py b/src/data/injuries.py index db86233..ba93e83 100644 --- a/src/data/injuries.py +++ b/src/data/injuries.py @@ -178,8 +178,16 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict: return result -async def get_injuries_for_match(db, team_id: int, match_date) -> list[Injury]: - """查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。""" +async def get_injuries_for_match(db, team_id: int, match_date, as_of=None) -> list[Injury]: + """查询某场比赛前某队的伤停名单(比赛日仍缺阵的)。 + + Args: + db: 数据库 session + team_id: 球队 ID + match_date: 比赛日期 + as_of: 截止时间(cutoff)。只返回 retrieved_at <= as_of 的记录。 + 用于回测时防止"未来采集的数据"泄漏到历史预测。 + """ from sqlalchemy import and_, or_, select from src.db.models import Injury @@ -192,7 +200,15 @@ async def get_injuries_for_match(db, team_id: int, match_date) -> list[Injury]: .where(Injury.team_id == team_id) .where(Injury.injury_date <= match_date) .where(or_(Injury.return_date.is_(None), Injury.return_date >= match_date)) - .order_by(Injury.injury_date.desc()) ) + + # 回测防泄漏: 只使用 as_of 时间点之前已采集的数据 + if as_of is not None: + if hasattr(as_of, "date"): + as_of = as_of.date() + stmt = stmt.where(Injury.retrieved_at.is_not(None)) + stmt = stmt.where(Injury.retrieved_at <= as_of) + + stmt = stmt.order_by(Injury.injury_date.desc()) result = await db.execute(stmt) return list(result.scalars().all()) diff --git a/src/db/base.py b/src/db/base.py index 829c649..c8c1d1e 100644 --- a/src/db/base.py +++ b/src/db/base.py @@ -53,6 +53,17 @@ async def get_db_read() -> AsyncIterator[AsyncSession]: async def init_db() -> None: - """开发/测试用:建表。生产建议用 alembic。""" + """验证数据库连接(不建表)。 + + 生产环境 schema 由 Alembic 管理。 + 本地开发/测试需要建表时调用 `create_all()`。 + """ + async with engine.begin() as conn: + # 只验证连接,不自动建表 + await conn.run_sync(lambda conn: None) + + +async def create_all() -> None: + """创建所有表(仅用于本地开发/测试)。""" async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) diff --git a/src/llm/agents/base.py b/src/llm/agents/base.py index b7dcd4f..1cbbeb3 100644 --- a/src/llm/agents/base.py +++ b/src/llm/agents/base.py @@ -99,36 +99,34 @@ def _stub_no_data(agent: str) -> AgentReport: def _parse_report(agent: str, parsed: dict, resp: LLMResponse, model: str) -> AgentReport: - """把 LLM JSON 输出解析为 AgentReport,字段宽容处理。""" - def _f(v, default=None): - try: - return float(v) if v is not None else default - except (TypeError, ValueError): - return default + """把 LLM JSON 输出解析为 AgentReport,经过严格校验。""" + from src.llm.validation import validate_agent_output - suff = str(parsed.get("data_sufficiency", "medium")).lower() - if suff not in ("high", "medium", "low", "none"): - suff = "medium" - - evidence = parsed.get("key_evidence") or [] - if isinstance(evidence, str): - evidence = [evidence] - - score = parsed.get("probable_score") - if isinstance(score, dict): - score = f"{score.get('home', '?')}-{score.get('away', '?')}" + try: + validated = validate_agent_output(parsed) + except Exception as e: + # 校验失败 → 返回 parse_error 而非静默降级 + return AgentReport( + agent=agent, + status="parse_error", + analysis=f"输出校验失败: {e}", + model=model, + latency_ms=resp.latency_ms, + prompt_tokens=resp.prompt_tokens, + completion_tokens=resp.completion_tokens, + ) return AgentReport( agent=agent, status="ok", - data_sufficiency=suff, - analysis=str(parsed.get("analysis", ""))[:600], - home_edge=_f(parsed.get("home_edge")), - confidence=_f(parsed.get("confidence")), - key_evidence=[str(e)[:120] for e in evidence[:5]], - exp_home_goals=_f(parsed.get("exp_home_goals")), - exp_away_goals=_f(parsed.get("exp_away_goals")), - probable_score=score if isinstance(score, str) else None, + data_sufficiency=validated.data_sufficiency, + analysis=validated.analysis, + home_edge=validated.home_edge, + confidence=validated.confidence, + key_evidence=validated.key_evidence, + exp_home_goals=validated.exp_home_goals, + exp_away_goals=validated.exp_away_goals, + probable_score=validated.probable_score, model=model, latency_ms=resp.latency_ms, prompt_tokens=resp.prompt_tokens, diff --git a/src/llm/agents/orchestrator.py b/src/llm/agents/orchestrator.py index cf7b283..3b560a2 100644 --- a/src/llm/agents/orchestrator.py +++ b/src/llm/agents/orchestrator.py @@ -183,6 +183,13 @@ async def predict_match_multi( if m is None: raise ValueError(f"match {match_id} not found") + # 严格校验终裁输出 + from src.llm.validation import validate_prediction_output + try: + validated = validate_prediction_output(final) + except Exception as e: + raise RuntimeError(f"终裁输出校验失败: {e}") + agent_weights = final.get("agent_weights") pred = Prediction( match_id=match_id, @@ -193,11 +200,11 @@ async def predict_match_multi( prompt_tokens=sum(r.prompt_tokens or 0 for r in reports) + agg_prompt_tokens, completion_tokens=sum(r.completion_tokens or 0 for r in reports) + agg_completion_tokens, latency_ms=latency_ms, - pred_home_goals=final.get("pred_home_goals"), - pred_away_goals=final.get("pred_away_goals"), - pred_1x2=final.get("1x2"), - confidence=final.get("confidence"), - reasoning=final.get("reasoning"), + pred_home_goals=validated.pred_home_goals, + pred_away_goals=validated.pred_away_goals, + pred_1x2=validated.pred_1x2, + confidence=validated.confidence, + reasoning=validated.reasoning, raw_response=final, agent_outputs=[r.to_dict() for r in reports], ) diff --git a/src/llm/backtest.py b/src/llm/backtest.py index 24ae305..8e4b8fb 100644 --- a/src/llm/backtest.py +++ b/src/llm/backtest.py @@ -117,7 +117,8 @@ async def run_backtest( for m in matches: try: - # 预测 (build_context 内部已用 before=match_date 防泄漏) + # 预测 (build_context 内部已用 before=match_date 防泄漏, + # injuries_slice 也使用 as_of=match_date 过滤 retrieved_at) result = await predict_match(m.id, mode=mode, model=model) # 用实际比分 settle diff --git a/src/llm/context_builder.py b/src/llm/context_builder.py index a70c989..4a97438 100644 --- a/src/llm/context_builder.py +++ b/src/llm/context_builder.py @@ -219,12 +219,16 @@ async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None) async def injuries_slice(header: MatchHeader, *, before=None) -> str: - """D - 阵容完整性切片: 伤停与停赛名单,评估战力缺失程度。""" + """D - 阵容完整性切片: 伤停与停赛名单,评估战力缺失程度。 + + before=cutoff: 只使用 cutoff 之前已采集的伤停数据,防回测泄漏。 + """ from src.data.injuries import get_injuries_for_match + cutoff = before or header.match_dt async with AsyncSessionLocal() as db: - home_injuries = await get_injuries_for_match(db, header.home_team_id, before or header.match_dt) - away_injuries = await get_injuries_for_match(db, header.away_team_id, before or header.match_dt) + home_injuries = await get_injuries_for_match(db, header.home_team_id, cutoff, as_of=cutoff) + away_injuries = await get_injuries_for_match(db, header.away_team_id, cutoff, as_of=cutoff) lines = ["── 阵容完整性 ──"] has_data = False diff --git a/src/llm/predict.py b/src/llm/predict.py index ece94b0..9455b37 100644 --- a/src/llm/predict.py +++ b/src/llm/predict.py @@ -130,6 +130,13 @@ async def _predict_single( parsed = resp.parsed or {} + # 3.5 严格校验 LLM 输出 + from src.llm.validation import validate_prediction_output + try: + validated = validate_prediction_output(parsed) + except Exception as e: + raise RuntimeError(f"LLM 输出校验失败: {e}") + # 4. 存预测(独立 session,因为 context 用的是自己的 session) async with AsyncSessionLocal() as db: # 验证 match 存在 @@ -145,11 +152,11 @@ async def _predict_single( prompt_tokens=resp.prompt_tokens, completion_tokens=resp.completion_tokens, latency_ms=resp.latency_ms, - pred_home_goals=parsed.get("pred_home_goals"), - pred_away_goals=parsed.get("pred_away_goals"), - pred_1x2=parsed.get("1x2"), - confidence=parsed.get("confidence"), - reasoning=parsed.get("reasoning"), + pred_home_goals=validated.pred_home_goals, + pred_away_goals=validated.pred_away_goals, + pred_1x2=validated.pred_1x2, + confidence=validated.confidence, + reasoning=validated.reasoning, raw_response=resp.raw, ) db.add(pred) diff --git a/src/llm/validation.py b/src/llm/validation.py new file mode 100644 index 0000000..08f8570 --- /dev/null +++ b/src/llm/validation.py @@ -0,0 +1,130 @@ +"""LLM 输出严格校验。 + +所有 LLM JSON 输出必须经过 Pydantic 校验 + 语义一致性检查后才能落库。 +""" +from __future__ import annotations + +from pydantic import BaseModel, Field, field_validator, model_validator + +from src.db.models import Prediction + + +class AgentReportSchema(BaseModel): + """单个专家 Agent 输出的校验 schema。""" + + data_sufficiency: str = "medium" + analysis: str = "" + home_edge: float | None = Field(None, ge=-1.0, le=1.0) + confidence: float | None = Field(None, ge=0.0, le=1.0) + key_evidence: list[str] = Field(default_factory=list) + exp_home_goals: float | None = Field(None, ge=0.0, le=10.0) + exp_away_goals: float | None = Field(None, ge=0.0, le=10.0) + probable_score: str | None = None + + @field_validator("data_sufficiency") + @classmethod + def validate_sufficiency(cls, v: str) -> str: + allowed = {"high", "medium", "low", "none"} + return v.lower() if v.lower() in allowed else "medium" + + @field_validator("key_evidence", mode="before") + @classmethod + def normalize_evidence(cls, v) -> list[str]: + if v is None: + return [] + if isinstance(v, str): + return [v] + if isinstance(v, list): + return [str(e)[:120] for e in v[:5]] + return [] + + @field_validator("analysis") + @classmethod + def truncate_analysis(cls, v: str) -> str: + return str(v)[:600] + + +class PredictionOutputSchema(BaseModel): + """最终预测输出的校验 schema。""" + + pred_home_goals: float = Field(ge=0.0, le=10.0) + pred_away_goals: float = Field(ge=0.0, le=10.0) + pred_1x2: str + confidence: float = Field(ge=0.0, le=1.0) + reasoning: str = "" + + @field_validator("pred_1x2") + @classmethod + def validate_1x2(cls, v: str) -> str: + if v not in ("1", "X", "2"): + raise ValueError(f"pred_1x2 must be '1', 'X', or '2', got '{v}'") + return v + + @model_validator(mode="after") + def check_consistency(self) -> "PredictionOutputSchema": + """验证比分与胜平负一致。""" + expected = _score_to_1x2(self.pred_home_goals, self.pred_away_goals) + if expected and self.pred_1x2 != expected: + # 自动修正而非拒绝(LLM 常见小错误) + self.pred_1x2 = expected + return self + + +def _score_to_1x2(home: float, away: float) -> str | None: + """从比分推导胜平负。""" + if home > away: + return "1" + if home == away: + return "X" + if home < away: + return "2" + return None + + +def validate_agent_output(raw: dict) -> AgentReportSchema: + """校验并规范化单个 Agent 输出。""" + return AgentReportSchema( + data_sufficiency=raw.get("data_sufficiency", "medium"), + analysis=raw.get("analysis", ""), + home_edge=_safe_float(raw.get("home_edge")), + confidence=_safe_float(raw.get("confidence")), + key_evidence=raw.get("key_evidence", []), + exp_home_goals=_safe_float(raw.get("exp_home_goals")), + exp_away_goals=_safe_float(raw.get("exp_away_goals")), + probable_score=_format_score(raw.get("probable_score")), + ) + + +def validate_prediction_output(raw: dict) -> PredictionOutputSchema: + """校验最终预测输出。""" + return PredictionOutputSchema( + pred_home_goals=float(raw.get("pred_home_goals", 0)), + pred_away_goals=float(raw.get("pred_away_goals", 0)), + pred_1x2=raw.get("1x2") or raw.get("pred_1x2", "X"), + confidence=float(raw.get("confidence", 0.5)), + reasoning=str(raw.get("reasoning", ""))[:1000], + ) + + +def _safe_float(v) -> float | None: + """安全转 float,失败返回 None。""" + if v is None: + return None + try: + f = float(v) + if not (f == f): # NaN check + return None + return f + except (TypeError, ValueError): + return None + + +def _format_score(v) -> str | None: + """格式化比分输出。""" + if v is None: + return None + if isinstance(v, str): + return v + if isinstance(v, dict): + return f"{v.get('home', '?')}-{v.get('away', '?')}" + return None diff --git a/tests/test_agents.py b/tests/test_agents.py index 6ac2b9e..15c79c7 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -96,16 +96,24 @@ class TestReportParsing: parsed = { "data_sufficiency": "bogus", # 非法 → medium - "home_edge": "very strong", # 非法 → None + "home_edge": "very strong", # 非法 → None (宽容降级) "confidence": None, "key_evidence": "单字符串", # → [str] } resp = LLMResponse(content="{}", parsed=parsed) r = _parse_report("form", parsed, resp, "m") + # 宽容降级: 不抛异常,非法字段变 None 或默认值 + assert r.status == "ok" assert r.data_sufficiency == "medium" assert r.home_edge is None assert r.key_evidence == ["单字符串"] + # 验证范围约束: confidence > 1 会被截断或拒绝 + parsed2 = {"confidence": 1.5, "home_edge": 2.0} + r2 = _parse_report("form", parsed2, resp, "m") + # Pydantic 会拒绝越界值 → parse_error + assert r2.status == "parse_error" + class TestRunAgent: """run_agent 执行器: 门控 + fail-open。"""