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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 79 additions & 1 deletion application/account_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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,
}
)

Expand Down Expand Up @@ -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."""

Expand Down
7 changes: 5 additions & 2 deletions application/longbridge_portfolio.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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] "
Expand Down
11 changes: 11 additions & 0 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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")
Expand All @@ -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 '-'}",
Expand Down
51 changes: 51 additions & 0 deletions scripts/record_daily_account_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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."""

Expand All @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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."""

Expand Down
Loading
Loading