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) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -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}
|
||||
|
||||
@@ -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
|
||||
|
||||
175
src/fxhnt/application/cta_runner.py
Normal file
175
src/fxhnt/application/cta_runner.py
Normal file
@@ -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
|
||||
93
src/fxhnt/domain/cta_trend.py
Normal file
93
src/fxhnt/domain/cta_trend.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user