Repository Public
1"""OHLCV access: PostgreSQL `candles` table (production) or local Parquet/CSV files (tests, offline research).23 df, quality = load_candles("ETH", "1h", start="2025-01-01", end="2026-01-01", root="data") # files4 df, quality = load_candles("ETH", "1h", database_url=os.environ["DATABASE_URL"], source="coinbase") # PostgreSQL56Every loader returns closed bars only, UTC, oldest first, de-duplicated and checked for gaps (see core.normalise).7"""8from __future__ import annotations910import os11from pathlib import Path1213import pandas as pd1415from .core import COLUMNS, DataError, Quality, align, bars_per_year, interval_seconds, normalise, to_utc1617__all__ = [18 "COLUMNS", "DataError", "Quality", "align", "bars_per_year", "interval_seconds", "normalise", "to_utc",19 "load_candles", "load_universe", "read_file", "file_path", "cross_source_deviation",20]212223def file_path(root, source: str, symbol: str, interval: str) -> Path:24 """Local layout: <root>/<source>/<SYMBOL>_<interval>.parquet (or .csv)."""25 base = Path(root) / source / f"{symbol.upper()}_{interval}"26 for ext in (".parquet", ".csv"):27 if base.with_suffix(ext).exists():28 return base.with_suffix(ext)29 return base.with_suffix(".parquet")303132def read_file(path) -> pd.DataFrame:33 """Read a raw Parquet or CSV candle file (column `ts` + OHLCV). No checks: use load_candles for that."""34 path = Path(path)35 if not path.exists():36 raise DataError(f"no candle file at {path}")37 if path.suffix == ".parquet":38 df = pd.read_parquet(path)39 return df.reset_index() if "ts" not in df.columns else df40 if path.suffix == ".csv":41 return pd.read_csv(path)42 raise DataError(f"unsupported file type {path.suffix!r} (use .parquet or .csv)")434445def load_candles(46 symbol: str,47 interval: str,48 start=None,49 end=None,50 *,51 source: str | None = None,52 root=None,53 path=None,54 database_url: str | None = None,55 as_of=None,56 on_duplicate: str = "raise",57 on_gap: str = "report",58) -> tuple[pd.DataFrame, Quality]:59 """Load one series. Exactly one origin is used, in this order: `path`, `root` (files), `database_url`.6061 If none is given, the environment decides: SPQ_DATA_DIR (files) first, then DATABASE_URL (PostgreSQL).62 start is inclusive and end exclusive, on bar-open time (UTC).63 """64 if path is None and root is None and database_url is None:65 root = os.environ.get("SPQ_DATA_DIR")66 if root is None:67 database_url = os.environ.get("DATABASE_URL")68 if root is None and database_url is None:69 raise DataError("no data origin: pass root=/path=/database_url= or set SPQ_DATA_DIR or DATABASE_URL")70 if path is not None or root is not None:71 src = source or os.environ.get("SPQ_CANDLE_SOURCE", "coinbase")72 raw = read_file(path if path is not None else file_path(root, src, symbol, interval))73 else:74 from .pg import read_pg7576 raw, src = read_pg(database_url, symbol, interval, start, end, source)77 return normalise(raw, symbol=symbol.upper(), interval=interval, source=src, start=start, end=end, as_of=as_of,78 on_duplicate=on_duplicate, on_gap=on_gap)798081def cross_source_deviation(root, symbol: str, interval: str, a: str = "coinbase", b: str = "kraken") -> dict | None:82 """Relative difference of the closes of two sources on their common bars (None if a file is missing)."""83 try:84 da, _ = load_candles(symbol, interval, root=root, source=a)85 db, _ = load_candles(symbol, interval, root=root, source=b)86 except DataError:87 return None88 common = da.index.intersection(db.index)89 if len(common) == 0:90 return None91 diff = (da.loc[common, "close"] / db.loc[common, "close"] - 1.0).abs()92 return {"bars": int(len(common)), "median_abs_rel_diff": float(diff.median()),93 "max_abs_rel_diff": float(diff.max())}949596def load_universe(symbols, interval: str, start=None, end=None, **kwargs):97 """Load several symbols. Returns (data aligned on common timestamps, {symbol: Quality})."""98 data, quality = {}, {}99 for s in symbols:100 data[s], quality[s] = load_candles(s, interval, start, end, **kwargs)101 return align(data), quality102