spb/crypto-trade-bot

1import os23import asyncpg4import pytest56from ktrade.db.pool import create_pool, dsn, migrate7from ktrade.ingestion.backfill import detect_gaps, upsert_candles8from ktrade.ingestion.stream import _parse_ts9from ktrade.kraken.models import normalize_symbol1011pytestmark = pytest.mark.skipif(os.environ.get("KTRADE_SKIP_DB") == "1", reason="db tests disabled")121314def test_symbol_normalisation():15    assert normalize_symbol("XBT/USD") == "BTC/USD"16    assert normalize_symbol("XDG/EUR") == "DOGE/EUR"17    assert normalize_symbol("ETH/USD") == "ETH/USD"181920def test_rfc3339_nanoseconds():21    assert abs(_parse_ts("2026-10-22T03:11:19.123456789Z") - 1792638679.123456) < 1e-3222324async def test_closed_flag_never_regresses_and_gaps_detected():25    url = os.environ["KTRADE_DATABASE_URL"]26    conn = await asyncpg.connect(dsn(url))27    await conn.execute("DROP SCHEMA public CASCADE; CREATE SCHEMA public")28    await conn.close()29    await migrate(url)30    pool = await create_pool(url)31    try:32        async with pool.acquire() as c:33            await upsert_candles(c, "T/USD", 1, [(60, 1, 1, 1, 1, 1, 1, 1, True), (120, 1, 1, 1, 2, 1, 1, 1, False)], "rest")34            # a later poll reports bar 120 as closed with the final close → overwritten35            await upsert_candles(c, "T/USD", 1, [(120, 1, 1, 1, 3, 1, 1, 1, True)], "ws")36            # a stale 'forming' snapshot must NOT overwrite a closed bar37            await upsert_candles(c, "T/USD", 1, [(120, 1, 1, 1, 9, 1, 1, 1, False)], "rest")38            # gap: bars 180, 240 missing39            await upsert_candles(c, "T/USD", 1, [(300, 1, 1, 1, 1, 1, 1, 1, True)], "rest")40        rows = await pool.fetch("SELECT ts, close, closed FROM candles WHERE symbol='T/USD' ORDER BY ts")41        assert [(r["ts"], r["close"], r["closed"]) for r in rows] == [(60, 1.0, True), (120, 3.0, True), (300, 1.0, True)]42        assert await detect_gaps(pool, "T/USD", 1) == 143        g = await pool.fetchrow("SELECT gap_start, gap_end, missing_bars FROM data_gaps")44        assert (g["gap_start"], g["gap_end"], g["missing_bars"]) == (180, 240, 2)45    finally:46        await pool.close()47