1"""Zip archive handling and CSV → canonical Arrow conversion for FirstRate bar files.23Vendor file facts: `{TICKER}_{period}_{timeframe}_{adjustment}.txt` (update archives) or `{TICKER}_{timeframe}.txt`4(full archives), no header, `yyyy-MM-dd HH:mm:ss,O,H,L,C,V` (daily: `yyyy-MM-dd,O,H,L,C,V[,OI]`), FX intraday splits5date and time into two fields, line endings may mix LF/CRLF, nested zips occur in bundle archives.6"""7from __future__ import annotations89import io10import re11import zipfile12from collections.abc import Iterator13from datetime import datetime14from pathlib import Path1516import polars as pl17import pyarrow as pa18from findata_schema import BARS_SCHEMA, Adjustment, AssetClass, Interval1920VENDOR_TF = {"1min": Interval.m1, "5min": Interval.m5, "30min": Interval.m30, "1hour": Interval.h1, "1day": Interval.d1}21VENDOR_ADJ = {"UNADJUSTED": Adjustment.raw, "adj_split": Adjustment.split, "adj_splitdiv": Adjustment.split_div}22_TICKER_RE = re.compile(r"^([A-Za-z0-9.\-]+?)(?:[_-](?:full|day|week|month|1min|5min|30min|1hour|1day|UNADJUSTED|adj_split|adj_splitdiv|contin.*))*\.(txt|csv)$", re.IGNORECASE)232425_TF_SUFFIX_RE = re.compile(r"_(1min|5min|30min|1hour|1day)$", re.I)262728def ticker_from_filename(name: str, contract: bool = False) -> str:29 base = Path(name).name30 if contract: # individual futures contracts: 'A6_F25_1day.txt' -> 'A6_F25'31 stem = Path(base).stem32 return _TF_SUFFIX_RE.sub("", stem).upper()33 m = _TICKER_RE.match(base)34 if m:35 return m.group(1).upper()36 return base.split("_")[0].upper()373839def iter_archive_files(zip_path: Path) -> Iterator[tuple[str, bytes]]:40 """Yield (member_name, bytes) for every data file, descending into nested zips."""41 with zipfile.ZipFile(zip_path) as zf:42 for info in zf.infolist():43 if info.is_dir():44 continue45 name = info.filename46 if name.lower().endswith(".zip"):47 data = zf.read(info)48 with zipfile.ZipFile(io.BytesIO(data)) as inner:49 for inner_info in inner.infolist():50 if not inner_info.is_dir() and inner_info.filename.lower().endswith((".txt", ".csv")):51 yield f"{name}!/{inner_info.filename}", inner.read(inner_info)52 elif name.lower().endswith((".txt", ".csv")):53 yield name, zf.read(info)545556def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval: Interval, adjustment: Adjustment,57 source_id: str, ingested_at: datetime, source_tz: str = "America/New_York",58 currency: str = "USD", venue: str | None = None, revision_id: int = 1) -> tuple[pa.Table, int]:59 """Parse one vendor CSV into the canonical bars table. Returns (table, raw_row_count).6061 Duplicate (symbol, timestamp) rows are *kept* here and flagged downstream; we never drop source records silently.62 Unparseable timestamps (e.g. non-existent local time at DST spring-forward) are dropped from the typed table and63 counted in `raw_row_count - table.num_rows`.64 """65 text = data.decode("utf-8", errors="replace").replace("\r\n", "\n").replace("\r", "\n")66 lines = [ln for ln in text.split("\n") if ln.strip()]67 if lines and lines[0].lower().startswith(("datetime", "date", "timestamp")):68 lines = lines[1:]69 raw_rows = len(lines)70 if raw_rows == 0:71 return pa.table({f.name: pa.array([], type=f.type) for f in BARS_SCHEMA}, schema=BARS_SCHEMA), 072 ncols = len(lines[0].split(","))73 buf = "\n".join(lines).encode()74 if ncols == 7 and re.match(r"^\d{8},\d{2}:\d{2}", lines[0]):75 names = ["d", "t", "open", "high", "low", "close", "volume"]76 df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0)77 df = df.with_columns((pl.col("d") + " " + pl.col("t")).alias("dt_raw")).drop("d", "t")78 fmt = "%Y%m%d %H:%M:%S"79 else:80 if ncols == 5: # indices: no volume column81 names = ["dt_raw", "open", "high", "low", "close"]82 else:83 names = ["dt_raw", "open", "high", "low", "close", "volume"] + (["open_interest"] if ncols == 7 else [])84 if ncols not in (5, 6, 7):85 raise ValueError(f"unexpected column count {ncols} in {symbol}")86 df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0)87 if ncols == 5:88 df = df.with_columns(pl.lit("0").alias("volume"))89 fmt = "%Y-%m-%d %H:%M:%S" if interval != Interval.d1 else None90 for c in ("open", "high", "low", "close", "volume"):91 df = df.with_columns(pl.col(c).str.strip_chars().cast(pl.Float64, strict=False))92 if interval == Interval.d1:93 local = pl.col("dt_raw").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False).cast(pl.Datetime("us"))94 else:95 local = pl.coalesce(96 pl.col("dt_raw").str.strip_chars().str.to_datetime(fmt, strict=False, time_unit="us"),97 pl.col("dt_raw").str.strip_chars().str.to_datetime("%Y-%m-%d %H:%M", strict=False, time_unit="us"),98 )99 df = df.with_columns(local.alias("local_naive"))100 df = df.filter(pl.col("local_naive").is_not_null() & pl.col("open").is_not_null() & pl.col("close").is_not_null())101 df = df.with_columns(102 pl.col("local_naive").dt.replace_time_zone(source_tz, ambiguous="earliest", non_existent="null").alias("ts_local")103 ).filter(pl.col("ts_local").is_not_null())104 df = df.with_columns(105 pl.col("ts_local").dt.convert_time_zone("UTC").alias("timestamp_utc"),106 pl.col("ts_local").dt.date().alias("session_date"),107 )108 n = df.height109 out = pl.DataFrame({110 "instrument_id": pl.repeat(f"{source_id}:{asset_class.value}:{symbol}", n, eager=True),111 "symbol_as_reported": pl.repeat(symbol, n, eager=True),112 "venue": pl.repeat(venue, n, eager=True, dtype=pl.Utf8),113 "asset_class": pl.repeat(asset_class.value, n, eager=True),114 "interval": pl.repeat(interval.value, n, eager=True),115 "timestamp_utc": df["timestamp_utc"],116 "session_date": df["session_date"],117 "open": df["open"], "high": df["high"], "low": df["low"], "close": df["close"],118 "volume": df["volume"].fill_null(0.0),119 "currency": pl.repeat(currency, n, eager=True),120 "trade_count": pl.repeat(None, n, eager=True, dtype=pl.Int32),121 "vwap": pl.repeat(None, n, eager=True, dtype=pl.Float64),122 "adjustment_type": pl.repeat(adjustment.value, n, eager=True),123 "source_id": pl.repeat(source_id, n, eager=True),124 "ingested_at": pl.repeat(ingested_at, n, eager=True, dtype=pl.Datetime("us", "UTC")),125 "quality_flags": pl.repeat([], n, eager=True, dtype=pl.List(pl.Utf8)),126 "revision_id": pl.repeat(revision_id, n, eager=True, dtype=pl.Int32),127 })128 return out.to_arrow().cast(BARS_SCHEMA), raw_rows129130131def parse_meta_pairs(data: bytes) -> list[tuple[str, float]]:132 """Splits/dividends per-ticker files: `yyyy-MM-dd,value` lines."""133 out = []134 for ln in data.decode("utf-8", errors="replace").splitlines():135 ln = ln.strip()136 if not ln or ln.lower().startswith("date"):137 continue138 parts = ln.split(",")139 if len(parts) >= 2:140 try:141 out.append((parts[0].strip(), float(parts[1])))142 except ValueError:143 continue144 return out145146147def parse_options_file(data: bytes, underlying: str, source_id: str, ingested_at: datetime) -> tuple[pa.Table, int]:148 """Vendor option-chain CSV (16 columns, no header): Trade Date, Strike, Expiry, Call/Put, Last, Bid, Ask, Bid IV,149 Ask IV, Open Interest, Volume, Delta, Gamma, Vega, Theta, Rho → OPTIONS_SCHEMA."""150 from findata_schema import OPTIONS_SCHEMA151 text = data.decode("utf-8", errors="replace").replace("\r\n", "\n").replace("\r", "\n")152 lines = [ln for ln in text.split("\n") if ln.strip()]153 if lines and lines[0].lower().startswith(("trade", "date")):154 lines = lines[1:]155 raw_rows = len(lines)156 empty = pa.table({f.name: pa.array([], type=f.type) for f in OPTIONS_SCHEMA}, schema=OPTIONS_SCHEMA)157 if raw_rows == 0:158 return empty, 0159 names = ["trade_date", "strike", "expiry", "call_put", "last_price", "bid", "ask", "bid_iv", "ask_iv", "open_interest",160 "volume", "delta", "gamma", "vega", "theta", "rho"]161 ncols = len(lines[0].split(","))162 if ncols != 16:163 raise ValueError(f"unexpected option-chain column count {ncols} for {underlying}")164 df = pl.read_csv("\n".join(lines).encode(), has_header=False, new_columns=names, infer_schema_length=0)165 num = [c for c in names if c not in ("trade_date", "expiry", "call_put")]166 df = df.with_columns([pl.col(c).str.strip_chars().cast(pl.Float64, strict=False) for c in num])167 df = df.with_columns(168 pl.col("trade_date").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False),169 pl.col("expiry").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False),170 pl.col("call_put").str.strip_chars().str.to_lowercase(),171 ).filter(pl.col("trade_date").is_not_null() & pl.col("expiry").is_not_null() & pl.col("strike").is_not_null())172 n = df.height173 out = df.with_columns(174 pl.lit(underlying).alias("underlying"), pl.lit(source_id).alias("source_id"),175 pl.lit(ingested_at).cast(pl.Datetime("us", "UTC")).alias("ingested_at"),176 ).select(OPTIONS_SCHEMA.names)177 return out.to_arrow().cast(OPTIONS_SCHEMA), raw_rows178