#!/usr/bin/env python3 """ Sports-only server data mapper. Reads the sports data areas on sports-new and writes a reproducible inventory, relationship checks, and first-pass correlations. It is intentionally read-only against source databases and files. """ from __future__ import annotations import argparse import csv import datetime as dt import json import math import os import re import sqlite3 import statistics from collections import Counter, defaultdict from pathlib import Path from typing import Any, Iterable WORKSPACE = Path("/root/.openclaw/workspace") BACKUP = Path("/root/.openclaw/workspace-clawopen-backup") DATA = WORKSPACE / "ladder-framework" / "data" KALSHI = WORKSPACE / "kalshi-data" SPORTS = WORKSPACE / "sports" BACKUP_DATA = BACKUP / "ladder-framework" / "data" BACKUP_KALSHI = BACKUP / "kalshi-data" BACKUP_MLB = BACKUP / "mlb-package" BACKUP_MLB_MODEL = BACKUP / "mlb-model" BACKUP_KALSHI_TRACES = BACKUP / "temp_zip" / "nba_kalshi_spread_mkts_and_scores_for_andre" SPORTS_ROOTS = [ DATA, KALSHI, SPORTS, BACKUP_DATA, BACKUP_KALSHI, BACKUP_MLB, BACKUP_MLB_MODEL, BACKUP_KALSHI_TRACES, ] SKIP_DIRS = {"__pycache__", ".git", "node_modules", "__MACOSX"} SKIP_FILES = {".kalshi_creds.json"} TEAM_ALIASES = { "ATL": "ATL", "ATLANTA": "ATL", "ATLANTA HAWKS": "ATL", "BOS": "BOS", "BOSTON": "BOS", "BOSTON CELTICS": "BOS", "BKN": "BKN", "BRK": "BKN", "BROOKLYN": "BKN", "BROOKLYN NETS": "BKN", "NJN": "BKN", "NJ": "BKN", "CHA": "CHA", "CHO": "CHA", "CHARLOTTE": "CHA", "CHARLOTTE HORNETS": "CHA", "CHH": "CHA", "CHI": "CHI", "CHICAGO": "CHI", "CHICAGO BULLS": "CHI", "CLE": "CLE", "CLEVELAND": "CLE", "CLEVELAND CAVALIERS": "CLE", "DAL": "DAL", "DALLAS": "DAL", "DALLAS MAVERICKS": "DAL", "DEN": "DEN", "DENVER": "DEN", "DENVER NUGGETS": "DEN", "DET": "DET", "DETROIT": "DET", "DETROIT PISTONS": "DET", "GS": "GSW", "GSW": "GSW", "GOLDEN STATE": "GSW", "GOLDEN STATE WARRIORS": "GSW", "HOU": "HOU", "HOUSTON": "HOU", "HOUSTON ROCKETS": "HOU", "IND": "IND", "INDIANA": "IND", "INDIANA PACERS": "IND", "LAC": "LAC", "LA CLIPPERS": "LAC", "LOS ANGELES CLIPPERS": "LAC", "LAL": "LAL", "LA LAKERS": "LAL", "LOS ANGELES LAKERS": "LAL", "MEM": "MEM", "MEMPHIS": "MEM", "MEMPHIS GRIZZLIES": "MEM", "MIA": "MIA", "MIAMI": "MIA", "MIAMI HEAT": "MIA", "MIL": "MIL", "MILWAUKEE": "MIL", "MILWAUKEE BUCKS": "MIL", "MIN": "MIN", "MINNESOTA": "MIN", "MINNESOTA TIMBERWOLVES": "MIN", "NO": "NOP", "NOH": "NOP", "NOK": "NOP", "NOP": "NOP", "NEW ORLEANS": "NOP", "NEW ORLEANS HORNETS": "NOP", "NEW ORLEANS PELICANS": "NOP", "NY": "NYK", "NYK": "NYK", "NEW YORK": "NYK", "NEW YORK KNICKS": "NYK", "OKC": "OKC", "OKLAHOMA CITY": "OKC", "OKLAHOMA CITY THUNDER": "OKC", "SEA": "OKC", "ORL": "ORL", "ORLANDO": "ORL", "ORLANDO MAGIC": "ORL", "PHI": "PHI", "PHILADELPHIA": "PHI", "PHILADELPHIA 76ERS": "PHI", "PHO": "PHX", "PHX": "PHX", "PHOENIX": "PHX", "PHOENIX SUNS": "PHX", "POR": "POR", "PORTLAND": "POR", "PORTLAND TRAIL BLAZERS": "POR", "SAC": "SAC", "SACRAMENTO": "SAC", "SACRAMENTO KINGS": "SAC", "SA": "SAS", "SAS": "SAS", "SAN ANTONIO": "SAS", "SAN ANTONIO SPURS": "SAS", "TOR": "TOR", "TORONTO": "TOR", "TORONTO RAPTORS": "TOR", "UTA": "UTA", "UTAH": "UTA", "UTAH JAZZ": "UTA", "WAS": "WAS", "WSH": "WAS", "WASHINGTON": "WAS", "WASHINGTON WIZARDS": "WAS", "VAN": "MEM", "VANCOUVER GRIZZLIES": "MEM", } def norm_team(value: Any) -> str: if value is None: return "" raw = str(value).strip() if not raw: return "" key = re.sub(r"[^A-Za-z0-9 ]+", " ", raw).upper() key = re.sub(r"\s+", " ", key).strip() return TEAM_ALIASES.get(key, TEAM_ALIASES.get(key[:3], key)) def as_float(value: Any) -> float | None: if value is None: return None if isinstance(value, bool): return 1.0 if value else 0.0 if isinstance(value, (int, float)): if math.isfinite(float(value)): return float(value) return None text = str(value).strip().replace(",", "") if not text or text.lower() in {"none", "null", "nan", "true", "false"}: if text.lower() == "true": return 1.0 if text.lower() == "false": return 0.0 return None try: val = float(text) except ValueError: return None return val if math.isfinite(val) else None def parse_date(value: Any) -> str | None: if value is None: return None text = str(value).strip() if not text: return None if re.match(r"^\d{4}-\d{2}-\d{2}", text): return text[:10] for fmt in ( "%A, %d %b %Y, %H:%M", "%A, %d %b %Y, %H:%M", "%m/%d/%Y %H:%M:%S %Z", "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M:%S", ): try: return dt.datetime.strptime(re.sub(r"\s+", " ", text), fmt).date().isoformat() except ValueError: pass match = re.match(r"^(\d{4})-\d{2}-(\d{2})(\d{2})$", text) if match: # kyleskom legacy Date format such as 2013-14-1029. season_start = int(match.group(1)) month = int(match.group(2)) day = int(match.group(3)) year = season_start if month >= 7 else season_start + 1 try: return dt.date(year, month, day).isoformat() except ValueError: return None return None def pearson(rows: Iterable[dict[str, Any]], x_key: str, y_key: str) -> dict[str, Any] | None: xs: list[float] = [] ys: list[float] = [] for row in rows: x = as_float(row.get(x_key)) y = as_float(row.get(y_key)) if x is None or y is None: continue xs.append(x) ys.append(y) n = len(xs) if n < 3: return None x_mean = statistics.fmean(xs) y_mean = statistics.fmean(ys) cov = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, ys)) sx = math.sqrt(sum((x - x_mean) ** 2 for x in xs)) sy = math.sqrt(sum((y - y_mean) ** 2 for y in ys)) if sx == 0 or sy == 0: return None return { "x": x_key, "y": y_key, "n": n, "r": round(cov / (sx * sy), 6), } def sqlite_connect(path: Path) -> sqlite3.Connection: return sqlite3.connect(f"file:{path}?mode=ro", uri=True) def safe_count(cur: sqlite3.Cursor, table: str) -> int | None: try: return int(cur.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]) except Exception: return None def path_root(path: Path) -> str: for root in SPORTS_ROOTS: try: path.relative_to(root) return str(root) except ValueError: pass return str(path.parent) def classify_file(path: Path) -> str: text = str(path).lower() if "pikkit" in text: return "pikkit" if "kalshi" in text or "kxnba" in text: return "kalshi" if "kyleskom" in text: return "kyleskom" if "kaggle" in text: return "kaggle" if "mlb" in text: return "mlb" if "odds" in text or "spread" in text or "bet" in text: return "odds" if "nba" in text or "pbp" in text or "game" in text: return "nba" return "sports_model" def iter_sports_files() -> Iterable[Path]: seen: set[Path] = set() for root in SPORTS_ROOTS: if not root.exists(): continue if root.is_file(): yield root continue for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS] for name in filenames: if name in SKIP_FILES: continue path = Path(dirpath) / name if path in seen: continue seen.add(path) yield path def summarize_files() -> dict[str, Any]: files = [] by_root: dict[str, dict[str, Any]] = {} by_category: Counter[str] = Counter() by_extension: Counter[str] = Counter() duplicates: dict[str, list[str]] = defaultdict(list) total_bytes = 0 for path in iter_sports_files(): try: st = path.stat() except OSError: continue category = classify_file(path) ext = path.suffix.lower() or "[none]" root = path_root(path) total_bytes += st.st_size by_category[category] += 1 by_extension[ext] += 1 by_root.setdefault(root, {"files": 0, "bytes": 0}) by_root[root]["files"] += 1 by_root[root]["bytes"] += st.st_size duplicates[f"{path.name}|{st.st_size}"].append(str(path)) files.append( { "path": str(path), "name": path.name, "root": root, "category": category, "extension": ext, "bytes": st.st_size, "modified": dt.datetime.fromtimestamp(st.st_mtime, dt.timezone.utc).isoformat(), "backup": str(path).startswith(str(BACKUP)), } ) files.sort(key=lambda item: item["bytes"], reverse=True) duplicate_groups = [ {"name_size": key, "paths": paths} for key, paths in duplicates.items() if len(paths) > 1 ] duplicate_groups.sort(key=lambda item: len(item["paths"]), reverse=True) return { "total_files": len(files), "total_bytes": total_bytes, "by_root": by_root, "by_category": dict(by_category), "by_extension": dict(by_extension), "largest_files": files[:40], "duplicate_name_size_groups": duplicate_groups[:50], } def inspect_sqlite(path: Path, detail_table_limit: int = 60) -> dict[str, Any]: item: dict[str, Any] = {"path": str(path), "exists": path.exists()} if not path.exists(): return item item["bytes"] = path.stat().st_size try: con = sqlite_connect(path) cur = con.cursor() tables = [r[0] for r in cur.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY 1")] item["table_count"] = len(tables) total_rows = 0 table_details = [] for idx, table in enumerate(tables): cols = [r[1] for r in cur.execute(f'PRAGMA table_info("{table}")')] count = safe_count(cur, table) if count is not None: total_rows += count if idx < detail_table_limit: detail: dict[str, Any] = { "table": table, "rows": count, "columns": cols[:40], } for col in ("game_date", "date", "Date", "ts", "timestamp", "created_at", "snapshot_ts"): if col in cols: try: detail[f"{col}_range"] = cur.execute( f'SELECT MIN("{col}"), MAX("{col}") FROM "{table}"' ).fetchone() except Exception: pass for col in ("game_id", "nba_game_id", "game_key", "ticker", "event_ticker"): if col in cols: try: detail[f"distinct_{col}"] = cur.execute( f'SELECT COUNT(DISTINCT "{col}") FROM "{table}"' ).fetchone()[0] except Exception: pass table_details.append(detail) item["total_rows_counted"] = total_rows item["tables"] = table_details con.close() except Exception as exc: item["error"] = str(exc) return item def summarize_csv(path: Path) -> dict[str, Any]: result: dict[str, Any] = {"path": str(path), "exists": path.exists()} if not path.exists(): return result rows = 0 date_values: list[str] = [] teams = set() numeric_presence = Counter() with path.open(newline="", encoding="utf-8", errors="replace") as f: reader = csv.DictReader(f) headers = reader.fieldnames or [] for row in reader: rows += 1 for key in ("date", "Date", "game_date", "GAME_DATE", "time_placed_iso", "time_placed"): parsed = parse_date(row.get(key)) if parsed: date_values.append(parsed) break for key in ("home", "away", "Home", "Away", "Home_name", "Away_name", "home_team", "away_team"): team = norm_team(row.get(key)) if team: teams.add(team) for key, value in row.items(): if as_float(value) is not None: numeric_presence[key] += 1 result.update( { "rows": rows, "columns": headers, "date_min": min(date_values) if date_values else None, "date_max": max(date_values) if date_values else None, "distinct_teams_seen": len(teams), "numeric_columns_top": numeric_presence.most_common(20), } ) return result def summarize_json(path: Path) -> dict[str, Any]: result: dict[str, Any] = {"path": str(path), "exists": path.exists()} if not path.exists(): return result result["bytes"] = path.stat().st_size try: with path.open(encoding="utf-8", errors="replace") as f: obj = json.load(f) except Exception as exc: result["error"] = str(exc) return result if isinstance(obj, list): result["shape"] = "list" result["records"] = len(obj) if obj and isinstance(obj[0], dict): result["sample_keys"] = sorted(obj[0].keys())[:60] elif isinstance(obj, dict): result["shape"] = "dict" result["keys"] = sorted(obj.keys())[:60] result["top_level_lengths"] = { key: (len(value) if hasattr(value, "__len__") and not isinstance(value, (str, bytes)) else None) for key, value in list(obj.items())[:30] } else: result["shape"] = type(obj).__name__ return result def load_backtest_games() -> dict[str, Any]: db = DATA / "backtest.db" con = sqlite_connect(db) cur = con.cursor() rows = cur.execute( """ SELECT game_id, game_date, home_team, away_team, home_score, away_score, season FROM games """ ).fetchall() games = {} by_date_team = {} by_score = {} for game_id, date, home, away, home_score, away_score, season in rows: home_tri = norm_team(home) away_tri = norm_team(away) record = { "game_id": game_id, "date": date, "home": home_tri, "away": away_tri, "home_score": home_score, "away_score": away_score, "season": season, } games[game_id] = record by_date_team[(date, home_tri, away_tri)] = game_id by_score[(date, home_tri, away_tri, int(home_score or 0), int(away_score or 0))] = game_id con.close() return { "games": games, "by_date_team": by_date_team, "by_score": by_score, } def relationship_backtest_integrity(backtest: dict[str, Any]) -> dict[str, Any]: game_ids = set(backtest["games"]) con = sqlite_connect(DATA / "backtest.db") cur = con.cursor() result: dict[str, Any] = {"games": len(game_ids)} for table, col in ( ("pbp_actions", "game_id"), ("box_scores", "game_id"), ("historical_odds", "nba_game_id"), ("historical_odds_events", "nba_game_id"), ): ids = {r[0] for r in cur.execute(f'SELECT DISTINCT "{col}" FROM "{table}" WHERE "{col}" IS NOT NULL')} result[table] = { "distinct_ids": len(ids), "ids_not_in_games": len(ids - game_ids), "games_missing_table_id": len(game_ids - ids), "sample_ids_not_in_games": sorted(ids - game_ids)[:20], } con.close() return result def relationship_pikkit(backtest: dict[str, Any]) -> list[dict[str, Any]]: game_ids = set(backtest["games"]) con = sqlite_connect(DATA / "backtest.db") cur = con.cursor() pbp_max_action = { game_id: max_action for game_id, max_action in cur.execute("SELECT game_id, MAX(action_number) FROM pbp_actions GROUP BY game_id") } con.close() checks = [] for name in ("pikkit_full_pbp_enriched.json", "pikkit_matched_v6.json", "pikkit_enriched.json", "pikkit_historical_ev.json"): path = DATA / name if not path.exists(): continue rows = json.load(path.open()) row_game_ids = {str(r.get("game_id")) for r in rows if r.get("game_id")} check = { "source": name, "rows": len(rows), "distinct_game_ids": len(row_game_ids), "game_ids_missing_from_backtest": len(row_game_ids - game_ids), "sample_missing_game_ids": sorted(row_game_ids - game_ids)[:20], } if name == "pikkit_full_pbp_enriched.json": bad_action = 0 for row in rows: action_at = as_float(row.get("action_at")) max_action = pbp_max_action.get(str(row.get("game_id"))) if action_at is not None and max_action is not None and not (0 <= action_at <= max_action): bad_action += 1 check["action_number_out_of_bounds_vs_backtest_pbp"] = bad_action checks.append(check) return checks def relationship_live(backtest: dict[str, Any]) -> list[dict[str, Any]]: game_ids = set(backtest["games"]) checks = [] for db_name, table, col in ( ("live_spreads.db", "spread_snapshots", "game_id"), ("arb_log.db", "pbp_snapshots", "game_id"), ("arb_log.db", "run_events", "game_id"), ): path = DATA / db_name if not path.exists(): continue con = sqlite_connect(path) cur = con.cursor() ids = {r[0] for r in cur.execute(f'SELECT DISTINCT "{col}" FROM "{table}" WHERE "{col}" IS NOT NULL')} count = safe_count(cur, table) con.close() checks.append( { "source": f"{db_name}/{table}", "rows": count, "distinct_game_ids": len(ids), "game_ids_missing_from_backtest": len(ids - game_ids), "sample_missing_game_ids": sorted(ids - game_ids)[:20], } ) return checks def kalshi_teams_from_ticker(*values: Any) -> tuple[str, str] | None: for value in values: if not value: continue text = str(value).upper() match = re.search(r"(\d{2}[A-Z]{3}\d{2})([A-Z]{6})", text) if match: teams = match.group(2) return norm_team(teams[:3]), norm_team(teams[3:]) parts = re.findall(r"[A-Z]{6,}", text) for part in parts: suffix = part[-6:] away = norm_team(suffix[:3]) home = norm_team(suffix[3:]) if away in set(TEAM_ALIASES.values()) and home in set(TEAM_ALIASES.values()): return away, home return None def relationship_kalshi(backtest: dict[str, Any]) -> dict[str, Any]: by_date_team = backtest["by_date_team"] db = KALSHI / "kalshi_nba.db" if not db.exists(): return {"exists": False} con = sqlite_connect(db) cur = con.cursor() rows = cur.execute( """ SELECT game_id, date, away, home, event_ticker_ml, event_ticker_spread, event_ticker_total FROM games """ ).fetchall() con.close() matched = 0 parse_fail = 0 missing = [] date_missing = 0 team_missing = 0 canonical_dates = {row["date"] for row in backtest["games"].values()} for game_id, date, away, home, ml, spread, total in rows: parsed = kalshi_teams_from_ticker(ml, spread, total) if parsed: away_tri, home_tri = parsed else: away_tri, home_tri = norm_team(away), norm_team(home) if not away_tri or not home_tri: parse_fail += 1 continue key = (date, home_tri, away_tri) if key in by_date_team: matched += 1 else: missing.append({"kalshi_game_id": game_id, "date": date, "away": away_tri, "home": home_tri}) if date not in canonical_dates: date_missing += 1 else: team_missing += 1 return { "rows": len(rows), "matched_to_backtest": matched, "missing_from_backtest": len(rows) - matched, "parse_failures": parse_fail, "missing_because_date_absent": date_missing, "missing_because_game_absent_on_date": team_missing, "sample_missing": missing[:30], } def relationship_kaggle(backtest: dict[str, Any]) -> dict[str, Any]: path = DATA / "kaggle_nba_betting" / "nba_2008-2025.csv" if not path.exists(): return {"exists": False} by_date_team = backtest["by_date_team"] by_score = backtest["by_score"] canonical_dates = {row["date"] for row in backtest["games"].values()} canonical_date_min = min(canonical_dates) canonical_date_max = max(canonical_dates) rows = 0 inside_date_range = 0 on_canonical_dates = 0 exact_score = 0 team_date_only = 0 bad_team = 0 with path.open(newline="", encoding="utf-8", errors="replace") as f: reader = csv.DictReader(f) for row in reader: rows += 1 date = parse_date(row.get("date")) home = norm_team(row.get("home")) away = norm_team(row.get("away")) if not home or not away: bad_team += 1 continue if date and canonical_date_min <= date <= canonical_date_max: inside_date_range += 1 if date in canonical_dates: on_canonical_dates += 1 home_score = as_float(row.get("score_home")) away_score = as_float(row.get("score_away")) if date and home_score is not None and away_score is not None: if (date, home, away, int(home_score), int(away_score)) in by_score: exact_score += 1 elif (date, home, away) in by_date_team: team_date_only += 1 return { "rows": rows, "canonical_date_min": canonical_date_min, "canonical_date_max": canonical_date_max, "rows_inside_canonical_date_range": inside_date_range, "rows_on_canonical_dates": on_canonical_dates, "exact_score_matches": exact_score, "team_date_only_matches": team_date_only, "team_parse_failures": bad_team, } def relationship_kyleskom(backtest: dict[str, Any]) -> dict[str, Any]: db = DATA / "kyleskom-nba" / "Data" / "OddsData.sqlite" if not db.exists(): return {"exists": False} by_date_team = backtest["by_date_team"] by_score_like: dict[tuple[Any, ...], str] = {} for game_id, row in backtest["games"].items(): total = (row["home_score"] or 0) + (row["away_score"] or 0) margin = (row["home_score"] or 0) - (row["away_score"] or 0) by_score_like[(row["date"], row["home"], row["away"], int(total), int(margin))] = game_id con = sqlite_connect(db) cur = con.cursor() tables = [r[0] for r in cur.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY 1")] details = [] total_rows = 0 total_exact = 0 total_team_date = 0 source_tables_scanned = 0 for table in tables: cols = [r[1] for r in cur.execute(f'PRAGMA table_info("{table}")')] if not {"Date", "Home", "Away", "Points", "Win_Margin"}.issubset(set(cols)): continue source_tables_scanned += 1 rows = cur.execute(f'SELECT Date, Home, Away, Points, Win_Margin FROM "{table}"').fetchall() exact = 0 team_date = 0 for date_raw, home_raw, away_raw, points_raw, margin_raw in rows: date = parse_date(date_raw) home = norm_team(home_raw) away = norm_team(away_raw) points = as_float(points_raw) margin = as_float(margin_raw) if not date or not home or not away: continue if points is not None and margin is not None and (date, home, away, int(points), int(margin)) in by_score_like: exact += 1 elif (date, home, away) in by_date_team: team_date += 1 total_rows += len(rows) total_exact += exact total_team_date += team_date if exact or team_date or table in {"2024-25", "odds_2025-26", "2023-24"}: details.append({"table": table, "rows": len(rows), "exact_score_matches": exact, "team_date_only_matches": team_date}) con.close() return { "source_tables_scanned": source_tables_scanned, "tables_with_matches_or_priority": len(details), "rows_checked": total_rows, "exact_score_matches": total_exact, "team_date_only_matches": total_team_date, "table_details": details[:80], } def relationship_missing_games(backtest: dict[str, Any]) -> dict[str, Any]: path = DATA / "missing_games.json" if not path.exists(): return {"exists": False} rows = json.load(path.open()) game_ids = set(backtest["games"]) by_season = Counter() blank_date_team = 0 already_present = 0 for row in rows: by_season[row.get("season") or "unknown"] += 1 if not row.get("game_date") or not row.get("home_team") or not row.get("away_team"): blank_date_team += 1 if row.get("game_id") in game_ids: already_present += 1 return { "rows": len(rows), "by_season": dict(by_season), "blank_date_or_team": blank_date_team, "already_present_in_backtest": already_present, "remaining_candidate_rows": len(rows) - already_present, } def correlation_kaggle() -> list[dict[str, Any]]: path = DATA / "kaggle_nba_betting" / "nba_2008-2025.csv" rows = [] if not path.exists(): return rows with path.open(newline="", encoding="utf-8", errors="replace") as f: reader = csv.DictReader(f) for row in reader: home_score = as_float(row.get("score_home")) away_score = as_float(row.get("score_away")) if home_score is not None and away_score is not None: row["actual_points"] = home_score + away_score row["home_margin"] = home_score - away_score rows.append(row) return [ item for item in ( pearson(rows, "total", "actual_points"), pearson(rows, "spread", "home_margin"), pearson(rows, "h2_total", "actual_points"), pearson(rows, "h2_spread", "home_margin"), ) if item ] def correlation_json_file(path: Path, pairs: list[tuple[str, str]]) -> list[dict[str, Any]]: if not path.exists(): return [] obj = json.load(path.open()) if not isinstance(obj, list): return [] out = [] for x, y in pairs: item = pearson(obj, x, y) if item: item["source"] = path.name out.append(item) return out def correlation_signals_db() -> list[dict[str, Any]]: db = DATA / "signals.db" if not db.exists(): return [] con = sqlite_connect(db) cur = con.cursor() cols = [r[1] for r in cur.execute("PRAGMA table_info(backtest_games)")] rows = [dict(zip(cols, r)) for r in cur.execute("SELECT * FROM backtest_games")] con.close() pairs = [ ("spread", "dog_covered"), ("final_margin", "dog_covered"), ("half_margin", "dog_covered"), ("q3_margin", "dog_covered"), ("q3_lead_pct", "dog_covered"), ("q3_max_lead", "dog_covered"), ("q3_erosion", "dog_covered"), ("q3_stability", "dog_covered"), ("to_diff", "dog_covered"), ("dog_tov", "dog_covered"), ("fav_tov", "dog_covered"), ] out = [] for x, y in pairs: item = pearson(rows, x, y) if item: item["source"] = "signals.db/backtest_games" out.append(item) return out def correlation_kyleskom() -> list[dict[str, Any]]: db = DATA / "kyleskom-nba" / "Data" / "OddsData.sqlite" if not db.exists(): return [] con = sqlite_connect(db) cur = con.cursor() tables = [r[0] for r in cur.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY 1")] rows = [] for table in tables: cols = [r[1] for r in cur.execute(f'PRAGMA table_info("{table}")')] if not {"OU", "Spread", "Points", "Win_Margin"}.issubset(set(cols)): continue # Prefer normalized tables without the old pandas index columns to avoid double counting. if "index" in cols and f"{table}_new" in tables: continue for ou, spread, points, margin in cur.execute(f'SELECT OU, Spread, Points, Win_Margin FROM "{table}"'): rows.append({"OU": ou, "Spread": spread, "Points": points, "Win_Margin": margin}) con.close() out = [] for x, y in (("OU", "Points"), ("Spread", "Win_Margin")): item = pearson(rows, x, y) if item: item["source"] = "kyleskom/OddsData.sqlite" out.append(item) return out def run_correlations() -> list[dict[str, Any]]: out = [] for item in correlation_kaggle(): item["source"] = "kaggle_nba_betting/nba_2008-2025.csv" out.append(item) out.extend(correlation_kyleskom()) out.extend(correlation_signals_db()) out.extend( correlation_json_file( DATA / "pikkit_matched_v6.json", [ ("edge", "profit"), ("edge", "won"), ("margin", "won"), ("period_at_bet", "won"), ("spread", "won"), ("odds", "profit"), ("amount", "profit"), ], ) ) out.extend( correlation_json_file( DATA / "pikkit_full_pbp_enriched.json", [ ("margin", "won"), ("period", "won"), ("spread", "won"), ("odds", "profit"), ("amount", "profit"), ("to_diff", "won"), ("reb_diff", "won"), ("poss_diff", "won"), ("bt_fg_pct", "won"), ("bt_fg3_pct", "won"), ], ) ) out.extend( correlation_json_file( DATA / "live_line_correlation_data.json", [ ("spread", "dog_covered"), ("total", "final_margin"), ("half_margin", "dog_covered"), ("half_to_diff", "dog_covered"), ("q3_margin", "dog_covered"), ("q3_to_diff", "dog_covered"), ("final_to_diff", "dog_covered"), ], ) ) out.extend( correlation_json_file( DATA / "universal_game_state.json", [ ("period", "dog_covered"), ("spread", "dog_covered"), ("margin", "dog_covered"), ("chase_gap", "dog_covered"), ("to_diff", "dog_covered"), ("reb_diff", "dog_covered"), ("fg_pct_diff", "dog_covered"), ("fg3_pct_diff", "dog_covered"), ], ) ) out.extend( correlation_json_file( DATA / "pregame_features.json", [ ("spread", "home_covered"), ("total", "margin"), ("to_diff", "home_covered"), ("reb_diff", "home_covered"), ("fg_pct_diff", "home_covered"), ("3pt_pct_diff", "home_covered"), ("home_rest", "home_covered"), ("away_rest", "home_covered"), ], ) ) out.sort(key=lambda item: abs(item["r"]), reverse=True) return out def summarize_known_files() -> dict[str, Any]: csvs = [ DATA / "kaggle_nba_betting" / "nba_2008-2025.csv", DATA / "pikkit_bets.csv", DATA / "nba_2013_2014.csv", DATA / "nba_2014_2015.csv", DATA / "nba_2015_2016.csv", DATA / "nba_2016_2017.csv", DATA / "nba_2017_2018.csv", BACKUP_MLB / "pikkit_mlb_raw.csv", BACKUP_MLB / "pikkit_mlb_enriched.csv", ] jsons = [ DATA / "pikkit_matched_v6.json", DATA / "pikkit_full_pbp_enriched.json", DATA / "pikkit_nba_spreads_full.json", DATA / "dynamic_game_states.json", DATA / "universal_game_state.json", DATA / "live_line_correlation_data.json", DATA / "pregame_features.json", KALSHI / "nba_events.json", KALSHI / "nba_spread_markets_raw.json", SPORTS / "screenshot_snapshots_raw.json", BACKUP_KALSHI_TRACES / "20251204_GSW_vs_PHI_spread_mkts.json", BACKUP_KALSHI_TRACES / "20260201_CHI_vs_MIA_spread_mkts.json", ] dbs = [ DATA / "backtest.db", DATA / "live_spreads.db", DATA / "arb_log.db", DATA / "signals.db", DATA / "odds_snapshots.db", DATA / "nuclear_players.db", SPORTS / "game_snapshots.db", KALSHI / "kalshi_nba.db", DATA / "kyleskom-nba" / "Data" / "OddsData.sqlite", DATA / "kyleskom-nba" / "Data" / "dataset.sqlite", DATA / "kyleskom-nba" / "Data" / "TeamData.sqlite", ] return { "csv": [summarize_csv(p) for p in csvs if p.exists()], "json": [summarize_json(p) for p in jsons if p.exists()], "sqlite": [inspect_sqlite(p) for p in dbs if p.exists()], } def build_report() -> dict[str, Any]: backtest = load_backtest_games() return { "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(), "scope": "sports-only: NBA, MLB package, odds, Pikkit, Kalshi, kyleskom, Kaggle, live sports captures, and sports model artifacts", "inventory": summarize_files(), "known_file_summaries": summarize_known_files(), "relationships": { "backtest_integrity": relationship_backtest_integrity(backtest), "pikkit_to_backtest": relationship_pikkit(backtest), "live_to_backtest": relationship_live(backtest), "kalshi_to_backtest": relationship_kalshi(backtest), "kaggle_to_backtest": relationship_kaggle(backtest), "kyleskom_to_backtest": relationship_kyleskom(backtest), "missing_games_queue": relationship_missing_games(backtest), }, "correlations": run_correlations(), } def write_markdown(report: dict[str, Any], path: Path) -> None: inv = report["inventory"] rel = report["relationships"] lines = [ "# Sports Data Map", "", f"Generated: {report['generated_at']}", "", f"Scope: {report['scope']}", "", "## Inventory", "", f"- Files mapped: {inv['total_files']:,}", f"- Bytes mapped: {inv['total_bytes']:,}", "", "### Largest Files", "", "| File | Size MB | Category | Backup |", "|---|---:|---|---|", ] for item in inv["largest_files"][:20]: lines.append( f"| `{item['path']}` | {item['bytes'] / 1024 / 1024:.2f} | {item['category']} | {item['backup']} |" ) lines.extend(["", "## Relationships", ""]) for name, value in rel.items(): lines.append(f"### {name}") lines.append("") lines.append("```json") lines.append(json.dumps(value, indent=2, sort_keys=True)[:12000]) lines.append("```") lines.append("") lines.extend(["## Top Correlations", "", "| Source | X | Y | N | r |", "|---|---|---|---:|---:|"]) for item in report["correlations"][:40]: lines.append( f"| `{item.get('source', '')}` | {item['x']} | {item['y']} | {item['n']:,} | {item['r']:.3f} |" ) path.write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--json-out", default="/var/www/sports-status/sports-data-map.json") parser.add_argument("--md-out", default="/var/www/sports-status/sports-data-map.md") args = parser.parse_args() report = build_report() json_path = Path(args.json_out) md_path = Path(args.md_out) json_path.parent.mkdir(parents=True, exist_ok=True) json_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") write_markdown(report, md_path) print(json.dumps({ "json": str(json_path), "markdown": str(md_path), "files_mapped": report["inventory"]["total_files"], "correlations": len(report["correlations"]), }, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())