P1-1 [context_builder] build_context 共享 session,切片函数传 db 参数,
回测 20 场并发连接需求从 100+ 降至每场 1 个
P1-2 [bzzoiro] 预加载改为按 raw_events 日期范围 ±30 天按需加载
P1-3 [understat] 批量查询球队 + 比赛,从 1140 次往返降到 3 次
P1-4 [injuries] 批量幂等检查 + 分批 flush,IntegrityError 逐条回退
P1-5 [predict] 删除 threading.Lock,dict 操作原子无需同步锁
P1-6 [models] 添加 (match_id, provider, model) 唯一约束 + 迁移
P2-1 [unit_of_work] get_uow 返回类型改为 AsyncIterator[AsyncSession]
P2-2 [normalize] _parse_date 失败时记录 warning 避免静默丢数据
P2-3 [repositories] find_by_teams_and_date 改用 match_date_date 等值匹配
P2-4 [migration] 幽灵列 cutoff_at 已在 0006 迁移删除(已有)
P2-5 [migration] injuries 约束命名对齐 ORM,UniqueConstraint → 唯一索引
215 lines
7.8 KiB
Python
215 lines
7.8 KiB
Python
"""数据规范化:任意数据源原始记录 → NormalizedMatch。
|
|
|
|
迁移自旧项目 app/data/normalize.py,简化:
|
|
- 去掉 XGBackoff 双轨(不再需要独立回填)
|
|
- 去掉 PIT 时间契约(无训练集要防泄漏)
|
|
- 保留核心清洗契约(队名归一、日期解析、数值范围)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VALID_STATUS = {"finished", "scheduled", "in_play", "paused", "postponed", "cancelled", "suspended"}
|
|
|
|
STATUS_MAP = {
|
|
"finished": "finished", "completed": "finished", "done": "finished", "awarded": "finished",
|
|
"scheduled": "scheduled", "upcoming": "scheduled",
|
|
"in_play": "in_play", "live": "in_play",
|
|
"paused": "paused", "postponed": "postponed",
|
|
"cancelled": "cancelled", "canceled": "cancelled", "abandoned": "cancelled",
|
|
"suspended": "suspended",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class NormalizedMatch:
|
|
"""清洗后的统一比赛记录(入库中间格式)。"""
|
|
|
|
league_type: str
|
|
date: datetime
|
|
home_team: str
|
|
away_team: str
|
|
match_status: str = "finished"
|
|
home_goals: int | None = None
|
|
away_goals: int | None = None
|
|
season_label: str = ""
|
|
home_xg: float | None = None
|
|
away_xg: float | None = None
|
|
home_shots: int | None = None
|
|
away_shots: int | None = None
|
|
home_shots_on_target: int | None = None
|
|
away_shots_on_target: int | None = None
|
|
home_corners: int | None = None
|
|
away_corners: int | None = None
|
|
home_possession: float | None = None
|
|
home_yellow_cards: int | None = None
|
|
away_yellow_cards: int | None = None
|
|
home_red_cards: int | None = None
|
|
away_red_cards: int | None = None
|
|
home_ht_goals: int | None = None
|
|
away_ht_goals: int | None = None
|
|
match_stage: str | None = None
|
|
|
|
def validate(self) -> None:
|
|
"""完整数据契约校验。"""
|
|
if self.match_status == "finished" and (self.home_goals is None or self.away_goals is None):
|
|
raise ValueError(f"Finished match must have score: {self.home_team} vs {self.away_team}")
|
|
|
|
def _finite(n, v):
|
|
if v is not None and isinstance(v, float) and not math.isfinite(v):
|
|
raise ValueError(f"{n} must be finite, got {v}")
|
|
|
|
def _range(n, v, lo, hi):
|
|
if v is not None and not (lo <= v <= hi):
|
|
raise ValueError(f"{n} out of range [{lo}, {hi}]: {v}")
|
|
|
|
for side in ("home", "away"):
|
|
_finite(f"{side}_goals", getattr(self, f"{side}_goals"))
|
|
_range(f"{side}_goals", getattr(self, f"{side}_goals"), 0, 30)
|
|
_finite(f"{side}_xg", getattr(self, f"{side}_xg"))
|
|
_range(f"{side}_xg", getattr(self, f"{side}_xg"), 0, 20)
|
|
for fld in ("shots", "shots_on_target", "corners"):
|
|
_range(f"{side}_{fld}", getattr(self, f"{side}_{fld}"), 0, 100)
|
|
for fld in ("yellow_cards", "red_cards"):
|
|
_range(f"{side}_{fld}", getattr(self, f"{side}_{fld}"), 0, 20)
|
|
_range("home_possession", self.home_possession, 0, 100)
|
|
if self.home_ht_goals is not None and self.home_goals is not None and self.home_ht_goals > self.home_goals:
|
|
raise ValueError(f"home_ht_goals({self.home_ht_goals}) > home_goals({self.home_goals})")
|
|
if self.away_ht_goals is not None and self.away_goals is not None and self.away_ht_goals > self.away_goals:
|
|
raise ValueError(f"away_ht_goals({self.away_ht_goals}) > away_goals({self.away_goals})")
|
|
|
|
|
|
def derive_season_label(date: datetime) -> str:
|
|
y = date.year
|
|
return f"{y}-{y + 1}" if date.month >= 8 else f"{y - 1}-{y}"
|
|
|
|
|
|
def _parse_date(value) -> datetime | None:
|
|
"""日期解析 → UTC datetime(带 tzinfo)。
|
|
|
|
P2-2: 解析失败时记录 warning,避免静默丢数据而无感知。
|
|
"""
|
|
if value in (None, ""):
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return datetime.fromtimestamp(value, tz=timezone.utc)
|
|
s = str(value).strip()
|
|
if not s:
|
|
return None
|
|
iso_s = s[:-1] + "+00:00" if s.endswith("Z") else s
|
|
try:
|
|
dt = datetime.fromisoformat(iso_s)
|
|
if dt.tzinfo is not None:
|
|
return dt.astimezone(timezone.utc)
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
pass
|
|
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d", "%d/%m/%Y", "%d/%m/%y"):
|
|
try:
|
|
return datetime.strptime(s[:19], fmt).replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
continue
|
|
# P2-2 修复: 记录被丢弃的原始值,便于排查数据源格式变更
|
|
logger.warning("_parse_date failed, dropping record: %r", value)
|
|
return None
|
|
|
|
|
|
def _to_int(v) -> int | None:
|
|
if v is None or (isinstance(v, str) and v.strip() in ("", "-")):
|
|
return None
|
|
if isinstance(v, bool):
|
|
return None
|
|
if isinstance(v, int):
|
|
return v
|
|
try:
|
|
f = float(str(v).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if not f.is_integer():
|
|
return None
|
|
return int(f)
|
|
|
|
|
|
def _to_float(v) -> float | None:
|
|
if v is None or (isinstance(v, str) and v.strip() in ("", "-")):
|
|
return None
|
|
try:
|
|
return float(str(v).strip())
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def normalize_bzzoiro(raw: dict, league_type: str) -> NormalizedMatch | None:
|
|
"""bzzoiro event → NormalizedMatch。"""
|
|
from src.data.team_names import normalize as normalize_name
|
|
|
|
date = _parse_date(raw.get("event_date"))
|
|
if date is None:
|
|
return None
|
|
raw_status = str(raw.get("status", "")).lower()
|
|
status = STATUS_MAP.get(raw_status)
|
|
if status is None:
|
|
return None
|
|
home = normalize_name(raw.get("home_team", ""))
|
|
away = normalize_name(raw.get("away_team", ""))
|
|
if not home or not away or home == away:
|
|
return None
|
|
m = NormalizedMatch(
|
|
league_type=league_type,
|
|
date=date,
|
|
home_team=home,
|
|
away_team=away,
|
|
match_status=status,
|
|
season_label=derive_season_label(date),
|
|
)
|
|
m.home_goals = _to_int(raw.get("home_score", raw.get("home_goals")))
|
|
m.away_goals = _to_int(raw.get("away_score", raw.get("away_goals")))
|
|
m.home_ht_goals = _to_int(raw.get("home_score_ht", raw.get("home_ht_goals")))
|
|
m.away_ht_goals = _to_int(raw.get("away_score_ht", raw.get("away_ht_goals")))
|
|
_rn = _to_int(raw.get("round_number"))
|
|
_rn_name = str(raw.get("round_name") or "").strip()
|
|
if _rn_name:
|
|
m.match_stage = _rn_name
|
|
elif _rn:
|
|
m.match_stage = f"第 {_rn} 轮"
|
|
if m.match_status == "finished" and m.home_goals is None:
|
|
m.match_status = "scheduled"
|
|
return m
|
|
|
|
|
|
def normalize_understat(raw: dict, league_type: str) -> NormalizedMatch | None:
|
|
"""understat 单场 → NormalizedMatch(仅 xG)。"""
|
|
from src.data.team_names import normalize as normalize_name
|
|
|
|
dt_str = raw.get("datetime") or raw.get("date")
|
|
if not dt_str:
|
|
return None
|
|
dt = _parse_date(dt_str)
|
|
if dt is None:
|
|
return None
|
|
home_info = raw.get("h", {})
|
|
away_info = raw.get("a", {})
|
|
home_name = home_info.get("title", "") if isinstance(home_info, dict) else ""
|
|
away_name = away_info.get("title", "") if isinstance(away_info, dict) else ""
|
|
home = normalize_name(home_name)
|
|
away = normalize_name(away_name)
|
|
if not home or not away or home == away:
|
|
return None
|
|
home_xg = raw.get("xG", {}).get("h") if isinstance(raw.get("xG"), dict) else None
|
|
away_xg = raw.get("xG", {}).get("a") if isinstance(raw.get("xG"), dict) else None
|
|
return NormalizedMatch(
|
|
league_type=league_type,
|
|
date=dt,
|
|
home_team=home,
|
|
away_team=away,
|
|
match_status="finished",
|
|
season_label=derive_season_label(dt),
|
|
home_xg=home_xg,
|
|
away_xg=away_xg,
|
|
)
|