1from __future__ import annotations23import base644import io5from datetime import date, datetime6from typing import Literal78import pyarrow as pa9import pyarrow.ipc as ipc10from fastapi import APIRouter, Depends, Query, Request, Response11from fastapi.responses import StreamingResponse1213from findata_api import auth as A14from findata_api import metrics as M15from findata_api.errors import ApiError16from findata_api.ratelimit import record_usage17from findata_core.config import get_settings18from findata_core.db import connection19from findata_ingest.reader import apply_adjustment, read_bars2021router = APIRouter(prefix="/market", tags=["market"])22INTERVAL_DS = {"1m": "market_bars.{a}.1m.raw", "5m": "market_bars.{a}.5m.raw", "1d": "market_bars.{a}.1d.raw"}23MAX_SPAN_DAYS = {"1m": 31, "5m": 186, "1d": 3660}24PAGE_MAX = 10_000252627def _dataset_for(symbol: str, interval: str) -> tuple[str, str]:28 with connection() as conn:29 inst = conn.execute("SELECT asset_class FROM instruments WHERE symbol=%s ORDER BY (asset_class='stock') DESC LIMIT 1", (symbol,)).fetchone()30 asset = inst["asset_class"] if inst else "stock"31 if interval not in INTERVAL_DS:32 raise ApiError(400, "bad_interval", "interval must be one of 1m, 5m, 1d")33 return INTERVAL_DS[interval].format(a=asset), asset343536@router.get("/bars")37def bars(request: Request, response: Response,38 symbol: str = Query(..., min_length=1, max_length=20), interval: Literal["1m", "5m", "1d"] = "1m",39 start: date = Query(...), end: date = Query(...), adjustment: Literal["raw", "split", "split_div"] = "raw",40 limit: int = Query(5000, ge=1, le=PAGE_MAX), cursor: str | None = None,41 format: Literal["json", "arrow", "csv"] = "json", version: int | None = None,42 p: A.Principal = Depends(A.require("read:market"))):43 sym = symbol.upper().strip()44 if end < start:45 raise ApiError(400, "bad_range", "end must be >= start")46 if (end - start).days > MAX_SPAN_DAYS[interval]:47 raise ApiError(413, "range_too_large", f"{interval} requests are limited to {MAX_SPAN_DAYS[interval]} days per call; "48 "use POST /api/v1/exports for larger ranges (async Parquet).", max_days=MAX_SPAN_DAYS[interval])49 dataset_id, asset = _dataset_for(sym, interval)50 table, meta = read_bars(dataset_id, [sym], start, end, version=version)51 notes: list[str] = []52 if adjustment != "raw":53 table, notes = apply_adjustment(table, adjustment)54 # cursor = base64(last timestamp_utc isoformat)55 if cursor:56 try:57 after = datetime.fromisoformat(base64.urlsafe_b64decode(cursor.encode()).decode())58 except Exception: # noqa: BLE00159 raise ApiError(400, "bad_cursor", "invalid cursor") from None60 import pyarrow.compute as pc61 table = table.filter(pc.greater(table.column("timestamp_utc"), pa.scalar(after, pa.timestamp("us", tz="UTC"))))62 total = table.num_rows63 page = table.slice(0, limit)64 next_cursor = None65 if total > limit:66 last = page.column("timestamp_utc")[-1].as_py()67 next_cursor = base64.urlsafe_b64encode(last.isoformat().encode()).decode()68 response.headers["X-Dataset-Id"] = dataset_id69 response.headers["X-Dataset-Version"] = str(meta.get("version"))70 M.ROWS.labels("bars").inc(page.num_rows)71 record_usage(p.label, rows=page.num_rows)72 if format == "arrow":73 sink = io.BytesIO()74 with ipc.new_stream(sink, page.schema) as w:75 w.write_table(page)76 headers = {"X-Dataset-Id": dataset_id, "X-Dataset-Version": str(meta.get("version")), "X-Next-Cursor": next_cursor or ""}77 return Response(sink.getvalue(), media_type="application/vnd.apache.arrow.stream", headers=headers)78 if format == "csv":79 import pyarrow.csv as pcsv80 sink = io.BytesIO()81 pcsv.write_csv(page.drop_columns(["quality_flags"]) if "quality_flags" in page.column_names else page, sink)82 return StreamingResponse(iter([sink.getvalue()]), media_type="text/csv",83 headers={"Content-Disposition": f'attachment; filename="{sym}_{interval}_{start}_{end}.csv"', "X-Next-Cursor": next_cursor or ""})84 rows = page.to_pylist()85 for r in rows:86 r["timestamp_utc"] = r["timestamp_utc"].isoformat().replace("+00:00", "Z")87 r["session_date"] = r["session_date"].isoformat()88 r["ingested_at"] = r["ingested_at"].isoformat().replace("+00:00", "Z")89 return {"data": rows, "meta": {**meta, "symbol": sym, "interval": interval, "adjustment": adjustment, "start": str(start), "end": str(end),90 "rows": len(rows), "rows_matching": total, "adjustment_notes": notes, "timestamp_convention": "bar_open",91 "license": "frd-private: no redistribution"}, "next_cursor": next_cursor}929394@router.get("/sessions")95def sessions(symbol: str, interval: Literal["1m", "5m"] = "1m", start: date | None = None, end: date | None = None,96 p: A.Principal = Depends(A.require("read:market"))):97 """Per-session coverage for one symbol (bar counts vs expected regular-hours bars)."""98 sym = symbol.upper()99 dataset_id, _ = _dataset_for(sym, interval)100 with connection() as conn:101 rows = conn.execute("SELECT session_date, bar_count, expected_bars, first_ts, last_ts, flags FROM coverage WHERE dataset_id=%s AND symbol=%s "102 "AND (%s::date IS NULL OR session_date>=%s) AND (%s::date IS NULL OR session_date<=%s) ORDER BY session_date DESC LIMIT 2000",103 (dataset_id, sym, start, start, end, end)).fetchall()104 return {"data": rows, "meta": {"dataset_id": dataset_id, "symbol": sym}}105