152 lines
5.3 KiB
Python
152 lines
5.3 KiB
Python
"""야구 예측 프롬프트용 데이터 블록 — 자체 DB(정산 결과) + DataCache(프리뷰·순위).
|
|
|
|
결과 동기화가 쌓은 종료 경기에서 팀 폼·시즌 성적·상대전적을 계산하고,
|
|
캐시된 선발투수·공식 상대전적·순위를 덧붙인다. 모두 없으면 None(이름만 예측).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import or_, select
|
|
|
|
from ..models import DataCache, Match
|
|
|
|
|
|
def _wdl(gf: int, ga: int) -> str:
|
|
return "W" if gf > ga else "L" if gf < ga else "D"
|
|
|
|
|
|
async def _team_games(db, league: str, code: str, before) -> list[dict]:
|
|
rows = (await db.execute(
|
|
select(Match)
|
|
.where(
|
|
Match.league == league,
|
|
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 # team_a = 원정
|
|
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
|
|
if gf is None or ga is None:
|
|
continue
|
|
out.append({
|
|
"r": _wdl(gf, ga), "gf": gf, "ga": ga,
|
|
"opp": m.team_b_short if is_a else m.team_a_short,
|
|
})
|
|
return out
|
|
|
|
|
|
def _team_lines(name: str, games: list[dict]) -> str:
|
|
head = f"[{name}]"
|
|
if not games:
|
|
return f"{head}\n (no season data yet — use general knowledge)"
|
|
n = len(games)
|
|
w = sum(1 for g in games if g["r"] == "W")
|
|
d = sum(1 for g in games if g["r"] == "D")
|
|
l = sum(1 for g in games if g["r"] == "L")
|
|
gf_avg = round(sum(g["gf"] for g in games) / n, 2)
|
|
ga_avg = round(sum(g["ga"] for g in games) / n, 2)
|
|
recent = games[-8:]
|
|
detail = "; ".join("{r} {gf}-{ga} vs {opp}".format(**g) for g in recent[-4:])
|
|
return (
|
|
f"{head}\n"
|
|
f" Season(in our data): {w}-{d}-{l} (W-D-L), "
|
|
f"runs avg {gf_avg} scored / {ga_avg} allowed over {n} games\n"
|
|
f" Form(last{len(recent)}): {' '.join(g['r'] for g in recent)} ({detail})"
|
|
)
|
|
|
|
|
|
async def _h2h_line(db, league: str, code_a: str, code_b: str, before) -> str:
|
|
rows = (await db.execute(
|
|
select(Match)
|
|
.where(
|
|
Match.league == league,
|
|
or_(
|
|
(Match.team_a_code == code_a) & (Match.team_b_code == code_b),
|
|
(Match.team_a_code == code_b) & (Match.team_b_code == code_a),
|
|
),
|
|
Match.result_outcome.isnot(None),
|
|
Match.kickoff_at < before,
|
|
)
|
|
.order_by(Match.kickoff_at)
|
|
)).scalars().all()
|
|
parts = [
|
|
f"{m.team_a_short} {m.result_score_a}-{m.result_score_b} {m.team_b_short}"
|
|
for m in rows[-5:]
|
|
if m.result_score_a is not None
|
|
]
|
|
return "; ".join(parts) if parts else "no meetings in our data yet"
|
|
|
|
|
|
def _starter_line(side: str, s: dict | None) -> str | None:
|
|
if not s or not s.get("name"):
|
|
return None
|
|
era = f", season ERA {s['era']}" if s.get("era") else ""
|
|
rec = (
|
|
f" ({s['w']}W-{s['l']}L)"
|
|
if s.get("w") is not None and s.get("l") is not None else ""
|
|
)
|
|
vs = f", ERA vs this opponent {s['vsEra']}" if s.get("vsEra") else ""
|
|
return f"[{side} starting pitcher] {s['name']}{era}{rec}{vs}"
|
|
|
|
|
|
def _standing_line(name: str, st: dict | None) -> str | None:
|
|
if not st:
|
|
return None
|
|
extra = f", last5 {st['last5']}" if st.get("last5") else ""
|
|
return (
|
|
f"[{name} standings] rank {st.get('rank')}, {st.get('w')}W-"
|
|
f"{st.get('l')}L (pct {st.get('wra')}){extra}"
|
|
)
|
|
|
|
|
|
async def _cached_extras(db, match: Match) -> list[str]:
|
|
lines: list[str] = []
|
|
prev = await db.get(DataCache, f"preview:{match.match_id}")
|
|
if prev:
|
|
p = prev.payload
|
|
for line in (
|
|
_starter_line("Away", p.get("starterA")),
|
|
_starter_line("Home", p.get("starterB")),
|
|
):
|
|
if line:
|
|
lines.append(line)
|
|
vs = p.get("seasonVs")
|
|
if vs and vs.get("aWin") is not None:
|
|
lines.append(
|
|
f"[Season head-to-head (official)] away {vs['aWin']}W - "
|
|
f"{vs.get('draw', 0)}D - home {vs.get('bWin')}W"
|
|
)
|
|
st_row = await db.get(DataCache, f"standings:{match.league}")
|
|
if st_row:
|
|
table = st_row.payload
|
|
for line in (
|
|
_standing_line(match.team_a_short, table.get(match.team_a_code)),
|
|
_standing_line(match.team_b_short, table.get(match.team_b_code)),
|
|
):
|
|
if line:
|
|
lines.append(line)
|
|
return lines
|
|
|
|
|
|
async def build_baseball_data_block(db, match: Match) -> str | None:
|
|
ga = await _team_games(db, match.league, match.team_a_code, match.kickoff_at)
|
|
gb = await _team_games(db, match.league, match.team_b_code, match.kickoff_at)
|
|
extras = await _cached_extras(db, match)
|
|
if not ga and not gb and not extras:
|
|
return None
|
|
h2h = await _h2h_line(
|
|
db, match.league, match.team_a_code, match.team_b_code, match.kickoff_at
|
|
)
|
|
return "\n".join([
|
|
"=== MATCH DATA (factual; weigh heavily over priors) ===",
|
|
"Away " + _team_lines(match.team_a_name, ga),
|
|
"Home " + _team_lines(match.team_b_name, gb),
|
|
f"[Head-to-head in our data] {h2h}",
|
|
*extras,
|
|
"===",
|
|
])
|