diff --git a/DATA_PIPELINE_LINEAGE.md b/DATA_PIPELINE_LINEAGE.md index e5ad4a1..65d5f29 100644 --- a/DATA_PIPELINE_LINEAGE.md +++ b/DATA_PIPELINE_LINEAGE.md @@ -29,19 +29,6 @@ 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: @@ -64,26 +51,9 @@ 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. 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. +HTTP/API projection contracts. Operationally, recheck `git status` in Dummy after +an import and confirm it is clean. diff --git a/README.md b/README.md index 2a2f482..fc51f52 100644 --- a/README.md +++ b/README.md @@ -183,20 +183,10 @@ 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 56b94f1..4efe776 100644 --- a/src/universal_sports_engine/__init__.py +++ b/src/universal_sports_engine/__init__.py @@ -8,10 +8,6 @@ simulate_many_analytical, simulate_profile_analytical, ) -from universal_sports_engine.dumbmoney import ( - DumbMoneyForecastExporter, - SportsForecastEnvelopeV1, -) from universal_sports_engine.simulation import ( counterfactual_game, simulate_game, @@ -23,8 +19,6 @@ __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 bebe437..dffca2d 100644 --- a/src/universal_sports_engine/data/models.py +++ b/src/universal_sports_engine/data/models.py @@ -32,14 +32,6 @@ 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 d70a99d..7c54e2d 100644 --- a/src/universal_sports_engine/data/pipeline.py +++ b/src/universal_sports_engine/data/pipeline.py @@ -16,116 +16,12 @@ 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: @@ -242,13 +138,7 @@ 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() - 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]] = {} + counts: dict[str, int] = {} with closing( sqlite3.connect( f"file:{source_database.as_posix()}?mode=ro", @@ -272,16 +162,17 @@ def import_dummy_history( break normalized_rows: list[GameRecord] = [] for row in rows: - 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: + 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: 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", @@ -291,145 +182,73 @@ 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("utf-8") + ).encode() normalized_rows.append( GameRecord( game_id=( f"espn:{league}:{external_id}" if source == "espn" - else hashlib.sha256(identity.encode("utf-8")).hexdigest() + else hashlib.sha256(identity.encode()).hexdigest() ), league=league, - season=season, + season=int(row["season"]) if row["season"] is not None else None, 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=home_score, - away_score=away_score, - neutral_site=neutral_site, + 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"]), 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=knowable_at.isoformat(), + observed_at=str( + row["received_at"] + or row["result_available_at"] + or row["as_of"] + or observed_at + ), payload_hash=hashlib.sha256(content).hexdigest(), ) ) - stored_counts[league] = stored_counts.get(league, 0) + 1 + counts[league] = counts.get(league, 0) + 1 self.store.upsert_games(normalized_rows) results: list[IngestResult] = [] for league in league_ids(): - 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()) + count = counts.get(league, 0) result = IngestResult( source="dummy_sports_history_read_only_import", league=league, window_start="historical", window_end="historical", - fetched=fetched, - stored=stored, - status="partial" if quarantined else "ok", + fetched=count, + stored=count, + status="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 9bae78f..a73313d 100644 --- a/src/universal_sports_engine/data/store.py +++ b/src/universal_sports_engine/data/store.py @@ -2,72 +2,16 @@ 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 ( - GameObservationRevision, - GameRecord, - IngestResult, - TeamRecord, -) +from universal_sports_engine.data.models import GameRecord, IngestResult, TeamRecord from universal_sports_engine.league_packs.registry import normalize_league_id -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() +SCHEMA_VERSION = 1 class SportsDataStore: @@ -131,43 +75,6 @@ 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, @@ -192,258 +99,64 @@ 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 - for game in rows: - _parse_utc(game.observed_at, "observed_at") - observation_values = [self._observation_values(game) for game in rows] + 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 + ] with closing(self._connect()) as connection: connection.executemany( """ - INSERT OR IGNORE INTO game_observations( - revision_id, game_id, league, season, start_time, status, + 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(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ) 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 """, - observation_values, + 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 deleted file mode 100644 index 6f447c6..0000000 --- a/src/universal_sports_engine/dumbmoney.py +++ /dev/null @@ -1,408 +0,0 @@ -"""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 d8c9711..771f2f8 100644 --- a/src/universal_sports_engine/projection.py +++ b/src/universal_sports_engine/projection.py @@ -96,7 +96,6 @@ class ProjectionContext: as_of: datetime history: tuple[GameRecord, ...] ratings: dict[str, TeamRating] - observation_revision_ids: tuple[str, ...] @dataclass(slots=True) @@ -285,18 +284,9 @@ def _history( self, league: str, as_of: datetime, - ) -> 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), - ) + ) -> 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) def prepare_context( self, @@ -306,14 +296,8 @@ def prepare_context( ) -> ProjectionContext: normalized = normalize_league_id(league) decision_time = (as_of or datetime.now(UTC)).astimezone(UTC) - history, ratings, revision_ids = self._history(normalized, decision_time) - return ProjectionContext( - normalized, - decision_time, - history, - ratings, - revision_ids, - ) + history, ratings = self._history(normalized, decision_time) + return ProjectionContext(normalized, decision_time, history, ratings) def project( self, @@ -342,7 +326,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, _revision_ids = self._history(normalized, decision_time) + history, ratings = 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 b1d3844..5131433 100644 --- a/src/universal_sports_engine/service.py +++ b/src/universal_sports_engine/service.py @@ -12,12 +12,11 @@ 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 PROJECTION_MODEL_VERSION, ProjectionEngine +from universal_sports_engine.projection import ProjectionEngine LEAGUE_PRESENTATION = { "mlb": {"short_name": "MLB", "accent": "coral"}, @@ -71,7 +70,6 @@ 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: @@ -203,32 +201,6 @@ 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 { @@ -236,10 +208,7 @@ 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 c79814a..4984000 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, datetime +from datetime import date from pathlib import Path from typing import Any @@ -39,27 +39,6 @@ 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 @@ -156,10 +135,6 @@ 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 deleted file mode 100644 index a4dfd46..0000000 --- a/tests/fixtures/dumbmoney_sports_forecast_envelope_v1.json +++ /dev/null @@ -1,90 +0,0 @@ -{ - "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 deleted file mode 100644 index 796ea30..0000000 --- a/tests/test_dumbmoney_forecast.py +++ /dev/null @@ -1,248 +0,0 @@ -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 7391f43..db2d001 100644 --- a/tests/test_projection_service_web.py +++ b/tests/test_projection_service_web.py @@ -152,23 +152,6 @@ 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 f454893..cdf0707 100644 --- a/tests/test_sports_data.py +++ b/tests/test_sports_data.py @@ -2,7 +2,7 @@ import json import sqlite3 -from datetime import UTC, date, datetime +from datetime import date from pathlib import Path from urllib.error import HTTPError @@ -142,143 +142,8 @@ 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",)