diff --git a/src/api/routes/ingest.py b/src/api/routes/ingest.py index 2e49ec9..5b44cf5 100644 --- a/src/api/routes/ingest.py +++ b/src/api/routes/ingest.py @@ -6,7 +6,7 @@ from fastapi import APIRouter, HTTPException from src.api.schemas import IngestBzzoiroRequest, IngestResponse, IngestUnderstatRequest, IngestInjuriesRequest, IngestSimpleResponse from src.data.sources import get_source from src.data.injuries import ingest_injuries -from src.db.base import AsyncSessionLocal +from src.db.unit_of_work import get_uow router = APIRouter(prefix="/api/v1", tags=["ingest"]) @@ -15,42 +15,41 @@ router = APIRouter(prefix="/api/v1", tags=["ingest"]) async def ingest_bzzoiro_route(req: IngestBzzoiroRequest): """触发 bzzoiro 采集。""" source = get_source("bzzoiro") - async with AsyncSessionLocal() as db: - try: + try: + async with get_uow() as uow: result = await source.ingest( - db, + uow.session, leagues=req.leagues, date_from=req.date_from, date_to=req.date_to, status=req.status, ) - await db.commit() - return IngestResponse(**result) - except Exception as e: - await db.rollback() - raise HTTPException(500, str(e)) + await uow.commit() + return IngestResponse(**result) + except Exception as e: + raise HTTPException(500, str(e)) @router.post("/ingest/understat", response_model=IngestSimpleResponse) async def ingest_understat_route(req: IngestUnderstatRequest): """触发 understat xG 回填。""" source = get_source("understat") - async with AsyncSessionLocal() as db: - try: - result = await source.ingest(db, league=req.league, season=req.season) - return IngestSimpleResponse(**result) - except Exception as e: - await db.rollback() - raise HTTPException(500, str(e)) + try: + async with get_uow() as uow: + result = await source.ingest(uow.session, league=req.league, season=req.season) + await uow.commit() + return IngestSimpleResponse(**result) + except Exception as e: + raise HTTPException(500, str(e)) @router.post("/ingest/injuries", response_model=IngestSimpleResponse) async def ingest_injuries_route(req: IngestInjuriesRequest): """触发伤停采集。""" - async with AsyncSessionLocal() as db: - try: - result = await ingest_injuries(db, date=req.date) - return IngestSimpleResponse(**result) - except Exception as e: - await db.rollback() - raise HTTPException(500, str(e)) + try: + async with get_uow() as uow: + result = await ingest_injuries(uow.session, date=req.date) + await uow.commit() + return IngestSimpleResponse(**result) + except Exception as e: + raise HTTPException(500, str(e)) diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 7528de8..2105563 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -1,6 +1,7 @@ """Bzzoiro 数据源:抓取 + 入库。 迁移自旧项目 app/data/sources/bzzoiro/,改成 async + 简化入库。 +使用 Repository 模式进行数据访问,不直接控制事务。 """ from __future__ import annotations @@ -18,10 +19,9 @@ from sqlalchemy import select from src.core.config import settings from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL -from src.data.match_lookup import find_existing_match, get_or_create_team from src.data.normalize import normalize_bzzoiro from src.data.sources import register -from src.db.models import League, Match, MatchStats +from src.db.models import League, Match, MatchStats, Team logger = logging.getLogger(__name__) @@ -125,7 +125,10 @@ class BzzoiroSource: date_to: str | None = None, status: str = "finished", ) -> dict: - """采集 bzzoiro → 入库。返回统计。""" + """采集 bzzoiro → 入库。返回统计。 + + 注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。 + """ result: dict = {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []} for code in leagues: @@ -159,29 +162,19 @@ class BzzoiroSource: 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))) + # 预加载已有比赛 + 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: @@ -255,7 +248,6 @@ class BzzoiroSource: league_r["inserted"] += 1 else: # 已有比赛: 需要查询对象来更新 - # 注意: 这里为了简化仍查询一次,但只在"已有"时触发 from sqlalchemy import func stmt = ( select(Match) @@ -297,7 +289,7 @@ class BzzoiroSource: if changed: league_r["updated"] += 1 - await db.commit() + # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 result["leagues"][code] = league_r result["total_inserted"] += league_r["inserted"] result["total_updated"] += league_r["updated"] diff --git a/src/data/injuries.py b/src/data/injuries.py index ba93e83..747abf5 100644 --- a/src/data/injuries.py +++ b/src/data/injuries.py @@ -100,7 +100,10 @@ async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = No async def ingest_injuries(db, *, date: str | None = None) -> dict: - """采集伤停数据并入库(injuries 表)。""" + """采集伤停数据并入库(injuries 表)。 + + 注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。 + """ from sqlalchemy import select from src.data.team_names import normalize as normalize_name @@ -173,7 +176,7 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict: except Exception as e: result["errors"].append(f"parse error: {e}") - await db.commit() + # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 logger.info("injuries: fetched %d, inserted %d for %s", result["count"], result["inserted"], date) return result diff --git a/src/data/match_lookup.py b/src/data/match_lookup.py deleted file mode 100644 index 83eb167..0000000 --- a/src/data/match_lookup.py +++ /dev/null @@ -1,40 +0,0 @@ -"""比赛匹配辅助函数(多数据源共用)。 - -bzzoiro / understat 等数据源在入库时都需要: - - 按队名获取或创建球队(get_or_create_team) - - 按联赛+主队+客队+日期找已有比赛(find_existing_match) -""" -from __future__ import annotations - -from sqlalchemy import func, select - -from src.db.models import Match, Team - - -async def get_or_create_team(db, name: str) -> Team: - """按名获取球队,不存在则创建。""" - stmt = select(Team).where(Team.name == name) - team = (await db.execute(stmt)).scalar_one_or_none() - if team is None: - team = Team(name=name) - db.add(team) - await db.flush() - return team - - -async def find_existing_match(db, league_id: int, home_name: str, away_name: str, date) -> Match | None: - """按联赛+主队+客队+日期找已有比赛(天级匹配,避免时间精度差异)。""" - home_team = (await db.execute(select(Team).where(Team.name == home_name))).scalar_one_or_none() - away_team = (await db.execute(select(Team).where(Team.name == away_name))).scalar_one_or_none() - if home_team is None or away_team is None: - return None - - date_only = date.date() if hasattr(date, "date") else date - 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_only) - ) - return (await db.execute(stmt)).scalar_one_or_none() diff --git a/src/data/understat.py b/src/data/understat.py index b244849..2ca5be1 100644 --- a/src/data/understat.py +++ b/src/data/understat.py @@ -1,6 +1,7 @@ """Understat xG 数据源。 迁移自旧项目 app/data/sources/understat.py,改成 async。 +使用 Repository 模式进行数据访问,不直接控制事务。 """ from __future__ import annotations @@ -10,12 +11,13 @@ import logging import random import re +from sqlalchemy import func, select + from src.core.http_client import get_client from src.data.config import FDCO_TO_UNDERSTAT, LEAGUE_NAMES -from src.data.match_lookup import find_existing_match from src.data.normalize import normalize_understat from src.data.sources import register -from src.db.models import League, Match, MatchStats +from src.db.models import League, Match, MatchStats, Team logger = logging.getLogger(__name__) @@ -80,9 +82,10 @@ class UnderstatSource: name = "understat" async def ingest(self, db, *, league: str, season: int) -> dict: - """采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。""" - from sqlalchemy import select + """采集 understat xG → 回填到现有 Match。只回填 xG 字段,不创建新 Match。 + 注意: 本方法不控制事务(commit/rollback),由调用方通过 UnitOfWork 控制。 + """ result = {"updated": 0, "skipped": 0, "unmatched": 0, "errors": []} try: @@ -111,8 +114,22 @@ class UnderstatSource: result["errors"].append(f"normalize: {e}") continue - # 匹配已有 Match(天级) - existing = await find_existing_match(db, league_obj.id, nm.home_team, nm.away_team, nm.date) + # 匹配已有 Match(天级) - 直接查询 + home_team = (await db.execute(select(Team).where(Team.name == nm.home_team))).scalar_one_or_none() + away_team = (await db.execute(select(Team).where(Team.name == nm.away_team))).scalar_one_or_none() + if home_team is None or away_team is None: + result["unmatched"] += 1 + continue + + date_only = nm.date.date() if hasattr(nm.date, "date") else nm.date + stmt = ( + select(Match) + .where(Match.league_id == league_obj.id) + .where(Match.home_team_id == home_team.id) + .where(Match.away_team_id == away_team.id) + .where(func.date(Match.match_date) == date_only) + ) + existing = (await db.execute(stmt)).scalar_one_or_none() if existing is None: result["unmatched"] += 1 continue @@ -129,5 +146,5 @@ class UnderstatSource: if existing.stats.away_xg is None and nm.away_xg is not None: existing.stats.away_xg = nm.away_xg - await db.commit() + # 注意: 不在此处 commit,由调用方 UnitOfWork 控制事务 return result diff --git a/src/db/repositories.py b/src/db/repositories.py new file mode 100644 index 0000000..3dcd144 --- /dev/null +++ b/src/db/repositories.py @@ -0,0 +1,126 @@ +"""Repository 层:封装数据访问。 + +Repository 只负责查询,不负责事务提交。 +事务由 Application Service 通过 UnitOfWork 控制。 +""" +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.orm import selectinload +from sqlalchemy.ext.asyncio import AsyncSession + +from src.db.models import League, Match, Prediction, Team + + +class MatchRepository: + """比赛数据访问。""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def get_by_id(self, match_id: int) -> Match | None: + return await self._session.get(Match, match_id) + + async def get_with_relations(self, match_id: int) -> Match | None: + stmt = ( + select(Match) + .options( + selectinload(Match.league), + selectinload(Match.home_team), + selectinload(Match.away_team), + selectinload(Match.stats), + ) + .where(Match.id == match_id) + ) + return (await self._session.execute(stmt)).scalar_one_or_none() + + async def find_by_teams_and_date( + self, league_id: int, home_team_id: int, away_team_id: int, date + ) -> Match | None: + """按联赛+主队+客队+日期查找比赛(天级匹配)。""" + from sqlalchemy import func + + if hasattr(date, "date"): + date = date.date() + + 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) + ) + return (await self._session.execute(stmt)).scalar_one_or_none() + + async def add(self, match: Match) -> None: + self._session.add(match) + await self._session.flush() + + +class TeamRepository: + """球队数据访问。""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def get_by_name(self, name: str) -> Team | None: + stmt = select(Team).where(Team.name == name) + return (await self._session.execute(stmt)).scalar_one_or_none() + + async def get_or_create(self, name: str) -> Team: + """按名获取球队,不存在则创建。""" + team = await self.get_by_name(name) + if team is None: + team = Team(name=name) + self._session.add(team) + await self._session.flush() + return team + + async def get_all_by_names(self, names: list[str]) -> dict[str, Team]: + """批量获取球队,返回 name → Team 映射。""" + if not names: + return {} + stmt = select(Team).where(Team.name.in_(names)) + teams = (await self._session.execute(stmt)).scalars().all() + return {t.name: t for t in teams} + + async def add(self, team: Team) -> None: + self._session.add(team) + await self._session.flush() + + +class LeagueRepository: + """联赛数据访问。""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def get_by_code(self, code: str) -> League | None: + stmt = select(League).where(League.code == code) + return (await self._session.execute(stmt)).scalar_one_or_none() + + async def get_or_create(self, code: str, name: str, country: str | None = None) -> League: + league = await self.get_by_code(code) + if league is None: + league = League(code=code, name=name, country=country) + self._session.add(league) + await self._session.flush() + return league + + async def add(self, league: League) -> None: + self._session.add(league) + await self._session.flush() + + +class PredictionRepository: + """预测记录数据访问。""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def get_by_id(self, prediction_id: int) -> Prediction | None: + return await self._session.get(Prediction, prediction_id) + + async def add(self, prediction: Prediction) -> None: + self._session.add(prediction) + await self._session.flush() diff --git a/src/db/unit_of_work.py b/src/db/unit_of_work.py new file mode 100644 index 0000000..1939205 --- /dev/null +++ b/src/db/unit_of_work.py @@ -0,0 +1,63 @@ +"""工作单元(Unit of Work):统一事务边界。 + +使用方式: + async with UnitOfWork(db) as uow: + await uow.matches.get_by_id(1) + await uow.matches.add(new_match) + # 退出时自动 commit,异常时 rollback +""" +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import AsyncGenerator + +from sqlalchemy.ext.asyncio import AsyncSession + +from src.db.base import AsyncSessionLocal + + +class UnitOfWork: + """工作单元:封装事务边界。""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + self.committed = False + + @property + def session(self) -> AsyncSession: + return self._session + + async def commit(self) -> None: + await self._session.commit() + self.committed = True + + async def rollback(self) -> None: + await self._session.rollback() + + async def close(self) -> None: + await self._session.close() + + async def __aenter__(self) -> "UnitOfWork": + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + if exc_type is not None: + await self.rollback() + await self.close() + + +@asynccontextmanager +async def get_uow() -> AsyncGenerator[UnitOfWork, None]: + """创建新的工作单元(用于非路由上下文)。""" + session = AsyncSessionLocal() + uow = UnitOfWork(session) + try: + yield uow + if not uow.committed: + await uow.commit() + except Exception: + await uow.rollback() + raise + finally: + await uow.close() diff --git a/src/llm/agents/orchestrator.py b/src/llm/agents/orchestrator.py index 8d787cd..f261d89 100644 --- a/src/llm/agents/orchestrator.py +++ b/src/llm/agents/orchestrator.py @@ -11,6 +11,7 @@ from dataclasses import dataclass from src.core.config import settings from src.db.base import AsyncSessionLocal from src.db.models import Match, Prediction +from src.db.unit_of_work import get_uow from src.llm.agents.base import AgentReport, AgentSpec, load_agent_prompt from src.llm.context_builder import ( MatchHeader, @@ -184,8 +185,9 @@ async def predict_match_multi( _reports_to_json(reports).encode("utf-8") ).hexdigest() - # 4. 存库 - async with AsyncSessionLocal() as db: + # 4. 存库(使用 UnitOfWork) + async with get_uow() as uow: + db = uow.session m = await db.get(Match, match_id) if m is None: raise ValueError(f"match {match_id} not found") @@ -218,7 +220,7 @@ async def predict_match_multi( input_hash=input_hash, ) db.add(pred) - await db.commit() + await uow.commit() await db.refresh(pred) return MultiPredictResult( diff --git a/src/llm/backtest.py b/src/llm/backtest.py index c50d19a..aa62a35 100644 --- a/src/llm/backtest.py +++ b/src/llm/backtest.py @@ -11,8 +11,8 @@ from dataclasses import dataclass, field from sqlalchemy import and_, select -from src.db.base import AsyncSessionLocal -from src.db.models import League, Match, Prediction +from src.db.models import League, Match +from src.db.unit_of_work import get_uow from src.llm.eval import settle_prediction from src.llm.predict import predict_match @@ -33,7 +33,7 @@ class BacktestMatchResult: pred_home: float | None pred_away: float | None pred_1x2: str | None - confidence: float | None + subjective_confidence: float | None correct_1x2: bool prediction_id: int @@ -45,7 +45,7 @@ class BacktestSummary: scored: int accuracy_1x2: float | None = None avg_score_rmse: float | None = None - avg_confidence: float | None = None + avg_subjective_confidence: float | None = None calibration: list[dict] = field(default_factory=list) results: list[BacktestMatchResult] = field(default_factory=list) @@ -108,9 +108,9 @@ async def run_backtest( Returns: BacktestSummary 含逐场结果 + 汇总统计 """ - async with AsyncSessionLocal() as db: + async with get_uow() as uow: matches = await _get_historical_matches( - db, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit + uow.session, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit ) summary = BacktestSummary(total=len(matches), scored=0) @@ -163,10 +163,10 @@ async def run_backtest( if errors: summary.avg_score_rmse = round(sum(errors) / len(errors), 2) - # 平均置信度 + # 平均主观置信度 confs = [r.subjective_confidence for r in summary.results if r.subjective_confidence is not None] if confs: - summary.avg_confidence = round(sum(confs) / len(confs), 2) + summary.avg_subjective_confidence = round(sum(confs) / len(confs), 2) # 校准:按置信度分桶,看实际准确率是否匹配 summary.calibration = _compute_calibration(summary.results) diff --git a/src/llm/eval.py b/src/llm/eval.py index 01cf0df..1f90522 100644 --- a/src/llm/eval.py +++ b/src/llm/eval.py @@ -5,23 +5,22 @@ import logging from sqlalchemy import select -from src.db.base import AsyncSessionLocal from src.db.models import Prediction +from src.db.unit_of_work import get_uow logger = logging.getLogger(__name__) async def settle_prediction(prediction_id: int, home_goals: int, away_goals: int) -> Prediction: """回填实际结果。""" - async with AsyncSessionLocal() as db: - pred = await db.get(Prediction, prediction_id) + async with get_uow() as uow: + pred = await uow.session.get(Prediction, prediction_id) if pred is None: raise ValueError(f"prediction {prediction_id} not found") pred.actual_home_goals = home_goals pred.actual_away_goals = away_goals pred.settled = True - await db.commit() - await db.refresh(pred) + await uow.commit() return pred @@ -36,12 +35,12 @@ def _actual_1x2(home: int, away: int) -> str: async def get_eval_summary() -> dict: """按 provider × 模型聚合评估。""" - async with AsyncSessionLocal() as db: + async with get_uow() as uow: stmt = ( select(Prediction) .where(Prediction.settled == True) ) - result = await db.execute(stmt) + result = await uow.session.execute(stmt) rows = list(result.scalars().all()) from collections import defaultdict @@ -76,6 +75,6 @@ async def get_eval_summary() -> dict: "total": b["total"], "accuracy_1x2": round(acc, 1), "avg_score_rmse": round(avg_err, 2) if avg_err is not None else None, - "avg_confidence": round(avg_conf, 2) if avg_conf is not None else None, + "avg_subjective_confidence": round(avg_conf, 2) if avg_conf is not None else None, }) return {"summary": summary} diff --git a/src/llm/predict.py b/src/llm/predict.py index 6408075..1f41c70 100644 --- a/src/llm/predict.py +++ b/src/llm/predict.py @@ -13,6 +13,7 @@ from threading import Lock from src.core.config import settings from src.db.base import AsyncSessionLocal from src.db.models import Match, Prediction +from src.db.unit_of_work import get_uow from src.llm.context_builder import build_context from src.llm.provider import LLMProvider, get_default_provider @@ -143,8 +144,9 @@ async def _predict_single( except Exception as e: raise RuntimeError(f"LLM 输出校验失败: {e}") - # 4. 存预测(独立 session,因为 context 用的是自己的 session) - async with AsyncSessionLocal() as db: + # 4. 存预测(使用 UnitOfWork 统一事务) + async with get_uow() as uow: + db = uow.session # 验证 match 存在 m = await db.get(Match, match_id) if m is None: @@ -188,4 +190,5 @@ async def _predict_single( # 5. 写入缓存 _set_cached(match_id, settings.LLM_PROVIDER, provider.model, version, result) + await uow.commit() return result