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 pa1819from findata_schema import BARS_SCHEMA, Adjustment, AssetClass, Interval2021VENDOR_TF = {"1min": Interval.m1, "5min": Interval.m5, "30min": Interval.m30, "1hour": Interval.h1, "1day": Interval.d1}22VENDOR_ADJ = {"UNADJUSTED": Adjustment.raw, "adj_split": Adjustment.split, "adj_splitdiv": Adjustment.split_div}23_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.I)242526def ticker_from_filename(name: str) -> str:27 base = Path(name).name28 m = _TICKER_RE.match(base)29 if m:30 return m.group(1).upper()31 return base.split("_")[0].upper()323334def iter_archive_files(zip_path: Path) -> Iterator[tuple[str, bytes]]:35 """Yield (member_name, bytes) for every data file, descending into nested zips."""36 with zipfile.ZipFile(zip_path) as zf:37 for info in zf.infolist():38 if info.is_dir():39 continue40 name = info.filename41 if name.lower().endswith(".zip"):42 data = zf.read(info)43 with zipfile.ZipFile(io.BytesIO(data)) as inner:44 for inner_info in inner.infolist():45 if not inner_info.is_dir() and inner_info.filename.lower().endswith((".txt", ".csv")):46 yield f"{name}!/{inner_info.filename}", inner.read(inner_info)47 elif name.lower().endswith((".txt", ".csv")):48 yield name, zf.read(info)495051def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval: Interval, adjustment: Adjustment,52 source_id: str, ingested_at: datetime, source_tz: str = "America/New_York",53 currency: str = "USD", venue: str | None = None, revision_id: int = 1) -> tuple[pa.Table, int]:54 """Parse one vendor CSV into the canonical bars table. Returns (table, raw_row_count).5556 Duplicate (symbol, timestamp) rows are *kept* here and flagged downstream; we never drop source records silently.57 Unparseable timestamps (e.g. non-existent local time at DST spring-forward) are dropped from the typed table and58 counted in `raw_row_count - table.num_rows`.59 """60 text = data.decode("utf-8", errors="replace").replace("\r\n", "\n").replace("\r", "\n")61 lines = [ln for ln in text.split("\n") if ln.strip()]62 if lines and lines[0].lower().startswith(("datetime", "date", "timestamp")):63 lines = lines[1:]64 raw_rows = len(lines)65 if raw_rows == 0:66 return pa.table({f.name: pa.array([], type=f.type) for f in BARS_SCHEMA}, schema=BARS_SCHEMA), 067 ncols = len(lines[0].split(","))68 buf = "\n".join(lines).encode()69 if ncols == 7 and re.match(r"^\d{8},\d{2}:\d{2}", lines[0]):70 names = ["d", "t", "open", "high", "low", "close", "volume"]71 df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0)72 df = df.with_columns((pl.col("d") + " " + pl.col("t")).alias("dt_raw")).drop("d", "t")73 fmt = "%Y%m%d %H:%M:%S"74 else:75 names = ["dt_raw", "open", "high", "low", "close", "volume"] + (["open_interest"] if ncols == 7 else [])76 if ncols not in (6, 7):77 raise ValueError(f"unexpected column count {ncols} in {symbol}")78 df = pl.read_csv(buf, has_header=False, new_columns=names, infer_schema_length=0)79 fmt = "%Y-%m-%d %H:%M:%S" if interval != Interval.d1 else None80 for c in ("open", "high", "low", "close", "volume"):81 df = df.with_columns(pl.col(c).str.strip_chars().cast(pl.Float64, strict=False))82 if interval == Interval.d1:83 local = pl.col("dt_raw").str.strip_chars().str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False).cast(pl.Datetime("us"))84 else:85 local = pl.col("dt_raw").str.strip_chars().str.to_datetime(fmt, strict=False, time_unit="us")86 df = df.with_columns(local.alias("local_naive"))87 df = df.filter(pl.col("local_naive").is_not_null() & pl.col("open").is_not_null() & pl.col("close").is_not_null())88 df = df.with_columns(89 pl.col("local_naive").dt.replace_time_zone(source_tz, ambiguous="earliest", non_existent="null").alias("ts_local")90 ).filter(pl.col("ts_local").is_not_null())91 df = df.with_columns(92 pl.col("ts_local").dt.convert_time_zone("UTC").alias("timestamp_utc"),93 pl.col("ts_local").dt.date().alias("session_date"),94 )95 n = df.height96 out = pl.DataFrame({97 "instrument_id": pl.repeat(f"{source_id}:{asset_class.value}:{symbol}", n, eager=True),98 "symbol_as_reported": pl.repeat(symbol, n, eager=True),99 "venue": pl.repeat(venue, n, eager=True, dtype=pl.Utf8),100 "asset_class": pl.repeat(asset_class.value, n, eager=True),101 "interval": pl.repeat(interval.value, n, eager=True),102 "timestamp_utc": df["timestamp_utc"],103 "session_date": df["session_date"],104 "open": df["open"], "high": df["high"], "low": df["low"], "close": df["close"],105 "volume": df["volume"].fill_null(0.0),106 "currency": pl.repeat(currency, n, eager=True),107 "trade_count": pl.repeat(None, n, eager=True, dtype=pl.Int32),108 "vwap": pl.repeat(None, n, eager=True, dtype=pl.Float64),109 "adjustment_type": pl.repeat(adjustment.value, n, eager=True),110 "source_id": pl.repeat(source_id, n, eager=True),111 "ingested_at": pl.repeat(ingested_at, n, eager=True, dtype=pl.Datetime("us", "UTC")),112 "quality_flags": pl.repeat([], n, eager=True, dtype=pl.List(pl.Utf8)),113 "revision_id": pl.repeat(revision_id, n, eager=True, dtype=pl.Int32),114 })115 return out.to_arrow().cast(BARS_SCHEMA), raw_rows116117118def parse_meta_pairs(data: bytes) -> list[tuple[str, float]]:119 """Splits/dividends per-ticker files: `yyyy-MM-dd,value` lines."""120 out = []121 for ln in data.decode("utf-8", errors="replace").splitlines():122 ln = ln.strip()123 if not ln or ln.lower().startswith("date"):124 continue125 parts = ln.split(",")126 if len(parts) >= 2:127 try:128 out.append((parts[0].strip(), float(parts[1])))129 except ValueError:130 continue131 return out132