fix(P0): 修复三处静默失效的阻断缺陷

P0-1 backtest 会话生命周期:
- _get_historical_matches 加 selectinload(league/home_team/away_team),
  并在 session 内物化为 BacktestCandidate 纯数据快照,避免 session
  关闭后访问惰性关系抛 MissingGreenlet
- except 收窄并改用 logger.exception 保留堆栈
- 移除未使用的 and_ / League 导入

P0-2 切片函数关系属性 MissingGreenlet:
- models.py 为 Match.league/home_team/away_team/stats 声明 lazy=selectin
- context_builder 的 _get_form/_get_h2h/_get_home_away 显式 selectinload
  (修复 form_slice/h2h_slice 恒定失败,被 fail-open 掩盖的问题)
- repositories.find_by_teams_and_date 加 selectinload(Match.stats)

P0-3 bzzoiro raw 变量泄漏导致血缘错乱:
- normalized_matches 改为携带 (nm, raw) 元组,内层循环解包
- source_event_id / source_record_id 现在取到正确 event id
- 已有比赛补建 stats 时补齐 source/source_event_id/retrieved_at/available_at
- 抽取 _match_key()/_to_date() 统一日期键构造 (P1-6)
This commit is contained in:
WorkBuddy
2026-09-15 16:45:09 +08:00
parent 60e4b89822
commit 235fb0de97
6 changed files with 168 additions and 44 deletions
+36 -8
View File
@@ -27,6 +27,26 @@ from src.db.models import League, Match, MatchStats, Team
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _to_date(value):
"""把 datetime / date / str 统一成 `date`。"""
if value is None:
return None
if hasattr(value, "date") and callable(value.date):
return value.date()
return value
def _match_key(home_team_id: int, away_team_id: int, match_date) -> tuple[int, int, str]:
"""比赛去重键:(主队, 客队, 天级日期 ISO 字符串)。
统一在这里构造,避免"预加载时用 str(date)、写入时用 isoformat()"这类
隐式格式依赖 —— 两者当前恰好相等,但一旦有人改动其一就会静默失配,
导致所有比赛被判为不存在而重复插入。
"""
d = _to_date(match_date)
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: def _fetch_json_sync(path: str, params: dict | None = None, max_retries: int = 3) -> dict | list:
"""同步 HTTP(bzzoiro 客户端保持同步,在 async 函数里 run_in_executor)。""" """同步 HTTP(bzzoiro 客户端保持同步,在 async 函数里 run_in_executor)。"""
base = settings.BZZOIRO_BASE.rstrip("/") base = settings.BZZOIRO_BASE.rstrip("/")
@@ -153,7 +173,9 @@ class BzzoiroSource:
# === 批量优化: 预加载球队和已有比赛到内存 === # === 批量优化: 预加载球队和已有比赛到内存 ===
team_name_to_id: dict[str, int] = {} team_name_to_id: dict[str, int] = {}
existing_matches: dict[tuple[int, int, str], Match] = {} # 完整对象,避免重复查询 existing_matches: dict[tuple[int, int, str], Match] = {} # 完整对象,避免重复查询
normalized_matches: list = [] # 缓存规范化结果,避免重复调用 # (NormalizedMatch, 原始 event) 成对保存:后续写 source_record_id 时
# 必须用配对的那条 event,不能依赖外层循环变量残留值。
normalized_matches: list[tuple] = []
if raw_events: if raw_events:
# 一次遍历: 收集球队名 + 规范化 # 一次遍历: 收集球队名 + 规范化
@@ -167,7 +189,7 @@ class BzzoiroSource:
logger.debug("normalize skip: %s", e) logger.debug("normalize skip: %s", e)
league_r["errors"].append(f"normalize: {e}") league_r["errors"].append(f"normalize: {e}")
continue continue
normalized_matches.append(nm) normalized_matches.append((nm, raw))
all_team_names.add(nm.home_team) all_team_names.add(nm.home_team)
all_team_names.add(nm.away_team) all_team_names.add(nm.away_team)
@@ -179,10 +201,10 @@ class BzzoiroSource:
# 预加载已有比赛(完整对象) # 预加载已有比赛(完整对象)
stmt = select(Match).where(Match.league_id == league.id) stmt = select(Match).where(Match.league_id == league.id)
for m in (await db.execute(stmt)).scalars(): for m in (await db.execute(stmt)).scalars():
key = (m.home_team_id, m.away_team_id, str(m.match_date_date)) key = _match_key(m.home_team_id, m.away_team_id, m.match_date_date)
existing_matches[key] = m existing_matches[key] = m
for nm in normalized_matches: for nm, raw in normalized_matches:
# 球队: 内存查找 + 按需创建 # 球队: 内存查找 + 按需创建
home_team_id = team_name_to_id.get(nm.home_team) home_team_id = team_name_to_id.get(nm.home_team)
if home_team_id is None: if home_team_id is None:
@@ -201,8 +223,7 @@ class BzzoiroSource:
team_name_to_id[nm.away_team] = away_team_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 = _match_key(home_team_id, away_team_id, nm.date)
match_key = (home_team_id, away_team_id, date_key)
existing_match = existing_matches.get(match_key) existing_match = existing_matches.get(match_key)
if existing_match is None: if existing_match is None:
@@ -212,7 +233,7 @@ class BzzoiroSource:
home_team_id=home_team_id, home_team_id=home_team_id,
away_team_id=away_team_id, away_team_id=away_team_id,
match_date=nm.date, match_date=nm.date,
match_date_date=nm.date.date() if hasattr(nm.date, "date") else nm.date, match_date_date=_to_date(nm.date),
match_status=nm.match_status, match_status=nm.match_status,
home_goals=nm.home_goals, home_goals=nm.home_goals,
away_goals=nm.away_goals, away_goals=nm.away_goals,
@@ -263,7 +284,14 @@ class BzzoiroSource:
existing_match.match_stage = nm.match_stage existing_match.match_stage = nm.match_stage
changed = True changed = True
if existing_match.stats is None and (nm.home_xg is not None or nm.away_xg is not None): 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) now = datetime.now(timezone.utc)
existing_match.stats = MatchStats(
match_id=existing_match.id,
source="bzzoiro",
source_event_id=str(raw.get("id", "")),
retrieved_at=now,
available_at=now,
)
db.add(existing_match.stats) db.add(existing_match.stats)
await db.flush() await db.flush()
if existing_match.stats is not None: if existing_match.stats is not None:
+14 -4
View File
@@ -75,10 +75,20 @@ class Match(Base):
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow, onupdate=_utcnow)
league: Mapped[League] = relationship(back_populates="matches") league: Mapped[League] = relationship(back_populates="matches", lazy="selectin")
home_team: Mapped[Team] = relationship(foreign_keys=[home_team_id], back_populates="home_matches") # lazy="selectin": 这些关系在业务里几乎总是一起读取(切片/回测/展示)。
away_team: Mapped[Team] = relationship(foreign_keys=[away_team_id], back_populates="away_matches") # 默认的 lazy="select" 在 async SQLAlchemy 下,于 session 之外或未显式
stats: Mapped["MatchStats | None"] = relationship(back_populates="match", cascade="all, delete-orphan") # eager-load 时访问会抛 MissingGreenlet —— 已因此导致 form/stats/h2h
# 三个专家切片静默失败。统一改为预加载,从根上消除这类问题。
home_team: Mapped[Team] = relationship(
foreign_keys=[home_team_id], back_populates="home_matches", lazy="selectin"
)
away_team: Mapped[Team] = relationship(
foreign_keys=[away_team_id], back_populates="away_matches", lazy="selectin"
)
stats: Mapped["MatchStats | None"] = relationship(
back_populates="match", cascade="all, delete-orphan", lazy="selectin"
)
predictions: Mapped[list["Prediction"]] = relationship(back_populates="match", cascade="all, delete-orphan") predictions: Mapped[list["Prediction"]] = relationship(back_populates="match", cascade="all, delete-orphan")
__table_args__ = ( __table_args__ = (
+6 -1
View File
@@ -37,12 +37,17 @@ class MatchRepository:
async def find_by_teams_and_date( async def find_by_teams_and_date(
self, league_id: int, home_team_id: int, away_team_id: int, date self, league_id: int, home_team_id: int, away_team_id: int, date
) -> Match | None: ) -> Match | None:
"""按联赛+主队+客队+日期查找比赛(天级匹配)。""" """按联赛+主队+客队+日期查找比赛(天级匹配)。
预加载 stats:调用方(understat 回填)会读取 existing.stats,
async session 下惰性加载会抛 MissingGreenlet。
"""
if hasattr(date, "date"): if hasattr(date, "date"):
date = date.date() date = date.date()
stmt = ( stmt = (
select(Match) select(Match)
.options(selectinload(Match.stats))
.where(Match.league_id == league_id) .where(Match.league_id == league_id)
.where(Match.home_team_id == home_team_id) .where(Match.home_team_id == home_team_id)
.where(Match.away_team_id == away_team_id) .where(Match.away_team_id == away_team_id)
+65 -20
View File
@@ -8,10 +8,12 @@ from __future__ import annotations
import logging import logging
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime
from sqlalchemy import and_, select from sqlalchemy import select
from sqlalchemy.orm import selectinload
from src.db.models import League, Match 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
@@ -38,6 +40,23 @@ class BacktestMatchResult:
prediction_id: int prediction_id: int
@dataclass
class BacktestCandidate:
"""回测候选比赛(字段快照,不持有 ORM 对象)。
session 关闭后仍可安全读取:所有需要的关系字段已在查询时物化为普通值,
避免在 session 之外访问惰性加载的关系属性(会抛 MissingGreenlet)。
"""
match_id: int
league_code: str | None
home_team: str
away_team: str
match_date: datetime
home_goals: int
away_goals: int
@dataclass @dataclass
class BacktestSummary: class BacktestSummary:
"""回测汇总统计。""" """回测汇总统计。"""
@@ -66,10 +85,20 @@ async def _get_historical_matches(
date_from: str | None = None, date_from: str | None = None,
date_to: str | None = None, date_to: str | None = None,
limit: int = 50, limit: int = 50,
) -> list[Match]: ) -> list[BacktestCandidate]:
"""查询已完赛且有比分的比赛(回测候选)。""" """查询已完赛且有比分的比赛(回测候选)。
返回普通值快照而非 ORM 对象:调用方在 session 关闭后仍需使用这些字段,
而 league / home_team / away_team 是惰性加载关系,在 async 下于 session
之外访问会抛 MissingGreenlet。这里用 selectinload 预加载后立即物化。
"""
stmt = ( stmt = (
select(Match) select(Match)
.options(
selectinload(Match.league),
selectinload(Match.home_team),
selectinload(Match.away_team),
)
.where(Match.match_status == "finished") .where(Match.match_status == "finished")
.where(Match.home_goals.is_not(None)) .where(Match.home_goals.is_not(None))
.where(Match.away_goals.is_not(None)) .where(Match.away_goals.is_not(None))
@@ -83,7 +112,19 @@ async def _get_historical_matches(
stmt = stmt.order_by(Match.match_date.desc()).limit(limit) stmt = stmt.order_by(Match.match_date.desc()).limit(limit)
result = await db.execute(stmt) result = await db.execute(stmt)
return list(result.scalars().all()) # 在 session 内物化为纯数据,切断与 ORM 会话的耦合
return [
BacktestCandidate(
match_id=m.id,
league_code=m.league.code if m.league else None,
home_team=m.home_team.name if m.home_team else "?",
away_team=m.away_team.name if m.away_team else "?",
match_date=m.match_date,
home_goals=m.home_goals,
away_goals=m.away_goals,
)
for m in result.scalars().all()
]
async def run_backtest( async def run_backtest(
@@ -109,32 +150,34 @@ async def run_backtest(
BacktestSummary 含逐场结果 + 汇总统计 BacktestSummary 含逐场结果 + 汇总统计
""" """
async with get_uow() as session: async with get_uow() as session:
matches = await _get_historical_matches( candidates = await _get_historical_matches(
session, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit session, league_id=league_id, date_from=date_from, date_to=date_to, limit=limit
) )
summary = BacktestSummary(total=len(matches), scored=0) summary = BacktestSummary(total=len(candidates), scored=0)
for m in matches: for c in candidates:
try: try:
# 预测 (build_context 内部已用 before=match_date 防泄漏, # 预测 (build_context 内部已用 before=match_date 防泄漏,
# injuries_slice 也使用 as_of=match_date 过滤 retrieved_at) # injuries_slice 也使用 as_of=match_date 过滤 retrieved_at)
result = await predict_match(m.id, mode=mode, model=model) # 回测必须禁用结果缓存: 否则命中缓存会复用同一 prediction_id,
# 导致 settle 反复覆盖同一条记录(见 P1-3)。
result = await predict_match(c.match_id, mode=mode, model=model, use_cache=False)
# 用实际比分 settle # 用实际比分 settle
await settle_prediction(result.prediction_id, m.home_goals, m.away_goals) await settle_prediction(result.prediction_id, c.home_goals, c.away_goals)
actual = _actual_1x2(m.home_goals, m.away_goals) actual = _actual_1x2(c.home_goals, c.away_goals)
correct = result.pred_1x2 == actual correct = result.pred_1x2 == actual
bt = BacktestMatchResult( bt = BacktestMatchResult(
match_id=m.id, match_id=c.match_id,
league_code=m.league.code if m.league else None, league_code=c.league_code,
home_team=m.home_team.name if m.home_team else "?", home_team=c.home_team,
away_team=m.away_team.name if m.away_team else "?", away_team=c.away_team,
match_date=m.match_date.strftime("%Y-%m-%d") if m.match_date else "?", match_date=c.match_date.strftime("%Y-%m-%d") if c.match_date else "?",
actual_home=m.home_goals, actual_home=c.home_goals,
actual_away=m.away_goals, actual_away=c.away_goals,
actual_1x2=actual, actual_1x2=actual,
pred_home=result.pred_home_goals, pred_home=result.pred_home_goals,
pred_away=result.pred_away_goals, pred_away=result.pred_away_goals,
@@ -146,8 +189,10 @@ async def run_backtest(
summary.results.append(bt) summary.results.append(bt)
summary.scored += 1 summary.scored += 1
except Exception as e: except Exception:
logger.warning("backtest match %s failed: %s", m.id, e) # 用 exception 而非 warning:保留堆栈,否则集成层缺陷(如惰性加载
# 在 session 外触发)会只剩一行无堆栈的 warning,极难定位。
logger.exception("backtest match %s failed", c.match_id)
# 汇总统计 # 汇总统计
if summary.scored > 0: if summary.scored > 0:
+25 -3
View File
@@ -328,9 +328,19 @@ async def _load_match(db, match_id: int) -> Match:
async def _get_form(db, team_id: int, before, *, limit: int = 5) -> list[Match]: async def _get_form(db, team_id: int, before, *, limit: int = 5) -> list[Match]:
"""某队近 N 场(已完赛)。before=None 表示不限制(预测赛前的场景由调用方保证)。""" """某队近 N 场(已完赛)。before=None 表示不限制(预测赛前的场景由调用方保证)。
必须预加载 stats / home_team / away_team:切片函数会读取这些关系,
而 async session 下惰性加载会抛 MissingGreenlet。
(models.py 已声明 lazy="selectin",此处显式声明以固化查询意图。)
"""
stmt = ( stmt = (
select(Match) select(Match)
.options(
selectinload(Match.stats),
selectinload(Match.home_team),
selectinload(Match.away_team),
)
.where(Match.match_status == "finished") .where(Match.match_status == "finished")
.where(Match.home_goals.is_not(None)) .where(Match.home_goals.is_not(None))
.where((Match.home_team_id == team_id) | (Match.away_team_id == team_id)) .where((Match.home_team_id == team_id) | (Match.away_team_id == team_id))
@@ -344,9 +354,13 @@ async def _get_form(db, team_id: int, before, *, limit: int = 5) -> list[Match]:
async def _get_h2h(db, home_id: int, away_id: int, before, *, limit: int = 5) -> list[Match]: async def _get_h2h(db, home_id: int, away_id: int, before, *, limit: int = 5) -> list[Match]:
"""两队交锋史。""" """两队交锋史。需预加载 home_team / away_team(切片输出队名)。"""
stmt = ( stmt = (
select(Match) select(Match)
.options(
selectinload(Match.home_team),
selectinload(Match.away_team),
)
.where(Match.match_status == "finished") .where(Match.match_status == "finished")
.where(Match.home_goals.is_not(None)) .where(Match.home_goals.is_not(None))
.where( .where(
@@ -363,9 +377,17 @@ async def _get_h2h(db, home_id: int, away_id: int, before, *, limit: int = 5) ->
async def _get_home_away(db, team_id: int, side: str, before, *, limit: int = 10) -> list[Match]: async def _get_home_away(db, team_id: int, side: str, before, *, limit: int = 10) -> list[Match]:
"""某队主场/客场近 N 场。side='home' 取主场,'away' 取客场。""" """某队主场/客场近 N 场。side='home' 取主场,'away' 取客场。
当前只用标量字段,但统一预加载以免后续扩展时踩坑。
"""
stmt = ( stmt = (
select(Match) select(Match)
.options(
selectinload(Match.stats),
selectinload(Match.home_team),
selectinload(Match.away_team),
)
.where(Match.match_status == "finished") .where(Match.match_status == "finished")
.where(Match.home_goals.is_not(None)) .where(Match.home_goals.is_not(None))
.order_by(Match.match_date.desc()) .order_by(Match.match_date.desc())
+22 -8
View File
@@ -81,11 +81,22 @@ async def predict_match(
model: str | None = None, model: str | None = None,
prompt_version: str | None = None, prompt_version: str | None = None,
mode: str = "multi", mode: str = "multi",
use_cache: bool = True,
) -> "PredictResult | MultiPredictResult": ) -> "PredictResult | MultiPredictResult":
"""预测入口。mode=multi(默认)走多 agent;mode=single 走单次调用。""" """预测入口。mode=multi(默认)走多 agent;mode=single 走单次调用。
Args:
use_cache: 是否允许返回进程内缓存结果。回测必须传 False——
缓存命中不会新建 prediction 行,调用方会对同一个 prediction_id
反复 settle,把不同比赛的真实比分覆盖到同一条记录上。
"""
if mode == "single": if mode == "single":
return await _predict_single( return await _predict_single(
match_id, provider=provider, model=model, prompt_version=prompt_version match_id,
provider=provider,
model=model,
prompt_version=prompt_version,
use_cache=use_cache,
) )
from src.llm.agents.orchestrator import predict_match_multi from src.llm.agents.orchestrator import predict_match_multi
@@ -98,6 +109,7 @@ async def _predict_single(
provider: LLMProvider | None = None, provider: LLMProvider | None = None,
model: str | None = None, model: str | None = None,
prompt_version: str | None = None, prompt_version: str | None = None,
use_cache: bool = True,
) -> PredictResult: ) -> PredictResult:
"""单次调用路径(原有实现)。""" """单次调用路径(原有实现)。"""
if provider is None: if provider is None:
@@ -107,10 +119,11 @@ async def _predict_single(
version = prompt_version or "v1" version = prompt_version or "v1"
# 0. 查缓存(同 match+provider+model+version 5 分钟内直接返) # 0. 查缓存(同 match+provider+model+version 5 分钟内直接返)
cached = _get_cached(match_id, settings.LLM_PROVIDER, provider.model, version) if use_cache:
if cached is not None: cached = _get_cached(match_id, settings.LLM_PROVIDER, provider.model, version)
logger.debug("predict cache hit match=%s", match_id) if cached is not None:
return cached logger.debug("predict cache hit match=%s", match_id)
return cached
# 1. 拼上下文 # 1. 拼上下文
ctx = await build_context(match_id) ctx = await build_context(match_id)
@@ -191,6 +204,7 @@ async def _predict_single(
raw=resp.raw, raw=resp.raw,
) )
# 5. 写入缓存 # 5. 写入缓存(仅当允许缓存时)
_set_cached(match_id, settings.LLM_PROVIDER, provider.model, version, result) if use_cache:
_set_cached(match_id, settings.LLM_PROVIDER, provider.model, version, result)
return result return result