spb/sp-wallet

1"""Read-only access to the engine's PostgreSQL table23    candles(source, symbol, interval, ts, open, high, low, close, volume)   PK (source, symbol, interval, ts)45as specified in docs/v3/ARCHITECTURE.md ("Market data"). `ts` is the bar-open time in UTC. The session is opened6read-only: the research zone never writes market data.7"""8from __future__ import annotations910import os1112import pandas as pd1314from .core import COLUMNS, DataError, to_utc1516TABLE = "candles"171819def read_pg(database_url: str, symbol: str, interval: str, start=None, end=None, source: str | None = None):20    """Return (raw frame with ts + OHLCV, source used).2122    `source` defaults to SPQ_CANDLE_SOURCE; if neither is set and the table holds exactly one source for this23    (symbol, interval), that one is used; with several sources the call fails and lists them (mixing sources24    silently would hide cross-source deviations).25    """26    import psycopg2728    source = source or os.environ.get("SPQ_CANDLE_SOURCE")29    with psycopg.connect(database_url, options="-c default_transaction_read_only=on") as conn:30        with conn.cursor() as cur:31            if source is None:32                cur.execute(33                    f"SELECT DISTINCT source FROM {TABLE} WHERE symbol = %s AND interval = %s ORDER BY 1",34                    (symbol.upper(), interval),35                )36                sources = [r[0] for r in cur.fetchall()]37                if len(sources) != 1:38                    raise DataError(39                        f"{symbol} {interval}: {len(sources)} sources in {TABLE} ({sources}); pass source= "40                        "or set SPQ_CANDLE_SOURCE"41                    )42                source = sources[0]43            sql = f"SELECT ts, open, high, low, close, volume FROM {TABLE} WHERE source = %s AND symbol = %s AND interval = %s"44            args: list = [source, symbol.upper(), interval]45            if start is not None:46                sql += " AND ts >= %s"47                args.append(to_utc(start).to_pydatetime())48            if end is not None:49                sql += " AND ts < %s"50                args.append(to_utc(end).to_pydatetime())51            cur.execute(sql + " ORDER BY ts", args)52            rows = cur.fetchall()53    if not rows:54        raise DataError(f"{symbol} {interval}: no rows in {TABLE} for source {source!r} in the requested range")55    df = pd.DataFrame(rows, columns=["ts", *COLUMNS])56    for c in COLUMNS:  # NUMERIC columns arrive as Decimal57        df[c] = df[c].astype("float64")58    return df, source59