Author SHA1 Message Date
shangfangjian 983b620659 refactor: fix P0/P1/P2 security and performance issues
P0 fixes:
- CORS: replace wildcard methods/headers with configurable lists
- deps.py: remove unsafe global _warned_unset variable

P1 fixes:
- http_client: read default timeout from Settings
- bzzoiro: replace sync urllib with async httpx
- bzzoiro: normalize validation failures use warning level only
- db pool: read pool config from Settings (default 5+10)
- backtest: add asyncio.Semaphore(8) for concurrent execution
- predict/context_builder: add backtest parameter for cutoff buffer

P2 improvements:
- injuries: enforce int conversion for player_id/fixture_id
- injuries: use system temp dir for cache
- utils.py: extract shared actual_1x2/is_correct_1x2
- validation: downgrade 1x2 mismatch log to debug
- docker-compose: use env vars for all credentials
- .env.example: add POSTGRES_USER/PASSWORD/PORT, API_PORT
2026-09-16 02:38:25 +08:00
shangfangjian fa69795d69 fix: 应用代码审查三项发现
- frontend: .tab 补 position:relative,修复激活 tab 下划线定位缺失定位上下文
- core/retry.py: 删除全项目零调用的重试模块(避免"看似生效实际未接入"的误判)
- llm/validation.py: KNOWN_AGENT_NAMES 改为从 orchestrator.SPECIALIST_SPECS 派生,单一权威源
2026-09-15 20:12:47 +08:00
shangfangjian c5f92c9a54 Merge pull request 'refactor(ui): 前端重做为报刊赛程版视觉' (#2) from frontend-beautify into main
Reviewed-on: #2
2026-09-15 20:00:05 +08:00
16 changed files with 159 additions and 201 deletions
+4
View File
@@ -3,6 +3,10 @@ APP_ENV=development
LOG_LEVEL=INFO LOG_LEVEL=INFO
# ---- 数据库 ---- # ---- 数据库 ----
POSTGRES_USER=football
POSTGRES_PASSWORD=football
POSTGRES_DB=football
POSTGRES_PORT=5432
DATABASE_URL=postgresql+asyncpg://football:football@localhost:5432/football DATABASE_URL=postgresql+asyncpg://football:football@localhost:5432/football
# ---- LLM (OpenAI-compatible,必填一个) ---- # ---- LLM (OpenAI-compatible,必填一个) ----
+6 -6
View File
@@ -2,15 +2,15 @@ services:
postgres: postgres:
image: postgres:16-alpine image: postgres:16-alpine
environment: environment:
POSTGRES_USER: football POSTGRES_USER: ${POSTGRES_USER:?POSTGRES_USER 未设置}
POSTGRES_PASSWORD: football POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?POSTGRES_PASSWORD 未设置}
POSTGRES_DB: football POSTGRES_DB: ${POSTGRES_DB:-football}
ports: ports:
- "5432:5432" - "${POSTGRES_PORT:-5432}:5432"
volumes: volumes:
- pgdata:/var/lib/postgresql/data - pgdata:/var/lib/postgresql/data
healthcheck: healthcheck:
test: ["CMD-SHELL", "pg_isready -U football"] test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:?POSTGRES_USER 未设置}"]
interval: 5s interval: 5s
timeout: 5s timeout: 5s
retries: 5 retries: 5
@@ -19,7 +19,7 @@ services:
build: . build: .
command: uvicorn src.api.app:app --host 0.0.0.0 --port 8000 --reload command: uvicorn src.api.app:app --host 0.0.0.0 --port 8000 --reload
ports: ports:
- "8000:8000" - "${API_PORT:-8000}:8000"
env_file: .env env_file: .env
depends_on: depends_on:
postgres: postgres:
+1 -1
View File
@@ -80,7 +80,7 @@
/* ── 版面切换文字标签(联赛/状态/模式) ── */ /* ── 版面切换文字标签(联赛/状态/模式) ── */
.tab { .tab {
@apply whitespace-nowrap px-0.5 py-1 text-sm text-ink-500 transition-colors hover:text-ink-900; @apply relative whitespace-nowrap px-0.5 py-1 text-sm text-ink-500 transition-colors hover:text-ink-900;
} }
.tab-on { .tab-on {
@apply font-medium text-press; @apply font-medium text-press;
+4 -2
View File
@@ -29,12 +29,14 @@ def create_app() -> FastAPI:
) )
origins = [o.strip() for o in settings.CORS_ORIGINS.split(",") if o.strip()] 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( app.add_middleware(
CORSMiddleware, CORSMiddleware,
allow_origins=origins, allow_origins=origins,
allow_credentials=True, allow_credentials=True,
allow_methods=["*"], allow_methods=methods,
allow_headers=["*"], allow_headers=headers,
) )
from src.api.routes.matches import router as matches_router from src.api.routes.matches import router as matches_router
-6
View File
@@ -19,24 +19,18 @@ from src.core.config import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_warned_unset = False
async def require_admin_key(x_api_key: str | None = Header(None, alias="X-API-Key")) -> None: 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)])` 用法: `@router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin_key)])`
""" """
global _warned_unset
expected = settings.ADMIN_API_KEY expected = settings.ADMIN_API_KEY
if not expected: if not expected:
if not _warned_unset:
logger.warning( logger.warning(
"ADMIN_API_KEY 未设置,采集/回测接口当前【无鉴权】。" "ADMIN_API_KEY 未设置,采集/回测接口当前【无鉴权】。"
"生产环境请设置该环境变量。" "生产环境请设置该环境变量。"
) )
_warned_unset = True
return return
if not x_api_key or not secrets.compare_digest(x_api_key, expected): if not x_api_key or not secrets.compare_digest(x_api_key, expected):
+11
View File
@@ -32,6 +32,17 @@ class Settings(BaseSettings):
# --- CORS --- # --- CORS ---
CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000" 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)。 # 采集 / 回测等高成本或写入型接口需要此 Key(请求头 X-API-Key)。
+5 -2
View File
@@ -2,24 +2,27 @@
使用方: 使用方:
- src/llm/provider.py: LLM 调用 - src/llm/provider.py: LLM 调用
- src/data/bzzoiro.py: bzzoiro 比赛数据
- src/data/understat.py: xG 抓取 - src/data/understat.py: xG 抓取
- src/data/injuries.py: 伤停抓取 - src/data/injuries.py: 伤停抓取
生命周期由 FastAPI lifespan 管理(关闭时 aclose)。 生命周期由 FastAPI lifespan 管理(关闭时 aclose)。
调用方可通过 `timeout` 参数覆盖 per-request 超时。
""" """
from __future__ import annotations from __future__ import annotations
import httpx import httpx
from src.core.config import settings
_shared_client: httpx.AsyncClient | None = None _shared_client: httpx.AsyncClient | None = None
_default_timeout = 30
def get_client() -> httpx.AsyncClient: def get_client() -> httpx.AsyncClient:
"""获取共享客户端(懒初始化)。""" """获取共享客户端(懒初始化)。"""
global _shared_client global _shared_client
if _shared_client is None or _shared_client.is_closed: 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 return _shared_client
-89
View File
@@ -1,89 +0,0 @@
"""重试工具:带指数退避的瞬态错误重试。
NOTE(审查报告 P3):当前全项目**无调用点** —— bzzoiro 在 `_fetch_json_sync`
里自带了一套重试逻辑,understat/injuries 各自也有。这里保留是作为后续统一
重试策略的落点,但请勿误以为它已在生效。
如果决定不引入统一重试,建议删除本文件以避免"看起来有重试、实际没有"的误判。
"""
from __future__ import annotations
import asyncio
import functools
import logging
import random
import time
from typing import Callable, Iterable, TypeVar
logger = logging.getLogger(__name__)
T = TypeVar("T")
def with_retry(
*,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
retryable_exceptions: Iterable[type[BaseException]] = (Exception,),
on_retry: Callable[[Exception, int], None] | None = None,
) -> Callable:
"""重试装饰器(同步/异步通用,指数退避 + 抖动)。
Args:
max_retries: 最大重试次数
base_delay: 基础延迟(秒)
max_delay: 最大延迟(秒)
retryable_exceptions: 触发重试的异常类型
on_retry: 重试回调(exception, attempt)
"""
retryable = tuple(retryable_exceptions)
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
last_exc: Exception | None = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except retryable as e:
last_exc = e
if attempt == max_retries:
break
delay = min(base_delay * (2 ** attempt), max_delay)
delay += random.uniform(0, delay * 0.1) # 抖动
logger.warning(
"%s failed (attempt %d/%d), retry in %.1fs: %s",
func.__name__, attempt + 1, max_retries, delay, e,
)
if on_retry:
on_retry(e, attempt + 1)
await asyncio.sleep(delay)
raise last_exc # type: ignore[misc]
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
last_exc: Exception | None = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except retryable as e:
last_exc = e
if attempt == max_retries:
break
delay = min(base_delay * (2 ** attempt), max_delay)
delay += random.uniform(0, delay * 0.1)
logger.warning(
"%s failed (attempt %d/%d), retry in %.1fs: %s",
func.__name__, attempt + 1, max_retries, delay, e,
)
if on_retry:
on_retry(e, attempt + 1)
time.sleep(delay)
raise last_exc # type: ignore[misc]
if asyncio.iscoroutinefunction(func):
return async_wrapper
return sync_wrapper
return decorator
+28 -31
View File
@@ -9,16 +9,13 @@ import asyncio
import json as _json import json as _json
import logging import logging
import random import random
import time as _time
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Iterable from collections.abc import Iterable
from datetime import datetime, timezone from datetime import datetime, timezone
from sqlalchemy import select from sqlalchemy import select
from src.core.config import settings 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.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES, REQUEST_INTERVAL
from src.data.normalize import normalize_bzzoiro from src.data.normalize import normalize_bzzoiro
from src.data.sources import register 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 "") 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: async def _fetch_json_async(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
"""步 HTTP(bzzoiro 客户端保持同步,在 async 函数里 run_in_executor)。""" """步 HTTP(bzzoiro 使用 httpx,不再阻塞事件循环线程池)。"""
base = settings.BZZOIRO_BASE.rstrip("/") base = settings.BZZOIRO_BASE.rstrip("/")
url = f"{base}/{path.lstrip('/')}" url = f"{base}/{path.lstrip('/')}"
if params:
url += "?" + urllib.parse.urlencode(params)
key = settings.BZZOIRO_KEY key = settings.BZZOIRO_KEY
if not key: if not key:
raise RuntimeError("BZZOIRO_KEY 未设置") raise RuntimeError("BZZOIRO_KEY 未设置")
headers = {
"Authorization": f"Token {key}",
"Accept": "application/json",
}
last_exc: Exception | None = None last_exc: Exception | None = None
for attempt in range(max_retries): for attempt in range(max_retries):
try: try:
req = urllib.request.Request(url) client = get_client()
req.add_header("Authorization", f"Token {key}") resp = await client.get(url, headers=headers, params=params, timeout=30)
req.add_header("Accept", "application/json") resp.raise_for_status()
with urllib.request.urlopen(req, timeout=30) as resp: return resp.json()
return _json.loads(resp.read().decode("utf-8")) except Exception as e:
except urllib.error.HTTPError as e:
last_exc = e last_exc = e
if e.code == 429: status = getattr(getattr(e, "response", None), "status_code", None)
# 指数退避: 429 通常意味着限速 if status == 429:
delay = min(2 ** attempt, 16) + random.uniform(0, 1) delay = min(2 ** attempt, 16) + random.uniform(0, 1)
logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay) logger.warning("bzzoiro 429, retry %d in %.1fs", attempt + 1, delay)
_time.sleep(delay) await asyncio.sleep(delay)
continue continue
if 500 <= e.code < 600: if 500 <= (status or 0) < 600:
delay = min(2 ** attempt, 16) + random.uniform(0, 1) delay = min(2 ** attempt, 16) + random.uniform(0, 1)
logger.warning("bzzoiro %d, retry %d in %.1fs", e.code, attempt + 1, delay) logger.warning("bzzoiro %d, retry %d in %.1fs", status, attempt + 1, delay)
_time.sleep(delay) await asyncio.sleep(delay)
continue continue
raise # 4xx 直接抛 # 网络错误(连接失败/超时)也退避重试
except (urllib.error.URLError, TimeoutError, ConnectionError) as e: if isinstance(e, (TimeoutError, ConnectionError, OSError)):
last_exc = e
delay = min(2 ** attempt, 16) + random.uniform(0, 1) delay = min(2 ** attempt, 16) + random.uniform(0, 1)
logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e) logger.warning("bzzoiro network error, retry %d in %.1fs: %s", attempt + 1, delay, e)
_time.sleep(delay) await asyncio.sleep(delay)
continue
raise
raise RuntimeError(f"bzzoiro request failed after {max_retries} attempts: {last_exc}") 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, date_to: str | None = None,
limit: int = 200, limit: int = 200,
) -> list[dict]: ) -> list[dict]:
"""抓取 bzzoiro 原始事件(异步包装)。""" """抓取 bzzoiro 原始事件(异步,无需 run_in_executor)。"""
league_id = BZZOIRO_LEAGUE_IDS.get(league_code) league_id = BZZOIRO_LEAGUE_IDS.get(league_code)
if league_id is None: if league_id is None:
raise ValueError(f"未知联赛代码: {league_code}") raise ValueError(f"未知联赛代码: {league_code}")
loop = asyncio.get_event_loop()
rows: list[dict] = [] rows: list[dict] = []
offset = 0 offset = 0
payload: dict | list = {}
while True: while True:
params: dict = { params: dict = {
"league_id": league_id, "league_id": league_id,
@@ -115,8 +113,7 @@ async def fetch_bzzoiro_events(
params["date_from"] = str(date_from)[:10] params["date_from"] = str(date_from)[:10]
if date_to: if date_to:
params["date_to"] = str(date_to)[:10] params["date_to"] = str(date_to)[:10]
# 显式位置参数,避免 lambda 闭包捕获循环变量 payload = await _fetch_json_async("/events/", params)
payload = await loop.run_in_executor(None, _fetch_json_sync, "/events/", params)
batch = payload.get("results") or [] batch = payload.get("results") or []
if not batch: if not batch:
break break
@@ -186,8 +183,8 @@ class BzzoiroSource:
try: try:
nm.validate() nm.validate()
except Exception as e: except Exception as e:
logger.debug("normalize skip: %s", e) # P1-3: 统一使用 warning,不追加到 errors(仅运行时错误入 errors)
league_r["errors"].append(f"normalize: {e}") logger.warning("normalize skip: %s", e)
continue continue
normalized_matches.append((nm, raw)) normalized_matches.append((nm, raw))
all_team_names.add(nm.home_team) all_team_names.add(nm.home_team)
+12 -2
View File
@@ -8,6 +8,7 @@ import asyncio
import json import json
import logging import logging
import random import random
import tempfile
import time import time
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
@@ -23,8 +24,8 @@ logger = logging.getLogger(__name__)
API_BASE = "https://v3.football.api-sports.io" API_BASE = "https://v3.football.api-sports.io"
DEFAULT_HOST = "v3.football.api-sports.io" DEFAULT_HOST = "v3.football.api-sports.io"
# 缓存目录 # P2-3: 缓存目录改用系统临时目录,避免源码树内写入
_CACHE_DIR = Path(__file__).resolve().parent.parent.parent / "data" / "injuries_cache" _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]: 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): except (ValueError, AttributeError):
pass pass
# P2-1: 强制 int 转换,API 可能返回字符串
player_id = player.get("id") 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") 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 = ( existing = (
+4 -2
View File
@@ -17,8 +17,10 @@ engine = create_async_engine(
settings.DATABASE_URL, settings.DATABASE_URL,
echo=False, echo=False,
pool_pre_ping=True, pool_pre_ping=True,
pool_size=10, pool_size=settings.DB_POOL_SIZE,
max_overflow=20, max_overflow=settings.DB_MAX_OVERFLOW,
pool_timeout=settings.DB_POOL_TIMEOUT,
pool_recycle=settings.DB_POOL_RECYCLE,
) )
AsyncSessionLocal = async_sessionmaker( AsyncSessionLocal = async_sessionmaker(
+20 -28
View File
@@ -3,9 +3,11 @@
核心机制: 核心机制:
- build_context 已内置 before=match_date,天然防未来信息泄漏 - build_context 已内置 before=match_date,天然防未来信息泄漏
- 对历史比赛跑预测 → 用实际比分 settle → 统计准确率 - 对历史比赛跑预测 → 用实际比分 settle → 统计准确率
- 并发控制: asyncio.Semaphore 限制同时 LLM 调用数
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import logging import logging
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime 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.db.unit_of_work import get_uow
from src.llm.eval import settle_prediction from src.llm.eval import settle_prediction
from src.llm.predict import predict_match from src.llm.predict import predict_match
from src.llm.utils import actual_1x2
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -69,15 +72,6 @@ class BacktestSummary:
results: list[BacktestMatchResult] = field(default_factory=list) 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( async def _get_historical_matches(
db, db,
*, *,
@@ -156,21 +150,16 @@ async def run_backtest(
summary = BacktestSummary(total=len(candidates), scored=0) summary = BacktestSummary(total=len(candidates), scored=0)
for c in candidates: # P1-6: 并发控制,同时最多 8 场预测(避免 LLM API 限流)
sem = asyncio.Semaphore(8)
async def _one(c: BacktestCandidate) -> BacktestMatchResult | None:
async with sem:
try: try:
# 预测 (build_context 内部已用 before=match_date 防泄漏, result = await predict_match(c.match_id, mode=mode, model=model, use_cache=False, backtest=True)
# 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)
# 用实际比分 settle
await settle_prediction(result.prediction_id, c.home_goals, c.away_goals) await settle_prediction(result.prediction_id, c.home_goals, c.away_goals)
actual = actual_1x2(c.home_goals, c.away_goals)
actual = _actual_1x2(c.home_goals, c.away_goals) return BacktestMatchResult(
correct = result.pred_1x2 == actual
bt = BacktestMatchResult(
match_id=c.match_id, match_id=c.match_id,
league_code=c.league_code, league_code=c.league_code,
home_team=c.home_team, home_team=c.home_team,
@@ -183,16 +172,19 @@ async def run_backtest(
pred_away=result.pred_away_goals, pred_away=result.pred_away_goals,
pred_1x2=result.pred_1x2, pred_1x2=result.pred_1x2,
subjective_confidence=result.subjective_confidence, subjective_confidence=result.subjective_confidence,
correct_1x2=correct, correct_1x2=result.pred_1x2 == actual,
prediction_id=result.prediction_id, prediction_id=result.prediction_id,
) )
summary.results.append(bt)
summary.scored += 1
except Exception: except Exception:
# 用 exception 而非 warning:保留堆栈,否则集成层缺陷(如惰性加载
# 在 session 外触发)会只剩一行无堆栈的 warning,极难定位。
logger.exception("backtest match %s failed", c.match_id) logger.exception("backtest match %s failed", c.match_id)
return None
# 并行执行,保持结果顺序
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
# 汇总统计 # 汇总统计
if summary.scored > 0: if summary.scored > 0:
+13 -6
View File
@@ -291,32 +291,39 @@ async def injuries_slice(header: MatchHeader, *, before=None) -> SliceResult:
# 单 agent 路径: 拼接全部切片(行为与旧版一致) # 单 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=比赛时间,防未来信息)。 """单 agent 路径的完整上下文: 拼接全部切片(before=比赛时间,防未来信息)。
has_stats / has_injuries 直接取切片显式声明的 has_data, has_stats / has_injuries 直接取切片显式声明的 has_data,
不再靠文案子串匹配(见审查报告 P2-1)。 不再靠文案子串匹配(见审查报告 P2-1)。
P2-6: backtest=True 时 cutoff = match_date - 1天,确保只用赛前数据。
""" """
header = await load_match_header(match_id) 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), ""] 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(form_res.text)
parts.append("") 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(h2h_res.text)
parts.append("") 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(stats_res.text)
parts.append("") 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(home_away_res.text)
parts.append("") 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) parts.append(injuries_res.text)
return MatchContext( return MatchContext(
+6 -2
View File
@@ -103,6 +103,7 @@ async def predict_match(
prompt_version: str | None = None, prompt_version: str | None = None,
mode: str = "multi", mode: str = "multi",
use_cache: bool = True, use_cache: bool = True,
backtest: bool = False,
) -> "PredictResult | MultiPredictResult": ) -> "PredictResult | MultiPredictResult":
"""预测入口。mode=multi(默认)走多 agent;mode=single 走单次调用。 """预测入口。mode=multi(默认)走多 agent;mode=single 走单次调用。
@@ -110,6 +111,7 @@ async def predict_match(
use_cache: 是否允许返回进程内缓存结果。回测必须传 False—— use_cache: 是否允许返回进程内缓存结果。回测必须传 False——
缓存命中不会新建 prediction 行,调用方会对同一个 prediction_id 缓存命中不会新建 prediction 行,调用方会对同一个 prediction_id
反复 settle,把不同比赛的真实比分覆盖到同一条记录上。 反复 settle,把不同比赛的真实比分覆盖到同一条记录上。
backtest: 是否回测模式。True 时 build_context 使用 match_date-1天 作为 cutoff。
""" """
if mode == "single": if mode == "single":
return await _predict_single( return await _predict_single(
@@ -118,6 +120,7 @@ async def predict_match(
model=model, model=model,
prompt_version=prompt_version, prompt_version=prompt_version,
use_cache=use_cache, use_cache=use_cache,
backtest=backtest,
) )
from src.llm.agents.orchestrator import predict_match_multi from src.llm.agents.orchestrator import predict_match_multi
@@ -131,6 +134,7 @@ async def _predict_single(
model: str | None = None, model: str | None = None,
prompt_version: str | None = None, prompt_version: str | None = None,
use_cache: bool = True, use_cache: bool = True,
backtest: bool = False,
) -> PredictResult: ) -> PredictResult:
"""单次调用路径(原有实现)。""" """单次调用路径(原有实现)。"""
if provider is None: if provider is None:
@@ -147,8 +151,8 @@ async def _predict_single(
logger.debug("predict cache hit match=%s", match_id) logger.debug("predict cache hit match=%s", match_id)
return cached return cached
# 1. 拼上下文 # 1. 拼上下文(P2-6: backtest 时使用 match_date-1天 作为 cutoff)
ctx = await build_context(match_id) ctx = await build_context(match_id, backtest=backtest)
# 1.5 计算快照元数据(用于可复现性) # 1.5 计算快照元数据(用于可复现性)
now = datetime.now(timezone.utc) now = datetime.now(timezone.utc)
+19
View File
@@ -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
+5 -3
View File
@@ -10,8 +10,10 @@ from pydantic import BaseModel, Field, field_validator, model_validator
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 已知的 5 个专家 agent 名(与 orchestrator.SPECIALIST_SPECS 保持一致) # 单一权威源:从 orchestrator.SPECIALIST_SPECS 派生,避免两端独立定义导致静默偏离
KNOWN_AGENT_NAMES: tuple[str, ...] = ("form", "stats", "home_away", "injuries", "h2h") from src.llm.agents.orchestrator import SPECIALIST_SPECS
KNOWN_AGENT_NAMES: tuple[str, ...] = tuple(spec.name for spec in SPECIALIST_SPECS)
class AgentReportSchema(BaseModel): class AgentReportSchema(BaseModel):
@@ -74,7 +76,7 @@ class PredictionOutputSchema(BaseModel):
""" """
expected = _score_to_1x2(self.pred_home_goals, self.pred_away_goals) expected = _score_to_1x2(self.pred_home_goals, self.pred_away_goals)
if self.pred_1x2 != expected: if self.pred_1x2 != expected:
logger.warning( logger.debug(
"1x2 与比分不一致: 比分 %.1f-%.1f 推出 '%s',但 LLM 给出 '%s';以比分修正", "1x2 与比分不一致: 比分 %.1f-%.1f 推出 '%s',但 LLM 给出 '%s';以比分修正",
self.pred_home_goals, self.pred_away_goals, expected, self.pred_1x2, self.pred_home_goals, self.pred_away_goals, expected, self.pred_1x2,
) )