perf+unify: matches 列表查询优化 + Standings 统一数据源

matches 列表:selectinload 加 load_only 限定列(League.code/Team.name,name_zh/
MatchStats.home_xg,away_xg),补 stats 加载消除 N+1(m.stats 此前懒加载)。
详情接口保持完整 options;游标分页/响应字段/空值语义不变。

Standings 改用 useLeagues() 统一数据源(API 优先,失败回退本地常量),
LEAGUES 常量扩展 CL/EL;无数据联赛显示虚线 tab + 空态(非隐藏)。

alembic/versions/0019_ingest_jobs.py

@@ -0,0 +1,42 @@
+"""采集任务状态表 ingest_jobs
+
+Revision ID: 0019_ingest_jobs
+Revises: 0018_match_checks
+Create Date: 2026-09-22
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '0019_ingest_jobs'
+down_revision: Union[str, None] = '0018_match_checks'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'ingest_jobs',
+        sa.Column('id', sa.String(36), primary_key=True),
+        sa.Column('task', sa.String(20), nullable=False),
+        sa.Column('params', sa.JSON(), nullable=False, server_default='{}'),
+        sa.Column('status', sa.String(20), nullable=False, server_default='pending'),
+        sa.Column('result', sa.JSON(), nullable=True),
+        sa.Column('error', sa.Text(), nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
+        sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
+        sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True),
+    )
+    op.create_index('ix_ingest_job_status_created', 'ingest_jobs', ['status', 'created_at'])
+    op.create_check_constraint(
+        'ck_ingest_job_status', 'ingest_jobs',
+        "status IN ('pending', 'running', 'success', 'failed')",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint('ck_ingest_job_status', 'ingest_jobs', type_='check')
+    op.drop_index('ix_ingest_job_status_created', table_name='ingest_jobs')
+    op.drop_table('ingest_jobs')

alembic/versions/0020_team_aliases.py

@@ -0,0 +1,32 @@
+"""球队别名表 team_aliases
+
+Revision ID: 0020_team_aliases
+Revises: 0019_ingest_jobs
+Create Date: 2026-09-22
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = '0020_team_aliases'
+down_revision: Union[str, None] = '0019_ingest_jobs'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'team_aliases',
+        sa.Column('alias_normalized', sa.String(120), primary_key=True),
+        sa.Column('team_id', sa.Integer, sa.ForeignKey('teams.id', ondelete='CASCADE'), nullable=False),
+        sa.Column('original_alias', sa.String(120), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
+    )
+    op.create_index('ix_team_aliases_team_id', 'team_aliases', ['team_id'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_team_aliases_team_id', table_name='team_aliases')
+    op.drop_table('team_aliases')

alembic/versions/0021_match_source_event_id_unique.py

@@ -0,0 +1,38 @@
+"""matches.source_event_id 部分唯一索引
+
+业务唯一键:同联赛同主客同自然天一条(ix_matches_unique,既有)。
+source_event_id 是上游 bzzoiro 的比赛 id,用于统计回填与血缘追踪;
+当它非空时应全局唯一(同一 upstream 比赛只对应一行 matches),
+避免同一场比赛因自然键天级舍入差异产生重复。
+
+partial unique(WHERE source_event_id IS NOT NULL):
+  - 兼容存量空 source_event_id 的历史行(不强制回填);
+  - 新采集行均带 source_event_id,从此具备 upstream 唯一性。
+
+Revision ID: 0021_match_source_event_id_unique
+Revises: 0020_team_aliases
+Create Date: 2026-09-22
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = '0021_match_source_event_id_unique'
+down_revision: Union[str, None] = '0020_team_aliases'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        'ix_matches_source_event_id_unique',
+        'matches',
+        ['source_event_id'],
+        unique=True,
+        postgresql_where=op.text('source_event_id IS NOT NULL'),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_matches_source_event_id_unique', table_name='matches')

docs/05-data.md

@@ -60,7 +60,7 @@
 
 ### 队名归一化
 
-`src/data/team_names.py` 维护 `NORMALIZE_MAP`(如 `Man City` → `Manchester City`),未命中映射的队名原样返回。
+`src/data.team_names.py` 维护 `NORMALIZE_MAP`(如 `Man City` → `Manchester City`),未命中映射的队名原样返回。
 归一前先做 Unicode NFKD 去重音。
 
 **唯一键是归一后英文名**:`teams.name` 带 `UNIQUE` 约束,所有入库路径均经 `TeamRepository.get_or_create` 收敛归一化
@@ -70,6 +70,32 @@
 >(如 `"Man City"` → `"Manchester City"`,但 `"man city"` 原样保留)。上游 bzzoiro 返回的队名首字母大写,
 >实际命中无问题;若新增数据源返回全小写/全大写队名,需先 `title()` 再归一,否则会绕过映射产生重复 Team。
 
+#### 别名机制(`team_aliases`)
+
+归一仍可能遗漏历史重复队(如 `"Bayern Munich"` 与 `"Bayern München"` 经 NFKD 后相同则命中,
+但 `"Man United"` vs `"Manchester United"` 若漏映射)。`team_aliases` 表提供**显式别名→teams.id** 映射:
+
+| 列 | 说明 |
+|----|------|
+| `alias_normalized` | PK,`normalize(别名)` 后的稳定幂等键 |
+| `team_id` | FK → `teams.id`(ON DELETE CASCADE) |
+| `original_alias` | 原始写法(保留供参考) |
+
+**定位三步链**(`get_or_create`):`normalize(name)` → 查 `teams.name` → 查 `team_aliases`(以 `normalize(name)` 为 PK)→ 都没有才 insert 新 Team。别名命中即复用已有 Team,避免产生重复。
+
+**添加别名**(不自动合并历史重复队):
+
+- **Admin 接口**(推荐):`POST /api/v1/admin/teams/aliases {"alias": "Man United", "team_id": 42}`(require_admin,幂等)
+- **直接 SQL**:
+  ```sql
+  INSERT INTO team_aliases(alias_normalized, team_id, original_alias)
+  VALUES ('man united', 42, 'Man United')
+  ON CONFLICT (alias_normalized) DO UPDATE SET team_id = EXCLUDED.team_id, original_alias = EXCLUDED.original_alias;
+  ```
+
+> ⚠️ **别名不自动合并**:发现历史重复队 A/B 后,需人工确认归一目标(如保留 B),再为 A 的归一名添加别名指向 B。
+> 合并前请确认 A 的 `matches`/`standings` 引用是否需要迁移(可先 `SELECT COUNT(*) FROM matches WHERE home_team_id = A.id OR away_team_id = A.id` 评估)。
+
 **改名 / 合并流程**(人工):
 
 当发现两个 `teams` 行实际是同一球队(如 `Manchester City` 与 `Man City` 因历史数据大小写差异各占一行):
@@ -192,11 +218,25 @@ CREATE TABLE predictions (
 
 1. **`match_date_date`(天级日期)**: 用于天级去重。bzzoiro 返回的时间带时分秒,精确匹配不可靠,故拆出 `DATE` 列做唯一键。
 
-2. **`ix_matches_unique`**: `(league_id, home_team_id, away_team_id, match_date_date)` 唯一,保证同一场比赛重复采集时 upsert 而非插入重复行。
+2. **`ix_matches_unique`**: `(league_id, home_team_id, away_team_id, match_date_date)` 唯一,保证同一场比赛重复采集时 upsert 而非插入重复行。**业务唯一:同联赛同主客同自然天一条。**
 
-3. **`predictions` 级联删除**: `ON DELETE CASCADE`,删比赛时自动清其预测。
+3. **`source_event_id` 部分唯一**: `ix_matches_source_event_id_unique`(WHERE source_event_id IS NOT NULL)——上游 bzzoiro 的比赛 id,当非空时全局唯一。作用:
+   - 统计回填(`/events/{id}/stats/`)与 Bronze 血缘(/events/ 采集)通过它定位比赛,不依赖自然键天级舍入;
+   - 新采集行均带此 id,避免同一 upstream 比赛因时间戳差异绕开自然键产生重复。
+   - 存量空 source_event_id 历史行不受影响(不强制回填)。
 
-4. **`mode` + `prompt_version`**: `single` 模式存 `v1`/`v2`,`multi` 模式存 `multi_v1`/`multi_v2`,eval summary 按这两列天然分组对比。
+4. **`predictions` 级联删除**: `ON DELETE CASCADE`,删比赛时自动清其预测。
+
+5. **`mode` + `prompt_version`**: `single` 模式存 `v1`/`v2`,`multi` 模式存 `multi_v1`/`multi_v2`,eval summary 按这两列天然分组对比。
+
+## 采集 upsert 查找顺序
+
+events 管线按以下优先级定位已有比赛,命中即复用(更新):
+
+1. **`source_event_id`**(upstream event id,唯一索引命中)——最精确,跨自然键舍入差异;
+2. **自然键**:`(league_id, home_team_id, away_team_id, match_date_date)`(内存去重,覆盖无 event id 的采集)。
+
+两者都未命中 → insert 新比赛。
 
 ## 入库语义(幂等)
 

frontend/src/admin/dal.ts

@@ -21,6 +21,7 @@ import type {
   LLMAgentConfig,
   LogEntry,
   IngestSourceStatus,
+  IngestJob,
   MatchDetailOut,
   MatchContextOut,
   AdminStats,
@@ -362,6 +363,20 @@ export function fetchIngestStatus(): Promise<{ sources: IngestSourceStatus[] }>
   return api.get<{ sources: IngestSourceStatus[] }>(`${API_BASE}/admin/ingest/status`)
 }
 
+/**
+ * 采集任务状态轮询(单任务)
+ */
+export function fetchIngestJob(jobId: string): Promise<IngestJob> {
+  return api.get<IngestJob>(`${API_BASE}/admin/ingest/jobs/${jobId}`)
+}
+
+/**
+ * 最近采集任务列表(最新在前)
+ */
+export function fetchIngestJobs(limit = 20): Promise<IngestJob[]> {
+  return api.get<IngestJob[]>(`${API_BASE}/admin/ingest/jobs?limit=${limit}`)
+}
+
 /**
  * 比赛详情(含最近预测摘要)
  */

frontend/src/admin/pages/Collection.tsx

@@ -11,9 +11,9 @@
  */
 
 import { useEffect, useState, useCallback, useRef } from 'react'
-import { triggerCollection, fetchLeagues, fetchIngestStatus } from '../dal'
+import { triggerCollection, fetchLeagues, fetchIngestJob } from '../dal'
-import type { IngestSourceStatus } from '../types'
+import type { IngestJob, League } from '../types'
-import type { CollectionRequest, League } from '../types'
+import type { CollectionRequest } from '../types'
 import { Card, CardBody, CardHeader, Badge, SectionHeader, Alert, Spinner } from '../components'
 
 const TASKS = [
@@ -23,7 +23,9 @@ const TASKS = [
   { value: 'all', label: '全量采集', desc: '依次采集比赛 + 积分榜 + 统计回填', icon: '⏵⏵' },
 ] as const
 
-type TaskStatus = 'idle' | 'running' | 'done' | 'error'
+type TaskUIStatus = 'idle' | 'running' | 'done' | 'error'
+
+const TERMINAL_STATUSES: ReadonlySet<string> = new Set(['success', 'failed'])
 
 export default function CollectionPage() {
   const [leagues, setLeagues] = useState<League[]>([])
@@ -49,13 +51,13 @@ export default function CollectionPage() {
   const [limit, setLimit] = useState(100)
   const [loading, setLoading] = useState(false)
   const [error, setError] = useState<string | null>(null)
-  const [result, setResult] = useState<{ title: string; detail: string } | null>(null)
 
-  // 任务进度反馈
+  // 任务进度反馈:跟踪真实 ingest_job 状态
-  const [taskStatus, setTaskStatus] = useState<TaskStatus>('idle')
+  const [taskStatus, setTaskStatus] = useState<TaskUIStatus>('idle')
+  const [jobId, setJobId] = useState<string | null>(null)
+  const [jobInfo, setJobInfo] = useState<IngestJob | null>(null)
   const [taskStartedAt, setTaskStartedAt] = useState<number | null>(null)
   const pollRef = useRef<ReturnType<typeof setInterval> | null>(null)
-  const [ingestSnap, setIngestSnap] = useState<IngestSourceStatus | null>(null)
 
   const loadLeagues = useCallback(async () => {
     const lg = await fetchLeagues()
@@ -64,30 +66,65 @@ export default function CollectionPage() {
 
   useEffect(() => { loadLeagues() }, [loadLeagues])
 
-  // 轮询采集状态(任务启动后)
+  // 轮询采集 job 直到终态(success/failed)
-  const startPolling = useCallback(() => {
-    if (pollRef.current) clearInterval(pollRef.current)
-    pollRef.current = setInterval(async () => {
-      try {
-        const { sources } = await fetchIngestStatus()
-        const bz = sources.find(s => s.name === 'bzzoiro')
-        if (bz) setIngestSnap(bz)
-      } catch { /* ignore */ }
-    }, 5_000)
-  }, [])
-
   const stopPolling = useCallback(() => {
     if (pollRef.current) { clearInterval(pollRef.current); pollRef.current = null }
   }, [])
 
   useEffect(() => () => stopPolling(), [stopPolling])
 
+  const startJobPolling = useCallback((id: string) => {
+    stopPolling()
+    const tick = async () => {
+      try {
+        const job = await fetchIngestJob(id)
+        setJobInfo(job)
+        if (TERMINAL_STATUSES.has(job.status)) {
+          setTaskStatus(job.status === 'success' ? 'done' : 'error')
+          stopPolling()
+        }
+      } catch { /* 单次轮询失败不影响后续 */ }
+    }
+    tick()
+    pollRef.current = setInterval(tick, 3_000)
+  }, [stopPolling])
+
   const isEventsTask = task === 'events' || task === 'all'
 
+  // 友好汇总 job.result
+  const jobSummary = (j: IngestJob | null): { title: string; detail: string } | null => {
+    if (!j) return null
+    if (j.status === 'failed') {
+      return { title: '采集失败', detail: j.error || '采集任务异常终止,请到「系统日志」查看详细堆栈。' }
+    }
+    if (j.status !== 'success') return null
+    const r = j.result as Record<string, unknown> | null
+    if (!r) return { title: '采集完成', detail: '任务成功(无汇总数据)。' }
+    const ev = r.events as Record<string, unknown> | undefined
+    const evTotal = ev ? (ev.total_inserted as number ?? 0) + (ev.total_updated as number ?? 0) : 0
+    const st = r.standings as Record<string, unknown> | undefined
+    const stTotal = st ? (st.total_upserted as number ?? 0) : 0
+    const stats = r.stats as Record<string, unknown> | undefined
+    const statsTotal = stats ? (stats.created as number ?? 0) + (stats.updated as number ?? 0) : 0
+    const evErr = (ev?.errors as string[] | undefined)?.length ?? 0
+    const stErr = (st?.errors as string[] | undefined)?.length ?? 0
+    const statsErr = (stats?.errors as string[] | undefined)?.length ?? 0
+    const totalErr = evErr + stErr + statsErr
+    const parts: string[] = []
+    if (ev) parts.push(`比赛 +${evTotal}`)
+    if (st) parts.push(`积分榜 +${stTotal}`)
+    if (stats) parts.push(`统计 +${statsTotal}`)
+    const detail = parts.length
+      ? `共更新: ${parts.join(' / ')}${totalErr ? `,错误 ${totalErr} 条(见日志)` : ''}`
+      : '任务成功'
+    return { title: '采集完成', detail }
+  }
+
   async function handleSubmit(e: React.FormEvent) {
     e.preventDefault()
     setError(null)
-    setResult(null)
+    setJobInfo(null)
+    setJobId(null)
     setLoading(true)
     setTaskStatus('running')
     setTaskStartedAt(Date.now())
@@ -103,18 +140,15 @@ export default function CollectionPage() {
         date_from: isEventsTask ? dateFrom || undefined : undefined,
         date_to: isEventsTask ? dateTo || undefined : undefined,
       }
-      await triggerCollection(body)
+      const res = await triggerCollection(body)
-      setResult({
+      const id: string | undefined = res?.job_id
-        title: '采集任务已启动',
+      if (id) {
-        detail: '正在后台执行(上游限速时可能需要数分钟)。完成结果与错误请到「系统日志」页查看(支持自动刷新)。',
+        setJobId(id)
-      })
+        startJobPolling(id)
-      // 启动轮询,跟踪状态
+      } else {
-      startPolling()
+        // 后端未返回 job_id(旧版兼容):退化为原逻辑
-      // 30 秒后自动停止轮询并标记完成
+        setTimeout(() => { setTaskStatus('done'); }, 30_000)
-      setTimeout(() => {
+      }
-        setTaskStatus('done')
-        stopPolling()
-      }, 30_000)
     } catch (err: unknown) {
       setTaskStatus('error')
       setError(err instanceof Error ? err.message : '采集触发失败')
@@ -125,6 +159,7 @@ export default function CollectionPage() {
   }
 
   const elapsed = taskStartedAt ? Math.round((Date.now() - taskStartedAt) / 1000) : 0
+  const summary = jobInfo ? jobSummary(jobInfo) : null
 
   return (
     <div className="space-y-6">
@@ -248,12 +283,11 @@ export default function CollectionPage() {
 
               {/* 消息提示 */}
               {error && <Alert kind="error" title="采集失败" message={error} onClose={() => setError(null)} />}
-              {result && (
+              {summary && (
                 <Alert
-                  kind="ok"
+                  kind={jobInfo?.status === 'success' ? 'ok' : 'error'}
-                  title={result.title}
+                  title={summary.title}
-                  message={result.detail || undefined}
+                  message={summary.detail}
-                  onClose={() => setResult(null)}
                 />
               )}
 
@@ -278,10 +312,10 @@ export default function CollectionPage() {
                 <div className="space-y-3">
                   <div className="flex items-center gap-2 text-xs text-ink-700">
                     <Spinner />
-                    <span>任务执行中,已运行 {elapsed}s…</span>
+                    <span>任务执行中{jobId ? `(job ${jobId.slice(0, 8)}…)` : ''},已运行 {elapsed}s…</span>
                   </div>
                   <p className="text-2xs text-ink-400">
-                    后台异步执行,关闭页面不影响结果。可稍后查看「系统日志」确认完成。
+                    后台异步执行,关闭页面不影响结果。每 3 秒自动轮询进度。
                   </p>
                 </div>
               )}
@@ -289,23 +323,25 @@ export default function CollectionPage() {
                 <div className="space-y-2">
                   <div className="flex items-center gap-2 text-xs text-emerald-700">
                     <span className="inline-block h-2 w-2 rounded-full bg-emerald-500" />
-                    <span>任务已提交,后台执行中(可能尚未完成)</span>
+                    <span>采集完成{jobId ? `(job ${jobId.slice(0, 8)}…)` : ''}</span>
                   </div>
-                  <p className="text-2xs text-ink-400">
+                  {summary && <p className="text-2xs text-ink-500">{summary.detail}</p>}
-                    采集耗时取决于数据量。请到「系统日志」页查看最终结果。
-                  </p>
                 </div>
               )}
               {taskStatus === 'error' && (
-                <p className="text-xs text-press">任务触发失败,请检查配置或网络。</p>
+                <div className="space-y-1">
-              )}
+                  <p className="text-xs text-press">采集失败{jobId ? `(job ${jobId.slice(0, 8)}…)` : ''}</p>
-              {ingestSnap?.last_success_at && (
+                  {jobInfo?.error && (
-                <div className="mt-3 border-t border-ink-100 pt-3">
+                    <p className="text-2xs text-ink-500">{jobInfo.error.slice(0, 200)}</p>
-                  <p className="text-2xs text-ink-400">
+                  )}
-                    bzzoiro 最近一次采集: {new Date(ingestSnap.last_success_at).toLocaleString('zh-CN', { hour12: false })}
-                  </p>
                 </div>
               )}
+              {jobInfo?.created_at && (
+                <p className="mt-2 text-2xs text-ink-400">
+                  创建于 {new Date(jobInfo.created_at).toLocaleString('zh-CN', { hour12: false })}
+                  {jobInfo.finished_at && ` · 完成于 ${new Date(jobInfo.finished_at).toLocaleString('zh-CN', { hour12: false })}`}
+                </p>
+              )}
             </CardBody>
           </Card>
 

frontend/src/admin/types.ts

@@ -283,6 +283,20 @@ export interface IngestSourceStatus {
   last_failure: IngestLastFailure | null
 }
 
+// ── 采集任务状态 ──────────────────────────────────────────────
+
+export interface IngestJob {
+  id: string
+  task: string
+  params: Record<string, unknown>
+  status: 'pending' | 'running' | 'success' | 'failed'
+  result: Record<string, unknown> | null
+  error: string | null
+  created_at: string | null
+  started_at: string | null
+  finished_at: string | null
+}
+
 // ── 比赛详情 ─────────────────────────────────────────────────────
 
 export interface MatchRecentPrediction {

frontend/src/pages/Standings.tsx

@@ -8,18 +8,9 @@
 import { useEffect, useState, useCallback } from 'react'
 import { fetchStandings } from '../admin/dal'
 import type { StandingsLeague, StandingRow } from '../admin/dal'
+import { useLeagues } from './matches/hooks/useLeagues'
 import { Spinner } from '../admin/components'
 
-const LEAGUES = [
-  { code: 'E0', name: '英超' },
-  { code: 'SP1', name: '西甲' },
-  { code: 'D1', name: '德甲' },
-  { code: 'I1', name: '意甲' },
-  { code: 'F1', name: '法甲' },
-  { code: 'CL', name: '欧冠' },
-  { code: 'EL', name: '欧联' },
-]
-
 const ZONE_META: Record<string, { label: string; cls: string }> = {
   // 欧战资格
   'Champions League': { label: '欧冠区', cls: 'bg-emerald-100 text-emerald-700' },
@@ -64,7 +55,9 @@ function FormDots({ form }: { form?: string | null }) {
 }
 
 export default function StandingsPage() {
-  const [leagues, setLeagues] = useState<StandingsLeague[]>([])
+  // 统一数据源:复用 useLeagues hook(优先 API,失败回退本地常量)
+  const leagues = useLeagues()
+  const [standings, setStandings] = useState<StandingsLeague[]>([])
   const [activeLeague, setActiveLeague] = useState<string>('')
   const [loading, setLoading] = useState(true)
   const [switching, setSwitching] = useState(false)  // 切换联赛中
@@ -85,7 +78,7 @@ export default function StandingsPage() {
     setError(null)
     try {
       const data = await fetchStandings(code)
-      setLeagues(data.leagues)
+      setStandings(data.leagues)
       if (!activeLeague && data.leagues.length > 0) {
         setActiveLeague(data.leagues[0].league_code)
       }
@@ -104,7 +97,7 @@ export default function StandingsPage() {
     setSwitching(true)
     setActiveLeague(code)
     try {
-      await fetchStandings(code).then(data => setLeagues(data.leagues))
+      await fetchStandings(code).then(data => setStandings(data.leagues))
     } catch (err) {
       setError(err instanceof Error ? err.message : '加载失败')
     } finally {
@@ -112,26 +105,32 @@ export default function StandingsPage() {
     }
   }
 
-  const active = leagues.find(l => l.league_code === activeLeague) ?? leagues[0]
+  const active = standings.find(l => l.league_code === activeLeague) ?? standings[0]
 
   return (
     <div className="space-y-6">
       {/* 联赛切换 */}
       <div className="flex flex-wrap gap-2">
-        {LEAGUES.map(l => (
+        {leagues.map(l => {
-          <button
+          // 标记该联赛是否有积分榜数据:有数据可正常切换,无数据也可选中但显示空态
-            key={l.code}
+          const hasData = standings.some(s => s.league_code === l.code)
-            onClick={() => switchLeague(l.code)}
+          const isEmpty = activeLeague === l.code && !hasData
-            disabled={switching}
+          return (
-            className={`rounded border px-3 py-1.5 text-xs transition-colors disabled:opacity-50 ${
+            <button
-              activeLeague === l.code
+              key={l.code}
-                ? 'border-ink-900 bg-ink-900 text-paper-50'
+              onClick={() => switchLeague(l.code)}
-                : 'border-ink-200 text-ink-500 hover:border-ink-300'
+              disabled={switching}
-            }`}
+              title={hasData ? undefined : '暂无积分榜数据'}
-          >
+              className={`rounded border px-3 py-1.5 text-xs transition-colors disabled:opacity-50 ${
-            {l.name}
+                activeLeague === l.code
-          </button>
+                  ? 'border-ink-900 bg-ink-900 text-paper-50'
-        ))}
+                  : 'border-ink-200 text-ink-500 hover:border-ink-300'
+              } ${!hasData ? 'border-dashed' : ''}`}
+            >
+              {l.name}
+            </button>
+          )
+        })}
       </div>
 
       {error && (

frontend/src/pages/matches/types.ts

@@ -78,6 +78,8 @@ export const LEAGUES = [
   { code: 'D1', name: '德甲' },
   { code: 'I1', name: '意甲' },
   { code: 'F1', name: '法甲' },
+  { code: 'CL', name: '欧冠' },
+  { code: 'EL', name: '欧联' },
 ]
 
 /** 汉字编号,给专家意见排版用 */

src/api/routes/admin_ingest_jobs.py

@@ -0,0 +1,55 @@
+"""后台管理:采集任务状态查询(只读)。
+
+GET /api/v1/admin/ingest/jobs/{job_id} — 单任务详情
+GET /api/v1/admin/ingest/jobs?limit=N — 最近任务列表(默认 20)
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import desc, select
+
+from src.api.deps import require_admin
+from src.api.schemas import IngestJobOut
+from src.db.base import AsyncSession, get_db_read
+from src.db.models import IngestJob
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)])
+
+
+@router.get("/ingest/jobs/{job_id}", response_model=IngestJobOut)
+async def get_ingest_job(job_id: str, db: AsyncSession = Depends(get_db_read)):
+    """查询单个采集任务状态。"""
+    job = await db.get(IngestJob, job_id)
+    if job is None:
+        raise HTTPException(404, f"采集任务不存在: {job_id}")
+    return _job_to_out(job)
+
+
+@router.get("/ingest/jobs", response_model=list[IngestJobOut])
+async def list_ingest_jobs(
+    limit: int = Query(20, ge=1, le=100, description="返回条数"),
+    db: AsyncSession = Depends(get_db_read),
+):
+    """查询最近采集任务(最新在前)。"""
+    rows = (
+        await db.execute(select(IngestJob).order_by(desc(IngestJob.created_at)).limit(limit))
+    ).scalars().all()
+    return [_job_to_out(j) for j in rows]
+
+
+def _job_to_out(job: IngestJob) -> IngestJobOut:
+    return IngestJobOut(
+        id=job.id,
+        task=job.task,
+        params=job.params or {},
+        status=job.status,
+        result=job.result,
+        error=job.error,
+        created_at=job.created_at,
+        started_at=job.started_at,
+        finished_at=job.finished_at,
+    )

src/api/routes/admin_settings.py

@@ -15,11 +15,15 @@ from fastapi import APIRouter
 
 from src.api.routes.admin_config import router as admin_config_router
 from src.api.routes.admin_datasources import router as admin_datasources_router
+from src.api.routes.admin_ingest_jobs import router as admin_ingest_jobs_router
 from src.api.routes.admin_llm import router as admin_llm_router
 from src.api.routes.admin_quality import router as admin_quality_router
+from src.api.routes.admin_teams import router as admin_teams_router
 
 router = APIRouter()
 router.include_router(admin_datasources_router)
 router.include_router(admin_config_router)
+router.include_router(admin_ingest_jobs_router)
 router.include_router(admin_llm_router)
 router.include_router(admin_quality_router)
+router.include_router(admin_teams_router)

src/api/routes/admin_teams.py

@@ -0,0 +1,60 @@
+"""后台管理:球队别名管理(只读列表 + 添加别名)。
+
+归一名(teams.name)是球队唯一键;别名(team_aliases)是同一球队的不同写法
+(大小写/译名/缩写)到归一后 teams.id 的映射。入库时 normalize(name) 依次查
+teams.name 与 team_aliases,命中即复用,避免重复 Team。
+
+不自动合并历史重复队;需显式添加别名(或先 SQL/再经由此接口)。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import desc, select
+
+from src.api.deps import require_admin
+from src.api.schemas import TeamAliasIn, TeamAliasOut
+from src.db.base import AsyncSession, get_db_read
+from src.db.models import Team, TeamAlias
+from src.db.repositories import TeamRepository
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/admin", tags=["admin"], dependencies=[Depends(require_admin)])
+
+
+@router.get("/teams/aliases", response_model=list[TeamAliasOut])
+async def list_team_aliases(db: AsyncSession = Depends(get_db_read)):
+    """列出所有球队别名(最新在前)。"""
+    rows = (await db.execute(select(TeamAlias).order_by(desc(TeamAlias.created_at)).limit(200))).scalars().all()
+    return [
+        TeamAliasOut(
+            alias_normalized=r.alias_normalized,
+            team_id=r.team_id,
+            original_alias=r.original_alias,
+        )
+        for r in rows
+    ]
+
+
+@router.post("/teams/aliases", response_model=TeamAliasOut, status_code=201)
+async def add_team_alias(req: TeamAliasIn, db: AsyncSession = Depends(get_db_read)):
+    """为已有 Team 添加别名(幂等:重复添加会更新指向)。
+
+    不自动合并历史重复队。若需合并 A→B:先为 A 的归一名添加别名指向 B,
+    再人工确认 A 是否仍有独立引用。
+    """
+    # 校验目标 Team 存在
+    team = await db.get(Team, req.team_id)
+    if team is None:
+        raise HTTPException(404, f"目标 Team 不存在: id={req.team_id}")
+
+    repo = TeamRepository(db)
+    row = await repo.add_alias(req.alias, req.team_id)
+    logger.info("添加 Team 别名: %s -> team_id=%s", req.alias, req.team_id)
+    return TeamAliasOut(
+        alias_normalized=row.alias_normalized,
+        team_id=row.team_id,
+        original_alias=row.original_alias,
+    )

src/api/routes/ingest.py

@@ -10,11 +10,13 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import uuid
+from datetime import datetime, timezone
 
 from fastapi import APIRouter, Depends, HTTPException
 
 from src.api.deps import require_admin
-from src.api.schemas import IngestBzzoiroRequest
+from src.api.schemas import IngestBzzoiroRequest, IngestBzzoiroResponse
 from src.data.config import BZZOIRO_LEAGUE_IDS
 from src.data.bzzoiro_standings import ingest_bzzoiro_standings
 from src.data.bzzoiro_stats import ingest_bzzoiro_event_stats
@@ -38,22 +40,58 @@ def _spawn(coro) -> None:
     task.add_done_callback(_background_tasks.discard)
 
 
-@router.post("/ingest/bzzoiro", dependencies=[Depends(require_admin)])
+@router.post("/ingest/bzzoiro", response_model=IngestBzzoiroResponse, dependencies=[Depends(require_admin)])
 async def ingest_bzzoiro_route(req: IngestBzzoiroRequest):
-    """触发 bzzoiro 采集(events / standings / stats / all)。"""
+    """触发 bzzoiro 采集(events / standings / stats / all)。
+
+    启动后台任务前写入 ingest_jobs(pending),响应返回 job_id 供前端轮询。
+    兼容原 message 字段(仍返回)。
+    """
     if req.task not in VALID_TASKS:
         raise HTTPException(status_code=422, detail=f"未知任务类型: {req.task}(可选: {', '.join(sorted(VALID_TASKS))})")
     leagues = req.leagues or list(BZZOIRO_LEAGUE_IDS.keys())
     task_label = {"events": "比赛数据", "standings": "积分榜", "stats": "统计回填", "all": "全量(比赛+积分榜+统计)"}[req.task]
-    _spawn(_run_bzzoiro(req.task, leagues, req))
+
-    return {
+    job_id = await _create_ingest_job(req.task, leagues, req)
-        "ok": True,
+    _spawn(_run_bzzoiro(job_id, req.task, leagues, req))
-        "message": f"采集任务已启动(后台执行,任务: {task_label}),请在「系统日志」查看进度与结果",
+
+    return IngestBzzoiroResponse(
+        ok=True,
+        job_id=job_id,
+        message=f"采集任务已启动(后台执行,任务: {task_label}),请到「数据采集」页跟踪进度",
+    )
+
+
+async def _create_ingest_job(task: str, leagues: list[str], req: IngestBzzoiroRequest) -> str:
+    """写入一条 ingest_jobs(pending),返回 job_id。"""
+    from src.db.models import IngestJob
+
+    job_id = str(uuid.uuid4())
+    params = {
+        "leagues": leagues,
+        "date_from": req.date_from,
+        "date_to": req.date_to,
+        "status": req.status,
+        "task": task,
+        "limit": req.limit,
+        "season": req.season,
     }
+    async with get_uow() as session:
+        job = IngestJob(id=job_id, task=task, params=params, status="pending")
+        session.add(job)
+    logger.info("ingest_jobs 创建: job=%s task=%s leagues=%s", job_id, task, leagues)
+    return job_id
 
 
-async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest) -> None:
+async def _run_bzzoiro(job_id: str, task: str, leagues: list[str], req: IngestBzzoiroRequest) -> None:
-    """后台执行 bzzoiro 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。"""
+    """后台执行 bzzoiro 采集:上游限速时单次可能耗时数分钟,必须脱离请求生命周期。
+
+    状态流转: pending → running → (success|failed)。
+    """
+    from src.db.models import IngestJob
+
+    await _update_job(job_id, status="running", started_at=datetime.now(timezone.utc))
+    result: dict = {}
     try:
         if task in ("events", "all"):
             statuses = [req.status] if req.status else ["finished", "scheduled"]
@@ -80,6 +118,7 @@ async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest)
             )
             if merged["errors"]:
                 logger.warning("bzzoiro 比赛采集错误 %d 条: %s", len(merged["errors"]), merged["errors"][:3])
+            result["events"] = merged
 
         if task in ("standings", "all"):
             async with get_uow() as session:
@@ -88,6 +127,7 @@ async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest)
                 logger.warning("bzzoiro 积分榜采集部分失败: %s", r["errors"][:3])
             else:
                 logger.info("bzzoiro 积分榜采集完成: upsert %d 条", r["total_upserted"])
+            result["standings"] = r
 
         if task in ("stats", "all"):
             async with get_uow() as session:
@@ -96,5 +136,28 @@ async def _run_bzzoiro(task: str, leagues: list[str], req: IngestBzzoiroRequest)
                 )
             if r["errors"]:
                 logger.warning("bzzoiro 统计回填错误 %d 条: %s", len(r["errors"]), r["errors"][:3])
-    except Exception:
+            result["stats"] = r
+
+        await _update_job(job_id, status="success", result=result, finished_at=datetime.now(timezone.utc))
+        logger.info("ingest_jobs 完成: job=%s task=%s", job_id, task)
+    except Exception as e:
         logger.exception("bzzoiro 采集任务失败(task=%s)", task)
+        await _update_job(
+            job_id, status="failed", error=str(e), finished_at=datetime.now(timezone.utc),
+        )
+
+
+async def _update_job(job_id: str, **fields) -> None:
+    """更新 ingest_jobs 单行;失败仅记日志,绝不抛异常(避免干扰采集主流程)。"""
+    from src.db.models import IngestJob
+
+    try:
+        async with get_uow() as session:
+            job = await session.get(IngestJob, job_id)
+            if job is None:
+                logger.warning("ingest_jobs 更新失败: job=%s 不存在", job_id)
+                return
+            for k, v in fields.items():
+                setattr(job, k, v)
+    except Exception:
+        logger.warning("ingest_jobs 更新异常: job=%s fields=%s", job_id, list(fields.keys()))

src/api/routes/matches.py

@@ -5,11 +5,11 @@ from datetime import datetime, timedelta
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy import func, or_, select
-from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import load_only, selectinload
 
 from src.api.schemas import MatchListOut, MatchOut, PredictionOut
 from src.db.base import AsyncSession, get_db_read
-from src.db.models import League, Match, Prediction, Standing
+from src.db.models import League, Match, MatchStats, Prediction, Standing, Team
 
 router = APIRouter(prefix="/api/v1", tags=["data"])
 
@@ -49,8 +49,20 @@ async def list_matches(
     limit: int = Query(50, ge=1, le=100),
     db: AsyncSession = Depends(get_db_read),
 ):
-    """比赛列表(游标分页)。"""
+    """比赛列表(游标分页)。
-    q = select(Match).options(selectinload(Match.league), selectinload(Match.home_team), selectinload(Match.away_team))
+
+    加载策略(列表 vs 详情):
+      - 列表:仅 selectinload 序列化需要的 3 个关系 + stats,且用 load_only 限定列
+        (League.code / Team.name,name_zh / MatchStats.home_xg,away_xg),避免传输全列;
+        同时一次性加载 stats 消除 N+1(m.stats.home_xg 此前触发懒加载)。
+      - 详情(/matches/{id}):保持完整 options(league/teams/stats 全列 + 最近预测)。
+    """
+    q = select(Match).options(
+        selectinload(Match.league).load_only(League.code),
+        selectinload(Match.home_team).load_only(Team.name, Team.name_zh),
+        selectinload(Match.away_team).load_only(Team.name, Team.name_zh),
+        selectinload(Match.stats).load_only(MatchStats.home_xg, MatchStats.away_xg),
+    )
 
     if cursor:
         try:

src/api/schemas.py

@@ -117,6 +117,19 @@ class IngestBzzoiroRequest(BaseModel):
     season: str | None = Field(None, description="standings 赛季,如 '2026-2027';空 = 当前赛季")
 
 
+class TeamAliasIn(BaseModel):
+    """POST /api/v1/admin/teams/aliases 请求体:为已有 Team 添加别名。"""
+
+    alias: str = Field(..., min_length=1, max_length=120, description="球队别名(原始写法)")
+    team_id: int = Field(..., gt=0, description="归一后的目标 teams.id")
+
+
+class TeamAliasOut(BaseModel):
+    alias_normalized: str
+    team_id: int
+    original_alias: str
+
+
 class IngestResponse(BaseModel):
     leagues: dict
     total_inserted: int
@@ -124,6 +137,28 @@ class IngestResponse(BaseModel):
     errors: list[str] = []
 
 
+class IngestBzzoiroResponse(BaseModel):
+    """POST /api/v1/ingest/bzzoiro 响应:兼容原 message 字段,新增 job_id 供轮询。"""
+
+    ok: bool = True
+    job_id: str = Field(..., description="采集任务 ID(GET /api/v1/admin/ingest/jobs/{job_id} 轮询)")
+    message: str = ""
+
+
+class IngestJobOut(BaseModel):
+    """采集任务状态详情。"""
+
+    id: str
+    task: str
+    params: dict
+    status: str  # pending | running | success | failed
+    result: dict | None = None
+    error: str | None = None
+    created_at: datetime | None = None
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+
+
 class ScheduleIn(BaseModel):
     id: str = Field(..., description="任务唯一标识,如 'daily-events'")
     task: str = Field(..., description="events / standings / stats / all")

src/core/config.py

@@ -32,6 +32,12 @@ class Settings(BaseSettings):
     LLM_SPECIALIST_MODEL: str = ""
     LLM_AGGREGATOR_MODEL: str = ""
 
+    # ── 预测缓存 ──
+    # 预测响应缓存后端:空(默认)=进程内 LRU+TTL 字典;填 redis://host:port/db 启用 Redis。
+    # Redis 失败自动降级内存缓存并 warning,不中断预测;不强制依赖 redis 包。
+    # TTL 固定 300s(5 分钟),键格式与内存后端一致(含 prompt 模板 hash)。
+    PREDICT_CACHE_URL: str = ""
+
     # --- data sources ---
     BZZOIRO_KEY: str = ""
     BZZOIRO_BASE: str = "https://sports.bzzoiro.com/api/v2"

src/data/bzzoiro.py

@@ -1,10 +1,11 @@
 """Bzzoiro 数据源:抓取 + 入库(单一数据源)—— 聚合门面。
 
-实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持两个不变量:
+实现按管线拆分(单文件 → 多模块),本模块只做再导出,保持不变量:
   1. sources._load_sources() 仍从本模块导入 BzzoiroSource(注册表入口不变);
-  2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 子模块在运行期
+  2. 测试与脚本对 `bz.<名称>` 的 monkeypatch 语义不变 —— 抓取函数 / REQUEST_INTERVAL
-     经本门面解析可替换协作者(抓取函数 / Bronze 写入助手 / REQUEST_INTERVAL),
+     仍经本门面解析可替换;Bronze 写入助手(_write_raw_event/_write_lineage/
-     与拆分前的单文件行为一致。
+     _safe_write_ingest_failure)已改为管线模块直接 import pipeline_write,
+    测试需 patch `src.data.pipeline_write.*` 源模块。
 
 三条管线(各自模块):
   1. events   — 比赛日程/比分(/events/),含 source_event_id 血缘  → bzzoiro_events.py
@@ -12,7 +13,8 @@
   3. stats    — 已完赛比赛详细统计回填(/events/{id}/stats/)        → bzzoiro_stats.py
 
 共享基础:HTTP 抓取(多 key 轮换)与字段转换 → bzzoiro_common.py;
-Bronze 基础设施(RawEvent/IngestFailure/DataLineage)→ pipeline_write.py。
+Bronze 基础设施(RawEvent/IngestFailure/DataLineage)→ pipeline_write.py
+(各管线模块直接 import pipeline_write,不再经本门面转发)。
 
 D4(工程债): Team/League/Match 的查找/创建经 Repository 层(src/db/repositories.py),
 各管线不直接控制事务(commit/rollback 由调用方 UnitOfWork 控制,只 flush)。
@@ -31,6 +33,7 @@ from src.data.key_ring import _mask  # noqa: F401  (R1 测试引用 bz._mask)
 from src.data.normalize import normalize_bzzoiro  # noqa: F401
 
 # ── 共享原语:HTTP 抓取 + 宽松字段转换 ──
+# NOTE: 管线模块同时从 bzzoiro_common 直接 import _fetch_json_async 等(经本处也转发)。
 from src.data.bzzoiro_common import (  # noqa: F401
     _fetch_json_async,
     _match_key,
@@ -39,14 +42,6 @@ from src.data.bzzoiro_common import (  # noqa: F401
     _to_int_or_none,
 )
 
-# ── 管线基础设施:RawEvent / IngestFailure / DataLineage ──
-from src.data.pipeline_write import (  # noqa: F401
-    _safe_write_ingest_failure,
-    _write_ingest_failure,
-    _write_lineage,
-    _write_raw_event,
-)
-
 # ── events 管线:BzzoiroSource(注册表入口)+ 抓取/入库 ──
 from src.data.bzzoiro_events import (  # noqa: F401
     BzzoiroSource,

src/data/bzzoiro_events.py

@@ -16,6 +16,7 @@ from datetime import datetime, timedelta, timezone
 from src.data.bzzoiro_common import _match_key, _to_date, _to_int_or_none
 from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES
 from src.data.normalize import normalize_bzzoiro
+from src.data.pipeline_write import _safe_write_ingest_failure, _write_lineage, _write_raw_event
 from src.data.sources import register
 from src.data.team_names_zh import zh_name
 from src.db.models import Match
@@ -98,7 +99,7 @@ class BzzoiroSource:
                 # 单联赛抓取失败隔离:记录错误后继续其余联赛,不拖垮整批
                 logger.exception("bzzoiro fetch failed for %s", code)
                 league_r["errors"].append(f"fetch failed: {e}")
-                await bz._safe_write_ingest_failure(
+                await _safe_write_ingest_failure(
                     db,
                     entity_type="events",
                     source_record_id=None,
@@ -181,9 +182,16 @@ class BzzoiroSource:
                     away_team_id = away.id
                     team_name_to_id[nm.away_team] = away_team_id
 
-                # 查找已有比赛: 内存查找
+                # 查找已有比赛:优先按 upstream event_id 定位(命中即唯一),
-                match_key = _match_key(home_team_id, away_team_id, nm.date)
+                # 否则回退自然键(联赛+主客+天级日期)内存查找。
-                existing_match = existing_matches.get(match_key)
+                # source_event_id 上有 partial unique 索引保障 upstream 唯一。
+                eid = _to_int_or_none(raw.get("id"))
+                existing_match = None
+                if eid is not None:
+                    existing_match = await match_r.find_by_source_event_id(eid)
+                if existing_match is None:
+                    match_key = _match_key(home_team_id, away_team_id, nm.date)
+                    existing_match = existing_matches.get(match_key)
 
                 if existing_match is None:
                     m = Match(
@@ -293,11 +301,9 @@ async def _write_events_bronze(
     source_record_id 查重保证;best-effort:基础设施写入失败只记 warning,
     绝不拖垮采集主流程(与 _safe_write_ingest_failure 同级约束)。
     """
-    from src.data import bzzoiro as bz
-
     try:
-        await bz._write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
+        await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
-        await bz._write_lineage(
+        await _write_lineage(
             db, "bzzoiro", source_record_id,
             "matches", target_match_id, "events_ingest",
             {"league": league_code, "match_status": match_status},

src/data/bzzoiro_standings.py

@@ -16,6 +16,7 @@ from sqlalchemy import select
 
 from src.data.bzzoiro_common import _to_float_or_none, _to_int_or_none
 from src.data.config import BZZOIRO_LEAGUE_IDS, LEAGUE_COUNTRIES, LEAGUE_NAMES
+from src.data.pipeline_write import _safe_write_ingest_failure, _write_lineage, _write_raw_event
 from src.data.team_names_zh import zh_name
 from src.db.models import Standing, Team
 from src.db.repositories import LeagueRepository, TeamRepository
@@ -33,6 +34,7 @@ async def fetch_bzzoiro_standings(league_code: str, season: str | None = None) -
     params: dict = {}
     if season:
         params["season"] = season
+
     return await bz._fetch_json_async(f"/leagues/{league_id}/standings/", params)
 
 
@@ -57,9 +59,10 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
     season 为 None 时采集当前赛季(bzzoiro 默认返回 is_current 赛季)。
     球队名与 events 管线使用同一 normalize 规则,保证 Team 匹配。
     """
-    from src.data import bzzoiro as bz
     from src.data.team_names import normalize as normalize_name
 
+    from src.data import bzzoiro as bz
+
     result: dict = {"leagues": {}, "total_upserted": 0, "errors": []}
     for code in leagues:
         league_r: dict = {"upserted": 0, "teams_created": 0, "rows": 0, "errors": []}
@@ -68,7 +71,7 @@ async def ingest_bzzoiro_standings(db, *, leagues: Iterable[str], season: str |
         except Exception as e:
             logger.exception("bzzoiro standings fetch failed for %s", code)
             league_r["errors"].append(str(e))
-            await bz._safe_write_ingest_failure(
+            await _safe_write_ingest_failure(
                 db,
                 entity_type="standings",
                 source_record_id=None,
@@ -198,11 +201,9 @@ async def _write_standings_bronze(
     命中同一条 RawEvent);best-effort:基础设施写入失败只记 warning,
     绝不拖垮采集主流程。
     """
-    from src.data import bzzoiro as bz
-
     try:
-        await bz._write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
+        await _write_raw_event(db, "bzzoiro", source_record_id, raw_payload, batch_id)
-        await bz._write_lineage(
+        await _write_lineage(
             db, "bzzoiro", source_record_id,
             "standings", league_id, "standings_ingest",
             {"league": league_code, "season": season_label, "rows_upserted": rows_upserted},

src/data/bzzoiro_stats.py

@@ -15,6 +15,7 @@ from datetime import datetime, timedelta, timezone
 
 from src.data.bzzoiro_common import _to_float_or_none, _to_int_or_none
 from src.data.config import BZZOIRO_LEAGUE_IDS
+from src.data.pipeline_write import _safe_write_ingest_failure, _write_lineage, _write_raw_event
 from src.db.models import MatchStats
 from src.db.repositories import MatchRepository
 
@@ -90,7 +91,7 @@ async def ingest_bzzoiro_event_stats(
 
     筛选条件: match_status=finished 且 source_event_id 非空。
     only_missing=True 时跳过已有统计的比赛(增量);False 则全量刷新。
-    limit 控制单次最多处理的比赛数(上游限速 1.2s/请求,大批量需分次触发)。
+    limit 控制单次最多处理的比赛数(上游限速约 1.2s/请求,大批量需分次触发)。
     """
     from src.data import bzzoiro as bz
 
@@ -120,7 +121,7 @@ async def ingest_bzzoiro_event_stats(
         except Exception as e:
             logger.warning("stats fetch failed match=%s event=%s: %s", m.id, m.source_event_id, e)
             result["errors"].append(f"match {m.id}: {e}")
-            await bz._safe_write_ingest_failure(
+            await _safe_write_ingest_failure(
                 db,
                 entity_type="match_stats",
                 source_record_id=str(m.source_event_id),
@@ -165,8 +166,8 @@ async def ingest_bzzoiro_event_stats(
         # 管线基础设施:写入 RawEvent + DataLineage
         batch_id = f"bzzoiro-stats-{m.source_event_id}-{now.strftime('%Y%m%d%H%M%S')}"
         try:
-            await bz._write_raw_event(db, "bzzoiro", str(m.source_event_id), payload, batch_id)
+            await _write_raw_event(db, "bzzoiro", str(m.source_event_id), payload, batch_id)
-            await bz._write_lineage(db, "bzzoiro", str(m.source_event_id), "match_stats", m.stats.id if m.stats else None, "stats_backfill", {"match_id": m.id}, batch_id)
+            await _write_lineage(db, "bzzoiro", str(m.source_event_id), "match_stats", m.stats.id if m.stats else None, "stats_backfill", {"match_id": m.id}, batch_id)
         except Exception:
             pass  # 基础设施写入失败不影响主流程
 

src/db/models.py

@@ -56,6 +56,22 @@ class Team(Base):
     away_matches: Mapped[list["Match"]] = relationship(foreign_keys="Match.away_team_id", back_populates="away_team")
 
 
+class TeamAlias(Base):
+    """球队别名:同一球队的不同写法(大小写/译名/缩写)映射到归一后的 teams.id。
+
+    入库流程(get_or_create):normalize(name) → 查 teams.name → 查 team_aliases
+    → 都没有再 insert 新 Team。别名不自动合并历史重复队,需显式添加。
+    alias_normalized 为 normalize(别名)后的稳定幂等键,用作 PK 避免重复插入。
+    """
+    __tablename__ = "team_aliases"
+
+    # normalize(别名)后的值,稳定幂等,用作主键
+    alias_normalized: Mapped[str] = mapped_column(String(120), primary_key=True)
+    team_id: Mapped[int] = mapped_column(ForeignKey("teams.id", ondelete="CASCADE"), nullable=False)
+    original_alias: Mapped[str] = mapped_column(String(120), nullable=False)  # 原始写法(保留供参考)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_utcnow)
+
+
 class Match(Base):
     __tablename__ = "matches"
 
@@ -324,6 +340,35 @@ class RawEvent(Base):
     )
 
 
+class IngestJob(Base):
+    """采集任务状态:跟踪每次触后台采集任务的执行进度与结果。
+
+    POST /api/v1/ingest/bzzoiro 触发时写入(pending→running→success/failed),
+    前端 Collection 页据此轮询到终态,替代此前"30 秒后盲标完成"的模拟。
+    分批 get_uow / BzzoiroSource / IngestFailure / Bronze/Lineage 均不受影响
+    (本表仅作状态追踪,不介入采集事务)。
+    """
+    __tablename__ = "ingest_jobs"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True)  # uuid4
+    task: Mapped[str] = mapped_column(String(20), nullable=False)
+    params: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    status: Mapped[str] = mapped_column(String(20), nullable=False, server_default="pending")
+    result: Mapped[dict | None] = mapped_column(JSONB)
+    error: Mapped[str | None] = mapped_column(Text)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    __table_args__ = (
+        Index("ix_ingest_job_status_created", "status", "created_at"),
+        CheckConstraint(
+            "status IN ('pending', 'running', 'success', 'failed')",
+            name="ck_ingest_job_status",
+        ),
+    )
+
+
 class IngestFailure(Base):
     """采集失败死信:记录失败原因、重试次数与下次重试时间。
 

src/db/repositories.py

@@ -80,6 +80,20 @@ class MatchRepository:
         )
         return (await self._session.execute(stmt)).scalars().all()
 
+    async def find_by_source_event_id(self, source_event_id: int) -> Match | None:
+        """按上游 event id 查找比赛(唯一命中,用于 upsert 优先路径)。
+
+        source_event_id 上有 partial unique 索引(WHERE IS NOT NULL),
+        同联赛同主客同天(自然键)与上游 event_id 共同保障同一场比赛
+        重复采集时 upsert 而非插入重复行。
+        """
+        stmt = (
+            select(Match)
+            .options(selectinload(Match.stats))
+            .where(Match.source_event_id == source_event_id)
+        )
+        return (await self._session.execute(stmt)).scalar_one_or_none()
+
     async def find_finished_with_stats(self, league_ids: list[int], *, limit: int) -> list[Match]:
         """已完赛且有上游 event id 的比赛(按日期倒序),供统计回填逐场拉取。
 
@@ -114,24 +128,59 @@ class TeamRepository:
     async def get_or_create(self, name: str, *, name_zh: str | None = None) -> Team:
         """按名获取球队,不存在则创建(name_zh 供 bzzoiro 管线写中文名)。
 
-        归一化咽喉:所有入库 Team.name 必须经过 team_names.normalize,
+        归一化咽喉 + 别名查找,三步定位:
-        此处统一收敛,避免各调用点散落归一化逻辑导致重复 Team。
+          1) normalize(name) → 查 teams.name
+          2) 查 team_aliases(以 normalize(name) 为幂等键)→ 复用已映射的 teams.id
+          3) 都没有 → insert 新 Team(归一名)
         创建新 Team 时 info 打出原始名与归一后的规范名,便于排查重名。
+        不自动合并历史重复队;需显式添加别名。
         """
         from src.data.team_names import normalize as normalize_name
+        from src.db.models import TeamAlias
 
         normalized = normalize_name(name) or name.strip()
+
+        # 1) 归一名直查 teams
         team = await self.get_by_name(normalized)
-        if team is None:
+        if team is not None:
-            logger.info(
+            return team
-                "创建新 Team: %s -> %s",
+
-                name, normalized,
+        # 2) 别名查找:normalize(别名) 作为幂等键,命中即复用已有 Team
-            )
+        alias = await self._session.get(TeamAlias, normalized)
-            team = Team(name=normalized, name_zh=name_zh)
+        if alias is not None:
-            self._session.add(team)
+            team = await self._session.get(Team, alias.team_id)
-            await self._session.flush()
+            if team is not None:
+                logger.info("Team 别名命中: %s -> %s(已有 id=%s)", name, normalized, team.id)
+                return team
+
+        # 3) 新建 Team(归一名)
+        logger.info("创建新 Team: %s -> %s", name, normalized)
+        team = Team(name=normalized, name_zh=name_zh)
+        self._session.add(team)
+        await self._session.flush()
         return team
 
+    async def add_alias(self, alias: str, team_id: int) -> TeamAlias:
+        """为已有 Team 添加别名。
+
+        幂等:以 normalize(alias) 为 PK,重复添加同一别名会 upsert。
+        不自动合并历史重复队,仅建立别名映射。
+        """
+        from src.data.team_names import normalize as normalize_name
+        from src.db.models import TeamAlias
+
+        normalized = normalize_name(alias) or alias.strip()
+        existing = await self._session.get(TeamAlias, normalized)
+        if existing is not None:
+            existing.team_id = team_id  # 允许重新指向
+            existing.original_alias = alias
+            await self._session.flush()
+            return existing
+        row = TeamAlias(alias_normalized=normalized, team_id=team_id, original_alias=alias)
+        self._session.add(row)
+        await self._session.flush()
+        return row
+
     async def get_all_by_names(self, names: list[str]) -> dict[str, Team]:
         """批量获取球队,返回 name → Team 映射。"""
         if not names:

src/llm/context_builder.py

@@ -1,9 +1,9 @@
-"""上下文构建器:数据切片 + 拼接。
+"""上下文构建器:数据切片 + 拼接(聚合门面)。
 
-架构:
+实现按 slice 拆分(单文件 → slices 包),本模块只做再导出:
-    - match_header: 比赛基础信息(对阵双方/联赛/时间)
+    - 切片函数: 每个领域 agent 一个数据切片 → src/llm/slices/{form,h2h,stats,home_away,standings}.py
-    - 切片函数: 每个领域 agent 一个数据切片(h2h / form / standings / stats)
+    - 共享类型/头信息/查询助手                              → src/llm/slices/common.py
-    - build_context: 单 agent 路径,拼接全部切片(行为与旧版一致)
+    - build_context: 单 agent 路径,拼接全部切片(行为与旧版一致) → src/llm/slices/aggregate.py
 
 multi-agent 路径由 agents/orchestrator.py 调用切片函数,每个专家只拿自己的切片。
 
@@ -11,548 +11,33 @@ multi-agent 路径由 agents/orchestrator.py 调用切片函数,每个专家只
     build_context 创建一个共享 session 并传给所有切片函数,
     避免每个切片独立创建 session —— 回测 20 场并发时,
     5 个切片 × 20 场 = 100 个连接会耗尽连接池(pool_size=15)。
+
+消费方(路由/orchestrator/tests)仍从本模块 import,签名与拆分前完全一致。
 """
 from __future__ import annotations
 
-import logging
+# ── 共享类型与基础(判空/赛果/统计可用性) ──
-from dataclasses import dataclass
+from src.llm.slices.common import (  # noqa: F401
-from typing import TYPE_CHECKING
+    MatchContext,
-
+    MatchHeader,
-from sqlalchemy import select
+    SliceResult,
-from sqlalchemy.orm import selectinload
+    _is_stats_available,
-
+    _outcome,
-from src.db.base import AsyncSessionLocal
+    header_text,
-from src.db.models import Match
+    load_match_header,
-
+)
-if TYPE_CHECKING:
+
-    from sqlalchemy.ext.asyncio import AsyncSession
+# ── 领域切片 ──
-
+from src.llm.slices.form import form_slice  # noqa: F401
-logger = logging.getLogger(__name__)
+from src.llm.slices.h2h import h2h_slice  # noqa: F401
-
+from src.llm.slices.home_away import home_away_slice  # noqa: F401
-
+from src.llm.slices.standings import standings_slice  # noqa: F401
-def _outcome(home_goals: int, away_goals: int, side: str) -> str:
+from src.llm.slices.stats import stats_slice  # noqa: F401
-    """从某队视角看赛果: W/D/L。"""
+
-    if home_goals is None or away_goals is None:
+# ── 底层查询助手(切片函数共用;orchestrator/tests 直接引用) ──
-        return "?"
+from src.llm.slices.form import _get_form  # noqa: F401
-    if side == "home":
+from src.llm.slices.h2h import _get_h2h  # noqa: F401
-        return "W" if home_goals > away_goals else ("D" if home_goals == away_goals else "L")
+from src.llm.slices.home_away import _get_home_away  # noqa: F401
-    return "W" if away_goals > home_goals else ("D" if away_goals == home_goals else "L")
+
-
+# ── 单 agent 聚合入口 ──
-
+from src.llm.slices.aggregate import build_context  # noqa: F401
-def _is_stats_available(stats, before) -> bool:
-    """检查统计数据在 cutoff 时间是否已可用。
-
-    available_at 语义:该条统计「对外可被使用」的最早时间,
-    至少不得早于比赛结束。用于回测防泄漏。
-
-    规则:
-      - before is None(实盘):available_at 为 None 时允许(兼容旧数据)
-      - before is not None(回测):available_at 为 None 视为不可用(保守)
-      - available_at > cutoff:不可用(数据在 cutoff 之后才生成)
-    """
-    if before is None:
-        # 实盘模式:无时间信息时允许(兼容旧数据)
-        return True
-    # 回测模式(cutoff 不为 None):
-    # available_at 为 None → 无法确认是否在 cutoff 前可用,保守视为不可用
-    if stats.available_at is None:
-        return False
-    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
-class MatchContext:
-    match_id: int
-    text: str
-    has_stats: bool
-    has_standings: bool
-    match_dt: object | None = None  # 比赛时间(回测防泄漏 + 快照用)
-    cutoff: object | None = None  # 实际使用的数据截止时间(用于落库记录)
-
-
-@dataclass
-class MatchHeader:
-    """比赛基础信息(所有 agent 共享)。"""
-    match_id: int
-    home_name: str
-    away_name: str
-    league_name: str
-    season: str | None
-    match_date: str
-    match_dt: object  # 原始 datetime,回测防泄漏用
-    stage: str | None
-    home_team_id: int
-    away_team_id: int
-    league_id: int
-
-
-async def load_match_header(match_id: int, db: AsyncSession | None = None) -> MatchHeader:
-    """加载比赛头信息(各 agent 共用)。
-
-    Args:
-        match_id: 比赛 ID
-        db: 可选的共享 session。不传则自建(向后兼容)。
-    """
-    if db is not None:
-        m = await _load_match(db, match_id)
-        return _to_header(m)
-    async with AsyncSessionLocal() as new_db:
-        m = await _load_match(new_db, match_id)
-        return _to_header(m)
-
-
-def _to_header(m: Match) -> MatchHeader:
-    return MatchHeader(
-        match_id=m.id,
-        home_name=m.home_team.name_zh or m.home_team.name,
-        away_name=m.away_team.name_zh or m.away_team.name,
-        league_name=m.league.name if m.league else "?",
-        season=m.season,
-        match_date=m.match_date.strftime("%Y-%m-%d %H:%M UTC") if m.match_date else "?",
-        match_dt=m.match_date,
-        stage=m.match_stage,
-        home_team_id=m.home_team_id,
-        away_team_id=m.away_team_id,
-        league_id=m.league_id,
-    )
-
-
-def header_text(h: MatchHeader) -> str:
-    stage = f" {h.stage}" if h.stage else ""
-    return (
-        f"对阵: {h.home_name} vs {h.away_name} | {h.league_name} {h.season or '?'}{stage} | {h.match_date}"
-    )
-
-
-# ============================================================
-# 切片函数: 每个领域 agent 一个
-# ============================================================
-
-async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None, db: AsyncSession | None = None) -> SliceResult:
-    """E - 历史交锋切片: 过去数年 + 近期交手数据,提取交手规律。before=match_date 用于回测。
-
-    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
-    """
-    if db is not None:
-        h2h = await _get_h2h(db, header.home_team_id, header.away_team_id, before=before, limit=limit)
-    else:
-        async with AsyncSessionLocal() as new_db:
-            h2h = await _get_h2h(new_db, header.home_team_id, header.away_team_id, before=before, limit=limit)
-    lines = [f"── 历史交锋(近 {limit} 次) ──"]
-    n_with_score = 0
-    if h2h:
-        # 从当前主队视角统计:判断当前主队在每场交锋中是主是客
-        current_home_wins = current_home_draws = current_home_losses = 0
-        for hm in h2h:
-            d = hm.match_date.strftime("%Y-%m") if hm.match_date else "?"
-            if hm.home_goals is not None:
-                n_with_score += 1
-                # 判断当前主队当时是主队还是客队
-                if hm.home_team_id == header.home_team_id:
-                    # 当前主队当时是主队
-                    if hm.home_goals > hm.away_goals:
-                        current_home_wins += 1
-                    elif hm.home_goals == hm.away_goals:
-                        current_home_draws += 1
-                    else:
-                        current_home_losses += 1
-                else:
-                    # 当前主队当时是客队(从客队视角看赛果)
-                    if hm.away_goals > hm.home_goals:
-                        current_home_wins += 1
-                    elif hm.away_goals == hm.home_goals:
-                        current_home_draws += 1
-                    else:
-                        current_home_losses += 1
-                lines.append(f"  {d}: {hm.home_team.name} {hm.home_goals}-{hm.away_goals} {hm.away_team.name}")
-            else:
-                lines.append(f"  {d}: {hm.home_team.name} vs {hm.away_team.name} (无比分)")
-        total = current_home_wins + current_home_draws + current_home_losses
-        if total:
-            lines.append(
-                f"  总计 {total} 场(从当前主队 {header.home_name} 视角): "
-                f"{current_home_wins}胜 {current_home_draws}平 {current_home_losses}负"
-            )
-    else:
-        lines.append("  无数据")
-    # 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, db: AsyncSession | None = None) -> SliceResult:
-    """A - 近期状态切片: 两队近 N 场赛果、关键事件、走势判断。before=match_date 用于回测。
-
-    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
-    """
-    if db is not None:
-        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)
-    else:
-        async with AsyncSessionLocal() as new_db:
-            home_form = await _get_form(new_db, header.home_team_id, before=before, limit=limit)
-            away_form = await _get_form(new_db, header.away_team_id, before=before, limit=limit)
-    lines = []
-    n_scored = 0
-    # P0-1 修复:每场历史比赛必须根据「该队当时是主是客」判断 side,
-    # 不能用本场 side 硬套 —— 否则客场输球会被算成主场赢球。
-    for label, name, form, team_id in (
-        ("主队", header.home_name, home_form, header.home_team_id),
-        ("客队", header.away_name, away_form, header.away_team_id),
-    ):
-        lines.append(f"── {label}近况({name},近 {limit} 场) ──")
-        if form:
-            wins = draws = losses = 0
-            for fm in form:
-                is_home = (fm.home_team_id == team_id)
-                side = "home" if is_home else "away"
-                o = _outcome(fm.home_goals, fm.away_goals, side)
-                if o == "W": wins += 1
-                elif o == "D": draws += 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"
-                xg = ""
-                if fm.stats and _is_stats_available(fm.stats, before) and fm.stats.home_xg is not None:
-                    own = fm.stats.home_xg if is_home else fm.stats.away_xg
-                    xg = f" (xG {own:.1f})"
-                opp = fm.away_team.name if is_home else fm.home_team.name
-                lines.append(f"  {o} {score} vs {opp}{xg}")
-            lines.append(f"  近 {len(form)} 场: {wins}胜 {draws}平 {losses}负")
-        else:
-            lines.append("  无数据")
-    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, db: AsyncSession | None = None) -> SliceResult:
-    """B - 攻防数据切片: 进球、射门、控球,评估攻防强度。before=match_date 用于回测。
-
-    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
-    """
-    if db is not None:
-        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)
-    else:
-        async with AsyncSessionLocal() as new_db:
-            home_form = await _get_form(new_db, header.home_team_id, before=before, limit=limit)
-            away_form = await _get_form(new_db, header.away_team_id, before=before, limit=limit)
-    lines = [f"── 攻防数据(近 {limit} 场) ──"]
-    n_total = 0
-    # P0-1 修复:每场历史比赛必须根据「该队当时是主是客」判断 side,
-    # 不能用本场 side 硬套 —— 否则进球/失球/xG 全部算反。
-    for label, name, form, team_id in (
-        ("主队", header.home_name, home_form, header.home_team_id),
-        ("客队", header.away_name, away_form, header.away_team_id),
-    ):
-        if form:
-            gf = ga = shots = sot = poss = xg = xga = 0
-            n = n_shots = n_poss = n_xg = 0
-            for fm in form:
-                if fm.home_goals is None: continue
-                is_home = (fm.home_team_id == team_id)
-                gf += fm.home_goals if is_home else fm.away_goals
-                ga += fm.away_goals if is_home else fm.home_goals
-                n += 1
-                # 只使用 cutoff 之前已可用的统计数据
-                if fm.stats and _is_stats_available(fm.stats, before):
-                    if fm.stats.home_shots is not None:
-                        shots += fm.stats.home_shots if is_home else fm.stats.away_shots
-                        sot += fm.stats.home_shots_on_target if is_home else fm.stats.away_shots_on_target
-                        n_shots += 1
-                    if fm.stats.home_possession is not None:
-                        poss += fm.stats.home_possession if is_home else (100 - fm.stats.home_possession)
-                        n_poss += 1
-                    if fm.stats.home_xg is not None:
-                        xg += fm.stats.home_xg if is_home else fm.stats.away_xg
-                        xga += fm.stats.away_xg if is_home else fm.stats.home_xg
-                        n_xg += 1
-            n_total += n
-            if n > 0:
-                lines.append(f"  {label} {name}:")
-                lines.append(f"    场均进球 {gf/n:.2f}, 场均失球 {ga/n:.2f}")
-                if n_shots: lines.append(f"    场均射门 {shots/n_shots:.1f}, 射正 {sot/n_shots:.1f}")
-                if n_poss: lines.append(f"    平均控球 {poss/n_poss:.1f}%")
-                if n_xg: lines.append(f"    场均 xG {xg/n_xg:.2f}, 场均被 xG {xga/n_xg:.2f}")
-            else:
-                lines.append(f"  {label} {name}: 无比分数据")
-        else:
-            lines.append(f"  {label} {name}: 无数据")
-    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, db: AsyncSession | None = None) -> SliceResult:
-    """C - 主客因素切片: 主场战绩 vs 客场战绩,评估地理优势影响。before=match_date 用于回测。
-
-    db: 可选共享 session,避免每个切片独立建连(见模块 docstring)。
-    """
-    if db is not None:
-        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)
-    else:
-        async with AsyncSessionLocal() as new_db:
-            home_home = await _get_home_away(new_db, header.home_team_id, "home", before=before, limit=limit)
-            away_away = await _get_home_away(new_db, header.away_team_id, "away", before=before, limit=limit)
-    lines = ["── 主客因素 ──"]
-    n_total = 0
-    for label, name, matches, side in (
-        ("主队主场", header.home_name, home_home, "home"),
-        ("客队客场", header.away_name, away_away, "away"),
-    ):
-        if matches:
-            wins = draws = losses = gf = ga = 0
-            for m in matches:
-                if m.home_goals is None: continue
-                o = _outcome(m.home_goals, m.away_goals, side)
-                if o == "W": wins += 1
-                elif o == "D": draws += 1
-                else: losses += 1
-                gf += m.home_goals if side == "home" else m.away_goals
-                ga += m.away_goals if side == "home" else m.home_goals
-            n = wins + draws + losses
-            n_total += n
-            if n > 0:
-                pct = wins / n * 100
-                lines.append(f"  {label} {name}(近 {n} 场): {wins}胜 {draws}平 {losses}负, 胜率 {pct:.0f}%")
-                lines.append(f"    场均进球 {gf/n:.2f}, 场均失球 {ga/n:.2f}")
-            else:
-                lines.append(f"  {label} {name}: 无比分数据")
-        else:
-            lines.append(f"  {label} {name}: 无数据")
-    return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)
-
-
-async def standings_slice(header: MatchHeader, *, before=None, db: AsyncSession | None = None) -> SliceResult:
-    """D - 联赛排名切片: 两队积分榜位置、积分、近期走势(form)、分区,评估整体实力差距。
-
-    before 参数保留与其他切片一致的签名(积分榜是最新快照,无历史版本,不受 cutoff 影响)。
-    db: 可选共享 session(见模块 docstring)。
-
-    语义区分:
-      - 两队都有积分榜行 → has_data=True(明确的排名信息)
-      - 任一队缺失 → has_data=False(升班马/杯赛无榜,信息不完整时明确声明)
-    """
-    from src.db.models import League, Standing
-
-    if db is not None:
-        league = (await db.execute(select(League).where(League.id == header.league_id))).scalar_one_or_none()
-        rows = (
-            (
-                await db.execute(
-                    select(Standing)
-                    .options(selectinload(Standing.team))
-                    .where(Standing.league_id == header.league_id)
-                    .order_by(Standing.position.asc())
-                )
-            )
-            .scalars()
-            .all()
-            if league
-            else []
-        )
-    else:
-        async with AsyncSessionLocal() as new_db:
-            return await standings_slice(header, before=before, db=new_db)
-
-    lines = [f"── 联赛排名({header.league_name} 共 {len(rows)} 队) ──"]
-    n_records = 0
-
-    def _fmt(row) -> str:
-        zg = f" xG差 {row.xgd:+.1f}" if row.xg_for is not None and row.xg_against is not None and row.goal_diff is not None else ""
-        form = f" 近5场 {row.form}" if row.form else ""
-        zone = f" [{row.zone}]" if row.zone else ""
-        return (
-            f"    第 {row.position} 名: {row.points} 分 / {row.played} 场 "
-            f"({row.won}胜{row.drawn}平{row.lost}负, 进{row.goals_for}失{row.goals_against} 净胜{row.goal_diff:+d}"
-            f"{zg}){form}{zone}"
-        )
-
-    for label, team_id in (("主队", header.home_team_id), ("客队", header.away_team_id)):
-        row = next((r for r in rows if r.team_id == team_id), None)
-        if row is None:
-            lines.append(f"  {label}: 暂无积分榜数据(可能杯赛/赛季未开始)")
-        else:
-            n_records += 1
-            lines.append(f"  {label} {header.home_name if label == '主队' else header.away_name}:")
-            lines.append(_fmt(row))
-
-    # 两队排名对比摘要
-    home_row = next((r for r in rows if r.team_id == header.home_team_id), None)
-    away_row = next((r for r in rows if r.team_id == header.away_team_id), None)
-    if home_row and away_row:
-        diff = home_row.position - away_row.position  # 正数=主队排名更靠前(名次更小)
-        lead = f"主队排名高 {diff} 位" if diff > 0 else (f"客队排名高 {-diff} 位" if diff < 0 else "两队同排名结构")
-        pts_diff = home_row.points - away_row.points
-        lines.append(f"  排名对比: {lead}, 分差 {pts_diff:+d}")
-
-    # has_data: 两队都有行才算完整;只有一队时仍有价值,但标记不完整
-    has_data = n_records >= 1
-    return SliceResult(text="\n".join(lines), has_data=has_data, n_records=n_records)
-
-
-# ============================================================
-# 单 agent 路径: 拼接全部切片(行为与旧版一致)
-# ============================================================
-
-async def build_context(match_id: int, *, form_last: int = 5, h2h_last: int = 5, backtest: bool = False, cutoff_at=None) -> MatchContext:
-    """单 agent 路径的完整上下文: 拼接全部切片(before=cutoff,防未来信息)。
-
-    has_stats / has_standings 直接取切片显式声明的 has_data,
-    不再靠文案子串匹配(见审查报告 P2-1)。
-
-    P2-6: backtest=True 时 cutoff = match_date - 1天,确保只用赛前数据。
-    cutoff_at: 显式截止时间(优先于 backtest 自动计算)。
-
-    P1-1: 使用单个共享 session 贯穿所有切片查询,避免连接池耗尽。
-    """
-    async with AsyncSessionLocal() as db:
-        header = await load_match_header(match_id, db=db)
-        # 计算数据截止时间: 显式 > backtest 自动计算 > 默认(比赛时间)
-        if cutoff_at is not None:
-            cutoff = cutoff_at
-        elif backtest and header.match_dt:
-            from datetime import timedelta
-            cutoff = header.match_dt - timedelta(days=1)
-        else:
-            cutoff = header.match_dt
-        parts = [header_text(header), ""]
-
-        form_res = await form_slice(header, limit=form_last, before=cutoff, db=db)
-        parts.append(form_res.text)
-        parts.append("")
-
-        h2h_res = await h2h_slice(header, limit=h2h_last, before=cutoff, db=db)
-        parts.append(h2h_res.text)
-        parts.append("")
-
-        stats_res = await stats_slice(header, before=cutoff, db=db)
-        parts.append(stats_res.text)
-        parts.append("")
-
-        home_away_res = await home_away_slice(header, before=cutoff, db=db)
-        parts.append(home_away_res.text)
-        parts.append("")
-
-        standings_res = await standings_slice(header, before=cutoff, db=db)
-        parts.append(standings_res.text)
-
-        return MatchContext(
-            match_id=match_id,
-            text="\n".join(parts),
-            has_stats=form_res.has_data or stats_res.has_data,
-            has_standings=standings_res.has_data,
-            match_dt=header.match_dt,
-            cutoff=cutoff,
-        )
-
-
-# ============================================================
-# 底层查询(切片函数共用)
-# ============================================================
-
-async def _load_match(db, match_id: int) -> Match:
-    stmt = (
-        select(Match)
-        .where(Match.id == match_id)
-        .options(
-            selectinload(Match.league),
-            selectinload(Match.home_team),
-            selectinload(Match.away_team),
-            selectinload(Match.stats),
-        )
-    )
-    m = (await db.execute(stmt)).scalar_one_or_none()
-    if m is None:
-        raise ValueError(f"match {match_id} not found")
-    return m
-
-
-async def _get_form(db, team_id: int, before, *, limit: int = 5) -> list[Match]:
-    """某队近 N 场(已完赛)。before=None 表示不限制(预测赛前的场景由调用方保证)。
-
-    必须预加载 stats / home_team / away_team:切片函数会读取这些关系,
-    而 async session 下惰性加载会抛 MissingGreenlet。
-    (models.py 已声明 lazy="selectin",此处显式声明以固化查询意图。)
-    """
-    stmt = (
-        select(Match)
-        .options(
-            selectinload(Match.stats),
-            selectinload(Match.home_team),
-            selectinload(Match.away_team),
-        )
-        .where(Match.match_status == "finished")
-        .where(Match.home_goals.is_not(None))
-        .where((Match.home_team_id == team_id) | (Match.away_team_id == team_id))
-        .order_by(Match.match_date.desc())
-        .limit(limit)
-    )
-    if before is not None:
-        stmt = stmt.where(Match.match_date < before)
-    result = await db.execute(stmt)
-    return list(result.scalars().all())
-
-
-async def _get_h2h(db, home_id: int, away_id: int, before, *, limit: int = 5) -> list[Match]:
-    """两队交锋史。需预加载 home_team / away_team(切片输出队名)。"""
-    stmt = (
-        select(Match)
-        .options(
-            selectinload(Match.home_team),
-            selectinload(Match.away_team),
-        )
-        .where(Match.match_status == "finished")
-        .where(Match.home_goals.is_not(None))
-        .where(
-            ((Match.home_team_id == home_id) & (Match.away_team_id == away_id))
-            | ((Match.home_team_id == away_id) & (Match.away_team_id == home_id))
-        )
-        .order_by(Match.match_date.desc())
-        .limit(limit)
-    )
-    if before is not None:
-        stmt = stmt.where(Match.match_date < before)
-    result = await db.execute(stmt)
-    return list(result.scalars().all())
-
-
-async def _get_home_away(db, team_id: int, side: str, before, *, limit: int = 10) -> list[Match]:
-    """某队主场/客场近 N 场。side='home' 取主场,'away' 取客场。
-
-    当前只用标量字段,但统一预加载以免后续扩展时踩坑。
-    """
-    stmt = (
-        select(Match)
-        .options(
-            selectinload(Match.stats),
-            selectinload(Match.home_team),
-            selectinload(Match.away_team),
-        )
-        .where(Match.match_status == "finished")
-        .where(Match.home_goals.is_not(None))
-        .order_by(Match.match_date.desc())
-        .limit(limit)
-    )
-    if side == "home":
-        stmt = stmt.where(Match.home_team_id == team_id)
-    else:
-        stmt = stmt.where(Match.away_team_id == team_id)
-    if before is not None:
-        stmt = stmt.where(Match.match_date < before)
-    result = await db.execute(stmt)
-    return list(result.scalars().all())

src/llm/predict.py

@@ -1,4 +1,16 @@
-"""预测服务:拼上下文 → 调 LLM → 存预测。"""
+"""预测服务:拼上下文 → 调 LLM → 存预测。
+
+落库层级(table: predictions):
+
+| 模式      | 落库位置(服务层)                                            | 路由层(routes/predict.py) |
+|-----------|-------------------------------------------------------------|---------------------------|
+| single    | `_predict_single` → `_upsert_prediction`                    | 不读 DB,仅映射 result → PredictOut |
+| multi     | `orchestrator.predict_match_multi` → `_upsert_prediction`   | 不读 DB,仅映射 result → PredictOut |
+| baseline  | `predict_baseline` → `_upsert_prediction`                   | 不读 DB,仅映射 result → PredictOut |
+
+三种模式统一在服务层经 UnitOfWork 落库并回填真实 prediction_id;
+路由层永不写入 predictions,只读 result.prediction_id 做响应映射。
+"""
 from __future__ import annotations
 
 import functools
@@ -25,9 +37,7 @@ _PROMPT_DIR = Path(__file__).resolve().parent / "prompts"
 # ── LLM 响应缓存(match+provider+model+version → 结果) ──
 _CACHE_TTL_SEC = 300  # 5 分钟
 _CACHE_MAX_SIZE = 200  # P3-1: 有上限,避免长期运行内存无限增长
-# P1-5: 缓存仅在 asyncio 协程内同步访问(dict 操作 GIL 原子),无需 threading.Lock。
+_CACHE_PREFIX = "predict:"  # Redis key 前缀
-# 删除 _cache_lock,避免同步锁阻塞事件循环;dict 的 get/set 在 CPython 下原子。
-_cache: dict[str, tuple[float, PredictResult]] = {}
 
 
 def _cache_key(match_id: int, provider: str, model: str, version: str, tpl_hash: str) -> str:
@@ -36,30 +46,130 @@ def _cache_key(match_id: int, provider: str, model: str, version: str, tpl_hash:
     仅用 version 做键不够 —— 编辑器里改动 `match_prediction_v1.md` 而版本号
     不变时,进程内缓存仍会返回旧模板产生的旧结果(见审查报告 P2-6)。
     把模板内容 hash 纳入键,模板一改缓存自动失效。
+    内存与 Redis 共用同一键格式,TTL 一致。
     """
-    return f"{match_id}:{provider}:{model}:{version}:{tpl_hash[:12]}"
+    return f"{_CACHE_PREFIX}{match_id}:{provider}:{model}:{version}:{tpl_hash[:12]}"
 
 
-def _get_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str) -> PredictResult | None:
+# ── 缓存后端:内存(LRU+TTL),可选 Redis ──────────────────────────
-    # P1-5: 无锁访问。dict get/del 在 CPython GIL 下原子,且无 await 穿插。
+class _CacheBackend:
+    """缓存后端统一接口:_get 同步返回(命中时),_set 异步(Redis 为 async,内存同步)。"""
+
+    def _raw_key(self, key: str) -> str:
+        return key
+
+    def get(self, key: str) -> PredictResult | None:
+        raise NotImplementedError
+
+    async def set(self, key: str, result: PredictResult, ttl: int) -> None:
+        raise NotImplementedError
+
+
+class _MemoryCache(_CacheBackend):
+    """进程内 LRU+TTL 缓存(默认后端)。"""
+
+    def __init__(self) -> None:
+        self._store: dict[str, tuple[float, PredictResult]] = {}
+
+    def get(self, key: str) -> PredictResult | None:
+        entry = self._store.get(key)
+        if entry is not None:
+            ts, result = entry
+            if time.time() - ts < _CACHE_TTL_SEC:
+                return result
+            self._store.pop(key, None)
+        return None
+
+    async def set(self, key: str, result: PredictResult, ttl: int) -> None:
+        self._store[key] = (time.time(), result)
+        if len(self._store) > _CACHE_MAX_SIZE:
+            oldest_key = min(self._store, key=lambda k: self._store[k][0])
+            self._store.pop(oldest_key, None)
+
+
+class _RedisCache(_CacheBackend):
+    """可选 Redis 后端: PREDICT_CACHE_URL 非空时启用。
+
+    失败降级:读失败返回 None(跳过缓存),写失败打 warning;不中断预测主流程。
+    不强制依赖 redis 包——未安装时启动回退内存并 warning。
+    """
+
+    def __init__(self, url: str) -> None:
+        self._url = url
+        self._redis = None  # type: ignore[var-annotated]
+        self._memory_fallback = _MemoryCache()
+        self._available: bool | None = None  # None=未探测,True=可用,False=不可用
+
+    async def _ensure_conn(self) -> bool:
+        """懒初始化 Redis 连接;失败返回 False 并降级内存。"""
+        if self._available is not None:
+            return self._available
+        try:
+            from redis.asyncio import Redis
+
+            self._redis = Redis.from_url(self._url, decode_responses=True, socket_timeout=2.0)
+            await self._redis.ping()
+            self._available = True
+            logger.info("predict cache: Redis 后端已连接 %s", self._url.replace(self._url.split("@")[-1] if "@" in self._url else self._url, "***") if "://" in self._url else "redis")
+        except Exception as e:
+            self._available = False
+            logger.warning("predict cache: Redis 连接失败(%s),降级内存缓存", e)
+        return self._available
+
+    def get(self, key: str) -> PredictResult | None:
+        # Redis get 是 async 的,此处统一由调用方走 async 路径;
+        # 同步 get 仅用于不可降级场景——Redis 模式下直接返回 None,
+        # 实际读取通过 get_async 完成。
+        return None
+
+    async def get_async(self, key: str) -> PredictResult | None:
+        if not await self._ensure_conn():
+            return self._memory_fallback.get(key)
+        try:
+            import pickle
+
+            raw = await self._redis.get(key)  # type: ignore[union-attr]
+            if raw is None:
+                return None
+            return pickle.loads(raw.encode("latin-1")) if isinstance(raw, str) else pickle.loads(raw)
+        except Exception as e:
+            logger.warning("predict cache: Redis GET 失败(%s),跳过缓存", e)
+            return None
+
+    async def set(self, key: str, result: PredictResult, ttl: int) -> None:
+        if not await self._ensure_conn():
+            await self._memory_fallback.set(key, result, ttl)
+            return
+        try:
+            import pickle
+
+            payload = pickle.dumps(result).decode("latin-1")
+            await self._redis.set(key, payload, ex=ttl)  # type: ignore[union-attr]
+        except Exception as e:
+            logger.warning("predict cache: Redis SET 失败(%s),降级内存写入", e)
+            await self._memory_fallback.set(key, result, ttl)
+
+
+def _build_cache_backend() -> _CacheBackend:
+    url = getattr(settings, "PREDICT_CACHE_URL", None)
+    if url:
+        return _RedisCache(url)
+    return _MemoryCache()
+
+
+_cache_backend: _CacheBackend = _build_cache_backend()
+
+
+async 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, tpl_hash)
-    entry = _cache.get(key)
+    if isinstance(_cache_backend, _RedisCache):
-    if entry is not None:
+        return await _cache_backend.get_async(key)
-        ts, result = entry
+    return _cache_backend.get(key)
-        if time.time() - ts < _CACHE_TTL_SEC:
-            return result
-        _cache.pop(key, None)
-    return None
 
 
-def _set_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str, result: PredictResult) -> None:
+async def _set_cached(match_id: int, provider: str, model: str, version: str, tpl_hash: str, result: PredictResult) -> None:
-    # P1-5: 无锁写入。同上,dict set 原子。
     key = _cache_key(match_id, provider, model, version, tpl_hash)
-    _cache[key] = (time.time(), result)
+    await _cache_backend.set(key, result, _CACHE_TTL_SEC)
-    # P3-1: 超过上限时淘汰最旧条目(按时间戳排序)
-    if len(_cache) > _CACHE_MAX_SIZE:
-        oldest_key = min(_cache, key=lambda k: _cache[k][0])
-        _cache.pop(oldest_key, None)
 
 
 def clear_prompt_cache() -> None:
@@ -67,9 +177,12 @@ def clear_prompt_cache() -> None:
 
     lru_cache 的模板缓存是进程级的,改完 .md 需要重启进程才能生效;
     提供显式清理入口,避免"改了模板却看不到变化"的困惑(见审查报告 P2-5)。
+    同时清空预测响应缓存(内存后端);Redis 后端因共享不清除。
     """
     _load_prompt_template.cache_clear()
-    logger.info("prompt 模板缓存已清空")
+    if isinstance(_cache_backend, _MemoryCache):
+        _cache_backend._store.clear()
+    logger.info("prompt 模板缓存 + 预测响应缓存(内存)已清空")
 
 
 @functools.lru_cache(maxsize=8)
@@ -235,7 +348,7 @@ async def _predict_single(
 
     # 0. 查缓存(同 match+provider+model+version+模板hash 5 分钟内直接返)
     if use_cache:
-        cached = _get_cached(match_id, settings.LLM_PROVIDER, provider.model, version, tpl_hash)
+        cached = await _get_cached(match_id, settings.LLM_PROVIDER, provider.model, version, tpl_hash)
         if cached is not None:
             logger.debug("predict cache hit match=%s", match_id)
             return cached
@@ -334,7 +447,7 @@ async def _predict_single(
 
         # 5. 写入缓存(仅当允许缓存时)
         if use_cache:
-            _set_cached(match_id, settings.LLM_PROVIDER, provider.model, version, tpl_hash, result)
+            await _set_cached(match_id, settings.LLM_PROVIDER, provider.model, version, tpl_hash, result)
         logger.info(
             "预测完成 match=%s mode=%s status=%s pred=%s:%s (%s) latency=%sms",
             match_id, "single", "success",

src/llm/slices/__init__.py

@@ -0,0 +1,5 @@
+"""slices 包:按领域拆分的数据切片(form/stats/h2h/home_away/standings)。
+
+对外统一经 src.llm.context_builder 再导出;本包 __init__ 不承载导出,
+保持「context_builder 是唯一公开入口」的 import 约定。
+"""

src/llm/slices/aggregate.py

@@ -0,0 +1,65 @@
+"""单 agent 聚合路径: 拼接全部切片(build_context)。
+
+共享 session 贯穿所有切片(见 context_builder 模块 docstring 的性能说明)。
+"""
+from __future__ import annotations
+
+from src.db.base import AsyncSessionLocal
+from src.llm.slices.common import MatchContext, header_text, load_match_header
+from src.llm.slices.form import form_slice
+from src.llm.slices.h2h import h2h_slice
+from src.llm.slices.home_away import home_away_slice
+from src.llm.slices.standings import standings_slice
+from src.llm.slices.stats import stats_slice
+
+
+async def build_context(match_id: int, *, form_last: int = 5, h2h_last: int = 5, backtest: bool = False, cutoff_at=None) -> MatchContext:
+    """单 agent 路径的完整上下文: 拼接全部切片(before=cutoff,防未来信息)。
+
+    has_stats / has_standings 直接取切片显式声明的 has_data,
+    不再靠文案子串匹配(见审查报告 P2-1)。
+
+    P2-6: backtest=True 时 cutoff = match_date - 1天,确保只用赛前数据。
+    cutoff_at: 显式截止时间(优先于 backtest 自动计算)。
+
+    P1-1: 使用单个共享 session 贯穿所有切片查询,避免连接池耗尽。
+    """
+    async with AsyncSessionLocal() as db:
+        header = await load_match_header(match_id, db=db)
+        # 计算数据截止时间: 显式 > backtest 自动计算 > 默认(比赛时间)
+        if cutoff_at is not None:
+            cutoff = cutoff_at
+        elif backtest and header.match_dt:
+            from datetime import timedelta
+            cutoff = header.match_dt - timedelta(days=1)
+        else:
+            cutoff = header.match_dt
+        parts = [header_text(header), ""]
+
+        form_res = await form_slice(header, limit=form_last, before=cutoff, db=db)
+        parts.append(form_res.text)
+        parts.append("")
+
+        h2h_res = await h2h_slice(header, limit=h2h_last, before=cutoff, db=db)
+        parts.append(h2h_res.text)
+        parts.append("")
+
+        stats_res = await stats_slice(header, before=cutoff, db=db)
+        parts.append(stats_res.text)
+        parts.append("")
+
+        home_away_res = await home_away_slice(header, before=cutoff, db=db)
+        parts.append(home_away_res.text)
+        parts.append("")
+
+        standings_res = await standings_slice(header, before=cutoff, db=db)
+        parts.append(standings_res.text)
+
+        return MatchContext(
+            match_id=match_id,
+            text="\n".join(parts),
+            has_stats=form_res.has_data or stats_res.has_data,
+            has_standings=standings_res.has_data,
+            match_dt=header.match_dt,
+            cutoff=cutoff,
+        )

src/llm/slices/common.py

@@ -0,0 +1,149 @@
+"""切片共享基础:结果类型 / 比赛头信息 / 赛果与统计可用性判定。
+
+从 context_builder.py 按领域拆出(单文件 → slices 包),仅做搬迁无逻辑修改。
+各领域切片见同包 form/h2h/stats/home_away/standings 模块;
+聚合入口 build_context 见 aggregate.py;对外统一经 context_builder 再导出。
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from src.db.base import AsyncSessionLocal
+from src.db.models import Match
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+
+def _outcome(home_goals: int, away_goals: int, side: str) -> str:
+    """从某队视角看赛果: W/D/L。"""
+    if home_goals is None or away_goals is None:
+        return "?"
+    if side == "home":
+        return "W" if home_goals > away_goals else ("D" if home_goals == away_goals else "L")
+    return "W" if away_goals > home_goals else ("D" if away_goals == home_goals else "L")
+
+
+def _is_stats_available(stats, before) -> bool:
+    """检查统计数据在 cutoff 时间是否已可用。
+
+    available_at 语义:该条统计「对外可被使用」的最早时间,
+    至少不得早于比赛结束。用于回测防泄漏。
+
+    规则:
+      - before is None(实盘):available_at 为 None 时允许(兼容旧数据)
+      - before is not None(回测):available_at 为 None 视为不可用(保守)
+      - available_at > cutoff:不可用(数据在 cutoff 之后才生成)
+    """
+    if before is None:
+        # 实盘模式:无时间信息时允许(兼容旧数据)
+        return True
+    # 回测模式(cutoff 不为 None):
+    # available_at 为 None → 无法确认是否在 cutoff 前可用,保守视为不可用
+    if stats.available_at is None:
+        return False
+    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
+class MatchContext:
+    match_id: int
+    text: str
+    has_stats: bool
+    has_standings: bool
+    match_dt: object | None = None  # 比赛时间(回测防泄漏 + 快照用)
+    cutoff: object | None = None  # 实际使用的数据截止时间(用于落库记录)
+
+
+@dataclass
+class MatchHeader:
+    """比赛基础信息(所有 agent 共享)。"""
+    match_id: int
+    home_name: str
+    away_name: str
+    league_name: str
+    season: str | None
+    match_date: str
+    match_dt: object  # 原始 datetime,回测防泄漏用
+    stage: str | None
+    home_team_id: int
+    away_team_id: int
+    league_id: int
+
+
+async def _load_match(db, match_id: int) -> Match:
+    stmt = (
+        select(Match)
+        .where(Match.id == match_id)
+        .options(
+            selectinload(Match.league),
+            selectinload(Match.home_team),
+            selectinload(Match.away_team),
+            selectinload(Match.stats),
+        )
+    )
+    m = (await db.execute(stmt)).scalar_one_or_none()
+    if m is None:
+        raise ValueError(f"match {match_id} not found")
+    return m
+
+
+async def load_match_header(match_id: int, db: AsyncSession | None = None) -> MatchHeader:
+    """加载比赛头信息(各 agent 共用)。
+
+    Args:
+        match_id: 比赛 ID
+        db: 可选的共享 session。不传则自建(向后兼容)。
+    """
+    if db is not None:
+        m = await _load_match(db, match_id)
+        return _to_header(m)
+    async with AsyncSessionLocal() as new_db:
+        m = await _load_match(new_db, match_id)
+        return _to_header(m)
+
+
+def _to_header(m: Match) -> MatchHeader:
+    return MatchHeader(
+        match_id=m.id,
+        home_name=m.home_team.name_zh or m.home_team.name,
+        away_name=m.away_team.name_zh or m.away_team.name,
+        league_name=m.league.name if m.league else "?",
+        season=m.season,
+        match_date=m.match_date.strftime("%Y-%m-%d %H:%M UTC") if m.match_date else "?",
+        match_dt=m.match_date,
+        stage=m.match_stage,
+        home_team_id=m.home_team_id,
+        away_team_id=m.away_team_id,
+        league_id=m.league_id,
+    )
+
+
+def header_text(h: MatchHeader) -> str:
+    stage = f" {h.stage}" if h.stage else ""
+    return (
+        f"对阵: {h.home_name} vs {h.away_name} | {h.league_name} {h.season or '?'}{stage} | {h.match_date}"
+    )

src/llm/slices/form.py

@@ -0,0 +1,85 @@
+"""A - 近期状态切片: 近 N 场赛果 / 走势(form)。"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from src.db.base import AsyncSessionLocal
+from src.db.models import Match
+from src.llm.slices.common import MatchHeader, SliceResult, _is_stats_available, _outcome
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def form_slice(header: MatchHeader, *, limit: int = 5, before=None, db: AsyncSession | None = None) -> SliceResult:
+    """A - 近期状态切片: 两队近 N 场赛果、关键事件、走势判断。before=match_date 用于回测。
+
+    db: 可选共享 session,避免每个切片独立建连(见 context_builder 模块 docstring)。
+    """
+    if db is not None:
+        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)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_form = await _get_form(new_db, header.home_team_id, before=before, limit=limit)
+            away_form = await _get_form(new_db, header.away_team_id, before=before, limit=limit)
+    lines = []
+    n_scored = 0
+    # P0-1 修复:每场历史比赛必须根据「该队当时是主是客」判断 side,
+    # 不能用本场 side 硬套 —— 否则客场输球会被算成主场赢球。
+    for label, name, form, team_id in (
+        ("主队", header.home_name, home_form, header.home_team_id),
+        ("客队", header.away_name, away_form, header.away_team_id),
+    ):
+        lines.append(f"── {label}近况({name},近 {limit} 场) ──")
+        if form:
+            wins = draws = losses = 0
+            for fm in form:
+                is_home = (fm.home_team_id == team_id)
+                side = "home" if is_home else "away"
+                o = _outcome(fm.home_goals, fm.away_goals, side)
+                if o == "W": wins += 1
+                elif o == "D": draws += 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"
+                xg = ""
+                if fm.stats and _is_stats_available(fm.stats, before) and fm.stats.home_xg is not None:
+                    own = fm.stats.home_xg if is_home else fm.stats.away_xg
+                    xg = f" (xG {own:.1f})"
+                opp = fm.away_team.name if is_home else fm.home_team.name
+                lines.append(f"  {o} {score} vs {opp}{xg}")
+            lines.append(f"  近 {len(form)} 场: {wins}胜 {draws}平 {losses}负")
+        else:
+            lines.append("  无数据")
+    return SliceResult(text="\n".join(lines), has_data=n_scored > 0, n_records=n_scored)
+
+
+async def _get_form(db, team_id: int, before, *, limit: int = 5) -> list[Match]:
+    """某队近 N 场(已完赛)。before=None 表示不限制(预测赛前的场景由调用方保证)。
+
+    必须预加载 stats / home_team / away_team:切片函数会读取这些关系,
+    而 async session 下惰性加载会抛 MissingGreenlet。
+    (models.py 已声明 lazy="selectin",此处显式声明以固化查询意图。)
+    """
+    stmt = (
+        select(Match)
+        .options(
+            selectinload(Match.stats),
+            selectinload(Match.home_team),
+            selectinload(Match.away_team),
+        )
+        .where(Match.match_status == "finished")
+        .where(Match.home_goals.is_not(None))
+        .where((Match.home_team_id == team_id) | (Match.away_team_id == team_id))
+        .order_by(Match.match_date.desc())
+        .limit(limit)
+    )
+    if before is not None:
+        stmt = stmt.where(Match.match_date < before)
+    result = await db.execute(stmt)
+    return list(result.scalars().all())

src/llm/slices/h2h.py

@@ -0,0 +1,88 @@
+"""E - 历史交锋切片: 交手史与胜负规律(h2h)。"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from src.db.base import AsyncSessionLocal
+from src.db.models import Match
+from src.llm.slices.common import MatchHeader, SliceResult
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def h2h_slice(header: MatchHeader, *, limit: int = 8, before=None, db: AsyncSession | None = None) -> SliceResult:
+    """E - 历史交锋切片: 过去数年 + 近期交手数据,提取交手规律。before=match_date 用于回测。
+
+    db: 可选共享 session,避免每个切片独立建连(见 context_builder 模块 docstring)。
+    """
+    if db is not None:
+        h2h = await _get_h2h(db, header.home_team_id, header.away_team_id, before=before, limit=limit)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            h2h = await _get_h2h(new_db, header.home_team_id, header.away_team_id, before=before, limit=limit)
+    lines = [f"── 历史交锋(近 {limit} 次) ──"]
+    n_with_score = 0
+    if h2h:
+        # 从当前主队视角统计:判断当前主队在每场交锋中是主是客
+        current_home_wins = current_home_draws = current_home_losses = 0
+        for hm in h2h:
+            d = hm.match_date.strftime("%Y-%m") if hm.match_date else "?"
+            if hm.home_goals is not None:
+                n_with_score += 1
+                # 判断当前主队当时是主队还是客队
+                if hm.home_team_id == header.home_team_id:
+                    # 当前主队当时是主队
+                    if hm.home_goals > hm.away_goals:
+                        current_home_wins += 1
+                    elif hm.home_goals == hm.away_goals:
+                        current_home_draws += 1
+                    else:
+                        current_home_losses += 1
+                else:
+                    # 当前主队当时是客队(从客队视角看赛果)
+                    if hm.away_goals > hm.home_goals:
+                        current_home_wins += 1
+                    elif hm.away_goals == hm.home_goals:
+                        current_home_draws += 1
+                    else:
+                        current_home_losses += 1
+                lines.append(f"  {d}: {hm.home_team.name} {hm.home_goals}-{hm.away_goals} {hm.away_team.name}")
+            else:
+                lines.append(f"  {d}: {hm.home_team.name} vs {hm.away_team.name} (无比分)")
+        total = current_home_wins + current_home_draws + current_home_losses
+        if total:
+            lines.append(
+                f"  总计 {total} 场(从当前主队 {header.home_name} 视角): "
+                f"{current_home_wins}胜 {current_home_draws}平 {current_home_losses}负"
+            )
+    else:
+        lines.append("  无数据")
+    # has_data 以「有比分的交锋」为准:仅有对阵无比分时不足以支撑分析
+    return SliceResult(text="\n".join(lines), has_data=n_with_score > 0, n_records=n_with_score)
+
+
+async def _get_h2h(db, home_id: int, away_id: int, before, *, limit: int = 5) -> list[Match]:
+    """两队交锋史。需预加载 home_team / away_team(切片输出队名)。"""
+    stmt = (
+        select(Match)
+        .options(
+            selectinload(Match.home_team),
+            selectinload(Match.away_team),
+        )
+        .where(Match.match_status == "finished")
+        .where(Match.home_goals.is_not(None))
+        .where(
+            ((Match.home_team_id == home_id) & (Match.away_team_id == away_id))
+            | ((Match.home_team_id == away_id) & (Match.away_team_id == home_id))
+        )
+        .order_by(Match.match_date.desc())
+        .limit(limit)
+    )
+    if before is not None:
+        stmt = stmt.where(Match.match_date < before)
+    result = await db.execute(stmt)
+    return list(result.scalars().all())

src/llm/slices/home_away.py

@@ -0,0 +1,82 @@
+"""C - 主客因素切片: 主场战绩 vs 客场战绩(home_away)。"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from src.db.base import AsyncSessionLocal
+from src.db.models import Match
+from src.llm.slices.common import MatchHeader, SliceResult, _outcome
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def home_away_slice(header: MatchHeader, *, limit: int = 10, before=None, db: AsyncSession | None = None) -> SliceResult:
+    """C - 主客因素切片: 主场战绩 vs 客场战绩,评估地理优势影响。before=match_date 用于回测。
+
+    db: 可选共享 session,避免每个切片独立建连(见 context_builder 模块 docstring)。
+    """
+    if db is not None:
+        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)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_home = await _get_home_away(new_db, header.home_team_id, "home", before=before, limit=limit)
+            away_away = await _get_home_away(new_db, header.away_team_id, "away", before=before, limit=limit)
+    lines = ["── 主客因素 ──"]
+    n_total = 0
+    for label, name, matches, side in (
+        ("主队主场", header.home_name, home_home, "home"),
+        ("客队客场", header.away_name, away_away, "away"),
+    ):
+        if matches:
+            wins = draws = losses = gf = ga = 0
+            for m in matches:
+                if m.home_goals is None: continue
+                o = _outcome(m.home_goals, m.away_goals, side)
+                if o == "W": wins += 1
+                elif o == "D": draws += 1
+                else: losses += 1
+                gf += m.home_goals if side == "home" else m.away_goals
+                ga += m.away_goals if side == "home" else m.home_goals
+            n = wins + draws + losses
+            n_total += n
+            if n > 0:
+                pct = wins / n * 100
+                lines.append(f"  {label} {name}(近 {n} 场): {wins}胜 {draws}平 {losses}负, 胜率 {pct:.0f}%")
+                lines.append(f"    场均进球 {gf/n:.2f}, 场均失球 {ga/n:.2f}")
+            else:
+                lines.append(f"  {label} {name}: 无比分数据")
+        else:
+            lines.append(f"  {label} {name}: 无数据")
+    return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)
+
+
+async def _get_home_away(db, team_id: int, side: str, before, *, limit: int = 10) -> list[Match]:
+    """某队主场/客场近 N 场。side='home' 取主场,'away' 取客场。
+
+    当前只用标量字段,但统一预加载以免后续扩展时踩坑。
+    """
+    stmt = (
+        select(Match)
+        .options(
+            selectinload(Match.stats),
+            selectinload(Match.home_team),
+            selectinload(Match.away_team),
+        )
+        .where(Match.match_status == "finished")
+        .where(Match.home_goals.is_not(None))
+        .order_by(Match.match_date.desc())
+        .limit(limit)
+    )
+    if side == "home":
+        stmt = stmt.where(Match.home_team_id == team_id)
+    else:
+        stmt = stmt.where(Match.away_team_id == team_id)
+    if before is not None:
+        stmt = stmt.where(Match.match_date < before)
+    result = await db.execute(stmt)
+    return list(result.scalars().all())

src/llm/slices/standings.py

@@ -0,0 +1,81 @@
+"""D - 联赛排名切片: 积分榜位置与实力差距(standings)。"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from src.db.base import AsyncSessionLocal
+from src.llm.slices.common import MatchHeader, SliceResult
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def standings_slice(header: MatchHeader, *, before=None, db: AsyncSession | None = None) -> SliceResult:
+    """D - 联赛排名切片: 两队积分榜位置、积分、近期走势(form)、分区,评估整体实力差距。
+
+    before 参数保留与其他切片一致的签名(积分榜是最新快照,无历史版本,不受 cutoff 影响)。
+    db: 可选共享 session(见 context_builder 模块 docstring)。
+
+    语义区分:
+      - 两队都有积分榜行 → has_data=True(明确的排名信息)
+      - 任一队缺失 → has_data=False(升班马/杯赛无榜,信息不完整时明确声明)
+    """
+    from src.db.models import League, Standing
+
+    if db is not None:
+        league = (await db.execute(select(League).where(League.id == header.league_id))).scalar_one_or_none()
+        rows = (
+            (
+                await db.execute(
+                    select(Standing)
+                    .options(selectinload(Standing.team))
+                    .where(Standing.league_id == header.league_id)
+                    .order_by(Standing.position.asc())
+                )
+            )
+            .scalars()
+            .all()
+            if league
+            else []
+        )
+    else:
+        async with AsyncSessionLocal() as new_db:
+            return await standings_slice(header, before=before, db=new_db)
+
+    lines = [f"── 联赛排名({header.league_name} 共 {len(rows)} 队) ──"]
+    n_records = 0
+
+    def _fmt(row) -> str:
+        zg = f" xG差 {row.xgd:+.1f}" if row.xg_for is not None and row.xg_against is not None and row.goal_diff is not None else ""
+        form = f" 近5场 {row.form}" if row.form else ""
+        zone = f" [{row.zone}]" if row.zone else ""
+        return (
+            f"    第 {row.position} 名: {row.points} 分 / {row.played} 场 "
+            f"({row.won}胜{row.drawn}平{row.lost}负, 进{row.goals_for}失{row.goals_against} 净胜{row.goal_diff:+d}"
+            f"{zg}){form}{zone}"
+        )
+
+    for label, team_id in (("主队", header.home_team_id), ("客队", header.away_team_id)):
+        row = next((r for r in rows if r.team_id == team_id), None)
+        if row is None:
+            lines.append(f"  {label}: 暂无积分榜数据(可能杯赛/赛季未开始)")
+        else:
+            n_records += 1
+            lines.append(f"  {label} {header.home_name if label == '主队' else header.away_name}:")
+            lines.append(_fmt(row))
+
+    # 两队排名对比摘要
+    home_row = next((r for r in rows if r.team_id == header.home_team_id), None)
+    away_row = next((r for r in rows if r.team_id == header.away_team_id), None)
+    if home_row and away_row:
+        diff = home_row.position - away_row.position  # 正数=主队排名更靠前(名次更小)
+        lead = f"主队排名高 {diff} 位" if diff > 0 else (f"客队排名高 {-diff} 位" if diff < 0 else "两队同排名结构")
+        pts_diff = home_row.points - away_row.points
+        lines.append(f"  排名对比: {lead}, 分差 {pts_diff:+d}")
+
+    # has_data: 两队都有行才算完整;只有一队时仍有价值,但标记不完整
+    has_data = n_records >= 1
+    return SliceResult(text="\n".join(lines), has_data=has_data, n_records=n_records)

src/llm/slices/stats.py

@@ -0,0 +1,67 @@
+"""B - 攻防数据切片: 进球/射门/控球/xG 聚合(stats)。"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from src.db.base import AsyncSessionLocal
+from src.llm.slices.common import MatchHeader, SliceResult, _is_stats_available
+from src.llm.slices.form import _get_form
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def stats_slice(header: MatchHeader, *, limit: int = 10, before=None, db: AsyncSession | None = None) -> SliceResult:
+    """B - 攻防数据切片: 进球、射门、控球,评估攻防强度。before=match_date 用于回测。
+
+    db: 可选共享 session,避免每个切片独立建连(见 context_builder 模块 docstring)。
+    """
+    if db is not None:
+        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)
+    else:
+        async with AsyncSessionLocal() as new_db:
+            home_form = await _get_form(new_db, header.home_team_id, before=before, limit=limit)
+            away_form = await _get_form(new_db, header.away_team_id, before=before, limit=limit)
+    lines = [f"── 攻防数据(近 {limit} 场) ──"]
+    n_total = 0
+    # P0-1 修复:每场历史比赛必须根据「该队当时是主是客」判断 side,
+    # 不能用本场 side 硬套 —— 否则进球/失球/xG 全部算反。
+    for label, name, form, team_id in (
+        ("主队", header.home_name, home_form, header.home_team_id),
+        ("客队", header.away_name, away_form, header.away_team_id),
+    ):
+        if form:
+            gf = ga = shots = sot = poss = xg = xga = 0
+            n = n_shots = n_poss = n_xg = 0
+            for fm in form:
+                if fm.home_goals is None: continue
+                is_home = (fm.home_team_id == team_id)
+                gf += fm.home_goals if is_home else fm.away_goals
+                ga += fm.away_goals if is_home else fm.home_goals
+                n += 1
+                # 只使用 cutoff 之前已可用的统计数据
+                if fm.stats and _is_stats_available(fm.stats, before):
+                    if fm.stats.home_shots is not None:
+                        shots += fm.stats.home_shots if is_home else fm.stats.away_shots
+                        sot += fm.stats.home_shots_on_target if is_home else fm.stats.away_shots_on_target
+                        n_shots += 1
+                    if fm.stats.home_possession is not None:
+                        poss += fm.stats.home_possession if is_home else (100 - fm.stats.home_possession)
+                        n_poss += 1
+                    if fm.stats.home_xg is not None:
+                        xg += fm.stats.home_xg if is_home else fm.stats.away_xg
+                        xga += fm.stats.away_xg if is_home else fm.stats.home_xg
+                        n_xg += 1
+            n_total += n
+            if n > 0:
+                lines.append(f"  {label} {name}:")
+                lines.append(f"    场均进球 {gf/n:.2f}, 场均失球 {ga/n:.2f}")
+                if n_shots: lines.append(f"    场均射门 {shots/n_shots:.1f}, 射正 {sot/n_shots:.1f}")
+                if n_poss: lines.append(f"    平均控球 {poss/n_poss:.1f}%")
+                if n_xg: lines.append(f"    场均 xG {xg/n_xg:.2f}, 场均被 xG {xga/n_xg:.2f}")
+            else:
+                lines.append(f"  {label} {name}: 无比分数据")
+        else:
+            lines.append(f"  {label} {name}: 无数据")
+    return SliceResult(text="\n".join(lines), has_data=n_total > 0, n_records=n_total)

tests/test_events_bronze.py

@@ -20,7 +20,7 @@ from datetime import date, datetime, timezone
 import pytest
 
 import src.data.bzzoiro as bz
-from src.db.models import DataLineage, League, Match, RawEvent, Team
+from src.db.models import DataLineage, League, Match, RawEvent, Team, TeamAlias
 
 
 def _event(eid=1001, status="finished", home="Arsenal", away="Chelsea", hs=2, as_=1):
@@ -71,11 +71,20 @@ class _FakeDB:
             League: list(leagues),
             RawEvent: list(raw_events),
         }
-        self._next_id = 0
+        self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)}
+        self._aliases: dict[str, TeamAlias] = {}
+        self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0)
 
     def add(self, obj):
         self.added.append(obj)
 
+    async def get(self, cls, key):
+        if cls is Team:
+            return self._teams_by_id.get(key)
+        if cls is TeamAlias:
+            return self._aliases.get(key)
+        return None
+
     async def execute(self, stmt):
         entities = set()
         for d in (stmt.column_descriptions or []):
@@ -90,6 +99,8 @@ class _FakeDB:
             if getattr(obj, "id", None) is None:
                 self._next_id += 1
                 obj.id = self._next_id
+                if isinstance(obj, Team) and obj.id is not None:
+                    self._teams_by_id[obj.id] = obj
 
 
 @pytest.fixture(autouse=True)
@@ -244,11 +255,14 @@ class TestEventsBronzeOnUpdate:
 
 class TestEventsBronzeIsBestEffort:
     async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch):
+        import src.data.bzzoiro_events as bz_events
+
         async def _boom(*args, **kwargs):
             raise RuntimeError("infra down")
 
-        monkeypatch.setattr(bz, "_write_raw_event", _boom)
+        # Bronze 写入助手直接 import 到 bzzoiro_events 命名空间,需 patch 该处
-        monkeypatch.setattr(bz, "_write_lineage", _boom)
+        monkeypatch.setattr(bz_events, "_write_raw_event", _boom)
+        monkeypatch.setattr(bz_events, "_write_lineage", _boom)
         _patch_fetch(monkeypatch, [_event()])
         db = _FakeDB()
 

tests/test_h2h_perspective.py

@@ -70,7 +70,7 @@ class TestH2HCurrentHomePerspective:
             _make_h2h_match(101, home_id=2, away_id=1, home_goals=3, away_goals=1,
                             home_name="阿森纳", away_name="利物浦"),
         ]
-        import src.llm.context_builder as cb
+        import src.llm.slices.h2h as cb
         orig = cb._get_h2h
         async def mock_get_h2h(db, home_id, away_id, before, *, limit):
             return matches
@@ -100,7 +100,7 @@ class TestH2HCurrentHomePerspective:
             _make_h2h_match(201, home_id=1, away_id=2, home_goals=2, away_goals=1,
                             home_name="曼城", away_name="诺维奇"),
         ]
-        import src.llm.context_builder as cb
+        import src.llm.slices.h2h as cb
 
         async def mock_get_h2h(db, h, a, before, **kw):
             # 真实契约是 async(见 context_builder.py 的 `h2h = await _get_h2h(...)`),
@@ -122,7 +122,7 @@ class TestH2HCurrentHomePerspective:
             _make_h2h_match(301, home_id=2, away_id=1, home_goals=0, away_goals=2,
                             home_name="热刺", away_name="切尔西"),  # 切尔西客场 2-0 赢
         ]
-        import src.llm.context_builder as cb
+        import src.llm.slices.h2h as cb
         orig = cb._get_h2h
         async def mock_get_h2h(db, h, a, before, *, limit):
             return matches

tests/test_multi_agent_cutoff.py

@@ -206,7 +206,7 @@ class TestBacktestXgNotVisible:
 
         header = _make_header(match_dt)
 
-        import src.llm.context_builder as cb
+        import src.llm.slices.stats as cb
 
         async def mock_get_form(db, team_id, before, *, limit=10):
             # before=cutoff(1月13日),比赛在1月15日,满足 before 条件

tests/test_p0_home_away.py

@@ -101,7 +101,7 @@ class TestFormSliceHomeAwayIdentity:
             home_name="曼城",
             away_name="利物浦",
         )
-        import src.llm.context_builder as cb
+        import src.llm.slices.form as cb
         orig_get_form = cb._get_form
         async def mock_get_form(db, team_id, before, *, limit):
             return [hist_match] if team_id == 1 else []
@@ -132,7 +132,7 @@ class TestFormSliceHomeAwayIdentity:
             home_name="阿森纳",
             away_name="切尔西",
         )
-        import src.llm.context_builder as cb
+        import src.llm.slices.form as cb
         orig_get_form = cb._get_form
         async def mock_get_form(db, team_id, before, *, limit):
             return [hist_match] if team_id == 2 else []
@@ -169,7 +169,7 @@ class TestStatsSliceHomeAwayIdentity:
             stats=_make_stats(home_xg=2.5, away_xg=0.8, home_shots=15, away_shots=5,
                               home_sot=6, away_sot=2, home_poss=60.0),
         )
-        import src.llm.context_builder as cb
+        import src.llm.slices.stats as cb
         orig_get_form = cb._get_form
         async def mock_get_form(db, team_id, before, *, limit):
             return [hist_match] if team_id == 1 else []

tests/test_regressions.py

@@ -32,17 +32,24 @@ class TestEagerLoadCoverage:
 
         models.py 已声明 lazy="selectin" 兜底,但这里同时检查显式
         selectinload —— 显式声明是查询意图的固化,也被 P0 修复所依赖。
+        (context_builder 已按 slice 拆分到 src/llm/slices/,getter 随实现迁移。)
         """
-        src = _read("llm/context_builder.py")
+        for rel in ("llm/slices/form.py", "llm/slices/h2h.py", "llm/slices/home_away.py"):
+            src = _read(rel)
+            for fn in ("_get_form", "_get_h2h", "_get_home_away"):
+                # 截取函数体(仅当前文件定义了该函数才检查)
+                m = re.search(rf"async def {fn}\(.*?(?=\nasync def |\n# =|\Z)", src, re.S)
+                if not m:
+                    continue
+                body = m.group(0)
+                assert "selectinload" in body, (
+                    f"{fn} 查询 Match 但未 eager-load 关系 —— "
+                    "this would raise MissingGreenlet in async SQLAlchemy (P0-2)"
+                )
+        # 守卫完整性: 三个 getter 必须都能在 slices 包中找到
+        all_src = "\n".join(_read(r) for r in ("llm/slices/form.py", "llm/slices/h2h.py", "llm/slices/home_away.py"))
         for fn in ("_get_form", "_get_h2h", "_get_home_away"):
-            # 截取函数体
+            assert f"async def {fn}(" in all_src, f"{fn} 未在 slices 包中找到(拆分后迁移缺失?)"
-            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 关闭后访问关系必炸)。"""
@@ -165,7 +172,9 @@ class TestBzzoiroLineage:
             if re.search(r"source_event_id\s*(?:is|==|!=)", stripped):
                 continue
             if re.search(r"source_event_id\s*\.\s*\w+\s*\(", stripped):
-                continue  # 方法调用,不是赋值
+                continue  # 方法调用(obj.source_event_id(...)),不是赋值
+            if re.search(r"\w*source_event_id\s*\(", stripped):
+                continue  # 方法调用(如 find_by_source_event_id(eid)),不是赋值
             if self._ASSIGN_DIRECT.search(stripped):
                 continue  # 直接取配对 raw
             m_var = self._ASSIGN_VIA_VAR.search(stripped)

tests/test_review_required_fixes.py

@@ -17,6 +17,7 @@ import re
 
 import pytest
 
+from src.db.models import Team, TeamAlias
 from src.data.key_ring import _mask
 from src.llm import backtest as bt_mod
 from src.llm.agents import orchestrator as orch_mod
@@ -105,6 +106,15 @@ class _FakeDb:
         self.added: list = []
         self.flush_count = 0
         self._next_id = 1000
+        self._teams_by_id: dict[int, Team] = {}
+        self._aliases: dict[str, TeamAlias] = {}
+
+    async def get(self, cls, key):
+        if cls is Team:
+            return self._teams_by_id.get(key)
+        if cls is TeamAlias:
+            return self._aliases.get(key)
+        return None
 
     async def execute(self, _stmt):
         if self._results:
@@ -120,6 +130,10 @@ class _FakeDb:
             if getattr(obj, "id", None) is None:
                 self._next_id += 1
                 obj.id = self._next_id
+            if isinstance(obj, Team) and getattr(obj, "id", None) is not None:
+                self._teams_by_id[obj.id] = obj
+            if isinstance(obj, TeamAlias):
+                self._aliases[obj.alias_normalized] = obj
 
 
 async def test_r2_standings_actually_upserts(monkeypatch):

tests/test_standings_bronze.py

@@ -18,7 +18,7 @@ from __future__ import annotations
 import pytest
 
 import src.data.bzzoiro as bz
-from src.db.models import DataLineage, IngestFailure, League, RawEvent, Standing, Team
+from src.db.models import DataLineage, IngestFailure, League, RawEvent, Standing, Team, TeamAlias
 
 
 def _payload():
@@ -78,11 +78,21 @@ class _FakeDB:
             Standing: list(standings),
             RawEvent: list(raw_events),
         }
-        self._next_id = 0
+        # session.get 查找表(Team/TeamAlias)
+        self._teams_by_id: dict[int, Team] = {t.id: t for t in teams if getattr(t, "id", None)}
+        self._aliases: dict[str, TeamAlias] = {}
+        self._next_id = max((t.id for t in teams if getattr(t, "id", None)), default=0)
 
     def add(self, obj):
         self.added.append(obj)
 
+    async def get(self, cls, key):
+        if cls is Team:
+            return self._teams_by_id.get(key)
+        if cls is TeamAlias:
+            return self._aliases.get(key)
+        return None
+
     async def execute(self, stmt):
         entities = set()
         for d in (stmt.column_descriptions or []):
@@ -110,6 +120,9 @@ class _FakeDB:
             if getattr(obj, "id", None) is None:
                 self._next_id += 1
                 obj.id = self._next_id
+                # 同步 session.get 可查到新建 Team
+                if isinstance(obj, Team) and obj.id is not None:
+                    self._teams_by_id[obj.id] = obj
 
 
 def _preset_league():
@@ -248,11 +261,14 @@ class TestStandingsRawEventIdempotent:
 
 class TestStandingsBronzeIsBestEffort:
     async def test_bronze_write_failure_does_not_break_ingest(self, monkeypatch):
+        import src.data.bzzoiro_standings as bz_standings
+
         async def _boom(*args, **kwargs):
             raise RuntimeError("infra down")
 
-        monkeypatch.setattr(bz, "_write_raw_event", _boom)
+        # Bronze 写入助手直接 import 到 bzzoiro_standings 命名空间,需 patch 该处
-        monkeypatch.setattr(bz, "_write_lineage", _boom)
+        monkeypatch.setattr(bz_standings, "_write_raw_event", _boom)
+        monkeypatch.setattr(bz_standings, "_write_lineage", _boom)
         _patch_fetch(monkeypatch, _payload())
         db = _FakeDB(leagues=[_preset_league()])