From cb397d182f08a14964e80942732a2624fb8bdfa5 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 25 Oct 2026 03:14:30 +0800 Subject: [PATCH 1/2] Forward explicit stock inventory through the independent snapshot observer Co-Authored-By: Codex --- scripts/record_daily_account_snapshot.py | 41 +++- ...test_account_stock_inventory_projection.py | 215 ++++++++++++++++++ tests/test_daily_account_snapshot.py | 15 +- 3 files changed, 268 insertions(+), 3 deletions(-) create mode 100644 tests/test_account_stock_inventory_projection.py diff --git a/scripts/record_daily_account_snapshot.py b/scripts/record_daily_account_snapshot.py index c32ef3d..5d6a28f 100644 --- a/scripts/record_daily_account_snapshot.py +++ b/scripts/record_daily_account_snapshot.py @@ -44,6 +44,7 @@ _TARGET_ID = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[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_TEXT = re.compile(r"^-?(?:0|[1-9]\d*)(?:\.\d+)?$") _FILENAME = re.compile(r"^\d{12}Z\.json$") _FORBIDDEN_PREFIX_PARTS = ("execution-report", "execution_report", "runtime-report") @@ -409,10 +410,44 @@ def _independent_money_rows(value: object, fields: tuple[str, ...]) -> list[dict return rows +def _independent_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("snapshot_invalid") + rows: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in value: + if not isinstance(item, Mapping): + raise _Rejected("snapshot_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_TEXT.fullmatch(quantity) is None + ): + raise _Rejected("snapshot_invalid") + whole, _, fraction = quantity.lstrip("-").partition(".") + if len(whole) > 15 or len(fraction) > 8 or not Decimal(quantity).is_finite(): + raise _Rejected("snapshot_invalid") + if symbol in seen: + raise _Rejected("snapshot_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 _project_independent_history( payload: object, config: _IndependentConfig, started_at: datetime, now: datetime, ) -> dict[str, Any]: - """Discard private positions/orders; retain only the existing history contract.""" + """Project public stock inventory only; discard prices, private fields and orders.""" if not isinstance(payload, Mapping): raise _Rejected("snapshot_invalid") binding = payload.get("source_binding") @@ -439,6 +474,7 @@ def _project_independent_history( raise _Rejected("snapshot_invalid") balances = _independent_money_rows(payload.get("broker_reported_balances"), _BALANCE_FIELDS) cash = _independent_money_rows(payload.get("cash"), _CASH_FIELDS) + positions = _independent_position_rows(payload.get("positions")) record = { "schema_version": HISTORY_SCHEMA, "snapshot_schema_version": SNAPSHOT_SCHEMA, "account_scope": config.account_scope, "target_id": config.target_id, @@ -447,6 +483,9 @@ def _project_independent_history( "observed_finished_at": payload["observed_finished_at"], "snapshot_atomic": False, "observation_date": started.date().isoformat(), "broker_reported_balances": balances, "cash": cash, + "broker_reported_positions": positions, + "broker_reported_positions_scope": "stocks_only", + "broker_reported_positions_complete": True, } if "financing" in payload: record["financing"] = _producer_financing_rows( diff --git a/tests/test_account_stock_inventory_projection.py b/tests/test_account_stock_inventory_projection.py new file mode 100644 index 0000000..aafda4c --- /dev/null +++ b/tests/test_account_stock_inventory_projection.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts import record_daily_account_snapshot as snapshots # noqa: E402 + + +T0 = datetime(2026, 9, 28, 12, 0, tzinfo=timezone.utc) +BINDING = "a" * 64 +CONFIG = snapshots._IndependentConfig( + project_id="example-project", + region="asia-east2", + service="example-hk-reader", + service_url="https://example-hk-reader-abc123-uc.a.run.app", + source_binding_id=BINDING, + target_id="hk", + account_scope="HK", +) + + +def _payload(**overrides): + value = { + "schema_version": "longbridge_account_snapshot.v1", + "status": "partial", + "account_scope": "HK", + "observed_started_at": (T0 + timedelta(seconds=1)).isoformat(), + "observed_finished_at": (T0 + timedelta(seconds=2)).isoformat(), + "snapshot_atomic": False, + "broker_reported_balances": [{"currency": "USD", "net_assets": "10.25", "total_cash": "-2.5"}], + "cash": [{"currency": "HKD", "available_cash": "0", "frozen_cash": "0", "settling_cash": "0"}], + "positions": [{"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "3"}], + "positions_complete": True, + "cash_complete": True, + "no_order": True, + "live_authority_granted": False, + "source_binding": {"kind": "deployment_scope_token_version", "status": "bound", "id": BINDING}, + } + value.update(overrides) + return value + + +def _project(payload): + return snapshots._project_independent_history( + payload, CONFIG, T0, T0 + timedelta(seconds=5), + ) + + +def _rejects(payload): + with pytest.raises(snapshots._Rejected) as raised: + _project(payload) + assert raised.value.category == "snapshot_invalid" + + +def test_non_empty_stock_inventory_projects_quantity_without_price(): + record = _project(_payload(positions=[ + {"symbol": "700.HK", "currency": "HKD", "quantity": "100"}, + {"symbol": "AAPL.US", "currency": "USD", "quantity": "-0.5"}, + ])) + assert record["broker_reported_positions"] == [ + {"symbol": "700.HK", "quantity": "100", "currency": "HKD", "market_value": None}, + {"symbol": "AAPL.US", "quantity": "-0.5", "currency": "USD", "market_value": None}, + ] + assert record["broker_reported_positions_scope"] == "stocks_only" + assert record["broker_reported_positions_complete"] is True + + +def test_empty_positions_list_is_proven_empty_stock_inventory(): + record = _project(_payload(positions=[])) + assert record["broker_reported_positions"] == [] + assert record["broker_reported_positions_scope"] == "stocks_only" + assert record["broker_reported_positions_complete"] is True + + +_MISSING = object() + + +@pytest.mark.parametrize("positions", [_MISSING, None, {}, "[]", [None], ["SYNTH.HK"], 0]) +def test_missing_null_object_or_non_list_positions_are_rejected(positions): + payload = _payload() + if positions is _MISSING: + payload.pop("positions") + else: + payload["positions"] = positions + _rejects(payload) + + +@pytest.mark.parametrize("complete", [False, None, "true", 1, 0]) +def test_positions_complete_must_be_exactly_true(complete): + _rejects(_payload(positions_complete=complete)) + + +def test_missing_positions_complete_is_rejected(): + payload = _payload() + payload.pop("positions_complete") + _rejects(payload) + + +@pytest.mark.parametrize( + "quantity", + ["0", "-0", "-1.25", "0.00000001", "123456789012345.12345678", "-0.5"], +) +def test_fractional_short_and_zero_quantities_are_preserved(quantity): + record = _project(_payload(positions=[ + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": quantity}, + ])) + assert record["broker_reported_positions"][0]["quantity"] == quantity + + +@pytest.mark.parametrize( + "rows", + [ + [ + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "1"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "2"}, + ], + [ + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "1"}, + {"symbol": "SYNTH.HK", "currency": "USD", "quantity": "2"}, + ], + ], +) +def test_duplicate_symbols_are_rejected_without_aggregation(rows): + _rejects(_payload(positions=rows)) + + +@pytest.mark.parametrize( + "row", + [ + {"symbol": "synth.hk", "currency": "HKD", "quantity": "1"}, + {"symbol": " SYNTH.HK", "currency": "HKD", "quantity": "1"}, + {"symbol": "SYNTH.HK", "currency": "hkd", "quantity": "1"}, + {"symbol": "SYNTH.HK", "currency": "HK", "quantity": "1"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": True}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": 1.5}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "NaN"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "Infinity"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "1e3"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "00"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "1."}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": ".5"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "0.123456789"}, + {"symbol": "SYNTH.HK", "currency": "HKD", "quantity": "1234567890123456"}, + {"symbol": "SYNTH.HK", "currency": "HKD"}, + {"symbol": "SYNTH.HK", "quantity": "1"}, + {"currency": "HKD", "quantity": "1"}, + ], +) +def test_invalid_symbol_currency_or_quantity_is_rejected(row): + _rejects(_payload(positions=[row])) + + +def _rows(count): + return [ + {"symbol": f"SYNTH{index:02d}.HK", "currency": "HKD", "quantity": "1"} + for index in range(count) + ] + + +def test_more_than_sixty_four_rows_are_rejected(): + _rejects(_payload(positions=_rows(65))) + + +def test_sixty_four_rows_are_accepted(): + record = _project(_payload(positions=_rows(64))) + assert len(record["broker_reported_positions"]) == 64 + + +def test_private_position_fields_and_orders_are_not_projected(): + payload = _payload( + positions=[{ + "symbol": "SYNTH.HK", "currency": "HKD", "quantity": "3", + "average_cost": "private-cost", "market_value": "999", + "account_id": "private-account", + }], + known_non_terminal_orders_7d=[{"synthetic": "private-order"}], + known_recent_executions_7d_count=1, + ) + record = _project(payload) + assert record["broker_reported_positions"] == [ + {"symbol": "SYNTH.HK", "quantity": "3", "currency": "HKD", "market_value": None}, + ] + serialized = json.dumps(record) + assert "private" not in serialized and "999" not in serialized + + +def test_positions_are_sorted_by_currency_then_symbol_without_mutating_input(): + payload = _payload(positions=[ + {"symbol": "ZZZ.US", "currency": "USD", "quantity": "1"}, + {"symbol": "700.HK", "currency": "HKD", "quantity": "2"}, + {"symbol": "AAA.US", "currency": "USD", "quantity": "3"}, + {"symbol": "000.HK", "currency": "HKD", "quantity": "4"}, + ]) + before = json.loads(json.dumps(payload)) + record = _project(payload) + assert [row["symbol"] for row in record["broker_reported_positions"]] == [ + "000.HK", "700.HK", "AAA.US", "ZZZ.US", + ] + assert payload == before + + +def test_rejection_category_is_fixed_and_does_not_include_input(): + secret = "private-symbol-value" + with pytest.raises(snapshots._Rejected) as raised: + _project(_payload(positions=[{"symbol": secret, "currency": "HKD", "quantity": "1"}])) + assert raised.value.category == "snapshot_invalid" + assert secret not in raised.value.category diff --git a/tests/test_daily_account_snapshot.py b/tests/test_daily_account_snapshot.py index 9c0fb1c..e2bf98e 100644 --- a/tests/test_daily_account_snapshot.py +++ b/tests/test_daily_account_snapshot.py @@ -1684,7 +1684,10 @@ def _independent_snapshot(**overrides): "snapshot_atomic": False, "broker_reported_balances": [{"currency": "USD", "net_assets": "10.25", "total_cash": "-2.5"}], "cash": [{"currency": "HKD", "available_cash": "0", "frozen_cash": "0", "settling_cash": "0"}], - "positions": [{"symbol": "synthetic-private-symbol", "currency": "HKD", "quantity": "3"}], + "positions": [{ + "symbol": "SYNTH.HK", "currency": "HKD", "quantity": "3", + "synthetic_private_cost": "synthetic-private-cost", + }], "known_non_terminal_orders_7d": [{"synthetic": "private-order"}], "known_recent_executions_7d_count": 1, "positions_complete": True, @@ -1761,7 +1764,15 @@ def test_independent_hk_preserves_native_currency_and_only_publishes_existing_hi assert 0 < request["timeout"] <= snapshots.HTTP_TIMEOUT_SECONDS outgoing = calls["posts"][0][1]["data"] history = json.loads(outgoing) - assert set(history) == set(_history(target_id="hk")) + assert set(history) == set(_history(target_id="hk")) | { + "broker_reported_positions", "broker_reported_positions_scope", + "broker_reported_positions_complete", + } + assert history["broker_reported_positions"] == [{ + "symbol": "SYNTH.HK", "quantity": "3", "currency": "HKD", "market_value": None, + }] + assert history["broker_reported_positions_scope"] == "stocks_only" + assert history["broker_reported_positions_complete"] is True assert history["broker_reported_balances"] == _independent_snapshot()["broker_reported_balances"] assert history["cash"] == _independent_snapshot()["cash"] assert history["target_id"] == "hk" and history["account_scope"] == "HK" From 0825d71094b8e717ae182476e6baa1dd755619b6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 25 Oct 2026 09:32:30 +0800 Subject: [PATCH 2/2] Archive the native stock inventory in the existing probe cycle Co-Authored-By: Codex --- application/account_snapshot.py | 57 ++++++++++++++- application/longbridge_portfolio.py | 4 +- main.py | 11 ++- scripts/record_daily_account_snapshot.py | 36 ++++++++-- tests/test_account_snapshot_history.py | 92 ++++++++++++++++++++++++ tests/test_request_handling.py | 5 ++ 6 files changed, 196 insertions(+), 9 deletions(-) diff --git a/application/account_snapshot.py b/application/account_snapshot.py index b656505..caa8040 100644 --- a/application/account_snapshot.py +++ b/application/account_snapshot.py @@ -5,16 +5,18 @@ 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("longbridge_natural_cycle_history", default=None) def begin_natural_cycle_history() -> Token[dict[str, Any] | None]: - return _CAPTURE.set({"seen": False, "balance": None, "projection": None, "started": None, "finished": None}) + return _CAPTURE.set({"seen": False, "balance": None, "projection": None, "started": None, "finished": None, + "stocks_seen": False, "stocks": None, "stocks_error": None}) def end_natural_cycle_history(token: Token[dict[str, Any] | None]) -> None: @@ -45,6 +47,57 @@ def cycle_history_observation() -> dict[str, Any] | None: return {"projection": slot["projection"], "started": slot["started"], "finished": slot["finished"]} +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: if isinstance(account_balance, (str, bytes)) or not isinstance(account_balance, (list, tuple)) or not account_balance: return None diff --git a/application/longbridge_portfolio.py b/application/longbridge_portfolio.py index b8f70fe..807a70b 100644 --- a/application/longbridge_portfolio.py +++ b/application/longbridge_portfolio.py @@ -6,7 +6,7 @@ from typing import Any from quant_platform_kit.common.broker_reconciliation import calculate_broker_observation_sha256 -from application.account_snapshot import read_cycle_account_balance +from application.account_snapshot import read_cycle_account_balance, read_cycle_stock_positions def _broker_capital(account_balance, currency: str, observed_at: datetime) -> dict | None: @@ -148,7 +148,7 @@ def warn(message: str) -> None: position_rows: list[tuple[str, str, Any, Any]] = [] position_currency_by_symbol: dict[str, str | None] = {} 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 2d8bb5a..56b46cc 100644 --- a/main.py +++ b/main.py @@ -47,7 +47,7 @@ from application.longbridge_portfolio import fetch_strategy_account_state from application.account_snapshot import ( begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, - read_cycle_account_balance, + read_cycle_account_balance, read_cycle_stock_positions, ) from scripts.record_daily_account_snapshot import record_projected_daily_account from entrypoints.cloud_run import is_market_open_now @@ -1075,6 +1075,14 @@ def run_probe(*, response_body: str = "Probe OK"): quote_context, trade_context, source_binding = composer.build_account_snapshot_broker_contexts() probe_step = "balance" 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") @@ -1089,6 +1097,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 '-'}", flush=True) if result.status != "recorded": diff --git a/scripts/record_daily_account_snapshot.py b/scripts/record_daily_account_snapshot.py index 5d6a28f..fd1f9ab 100644 --- a/scripts/record_daily_account_snapshot.py +++ b/scripts/record_daily_account_snapshot.py @@ -153,6 +153,7 @@ def record_projected_daily_account( balances: Any, cash: Any, started: datetime, finished: datetime, open_store: Callable[[str], Any], financing: Any = None, + positions: Any = None, ) -> DailyAccountRecordResult: """Create one bounded history object from an already-read broker response.""" if str(env.get("ACCOUNT_HISTORY_RECORDING_ENABLED") or "").strip() != "true": @@ -164,7 +165,7 @@ def record_projected_daily_account( if account_scope != config.account_scope: raise _Rejected("scope") body, uri = _producer_history_object( - config, source_binding, balances, cash, started, finished, financing=financing, + config, source_binding, balances, cash, started, finished, financing=financing, positions=positions, ) except _Rejected as rejected: return DailyAccountRecordResult("skipped", rejected.category) @@ -204,7 +205,7 @@ def _producer_gcs_prefix(value: str) -> str: return f"gs://{parsed.netloc}/{'/'.join(segments)}" -def _producer_history_object(config, source_binding, balances, cash, started, finished, financing=None): +def _producer_history_object(config, source_binding, balances, cash, started, finished, financing=None, positions=None): if (not isinstance(source_binding, Mapping) or source_binding.get("kind") != SOURCE_KIND or source_binding.get("status") != "bound" or not isinstance(source_binding.get("id"), str) or not _BINDING_ID.fullmatch(source_binding["id"])): @@ -229,6 +230,12 @@ def _producer_history_object(config, source_binding, balances, cash, started, fi record["financing"] = _producer_financing_rows( financing, {row["currency"] for row in balance_rows}, ) + if positions is not None: + record.update( + broker_reported_positions=_independent_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) uri = f"{config.prefix}/{config.target_id}/{source_binding['id']}/{record['observation_date']}/{finished.strftime('%H%M%S%fZ.json')}" return body, uri @@ -444,6 +451,25 @@ def _independent_position_rows(value: object) -> list[dict[str, Any]]: return sorted(rows, key=lambda row: (row["currency"], row["symbol"])) +_HISTORY_POSITION_FIELDS = frozenset({ + "broker_reported_positions", "broker_reported_positions_scope", "broker_reported_positions_complete", +}) + + +def _history_positions_valid(payload: Mapping[str, Any]) -> bool: + present = set(payload) & _HISTORY_POSITION_FIELDS + if not present: + return True + if (present != _HISTORY_POSITION_FIELDS + or payload.get("broker_reported_positions_scope") != "stocks_only" + or payload.get("broker_reported_positions_complete") is not True): + return False + try: + return payload["broker_reported_positions"] == _independent_position_rows(payload["broker_reported_positions"]) + except _Rejected: + return False + + def _project_independent_history( payload: object, config: _IndependentConfig, started_at: datetime, now: datetime, ) -> dict[str, Any]: @@ -1488,11 +1514,12 @@ def inspect_archived_account_snapshot( scope_matches = payload.get("account_scope") == config.expected_scope payload_fields = set(payload) schema_matches = ( - required_fields <= payload_fields <= required_fields | {"financing"} + required_fields <= payload_fields <= required_fields | {"financing"} | _HISTORY_POSITION_FIELDS and payload.get("schema_version") == HISTORY_SCHEMA and payload.get("snapshot_schema_version") == SNAPSHOT_SCHEMA and payload.get("snapshot_atomic") is False and ("financing" not in payload or financing_shape_ok(payload.get("financing"))) + and _history_positions_valid(payload) ) trigger_time_matches = ( started is not None and finished is not None @@ -1620,7 +1647,8 @@ def _validate_history_object( binding = payload.get("source_binding") payload_fields = set(payload) if ( - not expected_fields <= payload_fields <= expected_fields | {"financing"} + not expected_fields <= payload_fields <= expected_fields | {"financing"} | _HISTORY_POSITION_FIELDS + or not _history_positions_valid(payload) or payload.get("schema_version") != HISTORY_SCHEMA or payload.get("snapshot_schema_version") != SNAPSHOT_SCHEMA or payload.get("account_scope") != config.expected_scope diff --git a/tests/test_account_snapshot_history.py b/tests/test_account_snapshot_history.py index dac98fb..2c6fc00 100644 --- a/tests/test_account_snapshot_history.py +++ b/tests/test_account_snapshot_history.py @@ -2,12 +2,15 @@ from datetime import datetime, timezone from decimal import Decimal from types import SimpleNamespace +import pytest from application.account_snapshot import ( begin_natural_cycle_history, cycle_history_observation, end_natural_cycle_history, read_cycle_account_balance, + read_cycle_stock_positions, + project_cycle_stock_positions, ) from scripts import record_daily_account_snapshot as history @@ -106,6 +109,95 @@ def test_capture_failure_is_not_retried_and_context_is_reset(): assert calls == [1, 3] +def test_stock_capture_keeps_full_inventory_once_and_discards_private_fields(): + from application.longbridge_portfolio import fetch_strategy_account_state + response = SimpleNamespace(channels=[SimpleNamespace(account_channel="private-channel", positions=[ + SimpleNamespace(symbol="EXAMPLE.US", currency="USD", quantity=Decimal("3.250"), cost_price="123", account_id="private"), + SimpleNamespace(symbol="OTHER.HK", currency="HKD", quantity=Decimal("7")), + ])]) + calls = [] + token = begin_natural_cycle_history() + try: + read_cycle_account_balance(lambda: [_balance("USD", "100", [("USD", "2")])]) + assert read_cycle_stock_positions(lambda: calls.append(1) or response) is response + assert read_cycle_stock_positions(lambda: calls.append(2)) is response + def unexpected_read(*_args, **_kwargs): + raise AssertionError("cached reads and filtered-out stocks need no further broker request") + strategy = fetch_strategy_account_state( + SimpleNamespace(quote=unexpected_read), + SimpleNamespace(account_balance=unexpected_read, stock_positions=unexpected_read), + ["DIFFERENT"], + ) + assert strategy["quantities"] == {"DIFFERENT": 0.0} + rows = cycle_history_observation()["projection"]["broker_reported_positions"] + assert rows == [ + {"symbol": "OTHER.HK", "currency": "HKD", "quantity": "7", "market_value": None}, + {"symbol": "EXAMPLE.US", "currency": "USD", "quantity": "3.25", "market_value": None}, + ] + assert calls == [1] + assert "private" not in json.dumps(rows) and "cost_price" not in json.dumps(rows) + finally: + end_natural_cycle_history(token) + + +def test_stock_read_failure_is_reused_without_losing_balance_or_claiming_empty(): + token = begin_natural_cycle_history() + calls = [] + try: + read_cycle_account_balance(lambda: [_balance("USD", "100", [("USD", "2")])]) + def fail(): + calls.append(1) + raise ValueError("synthetic unavailable") + for _ in range(2): + with pytest.raises(ValueError): + read_cycle_stock_positions(fail) + observed = cycle_history_observation() + assert "broker_reported_positions" not in observed["projection"] + assert observed["projection"]["broker_reported_balances"] + assert calls == [1] + finally: + end_natural_cycle_history(token) + + +@pytest.mark.parametrize("response", [None, SimpleNamespace(channels=None), SimpleNamespace(channels=[SimpleNamespace()]), + SimpleNamespace(channels=[SimpleNamespace(positions=[SimpleNamespace(symbol="BAD", currency="USD", quantity="NaN")])]), + SimpleNamespace(channels=[SimpleNamespace(positions=[SimpleNamespace(symbol="BAD", currency="USD", quantity="1")]*2)]), +]) +def test_malformed_stock_inventory_is_unknown(response): + assert project_cycle_stock_positions(response) is None + + +def test_native_empty_stock_inventory_is_proven_empty(): + assert project_cycle_stock_positions(SimpleNamespace(channels=[])) == [] + assert project_cycle_stock_positions(SimpleNamespace(channels=[SimpleNamespace(positions=[])])) == [] + + +@pytest.mark.parametrize("positions", [[], [{"symbol": "EXAMPLE.US", "currency": "USD", "quantity": "2", "private": "discard"}]]) +def test_archive_positions_round_trip_and_rejects_partial_markers(positions): + now = datetime.now(timezone.utc) + config = history._ProducerConfig("gs://generic-history-bucket/longbridge/account_snapshots", "hk", "HK", "generic-project") + body, uri = history._producer_history_object(config, {"kind": history.SOURCE_KIND, "status": "bound", "id": "a"*64}, + [{"currency": "USD", "net_assets": "100", "total_cash": "2"}], + [{"currency": "USD", "available_cash": "2", "frozen_cash": "0", "settling_cash": "0"}], now, now, positions=positions) + payload = json.loads(body) + consumer = history._Config("generic-project", "region", "service", "https://service.run.app", config.prefix, + "generic-history-bucket", "longbridge/account_snapshots", "hk", "HK", "a"*64, "job", "resource") + candidate = {"name": uri.removeprefix("gs://generic-history-bucket/")} + history._validate_history_object(payload, consumer, candidate, now, now) + assert payload["broker_reported_positions_complete"] is True + assert payload["broker_reported_positions_scope"] == "stocks_only" + assert "private" not in body + for key in history._HISTORY_POSITION_FIELDS: + malformed = dict(payload) + del malformed[key] + with pytest.raises(history._Rejected): + history._validate_history_object(malformed, consumer, candidate, now, now) + if positions: + payload["broker_reported_positions"][0]["market_value"] = "100" + with pytest.raises(history._Rejected): + history._validate_history_object(payload, consumer, candidate, now, now) + + def test_create_only_upload_has_finite_timeout_and_no_retry(): received = {} diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 4425c3b..1701b50 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -1172,6 +1172,7 @@ def test_probe_archives_real_portfolio_balance_once_with_bound_source(self): module = load_module() events, orders, balance_calls, uploads, parsed_objects, context_tokens = [], [], [], [], [], [] + stock_calls = [] response = [types.SimpleNamespace( currency="USD", net_assets="100.25", total_cash="80", cash_infos=[ @@ -1186,6 +1187,7 @@ def account_balance(self, **kwargs): return response def stock_positions(self): + stock_calls.append(1) return types.SimpleNamespace(channels=[]) trade_context = TradeContext() @@ -1266,12 +1268,15 @@ def build_notification_adapters(self): self.assertEqual(result, ("Probe OK", 200)) self.assertEqual(balance_calls, [{"currency": "USD"}]) + self.assertEqual(stock_calls, [1]) self.assertEqual(context_tokens, ["opaque-token"]) self.assertEqual(len(uploads), 1) payload, content_type, generation, timeout, retry = uploads[0] self.assertEqual(payload["source_binding"], expected_binding) self.assertEqual(payload["broker_reported_balances"], [{"currency": "USD", "net_assets": "100.25", "total_cash": "80"}]) self.assertEqual({row["currency"] for row in payload["cash"]}, {"USD", "HKD"}) + self.assertEqual(payload["broker_reported_positions"], []) + self.assertIs(payload["broker_reported_positions_complete"], True) started = datetime.fromisoformat(payload["observed_started_at"]) finished = datetime.fromisoformat(payload["observed_finished_at"]) self.assertIsNotNone(started.utcoffset())