diff --git a/.env.example b/.env.example index 5626b32..a7d5abc 100644 --- a/.env.example +++ b/.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 diff --git a/README.md b/README.md index 2b47de0..52c1b0b 100644 --- a/README.md +++ b/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 diff --git a/application/paper_execution_command_consumer.py b/application/paper_execution_command_consumer.py new file mode 100644 index 0000000..65411d9 --- /dev/null +++ b/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", +) diff --git a/application/runtime_broker_adapters.py b/application/runtime_broker_adapters.py index e77ce84..8e903e1 100644 --- a/application/runtime_broker_adapters.py +++ b/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) diff --git a/docs/paper_execution_command_consumer.md b/docs/paper_execution_command_consumer.md new file mode 100644 index 0000000..1816793 --- /dev/null +++ b/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. diff --git a/main.py b/main.py index 1e9a514..b8342d6 100644 --- a/main.py +++ b/main.py @@ -8,11 +8,14 @@ from dataclasses import replace from datetime import datetime, timezone from typing import Any +from zoneinfo import ZoneInfo from flask import Flask, jsonify, request from quant_platform_kit.common.health import register_health_endpoint +from quant_platform_kit.common.execution_commands import build_execution_command_store_from_env from quant_platform_kit.common.platform_runner import dispatch_due_monitors, load_monitor_targets +from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt from application.firstrade_client import ( FirstradeBrokerClient, FirstradeCredentials, @@ -24,7 +27,12 @@ evaluate_paper_dry_run_admission, paper_dry_run_admission_requested, ) +from application.paper_execution_command_consumer import ( + consume_due_paper_execution_commands, + resolve_paper_execution_command_consumer_enabled, +) from application.rebalance_service import run_strategy_cycle +from application.runtime_broker_adapters import build_runtime_broker_adapters from application.session_check_service import run_session_check from notifications.telegram import build_sender from quant_platform_kit.common.runtime_reports import ( @@ -40,6 +48,7 @@ load_platform_runtime_settings, ) from strategy_registry import get_platform_profile_status_matrix +from strategy_runtime import load_strategy_runtime MARKET_CALENDAR = os.getenv("FIRSTRADE_MARKET_CALENDAR", "NYSE") MARKET_TIMEZONE = os.getenv("FIRSTRADE_MARKET_TIMEZONE", "America/New_York") @@ -399,6 +408,141 @@ def _evaluate_paper_dry_run_admission() -> dict[str, object] | None: return evaluate_paper_dry_run_admission(runtime_target=runtime_target, env=os.environ) +def _paper_command_consumer_session_date() -> str: + return datetime.now(ZoneInfo(MARKET_TIMEZONE)).date().isoformat() + + +def _paper_command_consumer_runtime_is_isolated(settings: PlatformRuntimeSettings) -> bool: + """Require an explicitly disabled, cash-only paper runtime.""" + + runtime_target = settings.runtime_target + return bool( + runtime_target is not None + and settings.dry_run_only + and not settings.runtime_target_enabled + and settings.cash_only_execution + and str(getattr(runtime_target, "execution_mode", "") or "").strip().lower() == "paper" + ) + + +def run_paper_execution_command_consumer() -> dict[str, object]: + """Verify due paper commands through read-only Firstrade account evidence. + + This function is intentionally outside ``run_strategy_cycle``. It creates + neither an execution port nor a stock-order request, and opens the + Firstrade session only after the shared consumer accepts the release and + exact platform/account/strategy delivery binding. + """ + + settings = _runtime_settings() + if not _paper_command_consumer_runtime_is_isolated(settings): + raise RuntimeError( + "paper command consumer requires RUNTIME_TARGET_ENABLED=false, " + "FIRSTRADE_DRY_RUN_ONLY=true, a paper runtime target, and cash-only execution" + ) + if not resolve_paper_execution_command_consumer_enabled( + env_reader=os.getenv, + dry_run_only=settings.dry_run_only, + ): + raise RuntimeError("paper command consumer is not enabled") + requested_account = str(os.getenv("FIRSTRADE_ACCOUNT") or "").strip() + if not requested_account: + raise RuntimeError("paper command consumer requires an explicit FIRSTRADE_ACCOUNT") + + runtime_target = settings.runtime_target + expected_release = getattr(runtime_target, "strategy_release", None) + expected_binding = { + "platform": "firstrade", + "account_scope": str(getattr(runtime_target, "account_scope", "") or "unknown"), + "strategy_profile": str(getattr(runtime_target, "strategy_profile", "") or "unknown"), + } + store = build_execution_command_store_from_env( + platform_env_prefix="FIRSTRADE", + env_reader=os.getenv, + project_id=settings.project_id or get_project_id(), + ) + if not store.cloud_prefix_uri and not store.local_dir: + raise RuntimeError("Firstrade paper command consumer requires an execution command store") + strategy_runtime = load_strategy_runtime( + settings.strategy_profile, + runtime_settings=settings, + logger=lambda message: print(message, flush=True), + ) + managed_symbols = tuple(strategy_runtime.managed_symbols) + if not managed_symbols: + raise RuntimeError("Firstrade paper command consumer requires configured managed symbols") + + report = _build_runtime_report(settings, dry_run=True) + broker_adapters = None + market_data_port = None + + def _broker_adapters(): + nonlocal broker_adapters + if broker_adapters is None: + credentials = FirstradeCredentials.from_env() + client = FirstradeBrokerClient( + credentials, + live_trading_enabled=False, + ).connect() + account = client.select_account(requested_account) + broker_adapters = build_runtime_broker_adapters( + client=client, + account=account, + strategy_symbols=managed_symbols, + account_hash=mask_account_id(account), + live_orders=False, + live_order_ack=False, + cash_only_execution=True, + ) + return broker_adapters + + def _load_portfolio(): + return _broker_adapters().build_reconciled_paper_portfolio_snapshot() + + def _load_quote(symbol: str): + nonlocal market_data_port + if market_data_port is None: + market_data_port = _broker_adapters().build_market_data_port() + return market_data_port.get_quote(symbol) + + try: + result = consume_due_paper_execution_commands( + store=store, + as_of_session=_paper_command_consumer_session_date(), + claimant=_service_name(), + portfolio_loader=_load_portfolio, + quote_loader=_load_quote, + managed_symbols=managed_symbols, + runtime_release_receipt=build_runtime_loaded_receipt( + strategy_release=expected_release, + ), + expected_strategy_release=expected_release, + expected_command_binding=expected_binding, + ) + finalize_runtime_report( + report, + status="ok" if result.get("status") == "ok" else "skipped", + summary={"paper_execution_command_consumer": result}, + ) + return result + except Exception as exc: + append_runtime_report_error( + report, + stage="paper_execution_command_consumer", + message=_safe_exception_text(exc), + error_type=type(exc).__name__, + ) + finalize_runtime_report(report, status="error") + raise + finally: + try: + report_path = _persist_runtime_report(report) + if report_path: + print(f"execution_report {report_path}", flush=True) + except Exception as persist_exc: + print(f"failed to persist execution report: {persist_exc}", flush=True) + + @app.get("/") def service_info(): return jsonify( @@ -606,6 +750,38 @@ def dry_run(): ) +@app.post("/paper-command-consumer") +def paper_execution_command_consumer(): + """Manual-only endpoint for isolated paper command reconciliation.""" + + try: + return jsonify(run_paper_execution_command_consumer()) + except (FirstradePlatformError, EnvironmentError, ValueError) as exc: + notification_attempted = _handle_strategy_run_exception(exc) + return ( + jsonify( + { + "ok": False, + "error": _safe_exception_text(exc), + "runtime_error_notification_attempted": notification_attempted, + } + ), + 500, + ) + except Exception as exc: + notification_attempted = _handle_strategy_run_exception(exc) + return ( + jsonify( + { + "ok": False, + "error": _safe_exception_text(exc, include_type=True), + "runtime_error_notification_attempted": notification_attempted, + } + ), + 500, + ) + + @app.post("/probe") @app.get("/probe") def probe(): diff --git a/tests/test_paper_execution_command_consumer.py b/tests/test_paper_execution_command_consumer.py new file mode 100644 index 0000000..e97b417 --- /dev/null +++ b/tests/test_paper_execution_command_consumer.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from application.paper_execution_command_consumer import ( + FIRSTRADE_PAPER_EXECUTION_INTENT_SCHEMA_VERSION, + consume_due_paper_execution_commands, + resolve_paper_execution_command_consumer_enabled, +) +from quant_platform_kit.common.execution_commands import ExecutionCommand, ExecutionCommandState, ExecutionCommandStore +from quant_platform_kit.common.models import PortfolioSnapshot, Position, QuoteSnapshot +from quant_platform_kit.common.paper_execution_admission import ( + PAPER_RISK_ADMISSION_RECEIPT_INTENT_FIELD, + build_paper_risk_admission_receipt, +) +from quant_platform_kit.common.strategy_release import build_runtime_loaded_receipt + + +def _release() -> dict[str, str]: + return { + "release_id": "tqqq-p3-v6.20260824", + "manifest_sha256": "a" * 64, + "strategy_revision": "tqqq-p3-v6", + "config_sha256": "b" * 64, + "risk_policy_sha256": "c" * 64, + "evidence_sha256": "d" * 64, + "plugin_bundle_sha256": "e" * 64, + "effective_session": "2026-08-25", + } + + +def _command(*, platform: str = "firstrade") -> ExecutionCommand: + release = _release() + intent = { + "schema_version": FIRSTRADE_PAPER_EXECUTION_INTENT_SCHEMA_VERSION, + "target_mode": "value", + "targets": {"TQQQ": 300.0, "BOXX": 100.0}, + "strategy_symbols": ["TQQQ", "BOXX"], + "strategy_release": release, + } + decision_digest = hashlib.sha256( + json.dumps(intent, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8") + ).hexdigest() + intent[PAPER_RISK_ADMISSION_RECEIPT_INTENT_FIELD] = build_paper_risk_admission_receipt( + strategy_profile="tqqq_growth_income", + release_id=release["release_id"], + risk_policy_sha256=release["risk_policy_sha256"], + decision_digest=decision_digest, + effective_session="2026-08-25", + disposition="allow_new_risk", + reason_codes=(), + ).to_dict() + return ExecutionCommand.from_decision( + platform=platform, + account_scope="us", + strategy_profile="tqqq_growth_income", + execution_mode="paper", + signal_date="2026-08-24", + effective_date="2026-08-25", + execution_timing_contract="next_trading_day", + decision_digest=decision_digest, + intent=intent, + ) + + +def _portfolio() -> PortfolioSnapshot: + return PortfolioSnapshot( + as_of=datetime(2026, 8, 25, tzinfo=timezone.utc), + total_equity=1_000.0, + cash_balance=800.0, + buying_power=800.0, + positions=( + Position( + symbol="TQQQ", + quantity=20.0, + market_value=200.0, + average_cost=8.0, + currency="USD", + ), + ), + metadata={"market_currency_cash": 800.0}, + ) + + +def _quote(symbol: str) -> QuoteSnapshot: + return QuoteSnapshot( + symbol=symbol, + as_of=datetime(2026, 8, 25, tzinfo=timezone.utc), + last_price=10.0, + ) + + +def _binding() -> dict[str, str]: + return { + "platform": "firstrade", + "account_scope": "us", + "strategy_profile": "tqqq_growth_income", + } + + +def test_consumer_fills_reconciled_paper_command_without_order_client(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command() + assert store.enqueue(command) + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="firstrade-paper-command-consumer", + portfolio_loader=_portfolio, + quote_loader=_quote, + managed_symbols=("TQQQ", "BOXX"), + runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=_release()), + expected_strategy_release=_release(), + expected_command_binding=_binding(), + ) + + assert result["status"] == "ok" + assert result["commands"][0]["status"] == "filled" + assert store.current_state(command) is ExecutionCommandState.FILLED + proposals = store.events(command)[1].details["proposals"] + assert [proposal["exposure_effect"] for proposal in proposals] == ["increases", "increases"] + assert all("order" not in proposal["details"] for proposal in proposals) + + +def test_consumer_rejects_cross_platform_command_before_account_reads(tmp_path: Path) -> None: + store = ExecutionCommandStore(local_dir=tmp_path) + command = _command(platform="schwab") + assert store.enqueue(command) + reads = {"portfolio": 0, "quote": 0} + + def portfolio_loader(): + reads["portfolio"] += 1 + return _portfolio() + + def quote_loader(symbol: str): + reads["quote"] += 1 + return _quote(symbol) + + result = consume_due_paper_execution_commands( + store=store, + as_of_session="2026-08-25", + claimant="firstrade-paper-command-consumer", + portfolio_loader=portfolio_loader, + quote_loader=quote_loader, + managed_symbols=("TQQQ", "BOXX"), + runtime_release_receipt=build_runtime_loaded_receipt(strategy_release=_release()), + expected_strategy_release=_release(), + expected_command_binding=_binding(), + ) + + assert result["commands"][0]["status"] == "rejected" + assert reads == {"portfolio": 0, "quote": 0} + assert store.current_state(command) is ExecutionCommandState.REJECTED + + +def test_consumer_flag_cannot_be_enabled_outside_dry_run() -> None: + with pytest.raises(RuntimeError, match="FIRSTRADE_DRY_RUN_ONLY=true"): + resolve_paper_execution_command_consumer_enabled( + env_reader=lambda *_args: "true", + dry_run_only=False, + ) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 05035d0..d5412e3 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -30,6 +30,7 @@ def test_cloud_run_route_contracts_are_registered(): "/smoke": ["GET"], "/run": ["GET", "POST"], "/dry-run": ["GET", "POST"], + "/paper-command-consumer": ["POST"], "/monitor-dispatch": ["GET", "POST"], "/probe": ["GET", "POST"], "/static/": ["GET"], diff --git a/tests/test_runtime_broker_adapters.py b/tests/test_runtime_broker_adapters.py index 7c2ac0e..46450a3 100644 --- a/tests/test_runtime_broker_adapters.py +++ b/tests/test_runtime_broker_adapters.py @@ -2,6 +2,8 @@ from datetime import datetime, timezone +import pytest + from application.runtime_broker_adapters import build_runtime_broker_adapters @@ -98,6 +100,46 @@ def get_positions(self, _account): assert portfolio.metadata["total_equity_source"] == "cash_plus_positions" +def test_reconciled_paper_snapshot_keeps_unmanaged_positions_for_fail_closed_review(): + class MultiPositionClient(FakeClient): + def get_balances(self, _account): + return {"cash_balance": "20.00"} + + def get_positions(self, _account): + return { + "items": [ + {"symbol": "SPY", "quantity": "2", "market_value": "21.00"}, + {"symbol": "AAPL", "quantity": "3", "market_value": "300.00"}, + ] + } + + adapters = build_runtime_broker_adapters( + client=MultiPositionClient(), + account="12345678", + strategy_symbols=("SPY",), + ) + + snapshot = adapters.build_reconciled_paper_portfolio_snapshot() + + assert [position.symbol for position in snapshot.positions] == ["SPY", "AAPL"] + assert snapshot.total_equity == 341.0 + assert snapshot.metadata["reconciliation_source"] == "firstrade_current_balances_and_positions" + + +def test_reconciled_paper_snapshot_requires_current_market_value(): + class IncompletePositionClient(FakeClient): + def get_positions(self, _account): + return {"items": [{"symbol": "SPY", "quantity": "2"}]} + + adapters = build_runtime_broker_adapters( + client=IncompletePositionClient(), + account="12345678", + ) + + with pytest.raises(ValueError, match="current market value"): + adapters.build_reconciled_paper_portfolio_snapshot() + + def test_portfolio_snapshot_falls_back_to_cash_when_total_value_missing(): class CashOnlyClient(FakeClient): def get_balances(self, _account):