diff --git a/.env.example b/.env.example index 2ae7ea3..53b4e87 100644 --- a/.env.example +++ b/.env.example @@ -3,6 +3,10 @@ APP_ENV=development LOG_LEVEL=INFO # ---- 数据库 ---- +POSTGRES_USER=football +POSTGRES_PASSWORD=football +POSTGRES_DB=football +POSTGRES_PORT=5432 DATABASE_URL=postgresql+asyncpg://football:football@localhost:5432/football # ---- LLM (OpenAI-compatible,必填一个) ---- diff --git a/docker-compose.yml b/docker-compose.yml index fcf0052..8812d2e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,15 +2,15 @@ services: postgres: image: postgres:16-alpine environment: - POSTGRES_USER: football - POSTGRES_PASSWORD: football - POSTGRES_DB: football + POSTGRES_USER: ${POSTGRES_USER:?POSTGRES_USER 未设置} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?POSTGRES_PASSWORD 未设置} + POSTGRES_DB: ${POSTGRES_DB:-football} ports: - - "5432:5432" + - "${POSTGRES_PORT:-5432}:5432" volumes: - pgdata:/var/lib/postgresql/data healthcheck: - test: ["CMD-SHELL", "pg_isready -U football"] + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:?POSTGRES_USER 未设置}"] interval: 5s timeout: 5s retries: 5 @@ -19,7 +19,7 @@ services: build: . command: uvicorn src.api.app:app --host 0.0.0.0 --port 8000 --reload ports: - - "8000:8000" + - "${API_PORT:-8000}:8000" env_file: .env depends_on: postgres: diff --git a/src/api/app.py b/src/api/app.py index 1a1e3f3..2685b5e 100644 --- a/src/api/app.py +++ b/src/api/app.py @@ -29,12 +29,14 @@ def create_app() -> FastAPI: ) origins = [o.strip() for o in settings.CORS_ORIGINS.split(",") if o.strip()] + methods = [m.strip() for m in settings.CORS_METHODS.split(",") if m.strip()] + headers = [h.strip() for h in settings.CORS_HEADERS.split(",") if h.strip()] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + allow_methods=methods, + allow_headers=headers, ) from src.api.routes.matches import router as matches_router diff --git a/src/api/deps.py b/src/api/deps.py index 395bf9c..10ac38a 100644 --- a/src/api/deps.py +++ b/src/api/deps.py @@ -19,24 +19,18 @@ from src.core.config import settings logger = logging.getLogger(__name__) -_warned_unset = False - async def require_admin_key(x_api_key: str | None = Header(None, alias="X-API-Key")) -> None: """保护「写入型 / 高成本」接口的依赖。 用法: `@router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin_key)])` """ - global _warned_unset - expected = settings.ADMIN_API_KEY if not expected: - if not _warned_unset: - logger.warning( - "ADMIN_API_KEY 未设置,采集/回测接口当前【无鉴权】。" - "生产环境请设置该环境变量。" - ) - _warned_unset = True + logger.warning( + "ADMIN_API_KEY 未设置,采集/回测接口当前【无鉴权】。" + "生产环境请设置该环境变量。" + ) return if not x_api_key or not secrets.compare_digest(x_api_key, expected): diff --git a/src/core/config.py b/src/core/config.py index 63649ac..2a223aa 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -32,6 +32,17 @@ class Settings(BaseSettings): # --- CORS --- CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000" + CORS_METHODS: str = "GET,POST,PUT,DELETE,OPTIONS" + CORS_HEADERS: str = "Authorization,Content-Type,X-API-Key,Accept" + + # --- HTTP --- + HTTP_DEFAULT_TIMEOUT: int = 30 + + # --- database pool --- + DB_POOL_SIZE: int = 5 + DB_MAX_OVERFLOW: int = 10 + DB_POOL_TIMEOUT: int = 30 + DB_POOL_RECYCLE: int = 1800 # --- 管理接口鉴权 --- # 采集 / 回测等高成本或写入型接口需要此 Key(请求头 X-API-Key)。 diff --git a/src/core/http_client.py b/src/core/http_client.py index 8ef1482..8bd7ab2 100644 --- a/src/core/http_client.py +++ b/src/core/http_client.py @@ -2,24 +2,27 @@ 使用方: - src/llm/provider.py: LLM 调用 + - src/data/bzzoiro.py: bzzoiro 比赛数据 - src/data/understat.py: xG 抓取 - src/data/injuries.py: 伤停抓取 生命周期由 FastAPI lifespan 管理(关闭时 aclose)。 +调用方可通过 `timeout` 参数覆盖 per-request 超时。 """ from __future__ import annotations import httpx +from src.core.config import settings + _shared_client: httpx.AsyncClient | None = None -_default_timeout = 30 def get_client() -> httpx.AsyncClient: """获取共享客户端(懒初始化)。""" global _shared_client if _shared_client is None or _shared_client.is_closed: - _shared_client = httpx.AsyncClient(timeout=_default_timeout) + _shared_client = httpx.AsyncClient(timeout=settings.HTTP_DEFAULT_TIMEOUT) return _shared_client diff --git a/src/data/bzzoiro.py b/src/data/bzzoiro.py index 8c51fe8..7b9dca5 100644 --- a/src/data/bzzoiro.py +++ b/src/data/bzzoiro.py @@ -9,16 +9,13 @@ import asyncio import json as _json import logging import random -import time as _time -import urllib.error -import urllib.parse -import urllib.request from collections.abc import Iterable from datetime import datetime, timezone from sqlalchemy import select from src.core.config import settings +from src.core.http_client import get_client from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL from src.data.normalize import normalize_bzzoiro from src.data.sources import register @@ -47,43 +44,46 @@ def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, i return (home_team_id, away_team_id, d.isoformat() if d is not None else "") -def _fetch_json_sync(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list: - """同步 HTTP(bzzoiro 客户端保持同步,在 async 函数里 run_in_executor)。""" +async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list: + """异步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。""" base = settings.BZZOIRO_BASE.rstrip("/") url = f"{base}/{path.lstrip('/')}" - if params: - url += "?" + urllib.parse.urlencode(params) key = settings.BZZOIRO_KEY if not key: raise RuntimeError("BZZOIRO_KEY 未设置") + headers = { + "Authorization": f"Token {key}", + "Accept": "application/json", + } + last_exc: Exception | None = None for attempt in range(max_retries): try: - req = urllib.request.Request(url) - req.add_header("Authorization", f"Token {key}") - req.add_header("Accept", "application/json") - with urllib.request.urlopen(req, timeout=30) as resp: - return _json.loads(resp.read().decode("utf-8")) - except urllib.error.HTTPError as e: + client = get_client() + resp = await client.get(url, headers=headers, params=params, timeout=30) + resp.raise_for_status() + return resp.json() + except Exception as e: last_exc = e - if e.code == 429: - # 指数退避: 429 通常意味着限速 + status = getattr(getattr(e, "response", None), "status_code", None) + if status == 429: delay = min(2 ** attempt, 16) + random.uniform(0, 1) logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay) - _time.sleep(delay) + await asyncio.sleep(delay) continue - if 500 <= e.code < 600: + if 500 <= (status or 0) < 600: delay = min(2 ** attempt, 16) + random.uniform(0, 1) - logger.warning("bzzoiro %d, retry %d in %.1fs", e.code, attempt + 1, delay) - _time.sleep(delay) + logger.warning("bzzoiro %d, retry %d in %.1fs", status, attempt + 1, delay) + await asyncio.sleep(delay) continue - raise # 4xx 直接抛 - except (urllib.error.URLError, TimeoutError, ConnectionError) as e: - last_exc = e - delay = min(2 ** attempt, 16) + random.uniform(0, 1) - logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e) - _time.sleep(delay) + # 网络错误(连接失败/超时)也退避重试 + if isinstance(e, (TimeoutError, ConnectionError, OSError)): + delay = min(2 ** attempt, 16) + random.uniform(0, 1) + logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e) + await asyncio.sleep(delay) + continue + raise raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}") @@ -95,15 +95,13 @@ async def fetch_bzzoiro_events( date_to: str | None = None, limit: int = 200, ) -> list[dict]: - """抓取 bzzoiro 原始事件(异步包装)。""" + """抓取 bzzoiro 原始事件(纯异步,无需 run_in_executor)。""" league_id = BZZOIRO_LEAGUE_IDS.get(league_code) if league_id is None: raise ValueError(f"未知联赛代码: {league_code}") - loop = asyncio.get_event_loop() rows: list[dict] = [] offset = 0 - payload: dict | list = {} while True: params: dict = { "league_id": league_id, @@ -115,8 +113,7 @@ async def fetch_bzzoiro_events( params["date_from"] = str(date_from)[:10] if date_to: params["date_to"] = str(date_to)[:10] - # 显式位置参数,避免 lambda 闭包捕获循环变量 - payload = await loop.run_in_executor(None, _fetch_json_sync, "/events/", params) + payload = await _fetch_json_async("/events/", params) batch = payload.get("results") or [] if not batch: break @@ -186,8 +183,8 @@ class BzzoiroSource: try: nm.validate() except Exception as e: - logger.debug("normalize skip: %s", e) - league_r["errors"].append(f"normalize: {e}") + # P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors) + logger.warning("normalize skip: %s", e) continue normalized_matches.append((nm, raw)) all_team_names.add(nm.home_team) diff --git a/src/data/injuries.py b/src/data/injuries.py index 2b1da01..8656f2d 100644 --- a/src/data/injuries.py +++ b/src/data/injuries.py @@ -8,6 +8,7 @@ import asyncio import json import logging import random +import tempfile import time from datetime import datetime, timezone from pathlib import Path @@ -23,8 +24,8 @@ logger = logging.getLogger(__name__) API_BASE = "https://v3.football.api-sports.io" DEFAULT_HOST = "v3.football.api-sports.io" -# 缓存目录 -_CACHE_DIR = Path(__file__).resolve().parent.parent.parent / "data" / "injuries_cache" +# P2-3: 缓存目录改用系统临时目录,避免源码树内写入 +_CACHE_DIR = Path(tempfile.gettempdir()) / "profeto_injuries" async def fetch_injuries(*, date: str | None = None, fixture_id: int | None = None, league_id: int | None = None) -> list[dict]: @@ -144,8 +145,17 @@ async def ingest_injuries(db, *, date: str | None = None) -> dict: except (ValueError, AttributeError): pass + # P2-1: 强制 int 转换,API 可能返回字符串 player_id = player.get("id") + try: + player_id = int(player_id) if player_id is not None else None + except (ValueError, TypeError): + player_id = None fixture_id = fixture.get("id") + try: + fixture_id = int(fixture_id) if fixture_id is not None else None + except (ValueError, TypeError): + fixture_id = None # 幂等: 已存在则跳过 existing = ( diff --git a/src/db/base.py b/src/db/base.py index c8c1d1e..d1b6b65 100644 --- a/src/db/base.py +++ b/src/db/base.py @@ -17,8 +17,10 @@ engine = create_async_engine( settings.DATABASE_URL, echo=False, pool_pre_ping=True, - pool_size=10, - max_overflow=20, + pool_size=settings.DB_POOL_SIZE, + max_overflow=settings.DB_MAX_OVERFLOW, + pool_timeout=settings.DB_POOL_TIMEOUT, + pool_recycle=settings.DB_POOL_RECYCLE, ) AsyncSessionLocal = async_sessionmaker( diff --git a/src/llm/backtest.py b/src/llm/backtest.py index 1d947bc..c8061ab 100644 --- a/src/llm/backtest.py +++ b/src/llm/backtest.py @@ -3,9 +3,11 @@ 核心机制: - build_context 已内置 before=match_date,天然防未来信息泄漏 - 对历史比赛跑预测 → 用实际比分 settle → 统计准确率 + - 并发控制: asyncio.Semaphore 限制同时 LLM 调用数 """ from __future__ import annotations +import asyncio import logging from dataclasses import dataclass, field from datetime import datetime @@ -17,6 +19,7 @@ from src.db.models import Match from src.db.unit_of_work import get_uow from src.llm.eval import settle_prediction from src.llm.predict import predict_match +from src.llm.utils import actual_1x2 logger = logging.getLogger(__name__) @@ -69,15 +72,6 @@ class BacktestSummary: results: list[BacktestMatchResult] = field(default_factory=list) -def _actual_1x2(home: int, away: int) -> str: - """实际比分 → 胜平负。""" - if home > away: - return "1" - if home == away: - return "X" - return "2" - - async def _get_historical_matches( db, *, @@ -156,44 +150,42 @@ async def run_backtest( summary = BacktestSummary(total=len(candidates), scored=0) - for c in candidates: - try: - # 预测 (build_context 内部已用 before=match_date 防泄漏, - # injuries_slice 也使用 as_of=match_date 过滤 retrieved_at) - # 回测必须禁用结果缓存: 否则命中缓存会复用同一 prediction_id, - # 导致 settle 反复覆盖同一条记录(见 P1-3)。 - result = await predict_match(c.match_id, mode=mode, model=model, use_cache=False) + # P1-6: 并发控制,同时最多 8 场预测(避免 LLM API 限流) + sem = asyncio.Semaphore(8) - # 用实际比分 settle - await settle_prediction(result.prediction_id, c.home_goals, c.away_goals) + async def _one(c: BacktestCandidate) -> BacktestMatchResult | None: + async with sem: + try: + result = await predict_match(c.match_id, mode=mode, model=model, use_cache=False, backtest=True) + await settle_prediction(result.prediction_id, c.home_goals, c.away_goals) + actual = actual_1x2(c.home_goals, c.away_goals) + return BacktestMatchResult( + match_id=c.match_id, + league_code=c.league_code, + home_team=c.home_team, + away_team=c.away_team, + match_date=c.match_date.strftime("%Y-%m-%d") if c.match_date else "?", + actual_home=c.home_goals, + actual_away=c.away_goals, + actual_1x2=actual, + pred_home=result.pred_home_goals, + pred_away=result.pred_away_goals, + pred_1x2=result.pred_1x2, + subjective_confidence=result.subjective_confidence, + correct_1x2=result.pred_1x2 == actual, + prediction_id=result.prediction_id, + ) + except Exception: + logger.exception("backtest match %s failed", c.match_id) + return None - actual = _actual_1x2(c.home_goals, c.away_goals) - correct = result.pred_1x2 == actual - - bt = BacktestMatchResult( - match_id=c.match_id, - league_code=c.league_code, - home_team=c.home_team, - away_team=c.away_team, - match_date=c.match_date.strftime("%Y-%m-%d") if c.match_date else "?", - actual_home=c.home_goals, - actual_away=c.away_goals, - actual_1x2=actual, - pred_home=result.pred_home_goals, - pred_away=result.pred_away_goals, - pred_1x2=result.pred_1x2, - subjective_confidence=result.subjective_confidence, - correct_1x2=correct, - prediction_id=result.prediction_id, - ) - summary.results.append(bt) + # 并行执行,保持结果顺序 + results = await asyncio.gather(*[_one(c) for c in candidates]) + for r in results: + if r is not None: + summary.results.append(r) summary.scored += 1 - except Exception: - # 用 exception 而非 warning:保留堆栈,否则集成层缺陷(如惰性加载 - # 在 session 外触发)会只剩一行无堆栈的 warning,极难定位。 - logger.exception("backtest match %s failed", c.match_id) - # 汇总统计 if summary.scored > 0: correct_count = sum(1 for r in summary.results if r.correct_1x2) diff --git a/src/llm/context_builder.py b/src/llm/context_builder.py index a928465..fbbd879 100644 --- a/src/llm/context_builder.py +++ b/src/llm/context_builder.py @@ -291,32 +291,39 @@ async def injuries_slice(header: MatchHeader, *, before=None) -> SliceResult: # 单 agent 路径: 拼接全部切片(行为与旧版一致) # ============================================================ -async def build_context(match_id: int, *, form_last: int = 5, h2h_last: int = 5) -> MatchContext: +async def build_context(match_id: int, *, form_last: int = 5, h2h_last: int = 5, backtest: bool = False) -> MatchContext: """单 agent 路径的完整上下文: 拼接全部切片(before=比赛时间,防未来信息)。 has_stats / has_injuries 直接取切片显式声明的 has_data, 不再靠文案子串匹配(见审查报告 P2-1)。 + + P2-6: backtest=True 时 cutoff = match_date - 1天,确保只用赛前数据。 """ header = await load_match_header(match_id) + # P2-6: 回测模式下 cutoff 提前 1 天,防止比赛日数据泄漏 + cutoff = header.match_dt + if backtest and header.match_dt: + from datetime import timedelta + cutoff = header.match_dt - timedelta(days=1) parts = [header_text(header), ""] - form_res = await form_slice(header, limit=form_last, before=header.match_dt) + form_res = await form_slice(header, limit=form_last, before=cutoff) parts.append(form_res.text) parts.append("") - h2h_res = await h2h_slice(header, limit=h2h_last, before=header.match_dt) + h2h_res = await h2h_slice(header, limit=h2h_last, before=cutoff) parts.append(h2h_res.text) parts.append("") - stats_res = await stats_slice(header, before=header.match_dt) + stats_res = await stats_slice(header, before=cutoff) parts.append(stats_res.text) parts.append("") - home_away_res = await home_away_slice(header, before=header.match_dt) + home_away_res = await home_away_slice(header, before=cutoff) parts.append(home_away_res.text) parts.append("") - injuries_res = await injuries_slice(header, before=header.match_dt) + injuries_res = await injuries_slice(header, before=cutoff) parts.append(injuries_res.text) return MatchContext( diff --git a/src/llm/predict.py b/src/llm/predict.py index 260c5f1..0dbd5d2 100644 --- a/src/llm/predict.py +++ b/src/llm/predict.py @@ -103,6 +103,7 @@ async def predict_match( prompt_version: str | None = None, mode: str = "multi", use_cache: bool = True, + backtest: bool = False, ) -> "PredictResult | MultiPredictResult": """预测入口。mode=multi(默认)走多 agent;mode=single 走单次调用。 @@ -110,6 +111,7 @@ async def predict_match( use_cache: 是否允许返回进程内缓存结果。回测必须传 False—— 缓存命中不会新建 prediction 行,调用方会对同一个 prediction_id 反复 settle,把不同比赛的真实比分覆盖到同一条记录上。 + backtest: 是否回测模式。True 时 build_context 使用 match_date-1天 作为 cutoff。 """ if mode == "single": return await _predict_single( @@ -118,6 +120,7 @@ async def predict_match( model=model, prompt_version=prompt_version, use_cache=use_cache, + backtest=backtest, ) from src.llm.agents.orchestrator import predict_match_multi @@ -131,6 +134,7 @@ async def _predict_single( model: str | None = None, prompt_version: str | None = None, use_cache: bool = True, + backtest: bool = False, ) -> PredictResult: """单次调用路径(原有实现)。""" if provider is None: @@ -147,8 +151,8 @@ async def _predict_single( logger.debug("predict cache hit match=%s", match_id) return cached - # 1. 拼上下文 - ctx = await build_context(match_id) + # 1. 拼上下文(P2-6: backtest 时使用 match_date-1天 作为 cutoff) + ctx = await build_context(match_id, backtest=backtest) # 1.5 计算快照元数据(用于可复现性) now = datetime.now(timezone.utc) diff --git a/src/llm/utils.py b/src/llm/utils.py new file mode 100644 index 0000000..3cbd4cb --- /dev/null +++ b/src/llm/utils.py @@ -0,0 +1,19 @@ +"""LLM 模块共享工具函数。""" +from __future__ import annotations + + +def actual_1x2(home: int, away: int) -> str: + """实际比分 → 胜平负。 + + 单一权威源: backtest.py 和 eval.py 共用,避免重复定义。 + """ + if home > away: + return "1" + if home == away: + return "X" + return "2" + + +def is_correct_1x2(pred: str | None, actual: str) -> bool: + """预测是否命中胜平负。""" + return pred == actual diff --git a/src/llm/validation.py b/src/llm/validation.py index d1b9cd8..9b16ee4 100644 --- a/src/llm/validation.py +++ b/src/llm/validation.py @@ -76,7 +76,7 @@ class PredictionOutputSchema(BaseModel): """ expected = _score_to_1x2(self.pred_home_goals, self.pred_away_goals) if self.pred_1x2 != expected: - logger.warning( + logger.debug( "1x2 与比分不一致: 比分 %.1f-%.1f 推出 '%s',但 LLM 给出 '%s';以比分修正", self.pred_home_goals, self.pred_away_goals, expected, self.pred_1x2, )