4 Commits
Author SHA1 Message Date
WorkBuddy f965650c10 test+docs: P0 回归测试与文档同步
回归测试(P0 三项,不依赖 DB 的静态断言):
- tests/test_regressions.py:
  · context_builder 三个查询函数 + backtest 必须 eager-load 关系
  · Match.{stats,home_team,away_team,league} 必须 lazy="selectin"
  · bzzoiro normalized_matches 必须携带并解包 (nm, raw) 配对
  这些是「集成方式」缺陷,原先的单元测试(全 mock slice_fn/provider)
  照不出来,故此专项静态锁定。

文档同步(修正与实际不符的表述):
- docs/01-architecture.md: app.py 标注改为「lifespan 仅验证连接,不建表」;
  补 deps.py / repositories.py / unit_of_work.py / validation.py /
  backtest.py / sources.py / http_client.py / retry.py;迁移范围改 0001~0006
- docs/07-development.md: 目录树修正 leagues.py→matches.py,补齐
  backtest.py / deps.py 及 0003~0006 迁移文件

附带: src/core/retry.py 加 NOTE 说明其当前无调用点(P3),
避免「看似有重试实则未生效」的误判。
2026-09-15 16:55:54 +08:00
WorkBuddy c89bfe2af7 fix(P2): no_data 结构化、权重校验、鉴权与前端竞态
P2-1 no_data 门控依赖文案子串(脆弱):
- context_builder 新增 SliceResult(text/has_data/n_records),
  5 个切片函数改为显式声明 has_data
- base._slice_has_data() 优先取结构化结果,str 返回仍走文案回退
  (兼容既有测试 mock 与自定义切片)
- build_context 的 has_stats/has_injuries 直接取切片声明

P2-2 agent_weights 无校验即落库:
- validation 新增 AgentWeightsSchema / validate_agent_weights:
  未知专家名丢弃、越界值钳制、总和非 1 时归一化
- orchestrator 落库前对 agent_weights 做校验

P2-3 1x2 与比分不一致被静默修正:
- 仍以比分修正,但补 logger.warning 暴露 LLM 自相矛盾

P2-5/P2-6 prompt 缓存不可刷新 + 缓存键不含模板内容:
- 新增 clear_prompt_cache() 供改模板后显式失效
- 缓存键纳入模板内容 hash,模板一改缓存自动失效

P2-7 ingest/backtest/settle 接口无鉴权:
- 新增 require_admin_key 依赖(X-API-Key),
  ADMIN_API_KEY 未设置时放行并告警(不破坏本地开发)
- 挂到 3 个 ingest 接口 + backtest + eval/settle

P2-8 前端请求竞态 + 未使用游标分页:
- Matches.tsx 用递增 seq 丢弃过期响应,避免旧筛选结果覆盖新筛选
- 接入后端已有的 cursor 分页 + 「加载更多」按钮

附带: .env.example 补齐 LLM_TIMEOUT / 分档模型 / ADMIN_API_KEY;
tests 新增 10 个用例覆盖 P2-1/2/3。
2026-09-15 16:54:05 +08:00
WorkBuddy 71bf723a10 fix(P1): 消除迁移漂移、回测缓存污染与历史数据越界风险
P1-1 schema/ORM 漂移:
- 新增 0006 迁移: 回填后 drop 幽灵列 predictions.cutoff_at,
  并幂等重建 ix_match_stats_available_at / ix_predictions_cutoff_at
- models.py 为 MatchStats / Prediction 声明上述两个索引,
  使 autogenerate 不再误建议删除

P1-2 迁移对历史越界数据不安全:
- 0004 在建 CHECK 约束前先 op.execute 清洗越界行
  (负数进球/置信度越界/非法 1x2/mode),避免 ALTER 中途失败
  导致迁移卡在半完成状态

P1-3 回测缓存污染:
- predict_match / _predict_single 新增 use_cache 参数,
  use_cache=False 时既不读也不写进程内缓存
- run_backtest 显式传 use_cache=False,避免 settle 到旧记录

P1-5 understat 关系属性访问:
- MatchRepository.find_by_teams_and_date 加 selectinload(Match.stats)

P1-6 日期键构造不统一:
- bzzoiro 抽取 _to_date()/_match_key() 统一去重键构造,
  消除 str(date) 与 isoformat() 的隐式格式依赖
2026-09-15 16:48:03 +08:00
WorkBuddy 235fb0de97 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)
2026-09-15 16:45:09 +08:00
23 changed files with 817 additions and 118 deletions
+10
View File
@@ -10,6 +10,11 @@ LLM_PROVIDER=openai
LLM_API_KEY=sk-xxxx LLM_API_KEY=sk-xxxx
LLM_BASE_URL=https://api.openai.com/v1 LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o LLM_MODEL=gpt-4o
# 多 Agent 分档模型(留空则回落 LLM_MODEL)
LLM_SPECIALIST_MODEL=
LLM_AGGREGATOR_MODEL=
# 单次 LLM 调用超时(秒)
LLM_TIMEOUT=60
# ---- 数据源 ---- # ---- 数据源 ----
BZZOIRO_KEY= BZZOIRO_KEY=
@@ -17,3 +22,8 @@ API_FOOTBALL_KEY=
# ---- CORS ---- # ---- CORS ----
CORS_ORIGINS=http://localhost:5173,http://localhost:3000 CORS_ORIGINS=http://localhost:5173,http://localhost:3000
# ---- 管理接口鉴权 ----
# 采集/回测/回填接口的访问密钥(请求头 X-API-Key)。
# 留空 = 不启用鉴权(本地开发默认);生产环境必须设置强随机值。
ADMIN_API_KEY=
@@ -26,6 +26,21 @@ def upgrade() -> None:
op.add_column('predictions', sa.Column('input_hash', sa.String(length=64), nullable=True)) op.add_column('predictions', sa.Column('input_hash', sa.String(length=64), nullable=True))
# 3. 新增 CHECK 约束 # 3. 新增 CHECK 约束
# 注意: 对已有数据行加 CHECK 约束时,若存在越界数据 ALTER TABLE 会中途失败,
# 导致迁移卡在半完成状态。这里先做一次性清洗(把越界值收敛到合法域),
# 再建约束,保证在非空库上也能成功。
op.execute("UPDATE predictions SET pred_home_goals = 0 WHERE pred_home_goals IS NOT NULL AND pred_home_goals < 0")
op.execute("UPDATE predictions SET pred_away_goals = 0 WHERE pred_away_goals IS NOT NULL AND pred_away_goals < 0")
op.execute(
"UPDATE predictions SET subjective_confidence = "
"CASE WHEN subjective_confidence < 0 THEN 0 "
" WHEN subjective_confidence > 1 THEN 1 ELSE subjective_confidence END "
"WHERE subjective_confidence IS NOT NULL "
" AND (subjective_confidence < 0 OR subjective_confidence > 1)"
)
op.execute("UPDATE predictions SET pred_1x2 = NULL WHERE pred_1x2 IS NOT NULL AND pred_1x2 NOT IN ('1', 'X', '2')")
op.execute("UPDATE predictions SET mode = 'single' WHERE mode IS NOT NULL AND mode NOT IN ('single', 'multi')")
op.create_check_constraint('ck_pred_home_goals_nonneg', 'predictions', 'pred_home_goals >= 0') op.create_check_constraint('ck_pred_home_goals_nonneg', 'predictions', 'pred_home_goals >= 0')
op.create_check_constraint('ck_pred_away_goals_nonneg', 'predictions', 'pred_away_goals >= 0') op.create_check_constraint('ck_pred_away_goals_nonneg', 'predictions', 'pred_away_goals >= 0')
op.create_check_constraint('ck_confidence_range', 'predictions', 'subjective_confidence >= 0 AND subjective_confidence <= 1') op.create_check_constraint('ck_confidence_range', 'predictions', 'subjective_confidence >= 0 AND subjective_confidence <= 1')
@@ -0,0 +1,80 @@
"""清理 schema 与 ORM 模型的漂移
Revision ID: 0006_schema_model_drift_cleanup
Revises: 0005_prediction_status_and_stats_provenance
Create Date: 2026-09-15
背景(见代码审查报告 P1-1):
0005 迁移在 predictions 表留下了 `cutoff_at` 列(删除语句被注释掉),
但 ORM 模型 `Prediction` 中并无该字段 —— 这是一处会长期存在的 schema 漂移,
而 alembic autogenerate 会持续建议 drop 它,造成噪声。
同时 0005 创建的两个索引 `ix_match_stats_available_at` 与
`ix_predictions_cutoff_at` 未在 ORM 声明,autogenerate 会建议删除它们 ——
一旦被误删,数据血缘相关的时间过滤查询会退化为全表扫描。
本迁移做两件事:
1. drop 掉幽灵列 predictions.cutoff_at(数据已由 prediction_cutoff_at 承载,
迁移前先把非空值回填过去,避免丢数据)
2. 显式重建这两个索引(幂等:先 drop if exists 再 create),使 DB 状态与
修正后的 ORM 声明一致
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '0006_schema_model_drift_cleanup'
down_revision: Union[str, None] = '0005_prediction_status_and_stats_provenance'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
# --- 1. 幽灵列 cutoff_at: 先把数据回填到 prediction_cutoff_at 再删除 ---
pred_cols = {c["name"] for c in inspector.get_columns("predictions")}
if "cutoff_at" in pred_cols:
if "prediction_cutoff_at" in pred_cols:
# 仅回填尚未有值的行,避免覆盖更权威的数据
op.execute(
"UPDATE predictions "
"SET prediction_cutoff_at = cutoff_at "
"WHERE prediction_cutoff_at IS NULL AND cutoff_at IS NOT NULL"
)
op.drop_column("predictions", "cutoff_at")
# --- 2. 与 ORM 声明对齐的索引(幂等重建) ---
stats_idx = {i["name"] for i in inspector.get_indexes("match_stats")}
if "ix_match_stats_available_at" not in stats_idx:
op.create_index("ix_match_stats_available_at", "match_stats", ["available_at"])
pred_idx = {i["name"] for i in inspector.get_indexes("predictions")}
if "ix_predictions_cutoff_at" not in pred_idx:
op.create_index(
"ix_predictions_cutoff_at", "predictions", ["prediction_cutoff_at"]
)
def downgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
# 恢复幽灵列(与 0005 的最终状态一致:列存在但无值)
pred_cols = {c["name"] for c in inspector.get_columns("predictions")}
if "cutoff_at" not in pred_cols:
op.add_column(
"predictions",
sa.Column("cutoff_at", sa.DateTime(timezone=True), nullable=True),
)
stats_idx = {i["name"] for i in inspector.get_indexes("match_stats")}
if "ix_match_stats_available_at" in stats_idx:
op.drop_index("ix_match_stats_available_at", table_name="match_stats")
pred_idx = {i["name"] for i in inspector.get_indexes("predictions")}
if "ix_predictions_cutoff_at" in pred_idx:
op.drop_index("ix_predictions_cutoff_at", table_name="predictions")
+18 -8
View File
@@ -57,38 +57,48 @@
Profeto/ Profeto/
├── src/ ├── src/
│ ├── api/ │ ├── api/
│ │ ├── app.py # FastAPI 工厂(lifespan 建表) │ │ ├── app.py # FastAPI 工厂(lifespan 仅验证 DB 连接,不建表)
│ │ ├── deps.py # 依赖:管理接口鉴权(X-API-Key)
│ │ ├── schemas.py # Pydantic v2 请求/响应 │ │ ├── schemas.py # Pydantic v2 请求/响应
│ │ └── routes/ │ │ └── routes/
│ │ ├── matches.py # 联赛/比赛查询(游标分页) │ │ ├── matches.py # 联赛/比赛查询(游标分页)
│ │ ├── predict.py # 预测 + 预测历史 │ │ ├── predict.py # 预测 + 预测历史
│ │ ├── ingest.py # 采集触发(自管 session) │ │ ├── ingest.py # 采集触发(自管 session,需鉴权)
│ │ ── eval.py # 赛后回填 + 准确率汇总 │ │ ── eval.py # 赛后回填 + 准确率汇总
│ │ └── backtest.py # 历史回测(需鉴权)
│ ├── db/ │ ├── db/
│ │ ├── base.py # async engine + get_db/get_db_read │ │ ├── base.py # async engine + get_db/get_db_read
│ │ ── models.py # 6 张表 ORM │ │ ── models.py # 6 张表 ORM
│ │ ├── repositories.py # 仓储层
│ │ └── unit_of_work.py # 事务边界
│ ├── data/ │ ├── data/
│ │ ├── bzzoiro.py # 赛果采集 + 幂等入库 │ │ ├── bzzoiro.py # 赛果采集 + 幂等入库
│ │ ├── understat.py # xG 回填 │ │ ├── understat.py # xG 回填
│ │ ├── injuries.py # 伤停采集(带文件缓存) │ │ ├── injuries.py # 伤停采集(带文件缓存)
│ │ ├── normalize.py # NormalizedMatch 清洗契约 │ │ ├── normalize.py # NormalizedMatch 清洗契约
│ │ ├── team_names.py # 队名归一映射 │ │ ├── team_names.py # 队名归一映射
│ │ ├── sources.py # 数据源注册表
│ │ └── config.py # 联赛代码映射 │ │ └── config.py # 联赛代码映射
│ ├── llm/ │ ├── llm/
│ │ ├── provider.py # OpenAI-compatible 抽象(共享连接池/JSON 兜底解析) │ │ ├── provider.py # OpenAI-compatible 抽象(共享连接池/JSON 兜底解析)
│ │ ├── context_builder.py # 数据切片(h2h/form/stats/home_away/injuries)+ 单 agent 拼接 │ │ ├── context_builder.py # 数据切片(h2h/form/stats/home_away/injuries)+ 单 agent 拼接
│ │ ├── predict.py # 预测入口(mode 分派 + 缓存) │ │ ├── predict.py # 预测入口(mode 分派 + 缓存)
│ │ ├── eval.py # 准确率统计 │ │ ├── eval.py # 准确率统计
│ │ ├── validation.py # LLM 输出严格校验(Pydantic)
│ │ ├── backtest.py # 回测执行
│ │ ├── agents/ │ │ ├── agents/
│ │ │ ├── base.py # AgentSpec / AgentReport / run_agent │ │ │ ├── base.py # AgentSpec / AgentReport / run_agent
│ │ │ └── orchestrator.py # 并行专家 → 终裁 → 存库 │ │ │ └── orchestrator.py # 并行专家 → 终裁 → 存库
│ │ └── prompts/ │ │ └── prompts/
│ │ ├── match_prediction_v1/v2.md # 单 agent 模板 │ │ ├── match_prediction_v1/v2.md # 单 agent 模板
│ │ └── agents/{h2h,form,stats,home_away,injuries,aggregator}_v1.md │ │ └── agents/{h2h,form,stats,home_away,injuries,aggregator}_v1.md
│ └── core/config.py # pydantic-settings │ └── core/
├── alembic/versions/ # 0001 建表 + 0002 agent 字段 │ ├── config.py # pydantic-settings
├── frontend/src/pages/Matches.tsx # 单页(预测面板 + 专家报告折叠区) │ ├── http_client.py # 共享 httpx 客户端
├── tests/ # 33 项(核心 13 + agent 20) │ └── retry.py # 重试工具
├── alembic/versions/ # 0001~0006(0001 建表 → 0006 漂移清理)
├── frontend/src/pages/Matches.tsx # 单页(预测面板 + 专家报告折叠区 + 游标分页)
├── tests/ # 核心 + agent 测试
├── docker-compose.yml # api + postgres 两容器 ├── docker-compose.yml # api + postgres 两容器
└── docs/ # 本文档 └── docs/ # 本文档
``` ```
+18 -5
View File
@@ -35,27 +35,34 @@ Profeto/
├── src/ # 后端源码 ├── src/ # 后端源码
│ ├── api/ │ ├── api/
│ │ ├── routes/ # FastAPI 路由 │ │ ├── routes/ # FastAPI 路由
│ │ │ ├── leagues.py # 联赛/比赛查询 │ │ │ ├── matches.py # 联赛/比赛查询
│ │ │ ├── predict.py # 预测入口 │ │ │ ├── predict.py # 预测入口
│ │ │ ├── ingest.py # 数据采集 │ │ │ ├── ingest.py # 数据采集
│ │ │ ── eval.py # 评估回填 │ │ │ ── eval.py # 评估回填
│ │ │ └── backtest.py # 历史回测
│ │ ├── deps.py # 依赖:管理接口鉴权(X-API-Key)
│ │ ├── schemas.py # Pydantic 模型 │ │ ├── schemas.py # Pydantic 模型
│ │ └── app.py # FastAPI 工厂 │ │ └── app.py # FastAPI 工厂
│ ├── db/ │ ├── db/
│ │ ├── base.py # SQLAlchemy async engine + session │ │ ├── base.py # SQLAlchemy async engine + session
│ │ ── models.py # 6 张表 ORM │ │ ── models.py # 6 张表 ORM
│ │ ├── repositories.py # 仓储层(查询封装)
│ │ └── unit_of_work.py # 事务边界
│ ├── data/ │ ├── data/
│ │ ├── bzzoiro.py # bzzoiro 采集 + 入库 │ │ ├── bzzoiro.py # bzzoiro 采集 + 入库
│ │ ├── understat.py # understat xG 回填 │ │ ├── understat.py # understat xG 回填
│ │ ├── injuries.py # 伤停采集 │ │ ├── injuries.py # 伤停采集
│ │ ├── normalize.py # 数据清洗契约 │ │ ├── normalize.py # 数据清洗契约
│ │ ├── team_names.py # 队名归一化映射 │ │ ├── team_names.py # 队名归一化映射
│ │ ├── sources.py # 数据源注册表
│ │ └── config.py # 联赛映射常量 │ │ └── config.py # 联赛映射常量
│ ├── llm/ │ ├── llm/
│ │ ├── provider.py # LLM 提供商抽象(OpenAI-compatible) │ │ ├── provider.py # LLM 提供商抽象(OpenAI-compatible)
│ │ ├── context_builder.py # 数据切片 + 拼接 │ │ ├── context_builder.py # 数据切片 + 拼接
│ │ ├── predict.py # 预测入口(单/多模式分派) │ │ ├── predict.py # 预测入口(单/多模式分派)
│ │ ├── eval.py # 评估统计 │ │ ├── eval.py # 评估统计
│ │ ├── validation.py # LLM 输出严格校验
│ │ ├── backtest.py # 回测执行
│ │ ├── agents/ │ │ ├── agents/
│ │ │ ├── base.py # AgentSpec + run_agent │ │ │ ├── base.py # AgentSpec + run_agent
│ │ │ └── orchestrator.py # 多 agent 编排 │ │ │ └── orchestrator.py # 多 agent 编排
@@ -69,12 +76,18 @@ Profeto/
│ │ ├── h2h_v1.md │ │ ├── h2h_v1.md
│ │ └── aggregator_v1.md │ │ └── aggregator_v1.md
│ └── core/ │ └── core/
── config.py # pydantic-settings 配置 ── config.py # pydantic-settings 配置
│ ├── http_client.py # 共享 httpx 客户端
│ └── retry.py # 重试工具
├── frontend/ # React 单页前端 ├── frontend/ # React 单页前端
├── alembic/ # 数据库迁移 ├── alembic/ # 数据库迁移
│ └── versions/ │ └── versions/
│ ├── 0001_initial.py │ ├── 0001_initial.py
── 0002_agent_outputs.py ── 0002_agent_outputs.py
│ ├── 0003_injuries.py
│ ├── 0004_snapshot_and_constraints.py
│ ├── 0005_prediction_status_and_stats_provenance.py
│ └── 0006_schema_model_drift_cleanup.py
├── tests/ # 测试 ├── tests/ # 测试
│ ├── test_core.py # 核心逻辑测试 │ ├── test_core.py # 核心逻辑测试
│ └── test_agents.py # 多 agent 测试 │ └── test_agents.py # 多 agent 测试
+54 -3
View File
@@ -1,4 +1,4 @@
import { useCallback, useEffect, useState } from 'react' import { useCallback, useEffect, useRef, useState } from 'react'
interface Match { interface Match {
id: number id: number
@@ -69,31 +69,69 @@ export default function Matches() {
const [league, setLeague] = useState('E0') const [league, setLeague] = useState('E0')
const [status, setStatus] = useState('scheduled') const [status, setStatus] = useState('scheduled')
const [matches, setMatches] = useState<Match[]>([]) const [matches, setMatches] = useState<Match[]>([])
const [nextCursor, setNextCursor] = useState<string | null>(null)
const [loadingMore, setLoadingMore] = useState(false)
const [loading, setLoading] = useState(false) const [loading, setLoading] = useState(false)
const [predictingId, setPredictingId] = useState<number | null>(null) const [predictingId, setPredictingId] = useState<number | null>(null)
const [prediction, setPrediction] = useState<Prediction | null>(null) const [prediction, setPrediction] = useState<Prediction | null>(null)
const [error, setError] = useState<string | null>(null) const [error, setError] = useState<string | null>(null)
const [mode, setMode] = useState<'single' | 'multi'>('multi') const [mode, setMode] = useState<'single' | 'multi'>('multi')
// 请求竞态防护:切换联赛/状态很快时,先发的慢请求可能后返回,
// 把旧结果覆盖到新筛选上。用递增序号只认最后一次请求的响应。
const loadSeq = useRef(0)
const predictSeq = useRef(0)
const load = useCallback(async () => { const load = useCallback(async () => {
const seq = ++loadSeq.current
setLoading(true) setLoading(true)
// 切换筛选时作废进行中的「加载更多」,避免其标志位卡住
setLoadingMore(false)
setError(null) setError(null)
try { try {
const params = new URLSearchParams({ league, status, limit: '50' }) const params = new URLSearchParams({ league, status, limit: '50' })
const res = await fetch(`/api/v1/matches?${params}`) const res = await fetch(`/api/v1/matches?${params}`)
if (seq !== loadSeq.current) return // 已有更新的请求,丢弃本次结果
if (!res.ok) throw new Error(`HTTP ${res.status}`) if (!res.ok) throw new Error(`HTTP ${res.status}`)
const data = await res.json() const data = await res.json()
if (seq !== loadSeq.current) return
setMatches(data.items) setMatches(data.items)
setNextCursor(data.next_cursor ?? null)
} catch (e) { } catch (e) {
if (seq !== loadSeq.current) return
setError(e instanceof Error ? e.message : String(e)) setError(e instanceof Error ? e.message : String(e))
} finally { } finally {
setLoading(false) if (seq === loadSeq.current) setLoading(false)
} }
}, [league, status]) }, [league, status])
// 加载下一页(游标分页)。后端已支持 cursor,前端此前未使用,
// 导致 limit=50 之后的数据永远看不到(见审查报告 P2-8)。
const loadMore = async () => {
if (!nextCursor || loadingMore) return
const seq = loadSeq.current // 不做自增:切换筛选会自增,这里只跟随当前序列
setLoadingMore(true)
try {
const params = new URLSearchParams({ league, status, limit: '50', cursor: nextCursor })
const res = await fetch(`/api/v1/matches?${params}`)
if (seq !== loadSeq.current) return
if (!res.ok) throw new Error(`HTTP ${res.status}`)
const data = await res.json()
if (seq !== loadSeq.current) return
setMatches(prev => [...prev, ...data.items])
setNextCursor(data.next_cursor ?? null)
} catch (e) {
if (seq !== loadSeq.current) return
setError(e instanceof Error ? e.message : String(e))
} finally {
if (seq === loadSeq.current) setLoadingMore(false)
}
}
useEffect(() => { load() }, [load]) useEffect(() => { load() }, [load])
const predict = async (matchId: number) => { const predict = async (matchId: number) => {
const seq = ++predictSeq.current
setPredictingId(matchId) setPredictingId(matchId)
setError(null) setError(null)
setPrediction(null) setPrediction(null)
@@ -103,16 +141,19 @@ export default function Matches() {
headers: { 'Content-Type': 'application/json' }, headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ match_id: matchId, mode }), body: JSON.stringify({ match_id: matchId, mode }),
}) })
if (seq !== predictSeq.current) return
if (!res.ok) { if (!res.ok) {
const t = await res.text() const t = await res.text()
throw new Error(`HTTP ${res.status}: ${t}`) throw new Error(`HTTP ${res.status}: ${t}`)
} }
const data = await res.json() const data = await res.json()
if (seq !== predictSeq.current) return
setPrediction(data) setPrediction(data)
} catch (e) { } catch (e) {
if (seq !== predictSeq.current) return
setError(e instanceof Error ? e.message : String(e)) setError(e instanceof Error ? e.message : String(e))
} finally { } finally {
setPredictingId(null) if (seq === predictSeq.current) setPredictingId(null)
} }
} }
@@ -211,6 +252,16 @@ export default function Matches() {
</table> </table>
</div> </div>
{/* 分页: 加载更多 */}
{nextCursor && (
<div className="flex justify-center">
<button onClick={loadMore} disabled={loadingMore}
className="bg-white border text-gray-700 text-sm px-6 py-2 rounded hover:bg-gray-50 disabled:opacity-50">
{loadingMore ? '加载中...' : '加载更多'}
</button>
</div>
)}
{/* 预测中指示 */} {/* 预测中指示 */}
{predictingId && ( {predictingId && (
<div className="bg-blue-50 border border-blue-200 text-blue-700 px-4 py-3 rounded text-sm flex items-center gap-2"> <div className="bg-blue-50 border border-blue-200 text-blue-700 px-4 py-3 rounded text-sm flex items-center gap-2">
+43
View File
@@ -0,0 +1,43 @@
"""API 依赖:鉴权等横切关注点。
审查报告 P2-7:ingest / backtest / settle 这类「写入型或高成本」接口此前
完全无鉴权 —— 任何能访问到服务的人都可触发采集、或直接烧掉 LLM 额度。
策略(渐进式,不破坏本地开发):
- `ADMIN_API_KEY` 未配置 → 直接放行,并打一次 warning。
这样本地 `docker compose up` 无需额外配置即可用。
- 已配置 → 必须带匹配的 `X-API-Key` 请求头,否则 401。
"""
from __future__ import annotations
import logging
import secrets
from fastapi import Header, HTTPException
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
return
if not x_api_key or not secrets.compare_digest(x_api_key, expected):
raise HTTPException(status_code=401, detail="无效或缺失的 X-API-Key")
+6 -2
View File
@@ -3,9 +3,10 @@ from __future__ import annotations
import logging import logging
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from src.api.deps import require_admin_key
from src.llm.backtest import run_backtest from src.llm.backtest import run_backtest
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -22,10 +23,13 @@ class BacktestRequest(BaseModel):
model: str | None = Field(None, description="指定模型 (空=默认)") model: str | None = Field(None, description="指定模型 (空=默认)")
@router.post("/backtest") @router.post("/backtest", dependencies=[Depends(require_admin_key)])
async def backtest(req: BacktestRequest): async def backtest(req: BacktestRequest):
"""对历史比赛运行回测。 """对历史比赛运行回测。
该接口会对每场已完赛比赛各发起一次 LLM 预测,成本高 —— 因此需要
`X-API-Key` 鉴权(见审查报告 P2-7)。
对每场已完赛比赛: 对每场已完赛比赛:
1. 用比赛之前的数据构建上下文 (防未来信息泄漏) 1. 用比赛之前的数据构建上下文 (防未来信息泄漏)
2. 调 LLM 预测 2. 调 LLM 预测
+2 -1
View File
@@ -5,6 +5,7 @@ import logging
from fastapi import APIRouter, Depends, HTTPException from fastapi import APIRouter, Depends, HTTPException
from src.api.deps import require_admin_key
from src.api.schemas import EvalSummaryOut, SettleRequest from src.api.schemas import EvalSummaryOut, SettleRequest
from src.db.base import AsyncSession, get_db, get_db_read from src.db.base import AsyncSession, get_db, get_db_read
from src.llm.eval import get_eval_summary, settle_prediction from src.llm.eval import get_eval_summary, settle_prediction
@@ -14,7 +15,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1", tags=["eval"]) router = APIRouter(prefix="/api/v1", tags=["eval"])
@router.post("/eval/settle") @router.post("/eval/settle", dependencies=[Depends(require_admin_key)])
async def settle(req: SettleRequest, db: AsyncSession = Depends(get_db)): async def settle(req: SettleRequest, db: AsyncSession = Depends(get_db)):
"""回填实际结果。""" """回填实际结果。"""
try: try:
+5 -4
View File
@@ -3,8 +3,9 @@ from __future__ import annotations
import logging import logging
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, Depends, HTTPException
from src.api.deps import require_admin_key
from src.api.schemas import IngestBzzoiroRequest, IngestResponse, IngestUnderstatRequest, IngestInjuriesRequest, IngestSimpleResponse from src.api.schemas import IngestBzzoiroRequest, IngestResponse, IngestUnderstatRequest, IngestInjuriesRequest, IngestSimpleResponse
from src.data.sources import get_source from src.data.sources import get_source
from src.data.injuries import ingest_injuries from src.data.injuries import ingest_injuries
@@ -15,7 +16,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1", tags=["ingest"]) router = APIRouter(prefix="/api/v1", tags=["ingest"])
@router.post("/ingest/bzzoiro", response_model=IngestResponse) @router.post("/ingest/bzzoiro", response_model=IngestResponse, dependencies=[Depends(require_admin_key)])
async def ingest_bzzoiro_route(req: IngestBzzoiroRequest): async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
"""触发 bzzoiro 采集。""" """触发 bzzoiro 采集。"""
source = get_source("bzzoiro") source = get_source("bzzoiro")
@@ -34,7 +35,7 @@ async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
raise HTTPException(500, "数据采集失败,请查看服务器日志") raise HTTPException(500, "数据采集失败,请查看服务器日志")
@router.post("/ingest/understat", response_model=IngestSimpleResponse) @router.post("/ingest/understat", response_model=IngestSimpleResponse, dependencies=[Depends(require_admin_key)])
async def ingest_understat_route(req: IngestUnderstatRequest): async def ingest_understat_route(req: IngestUnderstatRequest):
"""触发 understat xG 回填。""" """触发 understat xG 回填。"""
source = get_source("understat") source = get_source("understat")
@@ -47,7 +48,7 @@ async def ingest_understat_route(req: IngestUnderstatRequest):
raise HTTPException(500, "xG 回填失败,请查看服务器日志") raise HTTPException(500, "xG 回填失败,请查看服务器日志")
@router.post("/ingest/injuries", response_model=IngestSimpleResponse) @router.post("/ingest/injuries", response_model=IngestSimpleResponse, dependencies=[Depends(require_admin_key)])
async def ingest_injuries_route(req: IngestInjuriesRequest): async def ingest_injuries_route(req: IngestInjuriesRequest):
"""触发伤停采集。""" """触发伤停采集。"""
try: try:
+6
View File
@@ -33,5 +33,11 @@ class Settings(BaseSettings):
# --- CORS --- # --- CORS ---
CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000" CORS_ORIGINS: str = "http://localhost:5173,http://localhost:3000"
# --- 管理接口鉴权 ---
# 采集 / 回测等高成本或写入型接口需要此 Key(请求头 X-API-Key)。
# 留空表示「未启用鉴权」(本地开发默认),生产环境必须设置。
# 见审查报告 P2-7:ingest/backtest 无鉴权可被任意调用并烧掉 LLM 额度。
ADMIN_API_KEY: str = ""
settings = Settings() settings = Settings()
+8 -1
View File
@@ -1,4 +1,11 @@
"""重试工具:带指数退避的瞬态错误重试。""" """重试工具:带指数退避的瞬态错误重试。
NOTE(审查报告 P3):当前全项目**无调用点** —— bzzoiro 在 `_fetch_json_sync`
里自带了一套重试逻辑,understat/injuries 各自也有。这里保留是作为后续统一
重试策略的落点,但请勿误以为它已在生效。
如果决定不引入统一重试,建议删除本文件以避免"看起来有重试、实际没有"的误判。
"""
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
+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:
+21 -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__ = (
@@ -123,6 +133,11 @@ class MatchStats(Base):
match: Mapped[Match] = relationship(back_populates="stats") match: Mapped[Match] = relationship(back_populates="stats")
__table_args__ = (
# 数据血缘时间过滤查询用(按 available_at 取「赛前已可得」的统计)
Index("ix_match_stats_available_at", "available_at"),
)
class Injury(Base): class Injury(Base):
"""球员伤停记录(api-football 数据源)。""" """球员伤停记录(api-football 数据源)。"""
@@ -185,6 +200,8 @@ class Prediction(Base):
__table_args__ = ( __table_args__ = (
Index("ix_predictions_match", "match_id"), Index("ix_predictions_match", "match_id"),
Index("ix_predictions_provider_model", "provider", "model"), Index("ix_predictions_provider_model", "provider", "model"),
# 数据截止时间过滤查询用(按 prediction_cutoff_at 取「赛前已生成」的预测)
Index("ix_predictions_cutoff_at", "prediction_cutoff_at"),
# 数据库级约束:最后一道防线 # 数据库级约束:最后一道防线
CheckConstraint("pred_home_goals >= 0", name="ck_pred_home_goals_nonneg"), CheckConstraint("pred_home_goals >= 0", name="ck_pred_home_goals_nonneg"),
CheckConstraint("pred_away_goals >= 0", name="ck_pred_away_goals_nonneg"), CheckConstraint("pred_away_goals >= 0", name="ck_pred_away_goals_nonneg"),
+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)
+23 -5
View File
@@ -12,7 +12,7 @@ import logging
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from src.llm.context_builder import MatchHeader from src.llm.context_builder import MatchHeader, SliceResult
from src.llm.provider import LLMProvider, LLMResponse from src.llm.provider import LLMProvider, LLMResponse
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -80,7 +80,12 @@ class AgentReport:
def _is_no_data(slice_text: str) -> bool: def _is_no_data(slice_text: str) -> bool:
"""切片是否全无数据(除了标题行全是无数据)。""" """兜底: 判断纯字符串切片是否全无数据。
仅用于 slice_fn 返回 `str`(未升级为 SliceResult)的场景。
新代码应让切片返回 SliceResult 并显式声明 has_data —— 字符串子串匹配
依赖具体文案(「无比分数据」「无伤停数据」等变体会漏判),不可靠。
"""
body = [ln.strip() for ln in slice_text.splitlines() if ln.strip()] body = [ln.strip() for ln in slice_text.splitlines() if ln.strip()]
# 去掉标题行(── 开头) # 去掉标题行(── 开头)
content = [ln for ln in body if not ln.startswith("──")] content = [ln for ln in body if not ln.startswith("──")]
@@ -89,6 +94,18 @@ def _is_no_data(slice_text: str) -> bool:
return all(any(s in ln for s in NO_DATA_SENTINELS) for ln in content) return all(any(s in ln for s in NO_DATA_SENTINELS) for ln in content)
def _slice_has_data(slice_result) -> tuple[str, bool]:
"""把切片返回值统一成 (text, has_data)。
优先使用 SliceResult.has_data(结构化,可信);若切片函数仍返回 str,
则回退到文案子串匹配(向后兼容)。
"""
if isinstance(slice_result, SliceResult):
return slice_result.text, slice_result.has_data
text = str(slice_result)
return text, not _is_no_data(text)
def _stub_no_data(agent: str) -> AgentReport: def _stub_no_data(agent: str) -> AgentReport:
return AgentReport( return AgentReport(
agent=agent, agent=agent,
@@ -145,13 +162,14 @@ async def run_agent(
"""执行单个专家 agent: 切片 → no_data 门控 → 调 LLM → 解析报告。""" """执行单个专家 agent: 切片 → no_data 门控 → 调 LLM → 解析报告。"""
# 1. 数据切片 # 1. 数据切片
try: try:
slice_text = await spec.slice_fn(header, before=before) slice_result = await spec.slice_fn(header, before=before)
except Exception as e: except Exception as e:
logger.exception("agent %s slice failed", spec.name) logger.exception("agent %s slice failed", spec.name)
return AgentReport(agent=spec.name, status="error", analysis=f"数据切片失败: {e}") return AgentReport(agent=spec.name, status="error", analysis=f"数据切片失败: {e}")
# 2. no_data 门控: 切片无数据 → 不调 LLM # 2. no_data 门控: 切片显式声明无数据 → 不调 LLM
if _is_no_data(slice_text): slice_text, has_data = _slice_has_data(slice_result)
if not has_data:
logger.debug("agent %s: slice is no_data, skipping LLM", spec.name) logger.debug("agent %s: slice is no_data, skipping LLM", spec.name)
return _stub_no_data(spec.name) return _stub_no_data(spec.name)
+3 -2
View File
@@ -195,13 +195,14 @@ async def predict_match_multi(
raise ValueError(f"match {match_id} not found") raise ValueError(f"match {match_id} not found")
# 严格校验终裁输出 # 严格校验终裁输出
from src.llm.validation import validate_prediction_output from src.llm.validation import validate_agent_weights, validate_prediction_output
try: try:
validated = validate_prediction_output(final) validated = validate_prediction_output(final)
except Exception as e: except Exception as e:
raise RuntimeError(f"终裁输出校验失败: {e}") raise RuntimeError(f"终裁输出校验失败: {e}")
agent_weights = final.get("agent_weights") # agent_weights 同样必须过校验(旧实现直接取 raw 值落库,未做任何检查)
agent_weights = validate_agent_weights(final.get("agent_weights"))
pred = Prediction( pred = Prediction(
match_id=match_id, match_id=match_id,
provider=settings.LLM_PROVIDER, provider=settings.LLM_PROVIDER,
+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:
+82 -38
View File
@@ -39,6 +39,22 @@ def _is_stats_available(stats, before) -> bool:
return stats.available_at <= before return stats.available_at <= before
@dataclass
class SliceResult:
"""数据切片的显式结果(替代「靠文案子串猜有无数据」)。
旧实现用 `"无数据" in slice_text` 判断,依赖具体文案 —— 一旦某个切片
写成「无比分数据」「无伤停数据」这类变体,判断就会静默失配
(见审查报告 P2-1)。这里让切片函数直接声明 `has_data`,不再猜。
"""
text: str
has_data: bool
n_records: int = 0
def __str__(self) -> str: # 让老调用点可直接当 str 用
return self.text
@dataclass @dataclass
class MatchContext: class MatchContext:
match_id: int match_id: int
@@ -98,16 +114,18 @@ def header_text(h: MatchHeader) -> str:
# 切片函数: 每个领域 agent 一个 # 切片函数: 每个领域 agent 一个
# ============================================================ # ============================================================
async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None) -> str: async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None) -> SliceResult:
"""E - 历史交锋切片: 过去数年 + 近期交手数据,提取交手规律。before=match_date 用于回测。""" """E - 历史交锋切片: 过去数年 + 近期交手数据,提取交手规律。before=match_date 用于回测。"""
async with AsyncSessionLocal() as db: async with AsyncSessionLocal() as db:
h2h = await _get_h2h(db, header.home_team_id, header.away_team_id, before=before, limit=limit) h2h = await _get_h2h(db, header.home_team_id, header.away_team_id, before=before, limit=limit)
lines = [f"── 历史交锋(近 {limit} 次) ──"] lines = [f"── 历史交锋(近 {limit} 次) ──"]
n_with_score = 0
if h2h: if h2h:
home_wins = draws = away_wins = 0 home_wins = draws = away_wins = 0
for hm in h2h: for hm in h2h:
d = hm.match_date.strftime("%Y-%m") if hm.match_date else "?" d = hm.match_date.strftime("%Y-%m") if hm.match_date else "?"
if hm.home_goals is not None: if hm.home_goals is not None:
n_with_score += 1
if hm.home_goals > hm.away_goals: home_wins += 1 if hm.home_goals > hm.away_goals: home_wins += 1
elif hm.home_goals == hm.away_goals: draws += 1 elif hm.home_goals == hm.away_goals: draws += 1
else: away_wins += 1 else: away_wins += 1
@@ -119,15 +137,17 @@ async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None) -> str:
lines.append(f" 总计 {total} 场: 主队 {home_wins}{draws}{away_wins}") lines.append(f" 总计 {total} 场: 主队 {home_wins}{draws}{away_wins}")
else: else:
lines.append(" 无数据") lines.append(" 无数据")
return "\n".join(lines) # has_data 以「有比分的交锋」为准:仅有对阵无比分时不足以支撑分析
return SliceResult(text="\n".join(lines), has_data=n_with_score > 0, n_records=n_with_score)
async def form_slice(header: MatchHeader, *, limit: int = 5, before=None) -> str: async def form_slice(header: MatchHeader, *, limit: int = 5, before=None) -> SliceResult:
"""A - 近期状态切片: 两队近 N 场赛果、关键事件、走势判断。before=match_date 用于回测。""" """A - 近期状态切片: 两队近 N 场赛果、关键事件、走势判断。before=match_date 用于回测。"""
async with AsyncSessionLocal() as db: async with AsyncSessionLocal() as db:
home_form = await _get_form(db, header.home_team_id, before=before, limit=limit) home_form = await _get_form(db, header.home_team_id, before=before, limit=limit)
away_form = await _get_form(db, header.away_team_id, before=before, limit=limit) away_form = await _get_form(db, header.away_team_id, before=before, limit=limit)
lines = [] lines = []
n_scored = 0
for label, name, form, side in ( for label, name, form, side in (
("主队", header.home_name, home_form, "home"), ("主队", header.home_name, home_form, "home"),
("客队", header.away_name, away_form, "away"), ("客队", header.away_name, away_form, "away"),
@@ -140,6 +160,8 @@ async def form_slice(header: MatchHeader, *, limit: int = 5, before=None) -> str
if o == "W": wins += 1 if o == "W": wins += 1
elif o == "D": draws += 1 elif o == "D": draws += 1
else: losses += 1 else: losses += 1
if fm.home_goals is not None:
n_scored += 1
score = f"{fm.home_goals}-{fm.away_goals}" if fm.home_goals is not None else "vs" score = f"{fm.home_goals}-{fm.away_goals}" if fm.home_goals is not None else "vs"
xg = "" xg = ""
if fm.stats and _is_stats_available(fm.stats, before) and fm.stats.home_xg is not None: if fm.stats and _is_stats_available(fm.stats, before) and fm.stats.home_xg is not None:
@@ -150,15 +172,16 @@ async def form_slice(header: MatchHeader, *, limit: int = 5, before=None) -> str
lines.append(f"{len(form)} 场: {wins}{draws}{losses}") lines.append(f"{len(form)} 场: {wins}{draws}{losses}")
else: else:
lines.append(" 无数据") lines.append(" 无数据")
return "\n".join(lines) return SliceResult(text="\n".join(lines), has_data=n_scored > 0, n_records=n_scored)
async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> str: async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> SliceResult:
"""B - 攻防数据切片: 进球、射门、控球,评估攻防强度。before=match_date 用于回测。""" """B - 攻防数据切片: 进球、射门、控球,评估攻防强度。before=match_date 用于回测。"""
async with AsyncSessionLocal() as db: async with AsyncSessionLocal() as db:
home_form = await _get_form(db, header.home_team_id, before=before, limit=limit) home_form = await _get_form(db, header.home_team_id, before=before, limit=limit)
away_form = await _get_form(db, header.away_team_id, before=before, limit=limit) away_form = await _get_form(db, header.away_team_id, before=before, limit=limit)
lines = [f"── 攻防数据(近 {limit} 场) ──"] lines = [f"── 攻防数据(近 {limit} 场) ──"]
n_total = 0
for label, name, form, side in ( for label, name, form, side in (
("主队", header.home_name, home_form, "home"), ("主队", header.home_name, home_form, "home"),
("客队", header.away_name, away_form, "away"), ("客队", header.away_name, away_form, "away"),
@@ -184,6 +207,7 @@ async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> s
xg += fm.stats.home_xg if side == "home" else fm.stats.away_xg xg += fm.stats.home_xg if side == "home" else fm.stats.away_xg
xga += fm.stats.away_xg if side == "home" else fm.stats.home_xg xga += fm.stats.away_xg if side == "home" else fm.stats.home_xg
n_xg += 1 n_xg += 1
n_total += n
if n > 0: if n > 0:
lines.append(f" {label} {name}:") lines.append(f" {label} {name}:")
lines.append(f" 场均进球 {gf/n:.2f}, 场均失球 {ga/n:.2f}") lines.append(f" 场均进球 {gf/n:.2f}, 场均失球 {ga/n:.2f}")
@@ -194,15 +218,16 @@ async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None) -> s
lines.append(f" {label} {name}: 无比分数据") lines.append(f" {label} {name}: 无比分数据")
else: else:
lines.append(f" {label} {name}: 无数据") lines.append(f" {label} {name}: 无数据")
return "\n".join(lines) return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)
async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None) -> str: async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None) -> SliceResult:
"""C - 主客因素切片: 主场战绩 vs 客场战绩,评估地理优势影响。before=match_date 用于回测。""" """C - 主客因素切片: 主场战绩 vs 客场战绩,评估地理优势影响。before=match_date 用于回测。"""
async with AsyncSessionLocal() as db: async with AsyncSessionLocal() as db:
home_home = await _get_home_away(db, header.home_team_id, "home", before=before, limit=limit) home_home = await _get_home_away(db, header.home_team_id, "home", before=before, limit=limit)
away_away = await _get_home_away(db, header.away_team_id, "away", before=before, limit=limit) away_away = await _get_home_away(db, header.away_team_id, "away", before=before, limit=limit)
lines = ["── 主客因素 ──"] lines = ["── 主客因素 ──"]
n_total = 0
for label, name, matches, side in ( for label, name, matches, side in (
("主队主场", header.home_name, home_home, "home"), ("主队主场", header.home_name, home_home, "home"),
("客队客场", header.away_name, away_away, "away"), ("客队客场", header.away_name, away_away, "away"),
@@ -218,6 +243,7 @@ async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None)
gf += m.home_goals if side == "home" else m.away_goals gf += m.home_goals if side == "home" else m.away_goals
ga += m.away_goals if side == "home" else m.home_goals ga += m.away_goals if side == "home" else m.home_goals
n = wins + draws + losses n = wins + draws + losses
n_total += n
if n > 0: if n > 0:
pct = wins / n * 100 pct = wins / n * 100
lines.append(f" {label} {name}(近 {n} 场): {wins}{draws}{losses}负, 胜率 {pct:.0f}%") lines.append(f" {label} {name}(近 {n} 场): {wins}{draws}{losses}负, 胜率 {pct:.0f}%")
@@ -226,10 +252,10 @@ async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None)
lines.append(f" {label} {name}: 无比分数据") lines.append(f" {label} {name}: 无比分数据")
else: else:
lines.append(f" {label} {name}: 无数据") lines.append(f" {label} {name}: 无数据")
return "\n".join(lines) return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)
async def injuries_slice(header: MatchHeader, *, before=None) -> str: async def injuries_slice(header: MatchHeader, *, before=None) -> SliceResult:
"""D - 阵容完整性切片: 伤停与停赛名单,评估战力缺失程度。 """D - 阵容完整性切片: 伤停与停赛名单,评估战力缺失程度。
before=cutoff: 只使用 cutoff 之前已采集的伤停数据,防回测泄漏。 before=cutoff: 只使用 cutoff 之前已采集的伤停数据,防回测泄漏。
@@ -242,10 +268,10 @@ async def injuries_slice(header: MatchHeader, *, before=None) -> str:
away_injuries = await get_injuries_for_match(db, header.away_team_id, cutoff, as_of=cutoff) away_injuries = await get_injuries_for_match(db, header.away_team_id, cutoff, as_of=cutoff)
lines = ["── 阵容完整性 ──"] lines = ["── 阵容完整性 ──"]
has_data = False n_records = 0
for label, injuries in (("主队", home_injuries), ("客队", away_injuries)): for label, injuries in (("主队", home_injuries), ("客队", away_injuries)):
if injuries: if injuries:
has_data = True n_records += len(injuries)
lines.append(f" {label}伤停({len(injuries)}人):") lines.append(f" {label}伤停({len(injuries)}人):")
for inj in injuries[:8]: # 最多显示 8 条 for inj in injuries[:8]: # 最多显示 8 条
reason = inj.reason or inj.injury_type or "未知" reason = inj.reason or inj.injury_type or "未知"
@@ -255,10 +281,10 @@ async def injuries_slice(header: MatchHeader, *, before=None) -> str:
else: else:
lines.append(f" {label}: 无伤停数据") lines.append(f" {label}: 无伤停数据")
if not has_data: if n_records == 0:
return "── 阵容完整性 ──\n 无数据" return SliceResult(text="── 阵容完整性 ──\n 无数据", has_data=False, n_records=0)
return "\n".join(lines) return SliceResult(text="\n".join(lines), has_data=True, n_records=n_records)
# ============================================================ # ============================================================
@@ -266,42 +292,38 @@ async def injuries_slice(header: MatchHeader, *, before=None) -> str:
# ============================================================ # ============================================================
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) -> MatchContext:
"""单 agent 路径的完整上下文: 拼接全部切片(before=比赛时间,防未来信息)。""" """单 agent 路径的完整上下文: 拼接全部切片(before=比赛时间,防未来信息)。
has_stats / has_injuries 直接取切片显式声明的 has_data,
不再靠文案子串匹配(见审查报告 P2-1)。
"""
header = await load_match_header(match_id) header = await load_match_header(match_id)
parts = [header_text(header), ""] parts = [header_text(header), ""]
has_stats = False
has_injuries = False
form_text = await form_slice(header, limit=form_last, before=header.match_dt) form_res = await form_slice(header, limit=form_last, before=header.match_dt)
if "无数据" not in form_text: parts.append(form_res.text)
has_stats = True
parts.append(form_text)
parts.append("") parts.append("")
h2h_text = await h2h_slice(header, limit=h2h_last, before=header.match_dt) h2h_res = await h2h_slice(header, limit=h2h_last, before=header.match_dt)
parts.append(h2h_text) parts.append(h2h_res.text)
parts.append("") parts.append("")
stats_text = await stats_slice(header, before=header.match_dt) stats_res = await stats_slice(header, before=header.match_dt)
if "无数据" not in stats_text: parts.append(stats_res.text)
has_stats = True
parts.append(stats_text)
parts.append("") parts.append("")
home_away_text = await home_away_slice(header, before=header.match_dt) home_away_res = await home_away_slice(header, before=header.match_dt)
parts.append(home_away_text) parts.append(home_away_res.text)
parts.append("") parts.append("")
injuries_text = await injuries_slice(header, before=header.match_dt) injuries_res = await injuries_slice(header, before=header.match_dt)
if "无数据" not in injuries_text: parts.append(injuries_res.text)
has_injuries = True
parts.append(injuries_text)
return MatchContext( return MatchContext(
match_id=match_id, match_id=match_id,
text="\n".join(parts), text="\n".join(parts),
has_stats=has_stats, has_stats=form_res.has_data or stats_res.has_data,
has_injuries=has_injuries, has_injuries=injuries_res.has_data,
match_dt=header.match_dt, match_dt=header.match_dt,
) )
@@ -328,9 +350,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 +376,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 +399,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())
+48 -12
View File
@@ -27,12 +27,18 @@ _cache: dict[str, tuple[float, PredictResult]] = {}
_cache_lock = Lock() _cache_lock = Lock()
def _cache_key(match_id: int, provider: str, model: str, version: str) -> str: def _cache_key(match_id: int, provider: str, model: str, version: str, tpl_hash: str) -> str:
return f"{match_id}:{provider}:{model}:{version}" """缓存键:含 prompt 模板内容 hash。
仅用 version 做键不够 编辑器里改动 `match_prediction_v1.md` 而版本号
不变时,进程内缓存仍会返回旧模板产生的旧结果(见审查报告 P2-6)
把模板内容 hash 纳入键,模板一改缓存自动失效
"""
return f"{match_id}:{provider}:{model}:{version}:{tpl_hash[:12]}"
def _get_cached(match_id: int, provider: str, model: str, version: str) -> PredictResult | None: def _get_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str) -> PredictResult | None:
key = _cache_key(match_id, provider, model, version) key = _cache_key(match_id, provider, model, version, tpl_hash)
with _cache_lock: with _cache_lock:
if key in _cache: if key in _cache:
ts, result = _cache[key] ts, result = _cache[key]
@@ -42,12 +48,22 @@ def _get_cached(match_id: int, provider: str, model: str, version: str) -> Predi
return None return None
def _set_cached(match_id: int, provider: str, model: str, version: str, result: PredictResult) -> None: def _set_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str, result: PredictResult) -> None:
key = _cache_key(match_id, provider, model, version) key = _cache_key(match_id, provider, model, version, tpl_hash)
with _cache_lock: with _cache_lock:
_cache[key] = (time.time(), result) _cache[key] = (time.time(), result)
def clear_prompt_cache() -> None:
"""清空 prompt 模板缓存(供开发/热更新时手动调用)。
lru_cache 的模板缓存是进程级的,改完 .md 需要重启进程才能生效;
提供显式清理入口,避免"改了模板却看不到变化"的困惑(见审查报告 P2-5)
"""
_load_prompt_template.cache_clear()
logger.info("prompt 模板缓存已清空")
@functools.lru_cache(maxsize=8) @functools.lru_cache(maxsize=8)
def _load_prompt_template(version: str = "v1") -> str: def _load_prompt_template(version: str = "v1") -> str:
"""缓存 prompt 模板(进程生命周期内每个版本只读一次)。""" """缓存 prompt 模板(进程生命周期内每个版本只读一次)。"""
@@ -58,6 +74,11 @@ def _load_prompt_template(version: str = "v1") -> str:
return f.read() return f.read()
def _prompt_template_hash(version: str) -> str:
"""prompt 模板内容 hash(用于缓存键,模板变更即失效)。"""
return hashlib.sha256(_load_prompt_template(version).encode("utf-8")).hexdigest()
@dataclass @dataclass
class PredictResult: class PredictResult:
prediction_id: int prediction_id: int
@@ -81,11 +102,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 +130,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:
@@ -105,9 +138,11 @@ async def _predict_single(
if model: if model:
provider.model = model provider.model = model
version = prompt_version or "v1" version = prompt_version or "v1"
tpl_hash = _prompt_template_hash(version)
# 0. 查缓存(同 match+provider+model+version 5 分钟内直接返) # 0. 查缓存(同 match+provider+model+version+模板hash 5 分钟内直接返)
cached = _get_cached(match_id, settings.LLM_PROVIDER, provider.model, version) if use_cache:
cached = _get_cached(match_id, settings.LLM_PROVIDER, provider.model, version, tpl_hash)
if cached is not None: if cached is not None:
logger.debug("predict cache hit match=%s", match_id) logger.debug("predict cache hit match=%s", match_id)
return cached return cached
@@ -191,6 +226,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, tpl_hash, result)
return result return result
+69 -1
View File
@@ -10,6 +10,9 @@ from pydantic import BaseModel, Field, field_validator, model_validator
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 已知的 5 个专家 agent 名(与 orchestrator.SPECIALIST_SPECS 保持一致)
KNOWN_AGENT_NAMES: tuple[str, ...] = ("form", "stats", "home_away", "injuries", "h2h")
class AgentReportSchema(BaseModel): class AgentReportSchema(BaseModel):
"""单个专家 Agent 输出的校验 schema。""" """单个专家 Agent 输出的校验 schema。"""
@@ -64,9 +67,17 @@ class PredictionOutputSchema(BaseModel):
@model_validator(mode="after") @model_validator(mode="after")
def check_consistency(self) -> "PredictionOutputSchema": def check_consistency(self) -> "PredictionOutputSchema":
"""验证比分与胜平负一致,不一致则自动修正。""" """验证比分与胜平负一致
不一致时以比分为准修正 pred_1x2(比分是更结构化的输出),
**必须告警** 静默修正会掩盖 LLM 的自相矛盾,让问题无法被发现
"""
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(
"1x2 与比分不一致: 比分 %.1f-%.1f 推出 '%s',但 LLM 给出 '%s';以比分修正",
self.pred_home_goals, self.pred_away_goals, expected, self.pred_1x2,
)
self.pred_1x2 = expected self.pred_1x2 = expected
return self return self
@@ -80,6 +91,63 @@ def _score_to_1x2(home: float, away: float) -> str:
return "X" return "X"
class AgentWeightsSchema(BaseModel):
"""终裁给出的各专家权重校验 schema。
权重含义:各专家报告在最终决策中的相对影响力约束:
- key 必须是已知的 5 个专家名
- value [0, 1]
- 总和允许有 0.05 的浮点误差(LLM 常凑不到精确 1.0),
超出则归一化到 1.0 而不是直接拒收
"""
weights: dict[str, float] = Field(default_factory=dict)
@field_validator("weights", mode="before")
@classmethod
def coerce_weights(cls, v):
if v is None:
return {}
if not isinstance(v, dict):
raise ValueError(f"agent_weights 必须是 dict,得到 {type(v).__name__}")
out: dict[str, float] = {}
for k, raw in v.items():
key = str(k).strip().lower()
if key not in KNOWN_AGENT_NAMES:
logger.warning("agent_weights 含未知专家 '%s',已忽略", k)
continue
f = _safe_float(raw)
if f is None:
logger.warning("agent_weights['%s']=%r 非数值,已忽略", k, raw)
continue
# 负数直接钳到 0;超过 1 的钳到 1
out[key] = min(max(f, 0.0), 1.0)
return out
@model_validator(mode="after")
def normalize_sum(self) -> "AgentWeightsSchema":
"""权重和不为 1 时归一化(而非拒收),并在偏离较大时告警。"""
if not self.weights:
return self
total = sum(self.weights.values())
if total <= 0:
return self
if abs(total - 1.0) > 0.05:
logger.warning("agent_weights 总和为 %.3f,已归一化到 1.0", total)
self.weights = {k: v / total for k, v in self.weights.items()}
return self
def validate_agent_weights(raw) -> dict[str, float]:
"""校验并规范化终裁给出的 agent_weights。非法输入返回空 dict。"""
if raw is None:
return {}
try:
return AgentWeightsSchema(weights=raw).weights
except Exception as e:
logger.warning("agent_weights 校验失败,丢弃: %s", e)
return {}
def validate_agent_output(raw: dict) -> AgentReportSchema: def validate_agent_output(raw: dict) -> AgentReportSchema:
"""校验并规范化单个 Agent 输出。""" """校验并规范化单个 Agent 输出。"""
return AgentReportSchema( return AgentReportSchema(
+97
View File
@@ -219,3 +219,100 @@ class TestOrchestratorAggregation:
) )
assert "{{match_header}}" not in rendered assert "{{match_header}}" not in rendered
assert "{{agent_reports}}" not in rendered assert "{{agent_reports}}" not in rendered
class TestSliceResultGate:
"""P2-1: 结构化 has_data 门控(替代脆弱的文案子串匹配)。"""
def test_sliceresult_empty_is_no_data(self):
from src.llm.agents.base import _slice_has_data
from src.llm.context_builder import SliceResult
text, has = _slice_has_data(SliceResult(text="── x ──\n 无数据", has_data=False))
assert has is False
def test_sliceresult_with_data_beats_text(self):
"""即使文案里出现「无数据」字样,结构化 has_data=True 也应胜出。
这正是旧实现的漏洞:文案匹配会把主队: 无伤停数据 / 客队: 2人伤停
这类混合输出这里显式验证结构化声明优先
"""
from src.llm.agents.base import _slice_has_data
from src.llm.context_builder import SliceResult
tricky = SliceResult(
text="── 阵容完整性 ──\n 主队: 无伤停数据\n 客队伤停(1人):\n - X: 拉伤",
has_data=True,
)
_, has = _slice_has_data(tricky)
assert has is True
def test_str_fallback_still_works(self):
"""旧式 str 切片(测试 mock / 自定义切片)仍走文案回退,保持兼容。"""
from src.llm.agents.base import _slice_has_data
assert _slice_has_data("── 伤停 ──\n 无数据")[1] is False
assert _slice_has_data("── 交锋 ──\n A 2-1 B")[1] is True
def test_sliceresult_str_compat(self):
"""SliceResult 可当 str 用(老调用点无需改)。"""
from src.llm.context_builder import SliceResult
s = SliceResult(text="hello", has_data=True)
assert str(s) == "hello"
class TestAgentWeightsValidation:
"""P2-2: agent_weights 必须过校验才能落库。"""
def test_unknown_agent_dropped(self):
from src.llm.validation import validate_agent_weights
w = validate_agent_weights({"form": 0.4, "h2h": 0.4, "bogus": 0.2, "stats": 0.4})
assert "bogus" not in w
assert set(w) <= {"form", "stats", "home_away", "injuries", "h2h"}
def test_out_of_range_clamped(self):
from src.llm.validation import validate_agent_weights
w = validate_agent_weights({"form": 5.0, "h2h": -1.0})
assert w["form"] == 1.0
assert w["h2h"] == 0.0
def test_sum_normalized(self):
from src.llm.validation import validate_agent_weights
w = validate_agent_weights({"form": 2.0, "stats": 2.0})
assert abs(sum(w.values()) - 1.0) < 1e-9
def test_no_weights_returns_empty(self):
from src.llm.validation import validate_agent_weights
assert validate_agent_weights(None) == {}
assert validate_agent_weights("not a dict") == {}
class TestPredictionConsistencyWarn:
"""P2-3: 比分与 1x2 不一致 → 以比分修正(且告警)。"""
def test_mismatch_is_corrected_to_score(self):
from src.llm.validation import validate_prediction_output
v = validate_prediction_output({
"pred_home_goals": 2.0,
"pred_away_goals": 1.0,
"pred_1x2": "X", # 与 2-1 矛盾
"subjective_confidence": 0.7,
})
assert v.pred_1x2 == "1" # 按比分修正
def test_consistent_passes_through(self):
from src.llm.validation import validate_prediction_output
v = validate_prediction_output({
"pred_home_goals": 0.0,
"pred_away_goals": 0.0,
"pred_1x2": "X",
"subjective_confidence": 0.5,
})
assert v.pred_1x2 == "X"
+99
View File
@@ -0,0 +1,99 @@
"""回归测试:锁定 P0 三项「静默失效」缺陷不再复发。
这些用例不依赖数据库 它们用静态分析检查代码结构,
因为三个 P0 的本质都是集成方式错误, mock slice_fn /
provider 的单元测试里永远照不出来(这正是它们当初漏网的原因)
- P0-1/P0-2: 查询 Match 的函数必须 eager-load 切片会访问的关系
- P0-3: bzzoiro source_event_id 时用的 raw 必须与 nm 配对
"""
from __future__ import annotations
import re
from pathlib import Path
SRC = Path(__file__).resolve().parent.parent / "src"
# 切片函数会读取的关系属性 → 查询时必须 eager-load
MATCH_RELATIONS = ("stats", "home_team", "away_team", "league")
def _read(rel: str) -> str:
return (SRC / rel).read_text(encoding="utf-8")
class TestEagerLoadCoverage:
"""P0-1 / P0-2: 凡是 select(Match) 且后续访问关系的函数,必须有 eager-load。"""
def test_context_builder_getters_eager_load(self):
"""_get_form / _get_h2h / _get_home_away 必须预加载关系。
models.py 已声明 lazy="selectin" 兜底,但这里同时检查显式
selectinload 显式声明是查询意图的固化,也被 P0 修复所依赖
"""
src = _read("llm/context_builder.py")
for fn in ("_get_form", "_get_h2h", "_get_home_away"):
# 截取函数体
m = re.search(rf"async def {fn}\(.*?(?=\nasync def |\n# =|\Z)", src, re.S)
assert m, f"{fn} 未找到"
body = m.group(0)
assert "selectinload" in body, (
f"{fn} 查询 Match 但未 eager-load 关系 —— "
"this would raise MissingGreenlet in async SQLAlchemy (P0-2)"
)
def test_backtest_candidates_eager_load(self):
"""回测取历史比赛必须 eager-load(否则 session 关闭后访问关系必炸)。"""
src = _read("llm/backtest.py")
assert "selectinload" in src, "backtest 未 eager-load 关系 (P0-1)"
def test_relationship_default_is_selectin(self):
"""models.py 中 Match 的高频关系应声明 lazy='selectin' 作为兜底。"""
src = _read("db/models.py")
# 找到 Match 类定义段
m = re.search(r"class Match\(Base\):.*?(?=\nclass )", src, re.S)
assert m, "Match 类未找到"
body = m.group(0)
for rel in MATCH_RELATIONS:
# 关系声明可能跨多行(stats/home_team/away_team 都是),因此按
# 「从 `rel: Mapped` 到下一个 `xxx: Mapped` 之前」整段匹配。
m_rel = re.search(
rf"^\s*{rel}: Mapped.*?(?=^\s*\w+: Mapped|\Z)", body, re.M | re.S
)
assert m_rel, f"Match.{rel} 未找到"
assert 'lazy="selectin"' in m_rel.group(0), (
f"Match.{rel} 未声明 lazy='selectin' —— 兜底缺失 (P0-2)"
)
class TestBzzoiroLineage:
"""P0-3: source_event_id 必须取配对的 raw,不能是循环残留变量。"""
def test_normalized_matches_carries_raw(self):
src = _read("data/bzzoiro.py")
# 规范化结果必须与原始 event 成对保存
assert "normalized_matches.append((nm, raw))" in src, (
"normalized_matches 未携带 (nm, raw) 元组 —— raw 变量泄漏会回归 (P0-3)"
)
# 内层消费循环必须解包成对变量
assert re.search(r"for nm, raw in normalized_matches", src), (
"消费循环未解包 (nm, raw) —— 血缘字段会取到错误 event (P0-3)"
)
def test_no_orphan_raw_use(self):
"""source_event_id 所在行必须在解包循环内(用缩进 + 上下文粗判)。"""
src = _read("data/bzzoiro.py")
lines = src.splitlines()
# 找到 "for nm, raw in normalized_matches" 所在行号
start = next(
(i for i, ln in enumerate(lines) if "for nm, raw in normalized_matches" in ln),
None,
)
assert start is not None
# 该循环之后、下一个同/更低缩进的顶层语句之前的范围
seg = "\n".join(lines[start:])
uses = [ln for ln in seg.splitlines() if "source_event_id" in ln]
assert uses, "未找到 source_event_id 赋值"
assert all("raw.get(" in ln for ln in uses), (
"source_event_id 未使用配对的 raw (P0-3)"
)