thinkingEverytime/QuantOracle

1#!/usr/bin/env python32"""EOD ingest: download OHLCV and write to local Parquet store.3 4Usage:5  python scripts/ingest_eod.py --symbols RELIANCE.NS,TCS.NS --period 5y6  python scripts/ingest_eod.py --symbols-file symbols.txt --period max7  python scripts/ingest_eod.py --universe data/universe/india_core.txt --period 5y8"""9 10# ruff: noqa: E402  (sys.path bootstrap must run before local imports)11 12from __future__ import annotations13 14import argparse15from pathlib import Path16import time17 18import pandas as pd19import yfinance as yf20 21import sys22 23from pathlib import Path as _Path24 25# Scripts are executed with `scripts/` as sys.path[0]. Add repo root + frontend for imports.26_ROOT = str(_Path(__file__).resolve().parents[1])27if _ROOT not in sys.path:28    sys.path.insert(0, _ROOT)29_FRONTEND = str(_Path(__file__).resolve().parents[1] / "frontend")30if _FRONTEND not in sys.path:31    sys.path.insert(0, _FRONTEND)32 33from services.store import write_ohlcv34 35 36def _normalize_symbol(sym: str) -> str:37    # Keep ingest independent of Streamlit/UI modules.38    return sym.strip().upper()39 40 41def _read_symbols(path: Path) -> list[str]:42    out = []43    for line in path.read_text(encoding="utf-8").splitlines():44        line = line.strip()45        if not line or line.startswith("#"):46            continue47        out.append(line)48    return out49 50 51def _chunks(xs: list[str], n: int) -> list[list[str]]:52    n = max(1, int(n))53    return [xs[i : i + n] for i in range(0, len(xs), n)]54 55 56def _download(syms: list[str], period: str, auto_adjust: bool, threads: bool, progress: bool) -> pd.DataFrame:57    try:58        df = yf.download(59            syms,60            period=period,61            group_by="ticker",62            auto_adjust=auto_adjust,63            threads=threads,64            progress=progress,65        )66        return df if isinstance(df, pd.DataFrame) else pd.DataFrame()67    except Exception:68        return pd.DataFrame()69 70 71def _extract(df: pd.DataFrame, sym: str) -> pd.DataFrame:72    if df is None or df.empty:73        return pd.DataFrame()74    if isinstance(df.columns, pd.MultiIndex):75        if sym not in df.columns.get_level_values(0):76            return pd.DataFrame()77        return df[sym].dropna(how="all")78    return df.dropna(how="all")79 80 81def main():82    ap = argparse.ArgumentParser()83    ap.add_argument("--symbols", default="", help="Comma-separated symbols")84    ap.add_argument("--symbols-file", default="", help="Text file with 1 symbol per line")85    ap.add_argument("--universe", default="", help="Alias for --symbols-file")86    ap.add_argument("--period", default="5y", help="yfinance period: 1y/5y/max")87    ap.add_argument("--no-adjust", action="store_true", help="Disable auto_adjust")88    ap.add_argument("--batch-size", type=int, default=10, help="Download tickers in batches (reduces yfinance flakiness)")89    ap.add_argument("--retries", type=int, default=2, help="Retries per batch/symbol on transient failures")90    ap.add_argument("--threads", action="store_true", help="Enable yfinance threads (faster, less reliable)")91    args = ap.parse_args()92 93    syms = []94    if args.symbols:95        syms += [s.strip() for s in args.symbols.split(",") if s.strip()]96    symbols_file = args.symbols_file or args.universe97    if symbols_file:98        syms += _read_symbols(Path(symbols_file))99    syms = [_normalize_symbol(s) for s in dict.fromkeys(syms)]  # de-dupe, keep order100 101    if not syms:102        raise SystemExit("No symbols provided")103 104    auto_adjust = not args.no_adjust105    batches = _chunks(syms, args.batch_size)106 107    wrote = 0108    failed_dl: list[str] = []109    failed_write: list[str] = []110 111    for batch in batches:112        df = pd.DataFrame()113        for attempt in range(args.retries + 1):114            df = _download(batch, args.period, auto_adjust, threads=args.threads, progress=(attempt == 0))115            if not df.empty:116                break117            time.sleep(0.75 * (attempt + 1))118 119        for sym in batch:120            h = _extract(df, sym)121 122            # Fallback: per-symbol download tends to be more reliable than multi-ticker.123            if h.empty:124                one = pd.DataFrame()125                for attempt in range(args.retries + 1):126                    one = _download([sym], args.period, auto_adjust, threads=False, progress=False)127                    h = _extract(one, sym)128                    if not h.empty:129                        break130                    time.sleep(0.75 * (attempt + 1))131 132            if h.empty:133                failed_dl.append(sym)134                print(f"Skip {sym}: empty (download failed)")135                continue136 137            out = write_ohlcv(sym, h)138            if out is None:139                failed_write.append(sym)140                print(f"Failed {sym}: could not write Parquet (is `duckdb` installed?)")141            else:142                wrote += 1143                print(f"Wrote {sym} -> {out}")144 145    if failed_dl:146        print(f"\nDownload failed ({len(failed_dl)}): {failed_dl}")147    if failed_write:148        print(f"\nWrite failed ({len(failed_write)}): {failed_write}")149 150    if wrote == 0:151        raise SystemExit("No data written. yfinance may be blocked/rate-limited; try again later or reduce universe.")152    return 2 if failed_dl or failed_write else 0153 154 155if __name__ == "__main__":156    raise SystemExit(main())157