1from __future__ import annotations23import csv4import hashlib5import io6import json7from collections.abc import Iterator8from datetime import UTC, date, datetime9from pathlib import Path10from typing import Any1112from findata_core.config import get_settings13from findata_quality import validate_bars14from findata_schema import AssetClass, Interval15from findata_schema.connector import (16 Capabilities,17 Checkpoint,18 Connector,19 FetchRequest,20 FetchResult,21 NormalizedBatch,22 ValidationReport,23)2425from findata_market_firstrate.archive import (26 VENDOR_ADJ,27 VENDOR_TF,28 iter_archive_files,29 parse_bar_file,30 parse_meta_pairs,31 ticker_from_filename,32)33from findata_market_firstrate.client import FRDClient3435SOURCE_ID = "firstrate"36SOURCE_TZ = {"crypto": "UTC"} # everything else is US Eastern373839class FirstRateMarketConnector(Connector):40 source_id = SOURCE_ID4142 def __init__(self, client: FRDClient | None = None):43 self._client = client4445 @property46 def client(self) -> FRDClient:47 if self._client is None:48 self._client = FRDClient(get_settings().frd_userid.get_secret_value())49 return self._client5051 # ------------------------------------------------------------------ discover / capabilities52 def discover(self) -> dict[str, Any]:53 out: dict[str, Any] = {"source_id": SOURCE_ID, "probed_at": datetime.now(UTC).isoformat(), "last_update": {}}54 for t in ("stock", "etf", "futures", "index", "fx", "crypto", "options"):55 try:56 out["last_update"][t] = self.client.last_update(t).isoformat()57 except Exception as e:58 out["last_update"][t] = f"error: {e}"59 return out6061 def capabilities(self) -> Capabilities:62 return Capabilities(63 source_id=SOURCE_ID,64 asset_classes=tuple(a.value for a in AssetClass),65 intervals=("1m", "5m", "30m", "1h", "1d"),66 adjustments=("raw", "split", "split_div", "continuous"),67 supports_symbol_filter=False,68 supports_date_range=False,69 incremental_windows=("day", "week", "month"),70 license_class="frd-private",71 notes={"raw_only_for": ["1m", "1d"], "zero_volume_bars": "absent at source",72 "timestamp_convention": "bar_open", "timezone": "America/New_York (crypto: UTC)"},73 )7475 # ------------------------------------------------------------------ fetch76 def fetch_range(self, request: FetchRequest, dest_dir: Path) -> FetchResult:77 p = dict(request.params)78 endpoint = p.pop("endpoint", "data_file")79 asset = p.get("type", "stock")80 vendor_last = None81 try:82 vendor_last = self.client.last_update(asset, full=(p.get("period") == "full"))83 except Exception:84 vendor_last = None85 stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")86 label = "_".join(f"{k}-{v}" for k, v in sorted(p.items()))87 dest = dest_dir / f"{endpoint}__{label}__{stamp}.zip"88 got = self.client.download(endpoint, p, dest)89 file_count = None90 try:91 import zipfile92 with zipfile.ZipFile(got.path) as zf:93 file_count = sum(1 for i in zf.infolist() if not i.is_dir())94 except Exception:95 pass96 return FetchResult(request=request, path=got.path, sha256=got.sha256, bytes=got.bytes, fetched_at=got.fetched_at,97 vendor_last_update=vendor_last, file_count=file_count,98 source_url_sanitized=self.client.sanitized_url(endpoint, p))99100 # ------------------------------------------------------------------ normalize101 def normalize(self, result: FetchResult, **kwargs: Any) -> Iterator[NormalizedBatch]:102 p = result.request.params103 asset = AssetClass(p.get("type", "stock"))104 interval = VENDOR_TF[p.get("timeframe", "1min")]105 adjustment = VENDOR_ADJ.get(p.get("adjustment", "UNADJUSTED"))106 if adjustment is None:107 from findata_schema import Adjustment108 adjustment = Adjustment.continuous109 ingested_at = datetime.now(UTC)110 only: set[str] | None = set(kwargs.get("symbols") or []) or None111 tz = SOURCE_TZ.get(asset.value, "America/New_York")112 for name, data in iter_archive_files(result.path):113 symbol = ticker_from_filename(name)114 if only and symbol not in only:115 continue116 table, raw_rows = parse_bar_file(data, symbol, asset, interval, adjustment, SOURCE_ID, ingested_at, source_tz=tz)117 if table.num_rows == 0:118 continue119 sd = table.column("session_date").to_pylist()120 yield NormalizedBatch(table=table, symbol=symbol, session_dates=(min(sd), max(sd)),121 row_count=table.num_rows, raw_rows=raw_rows)122123 # ------------------------------------------------------------------ validate124 def validate(self, batch: NormalizedBatch) -> ValidationReport:125 interval = Interval(batch.table.column("interval")[0].as_py())126 table, anomalies = validate_bars(batch.table, interval)127 batch.table = table128 flagged = sum(1 for f in table.column("quality_flags").to_pylist() if f)129 errors = [a for a in anomalies if a["severity"] == "error"]130 return ValidationReport(rows=table.num_rows, flagged_rows=flagged, anomalies=anomalies, ok=not errors)131132 # ------------------------------------------------------------------ checkpoint133 def checkpoint(self, result: FetchResult, batches_published: int) -> Checkpoint:134 p = result.request.params135 return Checkpoint(scope=f"dataset:{result.request.dataset_id}", key="last_archive",136 value={"sha256": result.sha256, "vendor_last_update": result.vendor_last_update.isoformat()137 if result.vendor_last_update else None, "params": p,138 "fetched_at": result.fetched_at.isoformat(), "batches": batches_published})139140 # ------------------------------------------------------------------ reference data141 def ticker_listing(self, type_: str) -> list[dict[str, Any]]:142 txt = self.client.ticker_listing(type_)143 rows = list(csv.DictReader(io.StringIO(txt)))144 out = []145 for r in rows:146 sym = (r.get("Ticker") or "").strip()147 if not sym:148 continue149 out.append({"symbol": sym, "name": (r.get("Name") or "").strip() or None,150 "first_date": _d(r.get("First Date")), "last_date": _d(r.get("Last Date")),151 "status": "delisted" if sym.endswith("-DELISTED") else "active", "asset_class": type_})152 return out153154 def delisted_log(self) -> list[dict[str, Any]]:155 txt = self.client.update_log("stock", "delisted")156 return [{"symbol": r["Ticker"].strip(), "last_date": _d(r.get("Last Date"))}157 for r in csv.DictReader(io.StringIO(txt)) if r.get("Ticker")]158159 def meta_actions(self, type_: str, kind: str, dest_dir: Path) -> tuple[FetchResult, list[dict[str, Any]]]:160 """kind = splits | dividends → list of {symbol, ex_date, value}."""161 req = FetchRequest(dataset_id=f"corporate_actions.{type_}", params={"endpoint": "meta_file", "type": type_, "metafile_type": kind})162 res = self.fetch_range(req, dest_dir)163 rows = []164 for name, data in iter_archive_files(res.path):165 if name.startswith("_") or "readme" in name.lower():166 continue167 sym = Path(name).stem.upper()168 for d, v in parse_meta_pairs(data):169 rows.append({"symbol": sym, "ex_date": d, "value": v, "action_type": "split" if kind == "splits" else "dividend"})170 return res, rows171172173def _d(s: str | None) -> date | None:174 if not s:175 return None176 try:177 return date.fromisoformat(s.strip()[:10])178 except ValueError:179 return None180181182def sanitized_request(params: dict[str, Any]) -> str:183 return json.dumps({k: v for k, v in params.items() if k != "userid"}, sort_keys=True)184185186def digest_params(params: dict[str, Any]) -> str:187 return hashlib.sha256(sanitized_request(params).encode()).hexdigest()[:16]188