futures: service de données, moteur de rolls/ajustements et les 7 endpoints /v1/futures

- rolls.py : calendriers de roll calendar/first_notice/volume/open_interest (fenêtre 30 séances), profondeur 1–3, écarts par roll, ajustements additif/ratio (dernier contrat non ajusté, écart null jamais inventé)
- service.py : barres d'un contrat (fusion archive+update, session rth/eth sur l'horloge ET, curseur, UTC ISO 8601 DST-aware), chaîne, continu (schedule quotidien appliqué à l'intraday), structure à terme, couverture, racines, contrats
- routes.py : /roots, /{root}/contracts, /contract/{symbol}/bars, /contract/{symbol}/coverage, /{root}/chain, /{root}/continuous, /{root}/term-structure — OpenAPI riche, x-errors, clamp_limit par tier, json/csv/parquet

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

Simon-Pierre Boucher committed 20 days ago (Sep 4, 2026) parent f48ce8a

3 changed files +1,037 −0

added hfmarketdata/api/futures/rolls.py +242 −0

		`@@ -0,0 +1,242 @@`
	1	+"""Roll schedules and continuous-series stitching (pure pandas, no I/O).
	2	+
	3	+Inputs: the contracts of a root ordered by expiration, and their merged daily bars
	4	+(`symbol, date, open, high, low, close, volume, open_interest`). Output: front-month segments
	5	+(`Segment(symbol, start, end)`), depth-N schedules and back/ratio-adjusted series.
	6	+
	7	+Roll methods (`roll_dates[i]` = first session on which the new contract is the front):
	8	+* `calendar` — the old contract is held through its `expiration_date`; roll the next session.
	9	+* `first_notice` — roll on `first_notice_date` (held through the session before FND); falls back to
	10	+ calendar when FND is null (cash-settled products).
	11	+* `volume` — roll after the 2nd consecutive session where the next contract's volume is greater
	12	+ than the front's (never later than the calendar roll).
	13	+* `open_interest` — same test on open interest.
	14	+
	15	+Adjustments (latest contract always unadjusted):
	16	+* `back_adjusted` — additive: gap_k = close_new − close_old on the last session of the old segment
	17	+ (nearest common session within 5 sessions); every bar before roll k gets
	18	+ `+ Σ gap_j (j ≥ k)`.
	19	+* `ratio_adjusted` — multiplicative: factor_k = close_new / close_old; bars before roll k are
	20	+ multiplied by `Π factor_j (j ≥ k)`.
	21	+Volume and open interest are never adjusted. If no common session exists, the gap is null and the
	22	+roll is flagged (`adjusted: false`) — no value is invented.
	23	+
	24	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	25	+"""
	26	+from __future__ import annotations
	27	+
	28	+from dataclasses import dataclass
	29	+from datetime import date, timedelta
	30	+from typing import Any, Literal
	31	+
	32	+import numpy as np
	33	+import pandas as pd
	34	+
	35	+from . import calendar_us as cal
	36	+
	37	+RollMethod = Literal["calendar", "first_notice", "volume", "open_interest"]
	38	+Adjust = Literal["none", "back_adjusted", "ratio_adjusted"]
	39	+ROLL_METHODS = ("calendar", "first_notice", "volume", "open_interest")
	40	+ADJUSTMENTS = ("none", "back_adjusted", "ratio_adjusted")
	41	+PRICE_COLS = ("open", "high", "low", "close")
	42	+
	43	+
	44	+@dataclass(frozen=True)
	45	+class Segment:
	46	+ symbol: str
	47	+ start: date # inclusive
	48	+ end: date \| None # inclusive; None = open-ended (current front)
	49	+ position: int # index in the contract sequence
	50	+
	51	+
	52	+def _d(v: Any) -> date \| None:
	53	+ if v is None or (isinstance(v, float) and np.isnan(v)):
	54	+ return None
	55	+ if isinstance(v, pd.Timestamp):
	56	+ return v.date()
	57	+ if isinstance(v, date):
	58	+ return v
	59	+ return pd.Timestamp(v).date()
	60	+
	61	+
	62	+def _sessions(daily: pd.DataFrame) -> np.ndarray:
	63	+ """Sorted unique session dates across all contracts (datetime64[D])."""
	64	+ if daily is None or daily.empty:
	65	+ return np.array([], dtype="datetime64[D]")
	66	+ return np.unique(daily["date"].values.astype("datetime64[D]"))
	67	+
	68	+
	69	+def _next_session(sessions: np.ndarray, d: date, calendar: str) -> date:
	70	+ """First session strictly after d (from the data), else the next business day."""
	71	+ if len(sessions):
	72	+ i = np.searchsorted(sessions, np.datetime64(d, "D"), side="right")
	73	+ if i < len(sessions):
	74	+ return sessions[i].astype("datetime64[D]").astype(date)
	75	+ return cal.next_business_day(d, calendar, inclusive=False)
	76	+
	77	+
	78	+ROLL_WINDOW_SESSIONS = 30 # volume/OI tests are only evaluated in the last N sessions of the front contract
	79	+
	80	+
	81	+def _metric_roll_date(daily: pd.DataFrame, old: str, new: str, metric: str, lower: date, upper: date,
	82	+ window: int = ROLL_WINDOW_SESSIONS) -> date \| None:
	83	+ """First session after the 2nd consecutive session in (lower, upper] where metric(new) > metric(old).
	84	+ Only the last `window` sessions of the old contract (≤ upper) are examined: far-dated months with
	85	+ near-zero, noisy volume must not trigger a roll a year early."""
	86	+ a = daily.loc[daily["symbol"] == old, ["date", metric]].set_index("date")[metric]
	87	+ b = daily.loc[daily["symbol"] == new, ["date", metric]].set_index("date")[metric]
	88	+ a = a[a.index <= pd.Timestamp(upper)].sort_index().tail(window)
	89	+ both = pd.concat([a.rename("old"), b.rename("new")], axis=1).dropna().sort_index()
	90	+ both = both[(both.index > pd.Timestamp(lower)) & (both.index <= pd.Timestamp(upper))]
	91	+ if both.empty:
	92	+ return None
	93	+ hit = (both["new"] > both["old"]).to_numpy()
	94	+ for i in range(1, len(hit)):
	95	+ if hit[i] and hit[i - 1]:
	96	+ # the roll takes effect on the next session (any contract) after the 2nd confirming session
	97	+ nxt = daily.loc[daily["date"] > both.index[i], "date"]
	98	+ return _d(nxt.min()) if not nxt.empty else None
	99	+ return None
	100	+
	101	+
	102	+def roll_schedule(contracts: list[dict[str, Any]], daily: pd.DataFrame \| None, method: str = "calendar",
	103	+ calendar: str = "us") -> list[Segment]:
	104	+ """Front-month segments for a root.
	105	+
	106	+ `contracts`: dicts with symbol, expiration_date, first_notice_date, first_data_date, last_data_date,
	107	+ ordered by expiration then symbol. Contracts already expired when the previous roll happens are skipped.
	108	+ """
	109	+ if method not in ROLL_METHODS:
	110	+ raise ValueError(f"unknown roll method {method}")
	111	+ seq = [c for c in contracts if _d(c.get("expiration_date")) is not None and _d(c.get("first_data_date")) is not None]
	112	+ seq.sort(key=lambda c: (_d(c["expiration_date"]), c["symbol"]))
	113	+ if not seq:
	114	+ return []
	115	+ sessions = _sessions(daily) if daily is not None else np.array([], dtype="datetime64[D]")
	116	+ segments: list[Segment] = []
	117	+ start = _d(seq[0]["first_data_date"])
	118	+ k = 0
	119	+ pos = 0
	120	+ while k < len(seq):
	121	+ cur = seq[k]
	122	+ exp = _d(cur["expiration_date"])
	123	+ # next contract = first later contract that is still alive after `exp`
	124	+ j = k + 1
	125	+ while j < len(seq) and _d(seq[j]["expiration_date"]) <= exp:
	126	+ j += 1
	127	+ if j >= len(seq):
	128	+ segments.append(Segment(cur["symbol"], start, None, pos))
	129	+ break
	130	+ nxt = seq[j]
	131	+ cal_roll = _next_session(sessions, exp, calendar)
	132	+ roll = cal_roll
	133	+ if method == "first_notice":
	134	+ fnd = _d(cur.get("first_notice_date"))
	135	+ if fnd is not None and start < fnd < cal_roll:
	136	+ roll = fnd
	137	+ elif method in ("volume", "open_interest") and daily is not None and not daily.empty:
	138	+ m = _metric_roll_date(daily, cur["symbol"], nxt["symbol"], method, start, exp)
	139	+ if m is not None and start < m < cal_roll:
	140	+ roll = m
	141	+ if roll <= start: # degenerate (contract with data only after its own expiry)
	142	+ roll = _next_session(sessions, start, calendar)
	143	+ segments.append(Segment(cur["symbol"], start, roll - timedelta(days=1), pos))
	144	+ start = roll
	145	+ k = j
	146	+ pos += 1
	147	+ return segments
	148	+
	149	+
	150	+def depth_schedule(front: list[Segment], contracts: list[dict[str, Any]], depth: int) -> list[Segment]:
	151	+ """Depth-N segments: same windows as the front schedule, symbol = N-th contract in the sequence
	152	+ used by the front (front = 1). Segments with no such contract are dropped."""
	153	+ if depth == 1:
	154	+ return front
	155	+ order = [s.symbol for s in front]
	156	+ # contracts after the last front symbol (for the tail of deeper schedules)
	157	+ seq = [c["symbol"] for c in sorted(
	158	+ (c for c in contracts if _d(c.get("expiration_date")) is not None), key=lambda c: (_d(c["expiration_date"]), c["symbol"]))]
	159	+ tail = [s for s in seq if s not in set(order) and (not order or seq.index(s) > seq.index(order[-1]))]
	160	+ chain = order + tail
	161	+ out: list[Segment] = []
	162	+ for s in front:
	163	+ i = chain.index(s.symbol) + depth - 1
	164	+ if i < len(chain):
	165	+ out.append(Segment(chain[i], s.start, s.end, s.position))
	166	+ return out
	167	+
	168	+
	169	+def roll_gaps(segments: list[Segment], daily: pd.DataFrame, lookback: int = 5) -> list[dict[str, Any]]:
	170	+ """One entry per roll (between consecutive segments): date, from_symbol, to_symbol, gap, ratio."""
	171	+ out: list[dict[str, Any]] = []
	172	+ if daily is None or daily.empty:
	173	+ daily = pd.DataFrame(columns=["symbol", "date", "close"])
	174	+ closes = daily.pivot_table(index="date", columns="symbol", values="close", aggfunc="last") if not daily.empty else pd.DataFrame()
	175	+ for prev, nxt in zip(segments, segments[1:]):
	176	+ roll_date = nxt.start
	177	+ gap = ratio = None
	178	+ common = None
	179	+ if not closes.empty and prev.symbol in closes.columns and nxt.symbol in closes.columns:
	180	+ # last `lookback` sessions of the old contract up to the end of its segment
	181	+ old = closes[prev.symbol].dropna()
	182	+ old = old[old.index <= pd.Timestamp(prev.end)]
	183	+ if not old.empty:
	184	+ both = closes.loc[old.index[-lookback:], [prev.symbol, nxt.symbol]].dropna()
	185	+ if not both.empty: # nearest common session
	186	+ common = both.index[-1]
	187	+ co, cn = float(both.iloc[-1][prev.symbol]), float(both.iloc[-1][nxt.symbol])
	188	+ gap = cn - co
	189	+ ratio = cn / co if co else None
	190	+ out.append({"date": str(roll_date), "from_symbol": prev.symbol, "to_symbol": nxt.symbol,
	191	+ "gap": None if gap is None else round(gap, 10), "ratio": None if ratio is None else round(ratio, 12),
	192	+ "gap_session": None if common is None else str(_d(common)), "adjusted": gap is not None})
	193	+ return out
	194	+
	195	+
	196	+def adjustment_offsets(rolls: list[dict[str, Any]], adjust: str) -> tuple[list[float], list[float]]:
	197	+ """Per-segment additive offset and multiplicative factor (segment i has rolls i.. after it)."""
	198	+ n = len(rolls) + 1
	199	+ add = [0.0] * n
	200	+ mul = [1.0] * n
	201	+ if adjust == "none":
	202	+ return add, mul
	203	+ cum_add, cum_mul = 0.0, 1.0
	204	+ for i in range(n - 2, -1, -1):
	205	+ r = rolls[i]
	206	+ if adjust == "back_adjusted" and r["gap"] is not None:
	207	+ cum_add += r["gap"]
	208	+ if adjust == "ratio_adjusted" and r["ratio"] is not None:
	209	+ cum_mul *= r["ratio"]
	210	+ add[i], mul[i] = cum_add, cum_mul
	211	+ return add, mul
	212	+
	213	+
	214	+def apply_adjustment(frames: list[pd.DataFrame], add: list[float], mul: list[float]) -> pd.DataFrame:
	215	+ """Concatenate per-segment bar frames applying each segment's offset/factor to OHLC."""
	216	+ parts = []
	217	+ for i, f in enumerate(frames):
	218	+ if f is None or f.empty:
	219	+ continue
	220	+ f = f.copy()
	221	+ if add[i] or mul[i] != 1.0:
	222	+ for c in PRICE_COLS:
	223	+ if c in f.columns:
	224	+ f[c] = f[c] * mul[i] + add[i]
	225	+ parts.append(f)
	226	+ if not parts:
	227	+ return pd.DataFrame(columns=["symbol", "datetime", *PRICE_COLS, "volume"])
	228	+ return pd.concat(parts, ignore_index=True).sort_values("datetime", kind="stable").reset_index(drop=True)
	229	+
	230	+
	231	+def stitch_daily(segments: list[Segment], daily: pd.DataFrame, adjust: str = "none") -> tuple[pd.DataFrame, list[dict[str, Any]]]:
	232	+ """Convenience for daily series/tests: build the continuous daily frame from the merged daily bars."""
	233	+ rolls = roll_gaps(segments, daily)
	234	+ add, mul = adjustment_offsets(rolls, adjust)
	235	+ frames = []
	236	+ for s in segments:
	237	+ f = daily[daily["symbol"] == s.symbol]
	238	+ f = f[f["date"] >= pd.Timestamp(s.start)]
	239	+ if s.end is not None:
	240	+ f = f[f["date"] <= pd.Timestamp(s.end)]
	241	+ frames.append(f.rename(columns={"date": "datetime"}))
	242	+ return apply_adjustment(frames, add, mul), rolls

added hfmarketdata/api/futures/routes.py +276 −0

		`@@ -0,0 +1,276 @@`
	1	+"""`/v1/futures/*` — individual contracts, chains, continuous series, term structure (UPGRADE-PLAN §2).
	2	+
	3	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	4	+"""
	5	+from __future__ import annotations
	6	+
	7	+import pandas as pd
	8	+from core.responses import clamp_limit, frame_response, json_response, parse_format
	9	+from fastapi import APIRouter, Path, Query, Request
	10	+
	11	+from . import (
	12	+ models, # noqa: F401 (create_all at import)
	13	+ service,
	14	+)
	15	+
	16	+router = APIRouter(prefix="/v1/futures", tags=["futures"])
	17	+
	18	+LIMIT_DEFAULT = 5_000
	19	+LIMIT_JSON_MAX = 200_000
	20	+LIMIT_FILE_MAX = 2_000_000
	21	+
	22	+_INTERVAL_DOC = "Bar interval: `1m`, `5m`, `30m`, `1h`, `1d` (aliases `1min`, `5min`, `30min`, `1hour`, `1day` accepted)."
	23	+_SESSION_DOC = ("Intraday session filter on the US/Eastern clock: `rth` = regular hours of the product "
	24	+ "(09:30–16:00 for equity indices, 09:00–14:30 CL/NG, 08:20–13:30 GC, 08:20–15:00 ZN/FX…), "
	25	+ "`eth` = everything else, `all` (default). Ignored for `1d`.")
	26	+_FROM_DOC = "Lower bound (inclusive): `YYYY-MM-DD` (Eastern trading date) or ISO 8601 datetime (`Z`/offset → converted; naive = UTC)."
	27	+_TO_DOC = "Upper bound (inclusive date / exclusive datetime), same formats as `from`."
	28	+_FORMAT_DOC = "`json` (envelope `{data, meta}`), `csv` or `parquet` (up to 2 000 000 rows)."
	29	+_LIMIT_DOC = "Rows per page (default 5 000; max 200 000 JSON / 2 000 000 CSV-Parquet, capped by your tier)."
	30	+_CURSOR_DOC = "Opaque cursor from `meta.next_cursor` to fetch the next page."
	31	+
	32	+_BARS_EXAMPLE = {"data": [
	33	+ {"symbol": "ESZ24", "datetime": "2024-12-19T14:30:00Z", "open": 5921.25, "high": 5923.0, "low": 5919.5, "close": 5922.0, "volume": 1834.0},
	34	+ {"symbol": "ESZ24", "datetime": "2024-12-19T14:31:00Z", "open": 5922.0, "high": 5922.75, "low": 5920.0, "close": 5920.25, "volume": 1211.0}],
	35	+ "meta": {"count": 2, "symbol": "ESZ24", "root": "ES", "interval": "1m", "session": "rth", "timezone": "UTC",
	36	+ "next_cursor": "IjIwMjQtMTItMTkgMDk6MzE6MDAi", "expiration_date": "2024-12-20", "expiration_source": "rule", "status": "expired"}}
	37	+
	38	+
	39	+def _limit(limit: int \| None, fmt: str, request: Request) -> int:
	40	+ return clamp_limit(limit, LIMIT_DEFAULT, LIMIT_FILE_MAX if fmt in ("csv", "parquet") else LIMIT_JSON_MAX, request=request)
	41	+
	42	+
	43	+@router.get(
	44	+ "/roots",
	45	+ summary="List futures roots",
	46	+ description="""Every futures product available as individual contracts (142 roots) with its reference
	47	+specification: name, exchange, asset class, currency, contract size, tick size/value, settlement type,
	48	+expiry rule key, month cycle, RTH window (US/Eastern) and the CME aliases accepted in symbols
	49	+(`6E` → lake root `E6`, `ZB` → `US`…). `source` is `reference` when the specification comes from the
	50	+exchange table, `derived` when only the name (FirstRate) is known — nulls are never invented.
	51	+
	52	+Coverage columns (`first_data_date`, `last_data_date`, `contracts_count`) are computed by the contracts backfill.""",
	53	+ responses={200: {"description": "Roots", "content": {"application/json": {"example": {"data": [
	54	+ {"root": "ES", "name": "E-mini S&P 500", "exchange": "CME", "asset_class": "equity_index", "currency": "USD",
	55	+ "contract_size": 50.0, "contract_size_unit": "USD x index", "tick_size": 0.25, "tick_value": 12.5,
	56	+ "settlement_type": "cash", "expiry_rule": "third_friday", "first_notice_rule": None, "calendar": "us",
	57	+ "month_cycle": "HMUZ", "rth_start": "09:30", "rth_end": "16:00", "aliases": [], "first_data_date": "2008-01-02",
	58	+ "last_data_date": "2026-09-03", "contracts_count": 78, "source": "reference"}],
	59	+ "meta": {"count": 1}}}}}},
	60	+ openapi_extra={"x-errors": []},
	61	+)
	62	+def list_roots(request: Request,
	63	+ asset_class: str \| None = Query(None, description="Filter: equity_index, energy, metals, rates, ags, fx, crypto, volatility, softs, livestock"),
	64	+ exchange: str \| None = Query(None, description="Filter by exchange code (CME, CBOT, NYMEX, COMEX, ICE US, EUREX…)"),
	65	+ search: str \| None = Query(None, description="Substring on root or name (case-insensitive)"),
	66	+ format: str = Query("json", description=_FORMAT_DOC)):
	67	+ fmt = parse_format(format)
	68	+ rows = service.roots()
	69	+ if asset_class:
	70	+ rows = [r for r in rows if (r.get("asset_class") or "").lower() == asset_class.lower()]
	71	+ if exchange:
	72	+ rows = [r for r in rows if (r.get("exchange") or "").lower() == exchange.lower()]
	73	+ if search:
	74	+ s = search.lower()
	75	+ rows = [r for r in rows if s in r["root"].lower() or s in (r.get("name") or "").lower()]
	76	+ for r in rows:
	77	+ r.pop("updated_at", None)
	78	+ if fmt != "json":
	79	+ return frame_response(pd.DataFrame(rows), fmt, request=request, filename="futures_roots")
	80	+ return json_response(service.jsonable(rows))
	81	+
	82	+
	83	+@router.get(
	84	+ "/{root}/contracts",
	85	+ summary="List the contracts of a root",
	86	+ description="""All individual contracts of a root (e.g. `ES`, `CL`, `6E`/`E6`) with expiration
	87	+(`expiration_source` = `rule` from the exchange rule, or `data` = last bar date), last trading and first
	88	+notice dates, real data range, status (`active`/`expired`), 20-session average volume, last open interest,
	89	+available intervals and lake files. Filter by `status` and by expiration window, sort with `sort`
	90	+(prefix `-` for descending).""",
	91	+ responses={200: {"description": "Contracts", "content": {"application/json": {"example": {"data": [
	92	+ {"symbol": "ESZ24", "root": "ES", "month_code": "Z", "contract_month": 12, "contract_year": 2024,
	93	+ "expiration_date": "2024-12-20", "expiration_source": "rule", "last_trading_date": "2024-12-20", "first_notice_date": None,
	94	+ "settlement_type": "cash", "contract_size": 50.0, "tick_size": 0.25, "tick_value": 12.5, "currency": "USD", "exchange": "CME",
	95	+ "first_data_date": "2021-06-04", "last_data_date": "2024-12-20", "status": "expired", "volume_avg_daily": 1418211.4,
	96	+ "open_interest_last": 506519.0, "bars_1day": 895,
	97	+ "timeframes": {"1min": {"first": "2023-09-27 08:14:00", "last": "2024-12-20 09:29:00", "rows": 123342}},
	98	+ "files": {"1day": ["…/futures_contracts/1day/archive/ES_Z24_1day.parquet", None]}}],
	99	+ "meta": {"count": 1, "root": "ES"}}}}}},
	100	+ openapi_extra={"x-errors": ["ROOT_NOT_FOUND", "INVALID_PARAMETER"]},
	101	+)
	102	+def list_contracts(request: Request,
	103	+ root: str = Path(..., description="Root code (lake or CME alias, case-insensitive)"),
	104	+ status: str \| None = Query(None, description="`active` or `expired`"),
	105	+ from_: str \| None = Query(None, alias="from", description="Expiration date ≥ (YYYY-MM-DD)"),
	106	+ to: str \| None = Query(None, description="Expiration date ≤ (YYYY-MM-DD)"),
	107	+ sort: str = Query("expiration_date", description="expiration_date \| symbol \| first_data_date \| last_data_date \| volume_avg_daily \| open_interest_last (prefix `-` = desc)"),
	108	+ format: str = Query("json", description=_FORMAT_DOC)):
	109	+ fmt = parse_format(format)
	110	+ rows = service.contracts(root, status, service.parse_date(from_, "from"), service.parse_date(to, "to"), sort)
	111	+ for r in rows:
	112	+ r.pop("updated_at", None)
	113	+ if fmt != "json":
	114	+ flat = [{**r, "timeframes": ",".join(r["timeframes"] or {}) if isinstance(r.get("timeframes"), dict) else r.get("timeframes"),
	115	+ "files": None} for r in rows]
	116	+ return frame_response(pd.DataFrame(flat), fmt, request=request, filename=f"{service.normalize_root(root)}_contracts")
	117	+ return json_response(service.jsonable(rows), meta={"root": service.normalize_root(root), "status": status, "sort": sort})
	118	+
	119	+
	120	+@router.get(
	121	+ "/contract/{symbol}/bars",
	122	+ summary="Bars of one individual contract",
	123	+ description="""OHLCV bars of a single futures contract (`ESZ24`, `ESZ2024`, `ES_Z24`, `6EZ24`…), 1-minute to daily.
	124	+
	125	+* Archive and update files are merged and deduplicated on the bar timestamp (update wins).
	126	+* Intraday timestamps are converted from the exchange's US/Eastern clock to UTC (`…Z`), DST-aware;
	127	+ daily bars are dates and carry `open_interest`.
	128	+* `session=rth\|eth` filters on the product's regular-hours window (see `/v1/futures/roots`).
	129	+* Paginate with `limit` + `meta.next_cursor`; for large extracts use `format=parquet` (half the row cost).""",
	130	+ responses={200: {"description": "Bars", "content": {"application/json": {"example": _BARS_EXAMPLE},
	131	+ "text/csv": {"example": "symbol,datetime,open,high,low,close,volume\nESZ24,2024-12-19T14:30:00Z,5921.25,5923.0,5919.5,5922.0,1834.0\n"}}}},
	132	+ openapi_extra={"x-errors": ["INVALID_CONTRACT_SYMBOL", "CONTRACT_NOT_FOUND", "INVALID_PARAMETER", "ROW_LIMIT_EXCEEDED"]},
	133	+)
	134	+def contract_bars(request: Request,
	135	+ symbol: str = Path(..., description="Contract symbol: `ESZ24`, `ESZ2024`, `ES_Z24`, `6EZ24` (aliases resolved)"),
	136	+ interval: str \| None = Query(None, description=_INTERVAL_DOC + " Default `1d`."),
	137	+ timeframe: str \| None = Query(None, description="Alias of `interval` (legacy names `1min`…`1day`)."),
	138	+ from_: str \| None = Query(None, alias="from", description=_FROM_DOC),
	139	+ to: str \| None = Query(None, description=_TO_DOC),
	140	+ session: str \| None = Query(None, description=_SESSION_DOC),
	141	+ cursor: str \| None = Query(None, description=_CURSOR_DOC),
	142	+ limit: int \| None = Query(None, description=_LIMIT_DOC),
	143	+ format: str = Query("json", description=_FORMAT_DOC)):
	144	+ fmt = parse_format(format)
	145	+ lim = _limit(limit, fmt, request)
	146	+ df, meta = service.contract_bars(symbol, interval, from_, to, session, cursor, lim, timeframe=timeframe)
	147	+ tf = service.parse_interval(interval, timeframe)
	148	+ return frame_response(service.format_for_output(df, tf, fmt), fmt, meta=meta, request=request, filename=f"{meta['symbol']}_{meta['interval']}")
	149	+
	150	+
	151	+@router.get(
	152	+ "/contract/{symbol}/coverage",
	153	+ summary="Coverage report of one contract",
	154	+ description="""What the lake really holds for a contract: per interval availability, first/last bar, row
	155	+count and files; expiration/last-trading/first-notice dates with their source (`rule` vs `data`); daily
	156	+gaps (> 3 business days without a bar) computed by the backfill; and explicit `notes` for every null.""",
	157	+ responses={200: {"description": "Coverage", "content": {"application/json": {"example": {"data": {
	158	+ "symbol": "CLZ24", "root": "CL", "status": "expired", "expiration_date": "2024-11-20", "expiration_source": "rule",
	159	+ "last_trading_date": "2024-11-20", "first_notice_date": "2024-11-21", "first_data_date": "2019-12-03", "last_data_date": "2024-11-20",
	160	+ "bars_1day": 1251, "volume_avg_daily": 402113.0, "open_interest_last": 8082.0,
	161	+ "intervals": {"1m": {"available": True, "first": "2020-01-02 09:00:00", "last": "2024-11-20 14:29:00", "rows": 1204411,
	162	+ "files": {"archive": "…/1min/archive/CL_Z24_1min.parquet"}, "open_interest": False},
	163	+ "1d": {"available": True, "first": "2019-12-03 00:00:00", "last": "2024-11-20 00:00:00", "rows": 1251,
	164	+ "files": {"archive": "…/1day/archive/CL_Z24_1day.parquet"}, "open_interest": True}},
	165	+ "gaps": [{"interval": "1d", "start": "2020-03-16", "end": "2020-03-27", "business_days_missing": 10}],
	166	+ "notes": [], "timezone_note": "intraday bars are converted from US/Eastern to UTC; daily bars are dates"},
	167	+ "meta": {"count": 1}}}}}},
	168	+ openapi_extra={"x-errors": ["INVALID_CONTRACT_SYMBOL", "CONTRACT_NOT_FOUND"]},
	169	+)
	170	+def contract_coverage(request: Request, symbol: str = Path(..., description="Contract symbol")):
	171	+ return json_response(service.coverage(symbol))
	172	+
	173	+
	174	+@router.get(
	175	+ "/{root}/chain",
	176	+ summary="Contract chain as of a date",
	177	+ description="""The contracts of a root that were live on `as_of` (data started, not yet expired), ordered by
	178	+expiration: position (1 = front), expiration/first-notice dates, days to expiry, and the last daily
	179	+close/volume/open interest at or before `as_of` (null if older than 7 days). Default `as_of` = today.""",
	180	+ responses={200: {"description": "Chain", "content": {"application/json": {"example": {"data": [
	181	+ {"position": 1, "symbol": "CLF25", "contract_month": 1, "contract_year": 2025, "expiration_date": "2024-12-19",
	182	+ "expiration_source": "rule", "first_notice_date": "2024-12-20", "days_to_expiry": 17, "status": "expired",
	183	+ "last_date": "2024-12-02", "close": 68.1, "volume": 312560.0, "open_interest": 301224.0, "volume_avg_daily": 289112.0},
	184	+ {"position": 2, "symbol": "CLG25", "contract_month": 2, "contract_year": 2025, "expiration_date": "2025-01-21",
	185	+ "expiration_source": "rule", "first_notice_date": "2025-01-22", "days_to_expiry": 50, "status": "expired",
	186	+ "last_date": "2024-12-02", "close": 67.9, "volume": 120014.0, "open_interest": 210998.0, "volume_avg_daily": 101223.0}],
	187	+ "meta": {"count": 2, "root": "CL", "as_of": "2024-12-02", "front": "CLF25", "note": None}}}}}},
	188	+ openapi_extra={"x-errors": ["ROOT_NOT_FOUND", "INVALID_PARAMETER"]},
	189	+)
	190	+def root_chain(request: Request, root: str = Path(..., description="Root code (lake or CME alias)"),
	191	+ as_of: str \| None = Query(None, description="YYYY-MM-DD (default today)"),
	192	+ format: str = Query("json", description=_FORMAT_DOC)):
	193	+ fmt = parse_format(format)
	194	+ rows, meta = service.chain(root, as_of)
	195	+ if fmt != "json":
	196	+ return frame_response(pd.DataFrame(rows), fmt, meta=meta, request=request, filename=f"{meta['root']}_chain_{meta['as_of']}")
	197	+ return json_response(rows, meta=meta)
	198	+
	199	+
	200	+@router.get(
	201	+ "/{root}/continuous",
	202	+ summary="Continuous series built from individual contracts",
	203	+ description="""A continuous front-month (or 2nd/3rd month, `depth`) series stitched from the individual contracts
	204	+with an explicit, reproducible roll schedule and optional price adjustment.
	205	+
	206	+Roll methods (`roll`): `calendar` (hold through the expiration date), `first_notice` (roll on the first
	207	+notice day, else calendar), `volume` (default — roll after 2 consecutive sessions where the next contract's
	208	+volume exceeds the front's), `open_interest` (same test on OI). Rolls are decided on daily data and applied to
	209	+intraday intervals by trading date.
	210	+
	211	+Adjustments (`adjust`): `none`, `back_adjusted` (additive — every bar before a roll is shifted by the
	212	+cumulative close gap `new − old`, latest contract unadjusted), `ratio_adjusted` (multiplicative — cumulative
	213	+close ratio `new / old`). Volume/OI are never adjusted. `meta.roll_dates` lists each roll in the returned window
	214	+(`date`, `from_symbol`, `to_symbol`, `gap`, `ratio`, `gap_session`, `adjusted`); a roll without a common
	215	+session has `gap: null` and is not adjusted (nothing is invented). Every row carries the source `symbol`.
	216	+
	217	+`request cost = 2` (multi-file scan).""",
	218	+ responses={200: {"description": "Continuous bars", "content": {"application/json": {"example": {"data": [
	219	+ {"symbol": "ESU24", "datetime": "2024-09-19", "open": 5670.0, "high": 5721.25, "low": 5657.0, "close": 5713.5, "volume": 1650211.0, "open_interest": 1650002.0},
	220	+ {"symbol": "ESZ24", "datetime": "2024-09-20", "open": 5766.5, "high": 5805.0, "low": 5750.25, "close": 5766.75, "volume": 1520114.0, "open_interest": 2090144.0}],
	221	+ "meta": {"count": 2, "root": "ES", "interval": "1d", "roll": "volume", "adjust": "back_adjusted", "depth": 1, "session": "all",
	222	+ "timezone": "UTC", "next_cursor": None,
	223	+ "roll_dates": [{"date": "2024-09-20", "from_symbol": "ESU24", "to_symbol": "ESZ24", "gap": 55.5, "ratio": 1.00971,
	224	+ "gap_session": "2024-09-19", "adjusted": True}],
	225	+ "rolls_total": 70, "segments": [{"symbol": "ESU24", "start": "2024-06-14", "end": "2024-09-19"}, {"symbol": "ESZ24", "start": "2024-09-20", "end": "2024-12-16"}],
	226	+ "unadjusted_symbol": "ESZ26", "adjustment_note": "additive: …"}}}}}},
	227	+ openapi_extra={"x-errors": ["ROOT_NOT_FOUND", "INVALID_PARAMETER", "ROW_LIMIT_EXCEEDED"]},
	228	+)
	229	+def root_continuous(request: Request, root: str = Path(..., description="Root code (lake or CME alias)"),
	230	+ roll: str \| None = Query(None, description="`calendar` \| `first_notice` \| `volume` (default) \| `open_interest`"),
	231	+ adjust: str \| None = Query(None, description="`none` (default) \| `back_adjusted` \| `ratio_adjusted`"),
	232	+ depth: int \| None = Query(None, description="1 = front month (default), 2 = second, 3 = third"),
	233	+ interval: str \| None = Query(None, description=_INTERVAL_DOC + " Default `1d`."),
	234	+ timeframe: str \| None = Query(None, description="Alias of `interval`."),
	235	+ from_: str \| None = Query(None, alias="from", description=_FROM_DOC),
	236	+ to: str \| None = Query(None, description=_TO_DOC),
	237	+ session: str \| None = Query(None, description=_SESSION_DOC),
	238	+ cursor: str \| None = Query(None, description=_CURSOR_DOC),
	239	+ limit: int \| None = Query(None, description=_LIMIT_DOC),
	240	+ format: str = Query("json", description=_FORMAT_DOC)):
	241	+ fmt = parse_format(format)
	242	+ lim = _limit(limit, fmt, request)
	243	+ request.state.request_cost = 2
	244	+ df, meta = service.continuous(root, roll, adjust, depth, interval, from_, to, session, cursor, lim, timeframe=timeframe)
	245	+ tf = service.parse_interval(interval, timeframe)
	246	+ return frame_response(service.format_for_output(df, tf, fmt), fmt, meta=meta, request=request,
	247	+ filename=f"{meta['root']}_continuous_{meta['roll']}_{meta['adjust']}_{meta['interval']}")
	248	+
	249	+
	250	+@router.get(
	251	+ "/{root}/term-structure",
	252	+ summary="Term structure (forward curve) as of a date",
	253	+ description="""Forward curve of a root on `as_of`: every live contract with its settle (last daily close at or
	254	+before `as_of`, ≤ 7 days old, else null), days to expiry, spread vs the front and the annualised slope
	255	+`(settle_i / settle_front − 1) × 365 / (dte_i − dte_front)`. `meta.structure` flags `contango`
	256	+(2nd > front), `backwardation` or `flat`. Default `as_of` = today.""",
	257	+ responses={200: {"description": "Term structure", "content": {"application/json": {"example": {"data": [
	258	+ {"position": 1, "symbol": "CLF25", "contract_month": 1, "contract_year": 2025, "expiration_date": "2024-12-19", "expiration_source": "rule",
	259	+ "first_notice_date": "2024-12-20", "days_to_expiry": 17, "status": "expired", "last_date": "2024-12-02", "settle": 68.1,
	260	+ "volume": 312560.0, "open_interest": 301224.0, "slope_annualized": None, "spread_vs_front": None},
	261	+ {"position": 2, "symbol": "CLG25", "contract_month": 2, "contract_year": 2025, "expiration_date": "2025-01-21", "expiration_source": "rule",
	262	+ "first_notice_date": "2025-01-22", "days_to_expiry": 50, "status": "expired", "last_date": "2024-12-02", "settle": 67.9,
	263	+ "volume": 120014.0, "open_interest": 210998.0, "slope_annualized": -0.03248, "spread_vs_front": -0.2}],
	264	+ "meta": {"count": 2, "root": "CL", "as_of": "2024-12-02", "front": "CLF25", "front_settle": 68.1, "structure": "backwardation",
	265	+ "curve_slope_annualized": -0.03248, "priced_contracts": 2,
	266	+ "settle_note": "settle = last daily close at or before as_of (≤ 7 days old), else null"}}}}}},
	267	+ openapi_extra={"x-errors": ["ROOT_NOT_FOUND", "INVALID_PARAMETER"]},
	268	+)
	269	+def root_term_structure(request: Request, root: str = Path(..., description="Root code (lake or CME alias)"),
	270	+ as_of: str \| None = Query(None, description="YYYY-MM-DD (default today)"),
	271	+ format: str = Query("json", description=_FORMAT_DOC)):
	272	+ fmt = parse_format(format)
	273	+ rows, meta = service.term_structure(root, as_of)
	274	+ if fmt != "json":
	275	+ return frame_response(pd.DataFrame(rows), fmt, meta=meta, request=request, filename=f"{meta['root']}_term_structure_{meta['as_of']}")
	276	+ return json_response(rows, meta=meta)

added hfmarketdata/api/futures/service.py +519 −0

		`@@ -0,0 +1,519 @@`
	1	+"""Data access for the futures module: contract bars, chains, continuous series, term structure, coverage.
	2	+
	3	+Conventions
	4	+* Lake timestamps are naive US/Eastern; intraday output is converted to UTC (`zoneinfo`, DST-aware);
	5	+ daily bars stay calendar dates.
	6	+* `from`/`to`: `YYYY-MM-DD` = whole Eastern trading dates; ISO datetimes with an offset/`Z` are converted
	7	+ to Eastern; naive datetimes are treated as UTC.
	8	+* `session`: `rth` keeps bars whose start time falls in the root's RTH window (Eastern, see specs);
	9	+ `eth` keeps the complement; `all` (default) keeps everything. Ignored for daily bars.
	10	+* Cursor pagination on the (Eastern) bar datetime; `limit + 1` rows are fetched to detect `next_cursor`.
	11	+
	12	+Author: Simon-Pierre Boucher <contact@spboucher.ai>
	13	+"""
	14	+from __future__ import annotations
	15	+
	16	+import json
	17	+import re
	18	+from datetime import date, datetime, timedelta, timezone
	19	+from typing import Any
	20	+from zoneinfo import ZoneInfo
	21	+
	22	+import numpy as np
	23	+import pandas as pd
	24	+from core.db import session
	25	+from core.duck import cached, con
	26	+from core.errors import ApiError
	27	+from core.responses import decode_cursor, encode_cursor
	28	+from sqlalchemy import select
	29	+
	30	+from . import lake
	31	+from .lake import INTERVAL_ALIASES, INTERVAL_OF_TF, TIMEFRAMES, files_for, merged_sql
	32	+from .models import FuturesContract, FuturesContractGap, FuturesRoot
	33	+from .rolls import (
	34	+ ADJUSTMENTS,
	35	+ ROLL_METHODS,
	36	+ Segment,
	37	+ adjustment_offsets,
	38	+ apply_adjustment,
	39	+ depth_schedule,
	40	+ roll_gaps,
	41	+ roll_schedule,
	42	+)
	43	+from .specs import spec_for
	44	+from .symbols import ContractSymbol, normalize_root, parse_symbol
	45	+
	46	+ET = ZoneInfo("America/New_York")
	47	+UTC = timezone.utc
	48	+SESSIONS = ("rth", "eth", "all")
	49	+BAR_COLS = ["datetime", "open", "high", "low", "close", "volume"]
	50	+_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
	51	+
	52	+
	53	+# ---- parameter parsing ---------------------------------------------------------------------------------
	54	+
	55	+def parse_interval(interval: str \| None, timeframe: str \| None = None) -> str:
	56	+ v = (interval or timeframe or "1d").lower()
	57	+ tf = INTERVAL_ALIASES.get(v)
	58	+ if not tf:
	59	+ raise ApiError(400, "INVALID_PARAMETER", f"interval must be one of 1m, 5m, 30m, 1h, 1d (got '{v}')",
	60	+ details={"interval": v})
	61	+ return tf
	62	+
	63	+
	64	+def parse_session(session_: str \| None) -> str:
	65	+ v = (session_ or "all").lower()
	66	+ if v not in SESSIONS:
	67	+ raise ApiError(400, "INVALID_PARAMETER", "session must be rth, eth or all", details={"session": v})
	68	+ return v
	69	+
	70	+
	71	+def parse_date(value: str \| None, name: str) -> date \| None:
	72	+ if value in (None, ""):
	73	+ return None
	74	+ try:
	75	+ return date.fromisoformat(value[:10]) if _DATE_RE.match(value.strip()) else datetime.fromisoformat(value).date()
	76	+ except ValueError:
	77	+ raise ApiError(400, "INVALID_PARAMETER", f"{name} must be an ISO date (YYYY-MM-DD)", details={name: value})
	78	+
	79	+
	80	+def parse_bound(value: str \| None, name: str, end: bool = False) -> datetime \| None:
	81	+ """`from`/`to` → naive Eastern datetime bound (inclusive lower / exclusive upper)."""
	82	+ if value in (None, ""):
	83	+ return None
	84	+ v = value.strip()
	85	+ try:
	86	+ if _DATE_RE.match(v):
	87	+ d = datetime.fromisoformat(v)
	88	+ return d + timedelta(days=1) if end else d
	89	+ dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
	90	+ except ValueError:
	91	+ raise ApiError(400, "INVALID_PARAMETER", f"{name} must be an ISO 8601 date or datetime", details={name: value})
	92	+ if dt.tzinfo is None:
	93	+ dt = dt.replace(tzinfo=UTC)
	94	+ return dt.astimezone(ET).replace(tzinfo=None)
	95	+
	96	+
	97	+def parse_choice(value: str \| None, name: str, choices: tuple[str, ...], default: str) -> str:
	98	+ v = (value or default).lower()
	99	+ if v not in choices:
	100	+ raise ApiError(400, "INVALID_PARAMETER", f"{name} must be one of {', '.join(choices)}", details={name: v})
	101	+ return v
	102	+
	103	+
	104	+def parse_depth(value: int \| None) -> int:
	105	+ d = 1 if value is None else int(value)
	106	+ if not 1 <= d <= 3:
	107	+ raise ApiError(400, "INVALID_PARAMETER", "depth must be 1, 2 or 3", details={"depth": value})
	108	+ return d
	109	+
	110	+
	111	+# ---- lookups --------------------------------------------------------------------------------------------
	112	+
	113	+def _row(obj) -> dict[str, Any]:
	114	+ """ORM row → plain dict (dates kept as `date` objects; JSON columns decoded)."""
	115	+ d = {c.name: getattr(obj, c.name) for c in obj.__table__.columns}
	116	+ for k in ("timeframes", "files", "aliases"):
	117	+ if k in d and isinstance(d[k], str):
	118	+ try:
	119	+ d[k] = json.loads(d[k])
	120	+ except ValueError:
	121	+ pass
	122	+ return d
	123	+
	124	+
	125	+def jsonable(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
	126	+ """ISO strings for date/datetime values (JSONResponse does not serialise them)."""
	127	+ return [{k: (v.isoformat() if isinstance(v, (date, datetime)) else v) for k, v in r.items()} for r in rows]
	128	+
	129	+
	130	+def resolve_root(root: str) -> str:
	131	+ """Canonical lake root or 404 ROOT_NOT_FOUND."""
	132	+ r = normalize_root(root)
	133	+ if not re.fullmatch(r"[A-Z0-9]{1,4}", r or ""):
	134	+ raise ApiError(404, "ROOT_NOT_FOUND", f"Unknown futures root '{root}'.", details={"root": root})
	135	+ with session() as s:
	136	+ row = s.get(FuturesRoot, r)
	137	+ if row is None and not lake.symbols_of_root(r):
	138	+ raise ApiError(404, "ROOT_NOT_FOUND", f"Unknown futures root '{root}'. See /v1/futures/roots.", details={"root": root})
	139	+ return r
	140	+
	141	+
	142	+def get_contract(symbol: str) -> tuple[ContractSymbol, dict[str, Any]]:
	143	+ """Parse + load the contract row; 404 CONTRACT_NOT_FOUND if neither the DB nor the lake knows it."""
	144	+ cs = parse_symbol(symbol)
	145	+ with session() as s:
	146	+ row = s.get(FuturesContract, cs.short)
	147	+ if row is None:
	148	+ idx = lake.file_index().get(cs.short)
	149	+ if not idx:
	150	+ raise ApiError(404, "CONTRACT_NOT_FOUND", f"No data for contract {cs.short}.",
	151	+ details={"symbol": cs.short, "root": cs.root})
	152	+ # lake has files but the metadata backfill has not run yet: minimal on-the-fly row
	153	+ return cs, {"symbol": cs.short, "root": cs.root, "month_code": cs.month_code, "contract_month": cs.month,
	154	+ "contract_year": cs.year, "timeframes": {tf: {} for tf in idx}, "files": None, "status": None,
	155	+ "expiration_date": None, "expiration_source": None}
	156	+ return cs, _row(row)
	157	+
	158	+
	159	+def roots() -> list[dict[str, Any]]:
	160	+ with session() as s:
	161	+ rows = s.execute(select(FuturesRoot).order_by(FuturesRoot.root)).scalars().all()
	162	+ return [_row(r) for r in rows]
	163	+
	164	+
	165	+def contracts(root: str, status: str \| None = None, from_: date \| None = None, to: date \| None = None,
	166	+ sort: str = "expiration_date") -> list[dict[str, Any]]:
	167	+ r = resolve_root(root)
	168	+ if status and status not in ("active", "expired"):
	169	+ raise ApiError(400, "INVALID_PARAMETER", "status must be active or expired", details={"status": status})
	170	+ sort_cols = {"expiration_date": FuturesContract.expiration_date, "symbol": FuturesContract.symbol,
	171	+ "first_data_date": FuturesContract.first_data_date, "last_data_date": FuturesContract.last_data_date,
	172	+ "volume_avg_daily": FuturesContract.volume_avg_daily, "open_interest_last": FuturesContract.open_interest_last}
	173	+ desc = sort.startswith("-")
	174	+ key = sort.lstrip("-")
	175	+ if key not in sort_cols:
	176	+ raise ApiError(400, "INVALID_PARAMETER", f"sort must be one of {', '.join(sort_cols)} (prefix - for descending)",
	177	+ details={"sort": sort})
	178	+ q = select(FuturesContract).where(FuturesContract.root == r)
	179	+ if status:
	180	+ q = q.where(FuturesContract.status == status)
	181	+ if from_:
	182	+ q = q.where(FuturesContract.expiration_date >= from_)
	183	+ if to:
	184	+ q = q.where(FuturesContract.expiration_date <= to)
	185	+ col = sort_cols[key]
	186	+ q = q.order_by(col.desc() if desc else col.asc(), FuturesContract.symbol)
	187	+ with session() as s:
	188	+ rows = s.execute(q).scalars().all()
	189	+ return [_row(x) for x in rows]
	190	+
	191	+
	192	+def _root_contracts(root: str) -> list[dict[str, Any]]:
	193	+ with session() as s:
	194	+ rows = s.execute(select(FuturesContract).where(FuturesContract.root == root)
	195	+ .order_by(FuturesContract.expiration_date, FuturesContract.symbol)).scalars().all()
	196	+ return [_row(x) for x in rows]
	197	+
	198	+
	199	+# ---- bars ------------------------------------------------------------------------------------------------
	200	+
	201	+def _session_sql(root: str, session_: str, tf: str) -> str:
	202	+ if session_ == "all" or tf == "1day":
	203	+ return ""
	204	+ start, end = spec_for(root).rth
	205	+ if start < end:
	206	+ cond = f"(CAST(datetime AS TIME) >= TIME '{start}' AND CAST(datetime AS TIME) < TIME '{end}')"
	207	+ else: # overnight window (e.g. cotton 21:00–14:20)
	208	+ cond = f"(CAST(datetime AS TIME) >= TIME '{start}' OR CAST(datetime AS TIME) < TIME '{end}')"
	209	+ return f" AND {cond}" if session_ == "rth" else f" AND NOT {cond}"
	210	+
	211	+
	212	+def to_utc(df: pd.DataFrame, tf: str, col: str = "datetime") -> pd.DataFrame:
	213	+ """Naive Eastern → tz-aware UTC for intraday frames (daily frames stay naive dates)."""
	214	+ if tf == "1day" or df.empty or col not in df:
	215	+ return df
	216	+ s = pd.to_datetime(df[col])
	217	+ if s.dt.tz is None:
	218	+ s = s.dt.tz_localize(ET, ambiguous=np.zeros(len(s), dtype=bool), nonexistent="shift_forward")
	219	+ df = df.copy()
	220	+ # ns resolution: DuckDB yields µs and core.responses only formats `datetime64[ns, UTC]` with the `Z` suffix
	221	+ df[col] = s.dt.tz_convert("UTC").astype("datetime64[ns, UTC]")
	222	+ return df
	223	+
	224	+
	225	+def format_for_output(df: pd.DataFrame, tf: str, fmt: str) -> pd.DataFrame:
	226	+ """CSV gets ISO strings (UTC `Z` / plain dates); JSON is handled by frame_response; Parquet keeps types."""
	227	+ if fmt != "csv" or df.empty or "datetime" not in df:
	228	+ return df
	229	+ df = df.copy()
	230	+ if tf == "1day":
	231	+ df["datetime"] = pd.to_datetime(df["datetime"]).dt.strftime("%Y-%m-%d")
	232	+ else:
	233	+ df["datetime"] = df["datetime"].dt.strftime("%Y-%m-%dT%H:%M:%SZ")
	234	+ return df
	235	+
	236	+
	237	+def contract_bars(symbol: str, interval: str \| None, from_: str \| None, to: str \| None, session_: str \| None,
	238	+ cursor: str \| None, limit: int, timeframe: str \| None = None) -> tuple[pd.DataFrame, dict[str, Any]]:
	239	+ tf = parse_interval(interval, timeframe)
	240	+ sess = parse_session(session_)
	241	+ lo, hi = parse_bound(from_, "from"), parse_bound(to, "to", end=True)
	242	+ cs, row = get_contract(symbol)
	243	+ paths = files_for(cs, tf)
	244	+ if not paths:
	245	+ avail = sorted(lake.file_index().get(cs.short, {}), key=TIMEFRAMES.index)
	246	+ raise ApiError(404, "CONTRACT_NOT_FOUND",
	247	+ f"{cs.short} has no {INTERVAL_OF_TF[tf]} bars. Available intervals: "
	248	+ f"{', '.join(INTERVAL_OF_TF[t] for t in avail) or 'none'}.",
	249	+ details={"symbol": cs.short, "interval": INTERVAL_OF_TF[tf],
	250	+ "available_intervals": [INTERVAL_OF_TF[t] for t in avail]})
	251	+ after = decode_cursor(cursor)
	252	+ conds, params = [], list(paths)
	253	+ if lo is not None:
	254	+ conds.append("datetime >= ?"); params.append(lo)
	255	+ if hi is not None:
	256	+ conds.append("datetime < ?"); params.append(hi)
	257	+ if after is not None:
	258	+ conds.append("datetime > ?"); params.append(str(after))
	259	+ where = (" WHERE " + " AND ".join(conds)) if conds else " WHERE 1=1"
	260	+ cols = "* EXCLUDE (ticker)"
	261	+ sql = f"SELECT {cols} FROM ({merged_sql(paths)}){where}{_session_sql(cs.root, sess, tf)} ORDER BY datetime LIMIT ?"
	262	+ params.append(limit + 1)
	263	+ df = con().execute(sql, params).df()
	264	+ has_more = len(df) > limit
	265	+ df = df.iloc[:limit]
	266	+ next_cursor = encode_cursor(str(df["datetime"].iloc[-1])) if has_more and not df.empty else None
	267	+ meta = {"symbol": cs.short, "root": cs.root, "interval": INTERVAL_OF_TF[tf], "session": sess, "timezone": "UTC",
	268	+ "next_cursor": next_cursor, "expiration_date": str(row.get("expiration_date")) if row.get("expiration_date") else None,
	269	+ "expiration_source": row.get("expiration_source"), "status": row.get("status")}
	270	+ df.insert(0, "symbol", cs.short)
	271	+ return to_utc(df, tf), meta
	272	+
	273	+
	274	+# ---- daily cache per root (chains, rolls, term structure) ----------------------------------------------------
	275	+
	276	+def root_daily(root: str) -> pd.DataFrame:
	277	+ """Merged daily bars of every contract of a root: symbol, date, open, high, low, close, volume, open_interest."""
	278	+ def build() -> pd.DataFrame:
	279	+ paths: list[str] = []
	280	+ for sym in lake.symbols_of_root(root):
	281	+ paths.extend(files_for(sym, "1day"))
	282	+ if not paths:
	283	+ return pd.DataFrame(columns=["symbol", "date", "open", "high", "low", "close", "volume", "open_interest"])
	284	+ sql = r"""
	285	+ WITH raw AS (
	286	+ SELECT replace(regexp_extract(filename, '([A-Z0-9]+_[FGHJKMNQUVXZ][0-9]{2})_1day\.parquet$', 1), '_', '') AS symbol,
	287	+ CASE WHEN filename LIKE '%/update/%' THEN 0 ELSE 1 END AS prio,
	288	+ datetime, open, high, low, close, volume, open_interest
	289	+ FROM read_parquet(?, filename=true, union_by_name=true)
	290	+ )
	291	+ SELECT symbol, CAST(datetime AS DATE) AS date, open, high, low, close, volume, open_interest FROM raw
	292	+ QUALIFY row_number() OVER (PARTITION BY symbol, datetime ORDER BY prio) = 1
	293	+ ORDER BY symbol, datetime
	294	+ """
	295	+ df = con().execute(sql, [paths]).df()
	296	+ df["date"] = pd.to_datetime(df["date"])
	297	+ return df
	298	+ return cached(f"futures\|daily\|{root}", build)
	299	+
	300	+
	301	+def _as_of(value: str \| None) -> date:
	302	+ return parse_date(value, "as_of") or date.today()
	303	+
	304	+
	305	+def _live_contracts(root: str, as_of: date) -> list[dict[str, Any]]:
	306	+ """Contracts with data started on/before as_of and not yet expired at as_of (expiration ≥ as_of)."""
	307	+ out = []
	308	+ for c in _root_contracts(root):
	309	+ fd, ex = c.get("first_data_date"), c.get("expiration_date")
	310	+ if fd is None or ex is None:
	311	+ continue
	312	+ if fd <= as_of <= ex:
	313	+ out.append(c)
	314	+ out.sort(key=lambda c: (c["expiration_date"], c["symbol"]))
	315	+ return out
	316	+
	317	+
	318	+def _last_bar_as_of(daily: pd.DataFrame, symbol: str, as_of: date, max_age_days: int = 7) -> dict[str, Any] \| None:
	319	+ d = daily[(daily["symbol"] == symbol) & (daily["date"] <= pd.Timestamp(as_of))]
	320	+ if d.empty:
	321	+ return None
	322	+ last = d.iloc[-1]
	323	+ if (pd.Timestamp(as_of) - last["date"]).days > max_age_days:
	324	+ return None
	325	+ return {"date": last["date"].date(), "close": float(last["close"]), "volume": float(last["volume"]) if pd.notna(last["volume"]) else None,
	326	+ "open_interest": float(last["open_interest"]) if pd.notna(last.get("open_interest")) else None}
	327	+
	328	+
	329	+def chain(root: str, as_of: str \| None) -> tuple[list[dict[str, Any]], dict[str, Any]]:
	330	+ r = resolve_root(root)
	331	+ asof = _as_of(as_of)
	332	+ live = _live_contracts(r, asof)
	333	+ daily = root_daily(r)
	334	+ rows = []
	335	+ for i, c in enumerate(live, start=1):
	336	+ bar = _last_bar_as_of(daily, c["symbol"], asof)
	337	+ rows.append({"position": i, "symbol": c["symbol"], "contract_month": c["contract_month"], "contract_year": c["contract_year"],
	338	+ "expiration_date": str(c["expiration_date"]), "expiration_source": c["expiration_source"],
	339	+ "first_notice_date": str(c["first_notice_date"]) if c.get("first_notice_date") else None,
	340	+ "days_to_expiry": (c["expiration_date"] - asof).days, "status": c["status"],
	341	+ "last_date": str(bar["date"]) if bar else None, "close": bar["close"] if bar else None,
	342	+ "volume": bar["volume"] if bar else None, "open_interest": bar["open_interest"] if bar else None,
	343	+ "volume_avg_daily": c.get("volume_avg_daily")})
	344	+ meta = {"root": r, "as_of": str(asof), "front": rows[0]["symbol"] if rows else None,
	345	+ "note": None if rows else "no contract live at as_of (before first data or all expired)"}
	346	+ return rows, meta
	347	+
	348	+
	349	+def term_structure(root: str, as_of: str \| None) -> tuple[list[dict[str, Any]], dict[str, Any]]:
	350	+ rows, meta = chain(root, as_of)
	351	+ asof = date.fromisoformat(meta["as_of"])
	352	+ priced = [x for x in rows if x["close"] is not None]
	353	+ front = priced[0] if priced else None
	354	+ for x in rows:
	355	+ x["settle"] = x.pop("close")
	356	+ x["slope_annualized"] = None
	357	+ x["spread_vs_front"] = None
	358	+ if front and x["settle"] is not None and x is not front:
	359	+ dte_gap = x["days_to_expiry"] - front["days_to_expiry"]
	360	+ x["spread_vs_front"] = round(x["settle"] - front["settle"], 10)
	361	+ if dte_gap > 0 and front["settle"]:
	362	+ x["slope_annualized"] = round((x["settle"] / front["settle"] - 1.0) * 365.0 / dte_gap, 8)
	363	+ x.pop("volume_avg_daily", None)
	364	+ structure = None
	365	+ if len(priced) >= 2:
	366	+ structure = "contango" if priced[1]["settle"] > priced[0]["settle"] else ("backwardation" if priced[1]["settle"] < priced[0]["settle"] else "flat")
	367	+ last = priced[-1] if len(priced) >= 2 else None
	368	+ meta.update({"structure": structure, "front": front["symbol"] if front else None,
	369	+ "front_settle": front["settle"] if front else None,
	370	+ "curve_slope_annualized": last["slope_annualized"] if last else None,
	371	+ "priced_contracts": len(priced), "settle_note": "settle = last daily close at or before as_of (≤ 7 days old), else null",
	372	+ "as_of": str(asof)})
	373	+ return rows, meta
	374	+
	375	+
	376	+# ---- continuous -------------------------------------------------------------------------------------------
	377	+
	378	+def _schedule(root: str, roll: str, depth: int) -> tuple[list[Segment], list[Segment], list[dict[str, Any]], list[dict[str, Any]]]:
	379	+ """(front segments, depth segments, contracts, roll_dates with gaps) — cached per root/roll/depth."""
	380	+ def build():
	381	+ contracts_ = _root_contracts(root)
	382	+ daily = root_daily(root)
	383	+ spec = spec_for(root)
	384	+ front = roll_schedule(contracts_, daily, roll, spec.calendar)
	385	+ segs = depth_schedule(front, contracts_, depth)
	386	+ rolls = roll_gaps(segs, daily)
	387	+ return front, segs, contracts_, rolls
	388	+ return cached(f"futures\|schedule\|{root}\|{roll}\|{depth}", build)
	389	+
	390	+
	391	+def continuous(root: str, roll: str \| None, adjust: str \| None, depth: int \| None, interval: str \| None,
	392	+ from_: str \| None, to: str \| None, session_: str \| None, cursor: str \| None, limit: int,
	393	+ timeframe: str \| None = None) -> tuple[pd.DataFrame, dict[str, Any]]:
	394	+ r = resolve_root(root)
	395	+ tf = parse_interval(interval, timeframe)
	396	+ sess = parse_session(session_)
	397	+ roll_m = parse_choice(roll, "roll", ROLL_METHODS, "volume")
	398	+ adj = parse_choice(adjust, "adjust", ADJUSTMENTS, "none")
	399	+ dep = parse_depth(depth)
	400	+ lo, hi = parse_bound(from_, "from"), parse_bound(to, "to", end=True)
	401	+ after = decode_cursor(cursor)
	402	+ _front, segs, _contracts, rolls = _schedule(r, roll_m, dep)
	403	+ if not segs:
	404	+ raise ApiError(404, "ROOT_NOT_FOUND", f"No contracts with metadata for root {r}. Run the contracts backfill.",
	405	+ details={"root": r})
	406	+ add, mul = adjustment_offsets(rolls, adj)
	407	+ # segments intersecting the requested window
	408	+ lo_d = lo.date() if lo else None
	409	+ hi_d = (hi - timedelta(microseconds=1)).date() if hi else None
	410	+ after_dt = pd.Timestamp(after) if after is not None else None
	411	+ frames: list[pd.DataFrame \| None] = [None] * len(segs)
	412	+ if tf == "1day":
	413	+ daily = root_daily(r)
	414	+ for i, s in enumerate(segs):
	415	+ if (hi_d and s.start > hi_d) or (lo_d and s.end is not None and s.end < lo_d):
	416	+ continue
	417	+ f = daily[daily["symbol"] == s.symbol]
	418	+ f = f[f["date"] >= pd.Timestamp(s.start)]
	419	+ if s.end is not None:
	420	+ f = f[f["date"] <= pd.Timestamp(s.end)]
	421	+ if lo_d:
	422	+ f = f[f["date"] >= pd.Timestamp(lo_d)]
	423	+ if hi_d:
	424	+ f = f[f["date"] <= pd.Timestamp(hi_d)]
	425	+ if after_dt is not None:
	426	+ f = f[f["date"] > after_dt]
	427	+ frames[i] = f.rename(columns={"date": "datetime"})
	428	+ else:
	429	+ parts, params = [], []
	430	+ sess_sql = _session_sql(r, sess, tf)
	431	+ for i, s in enumerate(segs):
	432	+ if (hi_d and s.start > hi_d) or (lo_d and s.end is not None and s.end < lo_d):
	433	+ continue
	434	+ paths = files_for(s.symbol, tf)
	435	+ if not paths:
	436	+ continue
	437	+ seg_lo = datetime.combine(max(s.start, lo_d) if lo_d else s.start, datetime.min.time())
	438	+ seg_hi = None
	439	+ if s.end is not None or hi_d:
	440	+ e = s.end if hi_d is None else (hi_d if s.end is None else min(s.end, hi_d))
	441	+ seg_hi = datetime.combine(e + timedelta(days=1), datetime.min.time())
	442	+ conds = ["datetime >= ?"]
	443	+ p = [*paths, seg_lo]
	444	+ if seg_hi is not None:
	445	+ conds.append("datetime < ?"); p.append(seg_hi)
	446	+ if after is not None:
	447	+ conds.append("datetime > ?"); p.append(str(after))
	448	+ parts.append(f"SELECT {i} AS _seg, '{s.symbol}' AS symbol, datetime, open, high, low, close, volume "
	449	+ f"FROM ({merged_sql(paths)}) WHERE {' AND '.join(conds)}{sess_sql}")
	450	+ params.extend(p)
	451	+ if parts:
	452	+ sql = " UNION ALL ".join(parts) + " ORDER BY datetime LIMIT ?"
	453	+ params.append(limit + 1)
	454	+ df_all = con().execute(sql, params).df()
	455	+ for seg_idx, g in df_all.groupby("_seg", sort=False):
	456	+ frames[int(seg_idx)] = g.drop(columns="_seg")
	457	+ df = apply_adjustment(frames, add, mul)
	458	+ if "symbol" in df.columns:
	459	+ df = df[["symbol", *[c for c in df.columns if c != "symbol"]]]
	460	+ has_more = len(df) > limit
	461	+ df = df.iloc[:limit]
	462	+ next_cursor = encode_cursor(str(df["datetime"].iloc[-1])) if has_more and not df.empty else None
	463	+ # roll dates: those falling inside the returned window (plus adjustment info)
	464	+ if not df.empty:
	465	+ first_d, last_d = pd.Timestamp(df["datetime"].iloc[0]).date(), pd.Timestamp(df["datetime"].iloc[-1]).date()
	466	+ shown = [x for x in rolls if first_d <= date.fromisoformat(x["date"]) <= last_d + timedelta(days=1)]
	467	+ else:
	468	+ shown = []
	469	+ meta = {"root": r, "interval": INTERVAL_OF_TF[tf], "roll": roll_m, "adjust": adj, "depth": dep, "session": sess,
	470	+ "timezone": "UTC", "next_cursor": next_cursor, "roll_dates": shown, "rolls_total": len(rolls),
	471	+ "segments": [{"symbol": s.symbol, "start": str(s.start), "end": str(s.end) if s.end else None} for s in segs
	472	+ if (not df.empty and (s.end is None or s.end >= first_d) and s.start <= last_d)],
	473	+ "unadjusted_symbol": segs[-1].symbol,
	474	+ "adjustment_note": {"none": None,
	475	+ "back_adjusted": "additive: bars before each roll shifted by the cumulative close gap (new − old) so the latest contract is unadjusted",
	476	+ "ratio_adjusted": "multiplicative: bars before each roll scaled by the cumulative close ratio (new / old)"}[adj]}
	477	+ return to_utc(df, tf), meta
	478	+
	479	+
	480	+# ---- coverage ----------------------------------------------------------------------------------------------
	481	+
	482	+def coverage(symbol: str) -> dict[str, Any]:
	483	+ cs, row = get_contract(symbol)
	484	+ idx = lake.file_index().get(cs.short, {})
	485	+ tfs = row.get("timeframes") or {}
	486	+ per_tf = {}
	487	+ for tf in TIMEFRAMES:
	488	+ info = tfs.get(tf) if isinstance(tfs, dict) else None
	489	+ files = idx.get(tf, {})
	490	+ if not files:
	491	+ per_tf[INTERVAL_OF_TF[tf]] = {"available": False, "reason": "no file in the lake for this timeframe"}
	492	+ continue
	493	+ per_tf[INTERVAL_OF_TF[tf]] = {"available": True, "first": (info or {}).get("first"), "last": (info or {}).get("last"),
	494	+ "rows": (info or {}).get("rows"),
	495	+ "files": {b: p for b, p in files.items()},
	496	+ "open_interest": tf == "1day"}
	497	+ with session() as s:
	498	+ gaps = s.execute(select(FuturesContractGap).where(FuturesContractGap.symbol == cs.short)
	499	+ .order_by(FuturesContractGap.gap_start)).scalars().all()
	500	+ notes = []
	501	+ if row.get("expiration_source") == "data":
	502	+ notes.append("expiration_date = last data date (no exchange rule for this root)")
	503	+ if row.get("first_notice_date") is None and row.get("settlement_type") == "physical":
	504	+ notes.append("first_notice_date unknown for this root (null)")
	505	+ if row.get("files") is None:
	506	+ notes.append("metadata backfill not run for this contract: only lake files are reported")
	507	+ return {"symbol": cs.short, "root": cs.root, "status": row.get("status"),
	508	+ "expiration_date": str(row["expiration_date"]) if row.get("expiration_date") else None,
	509	+ "expiration_source": row.get("expiration_source"),
	510	+ "last_trading_date": str(row["last_trading_date"]) if row.get("last_trading_date") else None,
	511	+ "first_notice_date": str(row["first_notice_date"]) if row.get("first_notice_date") else None,
	512	+ "first_data_date": str(row["first_data_date"]) if row.get("first_data_date") else None,
	513	+ "last_data_date": str(row["last_data_date"]) if row.get("last_data_date") else None,
	514	+ "bars_1day": row.get("bars_1day"), "volume_avg_daily": row.get("volume_avg_daily"),
	515	+ "open_interest_last": row.get("open_interest_last"),
	516	+ "intervals": per_tf,
	517	+ "gaps": [{"interval": INTERVAL_OF_TF.get(g.timeframe, g.timeframe), "start": str(g.gap_start), "end": str(g.gap_end),
	518	+ "business_days_missing": g.bars_missing} for g in gaps],
	519	+ "notes": notes, "timezone_note": "intraday bars are converted from US/Eastern to UTC; daily bars are dates"}
	520