复用现有 probe 原生持仓读取补齐归档库存

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

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] "

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":

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")
@@ -152,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":
@@ -163,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)
@@ -203,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"])):
@@ -228,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
@@ -409,10 +417,63 @@ 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"]))
+
+
+_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]:
-    """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 +500,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 +509,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(
@@ -1449,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
@@ -1581,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

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 = {}