Preserve ledger venue provenance (#39)

* feat: preserve ledger venue provenance * fix: satisfy ledger test lint ratchet * fix: advance image migration inventory
2026-09-19 19:30:32 +09:00
parent 567d662913
commit 996127f215
23 changed files with 768 additions and 35 deletions
@@ -12,10 +12,13 @@ from __future__ import annotations

 import os
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
+from threading import Barrier

 import pytest

+from app.repositories.portfolio_repository import LedgerTradeConflictError
 from app.repositories.postgres import PortfolioDBConfig, PostgresPortfolioRepository


@@ -133,6 +136,182 @@ def test_duplicate_broker_write_is_idempotent(repo, seeded_portfolio):
        assert cur.fetchone()[0] == 1


+def test_duplicate_venue_enrichment_is_metadata_only(repo, seeded_portfolio):
+    args = {
+        "portfolio_id": seeded_portfolio,
+        "order_id": "ord-venue-enrich",
+        "execution_id": "exec-venue-enrich",
+        "side": "buy",
+        "quantity": 5.0,
+        "price": 10_000.0,
+        "executed_at": _now(),
+    }
+    _first, created = repo.insert_ledger_trade(**args)
+    assert created is True
+    before = repo.get(seeded_portfolio)
+
+    enriched, created = repo.insert_ledger_trade(
+        **args,
+        execution_venue="NXT",
+        order_venue="SOR",
+    )
+
+    assert created is False
+    assert enriched.execution_venue == "NXT"
+    assert enriched.order_venue == "SOR"
+    after = repo.get(seeded_portfolio)
+    assert after.cash_balance == before.cash_balance
+    assert after.asset_quantity == before.asset_quantity
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            """
+            SELECT COUNT(*), MIN(execution_venue), MIN(order_venue)
+            FROM portfolio_ledger
+            WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s
+            """,
+            (seeded_portfolio, args["order_id"], args["execution_id"]),
+        )
+        assert cur.fetchone() == (1, "NXT", "SOR")
+
+
+def test_duplicate_venue_conflict_is_atomic(repo, seeded_portfolio):
+    args = {
+        "portfolio_id": seeded_portfolio,
+        "order_id": "ord-venue-conflict",
+        "execution_id": "exec-venue-conflict",
+        "side": "buy",
+        "quantity": 1.0,
+        "price": 10_000.0,
+        "executed_at": _now(),
+    }
+    repo.insert_ledger_trade(**args, execution_venue="KRX")
+    before = repo.get(seeded_portfolio)
+
+    with pytest.raises(LedgerTradeConflictError):
+        repo.insert_ledger_trade(
+            **args,
+            execution_venue="NXT",
+            order_venue="SOR",
+        )
+
+    after = repo.get(seeded_portfolio)
+    assert after.cash_balance == before.cash_balance
+    assert after.asset_quantity == before.asset_quantity
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            """
+            SELECT execution_venue, order_venue
+            FROM portfolio_ledger
+            WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s
+            """,
+            (seeded_portfolio, args["order_id"], args["execution_id"]),
+        )
+        assert cur.fetchone() == ("KRX", None)
+
+
+def test_duplicate_null_venues_do_not_erase_existing_values(repo, seeded_portfolio):
+    args = {
+        "portfolio_id": seeded_portfolio,
+        "order_id": "ord-venue-preserve",
+        "execution_id": "exec-venue-preserve",
+        "side": "buy",
+        "quantity": 1.0,
+        "price": 10_000.0,
+        "executed_at": _now(),
+    }
+    repo.insert_ledger_trade(
+        **args,
+        execution_venue="KRX",
+        order_venue="SOR",
+    )
+
+    duplicate, created = repo.insert_ledger_trade(**args)
+
+    assert created is False
+    assert duplicate.execution_venue == "KRX"
+    assert duplicate.order_venue == "SOR"
+
+
+def test_duplicate_economic_conflict_is_rejected(repo, seeded_portfolio):
+    args = {
+        "portfolio_id": seeded_portfolio,
+        "order_id": "ord-economic-conflict",
+        "execution_id": "exec-economic-conflict",
+        "side": "buy",
+        "quantity": 1.0,
+        "price": 10_000.0,
+        "executed_at": _now(),
+    }
+    repo.insert_ledger_trade(**args)
+
+    with pytest.raises(LedgerTradeConflictError):
+        repo.insert_ledger_trade(**{**args, "quantity": 2.0}, order_venue="SOR")
+
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            """
+            SELECT quantity::float8, order_venue
+            FROM portfolio_ledger
+            WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s
+            """,
+            (seeded_portfolio, args["order_id"], args["execution_id"]),
+        )
+        assert cur.fetchone() == (1.0, None)
+
+
+def test_concurrent_conflicting_venue_enrichment_has_no_last_write_wins(
+    repo, seeded_portfolio
+):
+    args = {
+        "portfolio_id": seeded_portfolio,
+        "order_id": "ord-venue-concurrent",
+        "execution_id": "exec-venue-concurrent",
+        "side": "buy",
+        "quantity": 1.0,
+        "price": 10_000.0,
+        "executed_at": _now(),
+    }
+    repo.insert_ledger_trade(**args)
+    barrier = Barrier(2)
+
+    def enrich(execution_venue: str):
+        worker_repo = PostgresPortfolioRepository(repo._config)
+        barrier.wait()
+        try:
+            entry, created = worker_repo.insert_ledger_trade(
+                portfolio_id=seeded_portfolio,
+                order_id="ord-venue-concurrent",
+                execution_id="exec-venue-concurrent",
+                side="buy",
+                quantity=1.0,
+                price=10_000.0,
+                executed_at=args["executed_at"],
+                execution_venue=execution_venue,
+            )
+            return ("ok", created, entry.execution_venue)
+        except LedgerTradeConflictError:
+            return ("conflict", False, None)
+        finally:
+            worker_repo.close()
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = list(executor.map(enrich, ("KRX", "NXT")))
+
+    assert sorted(result[0] for result in results) == ["conflict", "ok"]
+    winner = next(result[2] for result in results if result[0] == "ok")
+    assert winner in {"KRX", "NXT"}
+    with repo._connect().cursor() as cur:
+        cur.execute(
+            """
+            SELECT COUNT(*), MIN(execution_venue), MIN(order_venue)
+            FROM portfolio_ledger
+            WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s
+            """,
+            (seeded_portfolio, args["order_id"], args["execution_id"]),
+        )
+        assert cur.fetchone() == (1, winner, None)
+
+
 def test_negative_balance_allowed(repo, seeded_portfolio):
    """REQ-037 AC-3 + OQ-3: oversell produces negative asset_quantity without rejection."""
    entry, created = repo.insert_ledger_trade(