trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: persist unmanaged position lifecycle state

Browse Source 
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-09-12 13:24:51 +00:00
committed by GitHub
co-authored by genichin
parent ae2e598209
commit fe51d211d9
12 changed files with 1274 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapters/portfolio/test_unmanaged_position_store.py
+101
View File
@@ -0,0 +1,101 @@
"""PostgreSQLUnmanagedPositionStore integration tests."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import pytest
@pytest.fixture
def pg_unmanaged_store(pg_clean):
from trex_core.adapters.portfolio.unmanaged_position_store import (
PostgreSQLUnmanagedPositionStore,
)
return PostgreSQLUnmanagedPositionStore(pg_clean)
class TestMigration028:
def test_tables_exist_after_migration(self, pg_clean):
import psycopg
with psycopg.connect(pg_clean) as conn, conn.cursor() as cur:
cur.execute(
"SELECT table_name FROM information_schema.tables "
"WHERE table_schema = 'public' "
"AND table_name IN ('unmanaged_position_state', 'unmanaged_position_audit')"
)
assert {row[0] for row in cur.fetchall()} == {
"unmanaged_position_state",
"unmanaged_position_audit",
}
class TestCRUDRoundTrip:
def test_observe_change_resolve_and_audit(self, pg_unmanaged_store):
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
t2 = t1 + timedelta(minutes=1)
t3 = t2 + timedelta(minutes=1)
first = pg_unmanaged_store.observe(
instance_key="inst-A",
symbol="008110",
quantity=3,
observed_at=t1,
)
repeat = pg_unmanaged_store.observe(
instance_key="inst-A",
symbol="008110",
quantity=3,
observed_at=t2,
)
changed = pg_unmanaged_store.observe(
instance_key="inst-A",
symbol="008110",
quantity=5,
observed_at=t3,
)
recorded = pg_unmanaged_store.record_operator_action(
instance_key="inst-A",
symbol="008110",
action="approved_exception",
evidence_ref="kis:2026-09-11",
actor="cli",
note="broker-confirmed",
)
resolved = pg_unmanaged_store.resolve_absent(
instance_key="inst-A",
observed_symbols=set(),
observed_at=t3 + timedelta(minutes=1),
)
assert first["transition"] == "detected"
assert repeat["transition"] == "unchanged"
assert changed["transition"] == "quantity_changed"
assert recorded is not None
assert len(resolved) == 1
state = pg_unmanaged_store.get_state(instance_key="inst-A", symbol="008110")
assert state is not None
assert state["quantity"] == 5
assert state["first_seen_at"] == t1
assert state["last_seen_at"] == t3
assert state["quantity_changed_at"] == t3
assert state["lifecycle_state"] == "resolved"
assert state["evidence_ref"] == "kis:2026-09-11"
audit = pg_unmanaged_store.list_audit(instance_key="inst-A", symbol="008110")
assert [row["event_type"] for row in audit] == [
"detected",
"quantity_changed",
"operator_action",
"resolved",
]
def test_same_symbol_isolated_by_instance(self, pg_unmanaged_store):
pg_unmanaged_store.observe(instance_key="inst-A", symbol="008110", quantity=3)
pg_unmanaged_store.observe(instance_key="inst-B", symbol="008110", quantity=7)
pg_unmanaged_store.resolve_absent(instance_key="inst-A", observed_symbols=set())
rows_a = pg_unmanaged_store.list_states(instance_key="inst-A", lifecycle_state="all")
rows_b = pg_unmanaged_store.list_states(instance_key="inst-B", lifecycle_state="all")
assert len(rows_a) == 1 and rows_a[0]["lifecycle_state"] == "resolved"
assert len(rows_b) == 1 and rows_b[0]["quantity"] == 7
tests/application/test_unmanaged_position_lifecycle.py
+190
View File
@@ -0,0 +1,190 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import pytest
from trex_core.adapters.messaging.in_memory import InMemoryTransport
from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
from trex_core.application.portfolio import InMemoryUnmanagedPositionStore
from trex_core.application.reconciliation import reconcile_core_holdings
from trex_core.domain.slot import SlotService
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
INSTANCE = "inst-1"
class FakeBroker:
def __init__(self, balances):
self._balances = balances
def name(self):
return "fake"
def submit_order(self, req): ...
def poll_fill(self, *args, **kwargs): ...
def cancel_order(self, *args, **kwargs): ...
def get_balance(self):
return [BalanceItem(symbol=symbol, hldg_qty=qty) for symbol, qty in self._balances.items()]
class RaisingBroker(FakeBroker):
def get_balance(self):
raise BrokerBalanceError("broker_unavailable")
def _wire():
repo = InMemorySlotRepository()
return repo, SlotService(repo, instance_key=INSTANCE)
def _make_emitter():
transport = InMemoryTransport()
return AlarmEmitter(transport), transport
def test_in_memory_store_tracks_lifecycle_and_operator_audit():
store = InMemoryUnmanagedPositionStore()
t1 = datetime(2026, 9, 7, 0, 0, tzinfo=UTC)
t2 = t1 + timedelta(minutes=1)
t3 = t2 + timedelta(minutes=1)
first = store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t1)
repeat = store.observe(instance_key=INSTANCE, symbol="008110", quantity=3, observed_at=t2)
changed = store.observe(instance_key=INSTANCE, symbol="008110", quantity=4, observed_at=t3)
recorded = store.record_operator_action(
instance_key=INSTANCE,
symbol="008110",
action="investigating",
evidence_ref="kis-balance:2026-09-07",
note="confirmed at broker",
)
resolved = store.resolve_absent(
instance_key=INSTANCE,
observed_symbols=set(),
observed_at=t3 + timedelta(minutes=1),
)
assert first["transition"] == "detected"
assert repeat["transition"] == "unchanged"
assert changed["transition"] == "quantity_changed"
assert recorded is not None
assert recorded["evidence_ref"] == "kis-balance:2026-09-07"
assert len(resolved) == 1
state = store.get_state(instance_key=INSTANCE, symbol="008110")
assert state is not None
assert state["quantity"] == 4
assert state["first_seen_at"] == t1
assert state["last_seen_at"] == t3
assert state["quantity_changed_at"] == t3
assert state["lifecycle_state"] == "resolved"
assert state["resolved_at"] == t3 + timedelta(minutes=1)
audit = store.list_audit(instance_key=INSTANCE, symbol="008110")
assert [row["event_type"] for row in audit] == [
"detected",
"quantity_changed",
"operator_action",
"resolved",
]
def test_in_memory_store_isolates_same_symbol_by_instance():
store = InMemoryUnmanagedPositionStore()
store.observe(instance_key="inst-A", symbol="008110", quantity=3)
store.observe(instance_key="inst-B", symbol="008110", quantity=7)
store.resolve_absent(instance_key="inst-A", observed_symbols=set())
state_a = store.get_state(instance_key="inst-A", symbol="008110")
state_b = store.get_state(instance_key="inst-B", symbol="008110")
assert state_a is not None and state_a["lifecycle_state"] == "resolved"
assert state_b is not None and state_b["lifecycle_state"] == "active"
assert state_b["quantity"] == 7
def test_reconcile_core_holdings_suppresses_repeated_unmanaged_alarm_spam():
repo, svc = _wire()
emitter, transport = _make_emitter()
store = InMemoryUnmanagedPositionStore()
for _ in range(3):
decisions = reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({"008110": 3}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
)
assert decisions == {"broker:008110": "unmanaged_position"}
state = store.get_state(instance_key=INSTANCE, symbol="008110")
assert state is not None
assert state["quantity"] == 3
codes = [payload["code"] for _, payload in transport.alarm_events()]
assert codes == ["unmanaged_position_detected"]
audit = store.list_audit(instance_key=INSTANCE, symbol="008110")
assert [row["event_type"] for row in audit] == ["detected"]
def test_reconcile_core_holdings_tracks_change_and_resolution():
repo, svc = _wire()
emitter, transport = _make_emitter()
store = InMemoryUnmanagedPositionStore()
reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({"008110": 3}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
)
reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({"008110": 5}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
)
reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=FakeBroker({}),
instance_key=INSTANCE,
alarm_emitter=emitter,
unmanaged_position_store=store,
)
state = store.get_state(instance_key=INSTANCE, symbol="008110")
assert state is not None
assert state["quantity"] == 5
assert state["lifecycle_state"] == "resolved"
codes = [payload["code"] for _, payload in transport.alarm_events()]
assert codes == [
"unmanaged_position_detected",
"unmanaged_position_changed",
"unmanaged_position_resolved",
]
def test_broker_failure_does_not_resolve_unmanaged_state():
repo, svc = _wire()
store = InMemoryUnmanagedPositionStore()
store.observe(instance_key=INSTANCE, symbol="008110", quantity=3)
with pytest.raises(BrokerBalanceError):
reconcile_core_holdings(
slot_repo=repo,
slot_service=svc,
broker=RaisingBroker({}),
instance_key=INSTANCE,
unmanaged_position_store=store,
)
state = store.get_state(instance_key=INSTANCE, symbol="008110")
assert state is not None
assert state["lifecycle_state"] == "active"
tests/conftest.py
+2
View File
@@ -28,6 +28,8 @@ _PG_TABLES = (
"daily_slot",
"reconcile_notification_state",
"slot_strategy_state",
"unmanaged_position_audit",
"unmanaged_position_state",
)
tests/runtime/test_cli_operator.py
+169
View File
@@ -516,6 +516,175 @@ def test_verify_no_db_no_live_flag_needed(capsys, monkeypatch):
assert rc == 0
# ---------------------------------------------------------------------------
# unmanaged parser/tests
# ---------------------------------------------------------------------------
def test_unmanaged_parser_accepts_list_and_record():
parser = cli._build_parser()
args = parser.parse_args(["unmanaged", "list", "--status", "all", "--symbol", "008110"])
assert args.cmd == "unmanaged"
assert args.unmanaged_cmd == "list"
assert args.status == "all"
args = parser.parse_args(
[
"unmanaged",
"record",
"008110",
"investigating",
"--evidence-ref",
"kis-balance:2026-09-11",
]
)
assert args.unmanaged_cmd == "record"
assert args.action == "investigating"
def test_unmanaged_list_outputs_durable_state(monkeypatch, capsys):
class _Store:
def __init__(self, *_args, **_kwargs):
pass
def list_states(self, **kwargs):
assert kwargs["instance_key"] == "inst-1"
assert kwargs["lifecycle_state"] == "all"
return [
{
"instance_key": "inst-1",
"symbol": "008110",
"quantity": 3,
"lifecycle_state": "active",
"first_seen_at": "2026-09-07T00:00:00+00:00",
"last_seen_at": "2026-09-11T00:00:00+00:00",
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
"resolved_at": None,
"operator_action": "investigating",
"evidence_ref": "kis-balance:2026-09-11",
"operator_note": "confirmed",
"operator_actor": "operator",
"operator_updated_at": "2026-09-11T00:00:00+00:00",
"created_at": "2026-09-07T00:00:00+00:00",
"updated_at": "2026-09-11T00:00:00+00:00",
}
]
monkeypatch.setattr(
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
_Store,
)
args = SimpleNamespace(
cmd="unmanaged",
unmanaged_cmd="list",
status="all",
symbol=None,
instance="inst-1",
config=None,
config_path=None,
env_file=None,
json=True,
timeout=10.0,
)
rc = cli._cmd_unmanaged_list(args, SimpleNamespace(dsn="postgresql://x/y"))
assert rc == 0
out = json.loads(capsys.readouterr().out)
assert out["count"] == 1
assert out["states"][0]["symbol"] == "008110"
assert out["states"][0]["evidence_ref"] == "kis-balance:2026-09-11"
def test_unmanaged_record_requires_existing_state(monkeypatch, capsys):
class _Store:
def __init__(self, *_args, **_kwargs):
pass
def record_operator_action(self, **_kwargs):
return None
monkeypatch.setattr(
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
_Store,
)
args = SimpleNamespace(
cmd="unmanaged",
unmanaged_cmd="record",
symbol="008110",
action="approved_exception",
evidence_ref="kis-balance:2026-09-11",
note="confirmed",
actor="operator",
instance="inst-1",
config=None,
config_path=None,
env_file=None,
json=True,
timeout=10.0,
)
rc = cli._cmd_unmanaged_record(args, SimpleNamespace(dsn="postgresql://x/y"))
assert rc == 1
err = json.loads(capsys.readouterr().err)
assert "not found" in err["error"]
def test_unmanaged_record_persists_operator_action(monkeypatch, capsys):
class _Store:
def __init__(self, *_args, **_kwargs):
pass
def record_operator_action(self, **kwargs):
assert kwargs["instance_key"] == "inst-1"
assert kwargs["symbol"] == "008110"
assert kwargs["action"] == "managed_inclusion"
return {
"instance_key": "inst-1",
"symbol": "008110",
"quantity": 3,
"lifecycle_state": "active",
"first_seen_at": "2026-09-07T00:00:00+00:00",
"last_seen_at": "2026-09-11T00:00:00+00:00",
"quantity_changed_at": "2026-09-07T00:00:00+00:00",
"resolved_at": None,
"operator_action": "managed_inclusion",
"evidence_ref": kwargs["evidence_ref"],
"operator_note": kwargs["note"],
"operator_actor": kwargs["actor"],
"operator_updated_at": "2026-09-11T00:00:00+00:00",
"created_at": "2026-09-07T00:00:00+00:00",
"updated_at": "2026-09-11T00:00:00+00:00",
}
monkeypatch.setattr(
"trex_core.adapters.portfolio.unmanaged_position_store.PostgreSQLUnmanagedPositionStore",
_Store,
)
args = SimpleNamespace(
cmd="unmanaged",
unmanaged_cmd="record",
symbol="008110",
action="managed_inclusion",
evidence_ref="kis-balance:2026-09-11",
note="manual inclusion planned",
actor="operator",
instance="inst-1",
config=None,
config_path=None,
env_file=None,
json=True,
timeout=10.0,
)
rc = cli._cmd_unmanaged_record(args, SimpleNamespace(dsn="postgresql://x/y"))
assert rc == 0
out = json.loads(capsys.readouterr().out)
assert out["result"] == "recorded"
assert out["state"]["operator_action"] == "managed_inclusion"
assert out["state"]["evidence_ref"] == "kis-balance:2026-09-11"
# ---------------------------------------------------------------------------
# REL-010 regression: _cmd_verify broker import must not raise ImportError
# ---------------------------------------------------------------------------
trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql
+38
View File
@@ -0,0 +1,38 @@
CREATE TABLE IF NOT EXISTS unmanaged_position_state (
instance_key TEXT NOT NULL,
symbol TEXT NOT NULL,
quantity INTEGER NOT NULL,
lifecycle_state TEXT NOT NULL,
first_seen_at TIMESTAMPTZ NOT NULL,
last_seen_at TIMESTAMPTZ NOT NULL,
quantity_changed_at TIMESTAMPTZ NOT NULL,
resolved_at TIMESTAMPTZ NULL,
operator_action TEXT NULL,
evidence_ref TEXT NULL,
operator_note TEXT NOT NULL DEFAULT '',
operator_actor TEXT NULL,
operator_updated_at TIMESTAMPTZ NULL,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (instance_key, symbol)
);
CREATE INDEX IF NOT EXISTS idx_unmanaged_position_state_lifecycle
ON unmanaged_position_state (instance_key, lifecycle_state, updated_at DESC);
CREATE TABLE IF NOT EXISTS unmanaged_position_audit (
audit_id BIGSERIAL PRIMARY KEY,
instance_key TEXT NOT NULL,
symbol TEXT NOT NULL,
event_type TEXT NOT NULL,
action TEXT NULL,
quantity INTEGER NOT NULL,
lifecycle_state TEXT NOT NULL,
actor TEXT NOT NULL,
evidence_ref TEXT NULL,
note TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_unmanaged_position_audit_lookup
ON unmanaged_position_audit (instance_key, symbol, audit_id);
trex_core/adapters/portfolio/__init__.py
+2
View File
@@ -3,10 +3,12 @@ from trex_core.adapters.portfolio.hub_read import PortfolioHubReadAdapter
from trex_core.adapters.portfolio.pause_store import PostgreSQLPauseStore
from trex_core.adapters.portfolio.reconcile_warning_store import PostgreSQLReconcileWarningStore
from trex_core.adapters.portfolio.stub_read import StubPortfolioRead
from trex_core.adapters.portfolio.unmanaged_position_store import PostgreSQLUnmanagedPositionStore
__all__ = [
"PortfolioHubReadAdapter",
"PostgreSQLPauseStore",
"PostgreSQLReconcileWarningStore",
"PostgreSQLUnmanagedPositionStore",
"StubPortfolioRead",
]
trex_core/adapters/portfolio/unmanaged_position_store.py
+309
View File
@@ -0,0 +1,309 @@
"""PostgreSQL-backed unmanaged-position lifecycle store."""
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from trex_core.adapters.persistence.postgresql import _require_psycopg
_STATE_COLUMNS = (
"instance_key, symbol, quantity, lifecycle_state, "
"first_seen_at, last_seen_at, quantity_changed_at, resolved_at, "
"operator_action, evidence_ref, operator_note, operator_actor, "
"operator_updated_at, created_at, updated_at"
)
class PostgreSQLUnmanagedPositionStore:
"""CRUD + audit for unmanaged positions keyed by (instance_key, symbol)."""
def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None:
self._psycopg = _require_psycopg(psycopg_module)
self._dsn = dsn
def observe(
self,
*,
instance_key: str,
symbol: str,
quantity: int,
observed_at: datetime | None = None,
actor: str = "reconcile",
) -> dict:
now = observed_at or datetime.now(tz=UTC)
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
cur.execute(
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
"WHERE instance_key = %s AND symbol = %s "
"FOR UPDATE",
(instance_key, symbol),
)
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 None:
transition = "detected"
cur.execute(
"INSERT INTO unmanaged_position_state "
f"({_STATE_COLUMNS}) "
"VALUES (%s, %s, %s, 'active', %s, %s, %s, NULL, "
" NULL, NULL, '', NULL, NULL, %s, %s) "
f"RETURNING {_STATE_COLUMNS}",
(instance_key, symbol, quantity, now, now, now, now, now),
)
elif previous["lifecycle_state"] == "resolved":
transition = "detected"
cur.execute(
"UPDATE unmanaged_position_state "
"SET quantity = %s, lifecycle_state = 'active', "
" first_seen_at = %s, last_seen_at = %s, quantity_changed_at = %s, "
" resolved_at = NULL, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(quantity, now, now, now, now, instance_key, symbol),
)
elif int(previous["quantity"]) != int(quantity):
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 "
f"RETURNING {_STATE_COLUMNS}",
(quantity, now, now, now, instance_key, symbol),
)
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 transition != "unchanged":
note = (
f"previous_quantity={previous_quantity}"
if transition == "quantity_changed" and previous_quantity is not None
else ""
)
self._insert_audit(
cur,
instance_key=instance_key,
symbol=symbol,
event_type=transition,
action=None,
quantity=quantity,
lifecycle_state=state["lifecycle_state"],
actor=actor,
evidence_ref=state.get("evidence_ref"),
note=note,
created_at=now,
)
conn.commit()
return {
"transition": transition,
"previous_quantity": previous_quantity,
"state": state,
}
def resolve_absent(
self,
*,
instance_key: str,
observed_symbols: set[str],
observed_at: datetime | None = None,
actor: str = "reconcile",
) -> list[dict]:
now = observed_at or datetime.now(tz=UTC)
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
cur.execute(
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
"WHERE instance_key = %s AND lifecycle_state = 'active' "
"ORDER BY symbol "
"FOR UPDATE",
(instance_key,),
)
active = [self._row_to_state(row) for row in cur.fetchall()]
resolved: list[dict] = []
for state in active:
if state["symbol"] in observed_symbols:
continue
cur.execute(
"UPDATE unmanaged_position_state "
"SET lifecycle_state = 'resolved', resolved_at = %s, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(now, now, instance_key, state["symbol"]),
)
resolved_state = self._row_to_state(cur.fetchone())
resolved.append(resolved_state)
self._insert_audit(
cur,
instance_key=instance_key,
symbol=state["symbol"],
event_type="resolved",
action=None,
quantity=state["quantity"],
lifecycle_state="resolved",
actor=actor,
evidence_ref=resolved_state.get("evidence_ref"),
note="",
created_at=now,
)
conn.commit()
return resolved
def get_state(self, *, instance_key: str, symbol: str) -> dict | None:
with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
cur.execute(
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
"WHERE instance_key = %s AND symbol = %s "
"LIMIT 1",
(instance_key, symbol),
)
row = cur.fetchone()
return None if row is None else self._row_to_state(row)
def list_states(
self,
*,
instance_key: str,
lifecycle_state: str = "active",
symbol: str | None = None,
) -> list[dict]:
clauses = ["instance_key = %s"]
params: list[object] = [instance_key]
if lifecycle_state != "all":
clauses.append("lifecycle_state = %s")
params.append(lifecycle_state)
if symbol is not None:
clauses.append("symbol = %s")
params.append(symbol)
with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
cur.execute(
f"SELECT {_STATE_COLUMNS} FROM unmanaged_position_state "
f"WHERE {' AND '.join(clauses)} ORDER BY symbol",
tuple(params),
)
return [self._row_to_state(row) for row in cur.fetchall()]
def record_operator_action(
self,
*,
instance_key: str,
symbol: str,
action: str,
evidence_ref: str,
actor: str = "operator",
note: str = "",
) -> dict | None:
now = datetime.now(tz=UTC)
with self._psycopg.connect(self._dsn, autocommit=False) as conn, conn.cursor() as cur:
cur.execute(
"UPDATE unmanaged_position_state "
"SET operator_action = %s, evidence_ref = %s, operator_note = %s, "
" operator_actor = %s, operator_updated_at = %s, updated_at = %s "
"WHERE instance_key = %s AND symbol = %s "
f"RETURNING {_STATE_COLUMNS}",
(action, evidence_ref, note, actor, now, now, instance_key, symbol),
)
row = cur.fetchone()
if row is None:
conn.rollback()
return None
state = self._row_to_state(row)
self._insert_audit(
cur,
instance_key=instance_key,
symbol=symbol,
event_type="operator_action",
action=action,
quantity=state["quantity"],
lifecycle_state=state["lifecycle_state"],
actor=actor,
evidence_ref=evidence_ref,
note=note,
created_at=now,
)
conn.commit()
return state
def list_audit(self, *, instance_key: str, symbol: str) -> list[dict]:
with self._psycopg.connect(self._dsn) as conn, conn.cursor() as cur:
cur.execute(
"SELECT event_type, action, quantity, lifecycle_state, actor, "
" evidence_ref, note, created_at "
"FROM unmanaged_position_audit "
"WHERE instance_key = %s AND symbol = %s "
"ORDER BY audit_id",
(instance_key, symbol),
)
return [
{
"event_type": row[0],
"action": row[1],
"quantity": row[2],
"lifecycle_state": row[3],
"actor": row[4],
"evidence_ref": row[5],
"note": row[6],
"created_at": row[7],
}
for row in cur.fetchall()
]
def _insert_audit(
self,
cur,
*,
instance_key: str,
symbol: str,
event_type: str,
action: str | None,
quantity: int,
lifecycle_state: str,
actor: str,
evidence_ref: str | None,
note: str,
created_at: datetime,
) -> None:
cur.execute(
"INSERT INTO unmanaged_position_audit "
"(instance_key, symbol, event_type, action, quantity, lifecycle_state, "
" actor, evidence_ref, note, created_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
instance_key,
symbol,
event_type,
action,
quantity,
lifecycle_state,
actor,
evidence_ref,
note,
created_at,
),
)
@staticmethod
def _row_to_state(row) -> dict:
return {
"instance_key": row[0],
"symbol": row[1],
"quantity": row[2],
"lifecycle_state": row[3],
"first_seen_at": row[4],
"last_seen_at": row[5],
"quantity_changed_at": row[6],
"resolved_at": row[7],
"operator_action": row[8],
"evidence_ref": row[9],
"operator_note": row[10],
"operator_actor": row[11],
"operator_updated_at": row[12],
"created_at": row[13],
"updated_at": row[14],
}
trex_core/application/portfolio.py
+253 -1
View File
@@ -537,6 +537,259 @@ class InMemoryReconcileNotificationStore:
]
class UnmanagedPositionStore(Protocol):
"""Durable unmanaged-position lifecycle store keyed by (instance_key, symbol)."""
def observe(
self,
*,
instance_key: str,
symbol: str,
quantity: int,
observed_at: datetime | None = None,
actor: str = "reconcile",
) -> dict:
"""Return {"transition", "previous_quantity", "state"} for one observation."""
...
def resolve_absent(
self,
*,
instance_key: str,
observed_symbols: set[str],
observed_at: datetime | None = None,
actor: str = "reconcile",
) -> list[dict]:
"""Mark active positions absent from a successful broker read as resolved."""
...
def get_state(self, *, instance_key: str, symbol: str) -> dict | None:
...
def list_states(
self,
*,
instance_key: str,
lifecycle_state: str = "active",
symbol: str | None = None,
) -> list[dict]:
...
def record_operator_action(
self,
*,
instance_key: str,
symbol: str,
action: str,
evidence_ref: str,
actor: str = "operator",
note: str = "",
) -> dict | None:
"""Persist an explicit operator disposition and append an audit event."""
...
def list_audit(self, *, instance_key: str, symbol: str) -> list[dict]:
...
class InMemoryUnmanagedPositionStore:
"""Process-local unmanaged-position lifecycle store for tests + CI."""
def __init__(self) -> None:
self._lock = threading.Lock()
self._states: dict[tuple[str, str], dict] = {}
self._audit: list[dict] = []
def observe(
self,
*,
instance_key: str,
symbol: str,
quantity: int,
observed_at: datetime | None = None,
actor: str = "reconcile",
) -> dict:
now = observed_at or datetime.now(tz=UTC)
key = (instance_key, symbol)
with self._lock:
existing = self._states.get(key)
previous_quantity = existing.get("quantity") if existing is not None else None
preserved = {} if existing is None else {
"operator_action": existing.get("operator_action"),
"evidence_ref": existing.get("evidence_ref"),
"operator_note": existing.get("operator_note"),
"operator_actor": existing.get("operator_actor"),
"operator_updated_at": existing.get("operator_updated_at"),
"created_at": existing.get("created_at", now),
}
if existing is None or existing.get("lifecycle_state") == "resolved":
transition = "detected"
state = {
"instance_key": instance_key,
"symbol": symbol,
"quantity": quantity,
"lifecycle_state": "active",
"first_seen_at": now,
"last_seen_at": now,
"quantity_changed_at": now,
"resolved_at": None,
"operator_action": None,
"evidence_ref": None,
"operator_note": "",
"operator_actor": None,
"operator_updated_at": None,
"created_at": now,
**preserved,
"updated_at": now,
}
self._states[key] = state
elif int(existing["quantity"]) != int(quantity):
transition = "quantity_changed"
existing["quantity"] = quantity
existing["last_seen_at"] = now
existing["quantity_changed_at"] = now
existing["resolved_at"] = None
existing["lifecycle_state"] = "active"
existing["updated_at"] = now
state = existing
else:
transition = "unchanged"
existing["last_seen_at"] = now
existing["updated_at"] = now
state = existing
if transition != "unchanged":
self._audit.append(
{
"instance_key": instance_key,
"symbol": symbol,
"event_type": transition,
"action": None,
"quantity": quantity,
"lifecycle_state": state["lifecycle_state"],
"actor": actor,
"evidence_ref": state.get("evidence_ref"),
"note": (
f"previous_quantity={previous_quantity}"
if transition == "quantity_changed" and previous_quantity is not None
else ""
),
"created_at": now,
}
)
return {
"transition": transition,
"previous_quantity": previous_quantity,
"state": dict(state),
}
def resolve_absent(
self,
*,
instance_key: str,
observed_symbols: set[str],
observed_at: datetime | None = None,
actor: str = "reconcile",
) -> list[dict]:
now = observed_at or datetime.now(tz=UTC)
with self._lock:
resolved: list[dict] = []
for key, state in sorted(self._states.items()):
if key[0] != instance_key:
continue
if state.get("lifecycle_state") != "active":
continue
if state["symbol"] in observed_symbols:
continue
state["lifecycle_state"] = "resolved"
state["resolved_at"] = now
state["updated_at"] = now
resolved.append(dict(state))
self._audit.append(
{
"instance_key": state["instance_key"],
"symbol": state["symbol"],
"event_type": "resolved",
"action": None,
"quantity": state["quantity"],
"lifecycle_state": "resolved",
"actor": actor,
"evidence_ref": state.get("evidence_ref"),
"note": "",
"created_at": now,
}
)
return resolved
def get_state(self, *, instance_key: str, symbol: str) -> dict | None:
with self._lock:
state = self._states.get((instance_key, symbol))
return dict(state) if state is not None else None
def list_states(
self,
*,
instance_key: str,
lifecycle_state: str = "active",
symbol: str | None = None,
) -> list[dict]:
with self._lock:
rows = []
for state in self._states.values():
if state["instance_key"] != instance_key:
continue
if lifecycle_state != "all" and state["lifecycle_state"] != lifecycle_state:
continue
if symbol is not None and state["symbol"] != symbol:
continue
rows.append(dict(state))
return sorted(rows, key=lambda item: item["symbol"])
def record_operator_action(
self,
*,
instance_key: str,
symbol: str,
action: str,
evidence_ref: str,
actor: str = "operator",
note: str = "",
) -> dict | None:
now = datetime.now(tz=UTC)
with self._lock:
state = self._states.get((instance_key, symbol))
if state is None:
return None
state["operator_action"] = action
state["evidence_ref"] = evidence_ref
state["operator_note"] = note
state["operator_actor"] = actor
state["operator_updated_at"] = now
state["updated_at"] = now
self._audit.append(
{
"instance_key": instance_key,
"symbol": symbol,
"event_type": "operator_action",
"action": action,
"quantity": state["quantity"],
"lifecycle_state": state["lifecycle_state"],
"actor": actor,
"evidence_ref": evidence_ref,
"note": note,
"created_at": now,
}
)
return dict(state)
def list_audit(self, *, instance_key: str, symbol: str) -> list[dict]:
with self._lock:
return [
dict(row)
for row in self._audit
if row["instance_key"] == instance_key and row["symbol"] == symbol
]
def slot_to_dict(
s,
*,
@@ -637,4 +890,3 @@ def _sanitize(message: str) -> str:
out = out[:idx] + "<redacted>"
break
return out[:200]
trex_core/application/reconciliation.py
+97 -4
View File
@@ -33,7 +33,11 @@ from typing import cast
from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.broker_port import BrokerPort
from trex_core.application.portfolio import ReconcileWarningStore, SnapshotStore
from trex_core.application.portfolio import (
ReconcileWarningStore,
SnapshotStore,
UnmanagedPositionStore,
)
from trex_core.domain.slot.state import SlotState
_log = logging.getLogger(__name__)
@@ -341,6 +345,25 @@ def _retry_pending_resolved(
)
def _notify_unmanaged_position_transition(
alarm_emitter: AlarmEmitter,
*,
event_type: str,
instance_key: str,
symbol: str,
quantity: int,
) -> None:
alarm_emitter.publish_alarm(
f"unmanaged_position_{event_type}",
f"unmanaged position {event_type}: symbol={symbol}",
fields={
"instance_key": instance_key,
"symbol": symbol,
"broker_quantity": quantity,
},
)
def reconcile_core_holdings(
*,
slot_repo,
@@ -350,6 +373,7 @@ def reconcile_core_holdings(
alarm_emitter: AlarmEmitter | None = None,
warn_store: ReconcileWarningStore | None = None,
notification_store=None,
unmanaged_position_store: UnmanagedPositionStore | None = None,
) -> dict[str, str]:
"""dcy-020 / REQ-062 (BAT-029) + REQ-103 (BAT-050): surface core-slot holding
vs broker divergence and set durable reconcile-warning for B case.
@@ -377,6 +401,7 @@ def reconcile_core_holdings(
decisions: dict[str, str] = {}
matched: set[str] = set()
active_b_identity_keys: set[str] = set() # req-0007: track active B mismatches for retry
active_unmanaged_symbols: set[str] = set()
for slot in slot_repo.list_holding_slots(instance_key):
sym = slot.symbol
matched.add(sym)
@@ -458,10 +483,78 @@ def reconcile_core_holdings(
)
else:
decisions[f"broker:{sym}"] = "unmanaged_position"
_log.warning(
"unmanaged position: symbol=%s qty=%d instance_key=%s",
sym, bq, instance_key,
active_unmanaged_symbols.add(sym)
if unmanaged_position_store is None:
_log.warning(
"unmanaged position: symbol=%s qty=%d instance_key=%s",
sym, bq, instance_key,
)
else:
try:
observed = unmanaged_position_store.observe(
instance_key=instance_key,
symbol=sym,
quantity=bq,
)
except Exception: # noqa: BLE001
_log.exception(
"failed to persist unmanaged_position for symbol=%s instance=%s",
sym,
instance_key,
)
_log.warning(
"unmanaged position: symbol=%s qty=%d instance_key=%s",
sym, bq, instance_key,
)
else:
transition = observed["transition"]
if transition == "detected":
_log.warning(
"unmanaged position detected: symbol=%s qty=%d instance_key=%s",
sym, bq, instance_key,
)
if alarm_emitter is not None:
_notify_unmanaged_position_transition(
alarm_emitter,
event_type="detected",
instance_key=instance_key,
symbol=sym,
quantity=bq,
)
elif transition == "quantity_changed":
_log.warning(
"unmanaged position quantity changed: symbol=%s "
"previous_qty=%s qty=%d instance_key=%s",
sym, observed.get("previous_quantity"), bq, instance_key,
)
if alarm_emitter is not None:
_notify_unmanaged_position_transition(
alarm_emitter,
event_type="changed",
instance_key=instance_key,
symbol=sym,
quantity=bq,
)
if unmanaged_position_store is not None:
try:
for state in unmanaged_position_store.resolve_absent(
instance_key=instance_key,
observed_symbols=active_unmanaged_symbols,
):
_log.info(
"unmanaged position resolved: symbol=%s qty=%d instance_key=%s",
state["symbol"], state["quantity"], instance_key,
)
if alarm_emitter is not None:
_notify_unmanaged_position_transition(
alarm_emitter,
event_type="resolved",
instance_key=instance_key,
symbol=state["symbol"],
quantity=state["quantity"],
)
except Exception: # noqa: BLE001
_log.exception("reconcile_core_holdings: unmanaged_position resolve failed")
# req-0007 / ac-0024: retry pending_resolved for B-case identities no longer mismatched.
if alarm_emitter is not None and notification_store is not None:
try:
trex_core/application/snapshot_scheduler.py
+4
View File
@@ -23,6 +23,7 @@ from trex_core.application.portfolio import (
PortfolioHubReadPort,
PortfolioHubUnavailableError,
ReconcileNotificationStore,
UnmanagedPositionStore,
ReconcileWarningStore,
SnapshotStore,
TradingContext,
@@ -51,6 +52,7 @@ class SnapshotScheduler:
alarm_emitter: Optional[AlarmEmitter] = None,
warn_store: Optional[ReconcileWarningStore] = None,
notification_store: Optional[ReconcileNotificationStore] = None,
unmanaged_position_store: Optional[UnmanagedPositionStore] = None,
credential_fetch_enabled: bool = False,
kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None,
credential_refresh: Optional[Callable[[], object]] = None,
@@ -75,6 +77,7 @@ class SnapshotScheduler:
self._warn_store = warn_store
# req-0007 / ac-0024: durable notification deduplication store (Optional).
self._notification_store = notification_store
self._unmanaged_position_store = unmanaged_position_store
# dcy-036 / REQ-113/114/115: credential fetch opt-in + break-glass detection.
# credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch.
self._credential_fetch_enabled: bool = credential_fetch_enabled
@@ -336,6 +339,7 @@ class SnapshotScheduler:
alarm_emitter=self._alarm_emitter, # dcy-033 / REQ-100
warn_store=self._warn_store, # BAT-050 / REQ-103
notification_store=self._notification_store, # req-0007 / ac-0024
unmanaged_position_store=self._unmanaged_position_store,
)
except Exception: # noqa: BLE001
_log.exception("core-holding divergence reconcile failed")
trex_core/runtime/cli.py
+104
View File
@@ -11,6 +11,8 @@ Subcommands:
cancel <slot_id> — cancel a pending order (control RPC; safeguarded)
sync — sync slot qty to broker ground truth (direct-DB; REQ-105)
verify — re-check C-case reconcile-warning (direct-DB; REQ-106)
unmanaged list — query durable unmanaged-position lifecycle state
unmanaged record — record explicit operator disposition + evidence reference
ledger backfill — evidence-backed historical ledger recovery
ledger skips list — query durable skipped-ledger records
ledger skips resolve — explicitly resolve a skipped-ledger record
@@ -640,6 +642,37 @@ def _build_parser() -> argparse.ArgumentParser:
)
_add_common(p_verify)
p_unmanaged = sub.add_parser(
"unmanaged",
help="List or annotate durable unmanaged broker positions (direct-DB)",
)
p_unmanaged_sub = p_unmanaged.add_subparsers(dest="unmanaged_cmd", required=True)
p_unmanaged_list = p_unmanaged_sub.add_parser(
"list",
help="List unmanaged broker positions with lifecycle state",
)
p_unmanaged_list.add_argument(
"--status",
choices=("active", "resolved", "all"),
default="active",
)
p_unmanaged_list.add_argument("--symbol", default=None)
_add_common(p_unmanaged_list)
p_unmanaged_record = p_unmanaged_sub.add_parser(
"record",
help="Record an explicit operator disposition for an unmanaged position",
)
p_unmanaged_record.add_argument("symbol")
p_unmanaged_record.add_argument(
"action",
choices=("investigating", "approved_exception", "managed_inclusion"),
)
p_unmanaged_record.add_argument("--evidence-ref", required=True)
p_unmanaged_record.add_argument("--note", default="")
p_unmanaged_record.add_argument("--actor", default="operator")
_add_common(p_unmanaged_record)
p_ledger = sub.add_parser(
"ledger",
help="Ledger operator tools (direct-DB enqueue + immediate writer path)",
@@ -971,6 +1004,73 @@ def _cmd_verify(args, cfg) -> int:
return exit_code
def _cmd_unmanaged_list(args, cfg) -> int:
from trex_core.adapters.portfolio.unmanaged_position_store import (
PostgreSQLUnmanagedPositionStore,
)
instance = _resolve_instance(args)
states = PostgreSQLUnmanagedPositionStore(cfg.dsn).list_states(
instance_key=instance,
lifecycle_state=args.status,
symbol=args.symbol,
)
payload = {
"instance": instance,
"status": args.status,
"count": len(states),
"states": states,
}
if args.json:
_emit(payload, as_json=True)
else:
print(
f"instance={instance} status={args.status} count={len(states)} "
f"symbol={args.symbol or '*'}"
)
for state in states:
print(
f" {state['symbol']}\tqty={state['quantity']}\t"
f"lifecycle={state['lifecycle_state']}\t"
f"first_seen={state['first_seen_at']}\tlast_seen={state['last_seen_at']}\t"
f"resolved_at={state['resolved_at']}\t"
f"operator_action={state.get('operator_action') or '-'}\t"
f"evidence_ref={state.get('evidence_ref') or '-'}"
)
return 0
def _cmd_unmanaged_record(args, cfg) -> int:
from trex_core.adapters.portfolio.unmanaged_position_store import (
PostgreSQLUnmanagedPositionStore,
)
instance = _resolve_instance(args)
state = PostgreSQLUnmanagedPositionStore(cfg.dsn).record_operator_action(
instance_key=instance,
symbol=args.symbol,
action=args.action,
evidence_ref=args.evidence_ref,
actor=args.actor,
note=args.note,
)
if state is None:
_fail(
f"unmanaged position not found: instance={instance} symbol={args.symbol}",
args.json,
)
return 1
payload = {"result": "recorded", "state": state}
if args.json:
_emit(payload, as_json=True)
else:
print(
f"recorded unmanaged position instance={instance} symbol={args.symbol} "
f"action={args.action} evidence_ref={args.evidence_ref}"
)
return 0
def _cmd_report(args, cfg) -> int:
"""REQ-088/110/111: holdings / today's fills / WAC realized PnL / pending — direct-DB."""
import psycopg
@@ -1434,6 +1534,10 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
# REQ-106 (BAT-051): verify — direct-DB, C-case re-check (read-centric).
if args.cmd == "verify":
return _cmd_verify(args, cfg)
if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "list":
return _cmd_unmanaged_list(args, cfg)
if args.cmd == "unmanaged" and getattr(args, "unmanaged_cmd", None) == "record":
return _cmd_unmanaged_record(args, cfg)
if args.cmd == "ledger" and getattr(args, "ledger_cmd", None) == "backfill":
return _cmd_ledger_backfill(args, cfg, confirm)
trex_core/runtime/daemon.py
+5
View File
@@ -52,6 +52,9 @@ from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchSt
from trex_core.adapters.portfolio.reconcile_notification_store import (
PostgreSQLReconcileNotificationStore,
)
from trex_core.adapters.portfolio.unmanaged_position_store import (
PostgreSQLUnmanagedPositionStore,
)
from trex_core.application.alarm_emitter import AlarmEmitter
from trex_core.application.account_snapshot import AccountSnapshotService
from trex_core.application.broker_port import BrokerPort
@@ -202,6 +205,7 @@ class Daemon:
self.warn_store = PostgreSQLReconcileWarningStore(config.dsn)
# req-0007 / ac-0024: durable reconcile notification deduplication store.
self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn)
self.unmanaged_position_store = PostgreSQLUnmanagedPositionStore(config.dsn)
# BAT-012: day-scoped kill-switch exclusion store + service.
self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn)
self.portfolio_read = _build_portfolio_read(config)
@@ -278,6 +282,7 @@ class Daemon:
alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100
warn_store=self.warn_store, # BAT-050 / REQ-103: durable reconcile-warning
notification_store=self.notification_store, # req-0007 / ac-0024
unmanaged_position_store=self.unmanaged_position_store,
credential_fetch_enabled=config.credential_fetch_enabled, # dcy-036 / REQ-113
credential_refresh=self.broker_provider.refresh_day_start_credential,
# dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.