from __future__ import annotations import csv import hashlib import io import json from collections.abc import Iterator from datetime import UTC, date, datetime from pathlib import Path from typing import Any from findata_core.config import get_settings from findata_quality import validate_bars from findata_schema import AssetClass, Interval from findata_schema.connector import ( Capabilities, Checkpoint, Connector, FetchRequest, FetchResult, NormalizedBatch, ValidationReport, ) from findata_market_firstrate.archive import ( VENDOR_TF, iter_archive_files, parse_bar_file, parse_meta_pairs, ticker_from_filename, ) from findata_market_firstrate.catalog import ADJ_MAP from findata_market_firstrate.client import FRDClient SOURCE_ID = "firstrate" SOURCE_TZ = {"crypto": "UTC"} # everything else is US Eastern class FirstRateMarketConnector(Connector): source_id = SOURCE_ID def __init__(self, client: FRDClient | None = None): self._client = client @property def client(self) -> FRDClient: if self._client is None: self._client = FRDClient(get_settings().frd_userid.get_secret_value()) return self._client # ------------------------------------------------------------------ discover / capabilities def discover(self) -> dict[str, Any]: out: dict[str, Any] = {"source_id": SOURCE_ID, "probed_at": datetime.now(UTC).isoformat(), "last_update": {}} for t in ("stock", "etf", "futures", "index", "fx", "crypto", "options"): try: out["last_update"][t] = self.client.last_update(t).isoformat() except Exception as e: out["last_update"][t] = f"error: {e}" return out def capabilities(self) -> Capabilities: return Capabilities( source_id=SOURCE_ID, asset_classes=tuple(a.value for a in AssetClass), intervals=("1m", "5m", "30m", "1h", "1d"), adjustments=("raw", "split", "split_div", "continuous"), supports_symbol_filter=False, supports_date_range=False, incremental_windows=("day", "week", "month"), license_class="frd-private", notes={"raw_only_for": ["1m", "1d"], "zero_volume_bars": "absent at source", "timestamp_convention": "bar_open", "timezone": "America/New_York (crypto: UTC)"}, ) # ------------------------------------------------------------------ fetch def fetch_range(self, request: FetchRequest, dest_dir: Path) -> FetchResult: p = dict(request.params) endpoint = p.pop("endpoint", "data_file") asset = p.get("type", "stock") vendor_last = None try: vendor_last = self.client.last_update(asset, full=(p.get("period") == "full")) except Exception: vendor_last = None stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") label = "_".join(f"{k}-{v}" for k, v in sorted(p.items())) dest = dest_dir / f"{endpoint}__{label}__{stamp}.zip" got = self.client.download(endpoint, p, dest) file_count = None try: import zipfile with zipfile.ZipFile(got.path) as zf: file_count = sum(1 for i in zf.infolist() if not i.is_dir()) except Exception: pass return FetchResult(request=request, path=got.path, sha256=got.sha256, bytes=got.bytes, fetched_at=got.fetched_at, vendor_last_update=vendor_last, file_count=file_count, source_url_sanitized=self.client.sanitized_url(endpoint, p)) # ------------------------------------------------------------------ normalize def normalize(self, result: FetchResult, **kwargs: Any) -> Iterator[NormalizedBatch]: p = result.request.params asset = AssetClass(p.get("type", "stock")) interval = VENDOR_TF[p.get("timeframe", "1min")] adj_vendor = p.get("adjustment") or ("contin_UNadj" if asset == AssetClass.futures and not p.get("contracts") else "UNADJUSTED") adjustment_value = ADJ_MAP.get(adj_vendor, "raw") contracts = bool(p.get("contracts")) ingested_at = datetime.now(UTC) only: set[str] | None = set(kwargs.get("symbols") or []) or None tz = SOURCE_TZ.get(asset.value, "America/New_York") for name, data in iter_archive_files(result.path): base = Path(name).name.lower() if base.startswith("_") or "readme" in base or base.endswith((".csv", ".txt")) is False: continue symbol = ticker_from_filename(name, contract=contracts) if only and symbol not in only: continue table, raw_rows = parse_bar_file(data, symbol, asset, interval, _Adj(adjustment_value), SOURCE_ID, ingested_at, source_tz=tz) if table.num_rows == 0: continue sd = table.column("session_date").to_pylist() yield NormalizedBatch(table=table, symbol=symbol, session_dates=(min(sd), max(sd)), row_count=table.num_rows, raw_rows=raw_rows) # ------------------------------------------------------------------ validate def validate(self, batch: NormalizedBatch) -> ValidationReport: interval = Interval(batch.table.column("interval")[0].as_py()) asset = batch.table.column("asset_class")[0].as_py() table, anomalies = validate_bars(batch.table, interval, asset_class=asset) batch.table = table flagged = sum(1 for f in table.column("quality_flags").to_pylist() if f) errors = [a for a in anomalies if a["severity"] == "error"] return ValidationReport(rows=table.num_rows, flagged_rows=flagged, anomalies=anomalies, ok=not errors) # ------------------------------------------------------------------ checkpoint def checkpoint(self, result: FetchResult, batches_published: int) -> Checkpoint: p = result.request.params return Checkpoint(scope=f"dataset:{result.request.dataset_id}", key="last_archive", value={"sha256": result.sha256, "vendor_last_update": result.vendor_last_update.isoformat() if result.vendor_last_update else None, "params": p, "fetched_at": result.fetched_at.isoformat(), "batches": batches_published}) # ------------------------------------------------------------------ reference data def ticker_listing(self, type_: str) -> list[dict[str, Any]]: txt = self.client.ticker_listing(type_) rows = list(csv.DictReader(io.StringIO(txt))) out = [] for r in rows: sym = (r.get("Ticker") or "").strip() if not sym: continue out.append({"symbol": sym, "name": (r.get("Name") or "").strip() or None, "first_date": _d(r.get("First Date")), "last_date": _d(r.get("Last Date")), "status": "delisted" if sym.endswith("-DELISTED") else "active", "asset_class": type_}) return out def delisted_log(self) -> list[dict[str, Any]]: txt = self.client.update_log("stock", "delisted") return [{"symbol": r["Ticker"].strip(), "last_date": _d(r.get("Last Date"))} for r in csv.DictReader(io.StringIO(txt)) if r.get("Ticker")] def meta_actions(self, type_: str, kind: str, dest_dir: Path) -> tuple[FetchResult, list[dict[str, Any]]]: """kind = splits | dividends → list of {symbol, ex_date, value}.""" req = FetchRequest(dataset_id=f"corporate_actions.{type_}", params={"endpoint": "meta_file", "type": type_, "metafile_type": kind}) res = self.fetch_range(req, dest_dir) rows = [] for name, data in iter_archive_files(res.path): if name.startswith("_") or "readme" in name.lower(): continue sym = Path(name).stem.upper() for d, v in parse_meta_pairs(data): rows.append({"symbol": sym, "ex_date": d, "value": v, "action_type": "split" if kind == "splits" else "dividend"}) return res, rows class _Adj(str): """Adjustment label passthrough (dataset adjustment ids are richer than the Adjustment enum).""" @property def value(self) -> str: # mimics Enum.value return str(self) def _d(s: str | None) -> date | None: if not s: return None try: return date.fromisoformat(s.strip()[:10]) except ValueError: return None def sanitized_request(params: dict[str, Any]) -> str: return json.dumps({k: v for k, v in params.items() if k != "userid"}, sort_keys=True) def digest_params(params: dict[str, Any]) -> str: return hashlib.sha256(sanitized_request(params).encode()).hexdigest()[:16]