feat: close SOXL adjusted-last diagnostics

pyproject.toml

@@ -13,9 +13,9 @@ dependencies = [
   "google-cloud-storage>=2.18",
   "requests[socks]>=2.31",
   "yfinance>=0.2.40",
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@2f75b59289ef24ab47a3ed8d522c9ef8d6aea6b2",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@9618b4bd8e179760ac174914713598762cab15d7",
   "quant-strategy-plugins @ git+https://github.com/QuantStrategyLab/QuantStrategyPlugins.git@1f3a27b8fd83d71b583f4f5160a748e95fbefaa1",
-  "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@f799ad115660b17bc888cbe6e7461255ccee1735",
+  "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@69716807fa61746f3b472aad3bf072d3960bfbd8",
 ]
 
 [project.optional-dependencies]

src/us_equity_snapshot_pipelines/lifecycle/soxl_adjusted_last_acquisition.py

@@ -0,0 +1,173 @@
+"""Strict SOXL adjusted-history wrapper and sanitized failure packager."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+from datetime import date, datetime
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+from quant_platform_kit.ibkr import (
+    StrictAdjustedHistoryError,
+    StrictAdjustedHistoryRequestOutcome,
+    StrictAdjustedHistoryResult,
+    fetch_strict_adjusted_historical_price_candles,
+)
+
+
+_CLASSIFICATIONS = frozenset(
+    {
+        "transport_error",
+        "provider_error",
+        "completion_not_observed",
+        "empty_response",
+        "session_contract_mismatch",
+    }
+)
+_COUNT_KEYS = frozenset(
+    {
+        "expected_count",
+        "observed_in_window_count",
+        "missing_count",
+        "extra_count",
+        "duplicate_count",
+    }
+)
+_COMMITMENT_KEYS = frozenset(
+    {
+        "algorithm",
+        "canonicalization",
+        "missing_sessions_sha256",
+        "extra_sessions_sha256",
+        "duplicate_sessions_sha256",
+    }
+)
+
+
+class SoxlAdjustedLastDiagnosticError(ValueError):
+    """The sanitized adjusted-history diagnostic violated its closed contract."""
+
+
+def acquire_strict_adjusted_last(
+    ib: Any,
+    symbol: str,
+    *,
+    end_datetime: datetime,
+    duration: str,
+    expected_sessions: Sequence[date],
+    stock_factory: Callable[..., Any] | None = None,
+    requester: Callable[..., StrictAdjustedHistoryRequestOutcome],
+) -> StrictAdjustedHistoryResult:
+    """Run the QPK strict contract with explicit completion and error state."""
+    return fetch_strict_adjusted_historical_price_candles(
+        ib,
+        symbol,
+        end_datetime=end_datetime,
+        duration=duration,
+        expected_sessions=expected_sessions,
+        stock_factory=stock_factory,
+        requester=requester,
+    )
+
+
+def _is_sha256(value: object) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == 64
+        and all(character in "0123456789abcdef" for character in value)
+    )
+
+
+def _sanitized_payload(error: StrictAdjustedHistoryError) -> dict[str, Any]:
+    diagnostic = error.diagnostic
+    if diagnostic is None:
+        raise SoxlAdjustedLastDiagnosticError("missing sanitized diagnostic")
+    payload = diagnostic.to_dict()
+    if set(payload) != {
+        "schema_version",
+        "classification",
+        "request_completion_observed",
+        "counts",
+        "commitments",
+        "provider_error_code_counts",
+    }:
+        raise SoxlAdjustedLastDiagnosticError("invalid sanitized diagnostic")
+    if (
+        payload["schema_version"] != "strict_adjusted_history_diagnostic.v1"
+        or payload["classification"] not in _CLASSIFICATIONS
+        or not isinstance(payload["request_completion_observed"], bool)
+    ):
+        raise SoxlAdjustedLastDiagnosticError("invalid sanitized diagnostic")
+
+    counts = payload["counts"]
+    if (
+        not isinstance(counts, dict)
+        or set(counts) != _COUNT_KEYS
+        or any(
+            isinstance(value, bool) or not isinstance(value, int) or value < 0
+            for value in counts.values()
+        )
+    ):
+        raise SoxlAdjustedLastDiagnosticError("invalid sanitized diagnostic")
+
+    commitments = payload["commitments"]
+    if (
+        not isinstance(commitments, dict)
+        or set(commitments) != _COMMITMENT_KEYS
+        or commitments["algorithm"] != "sha256"
+        or commitments["canonicalization"]
+        != "sorted_unique_iso_sessions_json_utf8.v1"
+        or any(
+            not _is_sha256(commitments[key])
+            for key in (
+                "missing_sessions_sha256",
+                "extra_sessions_sha256",
+                "duplicate_sessions_sha256",
+            )
+        )
+    ):
+        raise SoxlAdjustedLastDiagnosticError("invalid sanitized diagnostic")
+
+    errors = payload["provider_error_code_counts"]
+    if not isinstance(errors, dict) or any(
+        not isinstance(code, str)
+        or not code.isdigit()
+        or isinstance(count, bool)
+        or not isinstance(count, int)
+        or count <= 0
+        for code, count in errors.items()
+    ):
+        raise SoxlAdjustedLastDiagnosticError("invalid sanitized diagnostic")
+    return payload
+
+
+def write_sanitized_adjusted_last_diagnostic(
+    destination: str | Path,
+    error: StrictAdjustedHistoryError,
+) -> None:
+    """Create one exclusive mode-0600 JSON diagnostic without raw market data."""
+    path = Path(destination)
+    payload = json.dumps(
+        _sanitized_payload(error),
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+        allow_nan=False,
+    ).encode("utf-8")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    descriptor = os.open(path, flags, 0o600)
+    try:
+        os.fchmod(descriptor, 0o600)
+        with os.fdopen(descriptor, "wb", closefd=False) as handle:
+            handle.write(payload)
+            handle.flush()
+            os.fsync(handle.fileno())
+    except BaseException:
+        path.unlink(missing_ok=True)
+        raise
+    finally:
+        os.close(descriptor)

src/us_equity_snapshot_pipelines/lifecycle/soxl_pit_input_packager.py

@@ -54,7 +54,7 @@
     "BOXX": "2022-12-27",
     "QQQI": "2024-01-29",
 }
-QPK_REVISION = "2f75b59289ef24ab47a3ed8d522c9ef8d6aea6b2"
+QPK_REVISION = "9618b4bd8e179760ac174914713598762cab15d7"
 INPUT_CONTRACT_ID = "soxl_p3_core_only_9_input.v1"
 MANDATE_ID = "soxl_p3_core_only_9_input_research_v1"
 _FROZEN_CALENDAR_SHA256 = "6e3bf4713cca22264987c583cf4c5c94923850de4a3d18e76f66f42e719f2290"

src/us_equity_snapshot_pipelines/lifecycle/soxl_pit_regime_component_producer.py

@@ -308,8 +308,8 @@ def validate_soxl_pit_regime_source_contract(
     expected_calendar = {
         "calendar_id": "XNYS",
         "timezone": "America/New_York",
-        "source": "exchange_calendars",
-        "source_revision": "4.13.2",
+        "source": "uesp_repo_local_xnys_holiday_rules",
+        "source_revision": "soxl_pit_input_packager.v1",
         "first_session": expected_sessions[0],
         "last_session": expected_sessions[-1],
         "session_count": len(expected_sessions),

src/us_equity_snapshot_pipelines/lifecycle/soxl_promotion_runner.py

@@ -61,8 +61,8 @@
     "QQQI",
     "QQQ",
 )
-_QPK_REVISION = "2f75b59289ef24ab47a3ed8d522c9ef8d6aea6b2"
-_UES_REVISION = "f799ad115660b17bc888cbe6e7461255ccee1735"
+_QPK_REVISION = "9618b4bd8e179760ac174914713598762cab15d7"
+_UES_REVISION = "69716807fa61746f3b472aad3bf072d3960bfbd8"
 _PROFILE = "soxl_soxx_trend_income"
 _DOMAIN = "us_equity"
 _MIN_INDICATOR_SESSIONS = 420

tests/test_soxl_adjusted_last_acquisition.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+import json
+from pathlib import Path
+import stat
+from types import SimpleNamespace
+
+import pytest
+
+from quant_platform_kit.ibkr import (
+    StrictAdjustedHistoryError,
+    StrictAdjustedHistoryRequestOutcome,
+)
+from us_equity_snapshot_pipelines.lifecycle.soxl_adjusted_last_acquisition import (
+    acquire_strict_adjusted_last,
+    write_sanitized_adjusted_last_diagnostic,
+)
+
+
+EXPECTED = (date(2026, 8, 1), date(2026, 8, 2))
+CUTOFF = datetime(2026, 8, 5, 3, 59, 59, tzinfo=timezone.utc)
+
+
+def _bar(session: date, close: float = 98_765.4321) -> SimpleNamespace:
+    return SimpleNamespace(
+        date=session,
+        open=close,
+        high=close + 1.0,
+        low=close - 1.0,
+        close=close,
+        volume=1_000.0,
+    )
+
+
+class OfflineIB:
+    def __init__(self) -> None:
+        self.history_calls = 0
+
+    def qualifyContracts(self, contract):
+        return [contract]
+
+    def reqHistoricalData(self, _contract, **_kwargs):
+        self.history_calls += 1
+        raise AssertionError("provider calls are forbidden in synthetic tests")
+
+
+def _stock(symbol: str, exchange: str, currency: str) -> SimpleNamespace:
+    return SimpleNamespace(symbol=symbol, exchange=exchange, currency=currency)
+
+
+def _requester(*, bars, completion_observed: bool, provider_error_codes=()):
+    def request(_contract, **kwargs):
+        assert kwargs["whatToShow"] == "ADJUSTED_LAST"
+        assert kwargs["useRTH"] is True
+        return StrictAdjustedHistoryRequestOutcome(
+            bars=bars,
+            completion_observed=completion_observed,
+            provider_error_codes=provider_error_codes,
+        )
+
+    return request
+
+
+@pytest.mark.parametrize(
+    ("classification", "bars", "completion_observed", "provider_error_codes", "counts"),
+    [
+        ("session_contract_mismatch", [_bar(EXPECTED[0])], True, (), (1, 0, 0)),
+        (
+            "session_contract_mismatch",
+            [_bar(EXPECTED[0]), _bar(EXPECTED[1]), _bar(date(2026, 8, 3))],
+            True,
+            (),
+            (0, 1, 0),
+        ),
+        ("session_contract_mismatch", [_bar(EXPECTED[0]), _bar(EXPECTED[0])], True, (), (1, 0, 1)),
+        ("empty_response", [], True, (), (2, 0, 0)),
+        ("provider_error", [_bar(EXPECTED[0])], False, (10089, 10089), (1, 0, 0)),
+        ("completion_not_observed", [_bar(EXPECTED[0])], False, (), (1, 0, 0)),
+    ],
+)
+def test_failures_emit_only_sanitized_diagnostic(
+    tmp_path: Path,
+    classification: str,
+    bars,
+    completion_observed: bool,
+    provider_error_codes,
+    counts: tuple[int, int, int],
+) -> None:
+    ib = OfflineIB()
+    with pytest.raises(StrictAdjustedHistoryError) as caught:
+        acquire_strict_adjusted_last(
+            ib,
+            "SOXL",
+            end_datetime=CUTOFF,
+            duration="9 Y",
+            expected_sessions=EXPECTED,
+            stock_factory=_stock,
+            requester=_requester(
+                bars=bars,
+                completion_observed=completion_observed,
+                provider_error_codes=provider_error_codes,
+            ),
+        )
+
+    destination = tmp_path / f"{classification}.json"
+    write_sanitized_adjusted_last_diagnostic(destination, caught.value)
+    payload = json.loads(destination.read_bytes())
+    assert payload["classification"] == classification
+    assert (
+        payload["counts"]["missing_count"],
+        payload["counts"]["extra_count"],
+        payload["counts"]["duplicate_count"],
+    ) == counts
+    assert payload["provider_error_code_counts"] == (
+        {"10089": 2} if provider_error_codes else {}
+    )
+    assert stat.S_IMODE(destination.stat().st_mode) == 0o600
+    assert ib.history_calls == 0
+
+    serialized = destination.read_text(encoding="utf-8")
+    for forbidden in (
+        "2026-08-01",
+        "2026-08-02",
+        "2026-08-03",
+        "98765.4321",
+        "open",
+        "high",
+        "low",
+        "close",
+        "volume",
+        "provider message",
+        "secret",
+    ):
+        assert forbidden not in serialized
+
+
+def test_exact_match_preserves_strict_request_and_returns_no_failure_artifact(
+    tmp_path: Path,
+) -> None:
+    ib = OfflineIB()
+    result = acquire_strict_adjusted_last(
+        ib,
+        "SOXL",
+        end_datetime=CUTOFF,
+        duration="9 Y",
+        expected_sessions=EXPECTED,
+        stock_factory=_stock,
+        requester=_requester(
+            bars=[_bar(EXPECTED[0]), _bar(EXPECTED[1])],
+            completion_observed=True,
+        ),
+    )
+
+    assert result.diagnostic.to_dict()["classification"] == "exact_match"
+    assert tuple(candle.session for candle in result.candles) == EXPECTED
+    assert list(tmp_path.iterdir()) == []
+    assert ib.history_calls == 0

tests/test_soxl_pit_input_packager.py

@@ -115,8 +115,8 @@ def _source_contract(rows: list[dict[str, object]]) -> dict[str, object]:
         "calendar": {
             "calendar_id": "XNYS",
             "timezone": "America/New_York",
-            "source": "exchange_calendars",
-            "source_revision": "4.13.2",
+            "source": "uesp_repo_local_xnys_holiday_rules",
+            "source_revision": "soxl_pit_input_packager.v1",
             "first_session": FROZEN_XNYS_SESSIONS[0],
             "last_session": FROZEN_XNYS_SESSIONS[-1],
             "session_count": len(FROZEN_XNYS_SESSIONS),