Add data-driven AI predictions via API-Football

- football_data.py: 수집·캐싱·데이터블록 조립 (48팀 ID/FIFA랭킹)
- FootballCache 테이블: 폼·H2H·스쿼드 캐시 (외부 DB)
- 프롬프트에 랭킹·폼·H2H·스쿼드·WC실제결과 주입
- 키 없으면 기존(이름만) 동작 폴백, 무료 한도 점진 수집

backend/.env.example

@@ -53,6 +53,11 @@ ANTHROPIC_MODEL=claude-opus-4-8
 GOOGLE_API_KEY=
 GOOGLE_MODEL=gemini-2.5-flash
 
+# ── 외부 연동: 축구 데이터 (API-Football 무료 티어) ──
+# 키 없으면 데이터 수집/주입 생략 → 기존(이름만) 예측으로 동작.
+# 발급: https://dashboard.api-football.com (무료 100req/일)
+FOOTBALL_API_KEY=
+
 # ── 외부 연동: 이메일 ──
 # 1순위: Azure Communication Services(ACS) Email (endpoint + accesskey)
 #   AZURE_ACS_SENDER 는 검증된 MailFrom 주소(예: donotreply@triplepick.o2o.kr)

backend/app/config.py

@@ -112,6 +112,20 @@ class Settings(BaseSettings):
     google_api_key: str = ""
     google_model: str = "gemini-2.5-flash"
 
+    # ── 외부 연동: 축구 데이터 (API-Football 무료 티어) ───────
+    # 키 미설정 시 데이터 수집/주입 전부 no-op → 기존(이름만) 예측으로 폴백.
+    football_api_key: str = ""
+    football_api_base: str = "https://v3.football.api-sports.io"
+    # 캐시 신선도(시간). 이 시간 지난 팀만 워커가 재수집(랭킹/폼은 자주 안 변함).
+    football_cache_ttl_hours: int = 168  # 7일
+    # 수집 잡 1회당 갱신할 최대 팀 수(무료 100/일·10/분 제한 분산).
+    football_refresh_batch: int = 8
+    # API 호출 간 최소 간격(초) — 무료 10req/분 제한 회피.
+    football_call_interval_sec: float = 7.0
+    # 수집 잡 시각(KST) — 예측 생성(00:05)보다 앞서 캐시를 채워둔다.
+    football_refresh_hour: int = 0
+    football_refresh_minute: int = 0
+
     # ── 외부 연동: 이메일 ─────────────────────────────────────
     # 1순위: Azure Communication Services(ACS) Email — endpoint + accesskey.
     #   AZURE_ACS_SENDER 는 검증된 MailFrom 주소(예: donotreply@triplepick.o2o.kr).

backend/app/models.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 from datetime import date, datetime
 
 from sqlalchemy import (
+    JSON,
     Boolean,
     Date,
     DateTime,
@@ -201,3 +202,23 @@ class PageVisit(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
+
+
+class FootballCache(Base):
+    """축구 데이터 캐시 (API-Football 수집 결과) — 예측 프롬프트 조립용.
+
+    key 규칙:
+      team:{CODE}      팀 베이스라인(폼·평균득실·클린시트·스쿼드)
+      h2h:{A}-{B}      상대전적
+      teamid:{CODE}    팀코드→API팀ID 매핑
+    payload 는 가공된 압축 JSON. fetched_at 으로 캐시 신선도(TTL) 판단.
+    api·worker 가 공유하는 유일한 영속 저장소가 DB 라 여기에 둔다.
+    """
+
+    __tablename__ = "football_cache"
+
+    key: Mapped[str] = mapped_column(String, primary_key=True)
+    payload: Mapped[dict] = mapped_column(JSON, default=dict)
+    fetched_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )

backend/app/services/ai.py

@@ -30,6 +30,7 @@ class MatchContext:
     team_b: str
     venue: str
     kickoff: str  # ISO
+    data_block: str | None = None  # 실데이터(폼·H2H·랭킹·WC결과) 주입 블록. 없으면 이름만.
 
 
 # 모델별 분석 관점(페르소나) — 동일 경기라도 서로 다른 시각으로 보게 해
@@ -54,6 +55,8 @@ PERSONA = {
 
 
 def _prompt(ctx: MatchContext, persona: str) -> str:
+    # 실데이터 블록이 있으면 팀/킥오프 다음에 삽입(없으면 빈 문자열 — 기존 동작 동일).
+    data = f"\n{ctx.data_block}\n" if ctx.data_block else ""
     return (
         f"{persona}\n"
         f"Predict the result of this 2026 FIFA World Cup match using YOUR perspective "
@@ -61,6 +64,7 @@ def _prompt(ctx: MatchContext, persona: str) -> str:
         f"pick when your perspective warrants it.\n"
         f"Team A: {ctx.team_a}\nTeam B: {ctx.team_b}\n"
         f"Venue: {ctx.venue}\nKickoff: {ctx.kickoff}\n"
+        f"{data}"
         f"Important: World Cup group-stage matches are played at NEUTRAL venues. "
         f"Neither team has home advantage unless it is a host nation "
         f"(Mexico, USA, or Canada). Do NOT claim home advantage otherwise.\n\n"

backend/app/services/football_data.py

@@ -0,0 +1,299 @@
+"""축구 데이터 연동 (API-Football 무료 티어) — 예측 프롬프트용 수집·캐싱·조립.
+
+설계 요약
+  - 수집(refresh): 워커가 다가오는 경기의 팀/H2H 를 무료 한도(100/일·10/분) 안에서
+    점진 갱신해 FootballCache(DB)에 저장. 호출 간 간격(throttle)으로 분당 제한 회피.
+  - 조립(build_data_block): 예측 생성 시 캐시만 읽어 프롬프트 블록을 만든다(외부 호출 X).
+    2026 실제 결과는 우리 DB(matches)에서 직접 계산해 보완(무료 API 가 2026 시즌 차단).
+  - FOOTBALL_API_KEY 미설정이면 전부 no-op → 기존(이름만) 예측으로 자연 폴백.
+
+무료 티어 제약 우회
+  - season=2026, last=N 파라미터는 막힘 → season(2024,2023) 조회로 폼/H2H 확보.
+  - FIFA 랭킹은 API 에 없음 → 아래 FIFA_RANK(수동 관리, 외부 무료 소스 기반)로 보완.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import timedelta
+
+import httpx
+from sqlalchemy import or_, select
+
+from ..config import settings
+from ..domain import ensure_aware, now_utc
+from ..models import FootballCache, Match
+
+log = logging.getLogger("triplepick.football")
+
+# 무료 접근 가능한 시즌(최신 우선). 폼 표본 확보용.
+_SEASONS = (2024, 2023)
+
+# FIFA 랭킹(수동 관리) — API-Football 에 없어 외부 무료 소스값을 코드코어로 둔다.
+# 없는 팀은 블록에서 랭킹 줄을 생략(폴백). 갱신은 가끔 수동.
+# FIFA 세계랭킹 — 공식 발표 기준(2026-06-11). API-Football 엔 없어 외부값을 코드에 둔다.
+# 갱신: FIFA 공식 랭킹 새로 나오면 숫자만 교체.
+FIFA_RANK: dict[str, int] = {
+    "MEX": 13, "RSA": 61, "KOR": 22, "CZE": 43,      # A
+    "SUI": 19, "BIH": 63, "CAN": 32, "QAT": 49,      # B
+    "SCO": 38, "MAR": 7, "BRA": 6, "HAI": 84,        # C
+    "USA": 15, "AUS": 23, "TUR": 26, "PAR": 42,      # D
+    "GER": 9, "CIV": 29, "ECU": 28, "CUW": 82,       # E
+    "NED": 8, "SWE": 35, "TUN": 56, "JPN": 18,       # F
+    "BEL": 10, "EGY": 30, "IRN": 20, "NZL": 85,      # G
+    "ESP": 2, "CPV": 67, "KSA": 60, "URU": 17,       # H
+    "FRA": 3, "SEN": 16, "IRQ": 57, "NOR": 31,       # I
+    "ARG": 1, "ALG": 27, "AUT": 24, "JOR": 64,       # J
+    "POR": 5, "COD": 45, "UZB": 50, "COL": 14,       # K
+    "ENG": 4, "CRO": 11, "GHA": 73, "PAN": 34,       # L
+}
+
+# 팀코드→API-Football 팀ID 직접 매핑(이름검색 오인 방지 — 청소년/여자/표기차).
+# 48개 본선 팀 전체. 2026 본선 확정명단 기준 시니어 대표팀 ID.
+TEAM_ID_OVERRIDE: dict[str, int] = {
+    "MEX": 16, "RSA": 1531, "KOR": 17, "CZE": 770,      # A
+    "SUI": 15, "BIH": 1113, "CAN": 5529, "QAT": 1569,   # B
+    "SCO": 1108, "MAR": 31, "BRA": 6, "HAI": 2386,      # C
+    "USA": 2384, "AUS": 20, "TUR": 777, "PAR": 2380,    # D
+    "GER": 25, "CIV": 1501, "ECU": 2382, "CUW": 5530,   # E
+    "NED": 1118, "SWE": 5, "TUN": 28, "JPN": 12,        # F
+    "BEL": 1, "EGY": 32, "IRN": 22, "NZL": 4673,        # G
+    "ESP": 9, "CPV": 1533, "KSA": 23, "URU": 7,         # H
+    "FRA": 2, "SEN": 13, "IRQ": 1567, "NOR": 1090,      # I
+    "ARG": 26, "ALG": 1532, "AUT": 775, "JOR": 1548,    # J
+    "POR": 27, "COD": 1508, "UZB": 1568, "COL": 8,      # K
+    "ENG": 10, "CRO": 3, "GHA": 1504, "PAN": 11,        # L
+}
+
+
+def _is_senior(name: str) -> bool:
+    """청소년/여자/올림픽 팀 제외(시니어 대표팀만)."""
+    n = name.upper()
+    if any(b in n for b in ("U23", "U21", "U20", "U19", "U17", "U-23", "WOMEN", "OLYMPIC")):
+        return False
+    return not n.endswith(" W")
+
+
+def enabled() -> bool:
+    return bool(settings.football_api_key)
+
+
+# ── 저수준: API 호출 (throttle 포함) ───────────────────────────
+async def _get(client: httpx.AsyncClient, path: str, **params) -> dict:
+    r = await client.get(
+        settings.football_api_base + path,
+        headers={"x-apisports-key": settings.football_api_key},
+        params=params or None,
+        timeout=25,
+    )
+    r.raise_for_status()
+    await asyncio.sleep(settings.football_call_interval_sec)  # 10/분 제한 회피
+    return r.json()
+
+
+# ── 캐시 upsert ────────────────────────────────────────────────
+async def _upsert(db, key: str, payload: dict) -> None:
+    row = await db.get(FootballCache, key)
+    if row:
+        row.payload = payload
+        row.fetched_at = now_utc()
+    else:
+        db.add(FootballCache(key=key, payload=payload, fetched_at=now_utc()))
+
+
+async def _team_id(db, client, code: str, name: str) -> int | None:
+    """팀코드→API팀ID. 1)직접매핑 2)캐시 3)이름검색(시니어 대표팀 필터)."""
+    if code in TEAM_ID_OVERRIDE:
+        return TEAM_ID_OVERRIDE[code]
+    cached = await db.get(FootballCache, f"teamid:{code}")
+    if cached and cached.payload.get("id"):
+        return cached.payload["id"]
+    query = name.split(" (")[0].strip()  # "Korea Republic (...)" → "Korea Republic"
+    data = await _get(client, "/teams", search=query)
+    nats = [
+        x["team"] for x in (data.get("response") or [])
+        if x["team"].get("national") and _is_senior(x["team"]["name"])
+    ]
+    if not nats:
+        return None
+    tid = nats[0]["id"]
+    await _upsert(db, f"teamid:{code}", {"id": tid, "name": nats[0]["name"]})
+    return tid
+
+
+# ── 팀 베이스라인 수집 → 압축 payload ──────────────────────────
+async def _fetch_team(db, client, code: str, name: str) -> bool:
+    tid = await _team_id(db, client, code, name)
+    if not tid:
+        log.warning("football: team id 미확인 (%s/%s)", code, name)
+        return False
+
+    sq = await _get(client, "/players/squads", team=tid)
+    players = (sq.get("response") or [{}])[0].get("players", []) if sq.get("response") else []
+
+    fixtures = []
+    for season in _SEASONS:
+        d = await _get(client, "/fixtures", team=tid, season=season)
+        fixtures += d.get("response", [])
+
+    results = []
+    for f in fixtures:
+        g = f["goals"]
+        home, away = f["teams"]["home"], f["teams"]["away"]
+        is_home = home["id"] == tid
+        gf = g["home"] if is_home else g["away"]
+        ga = g["away"] if is_home else g["home"]
+        if gf is None or ga is None:
+            continue
+        r = "W" if gf > ga else "L" if gf < ga else "D"
+        results.append({"date": f["fixture"]["date"][:10], "r": r, "gf": gf, "ga": ga})
+    results.sort(key=lambda x: x["date"])
+
+    n = max(len(results), 1)
+    w = sum(1 for x in results if x["r"] == "W")
+    d_ = sum(1 for x in results if x["r"] == "D")
+    l = sum(1 for x in results if x["r"] == "L")
+    payload = {
+        "team_id": tid,
+        "form": " ".join(x["r"] for x in results[-8:]),
+        "w": w, "d": d_, "l": l,
+        "gf_avg": round(sum(x["gf"] for x in results) / n, 2),
+        "ga_avg": round(sum(x["ga"] for x in results) / n, 2),
+        "clean_sheets": sum(1 for x in results if x["ga"] == 0),
+        "stars": [p["name"] for p in players if p.get("position") == "Attacker"][:5],
+        "squad_n": len(players),
+    }
+    await _upsert(db, f"team:{code}", payload)
+    return True
+
+
+async def _fetch_h2h(db, client, code_a, id_a, code_b, id_b) -> None:
+    d = await _get(client, "/fixtures/headtohead", h2h=f"{id_a}-{id_b}")
+    rows = []
+    for f in d.get("response", []):
+        g = f["goals"]
+        if g["home"] is None:
+            continue
+        rows.append({
+            "date": f["fixture"]["date"][:10],
+            "home": f["teams"]["home"]["name"], "hg": g["home"],
+            "ag": g["away"], "away": f["teams"]["away"]["name"],
+        })
+    await _upsert(db, f"h2h:{code_a}-{code_b}", {"rows": rows[-5:]})
+
+
+# ── 수집 잡 (워커 호출) ─────────────────────────────────────────
+async def refresh(db, matches: list[Match], limit: int | None = None) -> int:
+    """주어진 경기들의 팀/H2H 캐시를 stale 우선으로 갱신. 갱신한 팀 수 반환."""
+    if not enabled():
+        return 0
+    limit = limit if limit is not None else settings.football_refresh_batch
+    cutoff = now_utc() - timedelta(hours=settings.football_cache_ttl_hours)
+    refreshed = 0
+    async with httpx.AsyncClient() as client:
+        seen: set[str] = set()
+        for m in matches:
+            for code, name in ((m.team_a_code, m.team_a_name), (m.team_b_code, m.team_b_name)):
+                if code in seen or refreshed >= limit:
+                    continue
+                seen.add(code)
+                row = await db.get(FootballCache, f"team:{code}")
+                if row and ensure_aware(row.fetched_at) > cutoff:
+                    continue  # 신선 — skip
+                try:
+                    if await _fetch_team(db, client, code, name):
+                        refreshed += 1
+                except Exception as e:  # noqa: BLE001
+                    log.warning("football: 팀 %s 수집 실패 %s", code, e)
+        # H2H (양 팀 캐시가 있을 때만)
+        for m in matches:
+            h = await db.get(FootballCache, f"h2h:{m.team_a_code}-{m.team_b_code}")
+            if h and ensure_aware(h.fetched_at) > cutoff:
+                continue
+            ta = await db.get(FootballCache, f"team:{m.team_a_code}")
+            tb = await db.get(FootballCache, f"team:{m.team_b_code}")
+            if not (ta and tb):
+                continue
+            try:
+                await _fetch_h2h(db, client, m.team_a_code, ta.payload["team_id"],
+                                 m.team_b_code, tb.payload["team_id"])
+            except Exception as e:  # noqa: BLE001
+                log.warning("football: H2H %s 수집 실패 %s", m.match_id, e)
+    await db.commit()
+    if refreshed:
+        log.info("football: 팀 %d개 캐시 갱신", refreshed)
+    return refreshed
+
+
+# ── 라이브 WC 결과 (우리 DB) ───────────────────────────────────
+async def _live_wc(db, code: str, before) -> list[str]:
+    rows = (await db.execute(
+        select(Match)
+        .where(
+            or_(Match.team_a_code == code, Match.team_b_code == code),
+            Match.result_outcome.isnot(None),
+            Match.kickoff_at < before,
+        )
+        .order_by(Match.kickoff_at)
+    )).scalars().all()
+    out = []
+    for m in rows:
+        is_a = m.team_a_code == code
+        gf = m.result_score_a if is_a else m.result_score_b
+        ga = m.result_score_b if is_a else m.result_score_a
+        opp = m.team_b_short if is_a else m.team_a_short
+        if gf is None or ga is None:
+            continue
+        r = "W" if gf > ga else "L" if gf < ga else "D"
+        out.append(f"{r} {gf}-{ga} vs {opp}")
+    return out
+
+
+def _h2h_line(row, default_code: str) -> str:
+    if not row or not row.payload.get("rows"):
+        return "no recent meetings"
+    rows = row.payload["rows"][-3:]
+    return "; ".join(
+        f'{r["home"]} {r["hg"]}-{r["ag"]} {r["away"]} ({r["date"][:4]})' for r in rows
+    )
+
+
+async def _team_lines(db, code, name, row, kickoff) -> str:
+    rank = FIFA_RANK.get(code)
+    head = f"[{name}]" + (f"  FIFA #{rank}" if rank else "")
+    if not row:
+        return f"{head}\n  (no external data — use general knowledge)"
+    p = row.payload
+    live = await _live_wc(db, code, kickoff)
+    wc = "; ".join(live) if live else "(none yet)"
+    return (
+        f"{head}\n"
+        f"  Form(last8): {p.get('form','')} | {p.get('w',0)}-{p.get('d',0)}-{p.get('l',0)} "
+        f"(W-D-L), avg {p.get('gf_avg','?')}-{p.get('ga_avg','?')}, "
+        f"clean sheets {p.get('clean_sheets',0)}\n"
+        f"  WC2026 so far: {wc}\n"
+        f"  Key attackers: {', '.join(p.get('stars', []))}"
+    )
+
+
+async def build_data_block(db, match: Match) -> str | None:
+    """경기 예측 프롬프트에 끼울 데이터블록. 캐시만 읽음(외부 호출 X).
+    양 팀 모두 데이터가 없으면 None → 기존(이름만) 동작 폴백."""
+    if not enabled():
+        return None
+    ta = await db.get(FootballCache, f"team:{match.team_a_code}")
+    tb = await db.get(FootballCache, f"team:{match.team_b_code}")
+    # 라이브 WC 결과는 캐시 없어도 의미 있으므로, 캐시가 둘 다 없을 때만 폴백
+    has_live = bool(await _live_wc(db, match.team_a_code, match.kickoff_at)) or \
+        bool(await _live_wc(db, match.team_b_code, match.kickoff_at))
+    if not ta and not tb and not has_live:
+        return None
+    h2h = (await db.get(FootballCache, f"h2h:{match.team_a_code}-{match.team_b_code}")) or \
+        (await db.get(FootballCache, f"h2h:{match.team_b_code}-{match.team_a_code}"))
+    return "\n".join([
+        "=== MATCH DATA (factual, weigh heavily over priors) ===",
+        await _team_lines(db, match.team_a_code, match.team_a_name, ta, match.kickoff_at),
+        await _team_lines(db, match.team_b_code, match.team_b_name, tb, match.kickoff_at),
+        f"[Head-to-head] {_h2h_line(h2h, match.team_a_code)}",
+        "===",
+    ])

backend/app/worker.py

@@ -30,6 +30,7 @@ from .database import SessionLocal, init_db
 from .domain import compute_phase, ensure_aware, now_utc
 from .models import AIPrediction, Match, UserPrediction
 from .scoring import load_scoring_data
+from .services import football_data
 from .services.ai import MatchContext, PROVIDERS, ProviderUnavailable
 from .services.email import EmailUnavailable, build_result_email, send_email
 from .services.grading import apply_result
@@ -97,11 +98,14 @@ async def generate_ai_predictions(only_missing: bool = True) -> None:
         matches = [m for m in rows if ensure_aware(m.opens_at) <= horizon]
 
         for m in matches:
+            # 실데이터 블록(폼·H2H·랭킹·WC결과) — 캐시만 읽음. 없으면 None(이름만).
+            data_block = await football_data.build_data_block(db, m)
             ctx = MatchContext(
                 team_a=m.team_a_name,
                 team_b=m.team_b_name,
                 venue=m.venue,
                 kickoff=m.kickoff_at.isoformat(),
+                data_block=data_block,
             )
             for model, fn in PROVIDERS.items():
                 pred = next((p for p in m.predictions if p.model == model), None)
@@ -132,6 +136,28 @@ async def generate_ai_predictions(only_missing: bool = True) -> None:
         await db.commit()
 
 
+# ── 2.1) 축구 데이터 수집 (예측 생성 전에 캐시를 채워둔다) ──
+async def refresh_football_data() -> None:
+    """다가오는 미종료 경기의 팀/H2H 축구 데이터를 캐시에 갱신.
+
+    FOOTBALL_API_KEY 미설정이면 no-op. 무료 한도(100/일·10/분)를 호출 간격·배치로
+    분산하므로 한 번에 일부 팀만 갱신되고, 매일 잡이 누적해 채운다.
+    """
+    if not football_data.enabled():
+        return
+    async with SessionLocal() as db:
+        horizon = now_utc() + timedelta(hours=settings.ai_generate_lookahead_hours)
+        rows = (
+            await db.execute(select(Match).where(Match.result_outcome.is_(None)))
+        ).scalars().all()
+        # 투표 윈도우에 든(곧 들) 경기만, 가장 가까운 경기부터(배치 한도 시 우선 수집).
+        matches = sorted(
+            (m for m in rows if ensure_aware(m.opens_at) <= horizon),
+            key=lambda m: ensure_aware(m.opens_at),
+        )
+        await football_data.refresh(db, matches)
+
+
 # ── 2.5) 결과 자동 정산 (관리자 입력 불필요) ────────────────
 async def settle_matches() -> None:
     """킥오프가 지난 미정산 경기를, 소스가 FINISHED 로 보고하는 즉시 채점·종료.
@@ -259,6 +285,14 @@ def build_scheduler() -> AsyncIOScheduler:
         id="status_tick",
         next_run_time=now_utc(),
     )
+    # 축구 데이터 수집: 매일 KST 00:00 (예측 생성보다 먼저 캐시를 채움)
+    sched.add_job(
+        refresh_football_data,
+        "cron",
+        hour=settings.football_refresh_hour,
+        minute=settings.football_refresh_minute,
+        id="football_refresh",
+    )
     # AI 예측 생성: 매일 KST 00:05 (실 API)
     sched.add_job(
         generate_ai_predictions,
@@ -283,6 +317,7 @@ async def main() -> None:
     await init_db()  # 테이블 보장 (idempotent). 시드는 API 가 담당.
     # 기동 시 1회: 일정 동기화 → AI 예측 생성 → 밀린 경기 자동 정산
     await sync_schedule_job()
+    await refresh_football_data()  # 축구 데이터 캐시 선채움 (키 없으면 no-op)
     await generate_ai_predictions()
     await settle_matches()