Add integration tests for execution persistence, metrics, and opportunity writing

- Implemented integration tests for the execution writer to ensure trade orders and PnL are persisted correctly.
- Created integration tests for the metrics calculator to summarize execution data accurately.
- Added integration tests for the opportunity writer to verify event persistence.
- Established PostgreSQL schema validation tests to ensure all expected tables, columns, and constraints exist.
- Removed outdated unit tests that relied on DuckDB and replaced them with tests using PgStore.

tests/integration/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for PostgreSQL schema and connectivity."""

tests/integration/conftest.py

@@ -0,0 +1,25 @@
+"""pytest configuration for integration tests.
+
+Integration tests require a live PostgreSQL server at the configured host.
+They are skipped automatically if the server is unreachable.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+
+def pytest_ignore_collect(path: str, config: pytest.Config) -> bool:
+    """Skip integration tests unless --integration is passed."""
+    if "integration" in path and not config.getoption("--integration", False):
+        return True
+    return False
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    parser.addoption(
+        "--integration",
+        action="store_true",
+        default=False,
+        help="Run integration tests (requires PostgreSQL)",
+    )

tests/integration/test_audit_repository.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from datetime import UTC, datetime
+
+import pytest
+import pytest_asyncio
+
+from arbitrade.config.settings import get_settings
+from arbitrade.storage.pg_store import PgStore
+from arbitrade.storage.repositories import AuditRecord, AuditRepository
+
+pytestmark = pytest.mark.integration
+
+
+@asynccontextmanager
+async def _pg() -> AsyncIterator[PgStore]:
+    s = get_settings()
+    store = PgStore(s)
+    try:
+        await store.start()
+        await store.migrate()
+        yield store
+    finally:
+        await store.stop()
+
+
+@pytest.mark.asyncio
+async def test_audit_repository_inserts_and_lists_recent() -> None:
+    async with _pg() as store:
+        repository = AuditRepository(store)
+
+        await repository.insert(
+            AuditRecord(
+                occurred_at=datetime.now(UTC),
+                actor="dashboard_user",
+                event_type="dashboard.control.start",
+                decision="approved",
+                payload={"execution_status": "running"},
+                correlation_id="req-1",
+            )
+        )
+
+        recent = await repository.list_recent(limit=5)
+
+        assert len(recent) == 1
+        assert recent[0].actor == "dashboard_user"
+        assert recent[0].event_type == "dashboard.control.start"
+        assert recent[0].decision == "approved"
+        assert recent[0].payload == {"execution_status": "running"}
+        assert recent[0].correlation_id == "req-1"

tests/integration/test_dashboard.py

@@ -0,0 +1,384 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import httpx
+
+from arbitrade.api.app import create_app
+from arbitrade.config.settings import Settings
+
+
+class _FakeAlertNotifier:
+    def __init__(self) -> None:
+        self.events: list[dict[str, Any]] = []
+
+    async def notify(
+        self,
+        *,
+        category: str,
+        severity: str,
+        title: str,
+        message: str,
+        details: dict[str, str] | None = None,
+    ) -> bool:
+        self.events.append(
+            {
+                "category": category,
+                "severity": severity,
+                "title": title,
+                "message": message,
+                "details": details or {},
+            }
+        )
+        return True
+
+
+def _seed_metrics_data(app) -> None:
+    store = app.state.store
+    started = datetime.now(UTC)
+    finished = started + timedelta(seconds=20)
+    with store.connect() as conn:
+        conn.execute(
+            """
+            INSERT INTO trades (
+                trade_ref,
+                started_at,
+                finished_at,
+                status,
+                realized_pnl,
+                estimated_pnl,
+                capital_used,
+                cycle,
+                leg_count
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [
+                "trade-1",
+                started,
+                finished,
+                "filled",
+                15.0,
+                10.0,
+                100.0,
+                "USD->BTC->ETH->USD",
+                3,
+            ],
+        )
+        conn.execute(
+            """
+            INSERT INTO opportunities (
+                detected_at,
+                cycle,
+                gross_pct,
+                net_pct,
+                est_profit,
+                executed
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            [started, "USD->BTC->ETH->USD", 4.0, 3.0, 0.03, True],
+        )
+        conn.execute(
+            """
+            INSERT INTO orders (
+                trade_ref,
+                order_ref,
+                leg_index,
+                pair,
+                side,
+                volume,
+                user_ref,
+                status,
+                filled_volume,
+                avg_price,
+                raw_response,
+                recorded_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [
+                "trade-1",
+                "order-1",
+                0,
+                "BTC/USD",
+                "buy",
+                2.0,
+                100,
+                "closed",
+                2.0,
+                100.0,
+                "{}",
+                started,
+            ],
+        )
+        conn.execute(
+            """
+            INSERT INTO portfolio_snapshots (
+                snapshot_at,
+                balances,
+                total_value_usd
+            ) VALUES (?, ?, ?)
+            """,
+            [started, '{"USD": 1000.0, "BTC": 0.25}', 1250.0],
+        )
+        conn.execute(
+            """
+            INSERT INTO trades (
+                trade_ref,
+                started_at,
+                finished_at,
+                status,
+                realized_pnl,
+                estimated_pnl,
+                capital_used,
+                cycle,
+                leg_count
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            [
+                "trade-open",
+                started,
+                None,
+                "open",
+                None,
+                5.0,
+                50.0,
+                "USD->BTC->ETH->USD",
+                3,
+            ],
+        )
+
+
+async def test_dashboard_page_and_fragment_and_sse(tmp_path) -> None:
+    app = create_app(Settings(DUCKDB_PATH=tmp_path / "dash.duckdb"))
+    _seed_metrics_data(app)
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        page = await client.get("/")
+        fragment = await client.get("/dashboard/fragment/metrics")
+        stream = await client.get("/dashboard/stream/metrics")
+        overview = await client.get("/dashboard/fragment/overview")
+        overview_stream = await client.get("/dashboard/stream/overview")
+        controls = await client.get("/dashboard/fragment/controls")
+        charts = await client.get("/dashboard/fragment/charts")
+        audit = await client.get("/dashboard/fragment/audit")
+
+    assert page.status_code == 200
+    assert "EventSource" in page.text
+    assert "alpinejs" in page.text.lower()
+    assert "Chart.js" in page.text or "chart.umd.min.js" in page.text
+    assert 'hx-get="/dashboard/fragment/metrics"' in page.text
+    assert 'hx-get="/dashboard/fragment/controls"' in page.text
+    assert 'hx-get="/dashboard/fragment/charts"' in page.text
+
+    assert fragment.status_code == 200
+    assert "Realized P&L" in fragment.text
+    assert "15.00 USD" in fragment.text
+    assert "100.0%" in fragment.text
+
+    assert stream.status_code == 200
+    assert stream.headers["content-type"].startswith("text/event-stream")
+    assert "event: metrics" in stream.text
+    assert "Realized P&L" in stream.text
+
+    assert overview.status_code == 200
+    assert "live" in overview.text
+    assert "Balances Snapshot" in overview.text
+    assert "Open Trades" in overview.text
+    assert "Opportunity Feed" in overview.text
+    assert "1250.00 USD" in overview.text
+    assert "trade-open" in overview.text
+
+    assert overview_stream.status_code == 200
+    assert overview_stream.headers["content-type"].startswith("text/event-stream")
+    assert "event: overview" in overview_stream.text
+    assert "trade-open" in overview_stream.text
+
+    assert controls.status_code == 200
+    assert "Runtime Status" in controls.text
+    assert "running" in controls.text
+    assert "Alerting" in controls.text
+    assert "Last result" in controls.text
+    assert "Paper trading" in controls.text
+    assert "Tradable pairs" in controls.text
+    assert "Strategy mode" in controls.text
+
+    assert charts.status_code == 200
+    assert "Opportunity Trend" in charts.text
+    assert "opportunity-chart" in charts.text
+    assert "Hide chart" in charts.text or "Show chart" in charts.text
+
+    assert audit.status_code == 200
+    assert "Audit Trail" in audit.text
+
+
+async def test_dashboard_controls_update_runtime_state_and_config(tmp_path) -> None:
+    app = create_app(Settings(DUCKDB_PATH=tmp_path / "controls.duckdb"))
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        stop_response = await client.post("/dashboard/control/stop")
+        start_response = await client.post("/dashboard/control/start")
+        kill_response = await client.post(
+            "/dashboard/control/kill-switch",
+            data={"reason": "manual"},
+        )
+        config_response = await client.post(
+            "/dashboard/control/config",
+            data={
+                "trade_capital_usd": "250.50",
+                "max_trade_capital_usd": "300.00",
+                "max_concurrent_trades": "4",
+                "tradable_pairs": "BTC/USD, ETH/BTC, BTC/USD",
+                "strategy_mode": "paper",
+                "strategy_profit_threshold": "0.0025",
+                "strategy_max_depth_levels": "7",
+                "paper_trading_mode": "on",
+            },
+        )
+
+    assert stop_response.status_code == 200
+    assert ">stopped<" in stop_response.text
+
+    assert start_response.status_code == 200
+    assert ">running<" in start_response.text
+
+    assert kill_response.status_code == 200
+    assert ">active<" in kill_response.text
+    assert "manual" in kill_response.text
+
+    assert config_response.status_code == 200
+    assert "250.50 USD" in config_response.text
+    assert "300.00 USD" in config_response.text
+    assert "4" in config_response.text
+    assert "BTC/USD, ETH/BTC" in config_response.text
+    assert "paper" in config_response.text
+    assert "0.002500" in config_response.text
+    assert "7" in config_response.text
+    assert app.state.settings.trade_capital_usd == 250.5
+    assert app.state.settings.max_trade_capital_usd == 300.0
+    assert app.state.settings.max_concurrent_trades == 4
+    assert app.state.settings.paper_trading_mode is True
+    assert app.state.dashboard_controls.tradable_pairs == ["BTC/USD", "ETH/BTC"]
+    assert app.state.dashboard_controls.strategy_mode == "paper"
+    assert app.state.dashboard_controls.strategy_profit_threshold == 0.0025
+    assert app.state.dashboard_controls.strategy_max_depth_levels == 7
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        audit_recent = await client.get("/dashboard/api/audit/recent")
+
+    assert audit_recent.status_code == 200
+    entries = audit_recent.json()["entries"]
+    assert len(entries) >= 4
+    assert any(entry["event_type"] == "dashboard.control.stop" for entry in entries)
+    assert any(entry["event_type"] == "dashboard.control.start" for entry in entries)
+    assert any(entry["event_type"] == "dashboard.control.kill_switch" for entry in entries)
+    assert any(entry["event_type"] == "dashboard.control.config" for entry in entries)
+
+
+async def test_dashboard_controls_emit_alerts(tmp_path) -> None:
+    app = create_app(Settings(DUCKDB_PATH=tmp_path / "alerts.duckdb"))
+    fake_notifier = _FakeAlertNotifier()
+    app.state.alert_notifier = fake_notifier
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        await client.post("/dashboard/control/start")
+        await client.post("/dashboard/control/stop")
+        await client.post("/dashboard/control/kill-switch", data={"reason": "manual-test"})
+
+    assert len(fake_notifier.events) == 3
+    assert fake_notifier.events[0]["title"] == "Execution started"
+    assert fake_notifier.events[1]["title"] == "Execution stopped"
+    assert fake_notifier.events[2]["title"] == "Kill switch activated"
+    assert fake_notifier.events[2]["details"]["reason"] == "manual-test"
+
+
+async def test_dashboard_requires_basic_auth_when_configured(tmp_path) -> None:
+    app = create_app(
+        Settings(
+            DUCKDB_PATH=tmp_path / "auth.duckdb",
+            DASHBOARD_AUTH_USERNAME="admin",
+            DASHBOARD_AUTH_PASSWORD="secret",
+        )
+    )
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        unauthenticated = await client.get("/dashboard/fragment/overview")
+        authenticated = await client.get(
+            "/dashboard/fragment/overview",
+            auth=("admin", "secret"),
+        )
+        health = await client.get("/health")
+
+    assert unauthenticated.status_code == 401
+    assert unauthenticated.headers["www-authenticate"] == 'Basic realm="Arbitrade Dashboard"'
+    assert authenticated.status_code == 200
+    assert health.status_code == 200
+
+
+async def test_dashboard_alert_status_api_exposes_notifier_snapshot(tmp_path) -> None:
+    app = create_app(Settings(DUCKDB_PATH=tmp_path / "alerts-status.duckdb"))
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/dashboard/api/alerts/status")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["enabled"] is True
+    assert "configured_channels" in payload
+    assert "last_result" in payload
+
+
+async def test_backtesting_page_run_and_recent_reports_api(tmp_path) -> None:
+    app = create_app(Settings(DUCKDB_PATH=tmp_path / "backtesting-ui.duckdb"))
+
+    events_file = tmp_path / "replay.jsonl"
+    events_file.write_text(
+        "\n".join(
+            [
+                '{"timestamp":"2026-06-01T12:00:00Z","symbol":"BTC/USD","bids":[[99.5,10.0]],"asks":[[100.0,10.0]]}',
+                '{"timestamp":"2026-06-01T12:00:01Z","symbol":"ETH/BTC","bids":[[0.051,10.0]],"asks":[[0.050,10.0]]}',
+                '{"timestamp":"2026-06-01T12:00:02Z","symbol":"ETH/USD","bids":[[110.0,10.0]],"asks":[[110.5,10.0]]}',
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        page = await client.get("/dashboard/backtesting")
+        fragment = await client.get("/dashboard/fragment/backtesting")
+        run = await client.post(
+            "/dashboard/backtesting/run",
+            data={
+                "source": "db",
+                "starting_balances": "USD=1000.0",
+                "trade_capital": "100.0",
+                "min_profit_threshold": "0.0005",
+                "fee_profile": "api",
+                "slippage_bps": "4.0",
+                "execution_latency_ms": "20.0",
+            },
+        )
+        reports = await client.get("/dashboard/api/backtesting/reports")
+
+    assert page.status_code == 200
+    assert "Backtesting" in page.text
+    assert "/dashboard/fragment/backtesting" in page.text
+
+    assert fragment.status_code == 200
+    assert "Run Backtest" in fragment.text
+    assert "Recent Jobs" in fragment.text
+
+    assert run.status_code == 200
+    assert "submitted" in run.text
+    assert "queued" in run.text
+
+    assert reports.status_code == 200
+    payload = reports.json()
+    assert len(payload["reports"]) >= 1
+    assert payload["reports"][0]["status"] == "pending"

tests/integration/test_execution_persistence.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+import pytest
+
+from arbitrade.config.settings import get_settings
+from arbitrade.detection.engine import OpportunityEvent
+from arbitrade.execution.sequencer import TriangularExecutionSequencer
+from arbitrade.storage.pg_store import PgStore
+from arbitrade.storage.executions import AsyncExecutionWriter
+from arbitrade.storage.repositories import OrderRepository, PnLRepository, TradeRepository
+
+pytestmark = pytest.mark.integration
+
+
+@dataclass(slots=True)
+class _FakeRestClient:
+    calls: int = 0
+
+    async def place_market_order(self, *, pair: str, side: str, volume: float) -> dict[str, object]:
+        self.calls += 1
+        return {"txid": [f"tx-{self.calls}"], "status": "submitted"}
+
+
+def _sample_event() -> OpportunityEvent:
+    return OpportunityEvent(
+        detected_at=datetime.now(UTC),
+        cycle="USD->BTC->ETH->USD",
+        updated_pair="BTC/USD",
+        gross_rate=1.04,
+        net_rate=1.03,
+        gross_pct=4.0,
+        net_pct=3.0,
+        est_profit=0.03,
+    )
+
+
+@asynccontextmanager
+async def _pg() -> AsyncIterator[PgStore]:
+    s = get_settings()
+    store = PgStore(s)
+    try:
+        await store.start()
+        await store.migrate()
+        yield store
+    finally:
+        await store.stop()
+
+
+@pytest.mark.asyncio
+async def test_execution_writer_persists_trade_order_and_pnl() -> None:
+    async with _pg() as store:
+        writer = AsyncExecutionWriter(
+            TradeRepository(store),
+            OrderRepository(store),
+            PnLRepository(store),
+            max_queue_size=10,
+        )
+        await writer.start()
+
+        client = _FakeRestClient()
+        sequencer = TriangularExecutionSequencer(
+            client,
+            available_pairs=["BTC/USD", "ETH/BTC", "ETH/USD"],
+            execution_writer=writer,
+        )
+
+        result = await sequencer.execute(_sample_event())
+        await writer.stop()
+
+        assert result.success
+        assert client.calls == 3
+
+        async with store.pool.acquire() as conn:
+            trades = await conn.fetch(
+                "SELECT trade_ref, status, estimated_pnl, capital_used, cycle, leg_count FROM trades"
+            )
+            orders = await conn.fetch(
+                "SELECT trade_ref, order_ref, leg_index, pair, side, volume, status "
+                "FROM orders ORDER BY leg_index"
+            )
+            pnls = await conn.fetch("SELECT trade_ref, kind, pnl_usd, source FROM pnl_events")
+
+        assert len(trades) == 1
+        assert trades[0]["status"] == "filled"
+        assert trades[0]["estimated_pnl"] == 0.03
+        assert trades[0]["capital_used"] == 1.0
+        assert trades[0]["cycle"] == "USD->BTC->ETH->USD"
+        assert trades[0]["leg_count"] == 3
+
+        assert len(orders) == 3
+        assert orders[0]["leg_index"] == 0
+        assert orders[1]["leg_index"] == 1
+        assert orders[2]["leg_index"] == 2
+        assert orders[0]["status"] == "submitted"
+
+        assert len(pnls) == 1
+        assert pnls[0]["kind"] == "estimated"
+        assert pnls[0]["pnl_usd"] == 0.03

tests/integration/test_metrics.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from datetime import UTC, datetime, timedelta
+
+import pytest
+import pytest_asyncio
+
+from arbitrade.config.settings import get_settings
+from arbitrade.metrics import MetricsCalculator
+from arbitrade.storage.pg_store import PgStore
+
+pytestmark = pytest.mark.integration
+
+
+@asynccontextmanager
+async def _pg() -> AsyncIterator[PgStore]:
+    s = get_settings()
+    store = PgStore(s)
+    try:
+        await store.start()
+        await store.migrate()
+        yield store
+    finally:
+        await store.stop()
+
+
+@pytest.mark.asyncio
+async def test_metrics_calculator_summarizes_execution_data() -> None:
+    async with _pg() as store:
+        started = datetime.now(UTC)
+        finished = started + timedelta(seconds=30)
+        started_two = started + timedelta(minutes=1)
+        finished_two = started_two + timedelta(seconds=90)
+
+        async with store.pool.acquire() as conn:
+            await conn.execute(
+                """
+                INSERT INTO trades (
+                    trade_ref, started_at, finished_at, status,
+                    realized_pnl, estimated_pnl, capital_used, cycle, leg_count
+                ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9),
+                         ($10, $11, $12, $13, $14, $15, $16, $17, $18)
+                """,
+                "trade-1", started, finished, "filled", 12.5, 10.0, 100.0, "USD->BTC->ETH->USD", 3,
+                "trade-2", started_two, finished_two, "filled", -
+                4.5, -2.0, 200.0, "USD->ETH->BTC->USD", 3,
+            )
+            await conn.execute(
+                """
+                INSERT INTO opportunities (detected_at, cycle, gross_pct, net_pct, est_profit, executed)
+                VALUES ($1, $2, $3, $4, $5, $6),
+                       ($7, $8, $9, $10, $11, $12),
+                       ($13, $14, $15, $16, $17, $18)
+                """,
+                started, "USD->BTC->ETH->USD", 4.0, 3.0, 0.03, True,
+                started_two, "USD->ETH->BTC->USD", 2.0, 1.0, 0.01, False,
+                started_two +
+                timedelta(
+                    seconds=30), "USD->BTC->ETH->USD", 5.0, 4.0, 0.04, True,
+            )
+            await conn.execute(
+                """
+                INSERT INTO orders (
+                    trade_ref, order_ref, leg_index, pair, side, volume,
+                    user_ref, status, filled_volume, avg_price, raw_response, recorded_at
+                ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12),
+                         ($13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24)
+                """,
+                "trade-1", "order-1", 0, "BTC/USD", "buy", 2.0, 101, "closed", 2.0, 100.0, "{}", started,
+                "trade-2", "order-2", 0, "ETH/USD", "sell", 4.0, 202, "closed", 3.0, 200.0, "{}", started_two,
+            )
+
+        metrics = await MetricsCalculator(store).compute()
+
+        assert metrics.realized_pnl_usd == 8.0
+        assert metrics.win_rate == 0.5
+        assert metrics.avg_trade_duration_seconds == 60.0
+        assert metrics.opportunities_per_minute == 2.0
+        assert metrics.fill_rate == 0.875
+        assert metrics.latency_p50_seconds == 60.0
+        assert metrics.latency_p95_seconds == 87.0
+        assert metrics.latency_p99_seconds == pytest.approx(89.4)

tests/integration/test_opportunity_writer.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from datetime import UTC, datetime
+
+import pytest
+
+from arbitrade.config.settings import get_settings
+from arbitrade.detection.engine import OpportunityEvent
+from arbitrade.storage.pg_store import PgStore
+from arbitrade.storage.opportunities import AsyncOpportunityWriter
+from arbitrade.storage.repositories import OpportunityRepository
+
+pytestmark = pytest.mark.integration
+
+
+@asynccontextmanager
+async def _pg() -> AsyncIterator[PgStore]:
+    s = get_settings()
+    store = PgStore(s)
+    try:
+        await store.start()
+        await store.migrate()
+        yield store
+    finally:
+        await store.stop()
+
+
+@pytest.mark.asyncio
+async def test_async_opportunity_writer_persists_events() -> None:
+    async with _pg() as store:
+        repository = OpportunityRepository(store)
+        writer = AsyncOpportunityWriter(repository, max_queue_size=10)
+        await writer.start()
+
+        event = OpportunityEvent(
+            detected_at=datetime.now(UTC),
+            cycle="USD->BTC->ETH->USD",
+            updated_pair="BTC/USD",
+            gross_rate=1.04,
+            net_rate=1.03,
+            gross_pct=4.0,
+            net_pct=3.0,
+            est_profit=0.03,
+        )
+
+        await writer.enqueue(event)
+        await writer.stop()
+
+        async with store.pool.acquire() as conn:
+            rows = await conn.fetch(
+                "SELECT cycle, gross_pct, net_pct, est_profit, executed FROM opportunities"
+            )
+
+        assert len(rows) == 1
+        assert rows[0]["cycle"] == "USD->BTC->ETH->USD"
+        assert rows[0]["gross_pct"] == 4.0
+        assert rows[0]["net_pct"] == 3.0
+        assert rows[0]["est_profit"] == 0.03
+        assert rows[0]["executed"] is False

tests/integration/test_postgresql_schema.py

@@ -0,0 +1,359 @@
+"""Integration tests: verify PostgreSQL schema and connection.
+
+These tests connect to the PostgreSQL server at 192.168.88.35 and
+validate that all expected tables, columns, and constraints exist.
+They are skipped if the server is unreachable.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import pytest
+import pytest_asyncio
+
+from arbitrade.config.settings import Settings, get_settings
+from arbitrade.storage.pg_store import PgStore
+
+pytestmark = pytest.mark.integration
+
+
+# ── expected schema ──────────────────────────────────────────────────────────
+
+EXPECTED_TABLES: dict[str, list[str]] = {
+    "schema_migrations": ["version", "applied_at"],
+    "config_sections": ["id", "name", "description", "updated_at"],
+    "config_settings": [
+        "key", "section", "value_json", "value_type", "is_secret",
+        "is_runtime_reloadable", "updated_at", "updated_by",
+    ],
+    "config_pairings": [
+        "id", "base_asset", "quote_asset", "enabled", "source",
+        "created_at", "updated_at",
+    ],
+    "config_backtesting_defaults": [
+        "id", "starting_balances", "trade_capital", "min_profit_threshold",
+        "slippage_bps", "execution_latency_ms", "fee_source",
+    ],
+    "opportunities": [
+        "id", "detected_at", "cycle", "gross_pct", "net_pct",
+        "est_profit", "executed",
+    ],
+    "trades": [
+        "id", "trade_ref", "started_at", "finished_at", "status",
+        "realized_pnl", "estimated_pnl", "capital_used", "cycle", "leg_count",
+    ],
+    "orders": [
+        "id", "trade_ref", "order_ref", "leg_index", "pair", "side",
+        "volume", "user_ref", "status", "filled_volume", "avg_price",
+        "raw_response", "recorded_at",
+    ],
+    "pnl_events": [
+        "id", "trade_ref", "recorded_at", "kind", "pnl_usd", "source",
+    ],
+    "portfolio_snapshots": ["snapshot_at", "balances", "total_value_usd"],
+    "market_snapshots": ["snapshot_at", "symbol", "source", "payload", "latency_ms"],
+    "audit_events": [
+        "id", "occurred_at", "actor", "event_type", "decision",
+        "payload", "correlation_id",
+    ],
+    "runtime_state_snapshots": [
+        "snapshot_at", "is_running", "kill_switch_active", "kill_switch_reason",
+        "open_trade_count", "last_known_balances", "note",
+    ],
+    "kraken_account_snapshots": [
+        "snapshot_at", "fee_tier", "maker_fee", "taker_fee",
+        "thirty_day_volume", "trade_balance_raw", "fee_schedule_raw",
+    ],
+    "backtest_jobs": [
+        "id", "status", "events_path", "config", "report", "error",
+        "created_at", "started_at", "finished_at",
+    ],
+}
+
+# Tables that should have a primary key
+TABLES_WITH_PRIMARY_KEY: dict[str, str | list[str]] = {
+    "schema_migrations": "version",
+    "config_sections": "id",
+    "config_settings": "key",
+    "config_pairings": "id",
+    "config_backtesting_defaults": "id",
+    "opportunities": "id",
+    "trades": "id",
+    "orders": "id",
+    "pnl_events": "id",
+    "audit_events": "id",
+    "backtest_jobs": "id",
+}
+
+# Tables with a UNIQUE constraint beyond the primary key
+TABLES_WITH_UNIQUE_CONSTRAINTS: dict[str, list[str]] = {
+    "config_sections": ["name"],
+    "config_pairings": ["base_asset, quote_asset"],
+}
+
+
+# ── fixtures ────────────────────────────────────────────────────────────────
+
+@asynccontextmanager
+async def _pg_lifecycle() -> AsyncIterator[PgStore]:
+    """Connect, yield store, then disconnect."""
+    settings = get_settings()
+    store = PgStore(settings)
+    try:
+        await store.start()
+        yield store
+    finally:
+        await store.stop()
+
+
+@pytest_asyncio.fixture(name="pg")
+async def pg_fixture() -> AsyncIterator[PgStore]:
+    async with _pg_lifecycle() as store:
+        yield store
+
+
+# ── helpers ─────────────────────────────────────────────────────────────────
+
+async def _get_actual_tables(store: PgStore) -> dict[str, list[str]]:
+    """Return {table_name: [column_name, ...]} for the public schema."""
+    actual: dict[str, list[str]] = {}
+    async with store.pool.acquire() as conn:
+        rows = await conn.fetch(
+            "SELECT table_name, column_name FROM information_schema.columns "
+            "WHERE table_schema = 'public' ORDER BY table_name, ordinal_position"
+        )
+    for row in rows:
+        tbl: str = row["table_name"]
+        col: str = row["column_name"]
+        actual.setdefault(tbl, []).append(col)
+    return actual
+
+
+async def _table_row_count(store: PgStore, table: str) -> int:
+    async with store.pool.acquire() as conn:
+        row = await conn.fetchrow(f"SELECT COUNT(*) AS cnt FROM {table}")
+    return int(row["cnt"]) if row else 0
+
+
+# ── tests ───────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_pg_connect(pg: PgStore) -> None:
+    """Can connect to PostgreSQL and ping the server."""
+    async with pg.pool.acquire() as conn:
+        val = await conn.fetchval("SELECT 1 AS val")
+    assert val == 1
+
+
+@pytest.mark.asyncio
+async def test_pgcrypto_extension(pg: PgStore) -> None:
+    """The pgcrypto extension is available (gen_random_uuid)."""
+    async with pg.pool.acquire() as conn:
+        val = await conn.fetchval("SELECT gen_random_uuid()")
+    assert val is not None
+    # The result should be a UUID object
+    assert len(str(val)) == 36  # UUID string length
+
+
+@pytest.mark.asyncio
+async def test_schema_migration_applies(pg: PgStore) -> None:
+    """Migrate creates all expected tables."""
+    await pg.migrate()
+    actual = await _get_actual_tables(pg)
+
+    for table in EXPECTED_TABLES:
+        assert table in actual, (
+            f"Table '{table}' missing after migration. "
+            f"Found tables: {sorted(actual)}"
+        )
+
+
+@pytest.mark.asyncio
+async def test_migration_is_idempotent(pg: PgStore) -> None:
+    """Running migrate twice does not raise."""
+    await pg.migrate()
+    await pg.migrate()  # second call should be a no-op
+    actual = await _get_actual_tables(pg)
+    for table in EXPECTED_TABLES:
+        assert table in actual
+
+
+@pytest.mark.asyncio
+async def test_table_columns(pg: PgStore) -> None:
+    """Every expected table has the correct columns."""
+    await pg.migrate()
+    actual = await _get_actual_tables(pg)
+
+    for table, expected_cols in EXPECTED_TABLES.items():
+        actual_cols = actual.get(table, [])
+        for col in expected_cols:
+            assert col in actual_cols, (
+                f"Column '{col}' missing from table '{table}'. "
+                f"Actual columns: {actual_cols}"
+            )
+
+
+@pytest.mark.asyncio
+async def test_primary_keys(pg: PgStore) -> None:
+    """Tables that should have primary keys do."""
+    await pg.migrate()
+    async with pg.pool.acquire() as conn:
+        for table, expected_pk in TABLES_WITH_PRIMARY_KEY.items():
+            rows = await conn.fetch(
+                "SELECT kcu.column_name FROM information_schema.table_constraints tc "
+                "JOIN information_schema.key_column_usage kcu "
+                "ON tc.constraint_name = kcu.constraint_name "
+                "WHERE tc.table_schema = 'public' AND tc.table_name = $1 "
+                "AND tc.constraint_type = 'PRIMARY KEY' "
+                "ORDER BY kcu.ordinal_position",
+                table,
+            )
+            pk_columns = [r["column_name"] for r in rows]
+            expected_list = [expected_pk] if isinstance(expected_pk, str) else expected_pk
+            for col in expected_list:
+                assert col in pk_columns, (
+                    f"Table '{table}' should have PK column '{col}'. "
+                    f"Actual PK columns: {pk_columns}"
+                )
+
+
+@pytest.mark.asyncio
+async def test_unique_constraints(pg: PgStore) -> None:
+    """Tables that should have UNIQUE constraints do."""
+    await pg.migrate()
+    async with pg.pool.acquire() as conn:
+        for table, expected_ucs in TABLES_WITH_UNIQUE_CONSTRAINTS.items():
+            rows = await conn.fetch(
+                "SELECT kcu.column_name FROM information_schema.table_constraints tc "
+                "JOIN information_schema.key_column_usage kcu "
+                "ON tc.constraint_name = kcu.constraint_name "
+                "WHERE tc.table_schema = 'public' AND tc.table_name = $1 "
+                "AND tc.constraint_type = 'UNIQUE'",
+                table,
+            )
+            uc_columns = {r["column_name"] for r in rows}
+            for expected_cols in expected_ucs:
+                cols = [c.strip() for c in expected_cols.split(",")]
+                for col in cols:
+                    assert col in uc_columns, (
+                        f"Table '{table}' should have UNIQUE column '{col}'. "
+                        f"Actual UNIQUE columns: {uc_columns}"
+                    )
+
+
+@pytest.mark.asyncio
+async def test_table_row_count_is_zero(pg: PgStore) -> None:
+    """All tables start empty after migration."""
+    await pg.migrate()
+    for table in EXPECTED_TABLES:
+        count = await _table_row_count(pg, table)
+        assert count == 0, (
+            f"Table '{table}' should be empty after migration, "
+            f"but has {count} rows"
+        )
+
+
+@pytest.mark.asyncio
+async def test_schema_migration_version_recorded(pg: PgStore) -> None:
+    """schema_migrations has the expected version after migrate."""
+    from arbitrade.storage.pg_store import SCHEMA_VERSION
+
+    await pg.migrate()
+    async with pg.pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "SELECT MAX(version) AS v FROM schema_migrations"
+        )
+    assert row is not None
+    assert row["v"] == SCHEMA_VERSION, (
+        f"Expected schema version {SCHEMA_VERSION}, "
+        f"got {row['v']}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_create_and_query_row(pg: PgStore) -> None:
+    """Can INSERT a row and SELECT it back (round-trip for a simple table)."""
+    await pg.migrate()
+    async with pg.pool.acquire() as conn:
+        # ConfigSections round-trip
+        await conn.execute(
+            "INSERT INTO config_sections (name, description) VALUES ($1, $2)",
+            "test_section", "A test section for integration test",
+        )
+        row = await conn.fetchrow(
+            "SELECT name, description FROM config_sections WHERE name = $1",
+            "test_section",
+        )
+        assert row is not None
+        assert row["name"] == "test_section"
+        assert row["description"] == "A test section for integration test"
+
+        # Clean up
+        await conn.execute(
+            "DELETE FROM config_sections WHERE name = $1",
+            "test_section",
+        )
+
+
+@pytest.mark.asyncio
+async def test_config_pairings_upsert(pg: PgStore) -> None:
+    """ON CONFLICT ... DO UPDATE works on config_pairings (unique constraint)."""
+    await pg.migrate()
+    from arbitrade.config.service import ConfigPairing
+    from arbitrade.storage.repositories import ConfigPairingRepository
+
+    repo = ConfigPairingRepository(pg)
+
+    # Insert
+    p1 = await repo.upsert_pairing(
+        ConfigPairing(base_asset="XBT", quote_asset="USD", enabled=True, source="kraken")
+    )
+    assert p1.id is not None
+    assert p1.base_asset == "XBT"
+    assert p1.enabled is True
+
+    # Upsert (update)
+    p2 = await repo.upsert_pairing(
+        ConfigPairing(base_asset="XBT", quote_asset="USD", enabled=False, source="manual")
+    )
+    assert p2.id == p1.id  # same row
+    assert p2.enabled is False
+    assert p2.source == "manual"
+
+    # Clean up
+    deleted = await repo.delete_pairing("XBT", "USD")
+    assert deleted is True
+
+
+@pytest.mark.asyncio
+async def test_audit_list_recent(pg: PgStore) -> None:
+    """AuditRepository.list_recent returns records in desc order."""
+    await pg.migrate()
+    from datetime import UTC, datetime
+    from arbitrade.storage.repositories import AuditRecord, AuditRepository
+
+    repo = AuditRepository(pg)
+    now = datetime.now(UTC)
+
+    # Insert a few records
+    for i in range(3):
+        await repo.insert(
+            AuditRecord(
+                occurred_at=now,
+                actor="test",
+                event_type="integration_test",
+                decision=f"decision_{i}",
+                payload={"index": i},
+                correlation_id=f"corr_{i}",
+            )
+        )
+
+    recent = await repo.list_recent(limit=5)
+    assert len(recent) >= 3
+    assert recent[0].decision in ("decision_2", "decision_1", "decision_0")
+    # Verify payload serialization worked
+    first = recent[0]
+    if first.payload:
+        assert "index" in first.payload