Repository Public
1"""Normalisation and quality checks shared by every data source.23Conventions (identical to the `candles` table of docs/v3/ARCHITECTURE.md):4- index `ts` = bar-OPEN time, timezone-aware UTC, strictly increasing (oldest first);5- columns open, high, low, close, volume as float64;6- only CLOSED bars: a bar is closed when ts + interval <= as_of.7"""8from __future__ import annotations910import re11from dataclasses import asdict, dataclass, field12from datetime import datetime, timezone1314import numpy as np15import pandas as pd1617COLUMNS = ["open", "high", "low", "close", "volume"]1819_UNIT_SECONDS = {"m": 60, "h": 3600, "d": 86400, "w": 604800}202122class DataError(ValueError):23 """Raised when a series cannot be trusted (duplicates, unsorted, malformed, forbidden gaps)."""242526def interval_seconds(interval: str) -> int:27 """'1m' -> 60, '1h' -> 3600, '1d' -> 86400."""28 m = re.fullmatch(r"(\d+)([mhdw])", interval)29 if not m or int(m.group(1)) <= 0:30 raise DataError(f"unsupported interval {interval!r} (expected e.g. 1m, 5m, 1h, 4h, 1d)")31 return int(m.group(1)) * _UNIT_SECONDS[m.group(2)]323334def bars_per_year(interval: str) -> float:35 """Annualisation factor for a 24/7 market: number of bars in a 365-day year."""36 return 365.0 * 86400.0 / interval_seconds(interval)373839def to_utc(ts) -> pd.Timestamp:40 """Parse a timestamp; naive values are taken as UTC."""41 t = pd.Timestamp(ts)42 return t.tz_localize("UTC") if t.tzinfo is None else t.tz_convert("UTC")434445@dataclass46class Quality:47 """Result of the checks on one (symbol, interval) series."""4849 symbol: str50 interval: str51 source: str52 n_bars: int = 053 start: str | None = None54 end: str | None = None55 expected_bars: int = 056 missing_bars: int = 057 max_gap_bars: int = 058 gaps: list = field(default_factory=list) # [[first_missing_ts, last_missing_ts, n_missing], ...] (first 50)59 duplicates_dropped: int = 060 unclosed_dropped: int = 061 off_grid: int = 062 ohlc_violations: int = 06364 @property65 def missing_fraction(self) -> float:66 return self.missing_bars / self.expected_bars if self.expected_bars else 0.06768 def ok(self, max_missing_fraction: float = 0.005, max_gap_bars: int = 24) -> bool:69 """Hard yes/no: enough bars, on the grid, sane OHLC, few and short gaps."""70 return (71 self.n_bars > 072 and self.off_grid == 073 and self.ohlc_violations == 074 and self.missing_fraction <= max_missing_fraction75 and self.max_gap_bars <= max_gap_bars76 )7778 def to_dict(self) -> dict:79 d = asdict(self)80 d["missing_fraction"] = self.missing_fraction81 return d828384def normalise(85 raw: pd.DataFrame,86 *,87 symbol: str,88 interval: str,89 source: str,90 start=None,91 end=None,92 as_of=None,93 on_duplicate: str = "raise",94 on_gap: str = "report",95) -> tuple[pd.DataFrame, Quality]:96 """Turn a raw frame (column or index `ts` + OHLCV) into the canonical form and check it.9798 start is inclusive, end is exclusive (both on bar-open time). `as_of` (default: now) is the instant99 used to drop bars that are not closed yet. Duplicated timestamps raise unless100 on_duplicate='drop_identical' (rows that are exact copies are merged; conflicting copies always raise).101 Gaps are reported in the Quality object; on_gap='raise' turns any gap into an error.102 Missing bars are never filled.103 """104 step = interval_seconds(interval)105 df = raw.copy()106 if "ts" in df.columns:107 df = df.set_index("ts")108 missing = [c for c in COLUMNS if c not in df.columns]109 if missing:110 raise DataError(f"{symbol} {interval}: missing columns {missing}")111 idx = df.index112 if not isinstance(idx, pd.DatetimeIndex):113 if pd.api.types.is_numeric_dtype(idx):114 vals = np.asarray(idx, dtype="float64")115 unit = "ms" if len(vals) and np.nanmax(np.abs(vals)) > 1e11 else "s"116 idx = pd.to_datetime(vals, unit=unit, utc=True)117 else:118 idx = pd.to_datetime(idx, utc=True)119 idx = idx.tz_localize("UTC") if idx.tz is None else idx.tz_convert("UTC")120 df = df[COLUMNS].astype("float64")121 df.index = idx.as_unit("ns").rename("ts")122123 q = Quality(symbol=symbol, interval=interval, source=source)124125 df = df.sort_index(kind="stable")126 dup = df.index.duplicated(keep=False)127 if dup.any():128 if on_duplicate != "drop_identical":129 first = df.index[dup][0]130 raise DataError(f"{symbol} {interval}: {int(dup.sum())} rows share a timestamp (first: {first})")131 deduped = df[~df.index.duplicated(keep="first")]132 merged = df.reset_index().drop_duplicates()133 if len(merged) != len(deduped):134 raise DataError(f"{symbol} {interval}: conflicting rows for the same timestamp")135 q.duplicates_dropped = len(df) - len(deduped)136 df = deduped137138 if start is not None:139 df = df[df.index >= to_utc(start)]140 if end is not None:141 df = df[df.index < to_utc(end)]142143 now = to_utc(as_of) if as_of is not None else pd.Timestamp(datetime.now(timezone.utc))144 closed = df.index + pd.Timedelta(seconds=step) <= now145 q.unclosed_dropped = int((~closed).sum())146 df = df[closed]147148 epoch = df.index.asi8 // 10**9149 q.off_grid = int(((epoch % step) != 0).sum() + ((df.index.asi8 % 10**9) != 0).sum())150151 v = df.to_numpy()152 o, h, l, c, vol = (v[:, i] for i in range(5))153 bad = ~np.isfinite(v).all(axis=1)154 bad |= (np.minimum.reduce([o, h, l, c]) <= 0) | (vol < 0)155 bad |= (h < np.maximum(o, c)) | (l > np.minimum(o, c))156 q.ohlc_violations = int(bad.sum())157158 q.n_bars = len(df)159 if len(df):160 q.start, q.end = df.index[0].isoformat(), df.index[-1].isoformat()161 q.expected_bars = int((epoch[-1] - epoch[0]) // step) + 1162 jumps = np.diff(epoch) // step - 1163 where = np.nonzero(jumps > 0)[0]164 q.missing_bars = int(jumps[where].sum())165 q.max_gap_bars = int(jumps.max()) if len(jumps) else 0166 for i in where[:50]:167 a = df.index[i] + pd.Timedelta(seconds=step)168 b = df.index[i + 1] - pd.Timedelta(seconds=step)169 q.gaps.append([a.isoformat(), b.isoformat(), int(jumps[i])])170 if q.missing_bars and on_gap == "raise":171 raise DataError(f"{symbol} {interval}: {q.missing_bars} missing bars (first gap {q.gaps[0]})")172 return df, q173174175def align(data: dict[str, pd.DataFrame]) -> dict[str, pd.DataFrame]:176 """Keep only the timestamps present for every symbol (no filling)."""177 if not data:178 return {}179 common = None180 for df in data.values():181 common = df.index if common is None else common.intersection(df.index)182 return {s: df.loc[common] for s, df in data.items()}183