feat: gate Firstrade PAPER admission

application/paper_execution_admission.py

@@ -0,0 +1,146 @@
+"""Fail-closed, opt-in PAPER admission for Firstrade dry-run requests.
+
+The shared QPK contract verifies an immutable execution command, its embedded
+deterministic-risk receipt, and the release currently loaded by the runtime.
+This adapter intentionally only turns that pure result into a redacted HTTP
+audit record.  It neither creates commands nor reaches a broker, account,
+strategy, deployment, scheduler, or persistence service.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+
+from quant_platform_kit.common.execution_commands import ExecutionCommand
+from quant_platform_kit.common.paper_execution_admission import (
+    PaperExecutionAdmissionDecision,
+    evaluate_paper_execution_admission,
+)
+
+
+PAPER_ADMISSION_ENABLED_ENV = "QSL_PAPER_ADMISSION_ENABLED"
+PAPER_EXECUTION_COMMAND_ENV = "QSL_PAPER_EXECUTION_COMMAND_JSON"
+PAPER_EXECUTION_ADMISSION_AUDIT_SCHEMA_VERSION = "firstrade_paper_execution_admission_audit.v1"
+
+_TRUE_VALUES = frozenset({"1", "true", "yes", "on"})
+_FALSE_VALUES = frozenset({"", "0", "false", "no", "off"})
+_COMMAND_MISSING = "paper_execution_command_missing"
+_COMMAND_INVALID = "paper_execution_command_invalid"
+_PLATFORM_MISMATCH = "paper_execution_platform_mismatch"
+_ADMISSION_CONFIGURATION_INVALID = "paper_execution_admission_configuration_invalid"
+_ADMISSION_EVALUATION_FAILED = "paper_execution_admission_evaluation_failed"
+_RUNTIME_MODE_INVALID = "paper_runtime_mode_invalid"
+
+
+def paper_dry_run_admission_requested(env: Mapping[str, str | None]) -> bool:
+    """Return whether this request should be checked instead of using legacy dry-run.
+
+    The default is disabled.  A malformed non-empty enable value is treated as
+    requested so the caller receives a fail-closed audit rather than silently
+    falling back to the ungated legacy preview path.
+    """
+    raw_value = env.get(PAPER_ADMISSION_ENABLED_ENV)
+    if raw_value is None:
+        return False
+    return str(raw_value).strip().lower() not in _FALSE_VALUES
+
+
+def _enabled(env: Mapping[str, str | None]) -> bool:
+    return str(env.get(PAPER_ADMISSION_ENABLED_ENV) or "").strip().lower() in _TRUE_VALUES
+
+
+def _audit(
+    *,
+    disposition: str,
+    findings: tuple[str, ...] | list[str],
+    command_id: str | None = None,
+    receipt_sha256: str | None = None,
+) -> dict[str, object]:
+    admitted = disposition == "allow_new_risk" and not findings
+    return {
+        "schema_version": PAPER_EXECUTION_ADMISSION_AUDIT_SCHEMA_VERSION,
+        "admission_enabled": True,
+        "audit_color": "green" if admitted else "red",
+        "status": "admitted" if admitted else "blocked",
+        "command_id": command_id,
+        "disposition": disposition,
+        "integrity_findings": list(findings),
+        "receipt_sha256": receipt_sha256,
+    }
+
+
+def _audit_from_decision(decision: PaperExecutionAdmissionDecision) -> dict[str, object]:
+    """Return only QPK's stable decision metadata, never raw command intent."""
+    return _audit(
+        disposition=decision.disposition.value,
+        findings=decision.integrity_findings,
+        command_id=decision.command_id,
+        receipt_sha256=decision.receipt_sha256,
+    )
+
+
+def evaluate_paper_dry_run_admission(
+    *,
+    runtime_target: object | None,
+    env: Mapping[str, str | None],
+) -> dict[str, object] | None:
+    """Evaluate an optional QPK PAPER admission before any preview can start.
+
+    ``None`` preserves the existing dry-run route while the feature is disabled.
+    Once requested, every malformed or incomplete input produces a red audit
+    record and the caller must avoid invoking the strategy cycle.
+    """
+    if not paper_dry_run_admission_requested(env):
+        return None
+    if not _enabled(env):
+        return _audit(
+            disposition="halted",
+            findings=(_ADMISSION_CONFIGURATION_INVALID,),
+        )
+
+    raw_command = env.get(PAPER_EXECUTION_COMMAND_ENV)
+    if not raw_command:
+        return _audit(disposition="halted", findings=(_COMMAND_MISSING,))
+    try:
+        payload = json.loads(raw_command)
+        if not isinstance(payload, Mapping):
+            raise ValueError("command payload must be an object")
+        command = ExecutionCommand.from_dict(payload)
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return _audit(disposition="halted", findings=(_COMMAND_INVALID,))
+
+    if command.platform != "firstrade":
+        return _audit(
+            disposition="halted",
+            findings=(_PLATFORM_MISMATCH,),
+            command_id=command.command_id,
+        )
+    if getattr(runtime_target, "dry_run_only", None) is not True:
+        return _audit(
+            disposition="halted",
+            findings=(_RUNTIME_MODE_INVALID,),
+            command_id=command.command_id,
+        )
+    expected_release = getattr(runtime_target, "strategy_release", None)
+    try:
+        decision = evaluate_paper_execution_admission(
+            command=command,
+            expected_strategy_release=expected_release,
+        )
+    except (TypeError, ValueError):
+        return _audit(
+            disposition="halted",
+            findings=(_ADMISSION_EVALUATION_FAILED,),
+            command_id=command.command_id,
+        )
+    return _audit_from_decision(decision)
+
+
+__all__ = [
+    "PAPER_ADMISSION_ENABLED_ENV",
+    "PAPER_EXECUTION_ADMISSION_AUDIT_SCHEMA_VERSION",
+    "PAPER_EXECUTION_COMMAND_ENV",
+    "evaluate_paper_dry_run_admission",
+    "paper_dry_run_admission_requested",
+]

main.py

@@ -20,6 +20,10 @@
     is_live_trading_enabled,
     mask_account_id,
 )
+from application.paper_execution_admission import (
+    evaluate_paper_dry_run_admission,
+    paper_dry_run_admission_requested,
+)
 from application.rebalance_service import run_strategy_cycle
 from application.session_check_service import run_session_check
 from notifications.telegram import build_sender
@@ -384,6 +388,17 @@ def _run_strategy_cycle_with_report(
         raise
 
 
+def _evaluate_paper_dry_run_admission() -> dict[str, object] | None:
+    """Load only release identity needed for an opt-in pre-preview gate."""
+    if not paper_dry_run_admission_requested(os.environ):
+        return None
+    try:
+        runtime_target = _runtime_settings(dry_run_override=True).runtime_target
+    except (EnvironmentError, ValueError):
+        runtime_target = None
+    return evaluate_paper_dry_run_admission(runtime_target=runtime_target, env=os.environ)
+
+
 @app.get("/")
 def service_info():
     return jsonify(
@@ -542,13 +557,29 @@ def dry_run():
     if skip_for_market and skip_payload is not None:
         return jsonify(skip_payload), 200
     try:
-        return jsonify(
-            _run_strategy_cycle_with_report(
-                dry_run_override=True,
-                send_cycle_notification=False,
-                dispatch_plugin_alerts=False,
+        admission_audit = _evaluate_paper_dry_run_admission()
+        if admission_audit is not None and admission_audit["status"] != "admitted":
+            return (
+                jsonify(
+                    {
+                        "ok": False,
+                        "status": "blocked",
+                        "action_done": False,
+                        "submitted_orders": [],
+                        "skipped_orders": [{"reason": "paper_execution_admission_blocked"}],
+                        "paper_execution_admission": admission_audit,
+                    }
+                ),
+                409,
             )
+        result = _run_strategy_cycle_with_report(
+            dry_run_override=True,
+            send_cycle_notification=False,
+            dispatch_plugin_alerts=False,
         )
+        if admission_audit is not None:
+            result = {**result, "paper_execution_admission": admission_audit}
+        return jsonify(result)
     except (FirstradePlatformError, EnvironmentError, ValueError) as exc:
         notification_attempted = _handle_strategy_run_exception(exc)
         return (

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "pytest",
     "pytz",
     "requests",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@54d4ba901ae4e72e09c143c051747b900de55022",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@9f7e7f8335e97f83f66677ad5e73254a1a421759",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@cec2a6a7aac02bd06ff9c83703b14c61166b245a",
 ]
 license = "MIT"
@@ -82,5 +82,5 @@ show_missing = true
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@54d4ba901ae4e72e09c143c051747b900de55022",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@9f7e7f8335e97f83f66677ad5e73254a1a421759",
 ]

tests/test_paper_execution_admission.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import json
+
+from application.paper_execution_admission import (
+    PAPER_ADMISSION_ENABLED_ENV,
+    PAPER_EXECUTION_COMMAND_ENV,
+    evaluate_paper_dry_run_admission,
+)
+from quant_platform_kit.common.execution_commands import ExecutionCommand
+from quant_platform_kit.common.paper_execution_admission import build_paper_risk_admission_receipt
+from quant_platform_kit.common.runtime_target import build_runtime_target
+
+
+def _release() -> dict[str, str]:
+    return {
+        "release_id": "soxl-p2-v3.20260825",
+        "manifest_sha256": "a" * 64,
+        "strategy_revision": "soxl-p2-v3",
+        "config_sha256": "b" * 64,
+        "risk_policy_sha256": "c" * 64,
+        "evidence_sha256": "d" * 64,
+        "plugin_bundle_sha256": "e" * 64,
+        "effective_session": "2026-08-25",
+    }
+
+
+def _runtime_target():
+    return build_runtime_target(
+        platform_id="firstrade",
+        strategy_profile="soxl_soxx_trend_income",
+        dry_run_only=True,
+        strategy_release=_release(),
+    )
+
+
+def _command(
+    *,
+    platform: str = "firstrade",
+    include_receipt: bool = True,
+    disposition: str = "allow_new_risk",
+    reason_codes: tuple[str, ...] = (),
+    command_decision_digest: str = "f" * 64,
+) -> dict[str, object]:
+    release = _release()
+    intent: dict[str, object] = {"strategy_release": release, "targets": {"SOXL": 0.2}}
+    if include_receipt:
+        receipt = build_paper_risk_admission_receipt(
+            strategy_profile="soxl_soxx_trend_income",
+            release_id=release["release_id"],
+            risk_policy_sha256=release["risk_policy_sha256"],
+            decision_digest="f" * 64,
+            effective_session="2026-08-25",
+            disposition=disposition,
+            reason_codes=reason_codes,
+        )
+        intent["paper_risk_admission_receipt"] = receipt.to_dict()
+    return ExecutionCommand.from_decision(
+        platform=platform,
+        account_scope="paper",
+        strategy_profile="soxl_soxx_trend_income",
+        execution_mode="paper",
+        signal_date="2026-08-24",
+        effective_date="2026-08-25",
+        execution_timing_contract="next_trading_day",
+        decision_digest=command_decision_digest,
+        intent=intent,
+        created_at="2026-08-24T20:00:00+00:00",
+    ).to_dict()
+
+
+def test_paper_admission_is_disabled_by_default_even_without_a_command():
+    assert evaluate_paper_dry_run_admission(runtime_target=None, env={}) is None
+
+
+def test_paper_admission_requires_an_immutable_command_before_preview():
+    audit = evaluate_paper_dry_run_admission(
+        runtime_target=_runtime_target(),
+        env={PAPER_ADMISSION_ENABLED_ENV: "true"},
+    )
+
+    assert audit == {
+        "schema_version": "firstrade_paper_execution_admission_audit.v1",
+        "admission_enabled": True,
+        "audit_color": "red",
+        "status": "blocked",
+        "command_id": None,
+        "disposition": "halted",
+        "integrity_findings": ["paper_execution_command_missing"],
+        "receipt_sha256": None,
+    }
+
+
+def test_matching_command_receipt_and_runtime_release_admit_paper_preview():
+    audit = evaluate_paper_dry_run_admission(
+        runtime_target=_runtime_target(),
+        env={
+            PAPER_ADMISSION_ENABLED_ENV: "true",
+            PAPER_EXECUTION_COMMAND_ENV: json.dumps(_command()),
+        },
+    )
+
+    assert audit is not None
+    assert audit["status"] == "admitted"
+    assert audit["audit_color"] == "green"
+    assert audit["disposition"] == "allow_new_risk"
+    assert audit["integrity_findings"] == []
+
+
+def test_non_firstrade_or_missing_embedded_receipt_stays_redacted_and_blocked():
+    for command, finding in (
+        (_command(platform="longbridge"), "paper_execution_platform_mismatch"),
+        (_command(include_receipt=False), "paper_risk_admission_receipt_missing"),
+    ):
+        audit = evaluate_paper_dry_run_admission(
+            runtime_target=_runtime_target(),
+            env={
+                PAPER_ADMISSION_ENABLED_ENV: "true",
+                PAPER_EXECUTION_COMMAND_ENV: json.dumps(command),
+            },
+        )
+
+        assert audit is not None
+        assert audit["status"] == "blocked"
+        assert audit["audit_color"] == "red"
+        assert audit["integrity_findings"] == [finding]
+        assert "targets" not in audit
+        assert "account_scope" not in audit
+
+
+def test_reducing_only_receipt_is_blocked_until_a_platform_can_prove_reduction():
+    audit = evaluate_paper_dry_run_admission(
+        runtime_target=_runtime_target(),
+        env={
+            PAPER_ADMISSION_ENABLED_ENV: "true",
+            PAPER_EXECUTION_COMMAND_ENV: json.dumps(
+                _command(
+                    disposition="reducing_only",
+                    reason_codes=("DAILY_LOSS_LIMIT_EXCEEDED",),
+                )
+            ),
+        },
+    )
+
+    assert audit is not None
+    assert audit["status"] == "blocked"
+    assert audit["audit_color"] == "red"
+    assert audit["disposition"] == "reducing_only"
+    assert audit["integrity_findings"] == ["paper_risk_admission_reducing_only"]
+
+
+def test_mismatched_risk_decision_digest_is_red_and_never_admitted():
+    audit = evaluate_paper_dry_run_admission(
+        runtime_target=_runtime_target(),
+        env={
+            PAPER_ADMISSION_ENABLED_ENV: "true",
+            PAPER_EXECUTION_COMMAND_ENV: json.dumps(_command(command_decision_digest="e" * 64)),
+        },
+    )
+
+    assert audit is not None
+    assert audit["status"] == "blocked"
+    assert audit["audit_color"] == "red"
+    assert audit["disposition"] == "halted"
+    assert audit["integrity_findings"] == ["paper_risk_admission_command_mismatch"]