Blame for backend/data.py

"""
Read-only data layer over lake/gold/nfl/*.csv.

Deliberately does NOT touch lake/nfl.duckdb -- that file is gitignored and
not guaranteed to exist on a fresh clone (see docs/HANDOFF.md sec 5). Every
gold CSV under lake/gold/nfl/ is tracked in git and is the source of truth
this backend reads.

Per HANDOFF_CLAUDE_CODE.md sec 0, rule 1: only a chart whose _index.csv row
has status == "ok" is ever rendered with real data -- anything else returns
an explicit empty/stale/error state, never a silent fallback to placeholder
values.

Known real gap, documented rather than worked around silently: prop_best_price
/ prop_line_rotowire key players by RotoWire's own numeric player_id, not the
nflverse gsis_id used everywhere else in the lake (docs/HANDOFF.md sec 5.8 --
the id bridge table exists but has never been populated). This layer joins
those two worlds by normalized player name, the same workaround
scripts/matchup_report.py already uses for the same reason.
"""
from __future__ import annotations

import csv
import pathlib
import re
import threading
from functools import lru_cache

REPO = pathlib.Path(__file__).resolve().parent.parent
GOLD = REPO / "lake" / "gold" / "nfl"

_lock = threading.Lock()


def _read_csv(name: str) -> list[dict]:
    path = GOLD / f"{name}.csv"
    if not path.exists():
        return []
    with path.open(newline="", encoding="utf-8") as fh:
        return list(csv.DictReader(fh))


@lru_cache(maxsize=64)
def _cached_csv(name: str, _version: int) -> list[dict]:
    return _read_csv(name)


_VERSION = 0  # bump to invalidate the cache (not needed -- data is static per session)


def load(name: str) -> list[dict]:
    with _lock:
        return _cached_csv(name, _VERSION)


def normalize_name(name: str) -> str:
    """Loose match key: lowercase, strip punctuation/suffixes, collapse spaces."""
    n = name.lower()
    n = re.sub(r"[.'\-]", "", n)
    n = re.sub(r"\b(jr|sr|ii|iii|iv)\b", "", n)
    n = re.sub(r"\s+", " ", n).strip()
    return n


# ---------------------------------------------------------------------------
# _index.csv -- the chart-status gate every endpoint must honor
# ---------------------------------------------------------------------------

def chart_index() -> dict[str, dict]:
    rows = load("_index")
    return {r["chart_name"]: r for r in rows}


def chart_status(chart_name: str) -> str:
    row = chart_index().get(chart_name)
    return row["status"] if row else "error"


# ---------------------------------------------------------------------------
# Player dimension -- gsis_id -> {name, team, position}, built from whichever
# gold CSVs actually carry gsis ids + names (player_photos/_index.csv is the
# most complete display-name source; team/position resolved from the most
# recent week seen in player_scrimmage_week.csv).
# ---------------------------------------------------------------------------

@lru_cache(maxsize=1)
def player_dimension() -> dict[str, dict]:
    photos = {r["player_id"]: r["player_display_name"] for r in load("player_photos/_index") if r.get("player_id")}
    dim: dict[str, dict] = {}
    for row in load("player_scrimmage_week"):
        pid = row.get("player_id")
        if not pid:
            continue
        wk = int(row.get("week") or 0)
        existing = dim.get(pid)
        if existing is None or wk >= existing["_week"]:
            dim[pid] = {
                "player_id": pid,
                "name": photos.get(pid, pid),
                "team": row.get("team", ""),
                "_week": wk,
            }
    # position isn't on the weekly stat tables; player_usage.csv is the one
    # gold export that carries it, keyed by (season, team, player_id).
    for row in load("player_usage"):
        pid = row.get("player_id")
        if pid in dim:
            dim[pid]["position"] = row.get("position", "")
    for pid, row in dim.items():
        row.setdefault("name", photos.get(pid, pid))
        row.setdefault("position", "")
        row.pop("_week", None)
    return dim


def player_name(player_id: str) -> str:
    dim = player_dimension()
    if player_id in dim:
        return dim[player_id]["name"]
    photos = {r["player_id"]: r["player_display_name"] for r in load("player_photos/_index")}
    return photos.get(player_id, player_id)


def player_team(player_id: str) -> str:
    return player_dimension().get(player_id, {}).get("team", "")


@lru_cache(maxsize=1)
def _headshots() -> dict[str, str]:
    return {r["player_id"]: r["headshot_url"] for r in load("player_headshots")}


def _enhance_cloudinary(url: str | None, w: int = 400, h: int = 400) -> str | None:
    """Inject face-crop + size transforms into NFL.com Cloudinary headshot URLs."""
    if not url:
        return url
    # Pattern: .../image/upload/<existing_transforms>/...
    import re
    m = re.match(r"(https://static\.www\.nfl\.com/image/upload/)([^/]+)(/.+)", url)
    if m:
        return f"{m.group(1)}w_{w},h_{h},c_fill,g_face,{m.group(2)}{m.group(3)}"
    return url


def photo_url(player_id: str) -> str | None:
    """Tank01's ESPN headshot first (matched on name + team), nflverse's headshot as fallback.
    The browser loads the URL directly; the frontend shows an initials avatar if it fails."""
    d = player_dimension().get(player_id)
    if d:
        try:
            import tank01
            hits = tank01.player_photos().get(normalize_name(d["name"]), []) if tank01.available() else []
        except Exception:
            hits = []
        same_team = [u for t, u in hits if t == d.get("team")]
        if same_team:
            return same_team[0]
        if len(hits) == 1:
            return hits[0][1]
    return _enhance_cloudinary(_headshots().get(player_id))


def search_players(query: str, limit: int = 20) -> list[dict]:
    q = normalize_name(query)
    if not q:
        return []
    out = []
    for pid, row in player_dimension().items():
        if q in normalize_name(row["name"]):
            out.append({"playerId": pid, "name": row["name"], "team": row["team"]})
    return out[:limit]


# ---------------------------------------------------------------------------
# Prop lines -- RotoWire, name-matched to the gsis player dimension
# ---------------------------------------------------------------------------

MARKET_TO_STAT = {
    "rushyds": ("player_rushing_week", "rushing_yards"),
    "recyds": ("player_receiving_week", "receiving_yards"),
    "recs": ("player_receiving_week", "receptions"),
    "passyds": ("player_passing_week", "passing_yards"),
    "passtd": ("player_passing_week", "passing_tds"),
}

PROP_LABELS = {
    "rushyds": "Rush Yds",
    "recyds": "Rec Yds",
    "recs": "Receptions",
    "passyds": "Pass Yds",
    "passtd": "Pass TDs",
    "anytd": "Any TD",
}


@lru_cache(maxsize=1)
def _name_to_best_price() -> dict[str, list[dict]]:
    out: dict[str, list[dict]] = {}
    for row in load("prop_best_price"):
        key = normalize_name(row.get("player_name", ""))
        out.setdefault(key, []).append(row)
    return out


def best_price_for(player_name_str: str, market_slug: str) -> dict | None:
    key = normalize_name(player_name_str)
    for row in _name_to_best_price().get(key, []):
        if row.get("market_slug") == market_slug:
            return row
    return None


@lru_cache(maxsize=1)
def _name_to_rotowire_fanduel() -> dict[str, list[dict]]:
    out: dict[str, list[dict]] = {}
    for row in load("prop_line_rotowire"):
        if row.get("book_slug") != "fanduel":
            continue
        key = normalize_name(row.get("player_name", ""))
        out.setdefault(key, []).append(row)
    return out


def fanduel_line_for(player_name_str: str, market_slug: str) -> dict | None:
    """
    The real FanDuel line/price for a player+market, per the user's explicit
    choice (FanDuel only, not best-price-across-books). Reads
    prop_line_rotowire.csv filtered to book_slug == 'fanduel' -- the raw,
    single-book source -- rather than prop_best_price.csv, which mixes
    whichever book happened to have the best number per market. Falls back
    to prop_best_price only if no FanDuel row exists at all for this
    player/market, and labels the book honestly either way.
    """
    key = normalize_name(player_name_str)
    candidates = [r for r in _name_to_rotowire_fanduel().get(key, []) if r.get("market_slug") == market_slug]
    if candidates:
        # multiple pulls/snapshots can exist; take the most recently fetched
        latest = max(candidates, key=lambda r: r.get("fetched_at_utc", ""))
        return {
            "line": latest.get("line"),
            "book": "fanduel",
            "over_price": latest.get("over_price_american"),
            "under_price": latest.get("under_price_american"),
            "moneyline_price": latest.get("moneyline_american"),
        }
    fallback = best_price_for(player_name_str, market_slug)
    if fallback:
        return {
            "line": fallback.get("line"),
            "book": fallback.get("best_over_book") or fallback.get("best_moneyline_book") or "unknown",
            "over_price": fallback.get("best_over_price"),
            "under_price": fallback.get("best_under_price"),
            "moneyline_price": fallback.get("best_moneyline_price"),
        }
    return None


# ---------------------------------------------------------------------------
# Weekly game log for a player + prop, with hit-rate vs the current line
# ---------------------------------------------------------------------------

def _num(v, digits: int = 3):
    try:
        return round(float(v), digits)
    except (TypeError, ValueError):
        return None


def player_trust(player_id: str) -> dict | None:
    """Usage trust, red-zone tiers and TD conversion for one player (three gold tables)."""
    usage = next((r for r in load("player_usage") if r.get("player_id") == player_id), None)
    corr = next((r for r in load("player_usage_td_correlation") if r.get("player_id") == player_id), None)
    tiers = [r for r in load("redzone_tiers") if r.get("player_id") == player_id]
    if not (usage or corr or tiers):
        return None
    row = usage or corr or {}
    return {
        "usageIndex": _num(row.get("usage_index_score"), 1),
        "role": row.get("usage_role", ""),
        "shareOverall": _num(row.get("share_overall")),
        "shareCalm": _num(row.get("share_calm")),
        "shareStress": _num(row.get("share_stress")),
        "confidenceDelta": _num(row.get("confidence_delta")),
        "redZone": [
            {
                "type": t.get("opportunity_type", ""),
                "tiers": [
                    {"label": label, "opps": _num(t.get(f"opp_{k}"), 0), "tds": _num(t.get(f"td_{k}"), 0),
                     "share": _num(t.get(f"pct_{k}_intra"))}
                    for label, k in (("INSIDE 20", "i20"), ("INSIDE 10", "i10"), ("INSIDE 5", "i5"), ("GOAL TO GO", "g2g"))
                    if t.get(f"opp_{k}") not in (None, "")
                ],
            }
            for t in tiers
        ],
        "td": {
            "total": _num(corr.get("tds"), 0), "redZone": _num(corr.get("tds_red_zone"), 0),
            "goalToGo": _num(corr.get("tds_goal_to_go"), 0), "rushing": _num(corr.get("tds_rush"), 0),
            "receiving": _num(corr.get("tds_rec"), 0), "gamesWithTd": _num(corr.get("games_with_td"), 0),
            "perRedZoneOpp": _num(corr.get("td_per_redzone_opp")), "flag": corr.get("correlation_flag", ""),
        } if corr else None,
    }


def player_prop_chart(player_id: str, market_slug: str) -> dict:
    name = player_name(player_id)
    team = player_team(player_id)
    status = chart_status(MARKET_TO_STAT.get(market_slug, ("prop_best_price",))[0])

    if market_slug == "anytd":
        games = anytime_td_games(player_id)
        line_row = None
        line = 0.5
        book = None
    else:
        src, stat_col = MARKET_TO_STAT.get(market_slug, (None, None))
        if src is None:
            return {"error": f"unknown market_slug {market_slug}"}
        games = weekly_games(player_id, src, stat_col)
        line_row = fanduel_line_for(name, market_slug)
        line = float(line_row["line"]) if line_row and line_row.get("line") else None
        book = line_row.get("book") if line_row else None

    games_sorted = sorted(games, key=lambda g: g["week"])
    bars = [{"gameDate": f"Week {g['week']}", "opponent": g["opponent"], "value": g["value"]} for g in games_sorted]

    def hit_rate(window: list[dict]) -> float | None:
        if not window or line is None:
            return None
        hits = sum(1 for g in window if g["value"] >= line)
        return round(hits / len(window), 3)

    l5 = games_sorted[-5:]
    l10 = games_sorted[-10:]
    l20 = games_sorted[-20:]

    import bpl
    official = bpl.nfl_player_line(player_id, market_slug, name, team) if market_slug in bpl.NFL_MARKETS else None
    return {
        "playerId": player_id,
        "name": name,
        "team": team,
        "position": player_dimension().get(player_id, {}).get("position", ""),
        "bpl": official,
        "trust": player_trust(player_id),
        "prop": PROP_LABELS.get(market_slug, market_slug),
        "marketSlug": market_slug,
        "line": line,
        "book": book or "FanDuel",
        "games": bars,
        "splits": [
            {"label": "L5", "hitRate": hit_rate(l5)},
            {"label": "L10", "hitRate": hit_rate(l10)},
            {"label": "L20", "hitRate": hit_rate(l20)},
        ],
        "sourceStatus": status,
        "photoUrl": photo_url(player_id),
    }


def weekly_games(player_id: str, src: str, stat_col: str) -> list[dict]:
    out = []
    for row in load(src):
        if row.get("player_id") != player_id:
            continue
        try:
            value = float(row.get(stat_col) or 0)
        except ValueError:
            value = 0.0
        out.append({"week": int(row.get("week") or 0), "opponent": row.get("opponent_team", ""), "value": value})
    return out


def anytime_td_games(player_id: str) -> list[dict]:
    out = []
    for row in load("player_scoring_week"):
        if row.get("player_id") != player_id:
            continue
        tds = float(row.get("total_tds") or 0)
        out.append({"week": int(row.get("week") or 0), "opponent": row.get("opponent_team", ""), "value": tds})
    return out


def hit_rate_probability(player_id: str, market_slug: str, window: int = 10) -> float | None:
    """
    The parlay-leg 'probability' score, per the user's explicit choice:
    hit-rate-vs-line only (no usage/toxicity/redzone weighting for v1).
    This is a documented model output, not a raw lake fact -- see
    sql/nfl_parlay_probability_schema.sql for the equivalent view definition
    and its own _index.csv row (added for when the lake is rebuilt with
    DuckDB in an environment that has it).
    """
    chart = player_prop_chart(player_id, market_slug)
    if chart.get("line") is None:
        return None
    games = sorted(chart["games"], key=lambda g: g["gameDate"])[-window:]
    if not games:
        return None
    hits = sum(1 for g in games if g["value"] >= chart["line"])
    return round(hits / len(games), 3)