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

fix(reconcile): symbol을 슬롯 생성 시 함께 저장 + NULL 슬롯 backfill

Browse Source 
reconcile_universe 시그니처를 set[str] → dict[str, str] (portfolio_id→symbol)로 변경.
기존 symbol=NULL 슬롯은 다음 reconcile 실행 시 자동 backfill.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
genichin
2026-06-11 21:08:46 +09:00
co-authored by Claude Sonnet 4.6
parent 221a16342e
commit 24a6017457
29 changed files with 92 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapters/messaging/test_rabbitmq_intent_e2e.py
+1 -1
View File
@@ -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]}"
tests/adapters/persistence/test_postgresql_repository.py
+10 -10
View File
@@ -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,
tests/application/test_asset_centric_admission.py
+8 -8
View File
@@ -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"
tests/application/test_execution.py
+1 -1
View File
@@ -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
tests/application/test_execution_cancel.py
+1 -1
View File
@@ -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
tests/application/test_execution_ledger.py
+1 -1
View File
@@ -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
tests/application/test_execution_policy.py
+1 -1
View File
@@ -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
tests/application/test_execution_policy_handler.py
+2 -2
View File
@@ -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):
tests/application/test_holding_divergence.py
+1 -1
View File
@@ -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)
tests/application/test_intent_concurrency.py
+1 -1
View File
@@ -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
tests/application/test_killswitch_execution.py
+1 -1
View File
@@ -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,
tests/application/test_ledger_option_d.py
+1 -1
View File
@@ -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"]
tests/application/test_messaging_event_publisher.py
+1 -1
View File
@@ -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 "_")
tests/application/test_messaging_intent_consumer.py
+1 -1
View File
@@ -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
tests/application/test_safety_sell_visibility.py
+1 -1
View File
@@ -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)
tests/application/test_safety_sell_worker.py
+1 -1
View File
@@ -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,
tests/application/test_sell_pricing.py
+1 -1
View File
@@ -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
tests/application/test_slots_handler.py
+2 -2
View File
@@ -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")
tests/application/test_update_thresholds_handler.py
+2 -2
View File
@@ -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"
tests/domain/slot/test_execution_policy_lifecycle.py
+1 -1
View File
@@ -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]
tests/domain/slot/test_price_threshold_fields.py
+1 -1
View File
@@ -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")
tests/domain/slot/test_service.py
+1 -1
View File
@@ -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)
tests/domain/slot/test_service_sell.py
+1 -1
View File
@@ -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")
tests/domain/slot/test_universe_reconcile.py
+11 -11
View File
@@ -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
tests/runtime/test_reset.py
+2 -2
View File
@@ -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,
trex_core/adapters/persistence/postgresql.py
+20 -6
View File
@@ -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
trex_core/application/snapshot_scheduler.py
+5 -1
View File
@@ -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
trex_core/domain/slot/in_memory_repository.py
+9 -4
View File
@@ -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
trex_core/domain/slot/repository.py
+3 -2
View File
@@ -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