File size: 3,218 Bytes
ef20ebe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | """
Local, self-contained persistence: an immutable forecast registry plus
model-artifact metadata (spec sections 45-46, 66, 76, 90). SQLite only —
no external DB service, consistent with the single-application /
HF-Spaces-friendly deployment target (section 76).
"""
from __future__ import annotations
import sqlite3
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import numpy as np
SCHEMA = """
CREATE TABLE IF NOT EXISTS forecasts (
id TEXT PRIMARY KEY,
created_at TEXT,
symbol TEXT, timeframe TEXT, horizon INTEGER,
model_checkpoint TEXT, config_hash TEXT,
origin_ts TEXT, price_now REAL,
p10 REAL, p50 REAL, p90 REAL,
action TEXT, rationale TEXT,
realized_price REAL, realized_return REAL, forecast_error REAL,
direction_correct INTEGER
);
CREATE TABLE IF NOT EXISTS model_artifacts (
id TEXT PRIMARY KEY,
created_at TEXT,
checkpoint TEXT, library_versions TEXT,
feature_schema TEXT, seed INTEGER,
training_range TEXT, validation_range TEXT, test_range TEXT
);
"""
class Store:
def __init__(self, path: str = "registry.sqlite3"):
self.path = Path(path)
self.conn = sqlite3.connect(self.path)
self.conn.executescript(SCHEMA)
self.conn.commit()
def record_forecast(self, **fields) -> str:
"""A forecast row is written once and never mutated except by
match_realized_outcome() filling in the realized_* columns
later (section 90: 'A forecast should remain immutable after
creation')."""
fid = str(uuid.uuid4())
fields = {"id": fid, "created_at": datetime.now(timezone.utc).isoformat(), **fields}
cols = ",".join(fields.keys())
qs = ",".join(["?"] * len(fields))
self.conn.execute(f"INSERT INTO forecasts ({cols}) VALUES ({qs})", list(fields.values()))
self.conn.commit()
return fid
def match_realized_outcome(self, forecast_id: str, realized_price: float) -> None:
row = self.conn.execute(
"SELECT price_now, p50 FROM forecasts WHERE id=?", (forecast_id,)
).fetchone()
if row is None:
raise KeyError(f"No forecast with id {forecast_id}")
price_now, p50 = row
realized_return = realized_price / price_now - 1.0
error = abs(p50 - realized_return)
direction_correct: Optional[int] = (
int(np.sign(p50) == np.sign(realized_return)) if realized_return != 0 else None
)
self.conn.execute(
"UPDATE forecasts SET realized_price=?, realized_return=?, forecast_error=?, "
"direction_correct=? WHERE id=?",
(realized_price, realized_return, error, direction_correct, forecast_id),
)
self.conn.commit()
def save_model_artifact(self, **fields) -> str:
aid = str(uuid.uuid4())
fields = {"id": aid, "created_at": datetime.now(timezone.utc).isoformat(), **fields}
cols = ",".join(fields.keys())
qs = ",".join(["?"] * len(fields))
self.conn.execute(f"INSERT INTO model_artifacts ({cols}) VALUES ({qs})", list(fields.values()))
self.conn.commit()
return aid
|