SPB Git

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%
27.9 KB · 662 lines python
Raw Blame History
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