From c39d78fd7ba22b84fcbdabd418755c0d84098901 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 19 Jun 2026 23:07:30 +0200 Subject: [PATCH] feat(surfer): cross-asset futures curve loader + CTA harness + opt-in allocator quality gate Reusable infra from the diversifier research (the strategies themselves found no deployable edge on our 22-market panel, but the construction is validated): - dbn_local.load_curve: front + 2nd-nearby CONTINUOUS series + annualized carry (log(F1/F2)/gap_years) from parent-symbology outright expiries. Unlocks carry + basis-momentum without re-reading raw .dbn. - cta_trend.py / cta_runner.py: literature-grounded managed-futures trend (vol-normalized risk-adj momentum -> tanh(x/0.89) response -> 1/3/12mo blend -> inverse-vol -> equal-risk-per-sector budget -> portfolio vol-target -> long/short). Construction validated (reproduces SG Trend +29% in 2022); premium absent for us (Sharpe ~0 over 2010-26 on 22 markets, ends mid-historic-drawdown). - book_allocator: OPTIONAL edge-quality gate (min_sleeve_sharpe, default None=OFF) to bench dead/decaying sleeves so they can't be levered up by the vol-target. Off by default because a naive Sharpe floor also benches legitimate anticorrelated hedges; a correct quality/decay layer is future work. Suite green, mypy clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fxhnt/adapters/data/dbn_local.py | 63 +++++++++ src/fxhnt/application/book_allocator.py | 40 ++++-- src/fxhnt/application/cta_runner.py | 175 ++++++++++++++++++++++++ src/fxhnt/domain/cta_trend.py | 93 +++++++++++++ 4 files changed, 359 insertions(+), 12 deletions(-) create mode 100644 src/fxhnt/application/cta_runner.py create mode 100644 src/fxhnt/domain/cta_trend.py diff --git a/src/fxhnt/adapters/data/dbn_local.py b/src/fxhnt/adapters/data/dbn_local.py index a33d36d..5b0dd0a 100644 --- a/src/fxhnt/adapters/data/dbn_local.py +++ b/src/fxhnt/adapters/data/dbn_local.py @@ -69,8 +69,71 @@ def _front_to_series(symbol: str, asset_class: AssetClass, days: np.ndarray, ins close=synthetic, high=syn_high, low=syn_low) +def read_curve_arrays(path: str, symbol: str) -> tuple[np.ndarray, ...]: + """Read a parent-symbology .dbn and return the FRONT and 2ND-NEARBY curve, aligned per day. + + Front = highest-volume outright (matches `read_front_arrays`). 2nd-nearby = the nearest DEFERRED + expiry strictly beyond the front (tie-break by volume). The 1-digit contract year (e.g. ESM0=2010, + ESZ6=2026) is resolved to the unique year in [trade_year, trade_year+3] ending in that digit. + + Returns (days, front_inst, front_close, second_inst, second_close, gap_months) for days where both a + front and a 2nd contract exist. gap_months = (2nd expiry - front expiry) in calendar months, the + annualization denominator for carry = log(F1/F2)/(gap_months/12). + """ + import databento as db + import pandas as pd + + df = db.DBNStore.from_file(path).to_df().reset_index() + if df.empty or "close" not in df.columns: + raise ValueError(f"{path}: no OHLCV close") + df = df[df["close"] > 0].copy() + pattern = re.compile("^" + re.escape(symbol) + "[" + _MONTHS + r"]\d{1,2}$") + df = df[df["symbol"].astype(str).str.match(pattern)] + if df.empty: + raise ValueError(f"{path}: no outright contracts match {symbol}") + n = len(symbol) + df["day"] = df["ts_event"].astype("int64") // (86_400 * 10**9) + month_idx = {c: i for i, c in enumerate(_MONTHS)} + df["m"] = df["symbol"].str[n].map(month_idx) + ydig = df["symbol"].str[n + 1:] + trade_year = pd.to_datetime(df["ts_event"]).dt.year + y1 = ydig.str[-1:].astype(int) + one = ydig.str.len() == 1 + cy = np.where(one, trade_year + ((y1 - trade_year % 10) % 10), 2000 + ydig.str[-2:].astype(int)) + df["exp"] = cy.astype(np.int64) * 12 + df["m"].astype(np.int64) + # front = max-volume contract per day + front = df.loc[df.groupby("day")["volume"].idxmax(), + ["day", "exp", "instrument_id", "close"]].rename( + columns={"exp": "f_exp", "instrument_id": "f_inst", "close": "f_close"}) + merged = df.merge(front[["day", "f_exp"]], on="day") + cand = merged[merged["exp"] > merged["f_exp"]] + second = (cand.sort_values(["day", "exp", "volume"], ascending=[True, True, False]) + .groupby("day").first().reset_index()[["day", "exp", "instrument_id", "close"]] + .rename(columns={"exp": "s_exp", "instrument_id": "s_inst", "close": "s_close"})) + curve = front.merge(second, on="day").sort_values("day") + return (curve["day"].to_numpy(np.int64), curve["f_inst"].to_numpy(np.int64), + curve["f_close"].to_numpy(np.float64), curve["s_inst"].to_numpy(np.int64), + curve["s_close"].to_numpy(np.float64), + (curve["s_exp"] - curve["f_exp"]).to_numpy(np.float64)) + + class LocalDbnLoader: name = "dbn_local" def load(self, path: str, symbol: str, asset_class: AssetClass = AssetClass.FUTURE) -> PriceSeries: return _front_to_series(symbol, asset_class, *read_front_arrays(path, symbol)) + + def load_curve(self, path: str, symbol: str) -> dict[str, np.ndarray]: + """Front + 2nd-nearby CONTINUOUS (ratio back-adjusted) closes + per-day annualized carry. + + Returns {'days', 'front', 'second', 'carry'} numpy arrays aligned on `days`. `front`/`second` are + continuous front-adjusted price levels (base 100) for the 1st/2nd-nearby roll series; + `carry` = log(F1/F2)/(gap_months/12) (backwardation positive). Powers basis-momentum (R1-R2) and + cross-asset carry without re-reading the raw .dbn. + """ + days, f_inst, f_close, s_inst, s_close, gap_m = read_curve_arrays(path, symbol) + front, _, _ = continuous_front_adjust(f_inst, f_close, f_close, f_close) + second, _, _ = continuous_front_adjust(s_inst, s_close, s_close, s_close) + gap_years = np.where(gap_m > 0, gap_m / 12.0, np.nan) + carry = np.log(np.where((f_close > 0) & (s_close > 0), f_close / s_close, np.nan)) / gap_years + return {"days": days, "front": front, "second": second, "carry": carry} diff --git a/src/fxhnt/application/book_allocator.py b/src/fxhnt/application/book_allocator.py index 72ab76a..eed16b7 100644 --- a/src/fxhnt/application/book_allocator.py +++ b/src/fxhnt/application/book_allocator.py @@ -74,20 +74,30 @@ def _inverse_vol_weights(window: dict[str, np.ndarray], max_sleeve_weight: float def combine_book(series_by_edge: dict[str, EdgeReturns], *, vol_lookback: int = 60, target_vol: float = 0.12, kelly_fraction: float = 0.5, max_sleeve_weight: float = 0.4, periods_per_year: int = 365, - rebalance_every: int = 21) -> dict[int, float]: + rebalance_every: int = 21, + min_sleeve_sharpe: float | None = None) -> dict[int, float]: """Combine date-stamped edges into ONE risk-managed daily book return series. Method (uses ONLY trailing data — no lookahead): 1. Align edges to the union date axis (0.0-fill on flat days). - 2. Every `rebalance_every` days, recompute inverse-vol relative weights from the trailing - `vol_lookback` window, CAP each at `max_sleeve_weight`, renormalize. Hold between rebalances. + 2. OPTIONAL EDGE-QUALITY GATE (`min_sleeve_sharpe`, default None = OFF): when set, every + `rebalance_every` days bench any sleeve whose trailing per-period Sharpe over the window is + below the floor, so a dead/decaying low-vol sleeve cannot earn inverse-vol weight and be + levered up by the vol-target (the allocator's exploitable flaw — a dead cta-trend sleeve + once inflated the book Sharpe 3.27→3.70 purely via manufactured leverage). + CAVEAT — a naive Sharpe floor cannot distinguish a dead sleeve from a legitimate + anticorrelated HEDGE (low standalone Sharpe but vol-reducing): it benches both. So it is + OFF by default; a correct quality/decay layer (gate on a long window + correlation-benefit, + or deflated-Sharpe) is future work. Use only for return-seeking sleeves you trust are edges. + Then recompute inverse-vol weights over the ELIGIBLE sleeves only, CAP each at + `max_sleeve_weight`, renormalize. Benched sleeves get weight 0. Hold between rebalances. 3. Portfolio raw daily return r_p[t] = Σ_i w_i · r_i[t]. 4. Vol-target + fractional-Kelly leverage (recomputed each rebalance from trailing data): lev = kelly_fraction · (target_vol_daily / trailing_port_vol), capped to [0, _MAX_LEVERAGE] where target_vol_daily = target_vol / sqrt(periods_per_year) and trailing_port_vol is the realized daily std of r_p over the trailing `vol_lookback` window. fractional-Kelly is the overall scalar so we run a fraction of the full vol-targeted exposure. - 5. book[t] = lev · r_p[t]. + 5. book[t] = lev · r_p[t]. If NO sleeve is eligible, the book is flat (0.0) that period. Returns {epoch_day: book_return}. """ @@ -102,20 +112,26 @@ def combine_book(series_by_edge: dict[str, EdgeReturns], *, vol_lookback: int = weights: dict[str, float] = {e: 1.0 / len(edges) for e in edges} if edges else {} leverage = 0.0 for t in range(n): - if t % rebalance_every == 0: + if t % rebalance_every == 0 and t > 1: # need trailing data to estimate vol/edge lo = max(0, t - vol_lookback) - window = {e: arrs[e][lo:t] for e in edges} - if t > 1: # need trailing data to estimate vol - weights = _inverse_vol_weights(window, max_sleeve_weight) + # OPTIONAL edge-quality gate: keep only sleeves with usable vol and (if gating) a + # trailing Sharpe >= floor. With the gate OFF (default) all sleeves with vol are eligible. + eligible = {e: arrs[e][lo:t] for e in edges + if (arrs[e][lo:t].size > 1 and float(np.std(arrs[e][lo:t])) > 1e-12 + and (min_sleeve_sharpe is None + or per_period_sharpe(arrs[e][lo:t]) >= min_sleeve_sharpe))} + if eligible: + ew = _inverse_vol_weights(eligible, max_sleeve_weight) + weights = {e: ew.get(e, 0.0) for e in edges} # trailing raw portfolio vol under the NEW weights (no lookahead: uses [lo, t)) port_window = np.zeros(t - lo, dtype=float) for e in edges: port_window += weights[e] * arrs[e][lo:t] pv = float(np.std(port_window)) if port_window.size > 1 else 0.0 - if pv > 1e-12: - leverage = min(_MAX_LEVERAGE, kelly_fraction * (target_vol_daily / pv)) - else: - leverage = 0.0 + leverage = min(_MAX_LEVERAGE, kelly_fraction * (target_vol_daily / pv)) if pv > 1e-12 else 0.0 + else: # every sleeve bleeding → stand flat + weights = {e: 0.0 for e in edges} + leverage = 0.0 raw = sum(weights[e] * arrs[e][t] for e in edges) book[dates[t]] = leverage * raw return book diff --git a/src/fxhnt/application/cta_runner.py b/src/fxhnt/application/cta_runner.py new file mode 100644 index 0000000..ba6e2a6 --- /dev/null +++ b/src/fxhnt/application/cta_runner.py @@ -0,0 +1,175 @@ +"""Proper CTA / managed-futures trend backtest — sector risk budgeting + portfolio vol target. + +The "trend equation" assembled from the literature (see `domain/cta_trend.py` for signal provenance): + + 1. Per instrument: signed conviction in ~[-1,1] (cta_conviction) sized inverse-vol -> w_i = conv_i/σ_i. + Risk contribution is then |conv_i| (the σ cancels) — that identity is what makes the sector layer + covariance-free. + 2. Sector risk budget: normalize each ACTIVE sector's total risk (Σ|conv|) to 1/n_sectors, so the 12 + commodity/ag markets cannot dominate the 4 equity ones (the bug that sank the naive TrendRunner). + 3. Portfolio vol target: lever the whole (unlevered) book by σ_TGT / trailing-realized-vol, capped at + max_leverage (Baltas-Kosowski constant-vol overlay). Strictly past-only (no look-ahead). + 4. Long/short by default (crisis alpha lives in the short leg). Held between rebalances. + +A "price panel" = dict[str, dict[int, float]] = {symbol: {epoch_day: close}}. Returns the same +`TrendRunResult` shape as TrendRunner so it plugs into the gauntlet and the book allocator unchanged. +""" +from __future__ import annotations + +import math +from collections import defaultdict +from typing import Any + +import numpy as np + +from fxhnt.application.trend_runner import TrendRunResult, _json_floats +from fxhnt.domain.backtest import compute_stats +from fxhnt.domain.cta_trend import DEFAULT_HORIZONS, SECTOR_MAP, cta_conviction, ewma_daily_vol +from fxhnt.domain.gauntlet.core import evaluate, per_period_sharpe + + +class CtaRunner: + def __init__(self, prices: dict[str, dict[int, float]], *, + horizons: tuple[int, ...] = DEFAULT_HORIZONS, + sector_map: dict[str, str] | None = None, + target_vol: float = 0.15, com: float = 60.0, vol_lookback: int = 60, + max_leverage: float = 4.0, min_history: int = 300, cost_bps: float = 5.0, + rebalance_every: int = 5, long_short: bool = True, + periods_per_year: int = 252) -> None: + self._p = prices + self._horizons = horizons + self._sector = sector_map or SECTOR_MAP + self._target_vol = target_vol + self._com = com + self._vol_lookback = vol_lookback + self._max_lev = max_leverage + self._min_history = min_history + self._cost = cost_bps / 1e4 + self._rebalance_every = max(1, rebalance_every) + self._long_short = long_short + self._ppy = periods_per_year + + def _closes_asof(self, sym: str, day: int) -> list[float]: + series = self._p[sym] + return [series[d] for d in sorted(series) if d <= day] + + def _targets(self, day: int) -> dict[str, float]: + """Sector-risk-budgeted UNLEVERED signed weights as of `day` (uses only closes <= day).""" + conv: dict[str, float] = {} + sigma: dict[str, float] = {} + for sym, series in self._p.items(): + if day not in series: + continue + closes = self._closes_asof(sym, day) + if len(closes) < self._min_history: + continue + c = cta_conviction(closes, horizons=self._horizons, com=self._com) + dv = ewma_daily_vol(closes, com=self._com) + if c is None or dv is None: + continue + if not self._long_short: + c = max(0.0, c) + if c == 0.0: + continue + conv[sym] = c + sigma[sym] = dv * math.sqrt(self._ppy) # annualized ex-ante vol + if not conv: + return {} + # group risk (Σ|conv|) by sector; equalize each active sector to 1/n_sectors of total risk + by_sector: dict[str, list[str]] = defaultdict(list) + for sym in conv: + by_sector[self._sector.get(sym, "other")].append(sym) + n_sectors = len(by_sector) + weights: dict[str, float] = {} + for syms in by_sector.values(): + sector_risk = sum(abs(conv[s]) for s in syms) + if sector_risk <= 0.0: + continue + sector_scale = (1.0 / n_sectors) / sector_risk + for s in syms: + weights[s] = (conv[s] / sigma[s]) * sector_scale + return weights + + def run(self) -> TrendRunResult: + all_days = sorted({d for s in self._p.values() for d in s}) + # pass 1: unlevered held positions + unlevered realized returns + unlev: list[float] = [] + snaps: list[dict[str, float]] = [] + dates: list[int] = [] + pos: dict[str, float] = {} + assets_seen = 0 + for i in range(len(all_days) - 1): + d, nxt = all_days[i], all_days[i + 1] + if i % self._rebalance_every == 0: + pos = self._targets(d) + realized = 0.0 + for sym, w in pos.items(): + series = self._p[sym] + if d in series and nxt in series and series[d] > 0 and series[nxt] > 0: + realized += w * (series[nxt] / series[d] - 1.0) + unlev.append(realized) + snaps.append(dict(pos)) + dates.append(int(nxt)) + assets_seen += len(pos) + n = len(unlev) + if n == 0: + return TrendRunResult(returns=np.zeros(0), n_assets_avg=0.0, n_rebalances=0, dates=[]) + unlev_arr = np.asarray(unlev, dtype=float) + # pass 2: portfolio vol-target overlay (trailing realized vol, strictly past) + turnover on levered book + tgt_daily = self._target_vol / math.sqrt(self._ppy) + rets: list[float] = [] + prev_lev_pos: dict[str, float] = {} + for t in range(n): + if t >= self._vol_lookback: + rv = float(np.std(unlev_arr[t - self._vol_lookback:t])) + lev = min(self._max_lev, tgt_daily / rv) if rv > 0 else 1.0 + else: + lev = 1.0 + lev_pos = {s: w * lev for s, w in snaps[t].items()} + turnover = sum(abs(lev_pos.get(s, 0.0) - prev_lev_pos.get(s, 0.0)) + for s in set(lev_pos) | set(prev_lev_pos)) * self._cost / 2.0 + rets.append(lev * unlev_arr[t] - turnover) + prev_lev_pos = lev_pos + return TrendRunResult( + returns=np.asarray(rets, dtype=float), + n_assets_avg=(assets_seen / n) if n else 0.0, + n_rebalances=n, + dates=dates, + ) + + +def evaluate_cta(prices: dict[str, dict[int, float]], *, periods_per_year: int = 252, + configs: list[dict[str, Any]] | None = None, oos_fraction: float = 0.40, + dsr_min: float = 0.95, oos_min_sharpe: float = 0.0, + max_is_oos_decay: float = 0.50) -> dict[str, Any]: + """Run CtaRunner for each config and gauntlet every cell. Mirrors evaluate_trend's structure.""" + configs = configs or [{"label": "cta-default"}] + results: dict[str, np.ndarray] = {} + labels: list[str] = [] + for idx, cfg in enumerate(configs): + kwargs = {k: v for k, v in cfg.items() if k != "label"} + label = str(cfg.get("label", f"cfg{idx}")) + labels.append(label) + results[label] = CtaRunner(prices, periods_per_year=periods_per_year, **kwargs).run().returns + n_trials = len(configs) + is_sharpes = [per_period_sharpe(r[:int((1.0 - oos_fraction) * len(r))]) + for r in results.values() if len(r) > 3] + sr_variance = float(np.var(is_sharpes)) if len(is_sharpes) > 1 else 0.0 + out: dict[str, Any] = {"cells": {}, "n_trials": n_trials, + "settings": {"oos_fraction": oos_fraction, "periods_per_year": periods_per_year}} + for label in labels: + r = results[label] + stats = compute_stats(r, periods_per_year=periods_per_year) + if len(r) < 3: + out["cells"][label] = {"config_label": label, "stats": _json_floats(stats.model_dump()), + "verdict": {"passed": False, "dsr": 0.0, "is_sharpe": 0.0, + "oos_sharpe": 0.0, "n_trials": n_trials, + "reasons": ["insufficient data"], "pvalue": 1.0}} + continue + split = int((1.0 - oos_fraction) * len(r)) + v = evaluate(r[:split], r[split:], n_trials=n_trials, sr_variance=sr_variance, + dsr_min=dsr_min, oos_min_sharpe=oos_min_sharpe, + max_is_oos_decay=max_is_oos_decay, has_economic_rationale=True) + out["cells"][label] = {"config_label": label, "stats": _json_floats(stats.model_dump()), + "verdict": _json_floats(v.model_dump())} + return out diff --git a/src/fxhnt/domain/cta_trend.py b/src/fxhnt/domain/cta_trend.py new file mode 100644 index 0000000..995c069 --- /dev/null +++ b/src/fxhnt/domain/cta_trend.py @@ -0,0 +1,93 @@ +"""Proper CTA / managed-futures trend signal — vol-normalized momentum with a saturating response. + +Grounded in the trend-following literature: +- Per-instrument ex-ante vol: EWMA of squared daily log-returns, center-of-mass 60d (Moskowitz-Ooi- + Pedersen, "Time Series Momentum", JFE 2012). +- Signal: risk-adjusted momentum over a lookback = log-return / (daily_vol * sqrt(L)) ≈ the realized + Sharpe over the window (Baltas-Kosowski TREND rule). +- Response: tanh(x / 0.89) saturating nonlinearity (Lempérière-Deremble-Seager-Potters-Bouchaud, + "Two centuries of trend following", 2014 — verbatim s* = 0.89). +- Blend: equal-weight the per-horizon convictions over 1/3/12-month (MOP / AQR "A Century of Evidence"). + +Pure functions on an ascending-by-day closes list. No look-ahead: every value uses only closes <= the +as-of day. The runner (`application/cta_runner.py`) adds sector risk budgeting + a portfolio vol-target. +""" +from __future__ import annotations + +import math + +import numpy as np + +# Asset-class sector map for the CME continuous-front roots (`futures_fetch.DEFAULT_ROOTS`). The sector +# layer equalizes risk across these groups so the 12 commodity/ag markets cannot drown the 4 equity ones. +SECTOR_MAP: dict[str, str] = { + "ES": "equity", "NQ": "equity", "YM": "equity", "RTY": "equity", + "ZN": "rates", "ZB": "rates", "ZF": "rates", "ZT": "rates", + "6E": "fx", "6J": "fx", "6B": "fx", "6A": "fx", "6C": "fx", + "GC": "metals", "SI": "metals", "HG": "metals", + "CL": "energy", "NG": "energy", "RB": "energy", + "ZC": "ags", "ZS": "ags", "ZW": "ags", +} + +_S_STAR = 0.89 # Lempérière-Bouchaud saturation constant (verbatim) +DEFAULT_HORIZONS = (21, 63, 252) # 1 / 3 / 12 month in trading days (MOP / AQR standard) + + +def _ewma(values: np.ndarray, com: float) -> np.ndarray: + """Exponentially-weighted moving average, center-of-mass `com` (alpha = 1/(1+com)). Full series.""" + alpha = 1.0 / (1.0 + com) + out = np.empty(len(values), dtype=float) + acc = float(values[0]) + out[0] = acc + for i in range(1, len(values)): + acc = alpha * float(values[i]) + (1.0 - alpha) * acc + out[i] = acc + return out + + +def ewma_daily_vol(closes: list[float], *, com: float = 60.0) -> float | None: + """Ex-ante daily return vol = sqrt(EWMA of squared demeaned daily log-returns), com=60d (MOP 2012). + + Returns None if there is too little history or the estimate is non-positive. + """ + c = np.asarray(closes, dtype=float) + if len(c) < int(com) // 2 + 3 or np.any(c <= 0): + return None + r = np.diff(np.log(c)) + if len(r) < 2: + return None + m = _ewma(r, com) + v = _ewma((r - m) ** 2, com) + sd = math.sqrt(max(float(v[-1]), 0.0)) + return sd if sd > 0.0 else None + + +def risk_adjusted_momentum(closes: list[float], lookback: int, daily_vol: float) -> float | None: + """Log-return over `lookback` days scaled by the window vol = log(p_t/p_{t-L}) / (daily_vol*sqrt(L)). + + Dimensionless (~ realized Sharpe over the window), so comparable across instruments. None if short. + """ + c = np.asarray(closes, dtype=float) + if len(c) <= lookback or daily_vol <= 0.0 or c[-1] <= 0.0 or c[-1 - lookback] <= 0.0: + return None + mom = math.log(c[-1] / c[-1 - lookback]) + return mom / (daily_vol * math.sqrt(lookback)) + + +def cta_conviction(closes: list[float], *, horizons: tuple[int, ...] = DEFAULT_HORIZONS, + com: float = 60.0, s_star: float = _S_STAR) -> float | None: + """Signed conviction in ~[-1, 1]: mean over horizons of tanh(risk_adjusted_momentum / s_star). + + None when the daily-vol estimate or every horizon's momentum is unavailable (insufficient history). + """ + dv = ewma_daily_vol(closes, com=com) + if dv is None: + return None + convs: list[float] = [] + for lookback in horizons: + m = risk_adjusted_momentum(closes, lookback, dv) + if m is not None: + convs.append(math.tanh(m / s_star)) + if not convs: + return None + return sum(convs) / len(convs)