"""World Cup Prediction Agent — Web UI.
FastAPI app deployed on HF Spaces. Renders HTML directly (no Jinja2).
"""
import json
import subprocess
from datetime import datetime, timezone
from pathlib import Path
import markdown
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
BASE = Path(__file__).parent
app = FastAPI(title="World Cup Prediction Agent")
# FIFA 3-letter → ISO 2-letter → flag emoji
_FIFA_TO_ISO2 = {
"NED": "NL", "JPN": "JP", "CIV": "CI", "ECU": "EC", "SWE": "SE",
"TUN": "TN", "ESP": "ES", "CPV": "CV", "BEL": "BE", "EGY": "EG",
"KSA": "SA", "URU": "UY", "IRN": "IR", "NZL": "NZ", "FRA": "FR",
"SEN": "SN", "IRQ": "IQ", "NOR": "NO", "ARG": "AR", "ALG": "DZ",
"AUT": "AT", "JOR": "JO", "POR": "PT", "COD": "CD", "ENG": "GB",
"GHA": "GH", "PAN": "PA", "UZB": "UZ", "COL": "CO", "CZE": "CZ",
"RSA": "ZA", "SUI": "CH", "BIH": "BA", "CAN": "CA", "QAT": "QA",
"MEX": "MX", "KOR": "KR", "USA": "US", "AUS": "AU", "SCO": "GB",
"MAR": "MA", "BRA": "BR", "HAI": "HT", "TUR": "TR", "PAR": "PY",
"GER": "DE", "CUW": "CW", "CRO": "HR", "CRC": "CR", "IRE": "IE",
"WLS": "GB", "DEN": "DK", "SRB": "RS", "POL": "PL", "SLO": "SI",
"ROM": "RO", "HUN": "HU", "SVK": "SK", "GRE": "GR", "BUL": "BG",
"CHL": "CL", "PER": "PE", "BOL": "BO", "VEN": "VE", "HON": "HN",
"CUB": "CU", "JAM": "JM", "TRI": "TT", "CMR": "CM", "NGA": "NG",
"MLI": "ML", "BFA": "BF", "GUI": "GN", "COG": "CG", "BEN": "BJ",
"ZIM": "ZW", "ZAM": "ZM", "MOZ": "MZ", "ANG": "AO", "ETH": "ET",
"TAN": "TZ", "UGA": "UG", "RWA": "RW", "CTA": "CF", "GAB": "GA",
"IRQ": "IQ", "UAE": "AE", "KUW": "KW", "BHR": "BH", "OMA": "OM",
"YEM": "YE", "SYR": "SY", "LIB": "LB", "PAL": "PS", "AFG": "AF",
"IND": "IN", "BAN": "BD", "PAK": "PK", "SRI": "LK", "NEP": "NP",
"THA": "TH", "VIE": "VN", "IDN": "ID", "MAS": "MY", "PHI": "PH",
"CHN": "CN", "TPE": "TW", "HKG": "HK", "MAC": "MO", "MGL": "MN",
"RUS": "RU", "UKR": "UA", "BLR": "BY", "KAZ": "KZ", "GEO": "GE",
"ARM": "AM", "AZE": "AZ", "MDA": "MD", "LTU": "LT", "LAT": "LV",
"EST": "EE", "FIN": "FI", "ISL": "IS", "IRL": "IE", "NIG": "NE",
}
def _flag(code: str) -> str:
"""Return flag emoji for a FIFA team code, or empty string."""
iso2 = _FIFA_TO_ISO2.get(code.upper(), "")
if not iso2 or len(iso2) != 2:
return ""
return "".join(chr(0x1F1E6 + ord(c) - ord("A")) for c in iso2.upper())
def _pick_display(pick: str, home: str, away: str) -> str:
"""Translate home/away/draw to team code."""
if pick == "home":
return home
if pick == "away":
return away
return "DRAW"
def _team(code: str) -> str:
"""Team code with flag emoji."""
flag = _flag(code)
return f"{flag} {code}" if flag else code
def _team_names() -> dict[str, str]:
"""FIFA code → full country name from schedule.json."""
names: dict[str, str] = {}
for f in _load("schedule.json", []):
if f.get("home") and f.get("home_name"):
names[f["home"]] = f["home_name"]
if f.get("away") and f.get("away_name"):
names[f["away"]] = f["away_name"]
return names
def _team_full(code: str, names: dict[str, str] | None = None) -> str:
"""Team with flag emoji and full country name."""
flag = _flag(code)
name = (names or {}).get(code.upper(), "")
if name:
return f"{flag} {name}" if flag else name
return _team(code)
def _pick_display_full(pick: str, home: str, away: str, names: dict[str, str]) -> str:
"""Translate home/away/draw to full team label."""
if pick == "home":
return _team_full(home, names)
if pick == "away":
return _team_full(away, names)
return "DRAW"
def _prob_summary(pred: dict, home: str, away: str) -> str:
"""Win probability line when model outputs HOME/DRAW/AWAY percentages."""
probs = pred.get("probabilities") or {}
if not probs:
return ""
h = probs.get("home", "?")
d = probs.get("draw", "?")
a = probs.get("away", "?")
return f"model {h}%/{d}%/{a}% ({home}/{d}/{away})"
CSS = """
"""
def _load(filename, default):
p = BASE / filename
if not p.exists():
return default
try:
return json.loads(p.read_text())
except Exception:
return default
def _load_text(filename):
p = BASE / filename
return p.read_text() if p.exists() else ""
_STAGE_ORDER = ["Group Stage", "Round Of 32", "Round Of 16", "Quarterfinals", "Semifinals", "Final"]
def _stage_indicator(schedule: list, results: list) -> str:
"""Return HTML for a tournament stage progress indicator."""
now = datetime.now(timezone.utc)
predicted = {(r["home"], r["away"]) for r in results}
# Count past/future per stage; find current stage (earliest with future unpredicted games)
stage_stats: dict[str, dict] = {}
for f in schedule:
k = f.get("kickoff_utc", "")
try:
dt = datetime.fromisoformat(k.replace("Z", "+00:00"))
except Exception:
continue
st = f.get("stage", "")
if not st:
continue
s = stage_stats.setdefault(st, {"total": 0, "past": 0, "future_unpredicted": 0})
s["total"] += 1
if dt <= now:
s["past"] += 1
elif (f["home"], f["away"]) not in predicted:
s["future_unpredicted"] += 1
# Determine which stages exist in this schedule (in order)
present_stages = [s for s in _STAGE_ORDER if s in stage_stats]
if not present_stages:
return ""
# Current stage = first one that still has future unpredicted matches
current_stage = next(
(s for s in present_stages if stage_stats[s]["future_unpredicted"] > 0),
present_stages[-1],
)
# Build pipeline HTML
pills = ""
for i, st in enumerate(present_stages):
stats = stage_stats[st]
is_current = st == current_stage
is_done = present_stages.index(st) < present_stages.index(current_stage)
if is_current:
played = stats["past"]
total = stats["total"]
pct = int(played / total * 100) if total else 0
bg = "#1a2a3a"
border = "#4dabf7"
color = "#4dabf7"
label = f'{st} {played}/{total} played'
progress = f''
elif is_done:
bg = "#0d1a0d"
border = "#2a3a2a"
color = "#4caf50"
label = f'✓ {st}'
progress = ""
else:
bg = "#111"
border = "#222"
color = "#444"
label = f'{st}'
progress = ""
connector = '›
' if i < len(present_stages) - 1 else ""
pills += f'{connector}'
return f'{pills}
'
def _dedupe_results(results: list) -> list:
"""Keep the best entry per fixture — prefer reflected over unreflected."""
latest: dict[tuple, dict] = {}
for r in results:
key = (r.get("home"), r.get("away"))
prev = latest.get(key)
if prev is None:
latest[key] = r
elif r.get("reflected") and not prev.get("reflected"):
latest[key] = r
elif prev.get("reflected") and not r.get("reflected"):
continue
else:
latest[key] = r
return list(latest.values())
def _kickoff_dt(r_or_fixture) -> datetime | None:
k = r_or_fixture.get("kickoff") or r_or_fixture.get("kickoff_utc") or ""
try:
return datetime.fromisoformat(k.replace("Z", "+00:00"))
except Exception:
return None
def _upcoming(schedule, results):
now = datetime.now(timezone.utc)
predicted = {(r["home"], r["away"]) for r in results}
out = []
for f in schedule:
dt = _kickoff_dt(f)
if dt is None:
continue
if dt > now and (f["home"], f["away"]) not in predicted:
out.append((dt, f))
sorted_all = [f for _, f in sorted(out, key=lambda x: x[0])]
return sorted_all[:8], len(sorted_all)
def _match_state(r: dict) -> dict:
"""Classify a predicted match: reflected, or awaiting reflection (with sub-status)."""
if r.get("reflected"):
if r.get("correct") is True:
badge = '✓ REFLECTED'
else:
badge = '✗ REFLECTED'
return {
"bucket": "reflected",
"sublabel": "Final result scored · strategy updated",
"badge": badge,
}
kdt = _kickoff_dt(r)
now = datetime.now(timezone.utc)
if kdt is not None:
secs_since = (now - kdt).total_seconds()
if secs_since > 7200:
return {
"bucket": "awaiting_reflection",
"sublabel": "Match finished · reflection pending (~2.5h after kickoff)",
"badge": 'AWAITING REFLECTION',
}
if secs_since > 0:
return {
"bucket": "awaiting_reflection",
"sublabel": "Match in progress · reflection after final whistle",
"badge": '● LIVE',
}
rel, _ = _time_rel(r.get("kickoff", ""))
return {
"bucket": "awaiting_reflection",
"sublabel": f"Pick locked in · kicks off in {rel}",
"badge": 'PREDICTED',
}
return {
"bucket": "awaiting_reflection",
"sublabel": "Pick locked in · awaiting kickoff",
"badge": 'PREDICTED',
}
def _badge(r):
return _match_state(r)["badge"]
def _kickoff_passed(r: dict) -> bool:
kdt = _kickoff_dt(r)
if kdt is None:
return False
return (datetime.now(timezone.utc) - kdt).total_seconds() > 0
def _awaiting_reflection(r: dict) -> bool:
return not r.get("reflected") and _kickoff_passed(r)
def _is_live(r: dict) -> bool:
"""Match kicked off recently — likely still in progress (~first 2h)."""
if r.get("reflected") or not _kickoff_passed(r):
return False
kdt = _kickoff_dt(r)
if kdt is None:
return False
secs = (datetime.now(timezone.utc) - kdt).total_seconds()
return 0 < secs <= 7200
def _pending_reflection(r: dict) -> bool:
"""Kickoff passed, not reflected, and not still in the live window."""
return _awaiting_reflection(r) and not _is_live(r)
def _upcoming_predicted(r: dict) -> bool:
return not r.get("reflected") and not _kickoff_passed(r)
def _awaiting_reflect_sort_key(r: dict) -> tuple:
kdt = _kickoff_dt(r)
now = datetime.now(timezone.utc)
if kdt is None:
return (9, 0)
secs_since = (now - kdt).total_seconds()
if secs_since > 7200:
priority = 0
elif secs_since > 0:
priority = 1
else:
priority = 2
return (priority, -kdt.timestamp())
def _kickoff_sort_key(r: dict) -> float:
kdt = _kickoff_dt(r)
return kdt.timestamp() if kdt else 0.0
def _render_match_card(r: dict, names: dict[str, str] | None = None) -> str:
pred = r.get("prediction", {})
res = r.get("result")
used = r.get("services_used", [])
planned = r.get("services_planned", [])
state = _match_state(r)
if names is None:
names = _team_names()
x402_html = ""
if used:
rows = "".join(
f'| {_esc(s["source"])} | ${s["cost"]:.4f} | '
f'{_esc(s.get("reason", ""))} |
'
for s in used
)
x402_html = f'x402 data purchased
'
elif planned:
x402_html = f'{len(planned)} endpoint(s) considered, none purchased.
'
else:
x402_html = 'No x402 data for this match.
'
result_html = ""
if res and r.get("reflected"):
cls = "correct" if r.get("correct") else "incorrect"
eval_snippet = _esc((r.get("evaluation") or "")[:180])
result_html = f'''
Result:
{_team_full(r['home'], names)} {res['home_score']}–{res['away_score']} {_team_full(r['away'], names)}
{f'
{eval_snippet}{"…" if len(r.get("evaluation", "")) > 180 else ""}
' if eval_snippet else ""}
Provisional score:
{_team_full(r['home'], names)} {res['home_score']}–{res['away_score']} {_team_full(r['away'], names)}
Official reflection not run yet — win/loss not scored.
Pick
{_pick_display_full(pred.get('pick', '?'), r['home'], r['away'], names)}
Confidence
{pred.get("confidence", "?")}/10{_esc(pred.get("confidence_reason", "Hover reason not available for this prediction."))}
Research cost
${r.get("research_cost", 0):.4f}
"{_esc(pred.get("reasoning", ""))}"
{x402_html}
{result_html}
Full reasoning & reflection →
"""
def _pipeline_html(awaiting_pred: int, awaiting_reflect: int, reflected: int, wins: int, losses: int) -> str:
record = f"{wins}W–{losses}L" if reflected else "—"
return f"""
{awaiting_pred}
Awaiting prediction
→
{awaiting_reflect}
Predicted · awaiting reflection
→
{reflected}
Reflected ({record})
"""
def _sidebar_html(active_page: str, sections: list[dict] | None = None) -> str:
pages = [
("dashboard", "/", "Dashboard"),
("learnings", "/learnings", "Learnings"),
("strategy", "/strategy", "Strategy"),
]
page_links = ""
for key, href, label in pages:
cls = "sidebar-link active" if active_page == key else "sidebar-link"
page_links += f'{label}'
section_links = ""
if sections:
for sec in sections:
sid = sec["id"]
label = sec["label"]
count = sec.get("count")
extra = " live-link" if sid == "live" else ""
active = " active" if sec.get("active") else ""
badge = f'' if count is not None else ""
section_links += f''
sections_block = ""
if section_links:
sections_block = f""""""
return f""""""
def _time_rel(kickoff_str: str) -> tuple[str, str]:
"""Return (label, color) for a kickoff time relative to now."""
try:
kdt = datetime.fromisoformat(kickoff_str.replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
delta = kdt - now
secs = delta.total_seconds()
abs_secs = abs(secs)
if abs_secs < 3600:
mins = int(abs_secs / 60)
label = f"{mins}m {'away' if secs > 0 else 'ago'}"
elif abs_secs < 86400:
hrs = abs_secs / 3600
label = f"{hrs:.1f}h {'away' if secs > 0 else 'ago'}"
else:
days = abs_secs / 86400
label = f"{days:.1f}d {'away' if secs > 0 else 'ago'}"
if secs > 0:
color = "#4caf50" if secs > 86400 else ("#f9a825" if secs > 3600 else "#f44336")
else:
color = "#555"
return label, color
except Exception:
return "", "#555"
def _esc(s):
return str(s).replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """)
def _render_md(text: str) -> str:
if not text or not text.strip():
return ""
return markdown.markdown(text, extensions=["sane_lists", "nl2br"])
def _md(text: str, extra_class: str = "") -> str:
cls = f"strategy-md {extra_class}".strip()
return f'{_render_md(text)}
'
def _page(body, title="⚽ World Cup Prediction Agent", active_page="dashboard", sections=None):
header = """
⚽ World Cup Prediction Agent
"""
sidebar = _sidebar_html(active_page, sections)
return HTMLResponse(
f""
f""
f"{title}{CSS}"
f"{header}"
f"
{sidebar}{body}
"
f"
{_team(f['home'])} vs {_team(f['away'])} {_esc(f.get('stage', ''))}
Agent runs ~30 min before kickoff
AWAITING PREDICTION
{rel} ({f['kickoff_utc'][:16].replace('T', ' ')} UTC)
"""
footer = ""
if upcoming_total > len(upcoming):
footer = f'Showing next {len(upcoming)} of {upcoming_total} fixtures without a prediction
'
html += f"""
Awaiting Prediction
Fixtures on the schedule with no pick yet. predict.py runs automatically ~30 minutes before kickoff.
{stage_bar}{items}{footer}
"""
else:
html += """
Awaiting Prediction
Fixtures on the schedule with no pick yet. predict.py runs automatically ~30 minutes before kickoff.
No unpredicted fixtures coming up — every scheduled match has a pick, or the group stage is complete.
"""
# ── 2. Predicted — upcoming ───────────────────────────────────────────
if upcoming_predicted:
cards = "".join(_render_match_card(r, team_names) for r in upcoming_predicted)
html += f"""
Predicted — Upcoming
Pick is locked in before kickoff. Reflection runs automatically ~2.5h after the final whistle.
{cards}
"""
# ── 3. Live ─────────────────────────────────────────────────────────────
html += '● Live
'
html += '
Matches in progress — kickoff passed, final whistle not yet reflected.
'
if live_matches:
cards = "".join(_render_match_card(r, team_names) for r in live_matches)
html += cards + "
No live matches right now.
'
# ── 4. Predicted — awaiting reflection ────────────────────────────────
html += 'Predicted — Awaiting Reflection
'
html += '
Match finished. reflect.py fetches the result, scores the prediction, and updates strategy (~2.5h after kickoff).
'
if pending_reflect:
cards = "".join(_render_match_card(r, team_names) for r in pending_reflect)
html += cards + "
Nothing here — no finished matches waiting on reflection.
'
# ── 5. Reflected ────────────────────────────────────────────────────────
html += 'Reflected
'
html += '
Match complete. Prediction scored against the final result and learnings applied to strategy.
'
if reflected:
reflected_sorted = sorted(
reflected,
key=lambda r: r.get("reflected_at") or "",
reverse=True,
)
cards = "".join(_render_match_card(r, team_names) for r in reflected_sorted)
html += cards + "
No reflections yet — they appear here after the first match finishes and reflect.py runs.
'
# Strategy snippet
if strategy:
snippet = _md(strategy[:500])
more = '' if len(strategy) > 500 else ""
html += f''
return _page(html, sections=sections)
@app.get("/match/{match_key}", response_class=HTMLResponse)
async def match_detail(match_key: str):
home, _, away = match_key.partition("_")
home, away = home.upper(), away.upper()
results = _load("results.json", [])
entry = next(
(r for r in reversed(_dedupe_results(results))
if r.get("home") == home and r.get("away") == away),
None,
)
if not entry:
raise HTTPException(status_code=404, detail=f"No prediction found for {home} vs {away}")
pred = entry.get("prediction", {})
res = entry.get("result")
used = entry.get("services_used", [])
planned = entry.get("services_planned", [])
state = _match_state(entry)
result_html = ""
if res and entry.get("reflected"):
cls = "correct" if entry.get("correct") else "incorrect"
result_html = f'{_team(home)} {res["home_score"]}–{res["away_score"]} {_team(away)} {state["badge"]}
'
elif res:
result_html = f'''
{_team(home)} {res["home_score"]}–{res["away_score"]} {_team(away)} {state["badge"]}
Provisional score — win/loss not scored until reflection runs.
'''
else:
result_html = f'{state["badge"]}
{_esc(state["sublabel"])}
{_esc(s['source'])}
${s['cost']:.4f} USDC
Why purchased: {_esc(s.get('reason',''))}
Data received:
{data_content}
"""
# Summary table of all planned endpoints
plan_rows = ""
for s in planned:
if s["url"] in used_urls:
status = '✓ purchased'
else:
status = 'skipped'
plan_rows += f'| {_esc(s["url"])} | ${s.get("cost",0):.4f} | {status} | {_esc(s.get("reason",""))} |
'
x402_section = ""
if planned or used:
total = entry.get("research_cost", 0)
budget = 0.50
cards_html = endpoint_cards if endpoint_cards else 'No endpoints were purchased.
'
table_html = f'| Endpoint | Cost | Status | Reason |
|---|
{plan_rows}
' if plan_rows else ""
x402_section = f"""
x402 Data Purchases — ${total:.4f} spent of ${budget:.2f} budget
{len(used)} of {len(planned)} planned endpoints purchased
{cards_html}
{f'
All considered endpoints
{table_html}
' if table_html else ""}
"""
eval_html = f'Reflection & Evaluation
{_esc(entry.get("evaluation","(not yet reflected)"))}
Strategy at Prediction Time
{_render_md(entry.get("strategy_snapshot","")) or "(empty)"}
Analyst Information Needs (Stage 1)
{_esc(entry["information_needs"])}
| {_esc(g.get("category","?"))} | {_esc(g.get("need",""))} | {_esc(g.get("ideal_service",""))} |
'
for g in entry["research_gaps"]
)
gaps_html = f'''Research Gaps — No x402 Service Available
These intelligence needs had no matching service in the directory. This is a demand signal for the x402 ecosystem.
| Category | Need | Ideal Service |
|---|
{gap_rows}
{_esc(entry["match"])}
{entry.get("kickoff","")[:16].replace("T"," ")} UTC
{result_html}
Prediction
Pick
{_team(_pick_display(pred.get('pick','?'), home, away))}
Confidence
{pred.get("confidence","?")}/10{_esc(pred.get("confidence_reason","Hover reason not available for this prediction."))}
{f'
Win model: {_esc(_prob_summary(pred, home, away))}
' if _prob_summary(pred, home, away) else ""}
"{_esc(pred.get("reasoning",""))}"
Full Gemini Reasoning
{_esc(entry.get("full_reasoning","(none)"))}
{_md(strategy)}
' if strategy else 'No strategy yet — will be written after the first match reflection.
After {_esc(match)} ({date})
{_md(snap_text + suffix)}
Strategy Evolution
{items}'
body = f'Current Strategy
{current}{hist_html}'
return _page(body, "Strategy — World Cup Predictions", active_page="strategy")
def _strategy_history_from_results(results: list) -> list[dict]:
"""Build strategy evolution from reflected entries in results.json."""
by_match: dict[str, dict] = {}
for r in results:
if not r.get("reflected"):
continue
match = r.get("match") or ""
if not match:
continue
prev = by_match.get(match)
if prev is None or (r.get("reflected_at") or "") >= (prev.get("reflected_at") or ""):
by_match[match] = r
entries = sorted(by_match.values(), key=lambda r: r.get("reflected_at") or "")
if not entries:
return []
current = _load_text("strategy.md")
history = []
for i, r in enumerate(entries):
match = r["match"]
try:
dt = datetime.fromisoformat((r.get("reflected_at") or "").replace("Z", "+00:00"))
except Exception:
dt = None
prev_content = r.get("strategy_snapshot") or ""
if not prev_content.strip() and i > 0:
prev_content = history[i - 1]["content"]
content = r.get("strategy_after") or ""
if not content.strip():
if i == len(entries) - 1:
content = current
else:
nxt = entries[i + 1].get("strategy_snapshot") or entries[i + 1].get("strategy_after") or ""
content = nxt
history.append({
"match": match,
"date": dt,
"subject": f"reflect: {match}",
"content": content,
"prev_content": prev_content,
})
return history
def _strategy_history_from_git() -> list[dict]:
"""Return strategy evolution from git log (local dev fallback)."""
try:
log = subprocess.run(
["git", "log", "--pretty=format:%H|%ai|%s", "--", "strategy.md"],
cwd=BASE, capture_output=True, text=True, check=True,
)
except Exception:
return []
entries = []
lines = [l for l in log.stdout.strip().splitlines() if l]
for line in lines:
parts = line.split("|", 2)
if len(parts) != 3:
continue
sha, date_str, subject = parts
match_name = subject.replace("reflect: ", "").replace("init: ", "").strip()
try:
dt = datetime.fromisoformat(date_str.strip())
except Exception:
dt = None
try:
content = subprocess.run(
["git", "show", f"{sha}:strategy.md"],
cwd=BASE, capture_output=True, text=True, check=True,
).stdout
except Exception:
content = ""
entries.append({"match": match_name, "date": dt, "sha": sha, "subject": subject, "content": content})
entries.reverse()
latest_reflect_idx: dict[str, int] = {}
for i, e in enumerate(entries):
if e["subject"].startswith("reflect:"):
latest_reflect_idx[e["match"]] = i
entries = [
e for i, e in enumerate(entries)
if not e["subject"].startswith("reflect:") or latest_reflect_idx.get(e["match"]) == i
]
for i, e in enumerate(entries):
e["prev_content"] = entries[i - 1]["content"] if i > 0 else ""
return entries
def _strategy_history() -> list[dict]:
"""Strategy evolution for the Learnings page — results.json first, git as fallback."""
results = _load("results.json", [])
history = _strategy_history_from_results(results)
if history:
git_by_match = {
e["match"]: e for e in _strategy_history_from_git()
if e["subject"].startswith("reflect:")
}
for e in history:
g = git_by_match.get(e["match"])
if not g:
continue
if not (e.get("content") or "").strip():
e["content"] = g["content"]
if not (e.get("prev_content") or "").strip():
e["prev_content"] = g["prev_content"]
return history
return _strategy_history_from_git()
def _text_diff_html(old: str, new: str) -> str:
"""Produce simple line-level added/removed/unchanged HTML."""
old_lines = old.splitlines()
new_lines = new.splitlines()
old_set = set(old_lines)
new_set = set(new_lines)
parts = []
for line in new_lines:
esc = _esc(line)
if line not in old_set:
parts.append(f'+ {esc}
')
else:
parts.append(f' {esc}
')
for line in old_lines:
if line not in new_set:
esc = _esc(line)
parts.append(f'- {esc}
')
return "\n".join(parts) if parts else '(no change)
'
@app.get("/learnings", response_class=HTMLResponse)
async def learnings_page():
history = _strategy_history()
results = _load("results.json", [])
# Index evaluations by match name for quick lookup
evals = {r.get("match", ""): r for r in results if r.get("evaluation")}
DIFF_CSS = """
"""
body = DIFF_CSS
reflect_entries = [e for e in reversed(history) if e["subject"].startswith("reflect:")]
if not reflect_entries:
body += 'No reflections recorded yet — learnings appear after each match is reflected.
'
return _page(body, "Learnings — World Cup Predictions", active_page="learnings")
cards = ""
for e in reflect_entries:
match_name = e["match"]
date_str = e["date"].strftime("%Y-%m-%d %H:%M UTC") if e["date"] else ""
result_entry = evals.get(match_name, {})
# Outcome badge
if result_entry.get("correct") is True:
badge = '✓ CORRECT'
elif result_entry.get("correct") is False:
badge = '✗ INCORRECT'
else:
badge = ""
# Score
res = result_entry.get("result")
score_html = ""
if res:
score_html = f'{_esc(result_entry.get("home",""))} {res["home_score"]}–{res["away_score"]} {_esc(result_entry.get("away",""))}
'
# Evaluation
eval_text = result_entry.get("evaluation", "")
eval_html = ""
if eval_text:
eval_html = f'Evaluator assessment
{_esc(eval_text)}
'
# Strategy diff
diff_html = _text_diff_html(e["prev_content"], e["content"])
is_first = not e["prev_content"].strip()
diff_label = "Initial strategy written" if is_first else "Strategy change"
cards += f"""
{score_html}
{eval_html}
{diff_label}
{diff_html}
"""
body += f'Strategy Evolution — {len(reflect_entries)} reflection{"s" if len(reflect_entries) != 1 else ""}
{cards}'
return _page(body, "Learnings — World Cup Predictions", active_page="learnings")
@app.get("/api/results")
async def api_results():
return _load("results.json", [])
@app.get("/api/schedule")
async def api_schedule():
return _load("schedule.json", [])