From 24a6017457d2e9d4cee9ada6c68b723673b8d9f1 Mon Sep 17 00:00:00 2001 From: Justin Ko Date: Thu, 11 Jun 2026 21:08:46 +0900 Subject: [PATCH] =?UTF-8?q?fix(reconcile):=20symbol=EC=9D=84=20=EC=8A=AC?= =?UTF-8?q?=EB=A1=AF=20=EC=83=9D=EC=84=B1=20=EC=8B=9C=20=ED=95=A8=EA=BB=98?= =?UTF-8?q?=20=EC=A0=80=EC=9E=A5=20+=20NULL=20=EC=8A=AC=EB=A1=AF=20backfil?= =?UTF-8?q?l?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reconcile_universe 시그니처를 set[str] → dict[str, str] (portfolio_id→symbol)로 변경. 기존 symbol=NULL 슬롯은 다음 reconcile 실행 시 자동 backfill. Co-Authored-By: Claude Sonnet 4.6 --- .../messaging/test_rabbitmq_intent_e2e.py | 2 +- .../persistence/test_postgresql_repository.py | 20 +++++++------- .../test_asset_centric_admission.py | 16 ++++++------ tests/application/test_execution.py | 2 +- tests/application/test_execution_cancel.py | 2 +- tests/application/test_execution_ledger.py | 2 +- tests/application/test_execution_policy.py | 2 +- .../test_execution_policy_handler.py | 4 +-- tests/application/test_holding_divergence.py | 2 +- tests/application/test_intent_concurrency.py | 2 +- .../application/test_killswitch_execution.py | 2 +- tests/application/test_ledger_option_d.py | 2 +- .../test_messaging_event_publisher.py | 2 +- .../test_messaging_intent_consumer.py | 2 +- .../test_safety_sell_visibility.py | 2 +- tests/application/test_safety_sell_worker.py | 2 +- tests/application/test_sell_pricing.py | 2 +- tests/application/test_slots_handler.py | 4 +-- .../test_update_thresholds_handler.py | 4 +-- .../slot/test_execution_policy_lifecycle.py | 2 +- .../slot/test_price_threshold_fields.py | 2 +- tests/domain/slot/test_service.py | 2 +- tests/domain/slot/test_service_sell.py | 2 +- tests/domain/slot/test_universe_reconcile.py | 22 ++++++++-------- tests/runtime/test_reset.py | 4 +-- trex_core/adapters/persistence/postgresql.py | 26 ++++++++++++++----- trex_core/application/snapshot_scheduler.py | 6 ++++- trex_core/domain/slot/in_memory_repository.py | 13 +++++++--- trex_core/domain/slot/repository.py | 5 ++-- 29 files changed, 92 insertions(+), 68 deletions(-) diff --git a/tests/adapters/messaging/test_rabbitmq_intent_e2e.py b/tests/adapters/messaging/test_rabbitmq_intent_e2e.py index 327d7d5..f4cefe3 100644 --- a/tests/adapters/messaging/test_rabbitmq_intent_e2e.py +++ b/tests/adapters/messaging/test_rabbitmq_intent_e2e.py @@ -54,7 +54,7 @@ def test_intent_admission_rpc_dedups_duplicate_correlation_id_over_rabbitmq(mq_u svc = SlotService(repo) intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") # dcy-011: universe resolve - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) handler = make_intent_rpc_handler(intents) queue = f"trex.test.intent.{uuid.uuid4().hex[:8]}" diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index 8e6e99e..3a4979f 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/tests/adapters/persistence/test_postgresql_repository.py @@ -29,22 +29,22 @@ def test_universe_reconcile_pg(pg_clean): # partial-index upsert + find_slot_by_portfolio/count_active. repo = PostgreSQLSlotRepository(pg_clean) inst = "inst-pg" - r1 = repo.reconcile_universe(inst, {"pf-a", "pf-b"}) + r1 = repo.reconcile_universe(inst, {"pf-a": "SYM", "pf-b": "SYM"}) assert r1["created"] == 2 - r2 = repo.reconcile_universe(inst, {"pf-a", "pf-b"}) # idempotent (ON CONFLICT) + r2 = repo.reconcile_universe(inst, {"pf-a": "SYM", "pf-b": "SYM"}) # idempotent (ON CONFLICT) assert r2["created"] == 0 slot_a = repo.find_slot_by_portfolio(inst, "pf-a") assert slot_a is not None and slot_a.status is SlotState.INACTIVE assert repo.count_active(inst) == 0 # pf-b leaves universe while inactive -> soft-retire - r3 = repo.reconcile_universe(inst, {"pf-a"}) + r3 = repo.reconcile_universe(inst, {"pf-a": "SYM"}) assert r3["retired"] == 1 assert repo.find_slot_by_portfolio(inst, "pf-b").retired_at is not None def test_capacity_instance_scoped(pg_clean): repo, svc, _ = _wire(pg_clean) - repo.reconcile_universe(INST, set(_SYMS)) + repo.reconcile_universe(INST, _SYMS) cap = svc.compute_slot_capacity() assert cap.inactive_count == 3 and cap.available_count == 3 cap_m = svc.compute_slot_capacity(max_active_slots=2) @@ -54,7 +54,7 @@ def test_capacity_instance_scoped(pg_clean): def test_buy_intent_persists_transition(pg_clean): repo, svc, intents = _wire(pg_clean) - repo.reconcile_universe(INST, set(_SYMS)) + repo.reconcile_universe(INST, _SYMS) d = intents.submit_buy(BuyIntent(intent_id="pg-buy-1", qty=10, portfolio_id="PF-1")) assert d.admitted slot = repo.fetch_slot(d.slot_id) @@ -63,7 +63,7 @@ def test_buy_intent_persists_transition(pg_clean): def test_buy_intent_idempotent_on_pg(pg_clean): repo, svc, intents = _wire(pg_clean) - repo.reconcile_universe(INST, set(_SYMS)) + repo.reconcile_universe(INST, _SYMS) d1 = intents.submit_buy(BuyIntent(intent_id="pg-dup", portfolio_id="PF-1")) d2 = intents.submit_buy(BuyIntent(intent_id="pg-dup", portfolio_id="PF-1")) assert d1.slot_id == d2.slot_id and d2.status == "admitted" @@ -74,7 +74,7 @@ def test_buy_intent_idempotent_on_pg(pg_clean): def test_buy_by_slot_id_contract_pg(pg_clean): # dcy-020/REQ-061: slot_id is the canonical buy key against real PG. repo, svc, intents = _wire(pg_clean) - repo.reconcile_universe(INST, set(_SYMS)) + repo.reconcile_universe(INST, _SYMS) slot = repo.find_slot_by_portfolio(INST, "PF-2") d = intents.submit_buy(BuyIntent(intent_id="pg-sid", slot_id=slot.slot_id)) assert d.admitted and d.slot_id == slot.slot_id @@ -85,7 +85,7 @@ def test_full_lifecycle_returns_to_inactive_and_reuses(pg_clean): # dcy-020 (BAT-030): sell fill -> inactive (occupancy cleared, asset # binding kept); the same permanent slot activates again. repo, svc, intents = _wire(pg_clean) - repo.reconcile_universe(INST, set(_SYMS)) + repo.reconcile_universe(INST, _SYMS) d = intents.submit_buy(BuyIntent(intent_id="pg-life", portfolio_id="PF-1")) sid = d.slot_id svc.transit_slot(sid, SlotState.PENDING_BUY, SlotState.HOLDING, @@ -106,8 +106,8 @@ def test_dispatch_queue_instance_scoped(pg_clean): # dcy-020/REQ-056 (BAT-030): list_undispatched_pending / list_dispatched_pending # never cross instance boundaries in a shared DB. repo_a = PostgreSQLSlotRepository(pg_clean) - repo_a.reconcile_universe("inst-a", {"pf-x"}) - repo_a.reconcile_universe("inst-b", {"pf-x"}) + repo_a.reconcile_universe("inst-a", {"pf-x": "SYM"}) + repo_a.reconcile_universe("inst-b", {"pf-x": "SYM"}) svc_a = SlotService(repo_a, instance_key="inst-a") slot_a = repo_a.find_slot_by_portfolio("inst-a", "pf-x") svc_a.transit_slot(slot_a.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, diff --git a/tests/application/test_asset_centric_admission.py b/tests/application/test_asset_centric_admission.py index 2328353..7f62713 100644 --- a/tests/application/test_asset_centric_admission.py +++ b/tests/application/test_asset_centric_admission.py @@ -24,7 +24,7 @@ def _wire(max_active): def test_activates_asset_slot(): repo, svc, intents = _wire(max_active=2) - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) d = intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="pf-a")) assert d.status == "admitted" slot = repo.find_slot_by_portfolio(INST, "pf-a") @@ -34,7 +34,7 @@ def test_activates_asset_slot(): def test_activates_by_slot_id(): # dcy-020/REQ-061: slot_id is the canonical buy contract key. repo, svc, intents = _wire(max_active=2) - repo.reconcile_universe(INST, {"pf-a"}) + repo.reconcile_universe(INST, {"pf-a": "SYM"}) slot = repo.find_slot_by_portfolio(INST, "pf-a") d = intents.submit_buy(BuyIntent(intent_id="b1", slot_id=slot.slot_id)) assert d.status == "admitted" and d.slot_id == slot.slot_id @@ -45,14 +45,14 @@ def test_activates_by_slot_id(): def test_unknown_slot_for_bad_slot_id(): from uuid import uuid4 repo, svc, intents = _wire(max_active=2) - repo.reconcile_universe(INST, {"pf-a"}) + repo.reconcile_universe(INST, {"pf-a": "SYM"}) d = intents.submit_buy(BuyIntent(intent_id="b1", slot_id=uuid4())) assert d.status == "rejected" and d.reason == "unknown_slot" def test_already_active_is_rejected(): repo, svc, intents = _wire(max_active=2) - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="pf-a")) d2 = intents.submit_buy(BuyIntent(intent_id="b2", portfolio_id="pf-a")) assert d2.status == "rejected" and d2.reason == "already_active" @@ -60,7 +60,7 @@ def test_already_active_is_rejected(): def test_capacity_full_at_M_no_eviction(): repo, svc, intents = _wire(max_active=1) - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) d1 = intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="pf-a")) assert d1.status == "admitted" d2 = intents.submit_buy(BuyIntent(intent_id="b2", portfolio_id="pf-b")) @@ -71,7 +71,7 @@ def test_capacity_full_at_M_no_eviction(): def test_unknown_portfolio_when_no_asset_slot(): repo, svc, intents = _wire(max_active=2) - repo.reconcile_universe(INST, {"pf-a"}) + repo.reconcile_universe(INST, {"pf-a": "SYM"}) # pf-c is resolvable to a symbol but has no reconciled asset slot d = intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="pf-c")) assert d.status == "rejected" and d.reason == "unknown_portfolio" @@ -80,7 +80,7 @@ def test_unknown_portfolio_when_no_asset_slot(): def test_over_cap_blocks_new_keeps_existing(): # M=2: fill 2, then lower M to 1 (over-cap) -> new rejected, existing kept repo, svc, intents = _wire(max_active=2) - repo.reconcile_universe(INST, {"pf-a", "pf-b", "pf-c"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM", "pf-c": "SYM"}) intents.submit_buy(BuyIntent(intent_id="b1", portfolio_id="pf-a")) intents.submit_buy(BuyIntent(intent_id="b2", portfolio_id="pf-b")) assert repo.count_active(INST) == 2 @@ -94,7 +94,7 @@ def test_over_cap_blocks_new_keeps_existing(): def test_no_cap_admits_without_capacity_check(): # BAT-030: max_active_slots=None -> no capacity cap (asset-centric only path) repo, svc, intents = _wire(max_active=None) - repo.reconcile_universe(INST, {"pf-a", "pf-b", "pf-c"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM", "pf-c": "SYM"}) for i, pid in enumerate(sorted(_SYMS)): d = intents.submit_buy(BuyIntent(intent_id=f"b{i}", portfolio_id=pid)) assert d.status == "admitted" diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index b0447a9..4827aa9 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -24,7 +24,7 @@ def _wire(broker, max_active_slots=None): instance_key=INST, max_active_slots=max_active_slots, ) execution = ExecutionService(svc, repo, broker, instance_key=INST) - repo.reconcile_universe(INST, set(SYMBOLS)) # permanent asset slots + repo.reconcile_universe(INST, SYMBOLS) # permanent asset slots return repo, svc, intents, execution diff --git a/tests/application/test_execution_cancel.py b/tests/application/test_execution_cancel.py index a67bd7f..8baa58a 100644 --- a/tests/application/test_execution_cancel.py +++ b/tests/application/test_execution_cancel.py @@ -69,7 +69,7 @@ def _wire(broker, *, stale_sec=None): stale_pending_threshold_sec=stale_sec, ) writer = LedgerWriter(outbox, ledger) - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents, execution, ledger, writer diff --git a/tests/application/test_execution_ledger.py b/tests/application/test_execution_ledger.py index 8b6b771..6f4d904 100644 --- a/tests/application/test_execution_ledger.py +++ b/tests/application/test_execution_ledger.py @@ -33,7 +33,7 @@ def _wire(broker, *, ledger=None, clock=None): LedgerWriter(outbox, ledger or InMemoryLedger(), clock=clock) if clock else LedgerWriter(outbox, ledger or InMemoryLedger()) ) - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents, execution, outbox, writer diff --git a/tests/application/test_execution_policy.py b/tests/application/test_execution_policy.py index 8d3a120..ab775d3 100644 --- a/tests/application/test_execution_policy.py +++ b/tests/application/test_execution_policy.py @@ -62,7 +62,7 @@ def _wire(broker): intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") execution = ExecutionService(svc, repo, broker) - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents, execution diff --git a/tests/application/test_execution_policy_handler.py b/tests/application/test_execution_policy_handler.py index 3c2a94a..b9e948c 100644 --- a/tests/application/test_execution_policy_handler.py +++ b/tests/application/test_execution_policy_handler.py @@ -28,7 +28,7 @@ def _wire(): svc = SlotService(repo) intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents @@ -145,7 +145,7 @@ class TestDefaultOrderTimeoutSec: resolve_symbol=lambda pid: "005930", default_order_timeout_sec=default_timeout, ) - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents def test_config_default_applied_when_intent_has_no_timeout(self): diff --git a/tests/application/test_holding_divergence.py b/tests/application/test_holding_divergence.py index ab7ea5f..1036362 100644 --- a/tests/application/test_holding_divergence.py +++ b/tests/application/test_holding_divergence.py @@ -22,7 +22,7 @@ class FakeBroker: def _hold(repo, svc, pid, symbol, qty): - repo.reconcile_universe(INST, {pid}) + repo.reconcile_universe(INST, {pid: "SYM"}) slot = repo.find_slot_by_portfolio(INST, pid) svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, reason="r", actor="a", symbol=symbol, qty=qty) diff --git a/tests/application/test_intent_concurrency.py b/tests/application/test_intent_concurrency.py index e6c123a..903511f 100644 --- a/tests/application/test_intent_concurrency.py +++ b/tests/application/test_intent_concurrency.py @@ -24,7 +24,7 @@ def _wire(n_assets: int, max_active: int | None): resolve_symbol=lambda pid: f"SYM-{pid}", instance_key=INST, max_active_slots=max_active, ) - repo.reconcile_universe(INST, {f"PF-{i}" for i in range(n_assets)}) + repo.reconcile_universe(INST, {f"PF-{i}": "SYM" for i in range(n_assets)}) return repo, svc, intents diff --git a/tests/application/test_killswitch_execution.py b/tests/application/test_killswitch_execution.py index de613a2..3d3e880 100644 --- a/tests/application/test_killswitch_execution.py +++ b/tests/application/test_killswitch_execution.py @@ -36,7 +36,7 @@ def ks(repo, service, store): def _holding(service, repo, *, portfolio_id, symbol="005930"): - repo.reconcile_universe("default", {portfolio_id}) + repo.reconcile_universe("default", {portfolio_id: "SYM"}) slot = repo.find_slot_by_portfolio("default", portfolio_id) pending = service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, diff --git a/tests/application/test_ledger_option_d.py b/tests/application/test_ledger_option_d.py index 6584d7c..e81dde2 100644 --- a/tests/application/test_ledger_option_d.py +++ b/tests/application/test_ledger_option_d.py @@ -105,7 +105,7 @@ def test_apply_cash_estimate_overlays_pending_delta(): outbox.enqueue(_trade(execution_id="b", side="buy", qty=10, price=1000.0, fee=1.0)) store = _store_with(pid="PF-1", cash=1_000_000.0, qty=0.0) repo = InMemorySlotRepository() - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) handler = make_slots_rpc_handler(repo, snapshot_store=store, ledger_outbox=outbox) resp = handler(RpcRequest(payload={"filter": "all"}, correlation_id="test")) slots = resp.payload["slots"] diff --git a/tests/application/test_messaging_event_publisher.py b/tests/application/test_messaging_event_publisher.py index a3e704e..ecd57a8 100644 --- a/tests/application/test_messaging_event_publisher.py +++ b/tests/application/test_messaging_event_publisher.py @@ -25,7 +25,7 @@ def _wire(transport): outbox = InMemoryLedgerOutbox() ledger = InMemoryLedger() writer = LedgerWriter(outbox, ledger) - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) def emitter(event_type: str, slot) -> None: rk = topology.event_routing_key(event_type, slot.strategy_id or "_", slot.symbol or "_") diff --git a/tests/application/test_messaging_intent_consumer.py b/tests/application/test_messaging_intent_consumer.py index beaebfa..42e4bd1 100644 --- a/tests/application/test_messaging_intent_consumer.py +++ b/tests/application/test_messaging_intent_consumer.py @@ -22,7 +22,7 @@ def _wire(max_active=3): intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930", max_active_slots=max_active) - repo.reconcile_universe("default", {"PF-1", "PF-2", "PF-3", "PF-9"}) + repo.reconcile_universe("default", {"PF-1": "SYM", "PF-2": "SYM", "PF-3": "SYM", "PF-9": "SYM"}) return repo, svc, intents diff --git a/tests/application/test_safety_sell_visibility.py b/tests/application/test_safety_sell_visibility.py index 89bd48d..d5f7a68 100644 --- a/tests/application/test_safety_sell_visibility.py +++ b/tests/application/test_safety_sell_visibility.py @@ -49,7 +49,7 @@ def _wire(**cfg_over): def _holding(svc, repo, *, symbol="005930", **fields): from uuid import uuid4 pid = f"pf-{uuid4().hex[:8]}" - repo.reconcile_universe("inst-1", {pid}) + repo.reconcile_universe("inst-1", {pid: "SYM"}) slot = repo.find_slot_by_portfolio("inst-1", pid) p = svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t", symbol=symbol, **fields) diff --git a/tests/application/test_safety_sell_worker.py b/tests/application/test_safety_sell_worker.py index 891f6c8..9ec56af 100644 --- a/tests/application/test_safety_sell_worker.py +++ b/tests/application/test_safety_sell_worker.py @@ -72,7 +72,7 @@ def _make_holding(service, repo, *, symbol="005930", forced_close_at=None, stop_loss_price=None, take_profit_price=None): from uuid import uuid4 pid = f"pf-{uuid4().hex[:8]}" - repo.reconcile_universe("default", {pid}) + repo.reconcile_universe("default", {pid: "SYM"}) slot = repo.find_slot_by_portfolio("default", pid) pending = service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, diff --git a/tests/application/test_sell_pricing.py b/tests/application/test_sell_pricing.py index 981e5d8..d401894 100644 --- a/tests/application/test_sell_pricing.py +++ b/tests/application/test_sell_pricing.py @@ -35,7 +35,7 @@ def _wire(broker): intents = IntentService(svc, repo, InMemoryIntentStore(), resolve_symbol=lambda pid: "005930") execution = ExecutionService(svc, repo, broker) - repo.reconcile_universe("default", {"PF-1"}) + repo.reconcile_universe("default", {"PF-1": "SYM"}) return repo, svc, intents, execution diff --git a/tests/application/test_slots_handler.py b/tests/application/test_slots_handler.py index fec6375..7291417 100644 --- a/tests/application/test_slots_handler.py +++ b/tests/application/test_slots_handler.py @@ -17,7 +17,7 @@ from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository def _wire(items=None, *, max_active_slots=None): repo = InMemorySlotRepository() svc = SlotService(repo) - repo.reconcile_universe("default", {"pf-A", "pf-B"}) + repo.reconcile_universe("default", {"pf-A": "SYM", "pf-B": "SYM"}) store = SnapshotStore() store.swap(items or [ PortfolioItem(portfolio_id="pf-A", symbol="005930", @@ -146,7 +146,7 @@ def test_warned_true_when_qty_diverges(): PortfolioItem(portfolio_id="pf-A", symbol="005930", cash_balance=1_000_000, asset_quantity=3), ]) - repo.reconcile_universe("default", {"pf-A"}) + repo.reconcile_universe("default", {"pf-A": "SYM"}) slot_a = repo.find_slot_by_portfolio("default", "pf-A") pending = svc.transit_slot(slot_a.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="test", symbol="005930") diff --git a/tests/application/test_update_thresholds_handler.py b/tests/application/test_update_thresholds_handler.py index 7bb81fb..c89e551 100644 --- a/tests/application/test_update_thresholds_handler.py +++ b/tests/application/test_update_thresholds_handler.py @@ -45,7 +45,7 @@ def _holding(service, repo, *, symbol="005930", stop_loss_price=None, take_profit_price=None, forced_close_at=None): from uuid import uuid4 pid = f"pf-{uuid4().hex[:8]}" - repo.reconcile_universe("default", {pid}) + repo.reconcile_universe("default", {pid: "SYM"}) slot = repo.find_slot_by_portfolio("default", pid) pending = service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, @@ -111,7 +111,7 @@ def test_explicit_null_clears(handler, service, repo): def test_non_holding_ignored(handler, service, repo, watchdog): """AC-6: non-holding slot → ignored + not_holding reason, no watchdog reset.""" - repo.reconcile_universe("default", {"pf-inactive"}) + repo.reconcile_universe("default", {"pf-inactive": "SYM"}) inactive = repo.find_slot_by_portfolio("default", "pf-inactive") resp = handler(_req(slot_id=str(inactive.slot_id), stop_loss_price=50000)) assert resp.payload["status"] == "ignored" diff --git a/tests/domain/slot/test_execution_policy_lifecycle.py b/tests/domain/slot/test_execution_policy_lifecycle.py index 54c7e84..c15d5f9 100644 --- a/tests/domain/slot/test_execution_policy_lifecycle.py +++ b/tests/domain/slot/test_execution_policy_lifecycle.py @@ -24,7 +24,7 @@ def service(repo): def _seed(repo): - repo.reconcile_universe(INST, {"PF-1"}) + repo.reconcile_universe(INST, {"PF-1": "SYM"}) return repo.list_slots(INST)[0] diff --git a/tests/domain/slot/test_price_threshold_fields.py b/tests/domain/slot/test_price_threshold_fields.py index a82e4a9..bd68e25 100644 --- a/tests/domain/slot/test_price_threshold_fields.py +++ b/tests/domain/slot/test_price_threshold_fields.py @@ -24,7 +24,7 @@ def service(repo): def _inactive_slot(repo): - repo.reconcile_universe(INST, {"pf-1"}) + repo.reconcile_universe(INST, {"pf-1": "SYM"}) return repo.find_slot_by_portfolio(INST, "pf-1") diff --git a/tests/domain/slot/test_service.py b/tests/domain/slot/test_service.py index daa1abb..93f9f7e 100644 --- a/tests/domain/slot/test_service.py +++ b/tests/domain/slot/test_service.py @@ -36,7 +36,7 @@ def service(repo): def _seed(repo, n=1): """Create n permanent asset slots (inactive) via universe reconcile.""" - repo.reconcile_universe(INST, {f"pf-{i}" for i in range(n)}) + repo.reconcile_universe(INST, {f"pf-{i}": "SYM" for i in range(n)}) return sorted(repo.list_slots(INST), key=lambda s: s.portfolio_id) diff --git a/tests/domain/slot/test_service_sell.py b/tests/domain/slot/test_service_sell.py index 558b642..0490309 100644 --- a/tests/domain/slot/test_service_sell.py +++ b/tests/domain/slot/test_service_sell.py @@ -25,7 +25,7 @@ def service(repo): def _inactive_slot(repo): - repo.reconcile_universe(INST, {"pf-1"}) + repo.reconcile_universe(INST, {"pf-1": "SYM"}) return repo.find_slot_by_portfolio(INST, "pf-1") diff --git a/tests/domain/slot/test_universe_reconcile.py b/tests/domain/slot/test_universe_reconcile.py index 37b69c5..43a4614 100644 --- a/tests/domain/slot/test_universe_reconcile.py +++ b/tests/domain/slot/test_universe_reconcile.py @@ -15,17 +15,17 @@ def repo(): def test_creates_asset_slots_idempotently(repo): - r1 = repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + r1 = repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) assert r1["created"] == 2 # idempotent — second pass creates nothing - r2 = repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + r2 = repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) assert r2["created"] == 0 assert repo.find_slot_by_portfolio(INST, "pf-a").status is SlotState.INACTIVE def test_soft_retire_inactive_asset_that_left_universe(repo): - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) - r = repo.reconcile_universe(INST, {"pf-a"}) # pf-b left + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) + r = repo.reconcile_universe(INST, {"pf-a": "SYM"}) # pf-b left assert r["retired"] == 1 slot_b = repo.find_slot_by_portfolio(INST, "pf-b") assert slot_b.retired_at is not None # soft-retired, row preserved @@ -33,28 +33,28 @@ def test_soft_retire_inactive_asset_that_left_universe(repo): def test_tombstone_holding_asset_that_left_universe(repo): svc = SlotService(repo, instance_key=INST) - repo.reconcile_universe(INST, {"pf-a"}) + repo.reconcile_universe(INST, {"pf-a": "SYM"}) slot = repo.find_slot_by_portfolio(INST, "pf-a") svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, reason="r", actor="a", symbol="005930") svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="r", actor="a", avg_price=70_000) - r = repo.reconcile_universe(INST, set()) # pf-a left while holding + r = repo.reconcile_universe(INST, {}) # pf-a left while holding assert r["tombstoned"] == 1 and r["retired"] == 0 held = repo.find_slot_by_portfolio(INST, "pf-a") assert held.status is SlotState.HOLDING and held.retired_at is None # protected def test_allow_retire_false_skips_retire(repo): - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) - r = repo.reconcile_universe(INST, {"pf-a"}, allow_retire=False) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) + r = repo.reconcile_universe(INST, {"pf-a": "SYM"}, allow_retire=False) assert r["retired"] == 0 assert repo.find_slot_by_portfolio(INST, "pf-b").retired_at is None def test_count_active(repo): svc = SlotService(repo, instance_key=INST) - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) assert repo.count_active(INST) == 0 slot = repo.find_slot_by_portfolio(INST, "pf-a") svc.transit_slot(slot.slot_id, slot.status, SlotState.PENDING_BUY, @@ -63,8 +63,8 @@ def test_count_active(repo): def test_instance_isolation(repo): - repo.reconcile_universe("inst-1", {"pf-a"}) - repo.reconcile_universe("inst-2", {"pf-a"}) + repo.reconcile_universe("inst-1", {"pf-a": "SYM"}) + repo.reconcile_universe("inst-2", {"pf-a": "SYM"}) # same portfolio_id, different instances -> two distinct slots assert repo.find_slot_by_portfolio("inst-1", "pf-a").slot_id != \ repo.find_slot_by_portfolio("inst-2", "pf-a").slot_id diff --git a/tests/runtime/test_reset.py b/tests/runtime/test_reset.py index afacfe0..77b0554 100644 --- a/tests/runtime/test_reset.py +++ b/tests/runtime/test_reset.py @@ -17,7 +17,7 @@ def _seed(dsn): """Create asset slots + drive one into holding so there are rows to wipe.""" repo = PostgreSQLSlotRepository(dsn) svc = SlotService(repo, instance_key=INST) - repo.reconcile_universe(INST, {"pf-a", "pf-b"}) + repo.reconcile_universe(INST, {"pf-a": "SYM", "pf-b": "SYM"}) s = repo.find_slot_by_portfolio(INST, "pf-a") svc.transit_slot(s.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="t", symbol="005930", qty=10) @@ -70,7 +70,7 @@ def test_reset_usable_for_admission_after_wipe(pg_clean): run_reset(pg_clean, dry_run=False) repo = PostgreSQLSlotRepository(pg_clean) svc = SlotService(repo, instance_key=INST) - repo.reconcile_universe(INST, {"pf-x"}) + repo.reconcile_universe(INST, {"pf-x": "SYM"}) s = repo.find_slot_by_portfolio(INST, "pf-x") assert s.status is SlotState.INACTIVE svc.transit_slot(s.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 3f1815b..64bf925 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -118,7 +118,7 @@ class PostgreSQLSlotRepository: return [self._row_to_slot(r) for r in cur.fetchall()] def reconcile_universe( - self, instance_key: str, universe_portfolio_ids: set[str], + self, instance_key: str, universe: dict[str, str], *, allow_retire: bool = True, ) -> dict[str, int]: now = datetime.now(tz=timezone.utc) @@ -131,24 +131,38 @@ class PostgreSQLSlotRepository: (instance_key,), ) existing = {r[0]: (r[1], r[2]) for r in cur.fetchall()} + # backfill symbol on slots that were created before this field was populated + cur.execute( + "SELECT portfolio_id FROM daily_slot " + "WHERE instance_key = %s AND portfolio_id IS NOT NULL AND symbol IS NULL", + (instance_key,), + ) + null_symbol_pids = {r[0] for r in cur.fetchall()} + for pid in null_symbol_pids: + if pid in universe: + cur.execute( + "UPDATE daily_slot SET symbol = %s, updated_at = %s " + "WHERE instance_key = %s AND portfolio_id = %s AND symbol IS NULL", + (universe[pid], now, instance_key, pid), + ) # create new asset slots (idempotent via partial unique) - for pid in universe_portfolio_ids: + for pid, symbol in universe.items(): if pid in existing: continue cur.execute( "INSERT INTO daily_slot " - "(slot_id, status, instance_key, portfolio_id, " - "created_at, updated_at) VALUES (%s, 'inactive', %s, %s, %s, %s) " + "(slot_id, status, symbol, instance_key, portfolio_id, " + "created_at, updated_at) VALUES (%s, 'inactive', %s, %s, %s, %s, %s) " "ON CONFLICT (instance_key, portfolio_id) " "WHERE portfolio_id IS NOT NULL DO NOTHING", - (uuid4(), instance_key, pid, now, now), + (uuid4(), symbol, instance_key, pid, now, now), ) created += cur.rowcount # retire / tombstone assets that left the universe (skipped under # the mass-disappearance guard, dcy-020/OQ-2 D3) if allow_retire: for pid, (status, retired_at) in existing.items(): - if pid in universe_portfolio_ids: + if pid in universe: continue if status in ("holding", "pending_buy", "pending_sell"): tombstoned += 1 diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index df35f58..0ebddf7 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -128,7 +128,11 @@ class SnapshotScheduler: # ---- asset-centric slot reconcile (dcy-020/REQ-058, BAT-028) ---------- def _reconcile_slots(self, items: list) -> None: - universe = {it.portfolio_id for it in items if getattr(it, "portfolio_id", None)} + universe = { + it.portfolio_id: it.symbol + for it in items + if getattr(it, "portfolio_id", None) + } if not universe: _log.warning("slot reconcile skipped: empty universe (degrade guard)") return diff --git a/trex_core/domain/slot/in_memory_repository.py b/trex_core/domain/slot/in_memory_repository.py index f6db5bb..ee39596 100644 --- a/trex_core/domain/slot/in_memory_repository.py +++ b/trex_core/domain/slot/in_memory_repository.py @@ -84,7 +84,7 @@ class InMemorySlotRepository: ] def reconcile_universe( - self, instance_key: str, universe_portfolio_ids: set[str], + self, instance_key: str, universe: dict[str, str], *, allow_retire: bool = True, ) -> dict[str, int]: existing = { @@ -93,13 +93,18 @@ class InMemorySlotRepository: } now = datetime.now(tz=timezone.utc) created = retired = tombstoned = 0 + # backfill symbol on slots that were created before this field was populated + for s in self._slots.values(): + if (s.instance_key == instance_key and s.portfolio_id in universe + and s.symbol is None): + self._slots[s.slot_id] = replace(s, symbol=universe[s.portfolio_id], updated_at=now) # create asset slots for newly-appeared universe assets (idempotent) - for pid in universe_portfolio_ids: + for pid, symbol in universe.items(): if pid not in existing: sid = uuid4() self._slots[sid] = Slot( slot_id=sid, status=SlotState.INACTIVE, - instance_key=instance_key, portfolio_id=pid, + instance_key=instance_key, portfolio_id=pid, symbol=symbol, created_at=now, updated_at=now, ) created += 1 @@ -107,7 +112,7 @@ class InMemorySlotRepository: # mass-disappearance guard, dcy-020/OQ-2 D3 — caller passes allow_retire=False) if allow_retire: for pid, s in existing.items(): - if pid in universe_portfolio_ids: + if pid in universe: continue if s.status in self._ACTIVE: tombstoned += 1 # holding/pending — keep (safety-sell), no retire diff --git a/trex_core/domain/slot/repository.py b/trex_core/domain/slot/repository.py index 758ed73..06b536e 100644 --- a/trex_core/domain/slot/repository.py +++ b/trex_core/domain/slot/repository.py @@ -66,13 +66,14 @@ class SlotRepository(Protocol): ... def reconcile_universe( - self, instance_key: str, universe_portfolio_ids: set[str], + self, instance_key: str, universe: dict[str, str], *, allow_retire: bool = True, ) -> dict[str, int]: """Reconcile the slot set to the portfolio-hub universe (dcy-020/REQ-058). + universe: {portfolio_id -> symbol} mapping from PortfolioItem. - new portfolio_id (no slot for instance) -> create an inactive asset slot - (idempotent; keyed by (instance_key, portfolio_id), permanent). + with symbol set (idempotent; keyed by (instance_key, portfolio_id), permanent). - slot whose portfolio_id left the universe AND is inactive AND not yet retired -> soft-retire (set retired_at). - slot that left the universe but is holding/pending -> tombstone (left