feat: add isolated Firstrade paper command consumer

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

Author: Pigbibi

.env.example

@@ -16,6 +16,12 @@ FIRSTRADE_ACCOUNT=
 # Shared US equity strategy runtime.
 STRATEGY_PROFILE=
 FIRSTRADE_DRY_RUN_ONLY=true
+# Default-disabled verifier for durable paper commands. It requires
+# RUNTIME_TARGET_ENABLED=false, FIRSTRADE_DRY_RUN_ONLY=true,
+# RUNTIME_TARGET_JSON.execution_mode=paper, CASH_ONLY_EXECUTION=true, and an
+# explicit FIRSTRADE_ACCOUNT. It never calls the order API.
+FIRSTRADE_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED=false
+FIRSTRADE_EXECUTION_COMMAND_CLOUD_URI=
 FIRSTRADE_RUNTIME_EXECUTION_WINDOW_TRADING_DAYS=
 ACCOUNT_PREFIX=FIRSTRADE
 ACCOUNT_REGION=US

README.md

@@ -54,7 +54,7 @@ uv run --no-sync python scripts/check_qpk_pin_consistency.py
 
 ## Useful docs
 
-- No separate `docs/` directory yet; start with this README and the workflow files.
+- [Isolated paper command consumer](docs/paper_execution_command_consumer.md)
 
 ## Community and security
 

application/paper_execution_command_consumer.py

@@ -0,0 +1,215 @@
+"""Firstrade read-only reconciliation adapter for shared paper commands.
+
+This module deliberately has no order-request, execution-port, or order-client
+import. It consumes only normalized current portfolio and quote snapshots from
+the isolated endpoint after the shared command binding passes.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Callable, Mapping, Sequence
+from datetime import date
+from typing import Any
+
+from quant_platform_kit.common.execution_commands import ExecutionCommand, ExecutionCommandStore
+from quant_platform_kit.common.paper_execution_command_consumer import (
+    PaperExecutionProposal,
+    PaperExecutionReconciliation,
+    consume_due_paper_execution_commands as consume_shared_paper_execution_commands,
+)
+from quant_platform_kit.common.runtime_command_gate import RuntimeCommandExposureEffect
+from quant_platform_kit.common.strategy_release import StrategyReleaseIdentity
+
+
+FIRSTRADE_PAPER_EXECUTION_INTENT_SCHEMA_VERSION = "firstrade.paper-execution-intent.v1"
+_NOTIONAL_TOLERANCE = 0.01
+
+
+def resolve_paper_execution_command_consumer_enabled(*, env_reader, dry_run_only: bool) -> bool:
+    """Enable only the isolated local-dry-run command consumer."""
+
+    enabled = str(
+        env_reader("FIRSTRADE_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED", "") or ""
+    ).strip().lower() in {"1", "true", "t", "yes", "y", "on"}
+    if enabled and not dry_run_only:
+        raise RuntimeError("Firstrade paper command consumer requires FIRSTRADE_DRY_RUN_ONLY=true")
+    return enabled
+
+
+def _symbol(value: object) -> str:
+    return str(value or "").strip().upper()
+
+
+def _symbols(value: object) -> set[str]:
+    if not isinstance(value, (list, tuple, set)):
+        return set()
+    return {_symbol(item) for item in value if _symbol(item)}
+
+
+def _finite(value: object, *, field_name: str) -> float:
+    try:
+        number = float(value)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{field_name} must be numeric") from exc
+    if not math.isfinite(number):
+        raise ValueError(f"{field_name} must be finite")
+    return number
+
+
+def _cash_balance(portfolio: Any) -> float:
+    metadata = getattr(portfolio, "metadata", {})
+    if not isinstance(metadata, Mapping):
+        raise ValueError("portfolio metadata is unavailable")
+    return _finite(metadata.get("market_currency_cash"), field_name="portfolio.market_currency_cash")
+
+
+def _reconcile(
+    command: ExecutionCommand,
+    *,
+    portfolio: Any,
+    quote_loader: Callable[[str], Any],
+    managed_symbols: Sequence[str],
+) -> PaperExecutionReconciliation:
+    intent = command.intent
+    if str(intent.get("schema_version") or "") != FIRSTRADE_PAPER_EXECUTION_INTENT_SCHEMA_VERSION:
+        return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",))
+    if str(intent.get("target_mode") or "") != "value":
+        return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",))
+    raw_targets = intent.get("targets")
+    if not isinstance(raw_targets, Mapping):
+        return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",))
+    try:
+        targets = {
+            _symbol(symbol): _finite(value, field_name=f"targets[{symbol!r}]")
+            for symbol, value in raw_targets.items()
+            if _symbol(symbol)
+        }
+    except ValueError:
+        return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",))
+    strategy_symbols = _symbols(intent.get("strategy_symbols"))
+    expected_symbols = {_symbol(symbol) for symbol in managed_symbols if _symbol(symbol)}
+    if (
+        not strategy_symbols
+        or strategy_symbols != expected_symbols
+        or set(targets) != strategy_symbols
+        or any(value < 0.0 for value in targets.values())
+    ):
+        return PaperExecutionReconciliation(proposals=(), integrity_findings=("data_artifact_invalid",))
+
+    findings: list[str] = []
+    quantities: dict[str, float] = {}
+    current_values: dict[str, float] = {}
+    for position in tuple(getattr(portfolio, "positions", ()) or ()):
+        symbol = _symbol(getattr(position, "symbol", ""))
+        if symbol not in strategy_symbols:
+            findings.append("position_reconciliation_mismatch")
+            continue
+        try:
+            quantity = _finite(getattr(position, "quantity", None), field_name=f"position[{symbol}].quantity")
+            recorded_value = _finite(
+                getattr(position, "market_value", None),
+                field_name=f"position[{symbol}].market_value",
+            )
+            quote = quote_loader(symbol)
+            price = _finite(getattr(quote, "last_price", None), field_name=f"quote[{symbol}].last_price")
+            if price <= 0.0:
+                raise ValueError("quote price must be positive")
+        except Exception:
+            findings.append("position_reconciliation_mismatch")
+            continue
+        quote_value = quantity * price
+        tolerance = max(1.0, abs(quote_value) * 0.005)
+        if quantity < -_NOTIONAL_TOLERANCE or recorded_value < -_NOTIONAL_TOLERANCE:
+            findings.append("position_reconciliation_mismatch")
+        if abs(recorded_value - quote_value) > tolerance:
+            findings.append("position_reconciliation_mismatch")
+        quantities[symbol] = quantities.get(symbol, 0.0) + quantity
+        current_values[symbol] = current_values.get(symbol, 0.0) + quote_value
+
+    try:
+        cash_balance = _cash_balance(portfolio)
+        total_equity = _finite(getattr(portfolio, "total_equity", None), field_name="portfolio.total_equity")
+        tolerance = max(1.0, abs(total_equity) * 0.005)
+        if abs(cash_balance + sum(current_values.values()) - total_equity) > tolerance:
+            findings.append("position_reconciliation_mismatch")
+    except ValueError:
+        findings.append("position_reconciliation_mismatch")
+
+    proposals: list[PaperExecutionProposal] = []
+    for symbol in sorted(strategy_symbols):
+        current_value = current_values.get(symbol, 0.0)
+        target_value = targets[symbol]
+        delta_value = target_value - current_value
+        if abs(delta_value) <= _NOTIONAL_TOLERANCE:
+            continue
+        try:
+            quote = quote_loader(symbol)
+            price = _finite(getattr(quote, "last_price", None), field_name=f"quote[{symbol}].last_price")
+            if price <= 0.0:
+                raise ValueError("quote price must be positive")
+        except Exception:
+            findings.append("position_reconciliation_mismatch")
+            continue
+        if abs(target_value) < abs(current_value) - _NOTIONAL_TOLERANCE:
+            effect = RuntimeCommandExposureEffect.REDUCES
+        elif abs(target_value) > abs(current_value) + _NOTIONAL_TOLERANCE:
+            effect = RuntimeCommandExposureEffect.INCREASES
+        else:
+            effect = RuntimeCommandExposureEffect.NEUTRAL
+        proposals.append(
+            PaperExecutionProposal(
+                symbol=symbol,
+                exposure_effect=effect,
+                details={
+                    "side": "buy" if delta_value > 0.0 else "sell",
+                    "quantity": round(abs(delta_value) / price, 8),
+                    "reference_price": round(price, 8),
+                    "current_value": round(current_value, 8),
+                    "target_value": round(target_value, 8),
+                    "target_notional_delta": round(delta_value, 8),
+                    "current_quantity": round(quantities.get(symbol, 0.0), 8),
+                },
+            )
+        )
+    return PaperExecutionReconciliation(
+        proposals=tuple(proposals),
+        integrity_findings=tuple(dict.fromkeys(findings)),
+    )
+
+
+def consume_due_paper_execution_commands(
+    *,
+    store: ExecutionCommandStore | None,
+    as_of_session: date | str,
+    claimant: str,
+    portfolio_loader: Callable[[], Any],
+    quote_loader: Callable[[str], Any],
+    managed_symbols: Sequence[str],
+    runtime_release_receipt: Mapping[str, Any] | None,
+    expected_strategy_release: StrategyReleaseIdentity | Mapping[str, object] | None,
+    expected_command_binding: Mapping[str, object] | None,
+) -> dict[str, object]:
+    """Consume paper commands after shared release and delivery binding checks."""
+
+    return consume_shared_paper_execution_commands(
+        store=store,
+        as_of_session=as_of_session,
+        claimant=claimant,
+        reconcile_command=lambda command: _reconcile(
+            command,
+            portfolio=portfolio_loader(),
+            quote_loader=quote_loader,
+            managed_symbols=managed_symbols,
+        ),
+        runtime_release_receipt=runtime_release_receipt,
+        expected_strategy_release=expected_strategy_release,
+        expected_command_binding=expected_command_binding,
+    )
+
+
+__all__ = (
+    "FIRSTRADE_PAPER_EXECUTION_INTENT_SCHEMA_VERSION",
+    "consume_due_paper_execution_commands",
+    "resolve_paper_execution_command_consumer_enabled",
+)

application/runtime_broker_adapters.py

@@ -327,6 +327,66 @@ def build_portfolio_snapshot(self) -> PortfolioSnapshot:
             },
         )
 
+    def build_reconciled_paper_portfolio_snapshot(self) -> PortfolioSnapshot:
+        """Read complete, current account evidence for delayed paper commands.
+
+        The normal strategy snapshot intentionally filters to its managed
+        symbols.  A delayed-command consumer must instead see every position:
+        an unmanaged or malformed holding is a reason to reject the command,
+        not a reason to silently omit it.  This method is read-only and lives
+        beside the normal snapshot so no ordinary execution behavior changes.
+        """
+
+        balances = self.client.get_balances(self.account)
+        positions_payload = self.client.get_positions(self.account)
+        positions: list[Position] = []
+        for row in iter_position_rows(positions_payload):
+            raw_symbol = get_first(row, "symbol", "ticker", "security_symbol")
+            if not raw_symbol:
+                raise ValueError("Firstrade reconciliation received a position without a symbol.")
+            symbol = self.normalize_symbol(raw_symbol)
+            quantity = float_or_none(get_first(row, "quantity", "shares", "qty"))
+            market_value = float_or_none(
+                get_first(row, "market_value", "marketValue", "value", "current_value")
+            )
+            if quantity is None or market_value is None:
+                raise ValueError(
+                    f"Firstrade reconciliation requires quantity and current market value for {symbol}."
+                )
+            if quantity == 0.0:
+                continue
+            positions.append(
+                Position(
+                    symbol=symbol,
+                    quantity=quantity,
+                    market_value=market_value,
+                    average_cost=float_or_none(
+                        get_first(row, "average_cost", "avg_cost", "cost_basis", "averagePrice")
+                    ),
+                    currency="USD",
+                    account_id=mask_account_id(self.account),
+                )
+            )
+        cash_balance = _first_numeric_by_keyword_groups(balances, _CASH_BALANCE_KEYWORD_GROUPS)
+        if cash_balance is None:
+            raise ValueError("Firstrade reconciliation requires a current cash balance.")
+        total_equity = float(cash_balance) + sum(float(position.market_value) for position in positions)
+        return PortfolioSnapshot(
+            as_of=self.clock(),
+            total_equity=total_equity,
+            cash_balance=float(cash_balance),
+            buying_power=float(cash_balance),
+            positions=tuple(positions),
+            metadata={
+                "broker": "firstrade",
+                "account_hash": self.account_hash or mask_account_id(self.account),
+                "api_kind": "unofficial-reverse-engineered",
+                "cash_only_execution": True,
+                "market_currency_cash": float(cash_balance),
+                "reconciliation_source": "firstrade_current_balances_and_positions",
+            },
+        )
+
     def build_portfolio_port(self) -> PortfolioPort:
         return CallablePortfolioPort(self.build_portfolio_snapshot)
 

docs/paper_execution_command_consumer.md

@@ -0,0 +1,38 @@
+# Firstrade isolated paper command consumer
+
+`POST /paper-command-consumer` manually verifies delayed paper commands. It is
+not called by `/run`, `/dry-run`, scheduler workflows, or the normal strategy
+cycle, and it never constructs an execution port or order request.
+
+The endpoint uses the shared QuantPlatformKit paper lifecycle: approved release
+receipt, exact platform/account-scope/strategy-profile binding, create-only
+claim and events, paper-risk receipt, and an enforced command gate. Only after
+those checks pass does it open a read-only Firstrade session for current
+balances, all positions, and quotes.
+
+## Required isolation
+
+- `FIRSTRADE_PAPER_EXECUTION_COMMAND_CONSUMER_ENABLED=true`
+- `RUNTIME_TARGET_ENABLED=false`
+- `FIRSTRADE_DRY_RUN_ONLY=true`
+- `RUNTIME_TARGET_JSON.execution_mode=paper`
+- `CASH_ONLY_EXECUTION=true`
+- explicit `FIRSTRADE_ACCOUNT`
+- `FIRSTRADE_EXECUTION_COMMAND_CLOUD_URI` or
+  `FIRSTRADE_EXECUTION_COMMAND_DIR`
+
+The exact account identifier is required for this endpoint even if Firstrade
+currently returns one account; this prevents a newly added account from being
+selected implicitly. The consumer binds logical delivery using the runtime
+target's `account_scope`, never command-provided metadata.
+
+## Fail-closed reconciliation
+
+The account read includes all positions rather than only the strategy's managed
+symbols. Missing cash/current market values, an unmanaged position, a short,
+a stale or invalid quote, a mismatch between cash-plus-positions and equity, or
+a release/binding mismatch records a blocked or rejected paper event. The
+consumer never guesses a quantity, adjusts leverage, or submits a broker order.
+
+Turn the flag off again after a manual verification. Live rollout remains a
+separate, reviewed release-readiness decision.