spb/crypto-trade-bot

1"""Ingestion worker: pair discovery (hourly), OHLC accumulation (every N s), WS streaming, retention."""23from __future__ import annotations45import asyncio6import time78import redis.asyncio as aioredis910from ktrade.config import Settings11from ktrade.db.pool import create_pool12from ktrade.ingestion.backfill import backfill_all13from ktrade.ingestion.pairs import refresh_pairs14from ktrade.ingestion.stream import StreamIngestor15from ktrade.kraken.models import PairMeta16from ktrade.kraken.rest import KrakenREST17from ktrade.logging import get_logger1819log = get_logger("worker.ingest")202122async def run_worker(s: Settings) -> None:23    pool = await create_pool(s.database_url, max_size=8)24    r = aioredis.from_url(s.redis_url, decode_responses=True)25    rest = KrakenREST(s.kraken_rest_url, tier=s.kraken_tier)  # public endpoints only — no credentials in the ingest worker26    pairs: dict[str, PairMeta] = {}27    while not pairs:28        try:29            pairs = await refresh_pairs(rest, pool, s.universe)30        except Exception as e:  # noqa: BLE00131            log.warning("worker.pairs.retry", error=str(e))32            await asyncio.sleep(10)33    # Kraken WS v2 accepts a single ohlc interval per symbol per connection → stream 1-min bars only; REST accumulates the rest.34    stream = StreamIngestor(pool, r, s.kraken_ws_url, pairs, s.book_depth, [1])3536    async def backfill_loop() -> None:37        while True:38            t0 = time.time()39            try:40                res = await backfill_all(rest, pool, pairs, s.candle_intervals)41                await r.set("ktrade:backfill:last", str(time.time()))42                log.info("worker.backfill.done", pairs=len(pairs), rows=sum(res.values()), seconds=round(time.time() - t0, 1))43            except Exception as e:  # noqa: BLE00144                log.warning("worker.backfill.error", error=str(e))45            await asyncio.sleep(max(5, s.backfill_poll_seconds - (time.time() - t0)))4647    async def pairs_loop() -> None:48        while True:49            await asyncio.sleep(3600)50            try:51                new = await refresh_pairs(rest, pool, s.universe)52                pairs.update(new)53            except Exception as e:  # noqa: BLE00154                log.warning("worker.pairs.error", error=str(e))5556    async def retention_loop() -> None:57        while True:58            try:59                await pool.execute("DELETE FROM book_samples WHERE ts < $1", int(time.time()) - 14 * 86400)60                await pool.execute("DELETE FROM trades WHERE ts < $1", time.time() - 30 * 86400)61            except Exception as e:  # noqa: BLE00162                log.warning("worker.retention.error", error=str(e))63            await asyncio.sleep(3600)6465    log.info("worker.started", universe=list(pairs), intervals=s.candle_intervals)66    await asyncio.gather(backfill_loop(), pairs_loop(), retention_loop(), stream.run())67