portfolio-hub/tests/test_ledger_consolidation_integration.py

"""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)