spb/crypto-trade-bot

1"""Read-side queries for the market API (candles, ticker, book, trades, data quality, features)."""23from __future__ import annotations45import time6from typing import Any78import asyncpg9import numpy as np10import pandas as pd1112from ktrade.db.pool import rows_to_dicts131415async def list_pairs(pool: asyncpg.Pool, universe_only: bool = True, q: str | None = None) -> list[dict[str, Any]]:16    sql = "SELECT pair_id, altname, wsname, base, quote, status, ordermin, costmin, tick_size, pair_decimals, lot_decimals, cost_decimals, leverage_buy, leverage_sell, in_universe, updated_at FROM pairs"17    cond, args = [], []18    if universe_only:19        cond.append("in_universe")20    if q:21        args.append(f"%{q.upper()}%")22        cond.append(f"(wsname ILIKE ${len(args)} OR altname ILIKE ${len(args)})")23    if cond:24        sql += " WHERE " + " AND ".join(cond)25    sql += " ORDER BY in_universe DESC, wsname LIMIT 500"26    out = rows_to_dicts(await pool.fetch(sql, *args))27    for r in out:28        for k in ("ordermin", "costmin", "tick_size"):29            r[k] = float(r[k])30    return out313233async def get_candles(pool: asyncpg.Pool, symbol: str, interval: int, limit: int = 1000, start: int | None = None, end: int | None = None, closed_only: bool = False) -> list[dict[str, Any]]:34    cond = ["symbol=$1", "interval_min=$2"]35    args: list[Any] = [symbol, interval]36    if start is not None:37        args.append(start)38        cond.append(f"ts >= ${len(args)}")39    if end is not None:40        args.append(end)41        cond.append(f"ts <= ${len(args)}")42    if closed_only:43        cond.append("closed")44    args.append(min(max(limit, 1), 20000))45    rows = await pool.fetch(f"SELECT ts, open, high, low, close, vwap, volume, count, closed FROM candles WHERE {' AND '.join(cond)} ORDER BY ts DESC LIMIT ${len(args)}", *args)46    return [dict(r) for r in reversed(rows)]474849async def candles_frame(pool: asyncpg.Pool, symbol: str, interval: int, start: int | None = None, end: int | None = None) -> pd.DataFrame:50    rows = await get_candles(pool, symbol, interval, limit=20000, start=start, end=end, closed_only=True)51    if not rows:52        return pd.DataFrame(columns=["ts", "open", "high", "low", "close", "vwap", "volume", "count"])53    df = pd.DataFrame(rows).drop(columns=["closed"])54    return df.sort_values("ts").reset_index(drop=True)555657async def ticker(pool: asyncpg.Pool, symbol: str | None = None) -> list[dict[str, Any]]:58    if symbol:59        rows = await pool.fetch("SELECT symbol, payload, received_at FROM ticker_latest WHERE symbol=$1", symbol)60    else:61        rows = await pool.fetch("SELECT symbol, payload, received_at FROM ticker_latest ORDER BY symbol")62    now = time.time()63    return [{**r["payload"], "received_at": r["received_at"], "age_s": round(now - r["received_at"], 1)} for r in rows]646566async def book(pool: asyncpg.Pool, symbol: str) -> dict[str, Any] | None:67    r = await pool.fetchrow("SELECT payload, received_at, valid FROM book_latest WHERE symbol=$1", symbol)68    if not r:69        return None70    return {**r["payload"], "received_at": r["received_at"], "age_s": round(time.time() - r["received_at"], 1), "valid": r["valid"]}717273async def recent_trades(pool: asyncpg.Pool, symbol: str, limit: int = 100) -> list[dict[str, Any]]:74    rows = await pool.fetch("SELECT trade_id, ts, price, volume, side, order_type FROM trades WHERE symbol=$1 ORDER BY ts DESC LIMIT $2", symbol, min(limit, 1000))75    return rows_to_dicts(rows)767778async def book_samples(pool: asyncpg.Pool, symbol: str, seconds: int = 3600) -> list[dict[str, Any]]:79    since = int(time.time()) - seconds80    rows = await pool.fetch("SELECT ts, best_bid, best_ask, spread_bps, bid_depth, ask_depth, imbalance FROM book_samples WHERE symbol=$1 AND ts>=$2 ORDER BY ts", symbol, since)81    return rows_to_dicts(rows)828384async def data_quality(pool: asyncpg.Pool, stale_after: int) -> list[dict[str, Any]]:85    rows = await pool.fetch(86        """SELECT c.symbol, c.interval_min, count(*) AS bars, min(ts) AS first_ts, max(ts) AS last_ts,87                  (SELECT count(*) FROM data_gaps g WHERE g.symbol=c.symbol AND g.interval_min=c.interval_min) AS gaps,88                  (SELECT coalesce(sum(missing_bars),0) FROM data_gaps g WHERE g.symbol=c.symbol AND g.interval_min=c.interval_min) AS missing_bars89           FROM candles c WHERE closed GROUP BY c.symbol, c.interval_min ORDER BY c.symbol, c.interval_min"""90    )91    now = int(time.time())92    out = []93    for r in rows:94        d = dict(r)95        iv = d["interval_min"]96        expected_last = (now // (iv * 60) - 1) * iv * 6097        d["lag_bars"] = max(0, (expected_last - d["last_ts"]) // (iv * 60))98        d["stale"] = d["lag_bars"] * iv * 60 > max(stale_after, iv * 60 * 2)99        d["coverage_days"] = round((d["last_ts"] - d["first_ts"]) / 86400, 2)100        out.append(d)101    return out102103104def realized_vol(df: pd.DataFrame, interval_min: int, window: int = 30) -> dict[str, float | None]:105    if len(df) < window + 2:106        return {"rv_window": None, "rv_annualized": None}107    r = np.log(df["close"]).diff().dropna().tail(window)108    per_year = 365 * 24 * 60 / interval_min109    rv = float(r.std(ddof=1))110    return {"rv_window": rv, "rv_annualized": rv * np.sqrt(per_year)}111112113async def overview(pool: asyncpg.Pool, universe: list[str], stale_after: int) -> dict[str, Any]:114    tick = {t["symbol"]: t for t in await ticker(pool)}115    out = []116    for s in universe:117        t = tick.get(s)118        df = await candles_frame(pool, s, 60)119        rv = realized_vol(df, 60, 24 * 7) if not df.empty else {"rv_window": None, "rv_annualized": None}120        out.append({121            "symbol": s, "last": t.get("last") if t else None, "change_pct": t.get("change_pct") if t else None,122            "bid": t.get("bid") if t else None, "ask": t.get("ask") if t else None, "volume": t.get("volume") if t else None,123            "spread_bps": ((t["ask"] - t["bid"]) / ((t["ask"] + t["bid"]) / 2) * 1e4) if t and t.get("bid") and t.get("ask") else None,124            "age_s": t.get("age_s") if t else None, "stale": (t is None) or (t["age_s"] > stale_after), "rv_7d_annualized": rv["rv_annualized"],125            "hourly_bars": int(len(df)),126        })127    return {"symbols": out}128