From fe51d211d9ea4f3ff0cdd3efe794bc9c398fa91d Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:24:51 +0000 Subject: [PATCH] feat: persist unmanaged position lifecycle state Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../test_unmanaged_position_store.py | 101 ++++++ .../test_unmanaged_position_lifecycle.py | 190 +++++++++++ tests/conftest.py | 2 + tests/runtime/test_cli_operator.py | 169 ++++++++++ .../028_create_unmanaged_position_state.sql | 38 +++ trex_core/adapters/portfolio/__init__.py | 2 + .../portfolio/unmanaged_position_store.py | 309 ++++++++++++++++++ trex_core/application/portfolio.py | 254 +++++++++++++- trex_core/application/reconciliation.py | 101 +++++- trex_core/application/snapshot_scheduler.py | 4 + trex_core/runtime/cli.py | 104 ++++++ trex_core/runtime/daemon.py | 5 + 12 files changed, 1274 insertions(+), 5 deletions(-) create mode 100644 tests/adapters/portfolio/test_unmanaged_position_store.py create mode 100644 tests/application/test_unmanaged_position_lifecycle.py create mode 100644 trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql create mode 100644 trex_core/adapters/portfolio/unmanaged_position_store.py diff --git a/tests/adapters/portfolio/test_unmanaged_position_store.py b/tests/adapters/portfolio/test_unmanaged_position_store.py new file mode 100644 index 0000000..b882f76 --- /dev/null +++ b/tests/adapters/portfolio/test_unmanaged_position_store.py @@ -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 diff --git a/tests/application/test_unmanaged_position_lifecycle.py b/tests/application/test_unmanaged_position_lifecycle.py new file mode 100644 index 0000000..3344117 --- /dev/null +++ b/tests/application/test_unmanaged_position_lifecycle.py @@ -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" diff --git a/tests/conftest.py b/tests/conftest.py index fc525ef..cfa8acd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -28,6 +28,8 @@ _PG_TABLES = ( "daily_slot", "reconcile_notification_state", "slot_strategy_state", + "unmanaged_position_audit", + "unmanaged_position_state", ) diff --git a/tests/runtime/test_cli_operator.py b/tests/runtime/test_cli_operator.py index f110464..d87683f 100644 --- a/tests/runtime/test_cli_operator.py +++ b/tests/runtime/test_cli_operator.py @@ -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 # --------------------------------------------------------------------------- diff --git a/trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql b/trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql new file mode 100644 index 0000000..870cd7e --- /dev/null +++ b/trex_core/adapters/persistence/sql/028_create_unmanaged_position_state.sql @@ -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); diff --git a/trex_core/adapters/portfolio/__init__.py b/trex_core/adapters/portfolio/__init__.py index 794288e..4476194 100644 --- a/trex_core/adapters/portfolio/__init__.py +++ b/trex_core/adapters/portfolio/__init__.py @@ -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", ] diff --git a/trex_core/adapters/portfolio/unmanaged_position_store.py b/trex_core/adapters/portfolio/unmanaged_position_store.py new file mode 100644 index 0000000..694ab4b --- /dev/null +++ b/trex_core/adapters/portfolio/unmanaged_position_store.py @@ -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], + } diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index abe519b..cf2870e 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -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] + "" break return out[:200] - diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index aca2c5d..4c2b7b4 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -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: diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index b6ef0bc..d675656 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -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") diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 128ef21..e4ffe68 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -11,6 +11,8 @@ Subcommands: cancel — 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) diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index d232973..b0b0f2e 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -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.