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_market_firstrate.archive import (VENDOR_ADJ, VENDOR_TF, iter_archive_files, parse_bar_file,14 parse_meta_pairs, ticker_from_filename)15from findata_market_firstrate.client import FRDClient16from findata_quality import validate_bars17from findata_schema import AssetClass, Interval18from findata_schema.connector import (Capabilities, Checkpoint, Connector, FetchRequest, FetchResult,19 NormalizedBatch, ValidationReport)2021SOURCE_ID = "firstrate"22SOURCE_TZ = {"crypto": "UTC"} # everything else is US Eastern232425class FirstRateMarketConnector(Connector):26 source_id = SOURCE_ID2728 def __init__(self, client: FRDClient | None = None):29 self._client = client3031 @property32 def client(self) -> FRDClient:33 if self._client is None:34 self._client = FRDClient(get_settings().frd_userid.get_secret_value())35 return self._client3637 # ------------------------------------------------------------------ discover / capabilities38 def discover(self) -> dict[str, Any]:39 out: dict[str, Any] = {"source_id": SOURCE_ID, "probed_at": datetime.now(UTC).isoformat(), "last_update": {}}40 for t in ("stock", "etf", "futures", "index", "fx", "crypto", "options"):41 try:42 out["last_update"][t] = self.client.last_update(t).isoformat()43 except Exception as e: # noqa: BLE00144 out["last_update"][t] = f"error: {e}"45 return out4647 def capabilities(self) -> Capabilities:48 return Capabilities(49 source_id=SOURCE_ID,50 asset_classes=tuple(a.value for a in AssetClass),51 intervals=("1m", "5m", "30m", "1h", "1d"),52 adjustments=("raw", "split", "split_div", "continuous"),53 supports_symbol_filter=False,54 supports_date_range=False,55 incremental_windows=("day", "week", "month"),56 license_class="frd-private",57 notes={"raw_only_for": ["1m", "1d"], "zero_volume_bars": "absent at source",58 "timestamp_convention": "bar_open", "timezone": "America/New_York (crypto: UTC)"},59 )6061 # ------------------------------------------------------------------ fetch62 def fetch_range(self, request: FetchRequest, dest_dir: Path) -> FetchResult:63 p = dict(request.params)64 endpoint = p.pop("endpoint", "data_file")65 asset = p.get("type", "stock")66 vendor_last = None67 try:68 vendor_last = self.client.last_update(asset, full=(p.get("period") == "full"))69 except Exception: # noqa: BLE00170 vendor_last = None71 stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")72 label = "_".join(f"{k}-{v}" for k, v in sorted(p.items()))73 dest = dest_dir / f"{endpoint}__{label}__{stamp}.zip"74 got = self.client.download(endpoint, p, dest)75 file_count = None76 try:77 import zipfile78 with zipfile.ZipFile(got.path) as zf:79 file_count = sum(1 for i in zf.infolist() if not i.is_dir())80 except Exception: # noqa: BLE00181 pass82 return FetchResult(request=request, path=got.path, sha256=got.sha256, bytes=got.bytes, fetched_at=got.fetched_at,83 vendor_last_update=vendor_last, file_count=file_count,84 source_url_sanitized=self.client.sanitized_url(endpoint, p))8586 # ------------------------------------------------------------------ normalize87 def normalize(self, result: FetchResult, **kwargs: Any) -> Iterator[NormalizedBatch]:88 p = result.request.params89 asset = AssetClass(p.get("type", "stock"))90 interval = VENDOR_TF[p.get("timeframe", "1min")]91 adjustment = VENDOR_ADJ.get(p.get("adjustment", "UNADJUSTED"))92 if adjustment is None:93 from findata_schema import Adjustment94 adjustment = Adjustment.continuous95 ingested_at = datetime.now(UTC)96 only: set[str] | None = set(kwargs.get("symbols") or []) or None97 tz = SOURCE_TZ.get(asset.value, "America/New_York")98 for name, data in iter_archive_files(result.path):99 symbol = ticker_from_filename(name)100 if only and symbol not in only:101 continue102 table, raw_rows = parse_bar_file(data, symbol, asset, interval, adjustment, SOURCE_ID, ingested_at, source_tz=tz)103 if table.num_rows == 0:104 continue105 sd = table.column("session_date").to_pylist()106 yield NormalizedBatch(table=table, symbol=symbol, session_dates=(min(sd), max(sd)),107 row_count=table.num_rows, raw_rows=raw_rows)108109 # ------------------------------------------------------------------ validate110 def validate(self, batch: NormalizedBatch) -> ValidationReport:111 interval = Interval(batch.table.column("interval")[0].as_py())112 table, anomalies = validate_bars(batch.table, interval)113 batch.table = table114 flagged = sum(1 for f in table.column("quality_flags").to_pylist() if f)115 errors = [a for a in anomalies if a["severity"] == "error"]116 return ValidationReport(rows=table.num_rows, flagged_rows=flagged, anomalies=anomalies, ok=not errors)117118 # ------------------------------------------------------------------ checkpoint119 def checkpoint(self, result: FetchResult, batches_published: int) -> Checkpoint:120 p = result.request.params121 return Checkpoint(scope=f"dataset:{result.request.dataset_id}", key="last_archive",122 value={"sha256": result.sha256, "vendor_last_update": result.vendor_last_update.isoformat()123 if result.vendor_last_update else None, "params": p,124 "fetched_at": result.fetched_at.isoformat(), "batches": batches_published})125126 # ------------------------------------------------------------------ reference data127 def ticker_listing(self, type_: str) -> list[dict[str, Any]]:128 txt = self.client.ticker_listing(type_)129 rows = list(csv.DictReader(io.StringIO(txt)))130 out = []131 for r in rows:132 sym = (r.get("Ticker") or "").strip()133 if not sym:134 continue135 out.append({"symbol": sym, "name": (r.get("Name") or "").strip() or None,136 "first_date": _d(r.get("First Date")), "last_date": _d(r.get("Last Date")),137 "status": "delisted" if sym.endswith("-DELISTED") else "active", "asset_class": type_})138 return out139140 def delisted_log(self) -> list[dict[str, Any]]:141 txt = self.client.update_log("stock", "delisted")142 return [{"symbol": r["Ticker"].strip(), "last_date": _d(r.get("Last Date"))}143 for r in csv.DictReader(io.StringIO(txt)) if r.get("Ticker")]144145 def meta_actions(self, type_: str, kind: str, dest_dir: Path) -> tuple[FetchResult, list[dict[str, Any]]]:146 """kind = splits | dividends → list of {symbol, ex_date, value}."""147 req = FetchRequest(dataset_id=f"corporate_actions.{type_}", params={"endpoint": "meta_file", "type": type_, "metafile_type": kind})148 res = self.fetch_range(req, dest_dir)149 rows = []150 for name, data in iter_archive_files(res.path):151 if name.startswith("_") or "readme" in name.lower():152 continue153 sym = Path(name).stem.upper()154 for d, v in parse_meta_pairs(data):155 rows.append({"symbol": sym, "ex_date": d, "value": v, "action_type": "split" if kind == "splits" else "dividend"})156 return res, rows157158159def _d(s: str | None) -> date | None:160 if not s:161 return None162 try:163 return date.fromisoformat(s.strip()[:10])164 except ValueError:165 return None166167168def sanitized_request(params: dict[str, Any]) -> str:169 return json.dumps({k: v for k, v in params.items() if k != "userid"}, sort_keys=True)170171172def digest_params(params: dict[str, Any]) -> str:173 return hashlib.sha256(sanitized_request(params).encode()).hexdigest()[:16]174