Handle FX intraday split date+time columns; 7zz fallback for Deflate64
FX intraday files are {yyyyMMdd},{HH:mm:ss},O,H,L,C,V — the 7-column layout
was being parsed as datetime+OHLCV+open_interest, silently dropping every row.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Showing 1 changed file with +36 and −9
modified
frd_downloader.py
+36 −9
@@ -63,6 +63,7 @@ import argparse | ||
| 63 | 63 | import json |
| 64 | 64 | import logging |
| 65 | 65 | import os |
| 66 | +import re | |
| 66 | 67 | import shutil |
| 67 | 68 | import subprocess |
| 68 | 69 | import sys |
@@ -344,13 +345,27 @@ OPTIONS_SCHEMA = { | ||
| 344 | 345 | } |
| 345 | 346 | |
| 346 | 347 | |
| 347 | −def _sniff_column_count(csv_path: Path) -> int: | |
| 348 | +def _sniff_first_line(csv_path: Path) -> list[str]: | |
| 348 | 349 | with open(csv_path, "r", errors="replace") as f: |
| 349 | 350 | for line in f: |
| 350 | 351 | line = line.strip() |
| 351 | 352 | if line: |
| 352 | − return line.count(",") + 1 | |
| 353 | − return 0 | |
| 353 | + return line.split(",") | |
| 354 | + return [] | |
| 355 | + | |
| 356 | + | |
| 357 | +_TIME_RE = re.compile(r"^\d{2}:\d{2}(:\d{2})?$") | |
| 358 | + | |
| 359 | +# FX intraday layout: date and time come as two separate fields | |
| 360 | +# ({yyyyMMdd},{HH:mm:ss},O,H,L,C,V per the fx readme). | |
| 361 | +SPLIT_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'))") | |
| 354 | 369 | |
| 355 | 370 | |
| 356 | 371 | def _ticker_from_filename(path: Path) -> str: |
@@ -361,9 +376,11 @@ def _ticker_from_filename(path: Path) -> str: | ||
| 361 | 376 | def csv_to_parquet(con: duckdb.DuckDBPyConnection, csv_path: Path, |
| 362 | 377 | out_path: Path, asset_type: str) -> int: |
| 363 | 378 | """Convert one extracted csv/txt file to zstd Parquet. Returns row count.""" |
| 364 | − ncols = _sniff_column_count(csv_path) | |
| 379 | + first = _sniff_first_line(csv_path) | |
| 380 | + ncols = len(first) | |
| 365 | 381 | if ncols == 0: |
| 366 | 382 | return 0 |
| 383 | + split_dt = ncols == 7 and len(first) > 1 and _TIME_RE.match(first[1]) | |
| 367 | 384 | out_path.parent.mkdir(parents=True, exist_ok=True) |
| 368 | 385 | ticker = _ticker_from_filename(csv_path) |
| 369 | 386 | src = str(csv_path).replace("'", "''") |
@@ -373,7 +390,12 @@ def csv_to_parquet(con: duckdb.DuckDBPyConnection, csv_path: Path, | ||
| 373 | 390 | # rows are appended with CRLF onto LF history), which the strict CSV |
| 374 | 391 | # sniffer rejects outright. |
| 375 | 392 | common = "strict_mode=false, ignore_errors=true" |
| 376 | − if asset_type == "options" and ncols == len(OPTIONS_SCHEMA): | |
| 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): | |
| 377 | 399 | cols = json.dumps(OPTIONS_SCHEMA).replace('"', "'") |
| 378 | 400 | select = (f"SELECT '{ticker}' AS ticker, * FROM read_csv('{src}', " |
| 379 | 401 | f"header=false, columns={cols}, dateformat='%Y-%m-%d', " |
@@ -435,10 +457,15 @@ def _extract_recursive(zip_path: Path, dest: Path, depth: int = 0) -> None: | ||
| 435 | 457 | with zipfile.ZipFile(zip_path) as zf: |
| 436 | 458 | zf.extractall(dest) |
| 437 | 459 | except (NotImplementedError, RuntimeError, zipfile.BadZipFile): |
| 438 | − # Some FirstRate archives use Deflate64, which the stdlib zipfile | |
| 439 | − # cannot decompress — bsdtar (libarchive) handles it. | |
| 440 | − subprocess.run(["tar", "-xf", str(zip_path), "-C", str(dest)], | |
| 441 | − check=True, capture_output=True) | |
| 460 | + # Some FirstRate archives use Deflate64, which neither the stdlib | |
| 461 | + # 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 | + break | |
| 467 | + else: | |
| 468 | + raise | |
| 442 | 469 | for nested in list(dest.rglob("*.zip")): |
| 443 | 470 | sub = nested.with_suffix("") |
| 444 | 471 | sub.mkdir(exist_ok=True) |
| 445 | 472 | |