Add indicator ablation framework + apply ablation-driven cleanup
Framework: - TechnicalConfig.disabled_indicators skips named indicators in engine.analyze - Optimizer supports disable_indicator override for parallel ablation sweeps - main.py --ablate runs leave-one-out across all 10 directional indicators Cleanup applied (validated across 2024 trend / 2023 chop / 2026 recent): - Drop pivot by default (cumulative +9.5% to disable) - Drop candlestick by default (+3.4% cumulative) - Regime-gate ichimoku: 1.3x in trending, 0.2x in ranging (was a chop villain) - Trim RSI weight 1.0 -> 0.3 (cumulative noise across regimes) Result: +10.60% on 2024 trend (+33.24% -> +43.84%, Sharpe 2.17 -> 2.77), +15.71% on 2023 chop (-18.27% -> -2.56%). Recent regime still adverse. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Harwinder
@@ -446,6 +446,9 @@ def _apply_overrides(config: PlatformConfig, overrides: Dict[str, float]) -> Pla
|
||||
new_config.backtest.regime_filter_min_adx = float(overrides["regime_filter_min_adx"])
|
||||
if "risk_per_trade_pct" in overrides:
|
||||
new_config.backtest.risk_per_trade_pct = float(overrides["risk_per_trade_pct"])
|
||||
if "disable_indicator" in overrides:
|
||||
v = overrides["disable_indicator"]
|
||||
new_config.technical.disabled_indicators = [str(v)] if v else []
|
||||
|
||||
return new_config
|
||||
|
||||
|
||||
@@ -85,6 +85,24 @@ class TechnicalConfig:
|
||||
pivot_lookback: int = 1
|
||||
divergence_lookback: int = 14
|
||||
|
||||
# Ablation control — names listed here are skipped during analyze().
|
||||
# Use lowercase keys: ema_trend, rsi, macd, bollinger, stochastic,
|
||||
# adx, ichimoku, divergence, candlestick, pivot, atr, volume.
|
||||
#
|
||||
# Defaults disable indicators that empirically hurt P&L on USD/JPY across
|
||||
# 2024-trend / 2023-chop / 2026-recent ablation sweeps (see project memory):
|
||||
# - pivot: net +9.5% to disable across 3 windows
|
||||
# - candlestick: net +3.4% to disable
|
||||
disabled_indicators: List[str] = field(
|
||||
default_factory=lambda: ["pivot", "candlestick"]
|
||||
)
|
||||
|
||||
|
||||
ALL_INDICATORS: List[str] = [
|
||||
"ema_trend", "rsi", "macd", "bollinger", "stochastic",
|
||||
"adx", "ichimoku", "divergence", "candlestick", "pivot",
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class SignalConfig:
|
||||
|
||||
@@ -83,18 +83,20 @@ class TechnicalEngine:
|
||||
)
|
||||
|
||||
# ── Compute all indicators ─────────────────────────────────
|
||||
self._compute_ema_signals(df, result)
|
||||
self._compute_rsi(df, result)
|
||||
self._compute_macd(df, result)
|
||||
self._compute_bollinger(df, result)
|
||||
self._compute_stochastic(df, result)
|
||||
self._compute_adx(df, result)
|
||||
self._compute_atr(df, result)
|
||||
self._compute_ichimoku(df, result)
|
||||
self._detect_divergences(df, result)
|
||||
self._detect_candlestick_patterns(df, result)
|
||||
self._compute_support_resistance(df, result)
|
||||
self._compute_volume_analysis(df, result)
|
||||
disabled = set(self.config.disabled_indicators or [])
|
||||
|
||||
if "ema_trend" not in disabled: self._compute_ema_signals(df, result)
|
||||
if "rsi" not in disabled: self._compute_rsi(df, result)
|
||||
if "macd" not in disabled: self._compute_macd(df, result)
|
||||
if "bollinger" not in disabled: self._compute_bollinger(df, result)
|
||||
if "stochastic" not in disabled: self._compute_stochastic(df, result)
|
||||
if "adx" not in disabled: self._compute_adx(df, result)
|
||||
if "atr" not in disabled: self._compute_atr(df, result)
|
||||
if "ichimoku" not in disabled: self._compute_ichimoku(df, result)
|
||||
if "divergence" not in disabled: self._detect_divergences(df, result)
|
||||
if "candlestick" not in disabled: self._detect_candlestick_patterns(df, result)
|
||||
if "pivot" not in disabled: self._compute_support_resistance(df, result)
|
||||
if "volume" not in disabled: self._compute_volume_analysis(df, result)
|
||||
|
||||
# ── Classify market regime ─────────────────────────────────
|
||||
self._classify_regime(df, result)
|
||||
@@ -215,7 +217,7 @@ class TechnicalEngine:
|
||||
|
||||
result.indicators["rsi"] = IndicatorSignal(
|
||||
name="RSI", value=rsi_val, signal=signal,
|
||||
confidence=conf, description=desc, weight=1.0,
|
||||
confidence=conf, description=desc, weight=0.3,
|
||||
)
|
||||
|
||||
# Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix)
|
||||
@@ -847,10 +849,12 @@ class TechnicalEngine:
|
||||
total_weight = 0.0
|
||||
total_confidence = 0.0
|
||||
|
||||
# Regime-gate ichimoku: trend hero (-23% to disable in 2024 trend) but
|
||||
# chop villain (+18% to disable in 2023 chop & 2026 recent).
|
||||
regime_weights = {
|
||||
"trending_up": {"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "rsi": 0.7, "bollinger": 0.6},
|
||||
"trending_down": {"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "rsi": 0.7, "bollinger": 0.6},
|
||||
"ranging": {"rsi": 1.4, "stochastic": 1.3, "bollinger": 1.4, "ema_trend": 0.6, "adx": 0.5},
|
||||
"trending_up": {"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "ichimoku": 1.3, "rsi": 0.7, "bollinger": 0.6},
|
||||
"trending_down": {"ema_trend": 1.5, "adx": 1.3, "macd": 1.2, "ichimoku": 1.3, "rsi": 0.7, "bollinger": 0.6},
|
||||
"ranging": {"rsi": 1.4, "stochastic": 1.3, "bollinger": 1.4, "ema_trend": 0.6, "adx": 0.5, "ichimoku": 0.2},
|
||||
}
|
||||
|
||||
active_regime_weights = regime_weights.get(result.market_regime, {})
|
||||
|
||||
@@ -306,6 +306,8 @@ Examples:
|
||||
help="Gate signals on higher-TF ADX trend confirmation")
|
||||
parser.add_argument("--regime-min-adx", type=float, default=30.0,
|
||||
help="Min ADX for regime filter (default 30, requires --regime-filter)")
|
||||
parser.add_argument("--ablate", action="store_true",
|
||||
help="Run leave-one-out indicator ablation sweep (parallel)")
|
||||
|
||||
return parser.parse_args()
|
||||
|
||||
@@ -433,6 +435,74 @@ def run_optimize(args, config: PlatformConfig):
|
||||
print_backtest_report(results[0], params=results[0].params, verbose_journal=False)
|
||||
|
||||
|
||||
def run_ablation(args, config: PlatformConfig):
|
||||
"""Leave-one-out ablation: baseline + each indicator disabled, in parallel."""
|
||||
from backtest import Optimizer
|
||||
from config import ALL_INDICATORS
|
||||
|
||||
print_header("USD/JPY INDICATOR ABLATION SWEEP")
|
||||
|
||||
grid = {"disable_indicator": [None] + list(ALL_INDICATORS)}
|
||||
n_combos = len(grid["disable_indicator"])
|
||||
print(f"{Colors.GRAY} Grid: {n_combos} runs (1 baseline + {n_combos-1} ablations) "
|
||||
f"× {config.backtest.parallel_workers} workers{Colors.RESET}\n")
|
||||
|
||||
optimizer = Optimizer(config)
|
||||
results = optimizer.run_grid(grid)
|
||||
|
||||
# Find baseline (disable_indicator=None) for delta computation
|
||||
baseline = next(
|
||||
(r for r in results if not r.params.get("disable_indicator")),
|
||||
None,
|
||||
)
|
||||
base_ret = baseline.metrics.get("total_return_pct", 0.0) if baseline else 0.0
|
||||
base_pf = baseline.metrics.get("profit_factor", 0.0) if baseline else 0.0
|
||||
base_sharpe = baseline.metrics.get("sharpe", 0.0) if baseline else 0.0
|
||||
|
||||
# Sort: baseline first, then ablations by impact (largest drop = most useful indicator)
|
||||
ablations = [r for r in results if r is not baseline]
|
||||
ablations.sort(
|
||||
key=lambda r: r.metrics.get("total_return_pct", 0.0) - base_ret
|
||||
)
|
||||
|
||||
print(f" {Colors.BOLD}Indicator Ablation Report{Colors.RESET}")
|
||||
print(f" {Colors.GRAY}{'─' * 88}{Colors.RESET}")
|
||||
print(f" {'Disabled':<14} {'Return%':>9} {'ΔRet%':>8} {'PF':>6} {'ΔPF':>7} "
|
||||
f"{'Sharpe':>7} {'ΔSh':>7} {'Trades':>7}")
|
||||
print(f" {Colors.GRAY}{'─' * 88}{Colors.RESET}")
|
||||
|
||||
def fmt_row(label, m, deltas=True):
|
||||
ret = m.get("total_return_pct", 0.0)
|
||||
pf = m.get("profit_factor", 0.0)
|
||||
sh = m.get("sharpe", 0.0)
|
||||
tr = m.get("total_trades", 0)
|
||||
ret_color = Colors.GREEN if ret >= 0 else Colors.RED
|
||||
pf_str = "inf" if pf == float("inf") else f"{pf:.2f}"
|
||||
if deltas:
|
||||
d_ret = ret - base_ret
|
||||
d_pf = pf - base_pf if pf != float("inf") else 0.0
|
||||
d_sh = sh - base_sharpe
|
||||
d_ret_color = Colors.RED if d_ret < 0 else Colors.GREEN
|
||||
print(f" {label:<14} {ret_color}{ret:>+8.2f}%{Colors.RESET} "
|
||||
f"{d_ret_color}{d_ret:>+7.2f}%{Colors.RESET} "
|
||||
f"{pf_str:>6} {d_pf:>+7.2f} "
|
||||
f"{sh:>7.2f} {d_sh:>+7.2f} {tr:>7}")
|
||||
else:
|
||||
print(f" {Colors.BOLD}{label:<14}{Colors.RESET} "
|
||||
f"{ret_color}{ret:>+8.2f}%{Colors.RESET} {'':>8} "
|
||||
f"{pf_str:>6} {'':>7} {sh:>7.2f} {'':>7} {tr:>7}")
|
||||
|
||||
if baseline:
|
||||
fmt_row("baseline", baseline.metrics, deltas=False)
|
||||
print(f" {Colors.GRAY}{'─' * 88}{Colors.RESET}")
|
||||
for r in ablations:
|
||||
fmt_row(str(r.params.get("disable_indicator")), r.metrics)
|
||||
|
||||
print(f"\n {Colors.GRAY}Indicators ranked by importance — most negative ΔRet means "
|
||||
f"removing it hurt most.{Colors.RESET}")
|
||||
print(f" {Colors.GRAY}Positive ΔRet means the indicator is *hurting* the system.{Colors.RESET}\n")
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
|
||||
@@ -445,7 +515,9 @@ def main():
|
||||
|
||||
config = build_config(args)
|
||||
|
||||
if args.optimize:
|
||||
if args.ablate:
|
||||
run_ablation(args, config)
|
||||
elif args.optimize:
|
||||
run_optimize(args, config)
|
||||
elif args.backtest:
|
||||
run_backtest(args, config)
|
||||
|
||||
Reference in New Issue
Block a user