feat: 足球 LLM 预测服务初始提交

Profeto — 给 LLM 提供数据,让 LLM 预测足球比分。

核心模块:
- FastAPI 后端 + PostgreSQL (SQLAlchemy async)
- 多 Agent LLM 预测 (5 专家 + 终裁)
- 数据采集 (bzzoiro / understat / injuries)
- React 前端 (Vite + Tailwind)

包含:
- 数据源抽象 (DataSource 协议 + 注册表)
- Alembic 数据库迁移
- Prompt 模板 (单/多 Agent)
- 核心路径单元测试
This commit is contained in:
shangfangjian
2026-09-09 02:10:47 +08:00
commit 0a27b18c27
74 changed files with 5667 additions and 0 deletions
+40
View File
@@ -0,0 +1,40 @@
"""比赛匹配辅助函数(多数据源共用)。
bzzoiro / understat 等数据源在入库时都需要:
- 按队名获取或创建球队(get_or_create_team)
- 按联赛+主队+客队+日期找已有比赛(find_existing_match)
"""
from __future__ import annotations
from sqlalchemy import func, select
from src.db.models import Match, Team
async def get_or_create_team(db, name: str) -> Team:
"""按名获取球队,不存在则创建。"""
stmt = select(Team).where(Team.name == name)
team = (await db.execute(stmt)).scalar_one_or_none()
if team is None:
team = Team(name=name)
db.add(team)
await db.flush()
return team
async def find_existing_match(db, league_id: int, home_name: str, away_name: str, date) -> Match | None:
"""按联赛+主队+客队+日期找已有比赛(天级匹配,避免时间精度差异)。"""
home_team = (await db.execute(select(Team).where(Team.name == home_name))).scalar_one_or_none()
away_team = (await db.execute(select(Team).where(Team.name == away_name))).scalar_one_or_none()
if home_team is None or away_team is None:
return None
date_only = date.date() if hasattr(date, "date") else date
stmt = (
select(Match)
.where(Match.league_id == league_id)
.where(Match.home_team_id == home_team.id)
.where(Match.away_team_id == away_team.id)
.where(func.date(Match.match_date) == date_only)
)
return (await db.execute(stmt)).scalar_one_or_none()