frd: row groups de 65 536 lignes pour l'intraday + script de réécriture reprenable du lac

- frd_downloader.csv_to_parquet : ROW_GROUP_SIZE 65536 pour 1min/5min/30min/1hour (déduit du chemin), 1 000 000 pour 1day/options
- scripts/rewrite_row_groups.py : COPY (read_parquet) → tmp (ZSTD, ROW_GROUP_SIZE 65536) puis os.replace, fichier par
  fichier, manifeste JSON de reprise (state/row_groups.json), --dry-run, --asset, --timeframe, --workers, --max-files,
  --retry-failed ; ignore les fichiers déjà découpés ; vérifie le nombre de lignes avant le swap

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 18 days ago (Sep 6, 2026) parent a9d289b

3 changed files +322 −1

modified frd_downloader.py +17 −1

		`@@ -373,6 +373,22 @@ def _ticker_from_filename(path: Path) -> str:`
373	373	return path.stem.split("_")[0].upper()
374	374
375	375
	376	+# Parquet row groups: the API slices intraday files by narrow datetime windows (predicate pushdown skips
	377	+# whole row groups on min/max statistics). 1 M-row groups (≈ 2.5 files per stock) forced DuckDB to decode
	378	+# a third of a 40 MB file for a 5-minute window; 65 536-row groups (≈ 1 week of 1-minute bars) keep the
	379	+# scan proportional to the window. Daily files are small: one big group is fine.
	380	+INTRADAY_TIMEFRAMES = {"1min", "5min", "30min", "1hour"}
	381	+ROW_GROUP_INTRADAY = 65_536
	382	+ROW_GROUP_DAILY = 1_000_000
	383	+
	384	+
	385	+def row_group_size(out_path: Path) -> int:
	386	+ """Row group size for a converted file, from the timeframe directory in its path
	387	+ (`parquet/{type}/{tf}/{adj}/…`, `parquet/futures_contracts/{tf}/{bucket}/…`); options/meta get the big size."""
	388	+ parts = {p for p in out_path.parts}
	389	+ return ROW_GROUP_INTRADAY if parts & INTRADAY_TIMEFRAMES else ROW_GROUP_DAILY
	390	+
	391	+
376	392	def csv_to_parquet(con: duckdb.DuckDBPyConnection, csv_path: Path,
377	393	out_path: Path, asset_type: str) -> int:
378	394	"""Convert one extracted csv/txt file to zstd Parquet. Returns row count."""
		`@@ -413,7 +429,7 @@ def csv_to_parquet(con: duckdb.DuckDBPyConnection, csv_path: Path,`
413	429	f"{common})")
414	430
415	431	con.execute(f"COPY ({select}) TO '{dst}' "
416		− f"(FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 1000000)")
	432	+ f"(FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE {row_group_size(out_path)})")
417	433	rows = con.execute(f"SELECT count(*) FROM read_parquet('{dst}')").fetchone()[0]
418	434	if rows == 0 and csv_path.stat().st_size > 0:
419	435	log.warning("0 rows converted from non-empty file %s (ncols=%d)",

added scripts/rewrite_row_groups.py +214 −0

		`@@ -0,0 +1,214 @@`
	1	+#!/usr/bin/env python3
	2	+"""Rewrite the intraday Parquet files of the lake with small row groups (65 536 rows) — resumable, file by file.
	3	+
	4	+Why: the files written by frd_downloader.py before 2026-09 have 1 000 000-row groups (3 groups for a 2.5 M-row
	5	+1-minute file). A 5-minute window forces DuckDB to decode a whole third of the file; with 65 536-row groups
	6	+(≈ one week of 1-minute bars) the min/max statistics let it skip everything but one group.
	7	+
	8	+Each file is rewritten with DuckDB (`COPY (SELECT * FROM read_parquet(src)) TO tmp (FORMAT PARQUET, COMPRESSION
	9	+ZSTD, ROW_GROUP_SIZE 65536)`), verified (row count identical) and swapped in atomically with `os.replace`.
	10	+Progress is kept in a JSON manifest so the job can be stopped and resumed (nightly PM2 task).
	11	+
	12	+Usage (production node M3U96b — do NOT run while frd_downloader.py rewrites the same directory):
	13	+ cd ~/hfmarketdata && venv/bin/python scripts/rewrite_row_groups.py --data-root ~/firstratedata --dry-run
	14	+ … --asset stock --timeframe 1min one directory family
	15	+ … --workers 3 --row-group-size 65536 parallel files (each worker uses 2 DuckDB threads)
	16	+ … --max-files 500 bounded nightly batch
	17	+ … --manifest ~/firstratedata/state/row_groups.json
	18	+
	19	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	20	+"""
	21	+from __future__ import annotations
	22	+
	23	+import argparse
	24	+import json
	25	+import logging
	26	+import os
	27	+import sys
	28	+import tempfile
	29	+import threading
	30	+import time
	31	+from concurrent.futures import ThreadPoolExecutor, as_completed
	32	+from datetime import datetime, timezone
	33	+from pathlib import Path
	34	+
	35	+import duckdb
	36	+
	37	+log = logging.getLogger("rewrite_row_groups")
	38	+
	39	+ASSETS_WITH_ADJ = ("stock", "etf", "crypto", "index", "fx", "futures", "futures_contracts")
	40	+INTRADAY = ("1min", "5min", "30min", "1hour")
	41	+DEFAULT_ROW_GROUP = 65_536
	42	+
	43	+
	44	+# ---- manifest -------------------------------------------------------------------------------------------
	45	+
	46	+class Manifest:
	47	+ """{"files": {"<relative path>": {"status": "done"\|"failed"\|"skipped", "row_groups": n, "rows": n, "at": iso, ...}}}"""
	48	+
	49	+ def __init__(self, path: Path) -> None:
	50	+ self.path = path
	51	+ self.lock = threading.Lock()
	52	+ self.data: dict = {"version": 1, "row_group_size": None, "files": {}}
	53	+ if path.is_file():
	54	+ try:
	55	+ self.data = json.loads(path.read_text())
	56	+ except json.JSONDecodeError:
	57	+ log.warning("manifest %s unreadable, starting over", path)
	58	+ self.data.setdefault("files", {})
	59	+ self._dirty = 0
	60	+
	61	+ def status(self, rel: str) -> str \| None:
	62	+ return (self.data["files"].get(rel) or {}).get("status")
	63	+
	64	+ def mark(self, rel: str, status: str, **extra) -> None:
	65	+ with self.lock:
	66	+ self.data["files"][rel] = {"status": status, "at": datetime.now(timezone.utc).isoformat(timespec="seconds"), **extra}
	67	+ self._dirty += 1
	68	+ if self._dirty >= 20:
	69	+ self._flush()
	70	+
	71	+ def _flush(self) -> None:
	72	+ self.path.parent.mkdir(parents=True, exist_ok=True)
	73	+ tmp = self.path.with_suffix(".tmp")
	74	+ tmp.write_text(json.dumps(self.data, indent=1, sort_keys=True))
	75	+ os.replace(tmp, self.path)
	76	+ self._dirty = 0
	77	+
	78	+ def flush(self) -> None:
	79	+ with self.lock:
	80	+ self._flush()
	81	+
	82	+
	83	+# ---- discovery --------------------------------------------------------------------------------------------
	84	+
	85	+def candidate_files(parquet_root: Path, assets: list[str] \| None, timeframes: list[str]) -> list[Path]:
	86	+ out: list[Path] = []
	87	+ for asset in assets or ASSETS_WITH_ADJ:
	88	+ for tf in timeframes:
	89	+ base = parquet_root / asset / tf
	90	+ if not base.is_dir():
	91	+ continue
	92	+ for adj_dir in sorted(p for p in base.iterdir() if p.is_dir()):
	93	+ with os.scandir(adj_dir) as it:
	94	+ out.extend(Path(e.path) for e in it if e.name.endswith(".parquet") and e.is_file())
	95	+ return sorted(out)
	96	+
	97	+
	98	+def parquet_layout(con: duckdb.DuckDBPyConnection, path: Path) -> tuple[int, int]:
	99	+ """(row_groups, rows) from the file footer — cheap."""
	100	+ p = str(path).replace("'", "''")
	101	+ rg = con.execute(f"SELECT count(DISTINCT row_group_id), coalesce(sum(row_group_num_rows), 0) "
	102	+ f"FROM parquet_metadata('{p}') WHERE column_id = 0").fetchone()
	103	+ return int(rg[0]), int(rg[1])
	104	+
	105	+
	106	+# ---- rewrite ----------------------------------------------------------------------------------------------
	107	+
	108	+def rewrite_one(path: Path, row_group_size: int, threads: int, dry_run: bool) -> dict:
	109	+ con = duckdb.connect()
	110	+ con.execute(f"SET threads TO {threads}")
	111	+ con.execute("SET memory_limit = '2GB'")
	112	+ try:
	113	+ groups, rows = parquet_layout(con, path)
	114	+ if groups and rows and rows / groups <= row_group_size * 1.5:
	115	+ return {"status": "skipped", "reason": "row groups already small", "row_groups": groups, "rows": rows}
	116	+ if dry_run:
	117	+ return {"status": "dry-run", "row_groups": groups, "rows": rows,
	118	+ "target_groups": -(-rows // row_group_size) if rows else 0}
	119	+ src = str(path).replace("'", "''")
	120	+ fd, tmp_name = tempfile.mkstemp(prefix=f".{path.stem}.", suffix=".parquet.tmp", dir=str(path.parent))
	121	+ os.close(fd)
	122	+ tmp = Path(tmp_name)
	123	+ try:
	124	+ t0 = time.perf_counter()
	125	+ dst = str(tmp).replace("'", "''")
	126	+ con.execute(f"COPY (SELECT * FROM read_parquet('{src}')) TO '{dst}' "
	127	+ f"(FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE {int(row_group_size)})")
	128	+ new_groups, new_rows = parquet_layout(con, tmp)
	129	+ if new_rows != rows:
	130	+ raise RuntimeError(f"row count mismatch after rewrite: {rows} → {new_rows}")
	131	+ old_size, new_size = path.stat().st_size, tmp.stat().st_size
	132	+ os.replace(tmp, path) # atomic on the same filesystem; readers see old or new, never partial
	133	+ return {"status": "done", "row_groups": new_groups, "rows": rows, "old_row_groups": groups,
	134	+ "bytes_before": old_size, "bytes_after": new_size, "seconds": round(time.perf_counter() - t0, 2)}
	135	+ finally:
	136	+ if tmp.exists():
	137	+ tmp.unlink(missing_ok=True)
	138	+ finally:
	139	+ con.close()
	140	+
	141	+
	142	+def main() -> int:
	143	+ ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
	144	+ ap.add_argument("--data-root", type=Path, default=Path(os.environ.get("HFMD_DATA_ROOT", "/Volumes/ssd/firstratedata")))
	145	+ ap.add_argument("--asset", action="append", help="restrict to an asset directory (repeatable): stock, etf, futures, futures_contracts, crypto, index, fx")
	146	+ ap.add_argument("--timeframe", action="append", help="restrict to a timeframe (repeatable): 1min, 5min, 30min, 1hour (default: all intraday)")
	147	+ ap.add_argument("--row-group-size", type=int, default=DEFAULT_ROW_GROUP)
	148	+ ap.add_argument("--workers", type=int, default=2, help="files rewritten in parallel (default 2)")
	149	+ ap.add_argument("--threads-per-worker", type=int, default=2)
	150	+ ap.add_argument("--max-files", type=int, default=0, help="stop after N files rewritten (0 = no bound)")
	151	+ ap.add_argument("--manifest", type=Path, help="progress file (default <data-root>/state/row_groups.json)")
	152	+ ap.add_argument("--retry-failed", action="store_true", help="re-attempt files marked failed in the manifest")
	153	+ ap.add_argument("--dry-run", action="store_true", help="list what would be rewritten, touch nothing")
	154	+ ap.add_argument("--verbose", "-v", action="store_true")
	155	+ args = ap.parse_args()
	156	+ logging.basicConfig(level=logging.INFO if args.verbose or args.dry_run else logging.WARNING,
	157	+ format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout)
	158	+
	159	+ parquet_root = args.data_root / "parquet"
	160	+ if not parquet_root.is_dir():
	161	+ log.error("no parquet directory under %s", args.data_root)
	162	+ return 2
	163	+ tfs = args.timeframe or list(INTRADAY)
	164	+ bad = [t for t in tfs if t not in INTRADAY]
	165	+ if bad:
	166	+ log.error("only intraday timeframes are rewritten (%s); got %s", ", ".join(INTRADAY), ", ".join(bad))
	167	+ return 2
	168	+ manifest = Manifest(args.manifest or (args.data_root / "state" / "row_groups.json"))
	169	+ manifest.data["row_group_size"] = args.row_group_size
	170	+
	171	+ files = candidate_files(parquet_root, args.asset, tfs)
	172	+ todo = []
	173	+ for f in files:
	174	+ rel = str(f.relative_to(parquet_root))
	175	+ st = manifest.status(rel)
	176	+ if st in ("done", "skipped") or (st == "failed" and not args.retry_failed):
	177	+ continue
	178	+ todo.append((f, rel))
	179	+ if args.max_files:
	180	+ todo = todo[:args.max_files]
	181	+ log.info("%d candidate files, %d to process (%s)", len(files), len(todo), "dry run" if args.dry_run else "rewrite")
	182	+
	183	+ counts = {"done": 0, "skipped": 0, "failed": 0, "dry-run": 0}
	184	+ t0 = time.time()
	185	+
	186	+ def job(item):
	187	+ f, rel = item
	188	+ try:
	189	+ return rel, rewrite_one(f, args.row_group_size, args.threads_per_worker, args.dry_run)
	190	+ except Exception as e: # noqa: BLE001 — one bad file must not stop the batch
	191	+ return rel, {"status": "failed", "error": f"{e.__class__.__name__}: {str(e)[:300]}"}
	192	+
	193	+ try:
	194	+ with ThreadPoolExecutor(max_workers=max(1, args.workers)) as ex:
	195	+ for rel, res in (fut.result() for fut in as_completed([ex.submit(job, it) for it in todo])):
	196	+ counts[res["status"]] = counts.get(res["status"], 0) + 1
	197	+ if res["status"] == "dry-run":
	198	+ print(f"would rewrite {rel}: {res['row_groups']} groups → {res['target_groups']} ({res['rows']:,} rows)")
	199	+ continue
	200	+ manifest.mark(rel, **res)
	201	+ if res["status"] == "failed":
	202	+ log.error("FAILED %s: %s", rel, res["error"])
	203	+ else:
	204	+ log.info("%s %s (%s)", res["status"], rel, ", ".join(f"{k}={v}" for k, v in res.items() if k not in ("status",)))
	205	+ finally:
	206	+ if not args.dry_run:
	207	+ manifest.flush()
	208	+ summary = {"processed": len(todo), **counts, "seconds": round(time.time() - t0, 1), "manifest": str(manifest.path)}
	209	+ print(json.dumps(summary))
	210	+ return 1 if counts.get("failed") else 0
	211	+
	212	+
	213	+if __name__ == "__main__":
	214	+ sys.exit(main())

added tests/test_row_groups.py +91 −0

		`@@ -0,0 +1,91 @@`
	1	+"""Row-group sizing: frd_downloader writes 65 536-row groups for intraday files (1 M for daily), and
	2	+scripts/rewrite_row_groups.py rewrites existing files in place, resumably."""
	3	+from __future__ import annotations
	4	+
	5	+import importlib.util
	6	+import json
	7	+import sys
	8	+from pathlib import Path
	9	+
	10	+import duckdb
	11	+import pandas as pd
	12	+import pytest
	13	+
	14	+ROOT = Path(__file__).resolve().parents[1]
	15	+
	16	+
	17	+def _load(name: str, path: Path):
	18	+ spec = importlib.util.spec_from_file_location(name, path)
	19	+ mod = importlib.util.module_from_spec(spec)
	20	+ sys.modules[name] = mod
	21	+ spec.loader.exec_module(mod)
	22	+ return mod
	23	+
	24	+
	25	+@pytest.fixture(scope="module")
	26	+def frd():
	27	+ if importlib.util.find_spec("requests") is None: # downloader-only dependency, not in the API venv
	28	+ import types
	29	+ stub = types.ModuleType("requests")
	30	+ stub.Session, stub.HTTPError, stub.RequestException = object, Exception, Exception # type: ignore[attr-defined]
	31	+ sys.modules.setdefault("requests", stub)
	32	+ return _load("frd_downloader_under_test", ROOT / "frd_downloader.py")
	33	+
	34	+
	35	+@pytest.fixture(scope="module")
	36	+def rewriter():
	37	+ return _load("rewrite_row_groups_under_test", ROOT / "scripts" / "rewrite_row_groups.py")
	38	+
	39	+
	40	+def test_row_group_size_by_timeframe(frd):
	41	+ pq = Path("/lake/parquet")
	42	+ assert frd.row_group_size(pq / "stock" / "1min" / "UNADJUSTED" / "AAPL_1min.parquet") == 65_536
	43	+ assert frd.row_group_size(pq / "futures_contracts" / "1hour" / "update" / "ES_Z25_1hour.parquet") == 65_536
	44	+ assert frd.row_group_size(pq / "stock" / "1day" / "adj_splitdiv" / "AAPL_1day.parquet") == 1_000_000
	45	+ assert frd.row_group_size(pq / "options" / "2025_q2" / "AAPL_month_option_chain.parquet") == 1_000_000
	46	+
	47	+
	48	+def _big_file(path: Path, rows: int = 200_000) -> None:
	49	+ idx = pd.date_range("2024-01-02 09:30", periods=rows, freq="min")
	50	+ df = pd.DataFrame({"ticker": "AAA", "datetime": idx, "open": 1.0, "high": 1.0, "low": 1.0, "close": 1.0, "volume": 1.0})
	51	+ path.parent.mkdir(parents=True, exist_ok=True)
	52	+ con = duckdb.connect()
	53	+ con.register("df", df)
	54	+ con.execute(f"COPY df TO '{path}' (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 1000000)")
	55	+
	56	+
	57	+def test_rewrite_is_resumable_and_atomic(rewriter, tmp_path, monkeypatch, capsys):
	58	+ root = tmp_path / "lake"
	59	+ f1 = root / "parquet" / "stock" / "1min" / "UNADJUSTED" / "AAA_1min.parquet"
	60	+ f2 = root / "parquet" / "stock" / "5min" / "adj_splitdiv" / "AAA_5min.parquet"
	61	+ daily = root / "parquet" / "stock" / "1day" / "adj_splitdiv" / "AAA_1day.parquet"
	62	+ _big_file(f1)
	63	+ _big_file(f2, rows=100_000)
	64	+ _big_file(daily, rows=5_000)
	65	+ con = duckdb.connect()
	66	+ assert rewriter.parquet_layout(con, f1)[0] == 1
	67	+
	68	+ # dry run touches nothing
	69	+ monkeypatch.setattr(sys, "argv", ["x", "--data-root", str(root), "--dry-run"])
	70	+ assert rewriter.main() == 0
	71	+ assert rewriter.parquet_layout(con, f1)[0] == 1 and not (root / "state" / "row_groups.json").exists()
	72	+ assert "would rewrite" in capsys.readouterr().out
	73	+
	74	+ # one file only (--max-files 1), then resume for the rest
	75	+ monkeypatch.setattr(sys, "argv", ["x", "--data-root", str(root), "--max-files", "1", "--workers", "1"])
	76	+ assert rewriter.main() == 0
	77	+ manifest = json.loads((root / "state" / "row_groups.json").read_text())
	78	+ done = [k for k, v in manifest["files"].items() if v["status"] == "done"]
	79	+ assert len(done) == 1
	80	+ monkeypatch.setattr(sys, "argv", ["x", "--data-root", str(root), "--workers", "2"])
	81	+ assert rewriter.main() == 0
	82	+ manifest = json.loads((root / "state" / "row_groups.json").read_text())
	83	+ assert {v["status"] for v in manifest["files"].values()} == {"done"}
	84	+ assert len(manifest["files"]) == 2 # daily file never a candidate
	85	+ groups, rows = rewriter.parquet_layout(con, f1)
	86	+ assert rows == 200_000 and groups == -(-200_000 // 65_536)
	87	+ assert rewriter.parquet_layout(con, daily)[0] == 1
	88	+ assert not list(f1.parent.glob(".*.tmp")) # no temp files left behind
	89	+ # idempotent: a third run processes nothing
	90	+ assert rewriter.main() == 0
	91	+ assert json.loads(capsys.readouterr().out.strip().splitlines()[-1])["processed"] == 0
	92