genichin
996127f215
* feat: preserve ledger venue provenance * fix: satisfy ledger test lint ratchet * fix: advance image migration inventory
328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""Real-DB integration tests for BAT-016 ledger consolidation (REQ-036/037).
|
|
|
|
These exercise `insert_ledger_trade` against an actual PostgreSQL/TimescaleDB
|
|
instance (docker-compose.local-test). The mock-based suite cannot catch DB-level
|
|
defects such as the hypertable `ON CONFLICT` regression that BAT-016 originally
|
|
shipped; this module closes that gap.
|
|
|
|
Skipped automatically when the local DB is unreachable.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
def _db_config() -> PortfolioDBConfig:
|
|
return PortfolioDBConfig(
|
|
host=os.environ.get("TIMESCALEDB_HOST", "localhost"),
|
|
port=int(os.environ.get("TIMESCALEDB_PORT", "5432")),
|
|
dbname=os.environ.get("TIMESCALEDB_NAME", "portfolio_hub"),
|
|
user=os.environ.get("TIMESCALEDB_USER", "portfolio_hub_user"),
|
|
password=os.environ.get("TIMESCALEDB_PASSWORD", "change-me"),
|
|
sslmode=os.environ.get("TIMESCALEDB_SSLMODE", "prefer"),
|
|
connect_timeout_sec=3,
|
|
application_name="ledger-integration-test",
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def repo():
|
|
config = _db_config()
|
|
repository = PostgresPortfolioRepository(config)
|
|
try:
|
|
repository.ensure_schema()
|
|
except Exception as exc: # pragma: no cover - environment dependent
|
|
pytest.skip(f"local DB unreachable: {exc}")
|
|
yield repository
|
|
if callable(getattr(type(repository), "close", None)):
|
|
repository.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def seeded_portfolio(repo):
|
|
"""Seed a unique asset + portfolio, yield portfolio_id, then clean up."""
|
|
pid = f"itest-{uuid.uuid4()}"
|
|
symbol = f"ITEST-{uuid.uuid4().hex[:8]}"
|
|
conn = repo._connect()
|
|
with conn.transaction():
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO assets (symbol, markets) VALUES (%s, ARRAY[]::text[]) RETURNING asset_id",
|
|
(symbol,),
|
|
)
|
|
asset_id = cur.fetchone()[0]
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO portfolios (portfolio_id, name, cash_balance, asset_quantity, asset_id, broker, status)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
""",
|
|
(pid, "itest-pf", 1_000_000.0, 0.0, asset_id, "itest-broker", "enabled"),
|
|
)
|
|
yield pid
|
|
conn = repo._connect()
|
|
with conn.transaction():
|
|
with conn.cursor() as cur:
|
|
cur.execute("DELETE FROM portfolio_ledger WHERE portfolio_id = %s", (pid,))
|
|
cur.execute("DELETE FROM portfolios WHERE portfolio_id = %s", (pid,))
|
|
cur.execute("DELETE FROM assets WHERE symbol = %s", (symbol,))
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def test_new_trade_updates_portfolio_and_returns_state(repo, seeded_portfolio):
|
|
"""REQ-037 AC-1: new broker write applies accounting and returns post-trade state."""
|
|
entry, created = repo.insert_ledger_trade(
|
|
portfolio_id=seeded_portfolio,
|
|
order_id="ord-1",
|
|
execution_id="exec-1",
|
|
side="buy",
|
|
quantity=10.0,
|
|
price=80_000.0,
|
|
executed_at=_now(),
|
|
fee_amount=100.0,
|
|
tax_amount=0.0,
|
|
)
|
|
assert created is True
|
|
assert entry.cash_balance_after == pytest.approx(1_000_000.0 - 800_000.0 - 100.0)
|
|
assert entry.asset_quantity_after == pytest.approx(10.0)
|
|
|
|
portfolio = repo.get(seeded_portfolio)
|
|
assert float(portfolio.cash_balance) == pytest.approx(199_900.0)
|
|
assert float(portfolio.asset_quantity) == pytest.approx(10.0)
|
|
|
|
|
|
def test_duplicate_broker_write_is_idempotent(repo, seeded_portfolio):
|
|
"""REQ-037 AC-2 + OQ-4(b): duplicate does not re-insert; returns current authoritative state."""
|
|
args = dict(
|
|
portfolio_id=seeded_portfolio,
|
|
order_id="ord-dup",
|
|
execution_id="exec-dup",
|
|
side="buy",
|
|
quantity=5.0,
|
|
price=10_000.0,
|
|
executed_at=_now(),
|
|
fee_amount=0.0,
|
|
tax_amount=0.0,
|
|
)
|
|
first, created1 = repo.insert_ledger_trade(**args)
|
|
assert created1 is True
|
|
assert first.cash_balance_after == pytest.approx(950_000.0)
|
|
|
|
second, created2 = repo.insert_ledger_trade(**args)
|
|
assert created2 is False
|
|
# duplicate returns CURRENT authoritative portfolio state (OQ-4 b)
|
|
assert second.cash_balance_after == pytest.approx(950_000.0)
|
|
assert second.asset_quantity_after == pytest.approx(5.0)
|
|
|
|
# exactly one ledger row for this broker key
|
|
conn = repo._connect()
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT COUNT(*) FROM portfolio_ledger WHERE portfolio_id = %s AND order_id = %s AND execution_id = %s",
|
|
(seeded_portfolio, "ord-dup", "exec-dup"),
|
|
)
|
|
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(
|
|
portfolio_id=seeded_portfolio,
|
|
order_id="ord-neg",
|
|
execution_id="exec-neg",
|
|
side="sell",
|
|
quantity=999.0,
|
|
price=1_000.0,
|
|
executed_at=_now(),
|
|
)
|
|
assert created is True
|
|
assert entry.asset_quantity_after == pytest.approx(-999.0)
|