futures: filtre à l'intérieur de la source fusionnée + CTE matérialisée (un seul scan par fichier)

`merged_sql(paths, where=…)` applique la condition dans l'union archive/update avant la fenêtre de dédup et matérialise
le résultat (core.duck.topn_sql) ; contract_bars, continuous et root_daily n'émettent plus de QUALIFY nu suivi d'un
LIMIT. Test de plan : EXPLAIN sur le lac synthétique = un READ_PARQUET filtré par fichier, sans HASH_JOIN.

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

Simon-Pierre Boucher committed 18 days ago (Sep 6, 2026) parent b11789f

3 changed files +132 −22

modified hfmarketdata/api/futures/lake.py +12 −6

		`@@ -12,7 +12,7 @@ import os`
12	12	from pathlib import Path
13	13
14	14	from core.config import settings
15		−from core.duck import cached
	15	+from core.duck import cached, topn_sql
16	16
17	17	from .symbols import ContractSymbol, symbol_from_file_stem
18	18
		`@@ -62,14 +62,20 @@ def files_for(symbol: ContractSymbol \| str, tf: str) -> list[str]:`
62	62	return [b[k] for k in BUCKETS if k in b]
63	63
64	64
65		−def merged_sql(paths: list[str], columns: str = "*") -> str:
	65	+def merged_sql(paths: list[str], columns: str = "*", where: str = "") -> str:
66	66	"""SQL text selecting the merged bars of one contract (dedup on datetime, update wins).
67		− `paths` is ordered by priority (index 0 wins). Uses positional `?` placeholders → bind `paths`."""
	67	+ `paths` is ordered by priority (index 0 wins). Uses positional `?` placeholders → bind `paths`, then the
	68	+ parameters of `where` (a bare condition, no `WHERE` keyword).
	69	+
	70	+ The filter is applied INSIDE the union, before the dedup window, and the filtered union is materialized
	71	+ (`core.duck.topn_sql`): a `QUALIFY` over `read_parquet` followed by a `LIMIT` would otherwise be rewritten
	72	+ into a second, unfiltered scan of the files (late materialization)."""
	73	+ w = f" WHERE {where}" if where else ""
68	74	if len(paths) == 1:
69		− return f"SELECT {columns} FROM read_parquet(?)"
	75	+ return f"SELECT {columns} FROM read_parquet(?){w}"
70	76	parts = " UNION ALL BY NAME ".join(f"SELECT *, {i} AS _src FROM read_parquet(?)" for i in range(len(paths)))
71		− return (f"SELECT {columns} FROM (SELECT * EXCLUDE (_src) FROM ({parts}) "
72		− "QUALIFY row_number() OVER (PARTITION BY datetime ORDER BY _src) = 1)")
	77	+ return topn_sql(source=f"({parts})", where=where, partition="datetime", order="_src", n=1,
	78	+ select=f"{columns} EXCLUDE (_src)" if columns == "*" else columns)
73	79
74	80
75	81	def symbols_of_root(root: str) -> list[str]:

modified hfmarketdata/api/futures/service.py +12 −16

		`@@ -22,7 +22,7 @@ from zoneinfo import ZoneInfo`
22	22	import numpy as np
23	23	import pandas as pd
24	24	from core.db import session
25		−from core.duck import cached, con
	25	+from core.duck import cached, con, topn_sql
26	26	from core.errors import ApiError
27	27	from core.responses import decode_cursor, encode_cursor
28	28	from sqlalchemy import select
		`@@ -256,9 +256,9 @@ def contract_bars(symbol: str, interval: str \| None, from_: str \| None, to: str`
256	256	conds.append("datetime < ?"); params.append(hi)
257	257	if after is not None:
258	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 ?"
	259	+ # filters go INSIDE the merged source (single filtered scan, see lake.merged_sql)
	260	+ where = (" AND ".join(conds) if conds else "1=1") + _session_sql(cs.root, sess, tf)
	261	+ sql = f"SELECT * EXCLUDE (ticker) FROM ({merged_sql(paths, where=where)}) ORDER BY datetime LIMIT ?"
262	262	params.append(limit + 1)
263	263	df = con().execute(sql, params).df()
264	264	has_more = len(df) > limit
		`@@ -281,17 +281,13 @@ def root_daily(root: str) -> pd.DataFrame:`
281	281	paths.extend(files_for(sym, "1day"))
282	282	if not paths:
283	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		− """
	284	+ raw = (r"(SELECT replace(regexp_extract(filename, '([A-Z0-9]+_[FGHJKMNQUVXZ][0-9]{2})_1day\.parquet$', 1), '_', '') AS symbol, "
	285	+ "CASE WHEN filename LIKE '%/update/%' THEN 0 ELSE 1 END AS prio, "
	286	+ "datetime, open, high, low, close, volume, open_interest "
	287	+ "FROM read_parquet(?, filename=true, union_by_name=true))")
	288	+ sql = topn_sql(source=raw, partition="symbol, datetime", order="prio", n=1,
	289	+ select="symbol, CAST(datetime AS DATE) AS date, open, high, low, close, volume, open_interest",
	290	+ order_by="symbol, datetime")
295	291	df = con().execute(sql, [paths]).df()
296	292	df["date"] = pd.to_datetime(df["date"])
297	293	return df
		`@@ -446,7 +442,7 @@ def continuous(root: str, roll: str \| None, adjust: str \| None, depth: int \| Non`
446	442	if after is not None:
447	443	conds.append("datetime > ?"); p.append(str(after))
448	444	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}")
	445	+ f"FROM ({merged_sql(paths, where=' AND '.join(conds) + sess_sql)})")
450	446	params.extend(p)
451	447	if parts:
452	448	sql = " UNION ALL ".join(parts) + " ORDER BY datetime LIMIT ?"

added tests/test_duck_plan.py +108 −0

		`@@ -0,0 +1,108 @@`
	1	+"""DuckDB query plans: top-N-per-group queries must scan the Parquet files ONCE, with the datetime filter pushed
	2	+into the scan (no late-materialization re-scan), and the process must hold a single DuckDB database."""
	3	+from __future__ import annotations
	4	+
	5	+import threading
	6	+
	7	+import pytest
	8	+
	9	+
	10	+@pytest.fixture
	11	+def paths(app):
	12	+ import main
	13	+ idx = main.ticker_index("stock", "1min", "UNADJUSTED")
	14	+ return [idx[t] for t in ("AAPL", "MSFT", "SMCP")]
	15	+
	16	+
	17	+def _plan(sql, params):
	18	+ from core.duck import con, explain
	19	+ return explain(con(), sql, params)
	20	+
	21	+
	22	+def test_topn_plan_single_filtered_scan(paths):
	23	+ from core.duck import scan_count, topn_per_group
	24	+ lo, hi = "2025-06-30 09:30:00", "2025-06-30 09:34:59"
	25	+ sql = topn_per_group(None, paths, "datetime >= ? AND datetime <= ?", [lo, hi], "ticker", "datetime DESC", 100,
	26	+ order_by="ticker, datetime DESC", limit=300)
	27	+ plan = _plan(sql, [paths, lo, hi, 100, 300])
	28	+ assert scan_count(plan) == 1, plan
	29	+ assert "Filters" in plan and "datetime>=" in plan.replace(" ", "") , plan
	30	+ assert "HASH_JOIN" not in plan, plan
	31	+
	32	+
	33	+def test_naive_qualify_would_double_scan(paths):
	34	+ """Documents WHY the helper exists: the bare form is rewritten into two scans + a join."""
	35	+ from core.duck import scan_count
	36	+ lo, hi = "2025-06-30 09:30:00", "2025-06-30 09:34:59"
	37	+ sql = ("SELECT * FROM read_parquet(?) WHERE datetime >= ? AND datetime <= ? "
	38	+ "QUALIFY row_number() OVER (PARTITION BY ticker ORDER BY datetime DESC) <= ? ORDER BY ticker, datetime DESC LIMIT ?")
	39	+ plan = _plan(sql, [paths, lo, hi, 100, 300])
	40	+ assert scan_count(plan) >= 2 or "HASH_JOIN" in plan, plan
	41	+
	42	+
	43	+def test_snapshot_plan_single_scan(paths):
	44	+ from core.duck import scan_count, topn_per_group
	45	+ sql = topn_per_group(None, paths, "datetime <= ?", ["2025-06-30 10:00:00"], "ticker", "datetime DESC", 1,
	46	+ order_by="ticker", limit=3)
	47	+ plan = _plan(sql, [paths, "2025-06-30 10:00:00", 1, 3])
	48	+ assert scan_count(plan) == 1 and "HASH_JOIN" not in plan, plan
	49	+
	50	+
	51	+def test_futures_merged_plan_single_scan_per_file(app):
	52	+ from core.duck import scan_count
	53	+ from futures.lake import files_for, merged_sql
	54	+ files = files_for("ESH25", "1day")
	55	+ assert len(files) == 2, files # update + archive overlap for 2025 contracts (daily: both cover 2024-12)
	56	+ sql = f"SELECT * EXCLUDE (ticker) FROM ({merged_sql(files, where='datetime >= ? AND datetime < ?')}) ORDER BY datetime LIMIT ?"
	57	+ plan = _plan(sql, [*files, "2024-12-01", "2024-12-31", 5001])
	58	+ assert scan_count(plan) == 2, plan # one READ_PARQUET per file, none duplicated
	59	+ assert "HASH_JOIN" not in plan, plan
	60	+ from core.duck import con
	61	+ df = con().execute(sql, [*files, "2024-12-01", "2024-12-31", 5001]).df()
	62	+ assert len(df) == 21 and df["datetime"].is_unique # dedup on datetime across the two buckets
	63	+
	64	+
	65	+def test_run_topn_results(paths):
	66	+ from core.duck import con, run_topn
	67	+ df = run_topn(con(), paths, "datetime >= ? AND datetime <= ?", ["2025-06-30 09:30:00", "2025-06-30 09:34:59"],
	68	+ "ticker", "datetime DESC", 2, order_by="ticker, datetime DESC", limit=100)
	69	+ assert len(df) == 6 and list(df["ticker"].unique()) == ["AAPL", "MSFT", "SMCP"]
	70	+ assert str(df["datetime"].iloc[0]).startswith("2025-06-30 09:34")
	71	+
	72	+
	73	+def test_single_database_shared_across_threads(app):
	74	+ from core import duck
	75	+ seen = {}
	76	+
	77	+ def worker(i):
	78	+ c = duck.con()
	79	+ assert duck.con() is c # stable within the thread
	80	+ seen[i] = (c, c.execute("SELECT current_setting('memory_limit')").fetchone()[0])
	81	+
	82	+ ts = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
	83	+ for t in ts:
	84	+ t.start()
	85	+ for t in ts:
	86	+ t.join()
	87	+ assert len({id(v[0]) for v in seen.values()}) == 4 # one cursor per thread…
	88	+ assert len({v[1] for v in seen.values()}) == 1 # …one configured database
	89	+ limit = seen[0][1]
	90	+ assert limit.upper().startswith(("3.", "4", "3.7")) # 4GB → DuckDB reports ~3.7 GiB
	91	+ threads = duck.con().execute("SELECT current_setting('threads')").fetchone()[0]
	92	+ assert int(threads) == duck.settings.duck_threads
	93	+
	94	+
	95	+def test_cache_has_ttl_and_maxsize():
	96	+ from core.cache import TTLCache
	97	+ c = TTLCache(maxsize=3, ttl=1000)
	98	+ for k in "abcd":
	99	+ c.set(k, k)
	100	+ assert len(c) == 3 and c.get("a") is None and c.get("d") == "d"
	101	+ calls = {"n": 0}
	102	+
	103	+ def build():
	104	+ calls["n"] += 1
	105	+ return "v"
	106	+ assert c.get_or_build("x", build) == "v" and c.get_or_build("x", build) == "v" and calls["n"] == 1
	107	+ assert c.get_or_build("x", build, ttl=-1) == "v" and calls["n"] == 2 # caller-specific horizon expired
	108	+ assert c.invalidate("x") == 1 and c.get("x") is None
	109