#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ================================================================================ FirstRate Data — Full Historical Downloader & Parquet Converter ================================================================================ Downloads the COMPLETE historical datasets from the FirstRate Data API and stores them as query-optimized Parquet files (zstd compressed), ready to be served by an ultra-fast query layer (DuckDB / pyarrow). Coverage -------- * options : full archives per year/quarter (2010 -> today) + 'month' update for the current partial quarter * stock, etf : period=full, ticker_range A-Z, timeframes 1min/5min/30min/1hour/1day, adjustments adj_split / adj_splitdiv / UNADJUSTED (UNADJUSTED only exists for 1min & 1day) * futures (continuous) : timeframes x contin_UNadj / contin_adj_ratio / contin_adj_absolute * futures (individual) : contract_files=archive + update, all timeframes * crypto, index, fx : period=full, all timeframes * metadata : ticker listings, splits, dividends, update logs, futures continuous-series audit file * delisted stocks : optional (--include-delisted), archives 1-4 + 2026 update, all timeframe/adjustment combos Storage layout (under --data-root) ---------------------------------- parquet/{type}/{timeframe}/{adjustment}/{TICKER}.parquet (bar data) parquet/options/{year}_{quarter}/{file}.parquet (options) meta/… (csv metadata) zips/… (only w/ --keep-zips) state/manifest.json (resume state) logs/frd_downloader.log Every Parquet file carries a `ticker` column, so the whole tree can be queried directly, e.g.: SELECT * FROM read_parquet('parquet/stock/1day/adj_splitdiv/*.parquet') WHERE ticker = 'AAPL' AND datetime >= '2020-01-01'; Usage ----- python3 frd_downloader.py --data-root /Volumes/ssd/firstratedata python3 frd_downloader.py --list-jobs # preview the job plan python3 frd_downloader.py --types options stock # restrict asset types python3 frd_downloader.py --include-delisted # add delisted stocks The run is fully resumable: completed jobs are recorded in the manifest and skipped on re-run. Author : Simon-Pierre Boucher Contact : contact@spboucher.ai Created : 2026-08-09 ================================================================================ """ from __future__ import annotations import argparse import json import logging import os import re import shutil import subprocess import sys import tempfile import threading import time import zipfile from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import date from pathlib import Path from typing import Iterator from urllib.parse import urlencode import duckdb import requests __author__ = "Simon-Pierre Boucher" __contact__ = "contact@spboucher.ai" __version__ = "1.0.0" # ------------------------------------------------------------------------------ # Configuration # ------------------------------------------------------------------------------ BASE_URL = "https://firstratedata.com/api" # FirstRate Data customer id — never hardcoded: export FRD_USERID= USERID = os.environ.get("FRD_USERID", "") TIMEFRAMES = ["1day", "1hour", "30min", "5min", "1min"] # small -> large STOCK_ETF_ADJUSTMENTS = ["adj_split", "adj_splitdiv", "UNADJUSTED"] UNADJUSTED_TIMEFRAMES = {"1min", "1day"} # per API doc FUTURES_ADJUSTMENTS = ["contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"] TICKER_RANGES = [chr(c) for c in range(ord("A"), ord("Z") + 1)] OPTIONS_FIRST_YEAR = 2010 DOWNLOAD_TIMEOUT = 7200 # seconds per request (archives can be huge) CHUNK_SIZE = 4 * 1024 * 1024 # 4 MiB streaming chunks MAX_RETRIES = 5 RETRY_BACKOFF = 60 # seconds, multiplied by attempt number MIN_FREE_GB = 30 # abort downloads below this free-space floor log = logging.getLogger("frd") # ------------------------------------------------------------------------------ # Job model # ------------------------------------------------------------------------------ @dataclass class Job: """One downloadable unit: an API request plus its destination.""" key: str # unique id, used in the manifest endpoint: str # API endpoint path (e.g. 'data_file') params: dict # query parameters (userid added at request time) dest_subdir: str # where converted output lands, under data root kind: str = "bars" # 'bars' (zip of csv -> parquet) or 'meta' (raw text) priority: int = 50 # lower runs first @property def url(self) -> str: q = dict(self.params) q["userid"] = USERID return f"{BASE_URL}/{self.endpoint}?{urlencode(q)}" @property def display_url(self) -> str: return f"{BASE_URL}/{self.endpoint}?{urlencode(self.params)}&userid=***" def _current_quarter(today: date) -> tuple[int, int]: return today.year, (today.month - 1) // 3 + 1 def build_jobs(types: list[str], include_delisted: bool) -> list[Job]: """Build the full download plan, ordered so that the small/most-useful datasets (daily bars, metadata) land first and the huge ones last.""" jobs: list[Job] = [] today = date.today() # -- metadata (tiny, always useful, run first) ------------------------------ for t in ["stock", "etf", "futures", "crypto", "index", "fx"]: if t in types: jobs.append(Job( key=f"ticker_listing|{t}", endpoint="ticker_listing", params={"type": t}, dest_subdir=f"meta/{t}", kind="meta", priority=0)) for t in ["stock", "etf"]: if t in types: for mft in ["splits", "dividends"]: jobs.append(Job( key=f"meta|{t}|{mft}", endpoint="meta_file", params={"type": t, "metafile_type": mft}, dest_subdir=f"meta/{t}", kind="meta", priority=1)) if "stock" in types: for lt in ["delisted", "added", "changed"]: jobs.append(Job( key=f"update_log|stock|{lt}", endpoint="update_log", params={"type": "stock", "log_type": lt}, dest_subdir="meta/stock", kind="meta", priority=1)) if "futures" in types: jobs.append(Job( key="meta|futures|contin_audit", endpoint="meta_file", params={"type": "futures", "metafile_type": "contin_audit"}, dest_subdir="meta/futures", kind="meta", priority=1)) # -- stock / etf : full archives, per letter -------------------------------- for t in ["stock", "etf"]: if t not in types: continue for tf_i, tf in enumerate(TIMEFRAMES): for adj in STOCK_ETF_ADJUSTMENTS: if adj == "UNADJUSTED" and tf not in UNADJUSTED_TIMEFRAMES: continue for letter in TICKER_RANGES: jobs.append(Job( key=f"{t}|full|{tf}|{adj}|{letter}", endpoint="data_file", params={"type": t, "period": "full", "ticker_range": letter, "timeframe": tf, "adjustment": adj}, dest_subdir=f"parquet/{t}/{tf}/{adj}", priority=10 + tf_i * 4)) # -- futures : continuous series -------------------------------------------- if "futures" in types: for tf_i, tf in enumerate(TIMEFRAMES): for adj in FUTURES_ADJUSTMENTS: jobs.append(Job( key=f"futures|full|{tf}|{adj}", endpoint="data_file", params={"type": "futures", "period": "full", "timeframe": tf, "adjustment": adj}, dest_subdir=f"parquet/futures/{tf}/{adj}", priority=10 + tf_i * 4)) # individual contracts (pre-2026 archive + current-year update) for tf_i, tf in enumerate(TIMEFRAMES): for cf in ["archive", "update"]: jobs.append(Job( key=f"futures_contract|{cf}|{tf}", endpoint="futures_contract", params={"contract_files": cf, "timeframe": tf}, dest_subdir=f"parquet/futures_contracts/{tf}/{cf}", priority=12 + tf_i * 4)) # -- crypto / index / fx : full archives ------------------------------------- for t in ["crypto", "index", "fx"]: if t not in types: continue for tf_i, tf in enumerate(TIMEFRAMES): jobs.append(Job( key=f"{t}|full|{tf}", endpoint="data_file", params={"type": t, "period": "full", "timeframe": tf}, dest_subdir=f"parquet/{t}/{tf}/none", priority=10 + tf_i * 4)) # -- options : every quarter since 2010 + current partial quarter ------------ # Priority 5-6: options are the primary dataset, they download right after # the metadata and before all the bar archives. if "options" in types: cur_year, cur_q = _current_quarter(today) for year in range(OPTIONS_FIRST_YEAR, cur_year + 1): for q in range(1, 5): if year == cur_year and q >= cur_q: continue # current/future quarters have no full archive yet jobs.append(Job( key=f"options|{year}|q{q}", endpoint="data_file", params={"type": "options", "year": str(year), "quarter": f"q{q}"}, dest_subdir=f"parquet/options/{year}_q{q}", priority=5)) jobs.append(Job( key="options|update|month", endpoint="data_file", params={"type": "options", "period": "month"}, dest_subdir=f"parquet/options/{cur_year}_q{cur_q}_partial", priority=6)) # -- delisted stocks (optional, survivorship-bias-free research) ------------- if include_delisted and "stock" in types: for tf_i, tf in enumerate(TIMEFRAMES): for adj in STOCK_ETF_ADJUSTMENTS: if adj == "UNADJUSTED" and tf != "1min": continue # per API doc: UNADJUSTED delisted = 1min only for arc in ["1", "2", "3", "4"]: jobs.append(Job( key=f"delisted|archive{arc}|{tf}|{adj}", endpoint="delisted_data_file", params={"archive_number": arc, "timeframe": tf, "adjustment": adj}, dest_subdir=f"parquet/stock_delisted/{tf}/{adj}", priority=30 + tf_i * 4)) jobs.append(Job( key=f"delisted|update_year|{tf}|{adj}", endpoint="delisted_data_file", params={"update": "year", "timeframe": tf, "adjustment": adj}, dest_subdir=f"parquet/stock_delisted/{tf}/{adj}", priority=30 + tf_i * 4)) jobs.sort(key=lambda j: (j.priority, j.key)) return jobs # ------------------------------------------------------------------------------ # Manifest (resume state) # ------------------------------------------------------------------------------ class Manifest: """Thread-safe JSON manifest tracking completed/failed jobs.""" def __init__(self, path: Path): self.path = path self._lock = threading.Lock() self._data: dict = {"jobs": {}, "version": __version__} if path.exists(): try: self._data = json.loads(path.read_text()) except (json.JSONDecodeError, OSError): log.warning("Manifest unreadable, starting fresh: %s", path) def is_done(self, key: str) -> bool: return self._data["jobs"].get(key, {}).get("status") == "done" def mark(self, key: str, status: str, **info) -> None: with self._lock: entry = self._data["jobs"].setdefault(key, {}) entry.update(status=status, updated=time.strftime("%Y-%m-%d %H:%M:%S"), **info) tmp = self.path.with_suffix(".tmp") tmp.write_text(json.dumps(self._data, indent=1)) tmp.replace(self.path) def summary(self) -> dict: counts: dict[str, int] = {} for e in self._data["jobs"].values(): counts[e.get("status", "?")] = counts.get(e.get("status", "?"), 0) + 1 return counts # ------------------------------------------------------------------------------ # CSV -> Parquet conversion # ------------------------------------------------------------------------------ # Column layouts by column count (FirstRate bar files have no header row). # The first field is read as VARCHAR because FirstRate mixes date formats # ('2005-01-03 17:00:00' intraday vs '20100104' daily) — it is normalized to a # proper TIMESTAMP at conversion time (see _DT_PARSE). BAR_SCHEMAS = { 5: {"dt_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE", "low": "DOUBLE", "close": "DOUBLE"}, 6: {"dt_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE", "low": "DOUBLE", "close": "DOUBLE", "volume": "DOUBLE"}, 7: {"dt_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE", "low": "DOUBLE", "close": "DOUBLE", "volume": "DOUBLE", "open_interest": "DOUBLE"}, } _DT_PARSE = ("COALESCE(" "try_strptime(dt_raw, '%Y-%m-%d %H:%M:%S'), " "try_strptime(dt_raw, '%Y%m%d'), " "try_strptime(dt_raw, '%Y-%m-%d'), " "try_strptime(dt_raw, '%m/%d/%Y %H:%M:%S'), " "try_strptime(dt_raw, '%m/%d/%Y'))") # Options chain layout, per https://firstratedata.com/_readme/options.txt : # Trade Date, Strike, Expiry Date, Call/Put, Last Trade Price, Bid Price, # Ask Price, Bid IV, Ask IV, Open Interest, Volume, Delta, Gamma, Vega, # Theta, Rho OPTIONS_SCHEMA = { "trade_date": "DATE", "strike": "DOUBLE", "expiry": "DATE", "call_put": "VARCHAR", "last_price": "DOUBLE", "bid": "DOUBLE", "ask": "DOUBLE", "bid_iv": "DOUBLE", "ask_iv": "DOUBLE", "open_interest": "DOUBLE", "volume": "DOUBLE", "delta": "DOUBLE", "gamma": "DOUBLE", "vega": "DOUBLE", "theta": "DOUBLE", "rho": "DOUBLE", } def _sniff_first_line(csv_path: Path) -> list[str]: with open(csv_path, "r", errors="replace") as f: for line in f: line = line.strip() if line: return line.split(",") return [] _TIME_RE = re.compile(r"^\d{2}:\d{2}(:\d{2})?$") # FX intraday layout: date and time come as two separate fields # ({yyyyMMdd},{HH:mm:ss},O,H,L,C,V per the fx readme). SPLIT_DT_SCHEMA = { "date_raw": "VARCHAR", "time_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE", "low": "DOUBLE", "close": "DOUBLE", "volume": "DOUBLE", } _SPLIT_DT_PARSE = ("COALESCE(" "try_strptime(date_raw || ' ' || time_raw, '%Y%m%d %H:%M:%S'), " "try_strptime(date_raw || ' ' || time_raw, '%Y-%m-%d %H:%M:%S'), " "try_strptime(date_raw || ' ' || time_raw, '%Y%m%d %H:%M'))") def _ticker_from_filename(path: Path) -> str: """'AAPL_1min.txt' -> 'AAPL' ; 'ES_contin_adj.txt' -> 'ES'.""" return path.stem.split("_")[0].upper() def csv_to_parquet(con: duckdb.DuckDBPyConnection, csv_path: Path, out_path: Path, asset_type: str) -> int: """Convert one extracted csv/txt file to zstd Parquet. Returns row count.""" first = _sniff_first_line(csv_path) ncols = len(first) if ncols == 0: return 0 split_dt = ncols == 7 and len(first) > 1 and _TIME_RE.match(first[1]) out_path.parent.mkdir(parents=True, exist_ok=True) ticker = _ticker_from_filename(csv_path) src = str(csv_path).replace("'", "''") dst = str(out_path).replace("'", "''") # strict_mode=false: FirstRate files mix LF and CRLF line endings (daily # rows are appended with CRLF onto LF history), which the strict CSV # sniffer rejects outright. common = "strict_mode=false, ignore_errors=true" if split_dt: cols = json.dumps(SPLIT_DT_SCHEMA).replace('"', "'") select = (f"SELECT '{ticker}' AS ticker, {_SPLIT_DT_PARSE} AS datetime, " f"open, high, low, close, volume FROM read_csv('{src}', " f"header=false, columns={cols}, {common})") elif asset_type == "options" and ncols == len(OPTIONS_SCHEMA): cols = json.dumps(OPTIONS_SCHEMA).replace('"', "'") select = (f"SELECT '{ticker}' AS ticker, * FROM read_csv('{src}', " f"header=false, columns={cols}, dateformat='%Y-%m-%d', " f"{common})") elif asset_type != "options" and (schema := BAR_SCHEMAS.get(ncols)): cols = json.dumps(schema).replace('"', "'") value_cols = ", ".join(c for c in schema if c != "dt_raw") select = (f"SELECT '{ticker}' AS ticker, {_DT_PARSE} AS datetime, " f"{value_cols} FROM read_csv('{src}', " f"header=false, columns={cols}, {common})") else: # Unknown layout (options files etc.): let DuckDB auto-detect types, # header presence and column names — still fully queryable. select = (f"SELECT '{ticker}' AS ticker, * FROM read_csv_auto('{src}', " f"{common})") con.execute(f"COPY ({select}) TO '{dst}' " f"(FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 1000000)") rows = con.execute(f"SELECT count(*) FROM read_parquet('{dst}')").fetchone()[0] if rows == 0 and csv_path.stat().st_size > 0: log.warning("0 rows converted from non-empty file %s (ncols=%d)", csv_path.name, ncols) return rows def extract_and_convert(zip_path: Path, dest_dir: Path, asset_type: str) -> tuple[int, int]: """Extract a downloaded zip (recursively if it nests zips) and convert every csv/txt member to Parquet under dest_dir. Returns (files, rows).""" files = rows = 0 con = duckdb.connect() con.execute("SET threads TO 4") try: with tempfile.TemporaryDirectory(dir=zip_path.parent) as tmp: tmp_dir = Path(tmp) _extract_recursive(zip_path, tmp_dir) members = sorted(p for p in tmp_dir.rglob("*") if p.is_file() and p.suffix.lower() in (".txt", ".csv")) for member in members: out = dest_dir / (member.stem + ".parquet") try: n = csv_to_parquet(con, member, out, asset_type) if n > 0: files += 1 rows += n else: out.unlink(missing_ok=True) except duckdb.Error as e: log.error("Convert failed %s: %s", member.name, e) member.unlink() # free space as we go finally: con.close() return files, rows def _extract_recursive(zip_path: Path, dest: Path, depth: int = 0) -> None: if depth > 3: return try: with zipfile.ZipFile(zip_path) as zf: zf.extractall(dest) except (NotImplementedError, RuntimeError, zipfile.BadZipFile): # Some FirstRate archives use Deflate64, which neither the stdlib # zipfile nor macOS bsdtar can decompress — 7-Zip handles it. for tool in (["7zz", "x", "-y", f"-o{dest}", str(zip_path)], ["tar", "-xf", str(zip_path), "-C", str(dest)]): if shutil.which(tool[0]): subprocess.run(tool, check=True, capture_output=True) break else: raise for nested in list(dest.rglob("*.zip")): sub = nested.with_suffix("") sub.mkdir(exist_ok=True) _extract_recursive(nested, sub, depth + 1) nested.unlink() # ------------------------------------------------------------------------------ # Downloader # ------------------------------------------------------------------------------ @dataclass class Runner: data_root: Path manifest: Manifest keep_zips: bool = False min_free_gb: int = MIN_FREE_GB session: requests.Session = field(default_factory=requests.Session) def _free_gb(self) -> float: return shutil.disk_usage(self.data_root).free / 1e9 def _download(self, job: Job, dest: Path) -> Path | None: """Stream the response to dest (via .part). Returns path or None.""" part = dest.with_suffix(dest.suffix + ".part") for attempt in range(1, MAX_RETRIES + 1): if self._free_gb() < self.min_free_gb: raise RuntimeError( f"Only {self._free_gb():.1f} GB free (< {self.min_free_gb} GB floor)") try: with self.session.get(job.url, stream=True, timeout=DOWNLOAD_TIMEOUT) as r: if r.status_code in (429, 500, 502, 503, 504): raise requests.HTTPError(f"HTTP {r.status_code}") if r.status_code == 404: log.warning("[%s] 404 — no data for this combination", job.key) return None r.raise_for_status() dest.parent.mkdir(parents=True, exist_ok=True) size = 0 with open(part, "wb") as f: for chunk in r.iter_content(CHUNK_SIZE): f.write(chunk) size += len(chunk) part.replace(dest) log.info("[%s] downloaded %.1f MB", job.key, size / 1e6) return dest except (requests.RequestException, OSError) as e: part.unlink(missing_ok=True) if attempt == MAX_RETRIES: raise wait = RETRY_BACKOFF * attempt log.warning("[%s] attempt %d/%d failed (%s), retrying in %ds", job.key, attempt, MAX_RETRIES, e, wait) time.sleep(wait) return None def run_job(self, job: Job) -> None: if self.manifest.is_done(job.key): log.info("[%s] already done, skipping", job.key) return log.info("[%s] GET %s", job.key, job.display_url) t0 = time.time() try: if job.kind == "meta": self._run_meta(job) else: self._run_bars(job) except Exception as e: log.error("[%s] FAILED: %s", job.key, e) self.manifest.mark(job.key, "failed", error=str(e)) return log.info("[%s] done in %.0fs", job.key, time.time() - t0) def _run_meta(self, job: Job) -> None: out_dir = self.data_root / job.dest_subdir out_dir.mkdir(parents=True, exist_ok=True) fname = "_".join(str(v) for v in job.params.values()) + ".csv" out = out_dir / fname r = self.session.get(job.url, timeout=600) r.raise_for_status() out.write_bytes(r.content) self.manifest.mark(job.key, "done", bytes=len(r.content), file=str(out)) def _run_bars(self, job: Job) -> None: zip_dir = self.data_root / "zips" zip_dir.mkdir(parents=True, exist_ok=True) zip_path = zip_dir / (job.key.replace("|", "_") + ".zip") got = self._download(job, zip_path) if got is None: self.manifest.mark(job.key, "empty") return if not zipfile.is_zipfile(zip_path): head = zip_path.read_bytes()[:200] zip_path.unlink() raise RuntimeError(f"Response is not a zip (starts with {head[:60]!r})") asset_type = job.key.split("|")[0] dest_dir = self.data_root / job.dest_subdir files, rows = extract_and_convert(zip_path, dest_dir, asset_type) size = zip_path.stat().st_size if not self.keep_zips: zip_path.unlink() self.manifest.mark(job.key, "done", bytes=size, files=files, rows=rows) log.info("[%s] converted %d files / %s rows -> %s", job.key, files, f"{rows:,}", dest_dir) # ------------------------------------------------------------------------------ # CLI # ------------------------------------------------------------------------------ def setup_logging(data_root: Path) -> None: log_dir = data_root / "logs" log_dir.mkdir(parents=True, exist_ok=True) fmt = logging.Formatter("%(asctime)s %(levelname)-7s %(message)s", "%Y-%m-%d %H:%M:%S") for handler in (logging.StreamHandler(sys.stdout), logging.FileHandler(log_dir / "frd_downloader.log")): handler.setFormatter(fmt) log.addHandler(handler) log.setLevel(logging.INFO) def main() -> int: ap = argparse.ArgumentParser( description="FirstRate Data full-history downloader " f"(v{__version__} — {__author__} <{__contact__}>)") ap.add_argument("--data-root", type=Path, default=Path("/Volumes/ssd/firstratedata"), help="Root directory for all data (default: %(default)s)") ap.add_argument("--types", nargs="+", default=["options", "stock", "etf", "futures", "crypto", "index", "fx"], choices=["options", "stock", "etf", "futures", "crypto", "index", "fx"], help="Asset types to download (default: all)") ap.add_argument("--workers", type=int, default=2, help="Parallel downloads (default: %(default)s — be polite)") ap.add_argument("--keep-zips", action="store_true", help="Keep raw zip archives after conversion") ap.add_argument("--include-delisted", action="store_true", help="Also download the delisted-stocks archives") ap.add_argument("--min-free-gb", type=int, default=MIN_FREE_GB, help="Stop downloading below this free space (default: %(default)s)") ap.add_argument("--list-jobs", action="store_true", help="Print the job plan and exit") args = ap.parse_args() if not USERID: ap.error("FRD_USERID environment variable is not set " "(your FirstRate Data customer id)") jobs = build_jobs(args.types, args.include_delisted) if args.list_jobs: for j in jobs: print(f"{j.priority:3d} {j.key:45s} {j.display_url}") print(f"\nTotal: {len(jobs)} jobs") return 0 args.data_root.mkdir(parents=True, exist_ok=True) setup_logging(args.data_root) log.info("FirstRate Data downloader v%s — %s <%s>", __version__, __author__, __contact__) log.info("Data root: %s | types: %s | %d jobs | %d workers", args.data_root, ",".join(args.types), len(jobs), args.workers) manifest = Manifest(args.data_root / "state" / "manifest.json") (args.data_root / "state").mkdir(parents=True, exist_ok=True) runner = Runner(args.data_root, manifest, keep_zips=args.keep_zips, min_free_gb=args.min_free_gb) pending = [j for j in jobs if not manifest.is_done(j.key)] log.info("%d jobs pending (%d already complete)", len(pending), len(jobs) - len(pending)) with ThreadPoolExecutor(max_workers=args.workers) as pool: futures = {pool.submit(runner.run_job, j): j for j in pending} for fut in as_completed(futures): fut.result() # exceptions are handled inside run_job counts = manifest.summary() log.info("Run complete. Manifest: %s", counts) failed = counts.get("failed", 0) return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())