"""Build a tiny synthetic Parquet lake with the exact layout produced by frd_downloader.py. parquet/{stock|etf|crypto|index|fx}/{timeframe}/{adjustment}/{TICKER}_{timeframe}.parquet parquet/futures/{timeframe}/{contin_UNadj|contin_adj_ratio|contin_adj_absolute}/{ROOT}_{timeframe}.parquet parquet/futures_contracts/{timeframe}/{archive|update}/{ROOT}_{MonthCode}{YY}_{timeframe}.parquet parquet/options/{year}_{quarter}/{TICKER}_month_option_chain.parquet meta/futures/futures.csv Every bar file carries a `ticker` column (root only for futures contracts — the contract identity is in the file name, exactly like the real lake). Futures contracts get a realistic life-cycle: volume and open interest ramp up towards expiry and collapse in the last sessions (so volume/OI rolls are testable), daily data ends on the exchange expiry (3rd Friday for ES/E6, 20th of the preceding month for CL/NG), intraday bars cover RTH 09:30–16:00 plus a pre-market ETH block 08:00–09:29 (Eastern, naive). """ from __future__ import annotations import csv from datetime import date, timedelta from pathlib import Path import numpy as np import pandas as pd TIMEFRAMES = ["1min", "5min", "30min", "1hour", "1day"] TICKERS = {"stock": ["AAPL", "MSFT", "SMCP", "SHAK", "GOOG", "GOOGL"], "etf": ["SPY"], "crypto": ["BTCUSD"], "index": ["SPX"], "fx": ["EURUSD"]} ADJ = {"stock": ["adj_split", "adj_splitdiv", "UNADJUSTED"], "etf": ["adj_split", "adj_splitdiv", "UNADJUSTED"], "crypto": ["none"], "index": ["none"], "fx": ["none"]} ROOTS = ["ES", "CL", "NG", "E6"] MONTHS = {"F": 1, "G": 2, "H": 3, "J": 4, "K": 5, "M": 6, "N": 7, "Q": 8, "U": 9, "V": 10, "X": 11, "Z": 12} CYCLE = {"ES": "HMUZ", "CL": "FJNV", "NG": "FJNV", "E6": "HMUZ"} # CL/NG: quarterly subset to keep the lake small INTRADAY_DAYS = {"ES": 30} # sessions of intraday history per contract (default 5) def _third_friday(y: int, m: int) -> date: d = date(y, m, 15) return d + timedelta(days=(4 - d.weekday()) % 7) def contract_expiry(root: str, yy: int, month_code: str) -> date: y, m = 2000 + yy, MONTHS[month_code] if root in ("ES", "E6"): return _third_friday(y, m) pm = m - 1 or 12 return date(y - (m == 1), pm, 20) def _lifecycle(dte: np.ndarray, rng: np.random.Generator) -> tuple[np.ndarray, np.ndarray]: """(volume, open_interest) as a function of days to expiry: ramp up, collapse in the last sessions.""" w = np.where(dte > 8, (150.0 - dte) / 100.0, 1.42 * (dte / 8.0) ** 2 + 0.02) w = np.clip(w, 0.02, None) * rng.uniform(0.97, 1.03, len(dte)) vol = np.round(10_000 * w) oi = np.clip((200.0 - dte) / 150.0, 0.05, 1.0) * np.where(dte > 10, 1.0, dte / 10.0) oi = np.round(100_000 * oi * rng.uniform(0.98, 1.02, len(dte))) return vol, oi def _bars(ticker: str, start: date, end: date, tf: str, seed: int, oi: bool = False, base: float = 100.0, expiry: date | None = None, eth: bool = False, days: int = 5) -> pd.DataFrame: rng = np.random.default_rng(seed) if tf == "1day": idx = pd.bdate_range(start, end) else: step = {"1min": 1, "5min": 5, "30min": 30, "1hour": 60}[tf] sessions = pd.bdate_range(start, end)[-days:] starts = [timedelta(hours=9, minutes=30)] counts = [int(390 / step)] if eth and step <= 30: # pre-market block 08:00–09:29 starts.insert(0, timedelta(hours=8)) counts.insert(0, int(90 / step)) 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)]) n = len(idx) close = base + np.cumsum(rng.normal(0, 1, n)) df = pd.DataFrame({"ticker": ticker, "datetime": idx, "open": close + rng.normal(0, .2, n), "high": close + abs(rng.normal(0, .5, n)), "low": close - abs(rng.normal(0, .5, n)), "close": close}) if expiry is not None: dte = np.array([(expiry - d.date()).days for d in idx], dtype=float) vol, oi_v = _lifecycle(dte, rng) df["volume"] = vol if tf == "1day" else np.round(vol / 400) if oi: df["open_interest"] = oi_v else: df["volume"] = rng.integers(100, 10_000, n).astype(float) if oi: df["open_interest"] = rng.integers(1_000, 100_000, n).astype(float) return df def build_lake(root: Path, start: date = date(2023, 1, 2), end: date = date(2025, 6, 30)) -> None: pq = root / "parquet" seed = 1 for asset, tickers in TICKERS.items(): for tf in TIMEFRAMES: for adj in ADJ[asset]: if tf not in ("1min", "1day") and adj == "UNADJUSTED": continue d = pq / asset / tf / adj d.mkdir(parents=True, exist_ok=True) for t in tickers: seed += 1 _bars(t, start, end, tf, seed).to_parquet(d / f"{t}_{tf}.parquet", index=False) for tf in TIMEFRAMES: for adj in ("contin_UNadj", "contin_adj_ratio", "contin_adj_absolute"): d = pq / "futures" / tf / adj d.mkdir(parents=True, exist_ok=True) for r in ROOTS: seed += 1 _bars(r, start, end, tf, seed, oi=(tf == "1day")).to_parquet(d / f"{r}_{tf}.parquet", index=False) # individual contracts: archive = up to 2024, update = 2025+ (with overlap for 2025 contracts) for tf in TIMEFRAMES: for r in ROOTS: base = {"ES": 5000.0, "CL": 70.0, "NG": 3.0, "E6": 1.08}[r] for yy in (23, 24, 25, 26): for mc in CYCLE[r]: exp = contract_expiry(r, yy, mc) first = exp - timedelta(days=400) last = min(exp, end) if first > end: continue bucket = "update" if yy >= 25 else "archive" d = pq / "futures_contracts" / tf / bucket d.mkdir(parents=True, exist_ok=True) seed += 1 kw = dict(oi=(tf == "1day"), base=base + (yy * 4 + MONTHS[mc] / 3) * base / 400, expiry=exp, eth=True, days=INTRADAY_DAYS.get(r, 5)) _bars(r, max(first, date(2022, 1, 3)), last, tf, seed, **kw).to_parquet(d / f"{r}_{mc}{yy}_{tf}.parquet", index=False) if bucket == "update" and yy == 25: # archive also holds the first part of 2025 contracts d2 = pq / "futures_contracts" / tf / "archive" d2.mkdir(parents=True, exist_ok=True) _bars(r, max(first, date(2022, 1, 3)), min(last, date(2024, 12, 31)), tf, seed, **kw).to_parquet( d2 / f"{r}_{mc}{yy}_{tf}.parquet", index=False) # options: one quarter, one ticker d = pq / "options" / "2025_q2" d.mkdir(parents=True, exist_ok=True) rows = [] for td in pd.bdate_range("2025-04-01", "2025-04-10"): for k in (180, 190, 200, 210): for cp in ("c", "p"): rows.append({"ticker": "AAPL", "trade_date": td.date(), "strike": float(k), "expiry": date(2025, 6, 20), "call_put": cp, "bid": 1.0, "ask": 1.2, "last": 1.1, "volume": 10.0, "open_interest": 100.0, "iv": 0.25, "delta": 0.5 if cp == "c" else -0.5, "gamma": 0.01, "theta": -0.02, "vega": 0.1, "rho": 0.01, "underlying_price": 195.0}) pd.DataFrame(rows).to_parquet(d / "AAPL_month_option_chain.parquet", index=False) # metadata m = root / "meta" / "futures" m.mkdir(parents=True, exist_ok=True) with open(m / "futures.csv", "w", newline="") as f: w = csv.writer(f) w.writerow(["Ticker", "Name", "First Date", "Last Date"]) w.writerow(["ES", "E-mini S&P 500 (CME) ", "2008-01-02", str(end)]) w.writerow(["CL", "Crude Oil WTI (NYMEX) ", "2008-01-02", str(end)]) w.writerow(["NG", "Natural Gas (NYMEX) ", "2008-01-02", str(end)]) w.writerow(["E6", "Euro FX Futures (CME) ", "2008-01-02", str(end)]) w.writerow(["ZK", "Unknown Product (XXX) ", "2008-01-02", str(end)]) (root / "state").mkdir(exist_ok=True) if __name__ == "__main__": import sys build_lake(Path(sys.argv[1] if len(sys.argv) > 1 else "/tmp/hfmd-lake")) print("fixture lake built")