Harwinder
2e5d66db1b
Session filter restricts trades to a UTC hour window (default 07-16, London + NY) — opt-in via --session-filter, with --session-start / --session-end overrides. Wired into both the backtester (cheap hour check next to the regime gate) and live mode (downgrades signal to NEUTRAL/F when current UTC hour is outside the window). Auto-toggle (--auto-session-filter) tracks rolling mean pnl_pct over the last N closed trades; when it falls below threshold, the session filter activates, releasing once expectancy recovers. Default lookback 10 trades, threshold -0.3%. Trade dataclass extended with diagnostic fields captured at entry: entry_hour, entry_dow, ADX, ATR%, market_regime, volatility_regime, mtf_agreement. These let --journal-csv exports surface loss-pattern clusters when a window underperforms. Validated on 2024 trend / 2023 chop / 2026 recent windows. Auto-toggle preserves ~50% of trend edge while still cutting ~25% of recent-regime losses — pure regime filter is still best on cumulative, so both session flags ship default-off.
575 lines
25 KiB
Python
575 lines
25 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ForexBot2 — USD/JPY Signal Intelligence (single-pair)
|
||
=======================================================
|
||
Entry point for both live signal analysis and walk-forward backtesting.
|
||
|
||
Usage:
|
||
python main.py # Live signal on H1
|
||
python main.py --timeframe H4 --detail # Detailed H4 signal
|
||
python main.py --json # JSON output
|
||
python main.py --backtest # Walk-forward backtest
|
||
python main.py --backtest --optimize # Parallel grid search
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List
|
||
|
||
# Force UTF-8 output on Windows so box-drawing chars render
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
if hasattr(sys.stderr, "reconfigure"):
|
||
sys.stderr.reconfigure(encoding="utf-8")
|
||
|
||
from config import PlatformConfig, Timeframe, SignalStrength, USDJPY, DISPLAY
|
||
from signal_aggregator import SignalAggregator, SignalReport
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# TERMINAL FORMATTING
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
class Colors:
|
||
RESET = "\033[0m"
|
||
BOLD = "\033[1m"
|
||
DIM = "\033[2m"
|
||
UNDERLINE = "\033[4m"
|
||
RED = "\033[91m"
|
||
GREEN = "\033[92m"
|
||
YELLOW = "\033[93m"
|
||
BLUE = "\033[94m"
|
||
MAGENTA = "\033[95m"
|
||
CYAN = "\033[96m"
|
||
WHITE = "\033[97m"
|
||
GRAY = "\033[90m"
|
||
|
||
|
||
def signal_color(signal: SignalStrength) -> str:
|
||
return {
|
||
SignalStrength.STRONG_BUY: Colors.GREEN + Colors.BOLD,
|
||
SignalStrength.BUY: Colors.GREEN,
|
||
SignalStrength.NEUTRAL: Colors.GRAY,
|
||
SignalStrength.SELL: Colors.RED,
|
||
SignalStrength.STRONG_SELL: Colors.RED + Colors.BOLD,
|
||
}.get(signal, Colors.RESET)
|
||
|
||
|
||
def quality_color(q: str) -> str:
|
||
return {
|
||
"A+": Colors.GREEN + Colors.BOLD,
|
||
"A": Colors.GREEN,
|
||
"B": Colors.CYAN,
|
||
"C": Colors.YELLOW,
|
||
"D": Colors.GRAY,
|
||
"F": Colors.DIM,
|
||
}.get(q, Colors.RESET)
|
||
|
||
|
||
def print_header(title: str = "USD/JPY SIGNAL INTELLIGENCE"):
|
||
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
|
||
print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}")
|
||
print(f"{Colors.BOLD}{Colors.CYAN} {title}{Colors.RESET}")
|
||
print(f"{Colors.GRAY} Time: {now}{Colors.RESET}")
|
||
print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n")
|
||
|
||
|
||
# ── Live signal renderer ───────────────────────────────────────────
|
||
|
||
_INDICATOR_PREFIXES = (
|
||
"EMA Trend", "RSI", "MACD", "Bollinger Bands", "Stochastic",
|
||
"ADX", "Ichimoku", "Divergence", "Candlestick", "Pivot Points",
|
||
"News", "Regime",
|
||
)
|
||
|
||
|
||
def _compact_reason(reason: str) -> str:
|
||
"""Shorten a verbose reason string for the compact view."""
|
||
s = reason
|
||
if s.startswith("STRONG: "):
|
||
s = s[8:]
|
||
# Strip the leading "<Indicator>: " prefix — it's always an indicator name
|
||
for prefix in _INDICATOR_PREFIXES:
|
||
if s.startswith(prefix + ": "):
|
||
s = s[len(prefix) + 2:]
|
||
break
|
||
# Drop trailing detail after first ' | '
|
||
if " | " in s:
|
||
s = s.split(" | ", 1)[0]
|
||
return s.strip()
|
||
|
||
|
||
def print_signal_report(report: SignalReport, detail: bool = False):
|
||
sc = signal_color(report.final_signal)
|
||
qc = quality_color(report.signal_quality)
|
||
sig_name = report.final_signal.name.replace("_", " ")
|
||
|
||
# ── 1. DECISION ────────────────────────────────────────────────
|
||
mtf_tag = (f"{Colors.GREEN}MTF✓{Colors.RESET}" if report.mtf_agreement
|
||
else f"{Colors.YELLOW}MTF✗{Colors.RESET}")
|
||
print(f" {sc}▶ {sig_name}{Colors.RESET} "
|
||
f"{qc}{report.signal_quality}{Colors.RESET} · "
|
||
f"{sc}{report.final_score:+.1f}{Colors.RESET} · "
|
||
f"{report.final_confidence:.0f}% · {mtf_tag}")
|
||
print(f" {Colors.GRAY}{report.action_recommendation}{Colors.RESET}")
|
||
|
||
# ── 2. TRADE SETUP (if signal is actionable) ───────────────────
|
||
if report.trade_setup:
|
||
ts = report.trade_setup
|
||
dc = Colors.GREEN if ts.direction == "BUY" else Colors.RED
|
||
print(f"\n {Colors.BOLD}Setup{Colors.RESET} "
|
||
f"{dc}{ts.direction}{Colors.RESET} @ {Colors.BOLD}{ts.entry_price:.3f}{Colors.RESET} "
|
||
f"→ SL {Colors.RED}{ts.stop_loss:.3f}{Colors.RESET} "
|
||
f"({ts.risk_pips:.1f}p) "
|
||
f"→ TP {Colors.GREEN}{ts.take_profit_1:.3f}{Colors.RESET} "
|
||
f"({ts.reward_pips:.1f}p) "
|
||
f"RR 1:{ts.risk_reward:.1f}")
|
||
|
||
# ── 3. CONTEXT (price · regime · vol · MTF breakdown) ──────────
|
||
print(f"\n {Colors.WHITE}{report.display_name}{Colors.RESET} "
|
||
f"{Colors.BOLD}{report.current_price:.3f}{Colors.RESET} · "
|
||
f"{Colors.CYAN}{report.market_regime}{Colors.RESET} · "
|
||
f"vol {report.volatility_regime}")
|
||
|
||
if report.mtf_signals:
|
||
parts = []
|
||
for tf, sig in report.mtf_signals.items():
|
||
tf_sc = signal_color(sig)
|
||
sc_val = report.mtf_scores.get(tf, 0)
|
||
parts.append(f"{tf} {tf_sc}{sig.name.replace('_',' ')}{Colors.RESET} {sc_val:+.1f}")
|
||
print(f" {Colors.GRAY}MTF{Colors.RESET} " + " · ".join(parts))
|
||
|
||
# ── 4. WHY — top 3 each side, compact ──────────────────────────
|
||
bulls = [_compact_reason(r) for r in report.bullish_reasons[:3]]
|
||
bears = [_compact_reason(r) for r in report.bearish_reasons[:3]]
|
||
if bulls or bears:
|
||
print()
|
||
rows = max(len(bulls), len(bears))
|
||
for i in range(rows):
|
||
left = f"{Colors.GREEN}+{Colors.RESET} {bulls[i]}" if i < len(bulls) else ""
|
||
right = f"{Colors.RED}-{Colors.RESET} {bears[i]}" if i < len(bears) else ""
|
||
print(f" {left:<60} {right}")
|
||
|
||
# ── 5. WARNINGS / NEWS / EVENTS ────────────────────────────────
|
||
if report.warnings:
|
||
for w in report.warnings:
|
||
print(f" {Colors.YELLOW}⚠{Colors.RESET} {w}")
|
||
if report.event_description and report.event_risk != "clear":
|
||
print(f" {Colors.YELLOW}📅{Colors.RESET} {report.event_description}")
|
||
if report.news_description and report.news_description != "Neutral news sentiment":
|
||
print(f" {Colors.MAGENTA}📰{Colors.RESET} {report.news_description}")
|
||
|
||
# ── 6. DETAIL (only with --detail) ─────────────────────────────
|
||
if detail and report.primary_analysis:
|
||
print(f"\n {Colors.BOLD}Indicators ({report.primary_analysis.timeframe}){Colors.RESET}")
|
||
for name, ind in report.primary_analysis.indicators.items():
|
||
if name.startswith("_") or ind.weight == 0:
|
||
continue
|
||
isc = signal_color(ind.signal)
|
||
print(f" {ind.name:<16} {isc}{ind.signal.name.replace('_',' '):<11}{Colors.RESET} "
|
||
f"{ind.confidence:>3.0f}% {Colors.GRAY}│ {ind.description}{Colors.RESET}")
|
||
|
||
levels = report.primary_analysis.key_levels
|
||
if levels:
|
||
shown = [lvl for lvl in ["r3", "r2", "r1", "pivot", "s1", "s2", "s3"] if lvl in levels]
|
||
if shown:
|
||
print(f"\n {Colors.BOLD}Pivot Levels{Colors.RESET}")
|
||
for lvl in shown:
|
||
val = levels[lvl]
|
||
color = Colors.RED if val > report.current_price else Colors.GREEN
|
||
print(f" {lvl.upper():>6} {color}{val:.3f}{Colors.RESET}")
|
||
|
||
|
||
def report_to_dict(report: SignalReport) -> dict:
|
||
d = {
|
||
"symbol": report.symbol,
|
||
"display_name": report.display_name,
|
||
"timestamp": report.timestamp,
|
||
"current_price": report.current_price,
|
||
"signal": report.final_signal.name,
|
||
"score": round(report.final_score, 2),
|
||
"confidence": round(report.final_confidence, 2),
|
||
"quality": report.signal_quality,
|
||
"action": report.action_recommendation,
|
||
"market_regime": report.market_regime,
|
||
"volatility_regime": report.volatility_regime,
|
||
"mtf_agreement": report.mtf_agreement,
|
||
"mtf_signals": {k: v.name for k, v in report.mtf_signals.items()},
|
||
"event_risk": report.event_risk,
|
||
"event_description": report.event_description,
|
||
"news_sentiment": round(report.news_sentiment, 3),
|
||
"bullish_reasons": report.bullish_reasons,
|
||
"bearish_reasons": report.bearish_reasons,
|
||
"warnings": report.warnings,
|
||
}
|
||
if report.trade_setup:
|
||
ts = report.trade_setup
|
||
d["trade_setup"] = {
|
||
"direction": ts.direction,
|
||
"entry": ts.entry_price,
|
||
"stop_loss": ts.stop_loss,
|
||
"take_profit_1": ts.take_profit_1,
|
||
"take_profit_2": ts.take_profit_2,
|
||
"risk_reward": ts.risk_reward,
|
||
"risk_pips": round(ts.risk_pips, 1),
|
||
"reward_pips": round(ts.reward_pips, 1),
|
||
}
|
||
if report.primary_analysis:
|
||
d["indicators"] = {
|
||
name: {
|
||
"signal": ind.signal.name,
|
||
"confidence": round(ind.confidence, 1),
|
||
"description": ind.description,
|
||
}
|
||
for name, ind in report.primary_analysis.indicators.items()
|
||
if not name.startswith("_")
|
||
}
|
||
d["key_levels"] = {
|
||
k: round(v, 5) for k, v in report.primary_analysis.key_levels.items()
|
||
}
|
||
return d
|
||
|
||
|
||
# ── Backtest renderer ──────────────────────────────────────────────
|
||
|
||
def print_backtest_report(result, params: dict = None, verbose_journal: bool = False):
|
||
m = result.metrics
|
||
print(f" {Colors.BOLD}{Colors.WHITE}Backtest Summary — USD/JPY{Colors.RESET}")
|
||
if params:
|
||
params_str = ", ".join(f"{k}={v}" for k, v in params.items())
|
||
print(f" {Colors.GRAY}Params: {params_str}{Colors.RESET}")
|
||
print(f" {Colors.GRAY}{'─' * 76}{Colors.RESET}")
|
||
|
||
initial = result.initial_balance
|
||
final = result.final_balance
|
||
ret_pct = m.get("total_return_pct", 0)
|
||
ret_color = Colors.GREEN if ret_pct >= 0 else Colors.RED
|
||
|
||
print(f" Trades: {Colors.BOLD}{m.get('total_trades', 0)}{Colors.RESET}"
|
||
f" ({m.get('wins', 0)}W / {m.get('losses', 0)}L)")
|
||
print(f" Win Rate: {m.get('win_rate', 0):.1f}%")
|
||
print(f" Profit Factor: {m.get('profit_factor', 0):.2f}")
|
||
pf_inf = m.get('profit_factor', 0) == float('inf')
|
||
if pf_inf:
|
||
print(f" (no losing trades — infinite PF)")
|
||
print(f" Expectancy: {m.get('expectancy_pips', 0):+.2f} pips/trade"
|
||
f" ({m.get('expectancy_per_trade', 0):+.2f} per trade)")
|
||
print(f" Initial Balance: ${initial:,.2f}")
|
||
print(f" Final Balance: {ret_color}${final:,.2f}{Colors.RESET}")
|
||
print(f" Total Return: {ret_color}{ret_pct:+.2f}%{Colors.RESET}")
|
||
print(f" CAGR: {m.get('cagr', 0):+.2f}%")
|
||
print(f" Max Drawdown: {Colors.RED}{m.get('max_drawdown_pct', 0):.2f}%{Colors.RESET}")
|
||
print(f" Sharpe Ratio: {m.get('sharpe', 0):.2f}")
|
||
print(f" Sortino Ratio: {m.get('sortino', 0):.2f}")
|
||
print(f" Avg Duration: {m.get('avg_trade_duration_bars', 0):.1f} bars")
|
||
print(f" Win Streak: {m.get('longest_winning_streak', 0)} | "
|
||
f"Loss Streak: {m.get('longest_losing_streak', 0)}")
|
||
print(f" {Colors.GRAY}{'─' * 76}{Colors.RESET}")
|
||
|
||
if verbose_journal and result.trades:
|
||
print(f"\n {Colors.BOLD}Trade Journal (last 10):{Colors.RESET}")
|
||
print(f" {'#':<4} {'Entry Time':<22} {'Dir':<5} {'Entry':<9} "
|
||
f"{'Exit':<9} {'Reason':<6} {'Pips':>8} {'P&L%':>7}")
|
||
for i, t in enumerate(result.trades[-10:], start=max(1, len(result.trades) - 9)):
|
||
color = Colors.GREEN if t.pnl_currency > 0 else Colors.RED
|
||
entry_str = str(t.entry_time)[:19]
|
||
print(f" {i:<4} {entry_str:<22} {t.direction:<5} {t.entry_price:<9.3f} "
|
||
f"{t.exit_price:<9.3f} {t.exit_reason:<6} "
|
||
f"{color}{t.pnl_pips:>+8.1f}{Colors.RESET} "
|
||
f"{color}{t.pnl_pct:>+6.2f}%{Colors.RESET}")
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════
|
||
# CLI
|
||
# ══════════════════════════════════════════════════════════════════
|
||
|
||
def parse_args():
|
||
parser = argparse.ArgumentParser(
|
||
description="ForexBot2 — USD/JPY Signal Intelligence + Backtester",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
Examples:
|
||
python main.py Live signal (H1, summary)
|
||
python main.py --detail Live signal with full breakdown
|
||
python main.py --timeframe H4 --detail H4 timeframe
|
||
python main.py --json JSON output
|
||
python main.py --backtest Walk-forward backtest (full history)
|
||
python main.py --backtest --bars 2000 Backtest on last 2000 bars
|
||
python main.py --backtest --optimize Parallel grid search across CPU cores
|
||
""",
|
||
)
|
||
parser.add_argument("--timeframe", "-tf", default="H1",
|
||
choices=["M5", "M15", "M30", "H1", "H4", "D1"],
|
||
help="Primary analysis timeframe (default: H1)")
|
||
parser.add_argument("--detail", action="store_true",
|
||
help="Show full indicator breakdown (live mode)")
|
||
parser.add_argument("--json", action="store_true",
|
||
help="Output as JSON")
|
||
parser.add_argument("--verbose", "-v", action="store_true",
|
||
help="Debug logging")
|
||
|
||
# Backtest options
|
||
parser.add_argument("--backtest", action="store_true",
|
||
help="Run walk-forward backtest instead of live signal")
|
||
parser.add_argument("--optimize", action="store_true",
|
||
help="Run parallel parameter grid search (implies --backtest)")
|
||
parser.add_argument("--bars", type=int, default=5000,
|
||
help="Max bars of history (default 5000 — TD ceiling)")
|
||
parser.add_argument("--start", default=None,
|
||
help="Backtest window start (YYYY-MM-DD)")
|
||
parser.add_argument("--end", default=None,
|
||
help="Backtest window end (YYYY-MM-DD)")
|
||
parser.add_argument("--min-grade", default="B",
|
||
choices=["A+", "A", "B", "C", "D", "F"],
|
||
help="Minimum grade to open backtest trade (default B)")
|
||
parser.add_argument("--journal", action="store_true",
|
||
help="Print last 10 trades from backtest journal")
|
||
parser.add_argument("--journal-csv", default=None,
|
||
help="Write full trade journal to CSV path")
|
||
parser.add_argument("--regime-filter", action="store_true",
|
||
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("--session-filter", action="store_true",
|
||
help="Only trade during a UTC hour window (default 07-16, London + NY)")
|
||
parser.add_argument("--session-start", type=int, default=7,
|
||
help="Session start hour UTC, inclusive (default 7)")
|
||
parser.add_argument("--session-end", type=int, default=16,
|
||
help="Session end hour UTC, exclusive (default 16)")
|
||
parser.add_argument("--auto-session-filter", action="store_true",
|
||
help="Auto-toggle session filter when recent trades show adverse mean pnl%%")
|
||
parser.add_argument("--auto-session-lookback", type=int, default=10,
|
||
help="Lookback in closed trades for auto-session detection (default 10)")
|
||
parser.add_argument("--auto-session-threshold-pct", type=float, default=-0.3,
|
||
help="Mean pnl%% threshold below which auto-session activates (default -0.3)")
|
||
parser.add_argument("--ablate", action="store_true",
|
||
help="Run leave-one-out indicator ablation sweep (parallel)")
|
||
|
||
return parser.parse_args()
|
||
|
||
|
||
def build_config(args) -> PlatformConfig:
|
||
config = PlatformConfig()
|
||
|
||
tf_map = {
|
||
"M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30,
|
||
"H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1,
|
||
}
|
||
config.primary_timeframe = tf_map[args.timeframe]
|
||
|
||
tf_confirmations = {
|
||
"M5": [Timeframe.M1, Timeframe.M15],
|
||
"M15": [Timeframe.M5, Timeframe.H1],
|
||
"M30": [Timeframe.M15, Timeframe.H1],
|
||
"H1": [Timeframe.M15, Timeframe.H4],
|
||
"H4": [Timeframe.H1, Timeframe.D1],
|
||
"D1": [Timeframe.H4, Timeframe.W1],
|
||
}
|
||
config.confirmation_timeframes = tf_confirmations.get(
|
||
args.timeframe, [Timeframe.M15, Timeframe.H4]
|
||
)
|
||
|
||
# Backtest config
|
||
config.backtest.start_date = args.start
|
||
config.backtest.end_date = args.end
|
||
config.backtest.max_bars = args.bars
|
||
config.backtest.min_trade_grade = args.min_grade
|
||
config.backtest.optimize = args.optimize
|
||
config.backtest.regime_filter = args.regime_filter
|
||
config.backtest.regime_filter_min_adx = args.regime_min_adx
|
||
config.backtest.session_filter = args.session_filter
|
||
config.backtest.session_start_hour = args.session_start
|
||
config.backtest.session_end_hour = args.session_end
|
||
config.backtest.auto_session_filter = args.auto_session_filter
|
||
config.backtest.auto_session_lookback_trades = args.auto_session_lookback
|
||
config.backtest.auto_session_threshold_pct = args.auto_session_threshold_pct
|
||
|
||
return config
|
||
|
||
|
||
def run_live(args, config: PlatformConfig):
|
||
print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n")
|
||
aggregator = SignalAggregator(config)
|
||
report = aggregator.analyze_pair()
|
||
|
||
if args.json:
|
||
print(json.dumps(report_to_dict(report), indent=2))
|
||
return
|
||
|
||
print_header("USD/JPY SIGNAL INTELLIGENCE")
|
||
print_signal_report(report, detail=args.detail)
|
||
print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}")
|
||
print(f"{Colors.GRAY} Disclaimer: For educational/research purposes only.{Colors.RESET}")
|
||
print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n")
|
||
|
||
|
||
def run_backtest(args, config: PlatformConfig):
|
||
from backtest import WalkForwardBacktester
|
||
from metrics import trades_to_dataframe
|
||
|
||
print_header("USD/JPY WALK-FORWARD BACKTEST")
|
||
print(f"{Colors.GRAY} Loading data and replaying bars...{Colors.RESET}\n")
|
||
|
||
backtester = WalkForwardBacktester(config)
|
||
result = backtester.run()
|
||
|
||
if args.json:
|
||
out = {
|
||
"metrics": result.metrics,
|
||
"initial_balance": result.initial_balance,
|
||
"final_balance": result.final_balance,
|
||
"trades": [
|
||
{
|
||
"entry_time": str(t.entry_time),
|
||
"exit_time": str(t.exit_time),
|
||
"direction": t.direction,
|
||
"entry_price": t.entry_price,
|
||
"exit_price": t.exit_price,
|
||
"exit_reason": t.exit_reason,
|
||
"pnl_pips": t.pnl_pips,
|
||
"pnl_pct": t.pnl_pct,
|
||
"grade": t.grade,
|
||
}
|
||
for t in result.trades
|
||
],
|
||
}
|
||
print(json.dumps(out, indent=2))
|
||
else:
|
||
print_backtest_report(result, verbose_journal=args.journal)
|
||
|
||
if args.journal_csv and result.trades:
|
||
df = trades_to_dataframe(result.trades)
|
||
df.to_csv(args.journal_csv, index=False)
|
||
print(f"\n {Colors.GREEN}✓ Trade journal written to {args.journal_csv}{Colors.RESET}")
|
||
|
||
|
||
def run_optimize(args, config: PlatformConfig):
|
||
from backtest import Optimizer
|
||
|
||
print_header("USD/JPY BACKTEST PARAMETER OPTIMIZER")
|
||
grid = config.backtest.param_grid
|
||
n_combos = 1
|
||
for v in grid.values():
|
||
n_combos *= len(v)
|
||
print(f"{Colors.GRAY} Grid: {n_combos} combos × {config.backtest.parallel_workers} workers{Colors.RESET}\n")
|
||
|
||
optimizer = Optimizer(config)
|
||
results = optimizer.run_grid(grid)
|
||
|
||
print(f"\n {Colors.BOLD}Top 10 Parameter Combinations (by total return):{Colors.RESET}\n")
|
||
print(f" {'#':<3} {'Return%':>8} {'PF':>6} {'WinR':>6} {'Trades':>7} "
|
||
f"{'MaxDD%':>7} {'Sharpe':>7} Params")
|
||
print(f" {Colors.GRAY}{'─' * 90}{Colors.RESET}")
|
||
|
||
for i, r in enumerate(results[:10], start=1):
|
||
m = r.metrics
|
||
ret = m.get("total_return_pct", 0)
|
||
ret_color = Colors.GREEN if ret >= 0 else Colors.RED
|
||
params_str = ", ".join(f"{k}={v}" for k, v in r.params.items())
|
||
pf = m.get("profit_factor", 0)
|
||
pf_str = "inf" if pf == float("inf") else f"{pf:.2f}"
|
||
print(f" {i:<3} {ret_color}{ret:>+7.2f}%{Colors.RESET} {pf_str:>6} "
|
||
f"{m.get('win_rate', 0):>5.1f}% {m.get('total_trades', 0):>7} "
|
||
f"{m.get('max_drawdown_pct', 0):>+6.2f}% {m.get('sharpe', 0):>7.2f}"
|
||
f" {Colors.GRAY}{params_str}{Colors.RESET}")
|
||
|
||
print(f"\n {Colors.BOLD}Best combination details:{Colors.RESET}\n")
|
||
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()
|
||
|
||
log_level = logging.DEBUG if args.verbose else logging.WARNING
|
||
logging.basicConfig(
|
||
level=log_level,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
|
||
config = build_config(args)
|
||
|
||
if args.ablate:
|
||
run_ablation(args, config)
|
||
elif args.optimize:
|
||
run_optimize(args, config)
|
||
elif args.backtest:
|
||
run_backtest(args, config)
|
||
else:
|
||
run_live(args, config)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|