diff --git a/application/account_snapshot.py b/application/account_snapshot.py index dc3cf58..a035e82 100644 --- a/application/account_snapshot.py +++ b/application/account_snapshot.py @@ -5,10 +5,15 @@ from collections.abc import Mapping from contextvars import ContextVar, Token from datetime import datetime, timezone +import re from typing import Any from application.account_financing import project_native_financing -from application.broker_reconciliation import _normalize_broker_balance, _normalize_cash +from application.broker_reconciliation import ( + _normalize_broker_balance, + _normalize_cash, + _normalize_position, +) _CAPTURE: ContextVar[dict[str, Any] | None] = ContextVar( @@ -24,6 +29,9 @@ def begin_natural_cycle_history() -> Token[dict[str, Any] | None]: "projection": None, "started": None, "finished": None, + "stocks_seen": False, + "stocks": None, + "stocks_error": None, } ) @@ -64,6 +72,76 @@ def cycle_history_observation() -> dict[str, Any] | None: } +def read_cycle_stock_positions(reader): + """Read once in an active capture and archive the full native inventory. + + The probe pre-reads before strategy filtering; the portfolio later reuses + the same response (or error) without a second native call. + """ + slot = _CAPTURE.get() + if slot is None: + return reader() + if slot["stocks_seen"]: + if slot["stocks_error"] is not None: + raise slot["stocks_error"] + return slot["stocks"] + slot["stocks_seen"] = True + try: + response = reader() + except Exception as exc: + slot["stocks_error"] = exc + raise + finally: + slot["finished"] = datetime.now(timezone.utc) + slot["stocks"] = response + rows = project_cycle_stock_positions(response) + if rows is not None and isinstance(slot["projection"], dict): + slot["projection"]["broker_reported_positions"] = rows + return response + + +def project_cycle_stock_positions(response: Any) -> list[dict[str, Any]] | None: + """Missing/malformed is unknown; a successfully read empty inventory stays empty.""" + channels = getattr(response, "channels", None) + # No channel at all is not evidence of an empty book (see reconciliation). + if not isinstance(channels, (list, tuple)) or not channels: + return None + rows = [] + seen = set() + try: + for channel in channels: + positions = getattr(channel, "positions", None) + if not isinstance(positions, (list, tuple)): + return None + for position in positions: + row = _normalize_position(position, account_channel="") + symbol, currency, quantity = ( + row[key] for key in ("symbol", "currency", "quantity") + ) + whole, _, fraction = quantity.lstrip("-").partition(".") + if ( + re.fullmatch(r"[A-Z0-9][A-Z0-9./ -]{0,31}", symbol) is None + or re.fullmatch(r"[A-Z]{3}", currency) is None + or len(whole) > 15 + or len(fraction) > 8 + or symbol in seen + or len(rows) >= 64 + ): + return None + seen.add(symbol) + rows.append( + { + "symbol": symbol, + "currency": currency, + "quantity": quantity, + "market_value": None, + } + ) + except Exception: + return None + return sorted(rows, key=lambda row: (row["currency"], row["symbol"])) + + def project_cycle_history_balances(account_balance: Any) -> dict[str, list[dict[str, str]]] | None: """Keep complete, currency-specific broker facts as decimal strings.""" diff --git a/application/longbridge_portfolio.py b/application/longbridge_portfolio.py index 7c49436..5a4fc12 100644 --- a/application/longbridge_portfolio.py +++ b/application/longbridge_portfolio.py @@ -5,7 +5,10 @@ import math from typing import Any -from application.account_snapshot import read_cycle_account_balance +from application.account_snapshot import ( + read_cycle_account_balance, + read_cycle_stock_positions, +) from quant_platform_kit.common.broker_reconciliation import calculate_broker_observation_sha256 @@ -112,7 +115,7 @@ def warn(message: str) -> None: position_rows: list[tuple[str, str, Any, Any]] = [] try: - positions_response = t_ctx.stock_positions() + positions_response = read_cycle_stock_positions(t_ctx.stock_positions) except Exception as exc: warn( "[longbridge_stock_positions_failed] " diff --git a/main.py b/main.py index 6ada989..8d1bbc5 100644 --- a/main.py +++ b/main.py @@ -33,6 +33,7 @@ cycle_history_observation, end_natural_cycle_history, read_cycle_account_balance, + read_cycle_stock_positions, ) from scripts.record_daily_account_snapshot import record_projected_daily_account from application.durable_execution_commands import ( @@ -1010,6 +1011,15 @@ def run_probe(*, response_body: str = "Probe OK"): composer.build_account_snapshot_broker_contexts() ) read_cycle_account_balance(lambda: trade_context.account_balance(currency="USD")) + # Read the native stock inventory once before any strategy filtering so + # the archive keeps the complete book and the portfolio reuses it. + # A failed read leaves balances/cash known and inventory unknown. + stock_reader = getattr(trade_context, "stock_positions", None) + if callable(stock_reader): + try: + read_cycle_stock_positions(stock_reader) + except Exception: + pass observation = cycle_history_observation() if observation is None: raise RuntimeError("account history observation incomplete") @@ -1028,6 +1038,7 @@ def run_probe(*, response_body: str = "Probe OK"): finished=observation["finished"], open_store=_open_account_history_store, financing=observation["projection"].get("financing"), + positions=observation["projection"].get("broker_reported_positions"), ) print( f"account_history status={result.status} category={result.category or '-'}", diff --git a/scripts/record_daily_account_snapshot.py b/scripts/record_daily_account_snapshot.py index baccc41..ce8cb3c 100644 --- a/scripts/record_daily_account_snapshot.py +++ b/scripts/record_daily_account_snapshot.py @@ -23,6 +23,7 @@ _PROJECT_ID = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$") _BINDING_ID = re.compile(r"^[0-9a-f]{64}$") _CURRENCY = re.compile(r"^[A-Z]{3}$") +_POSITION_SYMBOL = re.compile(r"^[A-Z0-9][A-Z0-9./ -]{0,31}$") _DECIMAL = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d+)?$") _FORBIDDEN_PREFIX_PARTS = ("execution-report", "execution_report", "runtime-report") _BALANCE_FIELDS = ("currency", "net_assets", "total_cash") @@ -58,6 +59,7 @@ def record_projected_daily_account( finished: datetime, open_store: Callable[[str], Any], financing: Any = None, + positions: Any = None, ) -> DailyAccountRecordResult: """Validate one PAPER observation and confirm its create-only write.""" @@ -75,6 +77,7 @@ def record_projected_daily_account( started=started, finished=finished, financing=financing, + positions=positions, ) except _Rejected as rejected: return DailyAccountRecordResult("skipped", rejected.category) @@ -142,6 +145,7 @@ def _history_object( started: datetime, finished: datetime, financing: Any = None, + positions: Any = None, ) -> tuple[str, str]: binding_id = _binding_id(source_binding) if not isinstance(started, datetime) or started.tzinfo is None or started.utcoffset() is None: @@ -175,6 +179,12 @@ def _history_object( "broker_reported_balances": balance_rows, "cash": _money_rows(cash, _CASH_FIELDS), } + if positions is not None: + record.update( + broker_reported_positions=_position_rows(positions), + broker_reported_positions_scope="stocks_only", + broker_reported_positions_complete=True, + ) if financing is not None: record["financing"] = _financing_rows( financing, @@ -239,6 +249,47 @@ def _money_rows(value: object, fields: tuple[str, ...]) -> list[dict[str, str]]: return rows +def _position_rows(value: object) -> list[dict[str, Any]]: + """Project only the three public stock-inventory facts; never prices or extras.""" + if not isinstance(value, list) or len(value) > 64: + raise _Rejected("projection_invalid") + rows: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in value: + if not isinstance(item, Mapping): + raise _Rejected("projection_invalid") + symbol, currency, quantity = ( + item.get("symbol"), + item.get("currency"), + item.get("quantity"), + ) + if ( + not isinstance(symbol, str) + or _POSITION_SYMBOL.fullmatch(symbol) is None + or not isinstance(currency, str) + or _CURRENCY.fullmatch(currency) is None + or isinstance(quantity, bool) + or not isinstance(quantity, str) + or _DECIMAL.fullmatch(quantity) is None + ): + raise _Rejected("projection_invalid") + whole, _, fraction = quantity.lstrip("-").partition(".") + if len(whole) > 15 or len(fraction) > 8 or not Decimal(quantity).is_finite(): + raise _Rejected("projection_invalid") + if symbol in seen: + raise _Rejected("projection_invalid") + seen.add(symbol) + rows.append( + { + "symbol": symbol, + "quantity": quantity, + "currency": currency, + "market_value": None, + } + ) + return sorted(rows, key=lambda row: (row["currency"], row["symbol"])) + + def _create_bounded(store: Any, uri: str, body: str) -> bool: """Upload once with an explicit timeout, no retry, and no-overwrite guard.""" diff --git a/tests/test_paper_stock_history.py b/tests/test_paper_stock_history.py new file mode 100644 index 0000000..99b8483 --- /dev/null +++ b/tests/test_paper_stock_history.py @@ -0,0 +1,392 @@ +"""Synthetic coverage for the PAPER native stock-inventory archive port. + +Only in-memory broker doubles are used: one request-local native stock read is +reused by the portfolio, the full inventory is archived before strategy +filtering, and the producer accepts only the three public stock facts. +""" + +from __future__ import annotations + +import json +import os +from decimal import Decimal +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import pytest + +from application.account_snapshot import ( + begin_natural_cycle_history, + cycle_history_observation, + end_natural_cycle_history, + project_cycle_stock_positions, + read_cycle_account_balance, + read_cycle_stock_positions, +) +from application.longbridge_portfolio import fetch_strategy_account_state +from scripts.record_daily_account_snapshot import record_projected_daily_account + + +SOURCE = {"kind": "deployment_scope_token_version", "status": "bound", "id": "a" * 64} +PAPER_ENV = { + "ACCOUNT_HISTORY_RECORDING_ENABLED": "true", + "ACCOUNT_HISTORY_GCS_PREFIX": "gs://paper-history/longbridge/account_snapshots", + "ACCOUNT_HISTORY_TARGET_ID": "paper", + "ACCOUNT_HISTORY_EXPECTED_SCOPE": "PAPER", + "GOOGLE_CLOUD_PROJECT": "longbridgequant", +} + + +def _balance( + currency="USD", + *, + net_assets="100", + total_cash="100", + available_cash="50", + financing=None, +): + return SimpleNamespace( + currency=currency, + net_assets=Decimal(net_assets), + total_cash=Decimal(total_cash), + cash_infos=[ + SimpleNamespace( + currency=currency, + available_cash=Decimal(available_cash), + frozen_cash=Decimal("0"), + settling_cash=Decimal("0"), + ) + ], + **(financing or {}), + ) + + +def _position(symbol, currency, quantity, **extra): + return SimpleNamespace(symbol=symbol, currency=currency, quantity=quantity, **extra) + + +def _response(*positions): + return SimpleNamespace(channels=[SimpleNamespace(positions=list(positions))]) + + +class _Store: + def __init__(self): + self.calls = [] + self.client = self + + def _parse_uri(self, uri): + bucket, blob = uri.removeprefix("gs://").split("/", 1) + return bucket, blob + + def bucket(self, _bucket): + return self + + def blob(self, _name): + return self + + def upload_from_string(self, data, **_kwargs): + self.calls.append(data) + + +def _record(observation, **overrides): + store = _Store() + projection = observation["projection"] + values = { + "env": PAPER_ENV, + "account_scope": "PAPER", + "source_binding": SOURCE, + "balances": projection["broker_reported_balances"], + "cash": projection["cash"], + "started": observation["started"], + "finished": observation["finished"], + "financing": projection.get("financing"), + "positions": projection.get("broker_reported_positions"), + } + values.update(overrides) + result = record_projected_daily_account(**values, open_store=lambda _project: store) + record = json.loads(store.calls[0]) if store.calls else None + return result, record + + +def _run_probe_with_stock_inventory(): + """Exercise the real probe path with synthetic broker and store doubles.""" + import sys + from pathlib import Path + + tests_dir = str(Path(__file__).resolve().parent) + if tests_dir not in sys.path: + sys.path.insert(0, tests_dir) + from test_request_handling import load_module + + module = load_module() + module.ACCOUNT_REGION = "PAPER" + module._v7_application_candidate_bound = lambda: False + + observed = {"balance": 0, "stocks": 0, "portfolio": 0, "stored": None} + account = _balance( + financing={ + "max_finance_amount": Decimal("0.00"), + "buy_power": Decimal("12.00"), + "risk_level": 1, + } + ) + positions = [ + _position("700.HK", "HKD", Decimal("5"), available_quantity=Decimal("5")), + _position("AAPL.US", "USD", Decimal("2"), available_quantity=Decimal("2")), + ] + + class TradeContext: + def account_balance(self, **_kwargs): + observed["balance"] += 1 + return [account] + + def stock_positions(self): + observed["stocks"] += 1 + return _response(*positions) + + trade = TradeContext() + + class QuoteContext: + def quote(self, _symbols): + return [ + SimpleNamespace(symbol="AAPL.US", last_done="10.0"), + SimpleNamespace(symbol="700.HK", last_done="300.0"), + ] + + quote = QuoteContext() + + class FakePort: + def get_portfolio_snapshot(self): + observed["portfolio"] += 1 + state = fetch_strategy_account_state(quote, trade, ["AAPL"]) + return SimpleNamespace( + buying_power=state["available_cash"], + total_equity=state["total_strategy_equity"], + positions=(), + ) + + class FakeRuntime: + def portfolio_port_factory(self, *_args): + return FakePort() + + class FakeComposer: + strategy_adapters = SimpleNamespace(calculate_strategy_indicators=lambda _context: {}) + + def build_reporting_adapters(self): + return SimpleNamespace( + start_run=lambda: (None, {"status": "pending"}), + log_event=lambda *_args, **_kwargs: None, + persist_execution_report=lambda _report: None, + ) + + def build_rebalance_runtime(self, **_kwargs): + return FakeRuntime() + + def build_account_snapshot_broker_contexts(self): + return quote, trade, SOURCE + + class Store(_Store): + def upload_from_string(self, data, **_kwargs): + observed["stored"] = data + super().upload_from_string(data, **_kwargs) + + module.build_composer = lambda **_kwargs: FakeComposer() + module._open_account_history_store = lambda _project: Store() + submit = Mock() + module.submit_order = submit + + with patch.dict(os.environ, PAPER_ENV, clear=False): + result = module.run_probe() + return result, observed, submit + + +def test_probe_archives_full_inventory_before_filtering_and_never_trades(): + result, observed, submit = _run_probe_with_stock_inventory() + + assert result == ("Probe OK", 200) + assert observed["balance"] == 1 + assert observed["stocks"] == 1 + assert observed["portfolio"] == 1 + submit.assert_not_called() + + record = json.loads(observed["stored"]) + assert record["broker_reported_positions"] == [ + {"symbol": "700.HK", "quantity": "5", "currency": "HKD", "market_value": None}, + {"symbol": "AAPL.US", "quantity": "2", "currency": "USD", "market_value": None}, + ] + assert record["broker_reported_positions_scope"] == "stocks_only" + assert record["broker_reported_positions_complete"] is True + assert record["broker_reported_balances"] == [ + {"currency": "USD", "net_assets": "100", "total_cash": "100"} + ] + assert record["financing"] == [ + { + "currency": "USD", + "max_finance_amount": "0.00", + "buy_power": "12.00", + "risk_level": "1", + } + ] + + +def test_portfolio_reuses_one_native_stock_read_and_keeps_unfiltered_inventory(): + stock_calls = [] + t_ctx = SimpleNamespace( + account_balance=lambda **_: [_balance()], + stock_positions=lambda: ( + stock_calls.append(1) + or _response( + _position("AAPL.US", "USD", "2"), + _position("700.HK", "HKD", "5"), + ) + ), + ) + q_ctx = SimpleNamespace( + quote=lambda symbols: [ + SimpleNamespace(symbol=symbol, last_done="10.0") for symbol in symbols + ] + ) + token = begin_natural_cycle_history() + try: + first = fetch_strategy_account_state(q_ctx, t_ctx, ["AAPL"]) + second = fetch_strategy_account_state(q_ctx, t_ctx, ["AAPL"]) + observation = cycle_history_observation() + finally: + end_natural_cycle_history(token) + + assert len(stock_calls) == 1 + assert first["quantities"] == {"AAPL": 2.0} + assert second["quantities"] == first["quantities"] + assert observation["projection"]["broker_reported_positions"] == [ + {"symbol": "700.HK", "currency": "HKD", "quantity": "5", "market_value": None}, + {"symbol": "AAPL.US", "currency": "USD", "quantity": "2", "market_value": None}, + ] + + +def test_proven_empty_inventory_is_archived_as_empty_not_unknown(): + token = begin_natural_cycle_history() + try: + read_cycle_account_balance(lambda: [_balance()]) + read_cycle_stock_positions(lambda: _response()) + observation = cycle_history_observation() + finally: + end_natural_cycle_history(token) + + assert observation["projection"]["broker_reported_positions"] == [] + result, record = _record(observation) + assert result.status == "recorded" + assert record["broker_reported_positions"] == [] + assert record["broker_reported_positions_scope"] == "stocks_only" + assert record["broker_reported_positions_complete"] is True + + +def test_no_account_channel_is_unknown_not_a_proven_empty_book(): + # Reconciliation also treats zero channels as incomplete position evidence. + assert project_cycle_stock_positions(SimpleNamespace(channels=[])) is None + + +def test_stock_read_failure_keeps_balance_and_financing_and_leaves_inventory_unknown(): + calls = [] + financing = { + "max_finance_amount": Decimal("0.00"), + "buy_power": Decimal("12.00"), + "risk_level": 1, + } + + def failing_reader(): + calls.append(1) + raise RuntimeError("synthetic stock read failure") + + t_ctx = SimpleNamespace( + account_balance=lambda: [_balance(financing=financing)], + stock_positions=failing_reader, + ) + token = begin_natural_cycle_history() + try: + read_cycle_account_balance(t_ctx.account_balance) + try: + read_cycle_stock_positions(failing_reader) + except RuntimeError: + pass + observation = cycle_history_observation() + with pytest.raises(RuntimeError) as raised: + fetch_strategy_account_state(SimpleNamespace(quote=lambda _symbols: []), t_ctx, ["AAPL"]) + finally: + end_natural_cycle_history(token) + + assert str(raised.value) == "LongBridge stock positions unavailable" + assert len(calls) == 1 + projection = observation["projection"] + assert projection["broker_reported_balances"] == [ + {"currency": "USD", "net_assets": "100", "total_cash": "100"} + ] + assert projection["cash"] == [ + { + "currency": "USD", + "available_cash": "50", + "frozen_cash": "0", + "settling_cash": "0", + } + ] + assert projection["financing"] == [ + { + "currency": "USD", + "max_finance_amount": "0.00", + "buy_power": "12.00", + "risk_level": "1", + } + ] + assert "broker_reported_positions" not in projection + + result, record = _record(observation) + assert result.status == "recorded" + assert record["broker_reported_balances"] == projection["broker_reported_balances"] + assert record["cash"] == projection["cash"] + assert record["financing"] == projection["financing"] + assert "broker_reported_positions" not in record + assert "broker_reported_positions_scope" not in record + assert "broker_reported_positions_complete" not in record + + +def test_multi_currency_inventory_keeps_only_public_facts(): + rows = project_cycle_stock_positions( + _response( + _position( + "700.HK", + "HKD", + "3", + average_cost="private-cost", + market_value="999", + account_id="private-account", + ), + _position("AAPL.US", "USD", "2"), + ) + ) + + assert rows == [ + {"symbol": "700.HK", "currency": "HKD", "quantity": "3", "market_value": None}, + {"symbol": "AAPL.US", "currency": "USD", "quantity": "2", "market_value": None}, + ] + assert "private" not in json.dumps(rows) + assert "999" not in json.dumps(rows) + + +def test_duplicate_symbol_in_one_cycle_stays_unknown_without_fabricated_rows(): + duplicate = _response( + _position("700.HK", "HKD", "1"), + _position("700.HK", "USD", "2"), + ) + assert project_cycle_stock_positions(duplicate) is None + + token = begin_natural_cycle_history() + try: + read_cycle_account_balance(lambda: [_balance()]) + read_cycle_stock_positions(lambda: duplicate) + observation = cycle_history_observation() + finally: + end_natural_cycle_history(token) + + assert "broker_reported_positions" not in observation["projection"] + result, record = _record(observation) + assert result.status == "recorded" + assert "broker_reported_positions" not in record diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index ebb7b15..bb367bc 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -326,7 +326,8 @@ def bootstrap(self): raise AssertionError("history probe must use metadata contexts, not refresh bootstrap") return quote, trade, {} - portfolio_port_factory = lambda self, *_args: FakePort() + def portfolio_port_factory(self, *_args): + return FakePort() def calculate_indicators(_context): observed["indicators"] += 1