From a7522bf8d12e5054373f3b8586754f3c32ceb264 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 3 Jul 2026 01:15:17 +0200 Subject: [PATCH] =?UTF-8?q?dita=5Fv2:=20reconcile=20SOA=20=E2=80=94=20vend?= =?UTF-8?q?ored=20thread-safety=20hardening=20=E2=88=AA=20venue=20plane?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 3-way merge (base=pre-venue-plane upstream, ours=violet-main vendored hardening, theirs=upstream venue plane): RLock atomic-snapshot wrapping (account.py, real_control_plane, real_zinc_plane), lazy __getattr__ imports (__init__), account-core test coverage — merged with venue_region telemetry. Zero conflicts; all files AST-verified. Ends the two-way vendor drift found in DITAV2_SOA_SURVEY_20260702 §5 / UV_DITAV2_SOA_VERDICT_20260703. Co-Authored-By: Claude Fable 5 --- prod/clean_arch/dita_v2/__init__.py | 24 +- prod/clean_arch/dita_v2/account.py | 244 ++++++++++-------- prod/clean_arch/dita_v2/real_control_plane.py | 26 +- prod/clean_arch/dita_v2/real_zinc_plane.py | 25 +- .../dita_v2/test_account_core_v2.py | 42 ++- 5 files changed, 225 insertions(+), 136 deletions(-) diff --git a/prod/clean_arch/dita_v2/__init__.py b/prod/clean_arch/dita_v2/__init__.py index 0a10dcbc..1da37bef 100644 --- a/prod/clean_arch/dita_v2/__init__.py +++ b/prod/clean_arch/dita_v2/__init__.py @@ -36,8 +36,6 @@ from .contracts import ( ) from .journal import ClickHouseKernelJournal, KernelJournal, MemoryKernelJournal from .rust_backend import ExecutionKernel -from .bingx_venue import BingxVenueAdapter -from .launcher import DITAv2LauncherBundle, LauncherVenueMode, LauncherZincMode, build_launcher_bundle from .projection import HazelcastProjection, build_position_state_row, build_projection from .venue import VenueAdapter from .mock_venue import MockVenueAdapter, MockVenueScenario @@ -45,6 +43,28 @@ from .zinc_plane import InMemoryZincPlane, ZincPlane from .real_zinc_plane import RealZincPlane, RealZincUnavailable from .real_control_plane import RealZincControlPlane, RealZincUnavailable as RealZincControlUnavailable + +def __getattr__(name: str): + if name == "BingxVenueAdapter": + from .bingx_venue import BingxVenueAdapter + + return BingxVenueAdapter + if name in {"DITAv2LauncherBundle", "LauncherVenueMode", "LauncherZincMode", "build_launcher_bundle"}: + from .launcher import ( + DITAv2LauncherBundle, + LauncherVenueMode, + LauncherZincMode, + build_launcher_bundle, + ) + + return { + "DITAv2LauncherBundle": DITAv2LauncherBundle, + "LauncherVenueMode": LauncherVenueMode, + "LauncherZincMode": LauncherZincMode, + "build_launcher_bundle": build_launcher_bundle, + }[name] + raise AttributeError(name) + __all__ = [ "AccountProjection", "AccountSnapshot", diff --git a/prod/clean_arch/dita_v2/account.py b/prod/clean_arch/dita_v2/account.py index 19ae2718..cd8d5a81 100644 --- a/prod/clean_arch/dita_v2/account.py +++ b/prod/clean_arch/dita_v2/account.py @@ -8,6 +8,7 @@ from enum import Enum from typing import Any, Dict, Iterable, List, Optional import math import time +import threading from .contracts import TradeSide, TradeSlot, TradeStage from .utils import safe_float @@ -59,42 +60,49 @@ class AccountProjection: GIL guarantees single-field reference assignment is atomic, so readers that hold snap = kernel.account.snapshot before use see a consistent view. """ - cur = self.snapshot - self.snapshot = AccountSnapshot( - capital=kw.get("capital", cur.capital), - equity=kw.get("equity", cur.equity), - realized_pnl=kw.get("realized_pnl", cur.realized_pnl), - unrealized_pnl=kw.get("unrealized_pnl", cur.unrealized_pnl), - open_positions=kw.get("open_positions", cur.open_positions), - open_notional=kw.get("open_notional", cur.open_notional), - fees_paid=kw.get("fees_paid", cur.fees_paid), - trade_seq=kw.get("trade_seq", cur.trade_seq), - peak_capital=kw.get("peak_capital", cur.peak_capital), - capital_source=kw.get("capital_source", cur.capital_source), - e_wallet_balance=kw.get("e_wallet_balance", cur.e_wallet_balance), - event_seq=kw.get("event_seq", cur.event_seq), - ) + with self._lock: + cur = self.snapshot + self.snapshot = AccountSnapshot( + capital=kw.get("capital", cur.capital), + equity=kw.get("equity", cur.equity), + realized_pnl=kw.get("realized_pnl", cur.realized_pnl), + unrealized_pnl=kw.get("unrealized_pnl", cur.unrealized_pnl), + open_positions=kw.get("open_positions", cur.open_positions), + open_notional=kw.get("open_notional", cur.open_notional), + fees_paid=kw.get("fees_paid", cur.fees_paid), + trade_seq=kw.get("trade_seq", cur.trade_seq), + peak_capital=kw.get("peak_capital", cur.peak_capital), + capital_source=kw.get("capital_source", cur.capital_source), + e_wallet_balance=kw.get("e_wallet_balance", cur.e_wallet_balance), + event_seq=kw.get("event_seq", cur.event_seq), + ) + + def __post_init__(self) -> None: + self._lock = threading.RLock() def observe_slots(self, slots: Iterable[TradeSlot]) -> None: - open_positions = 0 - open_notional = 0.0 - unrealized_pnl = 0.0 - for slot in slots: - if slot.closed or slot.size <= 0: - continue - if slot.fsm_state in {TradeStage.POSITION_OPEN, TradeStage.POSITION_OPENED, TradeStage.ENTRY_WORKING, TradeStage.EXIT_WORKING}: - open_positions += 1 - mark = safe_float(slot.entry_price, 0.0) - mark = safe_float(slot.metadata.get("mark_price"), mark) - open_notional += abs(slot.size) * abs(mark) - unrealized_pnl += float(slot.unrealized_pnl or 0.0) - self._replace_snapshot( - open_positions=open_positions, - open_notional=open_notional, - unrealized_pnl=unrealized_pnl, - equity=self.snapshot.capital + unrealized_pnl if math.isfinite(self.snapshot.capital + unrealized_pnl) else self.snapshot.capital, - peak_capital=max(self.snapshot.peak_capital, self.snapshot.capital) if open_notional > 0 and self.snapshot.capital > 0 else self.snapshot.peak_capital, - ) + with self._lock: + open_positions = 0 + open_notional = 0.0 + unrealized_pnl = 0.0 + for slot in slots: + if slot.closed or slot.size <= 0: + continue + if slot.fsm_state in {TradeStage.POSITION_OPEN, TradeStage.POSITION_OPENED, TradeStage.ENTRY_WORKING, TradeStage.EXIT_WORKING}: + open_positions += 1 + mark = safe_float(slot.entry_price, 0.0) + mark = safe_float(slot.metadata.get("mark_price"), mark) + open_notional += abs(slot.size) * abs(mark) + unrealized_pnl += float(slot.unrealized_pnl or 0.0) + capital = self.snapshot.capital + peak_capital = self.snapshot.peak_capital + self._replace_snapshot( + open_positions=open_positions, + open_notional=open_notional, + unrealized_pnl=unrealized_pnl, + equity=capital + unrealized_pnl if math.isfinite(capital + unrealized_pnl) else capital, + peak_capital=max(peak_capital, capital) if open_notional > 0 and capital > 0 else peak_capital, + ) def anchor_to_exchange(self, wallet_balance: float, available_margin: float, event_seq: int) -> None: """Snap published capital to exchange wallet balance. @@ -106,39 +114,42 @@ class AccountProjection: Guards: wallet_balance must be > 0 and finite (the zero-wb frame lesson from ACCOUNT_UPDATE frames with no USDT balance entry). """ - wb = safe_float(wallet_balance, 0.0) - if wb <= 0.0 or not math.isfinite(wb): - return - self.snapshot.capital = wb - self.snapshot.e_wallet_balance = wb - self.snapshot.capital_source = "e_anchored" - self.snapshot.event_seq = int(event_seq) - self.snapshot.equity = wb + self.snapshot.unrealized_pnl - if not math.isfinite(self.snapshot.equity): - self.snapshot.equity = wb - self.snapshot.peak_capital = max(self.snapshot.peak_capital, wb) + with self._lock: + wb = safe_float(wallet_balance, 0.0) + if wb <= 0.0 or not math.isfinite(wb): + return + self.snapshot.capital = wb + self.snapshot.e_wallet_balance = wb + self.snapshot.capital_source = "e_anchored" + self.snapshot.event_seq = int(event_seq) + self.snapshot.equity = wb + self.snapshot.unrealized_pnl + if not math.isfinite(self.snapshot.equity): + self.snapshot.equity = wb + self.snapshot.peak_capital = max(self.snapshot.peak_capital, wb) def settle(self, realized_pnl: float, fees: float = 0.0) -> None: - rp = safe_float(realized_pnl, 0.0) - # Include fees in capital delta (today fees only accumulate in - # fees_paid while published capital ignores them between reseeds). - net = rp - safe_float(fees, 0.0) - new_capital = safe_float(self.snapshot.capital + net, self.snapshot.capital) - if self.max_capital is not None: - new_capital = min(new_capital, self.max_capital) - new_capital = max(self.min_capital, new_capital) - new_source = self.snapshot.capital_source - if new_source == "e_anchored" and abs(net) > 1e-12: - new_source = "k_bridged" - new_fees = self.snapshot.fees_paid + safe_float(fees, 0.0) - new_equity = new_capital + self.snapshot.unrealized_pnl - if not math.isfinite(new_equity): - new_equity = new_capital - self._replace_snapshot( - capital=new_capital, capital_source=new_source, - realized_pnl=self.snapshot.realized_pnl + rp, - fees_paid=new_fees, equity=new_equity, - ) + with self._lock: + cur = self.snapshot + rp = safe_float(realized_pnl, 0.0) + # Include fees in capital delta (today fees only accumulate in + # fees_paid while published capital ignores them between reseeds). + net = rp - safe_float(fees, 0.0) + new_capital = safe_float(cur.capital + net, cur.capital) + if self.max_capital is not None: + new_capital = min(new_capital, self.max_capital) + new_capital = max(self.min_capital, new_capital) + new_source = cur.capital_source + if new_source == "e_anchored" and abs(net) > 1e-12: + new_source = "k_bridged" + new_fees = cur.fees_paid + safe_float(fees, 0.0) + new_equity = new_capital + cur.unrealized_pnl + if not math.isfinite(new_equity): + new_equity = new_capital + self._replace_snapshot( + capital=new_capital, capital_source=new_source, + realized_pnl=cur.realized_pnl + rp, + fees_paid=new_fees, equity=new_equity, + ) def to_account_event( self, @@ -154,31 +165,32 @@ class AccountProjection: bars_held: int = 0, metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: - self.snapshot.equity = self.snapshot.capital + self.snapshot.unrealized_pnl - return { - "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp), - "runtime_namespace": self.runtime_namespace, - "strategy_namespace": self.strategy_namespace, - "event_namespace": self.event_namespace, - "actor_name": self.actor_name, - "exec_venue": self.exec_venue, - "data_venue": self.data_venue, - "ledger_authority": self.ledger_authority, - "capital": float(self.snapshot.capital), - "equity": float(self.snapshot.equity), - "open_positions": int(self.snapshot.open_positions), - "current_open_notional": float(self.snapshot.open_notional), - "current_account_leverage": float(self.snapshot.leverage), - "trade_id": trade_id, - "asset": asset, - "side": side.value, - "reason": reason, - "stage": stage.value, - "pnl": float(pnl), - "pnl_pct": float(pnl_pct), - "bars_held": int(bars_held), - "metadata": dict(metadata or {}), - } + with self._lock: + self.snapshot.equity = self.snapshot.capital + self.snapshot.unrealized_pnl + return { + "timestamp": timestamp.isoformat() if hasattr(timestamp, "isoformat") else str(timestamp), + "runtime_namespace": self.runtime_namespace, + "strategy_namespace": self.strategy_namespace, + "event_namespace": self.event_namespace, + "actor_name": self.actor_name, + "exec_venue": self.exec_venue, + "data_venue": self.data_venue, + "ledger_authority": self.ledger_authority, + "capital": float(self.snapshot.capital), + "equity": float(self.snapshot.equity), + "open_positions": int(self.snapshot.open_positions), + "current_open_notional": float(self.snapshot.open_notional), + "current_account_leverage": float(self.snapshot.leverage), + "trade_id": trade_id, + "asset": asset, + "side": side.value, + "reason": reason, + "stage": stage.value, + "pnl": float(pnl), + "pnl_pct": float(pnl_pct), + "bars_held": int(bars_held), + "metadata": dict(metadata or {}), + } # --------------------------------------------------------------------------- @@ -311,6 +323,7 @@ class AccountProjectionV2: self._min_capital = min_capital self._max_capital = max_capital self._cfg = reconcile_config or ReconcileConfig() + self._lock = threading.RLock() # Running K-value accumulators self._k_realized: float = 0.0 @@ -345,16 +358,18 @@ class AccountProjectionV2: fee: float, realized_pnl: float, ) -> None: - self._k_realized += _safe(realized_pnl) - self._k_fees += _safe(fee) - self._e_last_fill_price = _safe(fill_price) - self._e_last_fill_qty = _safe(fill_qty) - self._e_last_fill_fee = _safe(fee) - self._e_last_fill_realized = _safe(realized_pnl) + with self._lock: + self._k_realized += _safe(realized_pnl) + self._k_fees += _safe(fee) + self._e_last_fill_price = _safe(fill_price) + self._e_last_fill_qty = _safe(fill_qty) + self._e_last_fill_fee = _safe(fee) + self._e_last_fill_realized = _safe(realized_pnl) def apply_funding(self, amount: float) -> None: - self._k_funding += _safe(amount) - self._e_last_funding = _safe(amount) + with self._lock: + self._k_funding += _safe(amount) + self._e_last_funding = _safe(amount) def apply_balance_update( self, @@ -364,13 +379,15 @@ class AccountProjectionV2: used_margin: float, maint_margin: float, ) -> None: - self._e_wallet_balance = _safe(wallet_balance) - self._e_avail_margin = _safe(available_margin) - self._e_used_margin = _safe(used_margin) - self._e_maint_margin = _safe(maint_margin) + with self._lock: + self._e_wallet_balance = _safe(wallet_balance) + self._e_avail_margin = _safe(available_margin) + self._e_used_margin = _safe(used_margin) + self._e_maint_margin = _safe(maint_margin) def apply_position_update(self, positions: List[EPosition]) -> None: - self._e_positions = list(positions) + with self._lock: + self._e_positions = list(positions) # ------------------------------------------------------------------ # Snapshot construction (called after each ingestion step) @@ -382,21 +399,24 @@ class AccountProjectionV2: slots: Iterable[TradeSlot], ts: Optional[float] = None, ) -> AccountSnapshotV2: - self._event_seq += 1 - snap = self._build(self._event_seq, source_event_id, list(slots), ts or time.time()) - self._snapshot = snap - return snap + with self._lock: + self._event_seq += 1 + snap = self._build(self._event_seq, source_event_id, list(slots), ts or time.time()) + self._snapshot = snap + return snap @property def snapshot(self) -> AccountSnapshotV2: - return self._snapshot + with self._lock: + return self._snapshot @property def k_capital(self) -> float: - raw = self._seed + self._k_realized - self._k_fees - self._k_funding - if self._max_capital is not None: - raw = min(raw, self._max_capital) - return max(self._min_capital, raw) + with self._lock: + raw = self._seed + self._k_realized - self._k_fees - self._k_funding + if self._max_capital is not None: + raw = min(raw, self._max_capital) + return max(self._min_capital, raw) # ------------------------------------------------------------------ # Internal helpers diff --git a/prod/clean_arch/dita_v2/real_control_plane.py b/prod/clean_arch/dita_v2/real_control_plane.py index 0139b910..13842712 100644 --- a/prod/clean_arch/dita_v2/real_control_plane.py +++ b/prod/clean_arch/dita_v2/real_control_plane.py @@ -8,6 +8,8 @@ import sys from pathlib import Path from typing import Any, Dict, Optional +import threading + from .control import BackendMode, ControlPlane, ControlUpdate, KernelControlSnapshot, KernelMode, KernelVerbosity _ZINC_ADAPTER_PATH = Path(__file__).resolve().parents[3] / "zinc" / "adapters" / "python" @@ -70,6 +72,7 @@ class RealZincControlPlane(ControlPlane): require_real_zinc() base = prefix.strip("/").replace("/", "_") self.region_name = f"{base}_control" + self._lock = threading.RLock() self._seq = 0 self._snapshot = KernelControlSnapshot() if create: @@ -86,21 +89,24 @@ class RealZincControlPlane(ControlPlane): self.region.close() def read(self) -> KernelControlSnapshot: - payload = _decode_packet(self.region.as_buffer()) - control = payload.get("control") if isinstance(payload, dict) else None - if not isinstance(control, dict): + with self._lock: + payload = _decode_packet(self.region.as_buffer()) + control = payload.get("control") if isinstance(payload, dict) else None + if not isinstance(control, dict): + return self._snapshot + self._snapshot = KernelControlSnapshot(**control) return self._snapshot - self._snapshot = KernelControlSnapshot(**control) - return self._snapshot def update(self, update: ControlUpdate) -> KernelControlSnapshot: - self._snapshot = update.apply(self.read()) - self._seq += 1 - self._write_region(self._seq, self._snapshot.as_dict()) - return self._snapshot + with self._lock: + self._snapshot = update.apply(self.read()) + self._seq += 1 + self._write_region(self._seq, self._snapshot.as_dict()) + return self._snapshot def mirror(self) -> Dict[str, Any]: - return self._snapshot.as_dict() + with self._lock: + return self._snapshot.as_dict() def wait(self, timeout_ms: int = 1000) -> bool: try: diff --git a/prod/clean_arch/dita_v2/real_zinc_plane.py b/prod/clean_arch/dita_v2/real_zinc_plane.py index 75bd5dba..94c9f823 100644 --- a/prod/clean_arch/dita_v2/real_zinc_plane.py +++ b/prod/clean_arch/dita_v2/real_zinc_plane.py @@ -258,14 +258,16 @@ class RealZincPlane: self._write_region(self.state_region, self._state_seq, payload) def read_slots(self) -> List[TradeSlot]: - payload = _decode_packet(self.state_region.as_buffer()) - slots = payload.get("slots", []) if isinstance(payload, dict) else [] - return [_slot_from_payload(slot) for slot in sorted(slots, key=lambda row: int(row.get("slot_id", 0)))] + with self._lock: + payload = _decode_packet(self.state_region.as_buffer()) + slots = payload.get("slots", []) if isinstance(payload, dict) else [] + return [_slot_from_payload(slot) for slot in sorted(slots, key=lambda row: int(row.get("slot_id", 0)))] def read_intents(self) -> List[Dict[str, Any]]: - payload = _decode_packet(self.intent_region.as_buffer()) - items = payload.get("items", []) if isinstance(payload, dict) else [] - return list(items) + with self._lock: + payload = _decode_packet(self.intent_region.as_buffer()) + items = payload.get("items", []) if isinstance(payload, dict) else [] + return list(items) def update_control(self, control: KernelControlSnapshot) -> None: with self._lock: @@ -274,11 +276,12 @@ class RealZincPlane: self._write_region(self.control_region, self._control_seq, {"control": control.as_dict()}) def read_control(self) -> KernelControlSnapshot: - payload = _decode_packet(self.control_region.as_buffer()) - control = payload.get("control") if isinstance(payload, dict) else None - if not isinstance(control, dict): - return self._control_cache - return KernelControlSnapshot(**control) + with self._lock: + payload = _decode_packet(self.control_region.as_buffer()) + control = payload.get("control") if isinstance(payload, dict) else None + if not isinstance(control, dict): + return self._control_cache + return KernelControlSnapshot(**control) def wait_on_state(self, timeout_ms: int = 1000) -> bool: return bool(self.state_region.wait(timeout_ms)) diff --git a/prod/clean_arch/dita_v2/test_account_core_v2.py b/prod/clean_arch/dita_v2/test_account_core_v2.py index 59b90597..2521a7c7 100644 --- a/prod/clean_arch/dita_v2/test_account_core_v2.py +++ b/prod/clean_arch/dita_v2/test_account_core_v2.py @@ -13,6 +13,7 @@ from __future__ import annotations import math import sys +from concurrent.futures import ThreadPoolExecutor sys.path.insert(0, "/mnt/dolphinng5_predict") import pytest @@ -324,7 +325,46 @@ class TestReplayDeterminism: # --------------------------------------------------------------------------- -# 7. V1 backward compatibility (AccountProjection must be untouched) +# 7. Concurrency guard +# --------------------------------------------------------------------------- + +class TestConcurrencyGuard: + def test_apply_fill_is_serialized(self): + proj = _proj(10_000.0) + n_threads = 16 + per_thread = 250 + total_fee = 0.0 + total_realized = 0.0 + + def _worker(tid: int) -> tuple[float, float]: + local_fee = 0.0 + local_realized = 0.0 + for i in range(per_thread): + realized = float(tid * per_thread + i) + fee = float((i % 5) * 0.1) + proj.apply_fill( + fill_price=100.0, + fill_qty=1.0, + fee=fee, + realized_pnl=realized, + ) + local_fee += fee + local_realized += realized + return local_realized, local_fee + + with ThreadPoolExecutor(max_workers=n_threads) as ex: + for realized, fee in ex.map(_worker, range(n_threads)): + total_realized += realized + total_fee += fee + + snap = _snap(proj) + assert snap.k.realized_pnl == pytest.approx(total_realized) + assert snap.k.fees_paid == pytest.approx(total_fee) + assert snap.k.capital == pytest.approx(10_000.0 + total_realized - total_fee) + + +# --------------------------------------------------------------------------- +# 8. V1 backward compatibility (AccountProjection must be untouched) # --------------------------------------------------------------------------- class TestV1Compat: