Add offline E3 receipt validation

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

runtime_preflight.py

@@ -4,13 +4,28 @@
 import os
 import re
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
 from runtime_config_support import PlatformRuntimeSettings, load_platform_runtime_settings
 from strategy_loader import load_strategy_entrypoint_for_profile
 
 
+E3_RECEIPT_SCHEMA_VERSION = "qmt.e3.receipt.v1"
+E3_RECEIPT_REQUIRED_SUMMARY_COUNTS = ("accounts", "positions", "orders", "fills", "cash", "ledger")
+_E3_RECEIPT_FIELDS = frozenset(
+    {
+        "schema_version",
+        "as_of",
+        "freshness_seconds",
+        "no_order",
+        "verify_only",
+        "summary_counts",
+    }
+)
+
+
 @dataclass(frozen=True)
 class PreflightIssue:
     code: str
@@ -42,6 +57,109 @@ def to_payload(self) -> dict[str, Any]:
         }
 
 
+@dataclass(frozen=True)
+class E3ReceiptValidationReport:
+    status: str
+    schema_version: str | None
+    issues: tuple[PreflightIssue, ...]
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "status": self.status,
+            "schema_version": self.schema_version,
+            "required_summary_counts": list(E3_RECEIPT_REQUIRED_SUMMARY_COUNTS),
+            "issues": [
+                {"code": issue.code, "message": issue.message}
+                for issue in self.issues
+            ],
+        }
+
+
+def validate_e3_receipt(
+    receipt: object,
+    *,
+    now: datetime | None = None,
+) -> E3ReceiptValidationReport:
+    """Validate a local E3 receipt without connecting to miniQMT or a provider."""
+    if not isinstance(receipt, dict):
+        return E3ReceiptValidationReport(
+            status="error",
+            schema_version=None,
+            issues=(PreflightIssue("invalid_receipt", "Receipt must be a JSON object."),),
+        )
+
+    issues: list[PreflightIssue] = []
+    schema_version = receipt.get("schema_version")
+    reported_schema_version = schema_version if schema_version == E3_RECEIPT_SCHEMA_VERSION else None
+
+    receipt_fields = frozenset(receipt)
+    if _E3_RECEIPT_FIELDS - receipt_fields:
+        issues.append(PreflightIssue("missing_receipt_fields", "Receipt has required fields missing."))
+    if receipt_fields - _E3_RECEIPT_FIELDS:
+        issues.append(PreflightIssue("unexpected_receipt_fields", "Receipt must not contain detail-bearing fields."))
+    if schema_version != E3_RECEIPT_SCHEMA_VERSION:
+        issues.append(PreflightIssue("invalid_schema_version", "Receipt schema version is not accepted."))
+
+    _validate_e3_summary_counts(receipt.get("summary_counts"), issues)
+    _validate_e3_declarations(receipt, issues)
+    _validate_e3_freshness(receipt, now=now, issues=issues)
+
+    return E3ReceiptValidationReport(
+        status="ok" if not issues else "error",
+        schema_version=reported_schema_version,
+        issues=tuple(issues),
+    )
+
+
+def _validate_e3_summary_counts(value: object, issues: list[PreflightIssue]) -> None:
+    if not isinstance(value, dict) or frozenset(value) != frozenset(E3_RECEIPT_REQUIRED_SUMMARY_COUNTS):
+        issues.append(PreflightIssue("invalid_summary_counts", "Receipt must contain only the required summary counts."))
+        return
+    if any(not isinstance(count, int) or isinstance(count, bool) or count < 0 for count in value.values()):
+        issues.append(PreflightIssue("invalid_summary_counts", "Receipt summary counts must be non-negative integers."))
+
+
+def _validate_e3_declarations(receipt: dict[object, object], issues: list[PreflightIssue]) -> None:
+    if receipt.get("no_order") is not True:
+        issues.append(PreflightIssue("no_order_required", "Receipt must declare no_order=true."))
+    if receipt.get("verify_only") is not True:
+        issues.append(PreflightIssue("verify_only_required", "Receipt must declare verify_only=true."))
+
+
+def _validate_e3_freshness(
+    receipt: dict[object, object],
+    *,
+    now: datetime | None,
+    issues: list[PreflightIssue],
+) -> None:
+    freshness_seconds = receipt.get("freshness_seconds")
+    if not isinstance(freshness_seconds, int) or isinstance(freshness_seconds, bool) or freshness_seconds < 0:
+        issues.append(PreflightIssue("invalid_freshness", "Receipt freshness must be a non-negative integer."))
+        return
+
+    as_of = receipt.get("as_of")
+    if not isinstance(as_of, str):
+        issues.append(PreflightIssue("invalid_as_of", "Receipt as_of must be a timezone-aware timestamp."))
+        return
+    try:
+        as_of_time = datetime.fromisoformat(as_of.replace("Z", "+00:00"))
+    except ValueError:
+        issues.append(PreflightIssue("invalid_as_of", "Receipt as_of must be a timezone-aware timestamp."))
+        return
+    if as_of_time.tzinfo is None:
+        issues.append(PreflightIssue("invalid_as_of", "Receipt as_of must be a timezone-aware timestamp."))
+        return
+
+    reference_time = now or datetime.now(timezone.utc)
+    if reference_time.tzinfo is None:
+        reference_time = reference_time.replace(tzinfo=timezone.utc)
+    age_seconds = (reference_time - as_of_time).total_seconds()
+    if age_seconds < 0:
+        issues.append(PreflightIssue("invalid_as_of", "Receipt as_of cannot be in the future."))
+    elif age_seconds > freshness_seconds:
+        issues.append(PreflightIssue("stale_receipt", "Receipt is older than its declared freshness window."))
+
+
 def run_preflight(*, paper_admission: bool = False) -> QmtPreflightReport:
     issues: list[PreflightIssue] = []
     try:

scripts/preflight_qmt_runtime.py

@@ -10,7 +10,7 @@
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
-from runtime_preflight import run_preflight  # noqa: E402
+from runtime_preflight import run_preflight, validate_e3_receipt  # noqa: E402
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -20,11 +20,28 @@ def main(argv: list[str] | None = None) -> int:
         action="store_true",
         help="Validate the offline paper-admission contract without calling QMT or creating orders.",
     )
+    parser.add_argument(
+        "--e3-receipt",
+        type=Path,
+        help="Validate a local, sanitized E3 receipt without connecting to QMT or creating orders.",
+    )
     args = parser.parse_args(argv)
 
     report = run_preflight(paper_admission=args.paper_admission)
-    print(json.dumps(report.to_payload(), ensure_ascii=False, indent=2))
-    return 0 if report.status == "ok" else 2
+    payload = report.to_payload()
+    receipt_status = "ok"
+    if args.e3_receipt:
+        receipt_status = "error"
+        try:
+            receipt = json.loads(args.e3_receipt.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            receipt = None
+        receipt_report = validate_e3_receipt(receipt)
+        payload["e3_receipt"] = receipt_report.to_payload()
+        receipt_status = receipt_report.status
+
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+    return 0 if report.status == "ok" and receipt_status == "ok" else 2
 
 
 if __name__ == "__main__":

tests/test_runtime_preflight.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import hashlib
+import json
+from datetime import datetime, timezone
 from pathlib import Path
 
 import pytest
 
-from runtime_preflight import run_preflight
+from runtime_preflight import E3_RECEIPT_SCHEMA_VERSION, run_preflight, validate_e3_receipt
 from scripts.preflight_qmt_runtime import main
 
 
@@ -17,6 +19,24 @@ def _sha256(path: str) -> str:
     return hashlib.sha256(Path(path).read_bytes()).hexdigest()
 
 
+def _valid_e3_receipt() -> dict[str, object]:
+    return {
+        "schema_version": E3_RECEIPT_SCHEMA_VERSION,
+        "as_of": "2026-09-05T00:00:00+00:00",
+        "freshness_seconds": 60,
+        "no_order": True,
+        "verify_only": True,
+        "summary_counts": {
+            "accounts": 1,
+            "positions": 0,
+            "orders": 0,
+            "fills": 0,
+            "cash": 1,
+            "ledger": 1,
+        },
+    }
+
+
 def test_preflight_accepts_primary_dry_run_config(monkeypatch, market_history_csv: str):
     monkeypatch.setenv("STRATEGY_PROFILE", "cn_industry_etf_rotation")
     monkeypatch.setenv("QMT_DRY_RUN_ONLY", "true")
@@ -136,3 +156,79 @@ def test_preflight_rejects_research_only_dividend_profile(monkeypatch):
 
     assert report.status == "error"
     assert _issue_codes(report) == {"runtime_config_error"}
+
+
+def test_e3_receipt_validator_accepts_complete_sanitized_receipt():
+    report = validate_e3_receipt(
+        _valid_e3_receipt(),
+        now=datetime(2026, 9, 5, 0, 0, 30, tzinfo=timezone.utc),
+    )
+
+    assert report.status == "ok"
+    assert report.schema_version == E3_RECEIPT_SCHEMA_VERSION
+    assert report.issues == ()
+    assert report.to_payload() == {
+        "status": "ok",
+        "schema_version": E3_RECEIPT_SCHEMA_VERSION,
+        "required_summary_counts": ["accounts", "positions", "orders", "fills", "cash", "ledger"],
+        "issues": [],
+    }
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "expected_code"),
+    [
+        ("summary_counts", {"accounts": 1}, "invalid_summary_counts"),
+        ("as_of", "not-a-timestamp", "invalid_as_of"),
+        ("freshness_seconds", -1, "invalid_freshness"),
+        ("no_order", False, "no_order_required"),
+        ("verify_only", False, "verify_only_required"),
+    ],
+)
+def test_e3_receipt_validator_fails_closed_for_missing_or_unsafe_fields(field, value, expected_code):
+    receipt = _valid_e3_receipt()
+    receipt[field] = value
+
+    report = validate_e3_receipt(receipt, now=datetime(2026, 9, 5, tzinfo=timezone.utc))
+
+    assert report.status == "error"
+    assert expected_code in _issue_codes(report)
+
+
+def test_e3_receipt_validator_fails_closed_for_stale_or_detail_bearing_receipts():
+    stale_report = validate_e3_receipt(
+        _valid_e3_receipt(),
+        now=datetime(2026, 9, 5, 0, 6, tzinfo=timezone.utc),
+    )
+    detail_receipt = _valid_e3_receipt()
+    detail_receipt["account"] = {"id": "must-not-be-accepted"}
+    detail_report = validate_e3_receipt(detail_receipt, now=datetime(2026, 9, 5, tzinfo=timezone.utc))
+
+    assert _issue_codes(stale_report) == {"stale_receipt"}
+    assert _issue_codes(detail_report) == {"unexpected_receipt_fields"}
+    assert "must-not-be-accepted" not in str(detail_report.to_payload())
+
+
+def test_e3_receipt_validator_fails_closed_for_missing_required_fields():
+    receipt = _valid_e3_receipt()
+    receipt.pop("verify_only")
+
+    report = validate_e3_receipt(receipt, now=datetime(2026, 9, 5, tzinfo=timezone.utc))
+
+    assert _issue_codes(report) == {"missing_receipt_fields", "verify_only_required"}
+
+
+def test_preflight_cli_validates_e3_receipt_offline(monkeypatch, market_history_csv: str, tmp_path, capsys):
+    monkeypatch.setenv("STRATEGY_PROFILE", "cn_industry_etf_rotation")
+    monkeypatch.setenv("QMT_DRY_RUN_ONLY", "true")
+    monkeypatch.setenv("QMT_MARKET_HISTORY_PATH", market_history_csv)
+    monkeypatch.delenv("RUNTIME_TARGET_JSON", raising=False)
+    receipt_path = tmp_path / "e3-receipt.json"
+    receipt = _valid_e3_receipt()
+    receipt["as_of"] = datetime.now(timezone.utc).isoformat()
+    receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
+
+    assert main(["--e3-receipt", str(receipt_path)]) == 0
+
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["e3_receipt"]["status"] == "ok"