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_TF,27 iter_archive_files,28 parse_bar_file,29 parse_meta_pairs,30 ticker_from_filename,31)32from findata_market_firstrate.catalog import ADJ_MAP33from 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 = client44 self.skipped_files: list[dict[str, str]] = []4546 @property47 def client(self) -> FRDClient:48 if self._client is None:49 self._client = FRDClient(get_settings().frd_userid.get_secret_value())50 return self._client5152 # ------------------------------------------------------------------ discover / capabilities53 def discover(self) -> dict[str, Any]:54 out: dict[str, Any] = {"source_id": SOURCE_ID, "probed_at": datetime.now(UTC).isoformat(), "last_update": {}}55 for t in ("stock", "etf", "futures", "index", "fx", "crypto", "options"):56 try:57 out["last_update"][t] = self.client.last_update(t).isoformat()58 except Exception as e:59 out["last_update"][t] = f"error: {e}"60 return out6162 def capabilities(self) -> Capabilities:63 return Capabilities(64 source_id=SOURCE_ID,65 asset_classes=tuple(a.value for a in AssetClass),66 intervals=("1m", "5m", "30m", "1h", "1d"),67 adjustments=("raw", "split", "split_div", "continuous"),68 supports_symbol_filter=False,69 supports_date_range=False,70 incremental_windows=("day", "week", "month"),71 license_class="frd-private",72 notes={"raw_only_for": ["1m", "1d"], "zero_volume_bars": "absent at source",73 "timestamp_convention": "bar_open", "timezone": "America/New_York (crypto: UTC)"},74 )7576 # ------------------------------------------------------------------ fetch77 def fetch_range(self, request: FetchRequest, dest_dir: Path) -> FetchResult:78 p = dict(request.params)79 endpoint = p.pop("endpoint", "data_file")80 asset = p.get("type", "stock")81 vendor_last = None82 try:83 vendor_last = self.client.last_update(asset, full=(p.get("period") == "full"))84 except Exception:85 vendor_last = None86 stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")87 label = "_".join(f"{k}-{v}" for k, v in sorted(p.items()))88 dest = dest_dir / f"{endpoint}__{label}__{stamp}.zip"89 got = self.client.download(endpoint, p, dest)90 file_count = None91 try:92 import zipfile93 with zipfile.ZipFile(got.path) as zf:94 file_count = sum(1 for i in zf.infolist() if not i.is_dir())95 except Exception:96 pass97 return FetchResult(request=request, path=got.path, sha256=got.sha256, bytes=got.bytes, fetched_at=got.fetched_at,98 vendor_last_update=vendor_last, file_count=file_count,99 source_url_sanitized=self.client.sanitized_url(endpoint, p))100101 # ------------------------------------------------------------------ normalize102 def normalize(self, result: FetchResult, **kwargs: Any) -> Iterator[NormalizedBatch]:103 p = result.request.params104 asset = AssetClass(p.get("type", "stock"))105 interval = VENDOR_TF[p.get("timeframe", "1min")]106 adj_vendor = p.get("adjustment") or ("contin_UNadj" if asset == AssetClass.futures and not p.get("contracts") else "UNADJUSTED")107 adjustment_value = ADJ_MAP.get(adj_vendor, "raw")108 contracts = bool(p.get("contracts"))109 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 base = Path(name).name.lower()114 if base.startswith("_") or "readme" in base or base.endswith((".csv", ".txt")) is False:115 continue116 symbol = ticker_from_filename(name, contract=contracts)117 if only and symbol not in only:118 continue119 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 continue124 if table.num_rows == 0:125 continue126 sd = table.column("session_date").to_pylist()127 yield NormalizedBatch(table=table, symbol=symbol, session_dates=(min(sd), max(sd)),128 row_count=table.num_rows, raw_rows=raw_rows)129130 # ------------------------------------------------------------------ validate131 def validate(self, batch: NormalizedBatch) -> ValidationReport:132 interval = Interval(batch.table.column("interval")[0].as_py())133 asset = batch.table.column("asset_class")[0].as_py()134 table, anomalies = validate_bars(batch.table, interval, asset_class=asset)135 batch.table = table136 flagged = sum(1 for f in table.column("quality_flags").to_pylist() if f)137 errors = [a for a in anomalies if a["severity"] == "error"]138 return ValidationReport(rows=table.num_rows, flagged_rows=flagged, anomalies=anomalies, ok=not errors)139140 # ------------------------------------------------------------------ checkpoint141 def checkpoint(self, result: FetchResult, batches_published: int) -> Checkpoint:142 p = result.request.params143 return Checkpoint(scope=f"dataset:{result.request.dataset_id}", key="last_archive",144 value={"sha256": result.sha256, "vendor_last_update": result.vendor_last_update.isoformat()145 if result.vendor_last_update else None, "params": p,146 "fetched_at": result.fetched_at.isoformat(), "batches": batches_published})147148 # ------------------------------------------------------------------ reference data149 def ticker_listing(self, type_: str) -> list[dict[str, Any]]:150 txt = self.client.ticker_listing(type_)151 rows = list(csv.DictReader(io.StringIO(txt)))152 out = []153 for r in rows:154 sym = (r.get("Ticker") or "").strip()155 if not sym:156 continue157 out.append({"symbol": sym, "name": (r.get("Name") or "").strip() or None,158 "first_date": _d(r.get("First Date")), "last_date": _d(r.get("Last Date")),159 "status": "delisted" if sym.endswith("-DELISTED") else "active", "asset_class": type_})160 return out161162 def delisted_log(self) -> list[dict[str, Any]]:163 txt = self.client.update_log("stock", "delisted")164 return [{"symbol": r["Ticker"].strip(), "last_date": _d(r.get("Last Date"))}165 for r in csv.DictReader(io.StringIO(txt)) if r.get("Ticker")]166167 def meta_actions(self, type_: str, kind: str, dest_dir: Path) -> tuple[FetchResult, list[dict[str, Any]]]:168 """kind = splits | dividends → list of {symbol, ex_date, value}."""169 req = FetchRequest(dataset_id=f"corporate_actions.{type_}", params={"endpoint": "meta_file", "type": type_, "metafile_type": kind})170 res = self.fetch_range(req, dest_dir)171 rows = []172 for name, data in iter_archive_files(res.path):173 if name.startswith("_") or "readme" in name.lower():174 continue175 sym = Path(name).stem.upper()176 for d, v in parse_meta_pairs(data):177 rows.append({"symbol": sym, "ex_date": d, "value": v, "action_type": "split" if kind == "splits" else "dividend"})178 return res, rows179180181class _Adj(str):182 """Adjustment label passthrough (dataset adjustment ids are richer than the Adjustment enum)."""183184 @property185 def value(self) -> str: # mimics Enum.value186 return str(self)187188189def _d(s: str | None) -> date | None:190 if not s:191 return None192 try:193 return date.fromisoformat(s.strip()[:10])194 except ValueError:195 return None196197198def sanitized_request(params: dict[str, Any]) -> str:199 return json.dumps({k: v for k, v in params.items() if k != "userid"}, sort_keys=True)200201202def digest_params(params: dict[str, Any]) -> str:203 return hashlib.sha256(sanitized_request(params).encode()).hexdigest()[:16]204