feat: bind private TQQQ offline baseline snapshot

src/us_equity_strategies/research/__init__.py

@@ -0,0 +1 @@
+"""Offline research-only contracts."""

src/us_equity_strategies/research/tqqq_offline_baseline.py

@@ -0,0 +1,186 @@
+"""Private-artifact binding for the TQQQ research baseline.
+
+This module validates an explicitly supplied private Yahoo snapshot. It does not
+infer sessions, invoke network providers, run production controls, or persist data.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import csv
+from datetime import date
+import hashlib
+import json
+import math
+from pathlib import Path
+from typing import Any
+
+PROFILE = "tqqq_growth_income_research_baseline_v1"
+SCHEMA = "qsl.research.price_snapshot.v1"
+COLUMNS = ("symbol", "as_of", "open", "high", "low", "close", "volume")
+SYMBOLS = ("QQQ", "TQQQ")
+
+
+class OfflineBaselineContractError(ValueError):
+    """Raised for any malformed or mismatched private snapshot contract."""
+
+
+@dataclass(frozen=True)
+class PriceRow:
+    symbol: str
+    as_of: str
+    open: float
+    high: float
+    low: float
+    close: float
+    volume: float
+
+
+@dataclass(frozen=True)
+class OfflineBaselineInput:
+    profile: str
+    source_revision: str
+    request_start: str
+    request_end_exclusive: str
+    rows: tuple[PriceRow, ...]
+    canonical_bytes: bytes
+    input_digest: str
+    controls_disabled: bool = True
+    research_only: bool = True
+    provider_completeness: str = "unverified"
+    calendar_authority: str = "unverified"
+
+
+def _fail() -> None:
+    raise OfflineBaselineContractError("invalid offline baseline snapshot") from None
+
+
+def _exact_dict(value: Any) -> dict[str, Any]:
+    if type(value) is not dict:
+        _fail()
+    return value
+
+
+def _text(value: Any) -> str:
+    if type(value) is not str or not value or any(ord(ch) < 0x20 for ch in value):
+        _fail()
+    return value
+
+
+def _date_text(value: Any) -> str:
+    text = _text(value)
+    try:
+        parsed = date.fromisoformat(text)
+    except (TypeError, ValueError):
+        _fail()
+    if parsed.isoformat() != text:
+        _fail()
+    return text
+
+
+def _number(value: Any, *, positive: bool = False, nonnegative: bool = False) -> float:
+    if type(value) not in (int, float, str) or isinstance(value, bool):
+        _fail()
+    try:
+        number = float(value)
+    except (TypeError, ValueError, OverflowError):
+        _fail()
+    if not math.isfinite(number) or (positive and number <= 0) or (nonnegative and number < 0):
+        _fail()
+    return 0.0 if number == 0.0 else number
+
+
+def _canonical_csv(rows: tuple[PriceRow, ...]) -> bytes:
+    lines = [",".join(COLUMNS)]
+    for row in rows:
+        lines.append(",".join((row.symbol, row.as_of, *(format(value, ".17g") for value in (row.open, row.high, row.low, row.close, row.volume)))))
+    return ("\n".join(lines) + "\n").encode("utf-8")
+
+
+def _load_manifest(path: Path) -> dict[str, Any]:
+    try:
+        value = json.loads(path.read_bytes().decode("utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError):
+        _fail()
+    manifest = _exact_dict(value)
+    expected = {
+        "schema", "research_only", "provider", "price_field", "provider_completeness",
+        "calendar_authority", "source_revision", "retrieved_at", "symbols", "request", "sha256", "bytes",
+        "counts", "coverage",
+    }
+    if set(manifest) != expected:
+        _fail()
+    if manifest["schema"] != SCHEMA or manifest["research_only"] is not True:
+        _fail()
+    if manifest["provider"] != "yahoo_chart" or manifest["price_field"] != "adjusted_close":
+        _fail()
+    if manifest["provider_completeness"] != "unverified" or manifest["calendar_authority"] != "unverified":
+        _fail()
+    if _text(manifest["source_revision"]) != manifest["source_revision"]:
+        _fail()
+    if _text(manifest["retrieved_at"]) != manifest["retrieved_at"]:
+        _fail()
+    if manifest["symbols"] != list(SYMBOLS):
+        _fail()
+    request = _exact_dict(manifest["request"])
+    if set(request) != {"start", "end_exclusive"}:
+        _fail()
+    request_start, request_end = _date_text(request["start"]), _date_text(request["end_exclusive"])
+    if request_start >= request_end:
+        _fail()
+    if type(manifest["sha256"]) is not str or len(manifest["sha256"]) != 64:
+        _fail()
+    if type(manifest["bytes"]) is not int or isinstance(manifest["bytes"], bool) or manifest["bytes"] < 1:
+        _fail()
+    for key in ("counts", "coverage"):
+        value = _exact_dict(manifest[key])
+        if set(value) != set(SYMBOLS):
+            _fail()
+    return manifest
+
+
+def load_offline_baseline_input(manifest_path: str | Path, artifact_path: str | Path) -> OfflineBaselineInput:
+    """Read and validate one explicit private artifact without network or writes."""
+    manifest = _load_manifest(Path(manifest_path))
+    try:
+        raw = Path(artifact_path).read_bytes()
+    except OSError:
+        _fail()
+    if len(raw) != manifest["bytes"] or hashlib.sha256(raw).hexdigest() != manifest["sha256"]:
+        _fail()
+    try:
+        text = raw.decode("utf-8")
+        records = list(csv.DictReader(text.splitlines()))
+    except (UnicodeError, csv.Error):
+        _fail()
+    if not records or tuple(records[0]) != COLUMNS:
+        _fail()
+    rows_list: list[PriceRow] = []
+    for record in records:
+        if set(record) != set(COLUMNS):
+            _fail()
+        rows_list.append(PriceRow(_text(record["symbol"]), _date_text(record["as_of"]), *(_number(record[key], positive=key != "volume", nonnegative=key == "volume") for key in COLUMNS[2:])))
+    rows = tuple(rows_list)
+    if tuple((row.as_of, row.symbol) for row in rows) != tuple(sorted((row.as_of, row.symbol) for row in rows)):
+        _fail()
+    if len({(row.symbol, row.as_of) for row in rows}) != len(rows) or {row.symbol for row in rows} != set(SYMBOLS):
+        _fail()
+    request = manifest["request"]
+    if any(not (request["start"] <= row.as_of < request["end_exclusive"]) for row in rows):
+        _fail()
+    dates_by_symbol = {
+        symbol: {row.as_of for row in rows if row.symbol == symbol}
+        for symbol in SYMBOLS
+    }
+    if dates_by_symbol[SYMBOLS[0]] != dates_by_symbol[SYMBOLS[1]]:
+        _fail()
+    if _canonical_csv(rows) != raw:
+        _fail()
+    for symbol in SYMBOLS:
+        selected = [row for row in rows if row.symbol == symbol]
+        if len(selected) != manifest["counts"][symbol] or not selected:
+            _fail()
+        coverage = _exact_dict(manifest["coverage"][symbol])
+        if set(coverage) != {"start", "end"} or (coverage["start"], coverage["end"]) != (selected[0].as_of, selected[-1].as_of):
+            _fail()
+    digest = hashlib.sha256(raw).hexdigest()
+    return OfflineBaselineInput(PROFILE, manifest["source_revision"], manifest["request"]["start"], manifest["request"]["end_exclusive"], rows, raw, digest)

tests/test_tqqq_offline_baseline_binding.py

@@ -0,0 +1,101 @@
+import csv
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from us_equity_strategies.research.tqqq_offline_baseline import (
+    OfflineBaselineContractError,
+    load_offline_baseline_input,
+)
+
+
+def _artifact(tmp_path: Path):
+    rows = [
+        ["symbol", "as_of", "open", "high", "low", "close", "volume"],
+        ["QQQ", "2026-01-02", "100", "101", "99", "100.5", "1000"],
+        ["TQQQ", "2026-01-02", "50", "51", "49", "50.25", "2000"],
+        ["QQQ", "2026-01-05", "101", "102", "100", "101.5", "1100"],
+        ["TQQQ", "2026-01-05", "51", "52", "50", "51.25", "2100"],
+    ]
+    path = tmp_path / "prices.csv"
+    raw = "\n".join(",".join(row) for row in rows) + "\n"
+    path.write_bytes(raw.encode())
+    sha = hashlib.sha256(raw.encode()).hexdigest()
+    manifest = {
+        "schema": "qsl.research.price_snapshot.v1",
+        "research_only": True,
+        "provider": "yahoo_chart",
+        "price_field": "adjusted_close",
+        "provider_completeness": "unverified",
+        "calendar_authority": "unverified",
+        "source_revision": "yahoo_chart_request_v1",
+        "retrieved_at": "2026-07-16T08:00:00Z",
+        "symbols": ["QQQ", "TQQQ"],
+        "request": {"start": "2026-01-02", "end_exclusive": "2026-01-06"},
+        "sha256": sha,
+        "bytes": len(raw.encode()),
+        "counts": {"QQQ": 2, "TQQQ": 2},
+        "coverage": {
+            "QQQ": {"start": "2026-01-02", "end": "2026-01-05"},
+            "TQQQ": {"start": "2026-01-02", "end": "2026-01-05"},
+        },
+    }
+    manifest_path = tmp_path / "prices.manifest.json"
+    manifest_path.write_text(json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n")
+    return manifest_path, path, manifest
+
+
+def test_private_binding_validates_and_is_deterministic(tmp_path):
+    manifest, artifact, _ = _artifact(tmp_path)
+    first = load_offline_baseline_input(manifest, artifact)
+    second = load_offline_baseline_input(manifest, artifact)
+    assert first == second
+    assert first.input_digest == hashlib.sha256(first.canonical_bytes).hexdigest()
+    assert first.controls_disabled is True
+    assert first.rows[0].symbol == "QQQ"
+
+
+@pytest.mark.parametrize("field", ["sha256", "bytes", "counts", "coverage"])
+def test_manifest_tamper_fails_closed(tmp_path, field):
+    manifest_path, artifact, manifest = _artifact(tmp_path)
+    manifest[field] = {"bad": 1} if field in {"counts", "coverage"} else 1
+    manifest_path.write_text(json.dumps(manifest))
+    with pytest.raises(OfflineBaselineContractError):
+        load_offline_baseline_input(manifest_path, artifact)
+
+
+def test_artifact_tamper_and_misaligned_dates_fail_closed(tmp_path):
+    manifest, artifact, _ = _artifact(tmp_path)
+    artifact.write_bytes(artifact.read_bytes().replace(b"100.5", b"999.5"))
+    with pytest.raises(OfflineBaselineContractError):
+        load_offline_baseline_input(manifest, artifact)
+
+
+def test_unknown_controls_and_nonfinite_values_fail_closed(tmp_path):
+    manifest, artifact, data = _artifact(tmp_path)
+    data["calendar_authority"] = "trusted"
+    manifest.write_text(json.dumps(data))
+    with pytest.raises(OfflineBaselineContractError):
+        load_offline_baseline_input(manifest, artifact)
+    manifest, artifact, _ = _artifact(tmp_path / "nested") if False else _artifact(tmp_path)
+    lines = artifact.read_text().replace("50.25", "nan")
+    artifact.write_text(lines)
+    with pytest.raises(OfflineBaselineContractError):
+        load_offline_baseline_input(manifest, artifact)
+
+def test_mismatched_symbol_date_sets_fail_closed(tmp_path):
+    manifest, artifact, _ = _artifact(tmp_path)
+    raw = artifact.read_text().replace("TQQQ,2026-01-05", "TQQQ,2026-01-04")
+    artifact.write_text(raw)
+    with pytest.raises(OfflineBaselineContractError):
+        load_offline_baseline_input(manifest, artifact)
+
+
+def test_dates_outside_manifest_window_fail_closed(tmp_path):
+    manifest, artifact, data = _artifact(tmp_path)
+    data["request"] = {"start": "2026-01-03", "end_exclusive": "2026-01-06"}
+    manifest.write_text(json.dumps(data))
+    with pytest.raises(OfflineBaselineContractError):
+        load_offline_baseline_input(manifest, artifact)