spb/findata

Public

46commits
1branches
0releases
2.8 MBon disk
maindefault branch
21 h agolast push

options parser: ragged lines tolerated, malformed files skipped

Simon-Pierre Boucher committed yesterday (Nov 2, 2026) parent c41a8d8 Browse files

2 changed files +4 −3

modified connectors/market/src/findata_market_firstrate/archive.py +3 −2

		`@@ -201,10 +201,11 @@ def parse_options_file(data: bytes, underlying: str, source_id: str, ingested_at`
201	201	return empty, 0
202	202	names = ["trade_date", "strike", "expiry", "call_put", "last_price", "bid", "ask", "bid_iv", "ask_iv", "open_interest",
203	203	"volume", "delta", "gamma", "vega", "theta", "rho"]
204		− ncols = len(lines[0].split(","))
	204	+ from collections import Counter
	205	+ ncols = Counter(len(ln.split(",")) for ln in lines[:500]).most_common(1)[0][0]
205	206	if ncols != 16:
206	207	raise ValueError(f"unexpected option-chain column count {ncols} for {underlying}")
207		− df = pl.read_csv("\n".join(lines).encode(), has_header=False, new_columns=names, infer_schema_length=0)
	208	+ df = pl.read_csv("\n".join(lines).encode(), has_header=False, new_columns=names, infer_schema_length=0, truncate_ragged_lines=True)
208	209	num = [c for c in names if c not in ("trade_date", "expiry", "call_put")]
209	210	df = df.with_columns([pl.col(c).str.strip_chars().cast(pl.Float64, strict=False) for c in num])
210	211	df = df.with_columns(

modified services/ingest/src/findata_ingest/tasks/options.py +1 −1

		`@@ -72,7 +72,7 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:`
72	72	continue
73	73	try:
74	74	table, raw_rows = parse_options_file(data, und, "firstrate", ingested_at)
75		− except ValueError as e:
	75	+ except Exception as e:
76	76	stats["skipped_files"] += 1
77	77	log.warning("options.skip_file", file=name, error=str(e)[:120])
78	78	continue
79	79