parser: tolerate ragged vendor lines; skip unparseable files instead of failing the archive
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 changed files +22 −5
modified
connectors/market/src/findata_market_firstrate/archive.py
+5 −3
@@ -110,11 +110,13 @@ def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval: | ||
| 110 | 110 | raw_rows = len(lines) |
| 111 | 111 | if raw_rows == 0: |
| 112 | 112 | return pa.table({f.name: pa.array([], type=f.type) for f in BARS_SCHEMA}, schema=BARS_SCHEMA), 0 |
| 113 | − ncols = len(lines[0].split(",")) | |
| 113 | + # field count = the most common count over a sample (vendor files occasionally contain ragged lines) | |
| 114 | + from collections import Counter | |
| 115 | + ncols = Counter(len(ln.split(",")) for ln in lines[:500]).most_common(1)[0][0] | |
| 114 | 116 | buf = "\n".join(lines).encode() |
| 115 | 117 | if ncols == 7 and re.match(r"^\d{8},\d{2}:\d{2}", lines[0]): |
| 116 | 118 | names = ["d", "t", "open", "high", "low", "close", "volume"] |
| 117 | − df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0) | |
| 119 | + df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0, truncate_ragged_lines=True) | |
| 118 | 120 | df = df.with_columns((pl.col("d") + " " + pl.col("t")).alias("dt_raw")).drop("d", "t") |
| 119 | 121 | fmt = "%Y%m%d %H:%M:%S" |
| 120 | 122 | else: |
@@ -124,7 +126,7 @@ def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval: | ||
| 124 | 126 | names = ["dt_raw", "open", "high", "low", "close", "volume"] + (["open_interest"] if ncols == 7 else []) |
| 125 | 127 | if ncols not in (5, 6, 7): |
| 126 | 128 | raise ValueError(f"unexpected column count {ncols} in {symbol}") |
| 127 | − df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0) | |
| 129 | + df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0, truncate_ragged_lines=True) | |
| 128 | 130 | if ncols == 5: |
| 129 | 131 | df = df.with_columns(pl.lit("0").alias("volume")) |
| 130 | 132 | fmt = "%Y-%m-%d %H:%M:%S" if interval != Interval.d1 else None |
modified
connectors/market/src/findata_market_firstrate/connector.py
+6 −1
@@ -41,6 +41,7 @@ class FirstRateMarketConnector(Connector): | ||
| 41 | 41 | |
| 42 | 42 | def __init__(self, client: FRDClient | None = None): |
| 43 | 43 | self._client = client |
| 44 | + self.skipped_files: list[dict[str, str]] = [] | |
| 44 | 45 | |
| 45 | 46 | @property |
| 46 | 47 | def client(self) -> FRDClient: |
@@ -115,7 +116,11 @@ class FirstRateMarketConnector(Connector): | ||
| 115 | 116 | symbol = ticker_from_filename(name, contract=contracts) |
| 116 | 117 | if only and symbol not in only: |
| 117 | 118 | continue |
| 118 | − table, raw_rows = parse_bar_file(data, symbol, asset, interval, _Adj(adjustment_value), SOURCE_ID, ingested_at, source_tz=tz) | |
| 119 | + try: | |
| 120 | + table, raw_rows = parse_bar_file(data, symbol, asset, interval, _Adj(adjustment_value), SOURCE_ID, ingested_at, source_tz=tz) | |
| 121 | + except Exception as e: | |
| 122 | + self.skipped_files.append({"file": name, "error": str(e)[:200]}) | |
| 123 | + continue | |
| 119 | 124 | if table.num_rows == 0: |
| 120 | 125 | continue |
| 121 | 126 | sd = table.column("session_date").to_pylist() |
modified
services/ingest/src/findata_ingest/tasks/backfill.py
+5 −1
@@ -137,5 +137,9 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | ||
| 137 | 137 | if interval == Interval.m1 and stats["rows"] and dataset_id in ("market_bars.stock.1m.raw", "market_bars.etf.1m.raw"): |
| 138 | 138 | queue.enqueue("derive", dataset_id.replace(".1m.", ".5m."), {"source_dataset": dataset_id, "target_interval": "5m", |
| 139 | 139 | "symbols": sorted(done)}, priority=45, dedupe_key=f"derive:{dataset_id}:5m:backfill:{job['id']}") |
| 140 | − log.info("backfill.done", **stats) | |
| 140 | + if conn_fr.skipped_files: | |
| 141 | + stats["files_skipped_unparseable"] = len(conn_fr.skipped_files) | |
| 142 | + stats["skipped_examples"] = conn_fr.skipped_files[:5] | |
| 143 | + log.warning("backfill.skipped_files", n=len(conn_fr.skipped_files), examples=conn_fr.skipped_files[:3]) | |
| 144 | + log.info("backfill.done", **{k: v for k, v in stats.items() if k != "skipped_examples"}) | |
| 141 | 145 | return stats |
modified
tests/test_parse_firstrate.py
+6 −0
@@ -68,3 +68,9 @@ def test_daily_layout_with_open_interest_column(): | ||
| 68 | 68 | def test_duplicates_are_kept_for_flagging_not_dropped(): |
| 69 | 69 | t, raw = _parse("2026-10-30 09:30:00,1,2,0.5,1.5,10\n2026-10-30 09:30:00,1,2,0.5,1.5,10\n") |
| 70 | 70 | assert raw == 2 and t.num_rows == 2 |
| 71 | + | |
| 72 | + | |
| 73 | +def test_ragged_lines_do_not_break_parsing(): | |
| 74 | + txt = "2026-10-30 09:30:00,1,2,0.5,1.5,10\n2026-10-30 09:31:00,1,2,0.5,1.5,10,EXTRA\n2026-10-30 09:32:00,1,2,0.5,1.5\n2026-10-30 09:33:00,1,2,0.5,1.5,10\n" | |
| 75 | + t, raw = _parse(txt) | |
| 76 | + assert raw == 4 and t.num_rows >= 3 # extra field truncated; short line yields nulls and is dropped or kept with volume 0 | |
| 71 | 77 | |