gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(b3b): move candidate-universe SQL into store adapter; stable tiebreaker; tidy candidate panel load

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-18 09:03:46 +02:00
parent 103572a6eb
commit 76e07ec3be
3 changed files with 141 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
src/fxhnt/adapters/warehouse/duckdb_feature_store.py
View File

@@ -110,6 +110,53 @@ class DuckDbFeatureStore:
out[symbol][int(ts) // 86_400] = float(value)
return out
def select_candidate_universe(self, *, min_history: int, top_k: int) -> dict[str, float]:
"""Top-`top_k` most-liquid symbols (by PEAK calendar-year avg dollar-volume,
close*volume) having at least `min_history` bars where BOTH close and volume
are present. Returns {symbol: peak_yearly_avg_dollar_volume}. Survivorship-safe
(peak, not lifetime-avg, so delisted-but-once-liquid names are included).
`min_history` counts bars where BOTH close and volume are present (the
self-joined set), NOT adjclose-only bars. Ranking is by peak calendar-year
average dollar-volume; ties broken deterministically by symbol ascending.
ONE DuckDB query, all aggregation pushed into the engine (no rows pulled into
Python). Self-join `features` on (symbol, ts): one side close, other volume →
per-bar dollar-volume. Bucket by approximate calendar year, take the per-year
mean dollar-volume, then the PEAK year per symbol."""
sql = """
WITH dv AS (
SELECT c.symbol AS symbol,
c.ts AS ts,
c.value * v.value AS dollar_vol
FROM features c
JOIN features v
ON c.symbol = v.symbol AND c.ts = v.ts
WHERE c.feature = 'close' AND v.feature = 'volume'
),
yearly AS (
SELECT symbol,
CAST(ts / (86400 * 365.25) AS INTEGER) AS yr,
avg(dollar_vol) AS yr_avg_dv,
count(*) AS bars
FROM dv
GROUP BY symbol, yr
),
agg AS (
SELECT symbol,
max(yr_avg_dv) AS peak_yearly_dv,
sum(bars) AS total_bars
FROM yearly
GROUP BY symbol
)
SELECT symbol, peak_yearly_dv
FROM agg
WHERE total_bars >= ? AND peak_yearly_dv > 0.0
ORDER BY peak_yearly_dv DESC, symbol ASC
LIMIT ?
"""
rows = self._con.execute(sql, [min_history, top_k]).fetchall()
return {str(sym): float(dv) for sym, dv in rows}
# --- ticker membership (survivorship-free listing index) ---------------------------------
def upsert_membership(self, rows: list[tuple[str, str, str, str]]) -> None:

50
src/fxhnt/application/equity_backtest_runner.py
View File

@@ -53,51 +53,18 @@ def select_backtest_universe(
) -> dict[str, float]:
"""Bounded, survivorship-safe candidate universe for the backtest.
Thin wrapper that delegates to the store adapter's `select_candidate_universe`
(keeps the SQL inside the warehouse adapter, respecting the hexagonal boundary).
Returns {symbol: peak_yearly_avg_dollar_volume} for the `top_k` most-liquid
symbols that have at least `min_history` daily bars. Liquidity is ranked by
symbols that have at least `min_history` bars where BOTH close and volume are
present (the self-joined set, NOT adjclose-only bars). Liquidity is ranked by
PEAK calendar-year average dollar-volume (close*volume), NOT lifetime average,
so names that were highly liquid only while alive (later delisted) or only
recently (recent IPOs) are still included — avoiding survivorship bias in the
candidate set. This caps memory: the runner then loads only these symbols.
"""
# ONE DuckDB query, all aggregation pushed into the engine (no rows pulled into
# Python). Self-join `features` on (symbol, ts): one side close, other volume →
# per-bar dollar-volume. Bucket by approximate calendar year, take the per-year
# mean dollar-volume, then the PEAK year per symbol. Bar count comes from the
# 'close' feature. Filter bars >= min_history, rank by peak yearly dv desc.
sql = """
WITH dv AS (
SELECT c.symbol AS symbol,
c.ts AS ts,
c.value * v.value AS dollar_vol
FROM features c
JOIN features v
ON c.symbol = v.symbol AND c.ts = v.ts
WHERE c.feature = 'close' AND v.feature = 'volume'
),
yearly AS (
SELECT symbol,
CAST(ts / (86400 * 365.25) AS INTEGER) AS yr,
avg(dollar_vol) AS yr_avg_dv,
count(*) AS bars
FROM dv
GROUP BY symbol, yr
),
agg AS (
SELECT symbol,
max(yr_avg_dv) AS peak_yearly_dv,
sum(bars) AS total_bars
FROM yearly
GROUP BY symbol
)
SELECT symbol, peak_yearly_dv
FROM agg
WHERE total_bars >= ? AND peak_yearly_dv > 0.0
ORDER BY peak_yearly_dv DESC
LIMIT ?
"""
rows = store._con.execute(sql, [min_history, top_k]).fetchall()
return {str(sym): float(dv) for sym, dv in rows}
return store.select_candidate_universe(min_history=min_history, top_k=top_k)
class EquityBacktestRunner:
@@ -136,10 +103,11 @@ class EquityBacktestRunner:
top_k=self._candidate_top_k,
)
members = [m for m in members if m[0] in liq]
adj = self._store.read_panel(list(liq), "adjclose")
cand_syms = [m[0] for m in members]
adj = self._store.read_panel(cand_syms, "adjclose")
close: dict[str, dict[int, float]] = {}
vol: dict[str, dict[int, float]] = {}
syms = list(liq)
syms = cand_syms
else:
syms = [m[0] for m in members]
adj = self._store.read_panel(syms, "adjclose")

85
tests/integration/test_equity_backtest_runner.py
View File

@@ -206,6 +206,91 @@ def test_select_backtest_universe_top_k_drops_illiquid_tail(tmp_path):
assert "ILLIQUID" not in liq
def test_select_candidate_universe_tie_broken_by_symbol_deterministically(tmp_path):
# FIX #3: two symbols with identical peak dollar-volume must order
# deterministically by symbol ascending (stable tiebreaker).
store = DuckDbFeatureStore(str(tmp_path / "tie.duckdb"))
def write(sym: str, start_iso: str, n_bars: int, close: float, vol: float) -> None:
start = dt.date.fromisoformat(start_iso)
rows = []
for i in range(n_bars):
day = start + dt.timedelta(days=i)
ts = (day - dt.date(1970, 1, 1)).days * _SPD
rows.append((ts, {"adjclose": close, "close": close, "volume": vol}))
store.write_features(sym, rows)
store.upsert_membership([(sym, "NYSE", start_iso,
(start + dt.timedelta(days=n_bars)).isoformat())])
# ZTIE and ATIE have identical close*volume over identical date ranges →
# identical peak-yearly dollar-volume → pure tie. top_k=1 must always keep
# the symbol-ascending winner (ATIE), never ZTIE, on every run.
write("ZTIE", "2015-01-01", 400, close=100.0, vol=1_000_000.0)
write("ATIE", "2015-01-01", 400, close=100.0, vol=1_000_000.0)
full = store.select_candidate_universe(min_history=300, top_k=10)
assert full["ZTIE"] == full["ATIE"] # genuine tie in the ranking key
for _ in range(3):
top1 = store.select_candidate_universe(min_history=300, top_k=1)
assert set(top1) == {"ATIE"}
store.close()
def test_select_candidate_universe_filters_and_ranks_via_store(tmp_path):
# Store-level mirror of test_select_backtest_universe_filters_and_ranks:
# min_history exclusion, delisted-but-liquid inclusion, top_k limit.
store = _build_liquidity_warehouse(str(tmp_path / "liq.duckdb"))
liq = store.select_candidate_universe(min_history=300, top_k=2500)
assert "TOOSHORT" not in liq # min_history exclusion
assert "DELISTED" in liq # survivorship-safe inclusion
assert liq["DELISTED"] > liq["LIVELONG"] > liq["ILLIQUID"] # ranking
top2 = store.select_candidate_universe(min_history=300, top_k=2)
assert set(top2) == {"DELISTED", "LIVELONG"} # top_k limit
store.close()
def test_candidate_path_deterministic(tmp_path):
# FIX #3 / project determinism requirement: the candidate path must produce
# byte-identical series across repeated runs.
store = _build_warehouse(str(tmp_path / "wh.duckdb"))
runner = EquityBacktestRunner(
store, n=10, candidate_top_k=100, candidate_min_history=252,
)
a = runner.run().returns_by_construction["ls"]
b = runner.run().returns_by_construction["ls"]
store.close()
assert np.array_equal(a, b)
def test_candidate_path_with_large_top_k_matches_membership(tmp_path):
# Equivalence sanity: with top_k >= number of symbols (3) and a low
# candidate_min_history, all 3 synthetic names qualify for the candidate set.
# The candidate path ranks by STATIC peak-yearly liquidity while the None path
# ranks by TRAILING dollar-volume; with n=10 (>= 3 names) BOTH paths hold ALL
# three names every traded period, so the HELD sets — and thus weights and the
# daily 'ls' series — are identical. n_names_avg is therefore identical (~3.0).
store = _build_warehouse(str(tmp_path / "wh.duckdb"))
none_path = EquityBacktestRunner(
store, n=10, momentum_min_history=252,
).run()
cand_path = EquityBacktestRunner(
store, n=10, momentum_min_history=252,
candidate_top_k=100, candidate_min_history=10,
).run()
store.close()
# both paths hold all 3 names every traded period
assert abs(none_path.n_names_avg - 3.0) < 1e-9
assert abs(cand_path.n_names_avg - 3.0) < 1e-9
assert cand_path.n_names_avg == pytest.approx(none_path.n_names_avg)
# identical held sets → identical series across all three constructions
for c in ("long", "ls", "tilt"):
assert np.array_equal(
none_path.returns_by_construction[c],
cand_path.returns_by_construction[c],
)
def test_runner_candidate_path_matches_membership_gating(tmp_path):
store = _build_warehouse(str(tmp_path / "wh.duckdb"))
runner = EquityBacktestRunner(