SPB Git forge
spboucher.ai ↗

spb/hfmarketdata

Public

Open high-frequency market data platform — FirstRate full-history downloader, DuckDB/Parquet lake, open REST API and React docs platform (www.hfmarketdata.io)

127commits 1branches 0releases
24.7 MBsize
maindefault branch
11 days agolast push
JavaScript 53.7% Python 38.3% CSS 4.6% TypeScript 3.1%

futures: tables SQLite (roots/contracts/gaps), index des fichiers du lac et backfill idempotent + CLI

- models.py : futures_roots, futures_contracts, futures_contract_gaps (plan §1.1–1.3 + expiration_source, aliases, timeframes JSON détaillé)
- lake.py : découverte des fichiers futures_contracts/{tf}/{archive|update}, SQL de fusion dédupliquée sur datetime (update prioritaire)
- backfill.py : un agrégat DuckDB par répertoire (filename=true), un scan 1day pour volume 20 séances / OI / trous > 3 jours ouvrés, statut, upserts ; modes --roots et --since (contrats touchés ré-agrégés sur tous leurs fichiers)
- scripts/backfill_contracts.py : CLI (--roots, --since, --today, --json, -v)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Simon-Pierre Boucher committed 20 days ago (Sep 4, 2026) parent 3331a8d

4 changed files +556 −0

added hfmarketdata/api/futures/backfill.py +331 −0
@@ -0,0 +1,331 @@
1 +"""Idempotent backfill of `futures_roots`, `futures_contracts` and `futures_contract_gaps` from the lake.
2 +
3 +Strategy (must finish in minutes on ~90 k files):
4 +* one DuckDB aggregate per `{tf}/{bucket}` directory (`read_parquet(..., filename=true)`) for
5 + first/last datetime + row counts — intraday directories only need the `datetime` column;
6 +* one DuckDB scan of all `1day` files (archive ∪ update, dedup on datetime) grouped per contract for
7 + the last 20 sessions volume, last non-zero open interest and the list of session dates (gaps);
8 +* SQLite upserts in one transaction.
9 +
10 +`run_backfill(roots=None, since=None)` is the library entry point (PM2 cron / CLI).
11 +
12 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
13 +"""
14 +from __future__ import annotations
15 +
16 +import csv
17 +import json
18 +import logging
19 +import os
20 +import time
21 +from collections import defaultdict
22 +from dataclasses import dataclass, field
23 +from datetime import date, datetime, timedelta
24 +from pathlib import Path
25 +from typing import Any
26 +
27 +import numpy as np
28 +from core.config import settings
29 +from core.db import session
30 +from core.duck import con, invalidate
31 +from sqlalchemy import delete, select
32 +from sqlalchemy.dialects.sqlite import insert
33 +
34 +from . import calendar_us as cal
35 +from . import expiry
36 +from .lake import BUCKETS, TIMEFRAMES, contracts_root, symbol_from_file_stem
37 +from .models import FuturesContract, FuturesContractGap, FuturesRoot
38 +from .specs import SPEC_BY_ROOT, UNKNOWN_LAKE_ROOTS, derived_spec, spec_for
39 +from .symbols import ContractSymbol
40 +
41 +log = logging.getLogger("hfmarketdata.futures.backfill")
42 +
43 +GAP_BUSINESS_DAYS = 3 # a gap is > 3 business days without a daily bar
44 +ACTIVE_GRACE_DAYS = 7
45 +
46 +
47 +@dataclass
48 +class ContractAgg:
49 + symbol: ContractSymbol
50 + files: dict[str, dict[str, str]] = field(default_factory=dict) # tf -> bucket -> path
51 + stats: dict[str, dict[str, Any]] = field(default_factory=dict) # tf -> {first,last,rows}
52 +
53 +
54 +def _dir_stats(directory: Path, tf: str) -> dict[str, tuple[Any, Any, int]]:
55 + """filename -> (min datetime, max datetime, rows) for every parquet file in a directory."""
56 + if not directory.is_dir():
57 + return {}
58 + pattern = str(directory / "*.parquet")
59 + rows = con().execute(
60 + "SELECT filename, min(datetime), max(datetime), count(*) FROM read_parquet(?, filename=true, union_by_name=true) "
61 + "GROUP BY filename", [pattern]).fetchall()
62 + return {os.path.basename(r[0]): (r[1], r[2], int(r[3])) for r in rows}
63 +
64 +
65 +def _daily_metrics(agg: dict[str, ContractAgg]) -> dict[str, dict[str, Any]]:
66 + """Per contract: volume_avg_daily (last 20 sessions), open_interest_last, session dates (for gaps).
67 + Single DuckDB query over all 1day files with archive/update dedup."""
68 + paths: list[tuple[str, str, int]] = [] # (symbol, path, priority)
69 + for sym, a in agg.items():
70 + for prio, bucket in enumerate(BUCKETS):
71 + p = a.files.get("1day", {}).get(bucket)
72 + if p:
73 + paths.append((sym, p, prio))
74 + if not paths:
75 + return {}
76 + c = con()
77 + c.execute("CREATE OR REPLACE TEMP TABLE _bf_files(sym VARCHAR, path VARCHAR, prio INTEGER)")
78 + c.executemany("INSERT INTO _bf_files VALUES (?, ?, ?)", paths)
79 + sql = """
80 + WITH raw AS (
81 + SELECT f.sym, f.prio, b.datetime, b.volume, b.open_interest
82 + FROM read_parquet(?, filename=true, union_by_name=true) b
83 + JOIN _bf_files f ON f.path = b.filename
84 + ), dedup AS (
85 + SELECT sym, datetime, volume, open_interest FROM raw
86 + QUALIFY row_number() OVER (PARTITION BY sym, datetime ORDER BY prio) = 1
87 + ), ranked AS (
88 + SELECT *, row_number() OVER (PARTITION BY sym ORDER BY datetime DESC) AS rn FROM dedup
89 + )
90 + SELECT sym,
91 + avg(volume) FILTER (WHERE rn <= 20) AS vol20,
92 + arg_max(open_interest, datetime) FILTER (WHERE open_interest > 0) AS oi_last,
93 + list(CAST(datetime AS DATE) ORDER BY datetime) AS days,
94 + count(*) AS n
95 + FROM ranked GROUP BY sym
96 + """
97 + out: dict[str, dict[str, Any]] = {}
98 + for sym, vol20, oi_last, days, n in c.execute(sql, [[p for _, p, _ in paths]]).fetchall():
99 + out[sym] = {"vol20": float(vol20) if vol20 is not None else None,
100 + "oi_last": float(oi_last) if oi_last is not None else None,
101 + "days": days, "n": int(n)}
102 + c.execute("DROP TABLE IF EXISTS _bf_files")
103 + return out
104 +
105 +
106 +def _gaps(days: list[date], calendar: str) -> list[tuple[date, date, int]]:
107 + """(gap_start, gap_end, business days missing) for holes > GAP_BUSINESS_DAYS between consecutive sessions."""
108 + if len(days) < 2:
109 + return []
110 + arr = np.array(days, dtype="datetime64[D]")
111 + diffs = (arr[1:] - arr[:-1]).astype(int)
112 + out = []
113 + for i in np.nonzero(diffs > GAP_BUSINESS_DAYS + 1)[0]: # ≥ 5 calendar days is the first candidate
114 + a, b = days[int(i)], days[int(i) + 1]
115 + missing = cal.business_days_between(a, b, calendar)
116 + if missing > GAP_BUSINESS_DAYS:
117 + out.append((a + timedelta(days=1), b - timedelta(days=1), missing))
118 + return out
119 +
120 +
121 +def _status(last_data: date | None, expiration: date | None, year: int, month: int, today: date) -> str:
122 + if last_data is None:
123 + return "expired"
124 + month_passed = (year, month) < (today.year, today.month) or (expiration is not None and expiration < today)
125 + return "expired" if last_data < today - timedelta(days=ACTIVE_GRACE_DAYS) and month_passed else "active"
126 +
127 +
128 +def _read_meta_csv() -> dict[str, dict[str, str]]:
129 + p = settings.data_root / "meta" / "futures" / "futures.csv"
130 + if not p.is_file():
131 + return {}
132 + out = {}
133 + with open(p, newline="", encoding="utf-8", errors="replace") as f:
134 + for row in csv.DictReader(f):
135 + t = (row.get("Ticker") or "").strip().upper()
136 + if t:
137 + out[t] = {k.strip(): (v or "").strip() for k, v in row.items() if k}
138 + return out
139 +
140 +
141 +def _exchange_from_name(name: str | None) -> str | None:
142 + if name and name.rstrip().endswith(")") and "(" in name:
143 + return name.rstrip()[name.rfind("(") + 1:-1].strip() or None
144 + return None
145 +
146 +
147 +def _clean_name(name: str | None) -> str | None:
148 + if not name:
149 + return None
150 + n = name.replace("Â", "").replace("\xa0", " ").strip()
151 + if n.endswith(")") and "(" in n:
152 + n = n[:n.rfind("(")].strip()
153 + return n or None
154 +
155 +
156 +def run_backfill(roots: list[str] | None = None, since: date | None = None, today: date | None = None,
157 + verbose: bool = False) -> dict[str, Any]:
158 + """Scan the lake and upsert roots/contracts/gaps. Returns a summary dict.
159 +
160 + roots: restrict to these lake roots (e.g. ["ES", "CL"]).
161 + since: only (re)process contract files modified on/after this date (incremental cron mode).
162 + """
163 + t0 = time.time()
164 + today = today or date.today()
165 + roots_filter = {r.upper() for r in roots} if roots else None
166 + since_ts = datetime.combine(since, datetime.min.time()).timestamp() if since else None
167 + base = contracts_root()
168 +
169 + # 1. file discovery + per-directory stats -------------------------------------------------------
170 + # `since` mode: a contract touched by ANY modified file is re-aggregated over ALL its files, so that
171 + # `files`/`timeframes` never shrink after an incremental run.
172 + touched: set[str] | None = None
173 + if since_ts:
174 + touched = set()
175 + for tf in TIMEFRAMES:
176 + for bucket in BUCKETS:
177 + d = base / tf / bucket
178 + if not d.is_dir():
179 + continue
180 + with os.scandir(d) as it:
181 + for e in it:
182 + if e.name.endswith(".parquet") and e.stat().st_mtime >= since_ts:
183 + cs = symbol_from_file_stem(e.name[:-8])
184 + if cs is not None and (not roots_filter or cs.root in roots_filter):
185 + touched.add(cs.short)
186 + agg: dict[str, ContractAgg] = {}
187 + n_files = 0
188 + for tf in TIMEFRAMES:
189 + for bucket in BUCKETS:
190 + d = base / tf / bucket
191 + if not d.is_dir():
192 + continue
193 + wanted: dict[str, ContractSymbol] = {}
194 + with os.scandir(d) as it:
195 + for e in it:
196 + if not e.name.endswith(".parquet"):
197 + continue
198 + cs = symbol_from_file_stem(e.name[:-8])
199 + if cs is None or (roots_filter and cs.root not in roots_filter):
200 + continue
201 + if touched is not None and cs.short not in touched:
202 + continue
203 + wanted[e.name] = cs
204 + if not wanted:
205 + continue
206 + if roots_filter or touched is not None:
207 + # partial scan: aggregate only the selected files (list of paths)
208 + paths = [str(d / n) for n in wanted]
209 + rows = con().execute(
210 + "SELECT filename, min(datetime), max(datetime), count(*) FROM read_parquet(?, filename=true, union_by_name=true) "
211 + "GROUP BY filename", [paths]).fetchall()
212 + stats = {os.path.basename(r[0]): (r[1], r[2], int(r[3])) for r in rows}
213 + else:
214 + stats = _dir_stats(d, tf)
215 + for name, cs in wanted.items():
216 + st = stats.get(name)
217 + if not st or st[2] == 0:
218 + continue
219 + n_files += 1
220 + a = agg.setdefault(cs.short, ContractAgg(cs))
221 + a.files.setdefault(tf, {})[bucket] = str(d / name)
222 + cur = a.stats.get(tf)
223 + first, last = st[0], st[1]
224 + if cur is None:
225 + a.stats[tf] = {"first": first, "last": last, "rows": st[2]}
226 + else:
227 + cur["first"] = min(cur["first"], first)
228 + cur["last"] = max(cur["last"], last)
229 + cur["rows"] = max(cur["rows"], st[2]) # overlap → the larger file approximates the merged count
230 + if verbose:
231 + log.info("%s/%s: %d files", tf, bucket, len(wanted))
232 + if verbose:
233 + log.info("discovery done: %d contracts, %d files in %.1fs", len(agg), n_files, time.time() - t0)
234 +
235 + # 2. daily metrics (volume, OI, gaps) ----------------------------------------------------------------
236 + daily = _daily_metrics(agg)
237 + if verbose:
238 + log.info("daily metrics done in %.1fs", time.time() - t0)
239 +
240 + # 3. build rows ----------------------------------------------------------------------------------------
241 + meta = _read_meta_csv()
242 + now = datetime.utcnow().replace(microsecond=0)
243 + contract_rows: list[dict[str, Any]] = []
244 + gap_rows: list[dict[str, Any]] = []
245 + per_root: dict[str, dict[str, Any]] = defaultdict(lambda: {"first": None, "last": None, "n": 0})
246 + for sym, a in agg.items():
247 + cs = a.symbol
248 + spec = spec_for(cs.root)
249 + d1 = a.stats.get("1day")
250 + any_first = min(s["first"] for s in a.stats.values())
251 + any_last = max(s["last"] for s in a.stats.values())
252 + first_data = (d1["first"] if d1 else any_first)
253 + last_data = (d1["last"] if d1 else any_last)
254 + first_data = first_data.date() if isinstance(first_data, datetime) else first_data
255 + last_data = last_data.date() if isinstance(last_data, datetime) else last_data
256 + ltd = expiry.compute(spec.expiry_rule, cs.year, cs.month) if spec.expiry_rule != "data" else None
257 + if ltd is not None:
258 + expiration, source = ltd, "rule"
259 + else:
260 + expiration, source = last_data, "data"
261 + fnd = expiry.compute_first_notice(spec.first_notice_rule, cs.year, cs.month) if spec.settlement_type == "physical" else None
262 + dm = daily.get(sym, {})
263 + timeframes = {tf: {"first": str(s["first"]), "last": str(s["last"]), "rows": s["rows"]} for tf, s in a.stats.items()}
264 + files = {tf: [b.get("archive"), b.get("update")] for tf, b in a.files.items()}
265 + contract_rows.append({
266 + "symbol": sym, "root": cs.root, "month_code": cs.month_code, "contract_month": cs.month, "contract_year": cs.year,
267 + "expiration_date": expiration, "expiration_source": source, "last_trading_date": ltd,
268 + "first_notice_date": fnd, "settlement_type": spec.settlement_type, "contract_size": spec.contract_size,
269 + "tick_size": spec.tick_size, "tick_value": spec.tick_value, "currency": spec.currency, "exchange": spec.exchange,
270 + "first_data_date": first_data, "last_data_date": last_data,
271 + "status": _status(last_data, expiration, cs.year, cs.month, today),
272 + "volume_avg_daily": dm.get("vol20"), "open_interest_last": dm.get("oi_last"),
273 + "bars_1day": dm.get("n") if dm else None,
274 + "timeframes": json.dumps(timeframes), "files": json.dumps(files), "updated_at": now,
275 + })
276 + for gs, ge, missing in _gaps(dm.get("days") or [], spec.calendar):
277 + gap_rows.append({"symbol": sym, "timeframe": "1day", "gap_start": gs, "gap_end": ge, "bars_missing": missing})
278 + pr = per_root[cs.root]
279 + pr["first"] = first_data if pr["first"] is None else min(pr["first"], first_data)
280 + pr["last"] = last_data if pr["last"] is None else max(pr["last"], last_data)
281 + pr["n"] += 1
282 +
283 + # roots: every reference spec + every lake root (+ meta csv names for derived ones)
284 + root_rows: list[dict[str, Any]] = []
285 + all_roots = set(SPEC_BY_ROOT) | set(per_root) | set(meta) if not roots_filter else set(per_root) | (roots_filter & set(SPEC_BY_ROOT))
286 + for r in sorted(all_roots):
287 + spec = SPEC_BY_ROOT.get(r)
288 + m = meta.get(r, {})
289 + if spec is None:
290 + spec = derived_spec(r, _clean_name(m.get("Name")), _exchange_from_name(m.get("Name")))
291 + d = spec.as_dict()
292 + pr = per_root.get(r)
293 + root_rows.append({**{k: v for k, v in d.items() if k not in ("aliases",)}, "aliases": json.dumps(d["aliases"]),
294 + "first_data_date": pr["first"] if pr else None, "last_data_date": pr["last"] if pr else None,
295 + "contracts_count": pr["n"] if pr else 0, "updated_at": now})
296 +
297 + # 4. upsert -------------------------------------------------------------------------------------------
298 + with session() as s:
299 + if root_rows:
300 + stmt = insert(FuturesRoot).values(root_rows)
301 + upd = {c.name: getattr(stmt.excluded, c.name) for c in FuturesRoot.__table__.columns if c.name != "root"}
302 + if since_ts:
303 + # incremental run: only some contracts were scanned → don't clobber the root coverage/count
304 + for k in ("first_data_date", "last_data_date", "contracts_count"):
305 + upd.pop(k, None)
306 + s.execute(stmt.on_conflict_do_update(index_elements=["root"], set_=upd))
307 + for i in range(0, len(contract_rows), 500):
308 + chunk = contract_rows[i:i + 500]
309 + stmt = insert(FuturesContract).values(chunk)
310 + upd = {c.name: getattr(stmt.excluded, c.name) for c in FuturesContract.__table__.columns if c.name != "symbol"}
311 + s.execute(stmt.on_conflict_do_update(index_elements=["symbol"], set_=upd))
312 + syms = [r["symbol"] for r in contract_rows]
313 + for i in range(0, len(syms), 500):
314 + s.execute(delete(FuturesContractGap).where(FuturesContractGap.symbol.in_(syms[i:i + 500])))
315 + for i in range(0, len(gap_rows), 500):
316 + s.execute(insert(FuturesContractGap).values(gap_rows[i:i + 500]))
317 + invalidate("futures|")
318 + summary = {"contracts": len(contract_rows), "files": n_files, "roots": len(root_rows), "gaps": len(gap_rows),
319 + "active": sum(1 for r in contract_rows if r["status"] == "active"),
320 + "rule_based": sum(1 for r in contract_rows if r["expiration_source"] == "rule"),
321 + "seconds": round(time.time() - t0, 1), "today": str(today)}
322 + log.info("futures backfill: %s", summary)
323 + return summary
324 +
325 +
326 +def contracts_in_db() -> int:
327 + with session() as s:
328 + return len(s.execute(select(FuturesContract.symbol)).all())
329 +
330 +
331 +__all__ = ["UNKNOWN_LAKE_ROOTS", "contracts_in_db", "run_backfill"]
added hfmarketdata/api/futures/lake.py +76 −0
@@ -0,0 +1,76 @@
1 +"""Discovery of individual-contract Parquet files in the lake + DuckDB merge helpers.
2 +
3 +Layout: `parquet/futures_contracts/{tf}/{archive|update}/{ROOT}_{MonthCode}{YY}_{tf}.parquet`.
4 +`archive` ≤ 2025 and `update` ≥ 2025 overlap: bars are merged with a union deduplicated on `datetime`,
5 +`update` winning.
6 +
7 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
8 +"""
9 +from __future__ import annotations
10 +
11 +import os
12 +from pathlib import Path
13 +
14 +from core.config import settings
15 +from core.duck import cached
16 +
17 +from .symbols import ContractSymbol, symbol_from_file_stem
18 +
19 +TIMEFRAMES = ("1min", "5min", "30min", "1hour", "1day")
20 +BUCKETS = ("update", "archive") # priority order for dedup
21 +INTERVAL_ALIASES = {"1m": "1min", "1min": "1min", "5m": "5min", "5min": "5min", "30m": "30min", "30min": "30min",
22 + "1h": "1hour", "1hour": "1hour", "60m": "1hour", "1d": "1day", "1day": "1day", "d": "1day", "daily": "1day"}
23 +INTERVAL_OF_TF = {"1min": "1m", "5min": "5m", "30min": "30m", "1hour": "1h", "1day": "1d"}
24 +
25 +FileIndex = dict[str, dict[str, dict[str, str]]] # symbol -> tf -> bucket -> path
26 +
27 +
28 +def contracts_root() -> Path:
29 + return settings.parquet / "futures_contracts"
30 +
31 +
32 +def _scan() -> FileIndex:
33 + idx: FileIndex = {}
34 + base = contracts_root()
35 + for tf in TIMEFRAMES:
36 + for bucket in BUCKETS:
37 + d = base / tf / bucket
38 + if not d.is_dir():
39 + continue
40 + with os.scandir(d) as it:
41 + for e in it:
42 + if not e.name.endswith(".parquet"):
43 + continue
44 + cs = symbol_from_file_stem(e.name[:-8])
45 + if cs is None:
46 + continue
47 + idx.setdefault(cs.short, {}).setdefault(tf, {})[bucket] = e.path
48 + return idx
49 +
50 +
51 +def file_index(fresh: bool = False) -> FileIndex:
52 + """symbol -> tf -> bucket -> path, cached 5 minutes (≈90 k entries, one scandir per directory)."""
53 + if fresh:
54 + return _scan()
55 + return cached("futures|file_index", _scan)
56 +
57 +
58 +def files_for(symbol: ContractSymbol | str, tf: str) -> list[str]:
59 + """[update_path?, archive_path?] for a contract/timeframe (empty if none)."""
60 + key = symbol.short if isinstance(symbol, ContractSymbol) else symbol
61 + b = file_index().get(key, {}).get(tf, {})
62 + return [b[k] for k in BUCKETS if k in b]
63 +
64 +
65 +def merged_sql(paths: list[str], columns: str = "*") -> str:
66 + """SQL text selecting the merged bars of one contract (dedup on datetime, update wins).
67 + `paths` is ordered by priority (index 0 wins). Uses positional `?` placeholders → bind `paths`."""
68 + if len(paths) == 1:
69 + return f"SELECT {columns} FROM read_parquet(?)"
70 + parts = " UNION ALL BY NAME ".join(f"SELECT *, {i} AS _src FROM read_parquet(?)" for i in range(len(paths)))
71 + return (f"SELECT {columns} FROM (SELECT * EXCLUDE (_src) FROM ({parts}) "
72 + "QUALIFY row_number() OVER (PARTITION BY datetime ORDER BY _src) = 1)")
73 +
74 +
75 +def symbols_of_root(root: str) -> list[str]:
76 + return sorted(s for s in file_index() if s[:-3] == root)
added hfmarketdata/api/futures/models.py +94 −0
@@ -0,0 +1,94 @@
1 +"""SQLite tables for the futures module (UPGRADE-PLAN §1.1–1.3).
2 +
3 +Deviations from the plan (documented): `futures_roots.aliases` (JSON list of accepted CME codes),
4 +`futures_roots.contract_size_unit`, `first_notice_rule`, `calendar`, `rth_start/rth_end`;
5 +`futures_contracts.expiration_source` (rule|data), `bars_1day`; `futures_contracts.timeframes` holds a JSON
6 +object `{tf: {"first": …, "last": …, "rows": n}}` rather than a bare list (needed by /coverage).
7 +
8 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
9 +"""
10 +from __future__ import annotations
11 +
12 +from datetime import date, datetime
13 +
14 +from core.db import Base, create_all
15 +from sqlalchemy import Date, DateTime, Float, ForeignKey, Index, Integer, String, Text
16 +from sqlalchemy.orm import Mapped, mapped_column
17 +
18 +
19 +class FuturesRoot(Base):
20 + __tablename__ = "futures_roots"
21 +
22 + root: Mapped[str] = mapped_column(String(8), primary_key=True)
23 + name: Mapped[str | None] = mapped_column(Text)
24 + exchange: Mapped[str | None] = mapped_column(String(16))
25 + asset_class: Mapped[str | None] = mapped_column(String(16))
26 + currency: Mapped[str | None] = mapped_column(String(3))
27 + contract_size: Mapped[float | None] = mapped_column(Float)
28 + contract_size_unit: Mapped[str | None] = mapped_column(String(32))
29 + tick_size: Mapped[float | None] = mapped_column(Float)
30 + tick_value: Mapped[float | None] = mapped_column(Float)
31 + settlement_type: Mapped[str | None] = mapped_column(String(8))
32 + expiry_rule: Mapped[str] = mapped_column(String(32), default="data")
33 + first_notice_rule: Mapped[str | None] = mapped_column(String(32))
34 + calendar: Mapped[str] = mapped_column(String(8), default="us")
35 + month_cycle: Mapped[str | None] = mapped_column(String(12))
36 + rth_start: Mapped[str | None] = mapped_column(String(5))
37 + rth_end: Mapped[str | None] = mapped_column(String(5))
38 + aliases: Mapped[str | None] = mapped_column(Text) # JSON list
39 + first_data_date: Mapped[date | None] = mapped_column(Date)
40 + last_data_date: Mapped[date | None] = mapped_column(Date)
41 + contracts_count: Mapped[int] = mapped_column(Integer, default=0)
42 + source: Mapped[str] = mapped_column(String(10), default="derived")
43 + updated_at: Mapped[datetime | None] = mapped_column(DateTime)
44 +
45 +
46 +class FuturesContract(Base):
47 + __tablename__ = "futures_contracts"
48 +
49 + symbol: Mapped[str] = mapped_column(String(10), primary_key=True) # ESZ25
50 + root: Mapped[str] = mapped_column(String(8), ForeignKey("futures_roots.root"), nullable=False)
51 + month_code: Mapped[str] = mapped_column(String(1), nullable=False)
52 + contract_month: Mapped[int] = mapped_column(Integer, nullable=False)
53 + contract_year: Mapped[int] = mapped_column(Integer, nullable=False)
54 + expiration_date: Mapped[date | None] = mapped_column(Date)
55 + expiration_source: Mapped[str] = mapped_column(String(4), default="data") # rule | data
56 + last_trading_date: Mapped[date | None] = mapped_column(Date)
57 + first_notice_date: Mapped[date | None] = mapped_column(Date)
58 + settlement_type: Mapped[str | None] = mapped_column(String(8))
59 + contract_size: Mapped[float | None] = mapped_column(Float)
60 + tick_size: Mapped[float | None] = mapped_column(Float)
61 + tick_value: Mapped[float | None] = mapped_column(Float)
62 + currency: Mapped[str | None] = mapped_column(String(3))
63 + exchange: Mapped[str | None] = mapped_column(String(16))
64 + first_data_date: Mapped[date | None] = mapped_column(Date)
65 + last_data_date: Mapped[date | None] = mapped_column(Date)
66 + status: Mapped[str] = mapped_column(String(8), default="expired") # active | expired
67 + volume_avg_daily: Mapped[float | None] = mapped_column(Float)
68 + open_interest_last: Mapped[float | None] = mapped_column(Float)
69 + bars_1day: Mapped[int | None] = mapped_column(Integer)
70 + timeframes: Mapped[str | None] = mapped_column(Text) # JSON {tf: {first,last,rows}}
71 + files: Mapped[str | None] = mapped_column(Text) # JSON {tf: [archive_path|null, update_path|null]}
72 + updated_at: Mapped[datetime | None] = mapped_column(DateTime)
73 +
74 + __table_args__ = (
75 + Index("ix_fc_root_exp", "root", "expiration_date"),
76 + Index("ix_fc_status", "status"),
77 + Index("ix_fc_root_ym", "root", "contract_year", "contract_month"),
78 + )
79 +
80 +
81 +class FuturesContractGap(Base):
82 + __tablename__ = "futures_contract_gaps"
83 +
84 + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
85 + symbol: Mapped[str] = mapped_column(String(10), ForeignKey("futures_contracts.symbol", ondelete="CASCADE"), nullable=False)
86 + timeframe: Mapped[str] = mapped_column(String(6), nullable=False, default="1day")
87 + gap_start: Mapped[date] = mapped_column(Date, nullable=False)
88 + gap_end: Mapped[date] = mapped_column(Date, nullable=False)
89 + bars_missing: Mapped[int] = mapped_column(Integer, nullable=False)
90 +
91 + __table_args__ = (Index("ix_fcg_symbol", "symbol"),)
92 +
93 +
94 +create_all()
added scripts/backfill_contracts.py +55 −0
@@ -0,0 +1,55 @@
1 +#!/usr/bin/env python3
2 +"""Backfill the futures contracts metadata (SQLite) from the Parquet lake — idempotent.
3 +
4 +Usage (production node M3U96b):
5 + cd ~/hfmarketdata && HFMD_DATA_ROOT=~/firstratedata venv/bin/python scripts/backfill_contracts.py
6 + … --roots ES,CL,NG only these lake roots
7 + … --since 2026-09-01 only contract files modified on/after this date (cron mode)
8 + … --verbose progress logs
9 +
10 +Author: Simon-Pierre Boucher <contact@spboucher.ai>
11 +"""
12 +from __future__ import annotations
13 +
14 +import argparse
15 +import json
16 +import logging
17 +import sys
18 +from datetime import date
19 +from pathlib import Path
20 +
21 +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "hfmarketdata" / "api"))
22 +
23 +
24 +def main() -> int:
25 + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
26 + ap.add_argument("--roots", help="comma-separated lake roots, e.g. ES,CL (CME aliases like 6E accepted)")
27 + ap.add_argument("--since", help="YYYY-MM-DD: only files modified on/after this date")
28 + ap.add_argument("--today", help="YYYY-MM-DD: reference date for the active/expired status (tests)")
29 + ap.add_argument("--verbose", "-v", action="store_true")
30 + ap.add_argument("--json", action="store_true", help="print the summary as JSON")
31 + args = ap.parse_args()
32 + logging.basicConfig(level=logging.INFO if args.verbose else logging.WARNING, format="%(asctime)s %(levelname)s %(message)s")
33 +
34 + from core.config import settings
35 + from futures.backfill import run_backfill
36 + from futures.symbols import normalize_root
37 +
38 + roots = [normalize_root(r) for r in args.roots.split(",") if r.strip()] if args.roots else None
39 + since = date.fromisoformat(args.since) if args.since else None
40 + today = date.fromisoformat(args.today) if args.today else None
41 + if not (settings.parquet / "futures_contracts").is_dir():
42 + print(f"error: {settings.parquet / 'futures_contracts'} not found (set HFMD_DATA_ROOT)", file=sys.stderr)
43 + return 2
44 + summary = run_backfill(roots=roots, since=since, today=today, verbose=args.verbose)
45 + if args.json:
46 + print(json.dumps(summary))
47 + else:
48 + print(f"futures backfill done in {summary['seconds']}s — {summary['contracts']} contracts "
49 + f"({summary['active']} active, {summary['rule_based']} rule-based expirations), "
50 + f"{summary['files']} files, {summary['roots']} roots, {summary['gaps']} gaps → {settings.state_db}")
51 + return 0
52 +
53 +
54 +if __name__ == "__main__":
55 + raise SystemExit(main())
56