diff --git a/tests/adapters/portfolio/test_unmanaged_position_store.py b/tests/adapters/portfolio/test_unmanaged_position_store.py index 91d3e92..db69a38 100644 --- a/tests/adapters/portfolio/test_unmanaged_position_store.py +++ b/tests/adapters/portfolio/test_unmanaged_position_store.py @@ -161,13 +161,83 @@ class TestCRUDRoundTrip: audit = store.list_audit(instance_key="inst-A", symbol="008110") assert [row["event_type"] for row in audit] == ["detected"] - def test_concurrent_observe_and_resolve_leave_consistent_state_and_audit(self, pg_clean): + def test_older_absent_snapshot_applied_after_newer_observation_is_ignored(self, pg_unmanaged_store): + t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) + t2 = t1 + timedelta(minutes=1) + t3 = t2 + timedelta(minutes=1) + + pg_unmanaged_store.observe( + instance_key="inst-A", + symbol="008110", + quantity=3, + observed_at=t1, + ) + observed = pg_unmanaged_store.observe( + instance_key="inst-A", + symbol="008110", + quantity=5, + observed_at=t3, + ) + resolved = pg_unmanaged_store.resolve_absent( + instance_key="inst-A", + observed_symbols=set(), + observed_at=t2, + ) + + assert observed["transition"] == "quantity_changed" + assert resolved == [] + state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110") + assert state is not None + assert state["lifecycle_state"] == "active" + assert state["quantity"] == 5 + assert [row["event_type"] for row in pg_unmanaged_store.list_audit( + instance_key="inst-A", + symbol="008110", + )] == ["detected", "quantity_changed"] + + def test_older_observation_applied_after_newer_resolution_is_ignored(self, pg_unmanaged_store): + t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) + t2 = t1 + timedelta(minutes=1) + t3 = t2 + timedelta(minutes=1) + + pg_unmanaged_store.observe( + instance_key="inst-A", + symbol="008110", + quantity=3, + observed_at=t1, + ) + pg_unmanaged_store.resolve_absent( + instance_key="inst-A", + observed_symbols=set(), + observed_at=t3, + ) + stale = pg_unmanaged_store.observe( + instance_key="inst-A", + symbol="008110", + quantity=5, + observed_at=t2, + ) + + assert stale["transition"] == "stale" + state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110") + assert state is not None + assert state["lifecycle_state"] == "resolved" + assert state["quantity"] == 3 + assert [row["event_type"] for row in pg_unmanaged_store.list_audit( + instance_key="inst-A", + symbol="008110", + )] == ["detected", "resolved"] + + def test_concurrent_observe_and_resolve_respects_newest_snapshot_timestamp(self, pg_clean): from trex_core.adapters.portfolio.unmanaged_position_store import ( PostgreSQLUnmanagedPositionStore, ) store = PostgreSQLUnmanagedPositionStore(pg_clean) - store.observe(instance_key="inst-A", symbol="008110", quantity=3) + t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) + t2 = t1 + timedelta(minutes=1) + t3 = t2 + timedelta(minutes=1) + store.observe(instance_key="inst-A", symbol="008110", quantity=3, observed_at=t1) with ThreadPoolExecutor(max_workers=2) as pool: observe_future = pool.submit( @@ -175,32 +245,24 @@ class TestCRUDRoundTrip: instance_key="inst-A", symbol="008110", quantity=5, + observed_at=t3, ) resolve_future = pool.submit( PostgreSQLUnmanagedPositionStore(pg_clean).resolve_absent, instance_key="inst-A", observed_symbols=set(), + observed_at=t2, ) observe_result = observe_future.result() resolve_result = resolve_future.result() state = store.get_state(instance_key="inst-A", symbol="008110") assert state is not None - audit = store.list_audit(instance_key="inst-A", symbol="008110") - event_types = [row["event_type"] for row in audit] - assert event_types[0] == "detected" - assert event_types in ( - ["detected", "quantity_changed", "resolved"], - ["detected", "resolved", "detected"], - ) - if state["lifecycle_state"] == "resolved": - assert observe_result["transition"] == "quantity_changed" - assert len(resolve_result) == 1 - assert state["quantity"] == 5 - assert event_types[-1] == "resolved" - else: - assert state["lifecycle_state"] == "active" - assert observe_result["transition"] == "detected" - assert len(resolve_result) == 1 - assert state["quantity"] == 5 - assert event_types[-1] == "detected" + assert observe_result["transition"] == "quantity_changed" + assert resolve_result == [] + assert state["lifecycle_state"] == "active" + assert state["quantity"] == 5 + assert [row["event_type"] for row in store.list_audit( + instance_key="inst-A", + symbol="008110", + )] == ["detected", "quantity_changed"] diff --git a/tests/application/test_unmanaged_position_lifecycle.py b/tests/application/test_unmanaged_position_lifecycle.py index a6de1fd..3fe04a0 100644 --- a/tests/application/test_unmanaged_position_lifecycle.py +++ b/tests/application/test_unmanaged_position_lifecycle.py @@ -140,6 +140,49 @@ def test_in_memory_store_preserves_first_seen_when_resolved_symbol_reappears(): assert reopened["state"]["evidence_ref"] is None +def test_in_memory_store_ignores_stale_resolve_after_newer_observation(): + store = InMemoryUnmanagedPositionStore() + t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) + t2 = t1 + timedelta(minutes=1) + t3 = t2 + timedelta(minutes=1) + + store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1) + store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t3) + resolved = store.resolve_absent( + instance_key=INSTANCE, + observed_symbols=set(), + observed_at=t2, + ) + + assert resolved == [] + state = store.get_state(instance_key=INSTANCE, symbol="008110") + assert state is not None + assert state["lifecycle_state"] == "active" + assert state["quantity"] == 5 + assert store.list_audit(instance_key=INSTANCE, symbol="008110")[-1]["event_type"] == "quantity_changed" + + +def test_in_memory_store_ignores_stale_observation_after_newer_resolution(): + store = InMemoryUnmanagedPositionStore() + t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC) + t2 = t1 + timedelta(minutes=1) + t3 = t2 + timedelta(minutes=1) + + store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1) + store.resolve_absent(instance_key=INSTANCE, observed_symbols=set(), observed_at=t3) + stale = store.observe(instance_key=INSTANCE, symbol="008110", quantity=5, observed_at=t2) + + assert stale["transition"] == "stale" + state = store.get_state(instance_key=INSTANCE, symbol="008110") + assert state is not None + assert state["lifecycle_state"] == "resolved" + assert state["quantity"] == 3 + assert [row["event_type"] for row in store.list_audit(instance_key=INSTANCE, symbol="008110")] == [ + "detected", + "resolved", + ] + + def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam(): repo, svc = _wire() emitter, transport = _make_emitter() diff --git a/trex_core/adapters/portfolio/unmanaged_position_store.py b/trex_core/adapters/portfolio/unmanaged_position_store.py index f2f2f38..42ae613 100644 --- a/trex_core/adapters/portfolio/unmanaged_position_store.py +++ b/trex_core/adapters/portfolio/unmanaged_position_store.py @@ -21,6 +21,18 @@ class PostgreSQLUnmanagedPositionStore: self._psycopg = _require_psycopg(psycopg_module) self._dsn = dsn + @staticmethod + def _latest_snapshot_at(state: dict) -> datetime: + resolved_at = state.get("resolved_at") + last_seen_at = state.get("last_seen_at") + if isinstance(resolved_at, datetime) and ( + not isinstance(last_seen_at, datetime) or resolved_at >= last_seen_at + ): + return resolved_at + if isinstance(last_seen_at, datetime): + return last_seen_at + return state.get("updated_at") or state.get("created_at") or datetime.min.replace(tzinfo=UTC) + def observe( self, *, @@ -43,6 +55,13 @@ class PostgreSQLUnmanagedPositionStore: row = cur.fetchone() previous = self._row_to_state(row) if row is not None else None previous_quantity = previous.get("quantity") if previous is not None else None + if previous is not None and now < self._latest_snapshot_at(previous): + conn.rollback() + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": previous, + } if previous is None: transition = "detected" cur.execute( @@ -54,6 +73,13 @@ class PostgreSQLUnmanagedPositionStore: (instance_key, symbol, quantity, now, now, now, now, now), ) elif previous["lifecycle_state"] == "resolved": + if now <= self._latest_snapshot_at(previous): + conn.rollback() + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": previous, + } transition = "detected" cur.execute( "UPDATE unmanaged_position_state " @@ -63,29 +89,58 @@ class PostgreSQLUnmanagedPositionStore: " operator_actor = NULL, operator_updated_at = NULL, " " updated_at = %s " "WHERE instance_key = %s AND symbol = %s " + " AND lifecycle_state = 'resolved' AND resolved_at < %s " f"RETURNING {_STATE_COLUMNS}", - (quantity, now, now, now, instance_key, symbol), + (quantity, now, now, now, instance_key, symbol, now), ) elif int(previous["quantity"]) != int(quantity): + if now <= previous["last_seen_at"]: + conn.rollback() + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": previous, + } transition = "quantity_changed" cur.execute( "UPDATE unmanaged_position_state " "SET quantity = %s, last_seen_at = %s, quantity_changed_at = %s, " " resolved_at = NULL, lifecycle_state = 'active', updated_at = %s " "WHERE instance_key = %s AND symbol = %s " + " AND lifecycle_state = 'active' AND last_seen_at < %s " f"RETURNING {_STATE_COLUMNS}", - (quantity, now, now, now, instance_key, symbol), + (quantity, now, now, now, instance_key, symbol, now), ) else: transition = "unchanged" - cur.execute( - "UPDATE unmanaged_position_state " - "SET last_seen_at = %s, updated_at = %s " - "WHERE instance_key = %s AND symbol = %s " - f"RETURNING {_STATE_COLUMNS}", - (now, now, instance_key, symbol), - ) - state = self._row_to_state(cur.fetchone()) + if now > previous["last_seen_at"]: + cur.execute( + "UPDATE unmanaged_position_state " + "SET last_seen_at = %s, updated_at = %s " + "WHERE instance_key = %s AND symbol = %s " + " AND lifecycle_state = 'active' AND quantity = %s AND last_seen_at < %s " + f"RETURNING {_STATE_COLUMNS}", + (now, now, instance_key, symbol, quantity, now), + ) + fetched = cur.fetchone() + state = previous if fetched is None else self._row_to_state(fetched) + else: + conn.rollback() + return { + "transition": "unchanged", + "previous_quantity": previous_quantity, + "state": previous, + } + if transition != "unchanged": + fetched = cur.fetchone() + if fetched is None: + conn.rollback() + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": previous, + } + state = self._row_to_state(fetched) if transition != "unchanged": note = ( f"previous_quantity={previous_quantity}" @@ -139,14 +194,20 @@ class PostgreSQLUnmanagedPositionStore: for state in active: if state["symbol"] in observed_symbols: continue + if now <= state["last_seen_at"]: + continue cur.execute( "UPDATE unmanaged_position_state " "SET lifecycle_state = 'resolved', resolved_at = %s, updated_at = %s " "WHERE instance_key = %s AND symbol = %s " + " AND lifecycle_state = 'active' AND last_seen_at < %s " f"RETURNING {_STATE_COLUMNS}", - (now, now, instance_key, state["symbol"]), + (now, now, instance_key, state["symbol"], now), ) - resolved_state = self._row_to_state(cur.fetchone()) + row = cur.fetchone() + if row is None: + continue + resolved_state = self._row_to_state(row) resolved.append(resolved_state) self._insert_audit( cur, diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index ac5c985..729f651 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -625,6 +625,18 @@ class InMemoryUnmanagedPositionStore: self._states: dict[tuple[str, str], dict] = {} self._audit: list[dict] = [] + @staticmethod + def _latest_snapshot_at(state: dict) -> datetime: + resolved_at = state.get("resolved_at") + last_seen_at = state.get("last_seen_at") + if isinstance(resolved_at, datetime) and ( + not isinstance(last_seen_at, datetime) or resolved_at >= last_seen_at + ): + return resolved_at + if isinstance(last_seen_at, datetime): + return last_seen_at + return state.get("updated_at") or state.get("created_at") or datetime.min.replace(tzinfo=UTC) + def observe( self, *, @@ -642,7 +654,21 @@ class InMemoryUnmanagedPositionStore: preserved = {} if existing is None else { "created_at": existing.get("created_at", now), } + if existing is not None: + latest_snapshot_at = self._latest_snapshot_at(existing) + if now < latest_snapshot_at: + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": dict(existing), + } if existing is None or existing.get("lifecycle_state") == "resolved": + if existing is not None and now == self._latest_snapshot_at(existing): + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": dict(existing), + } transition = "detected" state = { "instance_key": instance_key, @@ -666,6 +692,12 @@ class InMemoryUnmanagedPositionStore: } self._states[key] = state elif int(existing["quantity"]) != int(quantity): + if now <= existing["last_seen_at"]: + return { + "transition": "stale", + "previous_quantity": previous_quantity, + "state": dict(existing), + } transition = "quantity_changed" existing["quantity"] = quantity existing["last_seen_at"] = now @@ -676,8 +708,9 @@ class InMemoryUnmanagedPositionStore: state = existing else: transition = "unchanged" - existing["last_seen_at"] = now - existing["updated_at"] = now + if now > existing["last_seen_at"]: + existing["last_seen_at"] = now + existing["updated_at"] = now state = existing if transition != "unchanged": self._audit.append( @@ -722,6 +755,8 @@ class InMemoryUnmanagedPositionStore: continue if state["symbol"] in observed_symbols: continue + if now <= state["last_seen_at"]: + continue state["lifecycle_state"] = "resolved" state["resolved_at"] = now state["updated_at"] = now diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index f117e35..e4edf8c 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -29,6 +29,7 @@ import hashlib import logging import math import re +from datetime import UTC, datetime from typing import cast from trex_core.application.alarm_emitter import AlarmEmitter @@ -407,6 +408,7 @@ def reconcile_core_holdings( Returns {key: 'error'|'qty_warning'|'ok'|'unmanaged_position'| 'managed_non_holding_divergence'} for telemetry. """ + snapshot_observed_at = datetime.now(tz=UTC) broker_qty: dict[str, int] = { item.symbol: int(item.hldg_qty) for item in broker.get_balance() } @@ -507,6 +509,7 @@ def reconcile_core_holdings( instance_key=instance_key, symbol=sym, quantity=bq, + observed_at=snapshot_observed_at, ) except Exception: # noqa: BLE001 _log.exception( @@ -555,6 +558,7 @@ def reconcile_core_holdings( for state in unmanaged_position_store.resolve_absent( instance_key=instance_key, observed_symbols=active_unmanaged_symbols, + observed_at=snapshot_observed_at, ): _log.info( "unmanaged position resolved: symbol=%s qty=%d instance_key=%s",