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.
This commit is contained in:
Harwinder
2026-05-07 17:33:07 -04:00
parent 045f4248f3
commit ce76684a62
4 changed files with 114 additions and 17 deletions
+73 -1
View File
@@ -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)