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 = 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 adj_vendor = p.get("adjustment") or ("contin_UNadj" if asset == AssetClass.futures and not p.get("contracts") else "UNADJUSTED")106 adjustment_value = ADJ_MAP.get(adj_vendor, "raw")107 contracts = bool(p.get("contracts"))108 ingested_at = datetime.now(UTC)109 only: set[str] | None = set(kwargs.get("symbols") or []) or None110 tz = SOURCE_TZ.get(asset.value, "America/New_York")111 for name, data in iter_archive_files(result.path):112 base = Path(name).name.lower()113 if base.startswith("_") or "readme" in base or base.endswith((".csv", ".txt")) is False:114 continue115 symbol = ticker_from_filename(name, contract=contracts)116 if only and symbol not in only:117 continue118 table, raw_rows = parse_bar_file(data, symbol, asset, interval, _Adj(adjustment_value), SOURCE_ID, ingested_at, source_tz=tz)119 if table.num_rows == 0:120 continue121 sd = table.column("session_date").to_pylist()122 yield NormalizedBatch(table=table, symbol=symbol, session_dates=(min(sd), max(sd)),123 row_count=table.num_rows, raw_rows=raw_rows)124125 # ------------------------------------------------------------------ validate126 def validate(self, batch: NormalizedBatch) -> ValidationReport:127 interval = Interval(batch.table.column("interval")[0].as_py())128 asset = batch.table.column("asset_class")[0].as_py()129 table, anomalies = validate_bars(batch.table, interval, asset_class=asset)130 batch.table = table131 flagged = sum(1 for f in table.column("quality_flags").to_pylist() if f)132 errors = [a for a in anomalies if a["severity"] == "error"]133 return ValidationReport(rows=table.num_rows, flagged_rows=flagged, anomalies=anomalies, ok=not errors)134135 # ------------------------------------------------------------------ checkpoint136 def checkpoint(self, result: FetchResult, batches_published: int) -> Checkpoint:137 p = result.request.params138 return Checkpoint(scope=f"dataset:{result.request.dataset_id}", key="last_archive",139 value={"sha256": result.sha256, "vendor_last_update": result.vendor_last_update.isoformat()140 if result.vendor_last_update else None, "params": p,141 "fetched_at": result.fetched_at.isoformat(), "batches": batches_published})142143 # ------------------------------------------------------------------ reference data144 def ticker_listing(self, type_: str) -> list[dict[str, Any]]:145 txt = self.client.ticker_listing(type_)146 rows = list(csv.DictReader(io.StringIO(txt)))147 out = []148 for r in rows:149 sym = (r.get("Ticker") or "").strip()150 if not sym:151 continue152 out.append({"symbol": sym, "name": (r.get("Name") or "").strip() or None,153 "first_date": _d(r.get("First Date")), "last_date": _d(r.get("Last Date")),154 "status": "delisted" if sym.endswith("-DELISTED") else "active", "asset_class": type_})155 return out156157 def delisted_log(self) -> list[dict[str, Any]]:158 txt = self.client.update_log("stock", "delisted")159 return [{"symbol": r["Ticker"].strip(), "last_date": _d(r.get("Last Date"))}160 for r in csv.DictReader(io.StringIO(txt)) if r.get("Ticker")]161162 def meta_actions(self, type_: str, kind: str, dest_dir: Path) -> tuple[FetchResult, list[dict[str, Any]]]:163 """kind = splits | dividends → list of {symbol, ex_date, value}."""164 req = FetchRequest(dataset_id=f"corporate_actions.{type_}", params={"endpoint": "meta_file", "type": type_, "metafile_type": kind})165 res = self.fetch_range(req, dest_dir)166 rows = []167 for name, data in iter_archive_files(res.path):168 if name.startswith("_") or "readme" in name.lower():169 continue170 sym = Path(name).stem.upper()171 for d, v in parse_meta_pairs(data):172 rows.append({"symbol": sym, "ex_date": d, "value": v, "action_type": "split" if kind == "splits" else "dividend"})173 return res, rows174175176class _Adj(str):177 """Adjustment label passthrough (dataset adjustment ids are richer than the Adjustment enum)."""178179 @property180 def value(self) -> str: # mimics Enum.value181 return str(self)182183184def _d(s: str | None) -> date | None:185 if not s:186 return None187 try:188 return date.fromisoformat(s.strip()[:10])189 except ValueError:190 return None191192193def sanitized_request(params: dict[str, Any]) -> str:194 return json.dumps({k: v for k, v in params.items() if k != "userid"}, sort_keys=True)195196197def digest_params(params: dict[str, Any]) -> str:198 return hashlib.sha256(sanitized_request(params).encode()).hexdigest()[:16]199