shangfangjian
53e602b4f6
- unit_of_work.py: 简化为纯 session 上下文管理器,消除 double-close 风险 - validation.py: 移除未使用的 import,简化 _score_to_1x2 逻辑 - repositories.py: 将 func 导入移到模块顶层 - 更新所有 get_uow() 调用点使用新接口(yield session 而非 uow 对象)
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
"""评估:赛后回填 + 统计。"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from sqlalchemy import select
|
||
|
||
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 get_uow() as session:
|
||
pred = await 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
|
||
return pred
|
||
|
||
|
||
def _actual_1x2(home: int, away: int) -> str:
|
||
"""根据实际比分返胜平负。"""
|
||
if home > away:
|
||
return "1"
|
||
if home == away:
|
||
return "X"
|
||
return "2"
|
||
|
||
|
||
async def get_eval_summary() -> dict:
|
||
"""按 provider × 模型聚合评估。"""
|
||
async with get_uow() as session:
|
||
stmt = (
|
||
select(Prediction)
|
||
.where(Prediction.settled == True)
|
||
)
|
||
result = await session.execute(stmt)
|
||
rows = list(result.scalars().all())
|
||
|
||
from collections import defaultdict
|
||
buckets: dict[tuple[str, str], dict] = defaultdict(lambda: {
|
||
"total": 0, "correct_1x2": 0, "score_errors": [], "conf_sum": 0.0, "conf_count": 0,
|
||
})
|
||
for p in rows:
|
||
key = (p.provider, p.model)
|
||
b = buckets[key]
|
||
b["total"] += 1
|
||
if p.actual_home_goals is None or p.actual_away_goals is None:
|
||
continue
|
||
actual = _actual_1x2(p.actual_home_goals, p.actual_away_goals)
|
||
if p.pred_1x2 == actual:
|
||
b["correct_1x2"] += 1
|
||
if p.pred_home_goals is not None and p.pred_away_goals is not None:
|
||
err = ((p.pred_home_goals - p.actual_home_goals) ** 2 +
|
||
(p.pred_away_goals - p.actual_away_goals) ** 2) ** 0.5
|
||
b["score_errors"].append(err)
|
||
if p.subjective_confidence is not None:
|
||
b["conf_sum"] += p.subjective_confidence
|
||
b["conf_count"] += 1
|
||
|
||
summary = []
|
||
for (prov, model), b in sorted(buckets.items()):
|
||
acc = (b["correct_1x2"] / b["total"] * 100) if b["total"] else 0
|
||
avg_err = (sum(b["score_errors"]) / len(b["score_errors"])) if b["score_errors"] else None
|
||
avg_conf = (b["conf_sum"] / b["conf_count"]) if b["conf_count"] else None
|
||
summary.append({
|
||
"provider": prov,
|
||
"model": model,
|
||
"total": b["total"],
|
||
"accuracy_1x2": round(acc, 1),
|
||
"avg_score_rmse": round(avg_err, 2) if avg_err is not None else None,
|
||
"avg_subjective_confidence": round(avg_conf, 2) if avg_conf is not None else None,
|
||
})
|
||
return {"summary": summary}
|