spb/hfmarketdata

1"""Build a tiny synthetic Parquet lake with the exact layout produced by frd_downloader.py.23    parquet/{stock|etf|crypto|index|fx}/{timeframe}/{adjustment}/{TICKER}_{timeframe}.parquet4    parquet/futures/{timeframe}/{contin_UNadj|contin_adj_ratio|contin_adj_absolute}/{ROOT}_{timeframe}.parquet5    parquet/futures_contracts/{timeframe}/{archive|update}/{ROOT}_{MonthCode}{YY}_{timeframe}.parquet6    parquet/options/{year}_{quarter}/{TICKER}_month_option_chain.parquet7    meta/futures/futures.csv89Every bar file carries a `ticker` column (root only for futures contracts — the contract identity is10in the file name, exactly like the real lake). Futures contracts get a realistic life-cycle: volume and11open interest ramp up towards expiry and collapse in the last sessions (so volume/OI rolls are testable),12daily data ends on the exchange expiry (3rd Friday for ES/E6, 20th of the preceding month for CL/NG),13intraday bars cover RTH 09:30–16:00 plus a pre-market ETH block 08:00–09:29 (Eastern, naive).14"""15from __future__ import annotations1617import csv18from datetime import date, timedelta19from pathlib import Path2021import numpy as np22import pandas as pd2324TIMEFRAMES = ["1min", "5min", "30min", "1hour", "1day"]25TICKERS = {"stock": ["AAPL", "MSFT", "SMCP", "SHAK", "GOOG", "GOOGL"], "etf": ["SPY"], "crypto": ["BTCUSD"],26           "index": ["SPX"], "fx": ["EURUSD"]}27ADJ = {"stock": ["adj_split", "adj_splitdiv", "UNADJUSTED"], "etf": ["adj_split", "adj_splitdiv", "UNADJUSTED"],28       "crypto": ["none"], "index": ["none"], "fx": ["none"]}29ROOTS = ["ES", "CL", "NG", "E6"]30MONTHS = {"F": 1, "G": 2, "H": 3, "J": 4, "K": 5, "M": 6, "N": 7, "Q": 8, "U": 9, "V": 10, "X": 11, "Z": 12}31CYCLE = {"ES": "HMUZ", "CL": "FJNV", "NG": "FJNV", "E6": "HMUZ"}      # CL/NG: quarterly subset to keep the lake small32INTRADAY_DAYS = {"ES": 30}                                             # sessions of intraday history per contract (default 5)333435def _third_friday(y: int, m: int) -> date:36    d = date(y, m, 15)37    return d + timedelta(days=(4 - d.weekday()) % 7)383940def contract_expiry(root: str, yy: int, month_code: str) -> date:41    y, m = 2000 + yy, MONTHS[month_code]42    if root in ("ES", "E6"):43        return _third_friday(y, m)44    pm = m - 1 or 1245    return date(y - (m == 1), pm, 20)464748def _lifecycle(dte: np.ndarray, rng: np.random.Generator) -> tuple[np.ndarray, np.ndarray]:49    """(volume, open_interest) as a function of days to expiry: ramp up, collapse in the last sessions."""50    w = np.where(dte > 8, (150.0 - dte) / 100.0, 1.42 * (dte / 8.0) ** 2 + 0.02)51    w = np.clip(w, 0.02, None) * rng.uniform(0.97, 1.03, len(dte))52    vol = np.round(10_000 * w)53    oi = np.clip((200.0 - dte) / 150.0, 0.05, 1.0) * np.where(dte > 10, 1.0, dte / 10.0)54    oi = np.round(100_000 * oi * rng.uniform(0.98, 1.02, len(dte)))55    return vol, oi565758def _bars(ticker: str, start: date, end: date, tf: str, seed: int, oi: bool = False, base: float = 100.0,59          expiry: date | None = None, eth: bool = False, days: int = 5) -> pd.DataFrame:60    rng = np.random.default_rng(seed)61    if tf == "1day":62        idx = pd.bdate_range(start, end)63    else:64        step = {"1min": 1, "5min": 5, "30min": 30, "1hour": 60}[tf]65        sessions = pd.bdate_range(start, end)[-days:]66        starts = [timedelta(hours=9, minutes=30)]67        counts = [int(390 / step)]68        if eth and step <= 30:                     # pre-market block 08:00–09:2969            starts.insert(0, timedelta(hours=8))70            counts.insert(0, int(90 / step))71        idx = pd.DatetimeIndex([d + s + timedelta(minutes=step * i) for d in sessions for s, n in zip(starts, counts) for i in range(n)])72    n = len(idx)73    close = base + np.cumsum(rng.normal(0, 1, n))74    df = pd.DataFrame({"ticker": ticker, "datetime": idx, "open": close + rng.normal(0, .2, n),75                       "high": close + abs(rng.normal(0, .5, n)), "low": close - abs(rng.normal(0, .5, n)),76                       "close": close})77    if expiry is not None:78        dte = np.array([(expiry - d.date()).days for d in idx], dtype=float)79        vol, oi_v = _lifecycle(dte, rng)80        df["volume"] = vol if tf == "1day" else np.round(vol / 400)81        if oi:82            df["open_interest"] = oi_v83    else:84        df["volume"] = rng.integers(100, 10_000, n).astype(float)85        if oi:86            df["open_interest"] = rng.integers(1_000, 100_000, n).astype(float)87    return df888990def build_lake(root: Path, start: date = date(2023, 1, 2), end: date = date(2025, 6, 30)) -> None:91    pq = root / "parquet"92    seed = 193    for asset, tickers in TICKERS.items():94        for tf in TIMEFRAMES:95            for adj in ADJ[asset]:96                if tf not in ("1min", "1day") and adj == "UNADJUSTED":97                    continue98                d = pq / asset / tf / adj99                d.mkdir(parents=True, exist_ok=True)100                for t in tickers:101                    seed += 1102                    _bars(t, start, end, tf, seed).to_parquet(d / f"{t}_{tf}.parquet", index=False)103    for tf in TIMEFRAMES:104        for adj in ("contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"):105            d = pq / "futures" / tf / adj106            d.mkdir(parents=True, exist_ok=True)107            for r in ROOTS:108                seed += 1109                _bars(r, start, end, tf, seed, oi=(tf == "1day")).to_parquet(d / f"{r}_{tf}.parquet", index=False)110    # individual contracts: archive = up to 2024, update = 2025+ (with overlap for 2025 contracts)111    for tf in TIMEFRAMES:112        for r in ROOTS:113            base = {"ES": 5000.0, "CL": 70.0, "NG": 3.0, "E6": 1.08}[r]114            for yy in (23, 24, 25, 26):115                for mc in CYCLE[r]:116                    exp = contract_expiry(r, yy, mc)117                    first = exp - timedelta(days=400)118                    last = min(exp, end)119                    if first > end:120                        continue121                    bucket = "update" if yy >= 25 else "archive"122                    d = pq / "futures_contracts" / tf / bucket123                    d.mkdir(parents=True, exist_ok=True)124                    seed += 1125                    kw = dict(oi=(tf == "1day"), base=base + (yy * 4 + MONTHS[mc] / 3) * base / 400, expiry=exp, eth=True,126                              days=INTRADAY_DAYS.get(r, 5))127                    _bars(r, max(first, date(2022, 1, 3)), last, tf, seed, **kw).to_parquet(d / f"{r}_{mc}{yy}_{tf}.parquet", index=False)128                    if bucket == "update" and yy == 25:  # archive also holds the first part of 2025 contracts129                        d2 = pq / "futures_contracts" / tf / "archive"130                        d2.mkdir(parents=True, exist_ok=True)131                        _bars(r, max(first, date(2022, 1, 3)), min(last, date(2024, 12, 31)), tf, seed, **kw).to_parquet(132                            d2 / f"{r}_{mc}{yy}_{tf}.parquet", index=False)133    # options: one quarter, one ticker134    d = pq / "options" / "2025_q2"135    d.mkdir(parents=True, exist_ok=True)136    rows = []137    for td in pd.bdate_range("2025-04-01", "2025-04-10"):138        for k in (180, 190, 200, 210):139            for cp in ("c", "p"):140                rows.append({"ticker": "AAPL", "trade_date": td.date(), "strike": float(k), "expiry": date(2025, 6, 20),141                             "call_put": cp, "bid": 1.0, "ask": 1.2, "last": 1.1, "volume": 10.0, "open_interest": 100.0,142                             "iv": 0.25, "delta": 0.5 if cp == "c" else -0.5, "gamma": 0.01, "theta": -0.02, "vega": 0.1,143                             "rho": 0.01, "underlying_price": 195.0})144    pd.DataFrame(rows).to_parquet(d / "AAPL_month_option_chain.parquet", index=False)145    # metadata146    m = root / "meta" / "futures"147    m.mkdir(parents=True, exist_ok=True)148    with open(m / "futures.csv", "w", newline="") as f:149        w = csv.writer(f)150        w.writerow(["Ticker", "Name", "First Date", "Last Date"])151        w.writerow(["ES", "E-mini S&P 500 (CME) ", "2008-01-02", str(end)])152        w.writerow(["CL", "Crude Oil WTI (NYMEX) ", "2008-01-02", str(end)])153        w.writerow(["NG", "Natural Gas (NYMEX) ", "2008-01-02", str(end)])154        w.writerow(["E6", "Euro FX Futures (CME) ", "2008-01-02", str(end)])155        w.writerow(["ZK", "Unknown Product (XXX) ", "2008-01-02", str(end)])156    (root / "state").mkdir(exist_ok=True)157158159if __name__ == "__main__":160    import sys161    build_lake(Path(sys.argv[1] if len(sys.argv) > 1 else "/tmp/hfmd-lake"))162    print("fixture lake built")163