Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .env.example
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
215 changes: 215 additions & 0 deletions application/paper_execution_command_consumer.py
Original file line number Diff line number Diff line change
@@ -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",
)
60 changes: 60 additions & 0 deletions application/runtime_broker_adapters.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
38 changes: 38 additions & 0 deletions docs/paper_execution_command_consumer.md
Original file line number Diff line number Diff line change
@@ -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.
Loading