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