diff --git a/application/account_snapshot.py b/application/account_snapshot.py index 3daa6f9..c9367b7 100644 --- a/application/account_snapshot.py +++ b/application/account_snapshot.py @@ -5,9 +5,14 @@ from collections.abc import Mapping from contextvars import ContextVar, Token from datetime import datetime, timezone +import re from typing import Any -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( @@ -23,6 +28,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, } ) @@ -63,6 +71,71 @@ def cycle_history_observation() -> dict[str, Any] | None: } +def read_cycle_stock_positions(reader): + """Reuse one native stock read; archive the full inventory before strategy filtering.""" + 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) + if not isinstance(channels, (list, tuple)): + 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 8626add..b505832 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 @@ -147,7 +150,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 5139ad8..66008e4 100644 --- a/main.py +++ b/main.py @@ -32,6 +32,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.rebalance_service import run_strategy as run_rebalance_cycle @@ -1014,6 +1015,14 @@ 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")) + # Keep known balances even when the optional stock read fails. The + # portfolio below consumes the same response/error, without a retry. + 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") @@ -1031,6 +1040,7 @@ def run_probe(*, response_body: str = "Probe OK"): started=observation["started"], finished=observation["finished"], open_store=_open_account_history_store, + positions=observation["projection"].get("broker_reported_positions"), ) print( f"account_history status={result.status} category={result.category or '-'}", diff --git a/qsl.toml b/qsl.toml index 879819f..b64b5b2 100644 --- a/qsl.toml +++ b/qsl.toml @@ -5,7 +5,7 @@ upgrade_ring = "ring_d" allow_legacy = false [qsl.requires] -quant_platform_kit = "c7646a7168b3dafa763ef7751a182d23e8de7790" +quant_platform_kit = "038a10fc625d8310e8e3f43a969309450380ff63" us_equity_strategies = "8412d8338879f335ef8830678c2733e02e53e7c7" hk_equity_strategies = "cc3045689ae9d76eb97b9e1aa3b8fb6288ee1bcc" diff --git a/scripts/record_daily_account_snapshot.py b/scripts/record_daily_account_snapshot.py index 8026909..1a15b65 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( started: datetime, finished: datetime, open_store: Callable[[str], Any], + positions: Any = None, ) -> DailyAccountRecordResult: """Validate one HK/SG observation and confirm its create-only write.""" @@ -76,6 +78,7 @@ def record_projected_daily_account( cash=cash, started=started, finished=finished, + positions=positions, ) except _Rejected as rejected: return DailyAccountRecordResult("skipped", rejected.category) @@ -148,6 +151,7 @@ def _history_object( cash: Any, started: datetime, finished: datetime, + 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: @@ -180,6 +184,12 @@ def _history_object( "broker_reported_balances": _money_rows(balances, _BALANCE_FIELDS), "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, + ) body = json.dumps(record, ensure_ascii=True, separators=(",", ":"), sort_keys=True) filename = finished_utc.strftime("%H%M%S%fZ.json") uri = ( @@ -230,6 +240,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_native_stock_archive.py b/tests/test_native_stock_archive.py new file mode 100644 index 0000000..f342e94 --- /dev/null +++ b/tests/test_native_stock_archive.py @@ -0,0 +1,368 @@ +"""Synthetic coverage for the audited native stock-inventory archive port. + +These tests exercise only in-memory broker doubles: one request-local stock +read reused by the portfolio, the unfiltered archive projection, and the +producer's optional three-field stock block. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest + +from application import account_snapshot +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} + + +def _balance(currency="USD", net_assets="100", total_cash="100", available_cash="50"): + return SimpleNamespace( + currency=currency, + net_assets=net_assets, + total_cash=total_cash, + cash_infos=[ + SimpleNamespace( + currency=currency, + available_cash=available_cash, + frozen_cash="0", + settling_cash="0", + ) + ], + ) + + +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))]) + + +def _quotes(symbols): + return [SimpleNamespace(symbol=symbol, last_done="10.0") for symbol in symbols] + + +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): + self.name = name + return self + + def upload_from_string(self, data, *, content_type, if_generation_match, timeout, retry): + self.calls.append((data, content_type, if_generation_match, timeout, retry)) + + +def _producer_inputs(**overrides): + now = datetime.now(timezone.utc) + values = { + "env": { + "ACCOUNT_HISTORY_RECORDING_ENABLED": "true", + "ACCOUNT_HISTORY_GCS_PREFIX": "gs://history/longbridge/account_snapshots", + "ACCOUNT_HISTORY_TARGET_ID": "hk", + "ACCOUNT_HISTORY_EXPECTED_SCOPE": "HK", + "GOOGLE_CLOUD_PROJECT": "generic-project", + }, + "account_scope": "HK", + "source_binding": SOURCE, + "balances": [{"currency": "USD", "net_assets": "1234.5", "total_cash": "234.5"}], + "cash": [ + { + "currency": "HKD", + "available_cash": "200.25", + "frozen_cash": "20", + "settling_cash": "14.25", + } + ], + "started": now - timedelta(seconds=2), + "finished": now - timedelta(seconds=1), + } + values.update(overrides) + return values + + +def test_cycle_stock_read_is_performed_once_and_reused(): + token = account_snapshot.begin_natural_cycle_history() + calls = [] + response = _response(_position("700.HK", "HKD", "100")) + + def reader(): + calls.append(1) + return response + + try: + assert account_snapshot.read_cycle_stock_positions(reader) is response + assert account_snapshot.read_cycle_stock_positions(reader) is response + finally: + account_snapshot.end_natural_cycle_history(token) + + assert len(calls) == 1 + + +def test_cycle_stock_failure_is_reused_without_a_retry(): + token = account_snapshot.begin_natural_cycle_history() + calls = [] + error = RuntimeError("synthetic stock read failure") + + def reader(): + calls.append(1) + raise error + + try: + with pytest.raises(RuntimeError) as first: + account_snapshot.read_cycle_stock_positions(reader) + with pytest.raises(RuntimeError) as second: + account_snapshot.read_cycle_stock_positions(reader) + finally: + account_snapshot.end_natural_cycle_history(token) + + assert first.value is error + assert second.value is error + assert len(calls) == 1 + + +def test_cycle_projection_keeps_full_inventory_outside_strategy_assets(): + token = account_snapshot.begin_natural_cycle_history() + 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=_quotes) + try: + state = fetch_strategy_account_state(q_ctx, t_ctx, ["AAPL"]) + second_state = fetch_strategy_account_state(q_ctx, t_ctx, ["AAPL"]) + observation = account_snapshot.cycle_history_observation() + finally: + account_snapshot.end_natural_cycle_history(token) + + assert len(stock_calls) == 1 + 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}, + ] + assert state["quantities"] == {"AAPL": 2.0} + assert state["market_values"] == {"AAPL": 20.0} + assert second_state["quantities"] == state["quantities"] + + +def test_unknown_stock_response_is_not_archived_as_empty(): + token = account_snapshot.begin_natural_cycle_history() + try: + account_snapshot.read_cycle_account_balance(lambda: [_balance()]) + account_snapshot.read_cycle_stock_positions( + lambda: SimpleNamespace(channels=None) + ) + observation = account_snapshot.cycle_history_observation() + finally: + account_snapshot.end_natural_cycle_history(token) + + assert "broker_reported_positions" not in observation["projection"] + + +def test_proven_empty_stock_inventory_is_archived_as_empty(): + token = account_snapshot.begin_natural_cycle_history() + try: + account_snapshot.read_cycle_account_balance(lambda: [_balance()]) + account_snapshot.read_cycle_stock_positions(lambda: _response()) + observation = account_snapshot.cycle_history_observation() + finally: + account_snapshot.end_natural_cycle_history(token) + + assert observation["projection"]["broker_reported_positions"] == [] + + +def test_projection_keeps_only_public_stock_facts(): + rows = account_snapshot.project_cycle_stock_positions( + _response( + _position( + "700.HK", + "HKD", + "3", + average_cost="private-cost", + market_value="999", + account_id="private-account", + ) + ) + ) + + assert rows == [ + {"symbol": "700.HK", "currency": "HKD", "quantity": "3", "market_value": None} + ] + assert "private" not in json.dumps(rows) + + +@pytest.mark.parametrize( + "response", + [ + SimpleNamespace(channels=None), + SimpleNamespace(), + SimpleNamespace(channels=[SimpleNamespace(positions=None)]), + _response(_position("700.HK", "HKD", "1"), _position("700.HK", "HKD", "2")), + _response(_position("700.HK", "HKD", "NaN")), + _response(_position("700.HK", "HKD", "1234567890123456")), + _response(_position("700.HK", "HKD", "0.123456789")), + _response(*[_position(f"SYNTH{index:02d}.HK", "HKD", "1") for index in range(65)]), + ], +) +def test_unknown_or_unbounded_stock_projection_stays_unknown(response): + assert account_snapshot.project_cycle_stock_positions(response) is None + + +def test_portfolio_reuses_the_probe_failure_without_reading_again(): + token = account_snapshot.begin_natural_cycle_history() + calls = [] + + def stock_reader(): + calls.append(1) + raise RuntimeError("synthetic stock read failure") + + t_ctx = SimpleNamespace( + account_balance=lambda **_: [_balance()], + stock_positions=stock_reader, + ) + try: + account_snapshot.read_cycle_account_balance(t_ctx.account_balance) + with pytest.raises(RuntimeError): + account_snapshot.read_cycle_stock_positions(stock_reader) + with pytest.raises(RuntimeError) as raised: + fetch_strategy_account_state(SimpleNamespace(), t_ctx, ["AAPL"]) + finally: + account_snapshot.end_natural_cycle_history(token) + + assert str(raised.value) == "LongBridge stock positions unavailable" + assert len(calls) == 1 + + +def test_producer_without_inventory_keeps_the_original_object(): + store = _Store() + result = record_projected_daily_account( + **_producer_inputs(), open_store=lambda _project: store + ) + + assert result.status == "recorded" + record = json.loads(store.calls[0][0]) + assert "broker_reported_positions" not in record + assert "broker_reported_positions_scope" not in record + assert "broker_reported_positions_complete" not in record + assert record["broker_reported_balances"] == [ + {"currency": "USD", "net_assets": "1234.5", "total_cash": "234.5"} + ] + assert record["cash"] == [ + { + "currency": "HKD", + "available_cash": "200.25", + "frozen_cash": "20", + "settling_cash": "14.25", + } + ] + + +def test_producer_explicit_none_inventory_keeps_the_original_object(): + store = _Store() + result = record_projected_daily_account( + **_producer_inputs(positions=None), open_store=lambda _project: store + ) + + assert result.status == "recorded" + assert "broker_reported_positions" not in json.loads(store.calls[0][0]) + + +def test_producer_proven_empty_inventory_carries_all_three_fields(): + store = _Store() + result = record_projected_daily_account( + **_producer_inputs(positions=[]), open_store=lambda _project: store + ) + + assert result.status == "recorded" + record = json.loads(store.calls[0][0]) + assert record["broker_reported_positions"] == [] + assert record["broker_reported_positions_scope"] == "stocks_only" + assert record["broker_reported_positions_complete"] is True + + +def test_producer_projects_whitelisted_rows_without_private_fields(): + store = _Store() + result = record_projected_daily_account( + **_producer_inputs( + positions=[ + { + "symbol": "ZZZ.US", + "currency": "USD", + "quantity": "1", + "average_cost": "private-cost", + "market_value": "999", + }, + {"symbol": "700.HK", "currency": "HKD", "quantity": "-0.5"}, + ] + ), + open_store=lambda _project: store, + ) + + assert result.status == "recorded" + body = store.calls[0][0] + record = json.loads(body) + assert record["broker_reported_positions"] == [ + {"symbol": "700.HK", "quantity": "-0.5", "currency": "HKD", "market_value": None}, + {"symbol": "ZZZ.US", "quantity": "1", "currency": "USD", "market_value": None}, + ] + assert record["broker_reported_positions_scope"] == "stocks_only" + assert record["broker_reported_positions_complete"] is True + assert "private-cost" not in body + assert "999" not in body + + +@pytest.mark.parametrize( + "positions", + [ + "not-a-list", + {}, + [None], + ["700.HK"], + [{"symbol": "700.HK", "currency": "HKD"}], + [{"symbol": "700.HK", "currency": "hkd", "quantity": "1"}], + [{"symbol": "700.HK", "currency": "HKD", "quantity": True}], + [{"symbol": "700.HK", "currency": "HKD", "quantity": 1.5}], + [{"symbol": "700.HK", "currency": "HKD", "quantity": "NaN"}], + [{"symbol": "700.HK", "currency": "HKD", "quantity": "1e3"}], + [{"symbol": "700.HK", "currency": "HKD", "quantity": "1234567890123456"}], + [{"symbol": "700.HK", "currency": "HKD", "quantity": "0.123456789"}], + [ + {"symbol": "700.HK", "currency": "HKD", "quantity": "1"}, + {"symbol": "700.HK", "currency": "HKD", "quantity": "2"}, + ], + [{"symbol": f"SYNTH{index:02d}.HK", "currency": "HKD", "quantity": "1"} for index in range(65)], + ], +) +def test_producer_rejects_invalid_or_unbounded_inventory(positions): + opened = [] + + result = record_projected_daily_account( + **_producer_inputs(positions=positions), + open_store=lambda project: opened.append(project), + ) + + assert result.status == "skipped" + assert result.category == "projection_invalid" + assert opened == []