fix: enforce actual venue persistence contract
This commit is contained in:
+2
-1
@@ -153,7 +153,8 @@
|
||||
- side :: TEXT CHECK ('buy','sell')
|
||||
- qty / price / fee / tax :: NUMERIC(18,4)
|
||||
- executed_at :: TIMESTAMPTZ?
|
||||
- venue :: TEXT NOT NULL DEFAULT 'KRX' :: 실제 제출 거래소. 기존 행은 KRX로 보존한다.
|
||||
- venue :: TEXT? CHECK (NULL 또는 'KRX'/'NXT') :: broker-authoritative 실제 체결 거래소. 과거 SOR row처럼 실제 거래소를 복원할 수 없으면 NULL이며 임의 추론하지 않는다.
|
||||
- order_venue :: TEXT NOT NULL DEFAULT 'KRX' CHECK ('KRX'/'NXT'/'SOR') :: 원주문/라우팅 venue. 취소·재시작 감사에 사용한다.
|
||||
- status :: TEXT CHECK ('pending','completed','conflict_blocked','delivery_failed') — `delivery_failed` 추가(dcy-015, 마이그레이션 007)
|
||||
- attempts :: INTEGER :: 재시도 누적
|
||||
- last_outcome :: TEXT? CHECK ('completed','duplicate_completed','retryable_failure','conflict_blocked','permanent_failure','dead_lettered')
|
||||
|
||||
@@ -8,9 +8,14 @@ from __future__ import annotations
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from trex_core.adapters.persistence.migrate import apply_migrations, migration_files
|
||||
from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository
|
||||
from trex_core.adapters.persistence.postgresql import (
|
||||
PostgreSQLLedgerOutbox,
|
||||
PostgreSQLSlotRepository,
|
||||
)
|
||||
from trex_core.application.ledger_port import LedgerTrade
|
||||
from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
|
||||
|
||||
_EXPECTED_TABLES = {
|
||||
@@ -91,3 +96,64 @@ def test_migration_replay_converts_existing_auto_venue_row_to_sor(pg_clean):
|
||||
"UPDATE daily_slot SET venue = 'AUTO' WHERE slot_id = %s",
|
||||
(slot.slot_id,),
|
||||
)
|
||||
|
||||
|
||||
def test_migration_024_separates_legacy_sor_and_is_replay_safe(pg_clean):
|
||||
outbox = PostgreSQLLedgerOutbox(pg_clean)
|
||||
normal = LedgerTrade(
|
||||
asset_id="PF-NORMAL",
|
||||
order_id="O-NORMAL",
|
||||
execution_id="E-NORMAL",
|
||||
side="buy",
|
||||
qty=1,
|
||||
price=1000,
|
||||
executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc),
|
||||
venue="NXT",
|
||||
order_venue="KRX",
|
||||
)
|
||||
legacy = LedgerTrade(
|
||||
asset_id="PF-LEGACY",
|
||||
order_id="O-LEGACY",
|
||||
execution_id="E-LEGACY",
|
||||
side="buy",
|
||||
qty=1,
|
||||
price=1000,
|
||||
executed_at=datetime(2026, 8, 19, tzinfo=timezone.utc),
|
||||
venue="KRX",
|
||||
order_venue="SOR",
|
||||
)
|
||||
outbox.enqueue(normal)
|
||||
outbox.enqueue(legacy)
|
||||
|
||||
# Recreate a pre-024 SOR row: one venue column carried routing provenance,
|
||||
# and no separate order_venue was available.
|
||||
with psycopg.connect(pg_clean) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("ALTER TABLE ledger_outbox DROP CONSTRAINT ledger_outbox_venue_check")
|
||||
cur.execute("ALTER TABLE ledger_outbox ALTER COLUMN order_venue DROP NOT NULL")
|
||||
cur.execute(
|
||||
"UPDATE ledger_outbox SET venue='SOR', order_venue=NULL "
|
||||
"WHERE correlation_id=%s",
|
||||
(legacy.correlation_id,),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
apply_migrations(pg_clean)
|
||||
apply_migrations(pg_clean)
|
||||
|
||||
normal_row = outbox.get(normal.correlation_id)
|
||||
legacy_row = outbox.get(legacy.correlation_id)
|
||||
assert normal_row is not None
|
||||
assert normal_row.trade.venue == "NXT"
|
||||
assert normal_row.trade.order_venue == "KRX"
|
||||
assert legacy_row is not None
|
||||
assert legacy_row.trade.venue is None
|
||||
assert legacy_row.trade.order_venue == "SOR"
|
||||
|
||||
with psycopg.connect(pg_clean) as conn:
|
||||
with conn.cursor() as cur:
|
||||
with pytest.raises(psycopg.errors.CheckViolation):
|
||||
cur.execute(
|
||||
"UPDATE ledger_outbox SET venue='SOR' WHERE correlation_id=%s",
|
||||
(normal.correlation_id,),
|
||||
)
|
||||
|
||||
@@ -1,6 +1,18 @@
|
||||
-- GitHub #61: keep the original submission venue separate from actual execution venue.
|
||||
-- GitHub #61: keep original submission venue separate from actual execution venue.
|
||||
-- The nullable add + NULL-only backfill makes replay safe: rows written after
|
||||
-- this migration already have order_venue and are never reinterpreted.
|
||||
ALTER TABLE ledger_outbox
|
||||
ADD COLUMN IF NOT EXISTS order_venue TEXT NOT NULL DEFAULT 'KRX';
|
||||
ADD COLUMN IF NOT EXISTS order_venue TEXT;
|
||||
|
||||
UPDATE ledger_outbox
|
||||
SET order_venue = venue
|
||||
WHERE order_venue IS NULL
|
||||
AND venue IN ('KRX', 'NXT', 'SOR');
|
||||
|
||||
ALTER TABLE ledger_outbox
|
||||
ALTER COLUMN order_venue SET DEFAULT 'KRX';
|
||||
ALTER TABLE ledger_outbox
|
||||
ALTER COLUMN order_venue SET NOT NULL;
|
||||
|
||||
ALTER TABLE ledger_outbox
|
||||
DROP CONSTRAINT IF EXISTS ledger_outbox_order_venue_check;
|
||||
@@ -8,5 +20,14 @@ ALTER TABLE ledger_outbox
|
||||
ADD CONSTRAINT ledger_outbox_order_venue_check
|
||||
CHECK (order_venue IN ('KRX', 'NXT', 'SOR'));
|
||||
|
||||
-- Historical rows used venue for both concepts. Preserve that provenance.
|
||||
UPDATE ledger_outbox SET order_venue = venue WHERE order_venue = 'KRX' AND venue <> 'KRX';
|
||||
-- Historical SOR in `venue` recorded routing provenance, not a proven KRX/NXT
|
||||
-- execution exchange. Preserve SOR in order_venue and mark actual as unknown.
|
||||
ALTER TABLE ledger_outbox
|
||||
ALTER COLUMN venue DROP NOT NULL;
|
||||
UPDATE ledger_outbox SET venue = NULL WHERE venue = 'SOR';
|
||||
|
||||
ALTER TABLE ledger_outbox
|
||||
DROP CONSTRAINT IF EXISTS ledger_outbox_venue_check;
|
||||
ALTER TABLE ledger_outbox
|
||||
ADD CONSTRAINT ledger_outbox_venue_check
|
||||
CHECK (venue IS NULL OR venue IN ('KRX', 'NXT'));
|
||||
|
||||
@@ -71,8 +71,8 @@ class OrderAck:
|
||||
accepted: bool
|
||||
broker_order_id: Optional[str] = None
|
||||
reason: Optional[str] = None
|
||||
# Actual venue used for submission. Execution persists this value with the
|
||||
# broker order id so restart/cancel never has to infer the original venue.
|
||||
# Original submission/routing venue. Execution persists this with the
|
||||
# broker order id so restart/cancel never infer it from wall-clock time.
|
||||
venue: Optional[str] = None
|
||||
raw: dict[str, Any] = field(default_factory=dict)
|
||||
submitted_at: Optional[datetime] = None
|
||||
|
||||
@@ -57,8 +57,9 @@ class LedgerTrade:
|
||||
fee: float = 0.0
|
||||
tax: float = 0.0
|
||||
executed_at: Optional[datetime] = None
|
||||
# Broker-authoritative execution exchange (KRX/NXT).
|
||||
venue: str = "KRX"
|
||||
# Broker-authoritative execution exchange (KRX/NXT). Historical rows whose
|
||||
# SOR execution exchange cannot be reconstructed are None.
|
||||
venue: Optional[str] = None
|
||||
# Original submission/routing venue (KRX/NXT/SOR), retained for audit.
|
||||
order_venue: str = "KRX"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user