diff --git a/DATA_PIPELINE_LINEAGE.md b/DATA_PIPELINE_LINEAGE.md index 65d5f29..e5ad4a1 100644 --- a/DATA_PIPELINE_LINEAGE.md +++ b/DATA_PIPELINE_LINEAGE.md @@ -29,6 +29,19 @@ Each copied observation receives an immutable provenance reference to the protec source database and source row. The target database can be discarded and rebuilt without altering the source. +For point-in-time safety, a copied row's `observed_at` is the latest valid UTC +timestamp among its source `received_at`, `result_available_at`, and `as_of` +fields—not the first non-empty field. The content digest commits to all three raw +timestamps, the derived knowledge time, and the derivation rule. A completed +result therefore cannot enter a projection before the source says that result was +available. + +Rows with malformed or offset-free timestamps, source timestamps later than the +import receipt, missing final scores, or a final result-availability time before +the logical game start are rejected. Import receipts report per-reason quarantine +counts and a canonical digest of the rejected source-row set without modifying the +protected database. + ## Explicit exclusions The implementation does not copy or activate: @@ -51,9 +64,26 @@ range fall back to bounded daily requests. Provider availability is not guaranteed, and provider terms govern use. A fetch failure stays visible as an ingest error and never becomes synthetic current data. +## Point-in-time DumbMoney export + +Normalized game observations are append-only, content-addressed revisions. +The existing `games` table remains a compatibility view of the latest revision; +it is not the historical source of truth. Point-in-time projections select the +latest unambiguous revision received on or before the requested cutoff. A later +score, schedule change, or correction is retained but cannot leak into an older +forecast. + +`SportsForecastEnvelopeV1` commits to the selected revision set and exports only +candidate research. Its three fixed authority fields prevent the envelope from +granting order or capital permissions. Source terms still govern use; the +envelope records references and limitations but does not grant redistribution +or model-training rights. + ## Verification Automated tests create temporary source and destination databases, verify that the source bytes remain unchanged, exercise normalization and aliases, and inspect the -HTTP/API projection contracts. Operationally, recheck `git status` in Dummy after -an import and confirm it is clean. +HTTP/API projection contracts. They also prove deterministic envelope identity, +future-leakage rejection, contradiction and staleness handling, and append-only +revision retention. Operationally, recheck `git status` in Dummy after an import +and confirm it is clean. diff --git a/README.md b/README.md index fc51f52..2a2f482 100644 --- a/README.md +++ b/README.md @@ -183,10 +183,20 @@ selected teams, point-in-time history, model version, and seed. | `GET` | `/api/v1/leagues/{league}/teams` | Current team picker | | `GET` | `/api/v1/week` | Rolling current-week projection board | | `POST` | `/api/v1/simulations/hypothetical` | Reproducible matchup simulation | +| `POST` | `/api/v1/exports/dumbmoney/sports-forecast` | Candidate-only point-in-time forecast envelope | | `POST` | `/api/v1/data/refresh` | Bounded public schedule refresh | OpenAPI is available at `/docs`. +The DumbMoney export is schema +`dumbmoney.sports-forecast-envelope.v1`. Its canonical SHA-256 identity commits +to the exact append-only observation revisions, cutoff, model/champion version, +seed, draw count, distributions, evidence, provenance, data-rights warnings, +expiry, and authority invariants. It always emits `candidate_only=true`, +`order_authority=false`, and `capital_authority=false`. It refuses future, +stale, ambiguous, in-progress, or completed target observations and contains no +wagering, brokerage, order, or capital interface. + ## Determinism and governance ```mermaid diff --git a/src/universal_sports_engine/__init__.py b/src/universal_sports_engine/__init__.py index 4efe776..56b94f1 100644 --- a/src/universal_sports_engine/__init__.py +++ b/src/universal_sports_engine/__init__.py @@ -8,6 +8,10 @@ simulate_many_analytical, simulate_profile_analytical, ) +from universal_sports_engine.dumbmoney import ( + DumbMoneyForecastExporter, + SportsForecastEnvelopeV1, +) from universal_sports_engine.simulation import ( counterfactual_game, simulate_game, @@ -19,6 +23,8 @@ __all__ = [ "AdaptiveAnalyticalResult", "AnalyticalResult", + "DumbMoneyForecastExporter", + "SportsForecastEnvelopeV1", "counterfactual_game", "simulate_analytical", "simulate_ensemble_analytical", diff --git a/src/universal_sports_engine/data/models.py b/src/universal_sports_engine/data/models.py index dffca2d..bebe437 100644 --- a/src/universal_sports_engine/data/models.py +++ b/src/universal_sports_engine/data/models.py @@ -32,6 +32,14 @@ class GameRecord: payload_hash: str +@dataclass(frozen=True, slots=True) +class GameObservationRevision: + """One immutable, content-addressed revision of a game observation.""" + + revision_id: str + game: GameRecord + + @dataclass(frozen=True, slots=True) class TeamRecord: league: str diff --git a/src/universal_sports_engine/data/pipeline.py b/src/universal_sports_engine/data/pipeline.py index 7c54e2d..d70a99d 100644 --- a/src/universal_sports_engine/data/pipeline.py +++ b/src/universal_sports_engine/data/pipeline.py @@ -16,12 +16,116 @@ from universal_sports_engine.league_packs.registry import league_ids, normalize_league_id DUMMY_LEAGUE_ALIASES = {"ncaamb": "ncaambb"} +SOURCE_TIME_FIELDS = ("received_at", "result_available_at", "as_of") +DUMMY_SOURCE_FIELDS = ( + "game_id", + "league", + "season", + "start_time", + "status", + "home", + "away", + "home_score", + "away_score", + "venue", + "neutral", + "source", + "provenance_url", + *SOURCE_TIME_FIELDS, + "extra", +) def _utc_now_iso() -> str: return datetime.now(UTC).isoformat() +def _canonical_hash(value: object) -> str: + return hashlib.sha256( + json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + ).hexdigest() + + +def _source_row_digest(row: sqlite3.Row) -> str: + return _canonical_hash( + { + field: str(row[field]) if row[field] is not None else None + for field in DUMMY_SOURCE_FIELDS + } + ) + + +class DummyHistoryRowRejected(ValueError): + """A source row cannot support a point-in-time-safe normalized observation.""" + + def __init__(self, reason: str) -> None: + super().__init__(reason) + self.reason = reason + + +def _parse_source_timestamp(value: object, field: str) -> datetime | None: + if value is None or not str(value).strip(): + return None + raw = str(value).strip() + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError as exc: + raise DummyHistoryRowRejected(f"malformed_timestamp:{field}") from exc + if parsed.tzinfo is None: + raise DummyHistoryRowRejected(f"timestamp_without_offset:{field}") + return parsed.astimezone(UTC) + + +def _dummy_knowable_at( + row: sqlite3.Row, + *, + imported_at: datetime, + status: str, +) -> tuple[datetime, dict[str, str | None]]: + """Derive the earliest defensible knowledge time without result leakage.""" + + start_time = _parse_source_timestamp(row["start_time"], "start_time") + if start_time is None: + raise DummyHistoryRowRejected("missing_timestamp:start_time") + raw_timestamps = { + field: str(row[field]).strip() if row[field] is not None else None + for field in SOURCE_TIME_FIELDS + } + parsed_timestamps = { + field: _parse_source_timestamp(row[field], field) + for field in SOURCE_TIME_FIELDS + } + future_fields = sorted( + field + for field, timestamp in parsed_timestamps.items() + if timestamp is not None and timestamp > imported_at + ) + if future_fields: + raise DummyHistoryRowRejected( + f"source_timestamp_after_import:{','.join(future_fields)}" + ) + result_available_at = parsed_timestamps["result_available_at"] + if ( + status == "final" + and result_available_at is not None + and result_available_at < start_time + ): + raise DummyHistoryRowRejected("final_result_available_before_start") + valid_timestamps = tuple( + timestamp for timestamp in parsed_timestamps.values() if timestamp is not None + ) + knowable_at = max(valid_timestamps, default=imported_at) + if status == "final" and knowable_at < start_time: + raise DummyHistoryRowRejected("final_observation_before_start") + return knowable_at, raw_timestamps + + def _chunks(start: date, end: date, days: int) -> Iterable[tuple[date, date]]: cursor = start while cursor <= end: @@ -138,7 +242,13 @@ def import_dummy_history( if not source_database.is_file(): raise FileNotFoundError(f"Dummy sports history database is missing: {source_database}") observed_at = _utc_now_iso() - counts: dict[str, int] = {} + imported_at = _parse_source_timestamp(observed_at, "imported_at") + if imported_at is None: + raise RuntimeError("Import receipt timestamp is unavailable") + fetched_counts: dict[str, int] = {} + stored_counts: dict[str, int] = {} + quarantine_counts: dict[str, dict[str, int]] = {} + quarantine_row_digests: dict[str, list[str]] = {} with closing( sqlite3.connect( f"file:{source_database.as_posix()}?mode=ro", @@ -162,17 +272,16 @@ def import_dummy_history( break normalized_rows: list[GameRecord] = [] for row in rows: - if not row["home"] or not row["away"] or not row["start_time"]: - continue - home_team_id = str(row["home"]).upper() - away_team_id = str(row["away"]).upper() - if home_team_id == away_team_id: + raw_league = str(row["league"]) + try: + league = DUMMY_LEAGUE_ALIASES.get(raw_league) + if league is None: + league = normalize_league_id(raw_league) + except ValueError: continue - league = DUMMY_LEAGUE_ALIASES.get( - str(row["league"]), normalize_league_id(str(row["league"])) - ) if league not in league_ids(): continue + fetched_counts[league] = fetched_counts.get(league, 0) + 1 raw_status = str(row["status"]).lower() status = { "post": "final", @@ -182,73 +291,145 @@ def import_dummy_history( "scheduled": "scheduled", "pre": "scheduled", }.get(raw_status, "unresolved") + try: + if not row["home"] or not row["away"] or not row["start_time"]: + raise DummyHistoryRowRejected("missing_required_identity") + home_team_id = str(row["home"]).strip().upper() + away_team_id = str(row["away"]).strip().upper() + if not home_team_id or not away_team_id: + raise DummyHistoryRowRejected("missing_required_identity") + if home_team_id == away_team_id: + raise DummyHistoryRowRejected("same_team_matchup") + external_id = str(row["game_id"] or "").strip() + if not external_id: + raise DummyHistoryRowRejected("missing_game_id") + try: + season = ( + int(row["season"]) if row["season"] is not None else None + ) + home_score = ( + int(row["home_score"]) + if row["home_score"] is not None + else None + ) + away_score = ( + int(row["away_score"]) + if row["away_score"] is not None + else None + ) + neutral_value = int(row["neutral"] or 0) + except (TypeError, ValueError, OverflowError) as exc: + raise DummyHistoryRowRejected( + "malformed_numeric_field" + ) from exc + if neutral_value not in {0, 1}: + raise DummyHistoryRowRejected("invalid_neutral_site") + neutral_site = bool(neutral_value) + if (home_score is not None and home_score < 0) or ( + away_score is not None and away_score < 0 + ): + raise DummyHistoryRowRejected("negative_score") + if status == "final" and ( + home_score is None or away_score is None + ): + raise DummyHistoryRowRejected("final_score_missing") + knowable_at, source_timestamps = _dummy_knowable_at( + row, + imported_at=imported_at, + status=status, + ) + except DummyHistoryRowRejected as exc: + reasons = quarantine_counts.setdefault(league, {}) + reasons[exc.reason] = reasons.get(exc.reason, 0) + 1 + quarantine_row_digests.setdefault(league, []).append( + _source_row_digest(row) + ) + continue source = str(row["source"] or "dummy_history") - external_id = str(row["game_id"]) identity = f"{source}:{league}:{external_id}" extra = str(row["extra"] or "") content = json.dumps( { "game_id": external_id, "league": league, + "source": source, "start_time": row["start_time"], "status": status, "home": row["home"], "away": row["away"], "home_score": row["home_score"], "away_score": row["away_score"], + "neutral": row["neutral"], + "venue": row["venue"], + "provenance_url": row["provenance_url"], + "source_timestamps": source_timestamps, + "knowable_at": knowable_at.isoformat(), + "knowable_at_rule": "latest_valid_source_timestamp_v1", "extra": extra, }, + allow_nan=False, + ensure_ascii=False, sort_keys=True, separators=(",", ":"), - ).encode() + ).encode("utf-8") normalized_rows.append( GameRecord( game_id=( f"espn:{league}:{external_id}" if source == "espn" - else hashlib.sha256(identity.encode()).hexdigest() + else hashlib.sha256(identity.encode("utf-8")).hexdigest() ), league=league, - season=int(row["season"]) if row["season"] is not None else None, + season=season, start_time=str(row["start_time"]), status=status, home_team_id=home_team_id, away_team_id=away_team_id, home_name=str(row["home"]), away_name=str(row["away"]), - home_score=( - int(row["home_score"]) if row["home_score"] is not None else None - ), - away_score=( - int(row["away_score"]) if row["away_score"] is not None else None - ), - neutral_site=bool(row["neutral"]), + home_score=home_score, + away_score=away_score, + neutral_site=neutral_site, venue=str(row["venue"]) if row["venue"] else None, source=f"dummy_read_only:{source}", source_url=str(row["provenance_url"] or "local-read-only-import"), - observed_at=str( - row["received_at"] - or row["result_available_at"] - or row["as_of"] - or observed_at - ), + observed_at=knowable_at.isoformat(), payload_hash=hashlib.sha256(content).hexdigest(), ) ) - counts[league] = counts.get(league, 0) + 1 + stored_counts[league] = stored_counts.get(league, 0) + 1 self.store.upsert_games(normalized_rows) results: list[IngestResult] = [] for league in league_ids(): - count = counts.get(league, 0) + fetched = fetched_counts.get(league, 0) + stored = stored_counts.get(league, 0) + quarantine = quarantine_counts.get(league, {}) + quarantined_row_ids = quarantine_row_digests.get(league, []) + quarantined = sum(quarantine.values()) result = IngestResult( source="dummy_sports_history_read_only_import", league=league, window_start="historical", window_end="historical", - fetched=count, - stored=count, - status="ok", + fetched=fetched, + stored=stored, + status="partial" if quarantined else "ok", observed_at=observed_at, + error=( + json.dumps( + { + "quarantined": quarantined, + "reasons": dict(sorted(quarantine.items())), + "row_set_digest": _canonical_hash( + sorted(quarantined_row_ids) + ), + }, + separators=(",", ":"), + sort_keys=True, + ) + if quarantined + else None + ), ) self.store.record_ingest(result) results.append(result) diff --git a/src/universal_sports_engine/data/store.py b/src/universal_sports_engine/data/store.py index a73313d..9bae78f 100644 --- a/src/universal_sports_engine/data/store.py +++ b/src/universal_sports_engine/data/store.py @@ -2,16 +2,72 @@ from __future__ import annotations +import hashlib +import json import sqlite3 from collections.abc import Iterable from contextlib import closing +from dataclasses import asdict +from datetime import UTC, datetime from pathlib import Path from typing import Any, cast -from universal_sports_engine.data.models import GameRecord, IngestResult, TeamRecord +from universal_sports_engine.data.models import ( + GameObservationRevision, + GameRecord, + IngestResult, + TeamRecord, +) from universal_sports_engine.league_packs.registry import normalize_league_id -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 + + +class FutureObservationError(ValueError): + """The requested game exists, but every observation arrived after the cutoff.""" + + +class ContradictoryObservationError(ValueError): + """Multiple incompatible observations occupy the same point-in-time revision.""" + + +def _parse_utc(value: str, field: str) -> datetime: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"{field} must be an ISO-8601 timestamp: {value!r}") from exc + if parsed.tzinfo is None: + raise ValueError(f"{field} must include a UTC offset: {value!r}") + return parsed.astimezone(UTC) + + +def _utc_datetime(value: datetime, field: str) -> datetime: + if value.tzinfo is None: + raise ValueError(f"{field} must be timezone-aware") + return value.astimezone(UTC) + + +def _canonical_json_bytes(value: object) -> bytes: + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +def _revision_id(game: GameRecord) -> str: + return hashlib.sha256(_canonical_json_bytes(asdict(game))).hexdigest() + + +def _truth_digest(game: GameRecord) -> str: + """Identify material game truth independently from source provenance.""" + + value = asdict(game) + for field in ("source", "source_url", "observed_at", "payload_hash"): + value.pop(field) + return hashlib.sha256(_canonical_json_bytes(value)).hexdigest() class SportsDataStore: @@ -75,6 +131,43 @@ def initialize(self) -> None: ON games(league, start_time); CREATE INDEX IF NOT EXISTS ix_games_league_status ON games(league, status, start_time); + CREATE TABLE IF NOT EXISTS game_observations( + revision_id TEXT PRIMARY KEY, + game_id TEXT NOT NULL, + league TEXT NOT NULL, + season INTEGER, + start_time TEXT NOT NULL, + status TEXT NOT NULL, + home_team_id TEXT NOT NULL, + away_team_id TEXT NOT NULL, + home_name TEXT NOT NULL, + away_name TEXT NOT NULL, + home_score INTEGER, + away_score INTEGER, + neutral_site INTEGER NOT NULL DEFAULT 0, + venue TEXT, + source TEXT NOT NULL, + source_url TEXT NOT NULL, + observed_at TEXT NOT NULL, + payload_hash TEXT NOT NULL, + CHECK(home_team_id <> away_team_id), + CHECK(home_score IS NULL OR home_score >= 0), + CHECK(away_score IS NULL OR away_score >= 0) + ); + CREATE INDEX IF NOT EXISTS ix_game_observations_game_time + ON game_observations(game_id, observed_at, revision_id); + CREATE INDEX IF NOT EXISTS ix_game_observations_league_time + ON game_observations(league, observed_at, game_id); + CREATE TRIGGER IF NOT EXISTS game_observations_no_update + BEFORE UPDATE ON game_observations + BEGIN + SELECT RAISE(ABORT, 'game observations are append-only'); + END; + CREATE TRIGGER IF NOT EXISTS game_observations_no_delete + BEFORE DELETE ON game_observations + BEGIN + SELECT RAISE(ABORT, 'game observations are append-only'); + END; CREATE TABLE IF NOT EXISTS ingest_runs( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, @@ -99,64 +192,258 @@ def initialize(self) -> None: (str(SCHEMA_VERSION),), ) connection.commit() + self._backfill_observations() + + @staticmethod + def _game_values(game: GameRecord) -> tuple[object, ...]: + return ( + game.game_id, + game.league, + game.season, + game.start_time, + game.status, + game.home_team_id, + game.away_team_id, + game.home_name, + game.away_name, + game.home_score, + game.away_score, + int(game.neutral_site), + game.venue, + game.source, + game.source_url, + game.observed_at, + game.payload_hash, + ) + + @classmethod + def _observation_values(cls, game: GameRecord) -> tuple[object, ...]: + return (_revision_id(game), *cls._game_values(game)) + + def _backfill_observations(self) -> None: + """Seed the append-only table when opening a schema-v1 database.""" + + with closing(self._connect()) as connection: + marker = connection.execute( + """ + SELECT value FROM metadata + WHERE key='game_observations_backfilled_v2' + """ + ).fetchone() + if marker is not None and str(marker["value"]) == "complete": + return + rows = connection.execute("SELECT * FROM games").fetchall() + observations = [ + self._observation_values(self._game(row)) + for row in rows + ] + if observations: + connection.executemany( + """ + INSERT OR IGNORE INTO game_observations( + revision_id, game_id, league, season, start_time, status, + home_team_id, away_team_id, home_name, away_name, + home_score, away_score, neutral_site, venue, + source, source_url, observed_at, payload_hash + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, + observations, + ) + connection.execute( + """ + INSERT INTO metadata(key, value) + VALUES('game_observations_backfilled_v2', 'complete') + ON CONFLICT(key) DO UPDATE SET value=excluded.value + """ + ) + connection.commit() def upsert_games(self, games: Iterable[GameRecord]) -> int: + """Append observations and refresh the compatibility latest-state view. + + The method name remains import-compatible with the existing pipeline. + Source truth is never updated or deleted: every distinct normalized + observation is retained in ``game_observations``. + """ + rows = tuple(games) if not rows: return 0 - values = [ - ( - game.game_id, - game.league, - game.season, - game.start_time, - game.status, - game.home_team_id, - game.away_team_id, - game.home_name, - game.away_name, - game.home_score, - game.away_score, - int(game.neutral_site), - game.venue, - game.source, - game.source_url, - game.observed_at, - game.payload_hash, - ) - for game in rows - ] + for game in rows: + _parse_utc(game.observed_at, "observed_at") + observation_values = [self._observation_values(game) for game in rows] with closing(self._connect()) as connection: connection.executemany( """ - INSERT INTO games( - game_id, league, season, start_time, status, + INSERT OR IGNORE INTO game_observations( + revision_id, game_id, league, season, start_time, status, home_team_id, away_team_id, home_name, away_name, home_score, away_score, neutral_site, venue, source, source_url, observed_at, payload_hash - ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - ON CONFLICT(game_id) DO UPDATE SET - season=excluded.season, - start_time=excluded.start_time, - status=excluded.status, - home_team_id=excluded.home_team_id, - away_team_id=excluded.away_team_id, - home_name=excluded.home_name, - away_name=excluded.away_name, - home_score=excluded.home_score, - away_score=excluded.away_score, - neutral_site=excluded.neutral_site, - venue=excluded.venue, - source=excluded.source, - source_url=excluded.source_url, - observed_at=excluded.observed_at, - payload_hash=excluded.payload_hash + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, - values, + observation_values, ) + for game_id in sorted({game.game_id for game in rows}): + observations = connection.execute( + "SELECT * FROM game_observations WHERE game_id=?", + (game_id,), + ).fetchall() + latest = max( + observations, + key=lambda row: ( + _parse_utc(str(row["observed_at"]), "observed_at"), + str(row["revision_id"]), + ), + ) + latest_game = self._game(latest) + connection.execute( + """ + INSERT INTO games( + game_id, league, season, start_time, status, + home_team_id, away_team_id, home_name, away_name, + home_score, away_score, neutral_site, venue, + source, source_url, observed_at, payload_hash + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(game_id) DO UPDATE SET + league=excluded.league, + season=excluded.season, + start_time=excluded.start_time, + status=excluded.status, + home_team_id=excluded.home_team_id, + away_team_id=excluded.away_team_id, + home_name=excluded.home_name, + away_name=excluded.away_name, + home_score=excluded.home_score, + away_score=excluded.away_score, + neutral_site=excluded.neutral_site, + venue=excluded.venue, + source=excluded.source, + source_url=excluded.source_url, + observed_at=excluded.observed_at, + payload_hash=excluded.payload_hash + """, + self._game_values(latest_game), + ) connection.commit() return len(rows) + @classmethod + def _revision(cls, row: sqlite3.Row) -> GameObservationRevision: + return GameObservationRevision( + revision_id=str(row["revision_id"]), + game=cls._game(row), + ) + + def game_revisions(self, game_id: str) -> tuple[GameObservationRevision, ...]: + """Return every immutable revision in point-in-time order.""" + + with closing(self._connect(read_only=True)) as connection: + rows = connection.execute( + "SELECT * FROM game_observations WHERE game_id=?", + (game_id,), + ).fetchall() + revisions = (self._revision(row) for row in rows) + return tuple( + sorted( + revisions, + key=lambda revision: ( + _parse_utc(revision.game.observed_at, "observed_at"), + revision.revision_id, + ), + ) + ) + + @staticmethod + def _latest_revision_as_of( + revisions: tuple[GameObservationRevision, ...], + cutoff: datetime, + ) -> GameObservationRevision: + eligible = tuple( + revision + for revision in revisions + if _parse_utc(revision.game.observed_at, "observed_at") <= cutoff + ) + if not eligible: + raise FutureObservationError( + "Every observation for the requested game was received after " + f"the as-of cutoff: {cutoff.isoformat()}" + ) + latest_time = max( + _parse_utc(revision.game.observed_at, "observed_at") + for revision in eligible + ) + latest = tuple( + revision + for revision in eligible + if _parse_utc(revision.game.observed_at, "observed_at") == latest_time + ) + if len({_truth_digest(revision.game) for revision in latest}) != 1: + raise ContradictoryObservationError( + "Conflicting game observations share the latest received-at " + f"timestamp: {latest_time.isoformat()}" + ) + return min(latest, key=lambda revision: revision.revision_id) + + def game_revision_as_of( + self, + game_id: str, + *, + as_of: datetime, + ) -> GameObservationRevision: + cutoff = _utc_datetime(as_of, "as_of") + revisions = self.game_revisions(game_id) + if not revisions: + raise KeyError(f"Unknown game observation: {game_id}") + return self._latest_revision_as_of(revisions, cutoff) + + def completed_game_revisions_as_of( + self, + league: str, + *, + as_of: datetime, + limit: int = 10_000, + ) -> tuple[GameObservationRevision, ...]: + """Select only final observations that were knowable at ``as_of``.""" + + normalized = normalize_league_id(league) + cutoff = _utc_datetime(as_of, "as_of") + with closing(self._connect(read_only=True)) as connection: + rows = connection.execute( + "SELECT * FROM game_observations WHERE league=?", + (normalized,), + ).fetchall() + by_game: dict[str, list[GameObservationRevision]] = {} + for row in rows: + revision = self._revision(row) + by_game.setdefault(revision.game.game_id, []).append(revision) + selected: list[GameObservationRevision] = [] + for revisions in by_game.values(): + eligible = tuple( + revision + for revision in revisions + if _parse_utc(revision.game.observed_at, "observed_at") <= cutoff + ) + if not eligible: + continue + latest = self._latest_revision_as_of(eligible, cutoff) + game = latest.game + game_time = _parse_utc(game.start_time, "start_time") + if ( + game.status == "final" + and game.home_score is not None + and game.away_score is not None + and game_time < cutoff + ): + selected.append(latest) + selected.sort( + key=lambda revision: ( + _parse_utc(revision.game.start_time, "start_time"), + revision.game.game_id, + ) + ) + return tuple(selected[-max(1, limit) :]) + def record_ingest(self, result: IngestResult) -> None: with closing(self._connect()) as connection: connection.execute( diff --git a/src/universal_sports_engine/dumbmoney.py b/src/universal_sports_engine/dumbmoney.py new file mode 100644 index 0000000..6f447c6 --- /dev/null +++ b/src/universal_sports_engine/dumbmoney.py @@ -0,0 +1,408 @@ +"""Candidate-only, point-in-time forecast exports for DumbMoney. + +This boundary deliberately contains no order, brokerage, wagering, portfolio, +or capital interface. It exports a deterministic research artifact that a +separate consumer must evaluate under its own authority controls. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from dataclasses import asdict, dataclass, field +from datetime import UTC, datetime, timedelta +from typing import Literal + +from universal_sports_engine.data.models import GameObservationRevision +from universal_sports_engine.data.store import FutureObservationError, SportsDataStore +from universal_sports_engine.projection import ( + PROJECTION_MODEL_VERSION, + MatchupProjection, + ProjectionEngine, + ScoreInterval, +) + +SCHEMA_ID: Literal["dumbmoney.sports-forecast-envelope.v1"] = ( + "dumbmoney.sports-forecast-envelope.v1" +) +DEFAULT_CHAMPION_ID = "waterboy.projection-incumbent" +DEFAULT_FIDELITY = "F0" + + +class StaleObservationError(ValueError): + """The target observation is too old to export as a current candidate.""" + + +def canonical_json_bytes(value: object) -> bytes: + """Encode canonical JSON with the shared DumbMoney convention.""" + + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + + +def _digest(value: object) -> str: + return hashlib.sha256(canonical_json_bytes(value)).hexdigest() + + +def _utc(value: datetime, field_name: str) -> datetime: + if value.tzinfo is None: + raise ValueError(f"{field_name} must be timezone-aware") + return value.astimezone(UTC) + + +def _parse_utc(value: str, field_name: str) -> datetime: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError(f"{field_name} must be an ISO-8601 timestamp: {value!r}") from exc + return _utc(parsed, field_name) + + +def _iso(value: datetime) -> str: + return _utc(value, "timestamp").isoformat() + + +def observation_set_digest(revision_ids: tuple[str, ...]) -> str: + """Commit to the exact append-only observations used by a forecast.""" + + if not revision_ids: + raise ValueError("A forecast observation set cannot be empty") + return _digest( + { + "schema": "waterboy.observation-set.v1", + "revision_ids": sorted(revision_ids), + } + ) + + +@dataclass(frozen=True, slots=True) +class ForecastTargetV1: + game_id: str + league: str + starts_at: str + home_team_id: str + away_team_id: str + home_team_name: str + away_team_name: str + neutral_site: bool + + +@dataclass(frozen=True, slots=True) +class ForecastModelV1: + model_id: str + model_version: str + champion_id: str + champion_version: str + seed: int + draws: int + fidelity: str + + +@dataclass(frozen=True, slots=True) +class ScoreDistributionV1: + mean: float + lower: int + upper: int + interval_mass: float + + +@dataclass(frozen=True, slots=True) +class OutcomeDistributionV1: + home_win: float + away_win: float + tie: float + + +@dataclass(frozen=True, slots=True) +class ForecastDistributionsV1: + home_score: ScoreDistributionV1 + away_score: ScoreDistributionV1 + total_score: ScoreDistributionV1 + expected_margin: float + outcome: OutcomeDistributionV1 + + +@dataclass(frozen=True, slots=True) +class SourceProvenanceV1: + source: str + observation_count: int + latest_observed_at: str + source_urls: tuple[str, ...] + payload_set_digest: str + + +@dataclass(frozen=True, slots=True) +class ForecastProvenanceV1: + target_revision_id: str + observation_count: int + sources: tuple[SourceProvenanceV1, ...] + + +@dataclass(frozen=True, slots=True) +class DataRightsV1: + status: str + source_terms_apply: bool + redistribution_status: str + model_training_status: str + references: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class SportsForecastEnvelopeV1: + """A deterministic research candidate with structurally absent authority.""" + + as_of: str + expires_at: str + observation_set_digest: str + target: ForecastTargetV1 + model: ForecastModelV1 + distributions: ForecastDistributionsV1 + confidence_status: str + calibration_status: str + evidence_status: str + freshness_status: str + provenance: ForecastProvenanceV1 + data_rights: DataRightsV1 + limitations: tuple[str, ...] + schema: Literal["dumbmoney.sports-forecast-envelope.v1"] = field( + default=SCHEMA_ID, + init=False, + ) + candidate_only: Literal[True] = field(default=True, init=False) + order_authority: Literal[False] = field(default=False, init=False) + capital_authority: Literal[False] = field(default=False, init=False) + canonical_digest: str = field(default="", init=False) + + def __post_init__(self) -> None: + as_of = _parse_utc(self.as_of, "as_of") + expires_at = _parse_utc(self.expires_at, "expires_at") + if expires_at <= as_of: + raise ValueError("expires_at must be after as_of") + if len(self.observation_set_digest) != 64: + raise ValueError("observation_set_digest must be a SHA-256 hex digest") + if self.model.seed < 0 or self.model.draws <= 0: + raise ValueError("Forecast model seed and draw count are invalid") + outcome = self.distributions.outcome + probabilities = (outcome.home_win, outcome.away_win, outcome.tie) + if any(not math.isfinite(value) or not 0.0 <= value <= 1.0 for value in probabilities): + raise ValueError("Outcome distribution contains an invalid probability") + if not math.isclose(sum(probabilities), 1.0, abs_tol=0.0002): + raise ValueError("Outcome probabilities must sum to one") + payload = asdict(self) + payload.pop("canonical_digest", None) + object.__setattr__(self, "canonical_digest", _digest(payload)) + + def as_dict(self) -> dict[str, object]: + return asdict(self) + + +def _score_distribution(mean: float, interval: ScoreInterval) -> ScoreDistributionV1: + return ScoreDistributionV1( + mean=mean, + lower=interval.low, + upper=interval.high, + interval_mass=0.8, + ) + + +def _source_provenance( + revisions: tuple[GameObservationRevision, ...], +) -> tuple[SourceProvenanceV1, ...]: + by_source: dict[str, list[GameObservationRevision]] = {} + for revision in revisions: + by_source.setdefault(revision.game.source, []).append(revision) + output: list[SourceProvenanceV1] = [] + for source in sorted(by_source): + source_revisions = by_source[source] + output.append( + SourceProvenanceV1( + source=source, + observation_count=len(source_revisions), + latest_observed_at=max( + _parse_utc(revision.game.observed_at, "observed_at") + for revision in source_revisions + ).isoformat(), + source_urls=tuple( + sorted({revision.game.source_url for revision in source_revisions}) + ), + payload_set_digest=_digest( + sorted(revision.game.payload_hash for revision in source_revisions) + ), + ) + ) + return tuple(output) + + +def _data_rights(revisions: tuple[GameObservationRevision, ...]) -> DataRightsV1: + references = tuple( + sorted( + { + revision.game.source_url + for revision in revisions + if revision.game.source_url + } + ) + ) + return DataRightsV1( + status="source_terms_and_local_authorization_must_be_verified", + source_terms_apply=True, + redistribution_status="not_granted_by_this_envelope", + model_training_status="not_granted_by_this_envelope", + references=references, + ) + + +def _distributions(projection: MatchupProjection) -> ForecastDistributionsV1: + return ForecastDistributionsV1( + home_score=_score_distribution( + projection.projected_home_score, + projection.home_score_interval_80, + ), + away_score=_score_distribution( + projection.projected_away_score, + projection.away_score_interval_80, + ), + total_score=_score_distribution( + projection.projected_total, + projection.total_interval_80, + ), + expected_margin=projection.projected_margin, + outcome=OutcomeDistributionV1( + home_win=projection.home_win_probability, + away_win=projection.away_win_probability, + tie=projection.tie_probability, + ), + ) + + +class DumbMoneyForecastExporter: + """Export Waterboy projections without importing or granting execution.""" + + def __init__( + self, + store: SportsDataStore, + projections: ProjectionEngine, + *, + max_target_observation_age: timedelta = timedelta(hours=24), + ) -> None: + if max_target_observation_age <= timedelta(0): + raise ValueError("max_target_observation_age must be positive") + self.store = store + self.projections = projections + self.max_target_observation_age = max_target_observation_age + + def export( + self, + game_id: str, + *, + as_of: datetime, + draws: int = 5_000, + seed: int = 20_260_726, + champion_id: str = DEFAULT_CHAMPION_ID, + champion_version: str = PROJECTION_MODEL_VERSION, + fidelity: str = DEFAULT_FIDELITY, + valid_for: timedelta = timedelta(minutes=15), + ) -> SportsForecastEnvelopeV1: + cutoff = _utc(as_of, "as_of") + if fidelity != DEFAULT_FIDELITY: + raise ValueError("Only evidence-supported F0 exports are available") + if valid_for <= timedelta(0): + raise ValueError("valid_for must be positive") + target_revision = self.store.game_revision_as_of(game_id, as_of=cutoff) + target = target_revision.game + observed_at = _parse_utc(target.observed_at, "target.observed_at") + if observed_at > cutoff: + raise FutureObservationError( + "The target observation was received after the as-of cutoff" + ) + observation_age = cutoff - observed_at + if observation_age > self.max_target_observation_age: + raise StaleObservationError( + "The target observation is stale at the as-of cutoff: " + f"age={observation_age}, maximum={self.max_target_observation_age}" + ) + if target.status != "scheduled": + raise ValueError( + "Candidate exports require the latest as-of target observation " + f"to be scheduled, not {target.status!r}" + ) + starts_at = _parse_utc(target.start_time, "target.start_time") + if starts_at <= cutoff: + raise ValueError("Candidate exports require a future scheduled start time") + context = self.projections.prepare_context(target.league, as_of=cutoff) + projection = self.projections.project_game( + target, + simulations=draws, + seed=seed, + context=context, + ) + history_revisions = tuple( + GameObservationRevision(revision_id, game) + for revision_id, game in zip( + context.observation_revision_ids, + context.history, + strict=True, + ) + ) + revisions = (target_revision, *history_revisions) + if any( + _parse_utc(revision.game.observed_at, "observed_at") > cutoff + for revision in revisions + ): + raise FutureObservationError( + "The selected forecast observation set crosses the as-of cutoff" + ) + revision_ids = tuple(revision.revision_id for revision in revisions) + expiry = min(cutoff + valid_for, starts_at) + sources = _source_provenance(revisions) + limitations = tuple( + dict.fromkeys( + ( + *projection.limitations, + "Candidate-only research artifact; it grants no order or capital authority.", + "Source terms and data rights must be verified by downstream consumers.", + ) + ) + ) + return SportsForecastEnvelopeV1( + as_of=_iso(cutoff), + expires_at=_iso(expiry), + observation_set_digest=observation_set_digest(revision_ids), + target=ForecastTargetV1( + game_id=target.game_id, + league=target.league, + starts_at=_iso(starts_at), + home_team_id=target.home_team_id, + away_team_id=target.away_team_id, + home_team_name=target.home_name, + away_team_name=target.away_name, + neutral_site=target.neutral_site, + ), + model=ForecastModelV1( + model_id="waterboy.projection-engine", + model_version=projection.model_version, + champion_id=champion_id, + champion_version=champion_version, + seed=seed, + draws=draws, + fidelity=fidelity, + ), + distributions=_distributions(projection), + confidence_status=projection.confidence_label, + calibration_status="provisional_not_independently_calibrated", + evidence_status=projection.evidence_status, + freshness_status="fresh_as_of_cutoff", + provenance=ForecastProvenanceV1( + target_revision_id=target_revision.revision_id, + observation_count=len(revisions), + sources=sources, + ), + data_rights=_data_rights(revisions), + limitations=limitations, + ) diff --git a/src/universal_sports_engine/projection.py b/src/universal_sports_engine/projection.py index 771f2f8..d8c9711 100644 --- a/src/universal_sports_engine/projection.py +++ b/src/universal_sports_engine/projection.py @@ -96,6 +96,7 @@ class ProjectionContext: as_of: datetime history: tuple[GameRecord, ...] ratings: dict[str, TeamRating] + observation_revision_ids: tuple[str, ...] @dataclass(slots=True) @@ -284,9 +285,18 @@ def _history( self, league: str, as_of: datetime, - ) -> tuple[tuple[GameRecord, ...], dict[str, TeamRating]]: - history = self.store.completed_games(league, before=as_of.isoformat(), limit=10_000) - return history, self.estimator.estimate(league, history, as_of=as_of) + ) -> tuple[tuple[GameRecord, ...], dict[str, TeamRating], tuple[str, ...]]: + revisions = self.store.completed_game_revisions_as_of( + league, + as_of=as_of, + limit=10_000, + ) + history = tuple(revision.game for revision in revisions) + return ( + history, + self.estimator.estimate(league, history, as_of=as_of), + tuple(revision.revision_id for revision in revisions), + ) def prepare_context( self, @@ -296,8 +306,14 @@ def prepare_context( ) -> ProjectionContext: normalized = normalize_league_id(league) decision_time = (as_of or datetime.now(UTC)).astimezone(UTC) - history, ratings = self._history(normalized, decision_time) - return ProjectionContext(normalized, decision_time, history, ratings) + history, ratings, revision_ids = self._history(normalized, decision_time) + return ProjectionContext( + normalized, + decision_time, + history, + ratings, + revision_ids, + ) def project( self, @@ -326,7 +342,7 @@ def project( raise ValueError(f"seed must be non-negative: {seed}") decision_time = (as_of or datetime.now(UTC)).astimezone(UTC) if context is None: - history, ratings = self._history(normalized, decision_time) + history, ratings, _revision_ids = self._history(normalized, decision_time) else: if context.league != normalized: raise ValueError( diff --git a/src/universal_sports_engine/service.py b/src/universal_sports_engine/service.py index 5131433..b1d3844 100644 --- a/src/universal_sports_engine/service.py +++ b/src/universal_sports_engine/service.py @@ -12,11 +12,12 @@ from universal_sports_engine.data.models import GameRecord from universal_sports_engine.data.pipeline import SportsDataPipeline from universal_sports_engine.data.store import SportsDataStore +from universal_sports_engine.dumbmoney import DumbMoneyForecastExporter from universal_sports_engine.league_packs.registry import ( league_identities, normalize_league_id, ) -from universal_sports_engine.projection import ProjectionEngine +from universal_sports_engine.projection import PROJECTION_MODEL_VERSION, ProjectionEngine LEAGUE_PRESENTATION = { "mlb": {"short_name": "MLB", "accent": "coral"}, @@ -70,6 +71,7 @@ def __init__( self.store = store self.pipeline = pipeline self.projections = projections + self.dumbmoney = DumbMoneyForecastExporter(store, projections) @classmethod def from_paths(cls, data_path: Path, cache_dir: Path) -> SportsEngineService: @@ -201,6 +203,32 @@ def hypothetical( ) return asdict(projection) + def dumbmoney_forecast( + self, + *, + game_id: str, + as_of: datetime, + draws: int = 5_000, + seed: int = 20_260_726, + champion_id: str = "waterboy.projection-incumbent", + champion_version: str = PROJECTION_MODEL_VERSION, + fidelity: str = "F0", + valid_for_seconds: int = 900, + ) -> dict[str, Any]: + """Export a candidate-only, point-in-time DumbMoney forecast envelope.""" + + envelope = self.dumbmoney.export( + game_id, + as_of=as_of, + draws=draws, + seed=seed, + champion_id=champion_id, + champion_version=champion_version, + fidelity=fidelity, + valid_for=timedelta(seconds=valid_for_seconds), + ) + return envelope.as_dict() + def health(self) -> dict[str, Any]: database = self.store.health() return { @@ -208,7 +236,10 @@ def health(self) -> dict[str, Any]: "engine_name": "Waterboy", "engine_version": "0.4.0", "database": database, + "candidate_only": True, "execution_authority": False, + "order_authority": False, + "capital_authority": False, "remote_writes": False, "supported_leagues": list(LEAGUE_ORDER), } diff --git a/src/universal_sports_engine/web_app.py b/src/universal_sports_engine/web_app.py index 4984000..c79814a 100644 --- a/src/universal_sports_engine/web_app.py +++ b/src/universal_sports_engine/web_app.py @@ -4,7 +4,7 @@ import os from collections.abc import Awaitable, Callable -from datetime import date +from datetime import date, datetime from pathlib import Path from typing import Any @@ -39,6 +39,27 @@ class RefreshRequest(BaseModel): force: bool = False +class DumbMoneyForecastRequest(BaseModel): + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + game_id: str = Field(min_length=1, max_length=160) + as_of: datetime + draws: int = Field(default=5_000, ge=100, le=50_000) + seed: int = Field(default=20_260_726, ge=0, le=2_147_483_647) + champion_id: str = Field( + default="waterboy.projection-incumbent", + min_length=1, + max_length=160, + ) + champion_version: str = Field( + default="hierarchical-recency-opponent-adjusted-v1", + min_length=1, + max_length=160, + ) + fidelity: str = Field(default="F0", min_length=2, max_length=16) + valid_for_seconds: int = Field(default=900, ge=1, le=86_400) + + def _runtime_path(primary_name: str, legacy_name: str, fallback: Path) -> Path: value = os.environ.get(primary_name) or os.environ.get(legacy_name) return Path(value) if value else fallback @@ -135,6 +156,10 @@ def refresh(request: RefreshRequest) -> dict[str, Any]: def hypothetical(request: HypotheticalRequest) -> dict[str, Any]: return application_service.hypothetical(**request.model_dump()) + @app.post("/api/v1/exports/dumbmoney/sports-forecast") + def dumbmoney_forecast(request: DumbMoneyForecastRequest) -> dict[str, Any]: + return application_service.dumbmoney_forecast(**request.model_dump()) + static_dir = Path(__file__).with_name("web") / "static" app.mount("/", StaticFiles(directory=static_dir, html=True), name="dashboard") return app diff --git a/tests/fixtures/dumbmoney_sports_forecast_envelope_v1.json b/tests/fixtures/dumbmoney_sports_forecast_envelope_v1.json new file mode 100644 index 0000000..a4dfd46 --- /dev/null +++ b/tests/fixtures/dumbmoney_sports_forecast_envelope_v1.json @@ -0,0 +1,90 @@ +{ + "as_of": "2026-07-26T12:00:00+00:00", + "calibration_status": "provisional_not_independently_calibrated", + "candidate_only": true, + "canonical_digest": "45af7737ff85ff731f34620f771be3afa16d4d84b277fb989cfe4b8c31279d21", + "capital_authority": false, + "confidence_status": "MEDIUM_DATA_COVERAGE", + "data_rights": { + "model_training_status": "not_granted_by_this_envelope", + "redistribution_status": "not_granted_by_this_envelope", + "references": [ + "https://example.test/source-terms" + ], + "source_terms_apply": true, + "status": "source_terms_and_local_authorization_must_be_verified" + }, + "distributions": { + "away_score": { + "interval_mass": 0.8, + "lower": 88, + "mean": 106.3, + "upper": 121 + }, + "expected_margin": 7.8, + "home_score": { + "interval_mass": 0.8, + "lower": 95, + "mean": 114.2, + "upper": 135 + }, + "outcome": { + "away_win": 0.36, + "home_win": 0.63, + "tie": 0.01 + }, + "total_score": { + "interval_mass": 0.8, + "lower": 195, + "mean": 220.5, + "upper": 246 + } + }, + "evidence_status": "historical_form_provisional_not_calibrated", + "expires_at": "2026-07-26T12:10:00+00:00", + "freshness_status": "fresh_as_of_cutoff", + "limitations": [ + "Projection is a simulation, not a guarantee or certified real-world forecast.", + "Roster, injury, player, coaching, weather, and travel inputs are not yet modeled.", + "Confidence describes data coverage and freshness, not demonstrated accuracy.", + "Candidate-only research artifact; it grants no order or capital authority.", + "Source terms and data rights must be verified by downstream consumers." + ], + "model": { + "champion_id": "waterboy.projection-incumbent", + "champion_version": "hierarchical-recency-opponent-adjusted-v1", + "draws": 100, + "fidelity": "F0", + "model_id": "waterboy.projection-engine", + "model_version": "hierarchical-recency-opponent-adjusted-v1", + "seed": 41 + }, + "observation_set_digest": "65098388c106edd44976b733f2360910b25fce29b42edc5c289e66be0a1e3bd2", + "order_authority": false, + "provenance": { + "observation_count": 13, + "sources": [ + { + "latest_observed_at": "2026-07-26T11:30:00+00:00", + "observation_count": 13, + "payload_set_digest": "1aa271bdf01dbd4b9016911aade4a2fbf29675a6f082253c6d2cf9f316fe6093", + "source": "licensed_fixture", + "source_urls": [ + "https://example.test/source-terms" + ] + } + ], + "target_revision_id": "fa733460d4a42699b0866868693ef452368567e1a454f540d12f3c361986cbbd" + }, + "schema": "dumbmoney.sports-forecast-envelope.v1", + "target": { + "away_team_id": "BET", + "away_team_name": "Betas", + "game_id": "target", + "home_team_id": "ALP", + "home_team_name": "Alphas", + "league": "nba", + "neutral_site": false, + "starts_at": "2026-07-27T01:00:00+00:00" + } +} diff --git a/tests/test_dumbmoney_forecast.py b/tests/test_dumbmoney_forecast.py new file mode 100644 index 0000000..796ea30 --- /dev/null +++ b/tests/test_dumbmoney_forecast.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from dataclasses import FrozenInstanceError, asdict, replace +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from universal_sports_engine.data.models import GameRecord +from universal_sports_engine.data.store import ( + ContradictoryObservationError, + FutureObservationError, + SportsDataStore, +) +from universal_sports_engine.dumbmoney import ( + DumbMoneyForecastExporter, + StaleObservationError, + canonical_json_bytes, +) +from universal_sports_engine.projection import ProjectionEngine + +CUTOFF = datetime(2026, 7, 26, 12, tzinfo=UTC) + + +def _game( + game_id: str, + *, + start: datetime, + observed_at: datetime, + status: str = "scheduled", + home: str = "ALP", + away: str = "BET", + home_score: int | None = None, + away_score: int | None = None, + venue: str = "Research Arena", +) -> GameRecord: + identity = ( + f"{game_id}|{start.isoformat()}|{observed_at.isoformat()}|{status}|" + f"{home_score}|{away_score}|{venue}" + ) + return GameRecord( + game_id=game_id, + league="nba", + season=2026, + start_time=start.isoformat(), + status=status, + home_team_id=home, + away_team_id=away, + home_name="Alphas", + away_name="Betas", + home_score=home_score, + away_score=away_score, + neutral_site=False, + venue=venue, + source="licensed_fixture", + source_url="https://example.test/source-terms", + observed_at=observed_at.isoformat(), + payload_hash=hashlib.sha256(identity.encode()).hexdigest(), + ) + + +def _records() -> tuple[GameRecord, ...]: + history: list[GameRecord] = [] + for index in range(12): + start = datetime(2026, 6, 1 + index, 20, tzinfo=UTC) + history.append( + _game( + f"history-{index}", + start=start, + observed_at=start + timedelta(hours=3), + status="final", + home_score=110 + index % 4, + away_score=102 - index % 3, + ) + ) + target = _game( + "target", + start=datetime(2026, 7, 27, 1, tzinfo=UTC), + observed_at=CUTOFF - timedelta(minutes=30), + ) + return (*history, target) + + +def _exporter(tmp_path: Path, records: tuple[GameRecord, ...]) -> DumbMoneyForecastExporter: + store = SportsDataStore(tmp_path / "sports.db") + store.upsert_games(records) + return DumbMoneyForecastExporter(store, ProjectionEngine(store)) + + +def test_envelope_identity_is_deterministic_and_authority_free(tmp_path: Path) -> None: + exporter = _exporter(tmp_path, _records()) + first = exporter.export( + "target", + as_of=CUTOFF, + draws=100, + seed=41, + valid_for=timedelta(minutes=10), + ) + second = exporter.export( + "target", + as_of=CUTOFF, + draws=100, + seed=41, + valid_for=timedelta(minutes=10), + ) + + assert first == second + assert first.schema == "dumbmoney.sports-forecast-envelope.v1" + assert first.candidate_only is True + assert first.order_authority is False + assert first.capital_authority is False + assert first.model.fidelity == "F0" + assert first.model.draws == 100 + assert first.provenance.observation_count == 13 + fixture = json.loads( + ( + Path(__file__).parent + / "fixtures" + / "dumbmoney_sports_forecast_envelope_v1.json" + ).read_text(encoding="utf-8") + ) + assert json.loads(canonical_json_bytes(first.as_dict())) == fixture + payload = asdict(first) + digest = payload.pop("canonical_digest") + assert digest == hashlib.sha256(canonical_json_bytes(payload)).hexdigest() + with pytest.raises(FrozenInstanceError): + first.order_authority = True # type: ignore[misc] + + +def test_as_of_export_rejects_future_target_and_ignores_future_history( + tmp_path: Path, +) -> None: + future_target = replace( + _records()[-1], + observed_at=(CUTOFF + timedelta(seconds=1)).isoformat(), + payload_hash="f" * 64, + ) + exporter = _exporter(tmp_path / "future-target", (*_records()[:-1], future_target)) + with pytest.raises(FutureObservationError, match="after the as-of cutoff"): + exporter.export("target", as_of=CUTOFF, draws=100) + + baseline = _exporter(tmp_path / "baseline", _records()) + future_final = _game( + "late-result", + start=CUTOFF - timedelta(days=1), + observed_at=CUTOFF + timedelta(seconds=1), + status="final", + home_score=200, + away_score=20, + ) + with_future = _exporter(tmp_path / "future-history", (*_records(), future_final)) + baseline_envelope = baseline.export("target", as_of=CUTOFF, draws=100, seed=9) + future_envelope = with_future.export("target", as_of=CUTOFF, draws=100, seed=9) + assert future_envelope == baseline_envelope + + future_target_revision = replace( + _records()[-1], + status="final", + home_score=150, + away_score=50, + observed_at=(CUTOFF + timedelta(hours=14)).isoformat(), + payload_hash="e" * 64, + ) + target_with_future_result = _exporter( + tmp_path / "future-target-revision", + (*_records(), future_target_revision), + ) + historical_target_envelope = target_with_future_result.export( + "target", + as_of=CUTOFF, + draws=100, + seed=9, + ) + assert historical_target_envelope == baseline_envelope + + +def test_stale_or_contradictory_target_data_fails_closed(tmp_path: Path) -> None: + stale_target = replace( + _records()[-1], + observed_at=(CUTOFF - timedelta(hours=25)).isoformat(), + payload_hash="a" * 64, + ) + stale = _exporter(tmp_path / "stale", (*_records()[:-1], stale_target)) + with pytest.raises(StaleObservationError, match="stale"): + stale.export("target", as_of=CUTOFF, draws=100) + + records = _records() + conflicting = replace( + records[-1], + venue="Contradictory Arena", + payload_hash="b" * 64, + ) + contradictory = _exporter(tmp_path / "contradictory", (*records, conflicting)) + with pytest.raises(ContradictoryObservationError, match="Conflicting"): + contradictory.export("target", as_of=CUTOFF, draws=100) + + +def test_game_observations_are_append_only_revisions(tmp_path: Path) -> None: + store = SportsDataStore(tmp_path / "sports.db") + scheduled = _game( + "revision-game", + start=CUTOFF + timedelta(hours=2), + observed_at=CUTOFF - timedelta(hours=2), + ) + in_progress = replace( + scheduled, + status="in_progress", + home_score=12, + away_score=10, + observed_at=(CUTOFF + timedelta(hours=2, minutes=10)).isoformat(), + payload_hash="c" * 64, + ) + final = replace( + scheduled, + status="final", + home_score=99, + away_score=95, + observed_at=(CUTOFF + timedelta(hours=5)).isoformat(), + payload_hash="d" * 64, + ) + store.upsert_games((scheduled,)) + store.upsert_games((in_progress,)) + store.upsert_games((final,)) + + revisions = store.game_revisions("revision-game") + assert [revision.game.status for revision in revisions] == [ + "scheduled", + "in_progress", + "final", + ] + latest = store.games_between( + (CUTOFF - timedelta(days=1)).isoformat(), + (CUTOFF + timedelta(days=1)).isoformat(), + ) + assert len(latest) == 1 + assert latest[0].status == "final" + assert latest[0].home_score == 99 + + connection = sqlite3.connect(store.path) + with pytest.raises(sqlite3.IntegrityError, match="append-only"): + connection.execute( + "UPDATE game_observations SET status='scheduled' WHERE game_id=?", + ("revision-game",), + ) + connection.close() diff --git a/tests/test_projection_service_web.py b/tests/test_projection_service_web.py index db2d001..7391f43 100644 --- a/tests/test_projection_service_web.py +++ b/tests/test_projection_service_web.py @@ -152,6 +152,23 @@ async def exercise_api() -> None: week = await client.get("/api/v1/week?start=2026-07-26&simulations=100") assert week.status_code == 200 assert len(week.json()["games"]) == 1 + dumbmoney = await client.post( + "/api/v1/exports/dumbmoney/sports-forecast", + json={ + "game_id": "upcoming", + "as_of": "2026-07-26T00:00:00+00:00", + "draws": 100, + "seed": 12, + }, + ) + assert dumbmoney.status_code == 200 + assert ( + dumbmoney.json()["schema"] + == "dumbmoney.sports-forecast-envelope.v1" + ) + assert dumbmoney.json()["candidate_only"] is True + assert dumbmoney.json()["order_authority"] is False + assert dumbmoney.json()["capital_authority"] is False matchup = await client.post( "/api/v1/simulations/hypothetical", json={ diff --git a/tests/test_sports_data.py b/tests/test_sports_data.py index cdf0707..f454893 100644 --- a/tests/test_sports_data.py +++ b/tests/test_sports_data.py @@ -2,7 +2,7 @@ import json import sqlite3 -from datetime import date +from datetime import UTC, date, datetime from pathlib import Path from urllib.error import HTTPError @@ -142,8 +142,143 @@ def test_store_and_read_only_dummy_import(tmp_path: Path) -> None: results = pipeline.import_dummy_history(source) assert source.read_bytes() == before assert sum(result.stored for result in results) == 1 + basketball_result = next(result for result in results if result.league == "ncaambb") + assert basketball_result.fetched == 2 + assert basketball_result.stored == 1 + assert basketball_result.status == "partial" + assert basketball_result.error is not None + assert '"same_team_matchup":1' in basketball_result.error games = store.completed_games("ncaam") assert len(games) == 1 assert games[0].league == "ncaambb" assert games[0].home_score == 81 + assert games[0].observed_at == "2026-03-01T21:01:00+00:00" assert {team.team_id for team in store.teams("ncaambb")} == {"AAA", "BBB"} + + +def test_dummy_import_uses_latest_knowable_time_and_quarantines_contradictions( + tmp_path: Path, +) -> None: + source = tmp_path / "dummy-timing.db" + connection = sqlite3.connect(source) + connection.execute( + """ + CREATE TABLE games( + game_id TEXT, league TEXT, season INTEGER, start_time TEXT, status TEXT, + home TEXT, away TEXT, home_score INTEGER, away_score INTEGER, + venue TEXT, neutral INTEGER, source TEXT, provenance_url TEXT, + result_available_at TEXT, received_at TEXT, as_of TEXT, extra TEXT + ) + """ + ) + connection.executemany( + """ + INSERT INTO games VALUES( + ?, 'nba', 2026, ?, 'final', ?, ?, ?, ?, 'Arena', 0, 'espn', + ?, ?, ?, ?, '{}' + ) + """, + ( + ( + "late-result", + "2026-07-01T18:00:00Z", + "AAA", + "BBB", + 101, + 99, + "https://example.test/late", + "2026-07-01T21:00:00Z", + "2026-07-01T19:00:00Z", + "2026-07-01T18:30:00Z", + ), + ( + "late-result", + "2026-07-01T18:00:00Z", + "AAA", + "BBB", + 101, + 99, + "https://example.test/late", + "2026-07-01T21:00:00Z", + "2026-07-01T22:00:00Z", + "2026-07-01T18:30:00Z", + ), + ( + "prestart-result", + "2026-07-02T18:00:00Z", + "CCC", + "DDD", + 90, + 80, + "https://example.test/prestart", + "2026-07-02T17:00:00Z", + "2026-07-02T19:00:00Z", + "2026-07-02T19:00:00Z", + ), + ( + "malformed-time", + "2026-07-03T18:00:00Z", + "EEE", + "FFF", + 88, + 87, + "https://example.test/malformed", + "not-a-timestamp", + "2026-07-03T21:00:00Z", + "2026-07-03T20:00:00Z", + ), + ( + "future-contradiction", + "2026-07-04T18:00:00Z", + "GGG", + "HHH", + 77, + 70, + "https://example.test/future", + "2099-07-04T21:00:00Z", + "2099-07-04T21:01:00Z", + "2099-07-04T21:01:00Z", + ), + ), + ) + connection.commit() + connection.close() + before = source.read_bytes() + + store = SportsDataStore(tmp_path / "target-timing.db") + pipeline = SportsDataPipeline( + store, + EspnScoreboardClient(tmp_path / "cache-timing", transport=lambda *_args: b"{}"), + ) + results = pipeline.import_dummy_history(source) + + assert source.read_bytes() == before + nba_result = next(result for result in results if result.league == "nba") + assert nba_result.fetched == 5 + assert nba_result.stored == 2 + assert nba_result.status == "partial" + assert nba_result.error is not None + quarantine = json.loads(nba_result.error) + assert quarantine["quarantined"] == 3 + assert quarantine["reasons"] == { + "final_result_available_before_start": 1, + "malformed_timestamp:result_available_at": 1, + "source_timestamp_after_import:as_of,received_at,result_available_at": 1, + } + assert len(quarantine["row_set_digest"]) == 64 + + revision = store.game_revisions("espn:nba:late-result") + assert len(revision) == 2 + assert revision[0].game.observed_at == "2026-07-01T21:00:00+00:00" + assert revision[1].game.observed_at == "2026-07-01T22:00:00+00:00" + assert revision[0].game.payload_hash != revision[1].game.payload_hash + before_result = store.completed_game_revisions_as_of( + "nba", + as_of=datetime(2026, 7, 1, 20, tzinfo=UTC), + ) + at_result = store.completed_game_revisions_as_of( + "nba", + as_of=datetime(2026, 7, 1, 21, tzinfo=UTC), + ) + assert before_result == () + assert tuple(item.game.game_id for item in at_result) == ("espn:nba:late-result",)