bulk: extraits Parquet annuels des fondamentaux (/v1/bulk/fundamentals/{year}.parquet) hors quota, ETag fort + If-None-Match → 304

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Simon-Pierre Boucher committed 20 days ago (Sep 5, 2026) parent 78a6940

3 changed files +170 −0

added hfmarketdata/api/bulk/__init__.py +4 −0

		`@@ -0,0 +1,4 @@`
	1	+"""Bulk Parquet extracts (`/v1/bulk/*`) — quota exempt, served with ETag / 304.
	2	+
	3	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	4	+"""

added hfmarketdata/api/bulk/build.py +99 −0

		`@@ -0,0 +1,99 @@`
	1	+"""Build the yearly fundamentals Parquet files under `data_root/bulk/`.
	2	+
	3	+`fundamentals_{year}.parquet` = the latest version of every standardized statement row whose fiscal year is
	4	+`year` (all companies, all three statements, wide layout identical to `fund_statements`, `coverage` as JSON
	5	+text). A sidecar `.meta.json` keeps the row count, build time and the strong ETag (sha256 of the file).
	6	+
	7	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	8	+"""
	9	+from __future__ import annotations
	10	+
	11	+import hashlib
	12	+import json
	13	+from datetime import datetime
	14	+from pathlib import Path
	15	+
	16	+import pandas as pd
	17	+from sqlalchemy import func, select
	18	+
	19	+from core.config import settings
	20	+from core.db import session
	21	+
	22	+
	23	+def bulk_dir() -> Path:
	24	+ d = settings.data_root / "bulk"
	25	+ d.mkdir(parents=True, exist_ok=True)
	26	+ return d
	27	+
	28	+
	29	+def fundamentals_path(year: int) -> Path:
	30	+ return bulk_dir() / f"fundamentals_{int(year)}.parquet"
	31	+
	32	+
	33	+def meta_path(year: int) -> Path:
	34	+ return bulk_dir() / f"fundamentals_{int(year)}.meta.json"
	35	+
	36	+
	37	+def available_years() -> list[int]:
	38	+ from fundamentals.models import fund_statements
	39	+ with session() as s:
	40	+ rows = s.execute(select(fund_statements.c.fiscal_year, func.count()).group_by(fund_statements.c.fiscal_year)
	41	+ .order_by(fund_statements.c.fiscal_year)).all()
	42	+ return [int(y) for y, _ in rows]
	43	+
	44	+
	45	+def last_ingest_time() -> datetime \| None:
	46	+ from fundamentals.models import FundIngestState
	47	+ with session() as s:
	48	+ return s.scalar(select(func.max(FundIngestState.last_run_at)))
	49	+
	50	+
	51	+def is_stale(year: int) -> bool:
	52	+ p, m = fundamentals_path(year), meta_path(year)
	53	+ if not p.is_file() or not m.is_file():
	54	+ return True
	55	+ built = datetime.fromisoformat(json.loads(m.read_text())["built_at"])
	56	+ last = last_ingest_time()
	57	+ return bool(last and last > built)
	58	+
	59	+
	60	+def build_year(year: int) -> dict:
	61	+ """(Re)build the file for one fiscal year; returns the sidecar metadata."""
	62	+ from fundamentals.models import fund_statements
	63	+ t = fund_statements
	64	+ rn = func.row_number().over(partition_by=[t.c.cik, t.c.statement, t.c.fiscal_year, t.c.fiscal_quarter],
	65	+ order_by=[t.c.filed_date.desc(), t.c.id.desc()]).label("rn")
	66	+ sub = select(t, rn).where(t.c.fiscal_year == int(year)).subquery()
	67	+ q = select(sub).where(sub.c.rn == 1).order_by(sub.c.ticker, sub.c.statement, sub.c.fiscal_quarter)
	68	+ with session() as s:
	69	+ df = pd.DataFrame([dict(r._mapping) for r in s.execute(q)])
	70	+ if not df.empty:
	71	+ df = df.drop(columns=["rn", "id"])
	72	+ df["coverage"] = df["coverage"].map(lambda v: json.dumps(v) if v is not None else None)
	73	+ for c in ("period_start", "period_end", "filed_date"):
	74	+ df[c] = pd.to_datetime(df[c])
	75	+ else:
	76	+ df = pd.DataFrame(columns=[c for c in t.c.keys() if c != "id"])
	77	+ p = fundamentals_path(year)
	78	+ tmp = p.with_suffix(".parquet.tmp")
	79	+ df.to_parquet(tmp, index=False, compression="zstd")
	80	+ tmp.replace(p)
	81	+ etag = '"' + hashlib.sha256(p.read_bytes()).hexdigest()[:32] + '"'
	82	+ from fundamentals import utcnow
	83	+ meta = {"year": int(year), "rows": int(len(df)), "built_at": utcnow().isoformat(), "etag": etag,
	84	+ "bytes": p.stat().st_size, "columns": list(df.columns)}
	85	+ meta_path(year).write_text(json.dumps(meta))
	86	+ return meta
	87	+
	88	+
	89	+def ensure_year(year: int) -> dict \| None:
	90	+ """Build when missing/stale; returns the metadata or None when the year has no data at all."""
	91	+ if is_stale(year):
	92	+ if int(year) not in available_years():
	93	+ return None
	94	+ return build_year(year)
	95	+ return json.loads(meta_path(year).read_text())
	96	+
	97	+
	98	+def build_all() -> list[dict]:
	99	+ return [build_year(y) for y in available_years()]

added hfmarketdata/api/bulk/routes.py +67 −0

		`@@ -0,0 +1,67 @@`
	1	+"""`/v1/bulk/*` — whole-universe Parquet extracts, outside the rows quota (`request.state.quota_exempt = True`).
	2	+
	3	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	4	+"""
	5	+from __future__ import annotations
	6	+
	7	+from fastapi import APIRouter, Request, Response
	8	+from fastapi.responses import FileResponse
	9	+
	10	+from core.errors import ApiError
	11	+from core.responses import json_response
	12	+
	13	+from . import build
	14	+
	15	+router = APIRouter(prefix="/v1/bulk", tags=["bulk"])
	16	+
	17	+FUNDAMENTALS_COLUMNS_NOTE = ("One row per (company, statement, fiscal period) — the latest version of each period; "
	18	+ "columns = fund_statements (cik, ticker, statement, fiscal_year, fiscal_quarter [0 = annual], "
	19	+ "period_start, period_end, calendar_quarter, form, accn, filed_date, derived, restated, currency, "
	20	+ "coverage (JSON text), mapping_version, then every standardized account).")
	21	+
	22	+
	23	+@router.get("/fundamentals", summary="List bulk fundamentals files",
	24	+ description="Fiscal years for which a `fundamentals_{year}.parquet` extract can be downloaded, with the row "
	25	+ "count and ETag of the files already built. Files are rebuilt lazily after each ingestion. "
	26	+ "Quota exempt.",
	27	+ responses={200: {"description": "Available years", "content": {"application/json": {"example": {
	28	+ "data": [{"year": 2024, "url": "/v1/bulk/fundamentals/2024.parquet", "rows": 21876, "etag": "\"3f2a…\"",
	29	+ "built_at": "2026-09-04T18:02:11"}], "meta": {"count": 1}}}}}},
	30	+ openapi_extra={"x-errors": []})
	31	+def list_fundamentals(request: Request):
	32	+ request.state.quota_exempt = True
	33	+ out = []
	34	+ for y in build.available_years():
	35	+ meta = None
	36	+ if not build.is_stale(y):
	37	+ import json
	38	+ meta = json.loads(build.meta_path(y).read_text())
	39	+ out.append({"year": y, "url": f"/v1/bulk/fundamentals/{y}.parquet", "rows": meta["rows"] if meta else None,
	40	+ "etag": meta["etag"] if meta else None, "built_at": meta["built_at"] if meta else None,
	41	+ "stale": meta is None})
	42	+ return json_response(out, meta={"note": FUNDAMENTALS_COLUMNS_NOTE})
	43	+
	44	+
	45	+@router.get("/fundamentals/{year}.parquet", summary="Download one fiscal year of standardized statements (Parquet)",
	46	+ description="All companies × all statements for fiscal year `{year}` as a zstd Parquet file (latest version of "
	47	+ "each period, point-in-time columns `accn`/`filed_date` kept). Strong `ETag`; send "
	48	+ "`If-None-Match` to get `304 Not Modified` for free. Quota exempt — this is the recommended way "
	49	+ "to load the whole universe.\n\n" + FUNDAMENTALS_COLUMNS_NOTE,
	50	+ responses={200: {"description": "Parquet file", "content": {"application/vnd.apache.parquet": {}},
	51	+ "headers": {"ETag": {"schema": {"type": "string"}}, "X-Row-Count": {"schema": {"type": "integer"}}}},
	52	+ 304: {"description": "Not modified (ETag matched)"}},
	53	+ openapi_extra={"x-errors": ["NOT_FOUND", "INVALID_PARAMETER"]})
	54	+def download_fundamentals(year: int, request: Request):
	55	+ request.state.quota_exempt = True
	56	+ if year < 2000 or year > 2100:
	57	+ raise ApiError(400, "INVALID_PARAMETER", "year must be a fiscal year between 2000 and 2100")
	58	+ meta = build.ensure_year(year)
	59	+ if meta is None:
	60	+ raise ApiError(404, "NOT_FOUND", f"No fundamentals for fiscal year {year}",
	61	+ details={"available_years": build.available_years()})
	62	+ inm = request.headers.get("if-none-match")
	63	+ if inm and meta["etag"] in [t.strip() for t in inm.split(",")]:
	64	+ return Response(status_code=304, headers={"ETag": meta["etag"], "X-Row-Count": "0", "Cache-Control": "public, max-age=3600"})
	65	+ return FileResponse(build.fundamentals_path(year), media_type="application/vnd.apache.parquet",
	66	+ filename=f"fundamentals_{year}.parquet",
	67	+ headers={"ETag": meta["etag"], "X-Row-Count": str(meta["rows"]), "Cache-Control": "public, max-age=3600"})
	68