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)
Python 46.7%
JavaScript 37.4%
CSS 14.9%
HTML 0.9%
1#!/usr/bin/env python32# -*- coding: utf-8 -*-3"""4================================================================================5 FirstRate Data — Full Historical Downloader & Parquet Converter6================================================================================78Downloads the COMPLETE historical datasets from the FirstRate Data API and9stores them as query-optimized Parquet files (zstd compressed), ready to be10served by an ultra-fast query layer (DuckDB / pyarrow).1112Coverage13--------14 * options : full archives per year/quarter (2010 -> today)15 + 'month' update for the current partial quarter16 * stock, etf : period=full, ticker_range A-Z,17 timeframes 1min/5min/30min/1hour/1day,18 adjustments adj_split / adj_splitdiv / UNADJUSTED19 (UNADJUSTED only exists for 1min & 1day)20 * futures (continuous) : timeframes x contin_UNadj / contin_adj_ratio /21 contin_adj_absolute22 * futures (individual) : contract_files=archive + update, all timeframes23 * crypto, index, fx : period=full, all timeframes24 * metadata : ticker listings, splits, dividends, update logs,25 futures continuous-series audit file26 * delisted stocks : optional (--include-delisted), archives 1-4 +27 2026 update, all timeframe/adjustment combos2829Storage layout (under --data-root)30----------------------------------31 parquet/{type}/{timeframe}/{adjustment}/{TICKER}.parquet (bar data)32 parquet/options/{year}_{quarter}/{file}.parquet (options)33 meta/… (csv metadata)34 zips/… (only w/ --keep-zips)35 state/manifest.json (resume state)36 logs/frd_downloader.log3738Every Parquet file carries a `ticker` column, so the whole tree can be39queried directly, e.g.:4041 SELECT * FROM read_parquet('parquet/stock/1day/adj_splitdiv/*.parquet')42 WHERE ticker = 'AAPL' AND datetime >= '2020-01-01';4344Usage45-----46 python3 frd_downloader.py --data-root /Volumes/ssd/firstratedata47 python3 frd_downloader.py --list-jobs # preview the job plan48 python3 frd_downloader.py --types options stock # restrict asset types49 python3 frd_downloader.py --include-delisted # add delisted stocks5051The run is fully resumable: completed jobs are recorded in the manifest and52skipped on re-run.5354Author : Simon-Pierre Boucher55Contact : contact@spboucher.ai56Created : 2026-08-0957================================================================================58"""5960from __future__ import annotations6162import argparse63import json64import logging65import os66import re67import shutil68import subprocess69import sys70import tempfile71import threading72import time73import zipfile74from concurrent.futures import ThreadPoolExecutor, as_completed75from dataclasses import dataclass, field76from datetime import date77from pathlib import Path78from typing import Iterator79from urllib.parse import urlencode8081import duckdb82import requests8384__author__ = "Simon-Pierre Boucher"85__contact__ = "contact@spboucher.ai"86__version__ = "1.0.0"8788# ------------------------------------------------------------------------------89# Configuration90# ------------------------------------------------------------------------------9192BASE_URL = "https://firstratedata.com/api"93# FirstRate Data customer id — never hardcoded: export FRD_USERID=<your id>94USERID = os.environ.get("FRD_USERID", "")9596TIMEFRAMES = ["1day", "1hour", "30min", "5min", "1min"] # small -> large97STOCK_ETF_ADJUSTMENTS = ["adj_split", "adj_splitdiv", "UNADJUSTED"]98UNADJUSTED_TIMEFRAMES = {"1min", "1day"} # per API doc99FUTURES_ADJUSTMENTS = ["contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"]100TICKER_RANGES = [chr(c) for c in range(ord("A"), ord("Z") + 1)]101OPTIONS_FIRST_YEAR = 2010102103DOWNLOAD_TIMEOUT = 7200 # seconds per request (archives can be huge)104CHUNK_SIZE = 4 * 1024 * 1024 # 4 MiB streaming chunks105MAX_RETRIES = 5106RETRY_BACKOFF = 60 # seconds, multiplied by attempt number107MIN_FREE_GB = 30 # abort downloads below this free-space floor108109log = logging.getLogger("frd")110111# ------------------------------------------------------------------------------112# Job model113# ------------------------------------------------------------------------------114115116@dataclass117class Job:118 """One downloadable unit: an API request plus its destination."""119120 key: str # unique id, used in the manifest121 endpoint: str # API endpoint path (e.g. 'data_file')122 params: dict # query parameters (userid added at request time)123 dest_subdir: str # where converted output lands, under data root124 kind: str = "bars" # 'bars' (zip of csv -> parquet) or 'meta' (raw text)125 priority: int = 50 # lower runs first126127 @property128 def url(self) -> str:129 q = dict(self.params)130 q["userid"] = USERID131 return f"{BASE_URL}/{self.endpoint}?{urlencode(q)}"132133 @property134 def display_url(self) -> str:135 return f"{BASE_URL}/{self.endpoint}?{urlencode(self.params)}&userid=***"136137138def _current_quarter(today: date) -> tuple[int, int]:139 return today.year, (today.month - 1) // 3 + 1140141142def build_jobs(types: list[str], include_delisted: bool) -> list[Job]:143 """Build the full download plan, ordered so that the small/most-useful144 datasets (daily bars, metadata) land first and the huge ones last."""145 jobs: list[Job] = []146 today = date.today()147148 # -- metadata (tiny, always useful, run first) ------------------------------149 for t in ["stock", "etf", "futures", "crypto", "index", "fx"]:150 if t in types:151 jobs.append(Job(152 key=f"ticker_listing|{t}", endpoint="ticker_listing",153 params={"type": t}, dest_subdir=f"meta/{t}",154 kind="meta", priority=0))155 for t in ["stock", "etf"]:156 if t in types:157 for mft in ["splits", "dividends"]:158 jobs.append(Job(159 key=f"meta|{t}|{mft}", endpoint="meta_file",160 params={"type": t, "metafile_type": mft},161 dest_subdir=f"meta/{t}", kind="meta", priority=1))162 if "stock" in types:163 for lt in ["delisted", "added", "changed"]:164 jobs.append(Job(165 key=f"update_log|stock|{lt}", endpoint="update_log",166 params={"type": "stock", "log_type": lt},167 dest_subdir="meta/stock", kind="meta", priority=1))168 if "futures" in types:169 jobs.append(Job(170 key="meta|futures|contin_audit", endpoint="meta_file",171 params={"type": "futures", "metafile_type": "contin_audit"},172 dest_subdir="meta/futures", kind="meta", priority=1))173174 # -- stock / etf : full archives, per letter --------------------------------175 for t in ["stock", "etf"]:176 if t not in types:177 continue178 for tf_i, tf in enumerate(TIMEFRAMES):179 for adj in STOCK_ETF_ADJUSTMENTS:180 if adj == "UNADJUSTED" and tf not in UNADJUSTED_TIMEFRAMES:181 continue182 for letter in TICKER_RANGES:183 jobs.append(Job(184 key=f"{t}|full|{tf}|{adj}|{letter}",185 endpoint="data_file",186 params={"type": t, "period": "full",187 "ticker_range": letter,188 "timeframe": tf, "adjustment": adj},189 dest_subdir=f"parquet/{t}/{tf}/{adj}",190 priority=10 + tf_i * 4))191192 # -- futures : continuous series --------------------------------------------193 if "futures" in types:194 for tf_i, tf in enumerate(TIMEFRAMES):195 for adj in FUTURES_ADJUSTMENTS:196 jobs.append(Job(197 key=f"futures|full|{tf}|{adj}",198 endpoint="data_file",199 params={"type": "futures", "period": "full",200 "timeframe": tf, "adjustment": adj},201 dest_subdir=f"parquet/futures/{tf}/{adj}",202 priority=10 + tf_i * 4))203 # individual contracts (pre-2026 archive + current-year update)204 for tf_i, tf in enumerate(TIMEFRAMES):205 for cf in ["archive", "update"]:206 jobs.append(Job(207 key=f"futures_contract|{cf}|{tf}",208 endpoint="futures_contract",209 params={"contract_files": cf, "timeframe": tf},210 dest_subdir=f"parquet/futures_contracts/{tf}/{cf}",211 priority=12 + tf_i * 4))212213 # -- crypto / index / fx : full archives -------------------------------------214 for t in ["crypto", "index", "fx"]:215 if t not in types:216 continue217 for tf_i, tf in enumerate(TIMEFRAMES):218 jobs.append(Job(219 key=f"{t}|full|{tf}",220 endpoint="data_file",221 params={"type": t, "period": "full", "timeframe": tf},222 dest_subdir=f"parquet/{t}/{tf}/none",223 priority=10 + tf_i * 4))224225 # -- options : every quarter since 2010 + current partial quarter ------------226 # Priority 5-6: options are the primary dataset, they download right after227 # the metadata and before all the bar archives.228 if "options" in types:229 cur_year, cur_q = _current_quarter(today)230 for year in range(OPTIONS_FIRST_YEAR, cur_year + 1):231 for q in range(1, 5):232 if year == cur_year and q >= cur_q:233 continue # current/future quarters have no full archive yet234 jobs.append(Job(235 key=f"options|{year}|q{q}",236 endpoint="data_file",237 params={"type": "options", "year": str(year),238 "quarter": f"q{q}"},239 dest_subdir=f"parquet/options/{year}_q{q}",240 priority=5))241 jobs.append(Job(242 key="options|update|month",243 endpoint="data_file",244 params={"type": "options", "period": "month"},245 dest_subdir=f"parquet/options/{cur_year}_q{cur_q}_partial",246 priority=6))247248 # -- delisted stocks (optional, survivorship-bias-free research) -------------249 if include_delisted and "stock" in types:250 for tf_i, tf in enumerate(TIMEFRAMES):251 for adj in STOCK_ETF_ADJUSTMENTS:252 if adj == "UNADJUSTED" and tf != "1min":253 continue # per API doc: UNADJUSTED delisted = 1min only254 for arc in ["1", "2", "3", "4"]:255 jobs.append(Job(256 key=f"delisted|archive{arc}|{tf}|{adj}",257 endpoint="delisted_data_file",258 params={"archive_number": arc,259 "timeframe": tf, "adjustment": adj},260 dest_subdir=f"parquet/stock_delisted/{tf}/{adj}",261 priority=30 + tf_i * 4))262 jobs.append(Job(263 key=f"delisted|update_year|{tf}|{adj}",264 endpoint="delisted_data_file",265 params={"update": "year", "timeframe": tf,266 "adjustment": adj},267 dest_subdir=f"parquet/stock_delisted/{tf}/{adj}",268 priority=30 + tf_i * 4))269270 jobs.sort(key=lambda j: (j.priority, j.key))271 return jobs272273274# ------------------------------------------------------------------------------275# Manifest (resume state)276# ------------------------------------------------------------------------------277278279class Manifest:280 """Thread-safe JSON manifest tracking completed/failed jobs."""281282 def __init__(self, path: Path):283 self.path = path284 self._lock = threading.Lock()285 self._data: dict = {"jobs": {}, "version": __version__}286 if path.exists():287 try:288 self._data = json.loads(path.read_text())289 except (json.JSONDecodeError, OSError):290 log.warning("Manifest unreadable, starting fresh: %s", path)291292 def is_done(self, key: str) -> bool:293 return self._data["jobs"].get(key, {}).get("status") == "done"294295 def mark(self, key: str, status: str, **info) -> None:296 with self._lock:297 entry = self._data["jobs"].setdefault(key, {})298 entry.update(status=status, updated=time.strftime("%Y-%m-%d %H:%M:%S"), **info)299 tmp = self.path.with_suffix(".tmp")300 tmp.write_text(json.dumps(self._data, indent=1))301 tmp.replace(self.path)302303 def summary(self) -> dict:304 counts: dict[str, int] = {}305 for e in self._data["jobs"].values():306 counts[e.get("status", "?")] = counts.get(e.get("status", "?"), 0) + 1307 return counts308309310# ------------------------------------------------------------------------------311# CSV -> Parquet conversion312# ------------------------------------------------------------------------------313314# Column layouts by column count (FirstRate bar files have no header row).315# The first field is read as VARCHAR because FirstRate mixes date formats316# ('2005-01-03 17:00:00' intraday vs '20100104' daily) — it is normalized to a317# proper TIMESTAMP at conversion time (see _DT_PARSE).318BAR_SCHEMAS = {319 5: {"dt_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE",320 "low": "DOUBLE", "close": "DOUBLE"},321 6: {"dt_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE",322 "low": "DOUBLE", "close": "DOUBLE", "volume": "DOUBLE"},323 7: {"dt_raw": "VARCHAR", "open": "DOUBLE", "high": "DOUBLE",324 "low": "DOUBLE", "close": "DOUBLE", "volume": "DOUBLE",325 "open_interest": "DOUBLE"},326}327328_DT_PARSE = ("COALESCE("329 "try_strptime(dt_raw, '%Y-%m-%d %H:%M:%S'), "330 "try_strptime(dt_raw, '%Y%m%d'), "331 "try_strptime(dt_raw, '%Y-%m-%d'), "332 "try_strptime(dt_raw, '%m/%d/%Y %H:%M:%S'), "333 "try_strptime(dt_raw, '%m/%d/%Y'))")334335# Options chain layout, per https://firstratedata.com/_readme/options.txt :336# Trade Date, Strike, Expiry Date, Call/Put, Last Trade Price, Bid Price,337# Ask Price, Bid IV, Ask IV, Open Interest, Volume, Delta, Gamma, Vega,338# Theta, Rho339OPTIONS_SCHEMA = {340 "trade_date": "DATE", "strike": "DOUBLE", "expiry": "DATE",341 "call_put": "VARCHAR", "last_price": "DOUBLE", "bid": "DOUBLE",342 "ask": "DOUBLE", "bid_iv": "DOUBLE", "ask_iv": "DOUBLE",343 "open_interest": "DOUBLE", "volume": "DOUBLE", "delta": "DOUBLE",344 "gamma": "DOUBLE", "vega": "DOUBLE", "theta": "DOUBLE", "rho": "DOUBLE",345}346347348def _sniff_first_line(csv_path: Path) -> list[str]:349 with open(csv_path, "r", errors="replace") as f:350 for line in f:351 line = line.strip()352 if line:353 return line.split(",")354 return []355356357_TIME_RE = re.compile(r"^\d{2}:\d{2}(:\d{2})?$")358359# FX intraday layout: date and time come as two separate fields360# ({yyyyMMdd},{HH:mm:ss},O,H,L,C,V per the fx readme).361SPLIT_DT_SCHEMA = {362 "date_raw": "VARCHAR", "time_raw": "VARCHAR", "open": "DOUBLE",363 "high": "DOUBLE", "low": "DOUBLE", "close": "DOUBLE", "volume": "DOUBLE",364}365_SPLIT_DT_PARSE = ("COALESCE("366 "try_strptime(date_raw || ' ' || time_raw, '%Y%m%d %H:%M:%S'), "367 "try_strptime(date_raw || ' ' || time_raw, '%Y-%m-%d %H:%M:%S'), "368 "try_strptime(date_raw || ' ' || time_raw, '%Y%m%d %H:%M'))")369370371def _ticker_from_filename(path: Path) -> str:372 """'AAPL_1min.txt' -> 'AAPL' ; 'ES_contin_adj.txt' -> 'ES'."""373 return path.stem.split("_")[0].upper()374375376def csv_to_parquet(con: duckdb.DuckDBPyConnection, csv_path: Path,377 out_path: Path, asset_type: str) -> int:378 """Convert one extracted csv/txt file to zstd Parquet. Returns row count."""379 first = _sniff_first_line(csv_path)380 ncols = len(first)381 if ncols == 0:382 return 0383 split_dt = ncols == 7 and len(first) > 1 and _TIME_RE.match(first[1])384 out_path.parent.mkdir(parents=True, exist_ok=True)385 ticker = _ticker_from_filename(csv_path)386 src = str(csv_path).replace("'", "''")387 dst = str(out_path).replace("'", "''")388389 # strict_mode=false: FirstRate files mix LF and CRLF line endings (daily390 # rows are appended with CRLF onto LF history), which the strict CSV391 # sniffer rejects outright.392 common = "strict_mode=false, ignore_errors=true"393 if split_dt:394 cols = json.dumps(SPLIT_DT_SCHEMA).replace('"', "'")395 select = (f"SELECT '{ticker}' AS ticker, {_SPLIT_DT_PARSE} AS datetime, "396 f"open, high, low, close, volume FROM read_csv('{src}', "397 f"header=false, columns={cols}, {common})")398 elif asset_type == "options" and ncols == len(OPTIONS_SCHEMA):399 cols = json.dumps(OPTIONS_SCHEMA).replace('"', "'")400 select = (f"SELECT '{ticker}' AS ticker, * FROM read_csv('{src}', "401 f"header=false, columns={cols}, dateformat='%Y-%m-%d', "402 f"{common})")403 elif asset_type != "options" and (schema := BAR_SCHEMAS.get(ncols)):404 cols = json.dumps(schema).replace('"', "'")405 value_cols = ", ".join(c for c in schema if c != "dt_raw")406 select = (f"SELECT '{ticker}' AS ticker, {_DT_PARSE} AS datetime, "407 f"{value_cols} FROM read_csv('{src}', "408 f"header=false, columns={cols}, {common})")409 else:410 # Unknown layout (options files etc.): let DuckDB auto-detect types,411 # header presence and column names — still fully queryable.412 select = (f"SELECT '{ticker}' AS ticker, * FROM read_csv_auto('{src}', "413 f"{common})")414415 con.execute(f"COPY ({select}) TO '{dst}' "416 f"(FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 1000000)")417 rows = con.execute(f"SELECT count(*) FROM read_parquet('{dst}')").fetchone()[0]418 if rows == 0 and csv_path.stat().st_size > 0:419 log.warning("0 rows converted from non-empty file %s (ncols=%d)",420 csv_path.name, ncols)421 return rows422423424def extract_and_convert(zip_path: Path, dest_dir: Path, asset_type: str) -> tuple[int, int]:425 """Extract a downloaded zip (recursively if it nests zips) and convert every426 csv/txt member to Parquet under dest_dir. Returns (files, rows)."""427 files = rows = 0428 con = duckdb.connect()429 con.execute("SET threads TO 4")430 try:431 with tempfile.TemporaryDirectory(dir=zip_path.parent) as tmp:432 tmp_dir = Path(tmp)433 _extract_recursive(zip_path, tmp_dir)434 members = sorted(p for p in tmp_dir.rglob("*")435 if p.is_file() and p.suffix.lower() in (".txt", ".csv"))436 for member in members:437 out = dest_dir / (member.stem + ".parquet")438 try:439 n = csv_to_parquet(con, member, out, asset_type)440 if n > 0:441 files += 1442 rows += n443 else:444 out.unlink(missing_ok=True)445 except duckdb.Error as e:446 log.error("Convert failed %s: %s", member.name, e)447 member.unlink() # free space as we go448 finally:449 con.close()450 return files, rows451452453def _extract_recursive(zip_path: Path, dest: Path, depth: int = 0) -> None:454 if depth > 3:455 return456 try:457 with zipfile.ZipFile(zip_path) as zf:458 zf.extractall(dest)459 except (NotImplementedError, RuntimeError, zipfile.BadZipFile):460 # Some FirstRate archives use Deflate64, which neither the stdlib461 # zipfile nor macOS bsdtar can decompress — 7-Zip handles it.462 for tool in (["7zz", "x", "-y", f"-o{dest}", str(zip_path)],463 ["tar", "-xf", str(zip_path), "-C", str(dest)]):464 if shutil.which(tool[0]):465 subprocess.run(tool, check=True, capture_output=True)466 break467 else:468 raise469 for nested in list(dest.rglob("*.zip")):470 sub = nested.with_suffix("")471 sub.mkdir(exist_ok=True)472 _extract_recursive(nested, sub, depth + 1)473 nested.unlink()474475476# ------------------------------------------------------------------------------477# Downloader478# ------------------------------------------------------------------------------479480481@dataclass482class Runner:483 data_root: Path484 manifest: Manifest485 keep_zips: bool = False486 min_free_gb: int = MIN_FREE_GB487 session: requests.Session = field(default_factory=requests.Session)488489 def _free_gb(self) -> float:490 return shutil.disk_usage(self.data_root).free / 1e9491492 def _download(self, job: Job, dest: Path) -> Path | None:493 """Stream the response to dest (via .part). Returns path or None."""494 part = dest.with_suffix(dest.suffix + ".part")495 for attempt in range(1, MAX_RETRIES + 1):496 if self._free_gb() < self.min_free_gb:497 raise RuntimeError(498 f"Only {self._free_gb():.1f} GB free (< {self.min_free_gb} GB floor)")499 try:500 with self.session.get(job.url, stream=True,501 timeout=DOWNLOAD_TIMEOUT) as r:502 if r.status_code in (429, 500, 502, 503, 504):503 raise requests.HTTPError(f"HTTP {r.status_code}")504 if r.status_code == 404:505 log.warning("[%s] 404 — no data for this combination", job.key)506 return None507 r.raise_for_status()508 dest.parent.mkdir(parents=True, exist_ok=True)509 size = 0510 with open(part, "wb") as f:511 for chunk in r.iter_content(CHUNK_SIZE):512 f.write(chunk)513 size += len(chunk)514 part.replace(dest)515 log.info("[%s] downloaded %.1f MB", job.key, size / 1e6)516 return dest517 except (requests.RequestException, OSError) as e:518 part.unlink(missing_ok=True)519 if attempt == MAX_RETRIES:520 raise521 wait = RETRY_BACKOFF * attempt522 log.warning("[%s] attempt %d/%d failed (%s), retrying in %ds",523 job.key, attempt, MAX_RETRIES, e, wait)524 time.sleep(wait)525 return None526527 def run_job(self, job: Job) -> None:528 if self.manifest.is_done(job.key):529 log.info("[%s] already done, skipping", job.key)530 return531 log.info("[%s] GET %s", job.key, job.display_url)532 t0 = time.time()533 try:534 if job.kind == "meta":535 self._run_meta(job)536 else:537 self._run_bars(job)538 except Exception as e:539 log.error("[%s] FAILED: %s", job.key, e)540 self.manifest.mark(job.key, "failed", error=str(e))541 return542 log.info("[%s] done in %.0fs", job.key, time.time() - t0)543544 def _run_meta(self, job: Job) -> None:545 out_dir = self.data_root / job.dest_subdir546 out_dir.mkdir(parents=True, exist_ok=True)547 fname = "_".join(str(v) for v in job.params.values()) + ".csv"548 out = out_dir / fname549 r = self.session.get(job.url, timeout=600)550 r.raise_for_status()551 out.write_bytes(r.content)552 self.manifest.mark(job.key, "done", bytes=len(r.content), file=str(out))553554 def _run_bars(self, job: Job) -> None:555 zip_dir = self.data_root / "zips"556 zip_dir.mkdir(parents=True, exist_ok=True)557 zip_path = zip_dir / (job.key.replace("|", "_") + ".zip")558559 got = self._download(job, zip_path)560 if got is None:561 self.manifest.mark(job.key, "empty")562 return563 if not zipfile.is_zipfile(zip_path):564 head = zip_path.read_bytes()[:200]565 zip_path.unlink()566 raise RuntimeError(f"Response is not a zip (starts with {head[:60]!r})")567568 asset_type = job.key.split("|")[0]569 dest_dir = self.data_root / job.dest_subdir570 files, rows = extract_and_convert(zip_path, dest_dir, asset_type)571 size = zip_path.stat().st_size572 if not self.keep_zips:573 zip_path.unlink()574 self.manifest.mark(job.key, "done", bytes=size, files=files, rows=rows)575 log.info("[%s] converted %d files / %s rows -> %s",576 job.key, files, f"{rows:,}", dest_dir)577578579# ------------------------------------------------------------------------------580# CLI581# ------------------------------------------------------------------------------582583584def setup_logging(data_root: Path) -> None:585 log_dir = data_root / "logs"586 log_dir.mkdir(parents=True, exist_ok=True)587 fmt = logging.Formatter("%(asctime)s %(levelname)-7s %(message)s",588 "%Y-%m-%d %H:%M:%S")589 for handler in (logging.StreamHandler(sys.stdout),590 logging.FileHandler(log_dir / "frd_downloader.log")):591 handler.setFormatter(fmt)592 log.addHandler(handler)593 log.setLevel(logging.INFO)594595596def main() -> int:597 ap = argparse.ArgumentParser(598 description="FirstRate Data full-history downloader "599 f"(v{__version__} — {__author__} <{__contact__}>)")600 ap.add_argument("--data-root", type=Path,601 default=Path("/Volumes/ssd/firstratedata"),602 help="Root directory for all data (default: %(default)s)")603 ap.add_argument("--types", nargs="+",604 default=["options", "stock", "etf", "futures",605 "crypto", "index", "fx"],606 choices=["options", "stock", "etf", "futures",607 "crypto", "index", "fx"],608 help="Asset types to download (default: all)")609 ap.add_argument("--workers", type=int, default=2,610 help="Parallel downloads (default: %(default)s — be polite)")611 ap.add_argument("--keep-zips", action="store_true",612 help="Keep raw zip archives after conversion")613 ap.add_argument("--include-delisted", action="store_true",614 help="Also download the delisted-stocks archives")615 ap.add_argument("--min-free-gb", type=int, default=MIN_FREE_GB,616 help="Stop downloading below this free space (default: %(default)s)")617 ap.add_argument("--list-jobs", action="store_true",618 help="Print the job plan and exit")619 args = ap.parse_args()620621 if not USERID:622 ap.error("FRD_USERID environment variable is not set "623 "(your FirstRate Data customer id)")624625 jobs = build_jobs(args.types, args.include_delisted)626627 if args.list_jobs:628 for j in jobs:629 print(f"{j.priority:3d} {j.key:45s} {j.display_url}")630 print(f"\nTotal: {len(jobs)} jobs")631 return 0632633 args.data_root.mkdir(parents=True, exist_ok=True)634 setup_logging(args.data_root)635 log.info("FirstRate Data downloader v%s — %s <%s>",636 __version__, __author__, __contact__)637 log.info("Data root: %s | types: %s | %d jobs | %d workers",638 args.data_root, ",".join(args.types), len(jobs), args.workers)639640 manifest = Manifest(args.data_root / "state" / "manifest.json")641 (args.data_root / "state").mkdir(parents=True, exist_ok=True)642 runner = Runner(args.data_root, manifest,643 keep_zips=args.keep_zips, min_free_gb=args.min_free_gb)644645 pending = [j for j in jobs if not manifest.is_done(j.key)]646 log.info("%d jobs pending (%d already complete)",647 len(pending), len(jobs) - len(pending))648649 with ThreadPoolExecutor(max_workers=args.workers) as pool:650 futures = {pool.submit(runner.run_job, j): j for j in pending}651 for fut in as_completed(futures):652 fut.result() # exceptions are handled inside run_job653654 counts = manifest.summary()655 log.info("Run complete. Manifest: %s", counts)656 failed = counts.get("failed", 0)657 return 1 if failed else 0658659660if __name__ == "__main__":661 sys.exit(main())662