spb/crypto-trade-bot

1"""Incremental OHLC accumulation.23Kraken REST /public/OHLC returns at most the 720 most recent bars per (pair, interval) — older data cannot4be retrieved from the exchange, so history is *accumulated* by polling: each poll stores closed bars, and the5still-forming last bar is stored with closed=false (and overwritten on the next poll). Coverage therefore grows6from first deployment onward (720 × interval backwards from the first run). Gaps are detected and recorded.7"""89from __future__ import annotations1011import time1213import asyncpg1415from ktrade.kraken.models import PairMeta16from ktrade.kraken.rest import KrakenREST17from ktrade.logging import get_logger1819log = get_logger("ingest.backfill")202122async def upsert_candles(conn: asyncpg.Connection, symbol: str, interval: int, rows: list[tuple], source: str) -> int:23    if not rows:24        return 025    await conn.executemany(26        """INSERT INTO candles (symbol, interval_min, ts, open, high, low, close, vwap, volume, count, closed, source, ingested_at)27           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12, now())28           ON CONFLICT (symbol, interval_min, ts) DO UPDATE SET29             open=EXCLUDED.open, high=EXCLUDED.high, low=EXCLUDED.low, close=EXCLUDED.close, vwap=EXCLUDED.vwap,30             volume=EXCLUDED.volume, count=EXCLUDED.count, closed=EXCLUDED.closed OR candles.closed, source=EXCLUDED.source, ingested_at=now()31           WHERE candles.closed = false OR EXCLUDED.closed = true""",32        [(symbol, interval, *r, source) for r in rows],33    )34    return len(rows)353637async def backfill_pair(rest: KrakenREST, pool: asyncpg.Pool, pair: PairMeta, interval: int) -> int:38    key = f"ohlc:{pair.wsname}:{interval}"39    state = await pool.fetchval("SELECT value FROM ingestion_state WHERE key=$1", key)40    since = int(state["last"]) if state and state.get("last") else None41    candles, last = await rest.ohlc(pair.pair_id, interval, since)42    if not candles:43        return 044    now = int(time.time())45    rows = []46    for c in candles:47        closed = c.closed_at(interval) <= now48        rows.append((c.ts, c.open, c.high, c.low, c.close, c.vwap, c.volume, c.count, closed))49    async with pool.acquire() as conn, conn.transaction():50        n = await upsert_candles(conn, pair.wsname, interval, rows, "rest")51        await conn.execute(52            "INSERT INTO ingestion_state (key, value, updated_at) VALUES ($1, $2, now()) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()",53            key, {"last": last, "polled_at": now, "rows": n, "first_ts": min(r[0] for r in rows)},54        )55    return n565758async def detect_gaps(pool: asyncpg.Pool, symbol: str, interval: int) -> int:59    """Record missing bars between consecutive stored bars. Kraken omits bars with zero trades on illiquid pairs,60    so a 'gap' may be genuine inactivity — they are reported, not silently filled."""61    step = interval * 6062    rows = await pool.fetch(63        """WITH s AS (SELECT ts, lag(ts) OVER (ORDER BY ts) AS prev FROM candles WHERE symbol=$1 AND interval_min=$2 AND closed)64           SELECT prev, ts FROM s WHERE prev IS NOT NULL AND ts - prev > $3""",65        symbol, interval, step,66    )67    n = 068    async with pool.acquire() as conn:69        for r in rows:70            missing = (r["ts"] - r["prev"]) // step - 171            await conn.execute(72                "INSERT INTO data_gaps (symbol, interval_min, gap_start, gap_end, missing_bars) VALUES ($1,$2,$3,$4,$5) ON CONFLICT (symbol, interval_min, gap_start) DO UPDATE SET gap_end=EXCLUDED.gap_end, missing_bars=EXCLUDED.missing_bars",73                symbol, interval, r["prev"] + step, r["ts"] - step, int(missing),74            )75            n += 176    return n777879async def backfill_all(rest: KrakenREST, pool: asyncpg.Pool, pairs: dict[str, PairMeta], intervals: list[int]) -> dict[str, int]:80    out: dict[str, int] = {}81    for sym, p in pairs.items():82        for iv in intervals:83            try:84                out[f"{sym}:{iv}"] = await backfill_pair(rest, pool, p, iv)85                await detect_gaps(pool, sym, iv)86            except Exception as e:  # noqa: BLE00187                log.warning("ingest.backfill.error", symbol=sym, interval=iv, error=str(e))88    return out89