"""Performance metrics and reporting for the intraday backtester. All metric functions are pure (no side effects). Uses rich for terminal output formatting. """ from __future__ import annotations import hashlib import json import math import statistics import uuid from datetime import datetime from pathlib import Path from typing import Any from libs.intraday.domain import ( DayResult, IntradayConfig, IntradayMetrics, IntradayTrade, SweepResult, ) # ── Metric Computation ───────────────────────────────────────────────────── def _get_initial_capital(config: IntradayConfig) -> float: """Get initial capital from the active strategy (ORB or momentum).""" if getattr(config, "strategy_mode", "momentum") == "orb" and config.orb_strategy: return config.orb_strategy.initial_capital return config.strategy.initial_capital def _idle_sleeve_open_count(day_result: DayResult) -> int: diagnostics = day_result.entry_diagnostics or {} try: return max(0, int(diagnostics.get("orb_idle_sleeve_positions_opened") or 0)) except (TypeError, ValueError): return 0 def _idle_sleeve_open_symbols(day_result: DayResult) -> list[str]: diagnostics = day_result.entry_diagnostics or {} raw = diagnostics.get("orb_idle_sleeve_symbols_opened") or [] if not isinstance(raw, list): return [] return [str(symbol) for symbol in raw if str(symbol)] def _idle_sleeve_open_labels(day_result: DayResult) -> list[str]: diagnostics = day_result.entry_diagnostics or {} raw = diagnostics.get("orb_idle_sleeve_labels_opened") or [] if not isinstance(raw, list): return [] return [str(label) for label in raw if str(label)] def _same_day_idle_trade_count(day_result: DayResult) -> int: return sum( 1 for trade in day_result.trades if getattr(trade, "trigger_type", None) == "orb_idle_sleeve" and getattr(trade, "exit_reason", None) == "idle_sleeve_same_day_close" ) def _day_activity_count(day_result: DayResult) -> int: pending_idle_open_count = max( 0, _idle_sleeve_open_count(day_result) - _same_day_idle_trade_count(day_result), ) return len(day_result.trades) + pending_idle_open_count class IntradayMetricsAccumulator: """Streaming metrics accumulator for bounded-memory intraday research runs.""" def __init__(self, config: IntradayConfig, run_id: str = "") -> None: self.config = config self.run_id = run_id or str(uuid.uuid4())[:8] self.initial_capital = _get_initial_capital(config) self.is_orb = getattr(config, "strategy_mode", "momentum") == "orb" self.active_strategy = ( config.orb_strategy if (self.is_orb and config.orb_strategy) else config.strategy ) self.n_days = 0 self.days_with_trades = 0 self.days_with_activity = 0 self.start_date = "" self.end_date = "" self.total_trades = 0 self.stop_loss_exits = 0 self.idle_sleeve_entry_days = 0 self.idle_sleeve_positions_opened = 0 self.win_count = 0 self.loss_count = 0 self.sum_win_pct = 0.0 self.sum_loss_pct = 0.0 self.gross_profit = 0.0 self.gross_loss = 0.0 self.hold_minutes_sum = 0.0 self.hold_minutes_count = 0 self.daily_returns: list[float] = [] self.equity = self.initial_capital self.max_equity = self.initial_capital self.max_drawdown = 0.0 def update(self, day_result: DayResult) -> None: if not self.start_date: self.start_date = day_result.date self.end_date = day_result.date self.n_days += 1 self.daily_returns.append(day_result.daily_return_pct) if day_result.trades: self.days_with_trades += 1 idle_open_count = _idle_sleeve_open_count(day_result) if idle_open_count > 0: self.idle_sleeve_entry_days += 1 self.idle_sleeve_positions_opened += idle_open_count if day_result.trades or idle_open_count > 0: self.days_with_activity += 1 for trade in day_result.trades: self.total_trades += 1 if trade.exit_reason == "stop_loss": self.stop_loss_exits += 1 if trade.pnl > 0: self.win_count += 1 self.sum_win_pct += trade.pnl_pct self.gross_profit += trade.pnl else: self.loss_count += 1 self.sum_loss_pct += trade.pnl_pct self.gross_loss += abs(trade.pnl) try: entry = datetime.fromisoformat(trade.entry_time.replace("Z", "+00:00")) exit_ = datetime.fromisoformat(trade.exit_time.replace("Z", "+00:00")) self.hold_minutes_sum += (exit_ - entry).total_seconds() / 60 self.hold_minutes_count += 1 except Exception: pass self.equity += day_result.daily_pnl self.max_equity = max(self.max_equity, self.equity) if self.max_equity > 0: dd = (self.equity - self.max_equity) / self.max_equity self.max_drawdown = min(self.max_drawdown, dd) def extend(self, day_results: list[DayResult]) -> None: for day_result in day_results: self.update(day_result) def snapshot(self) -> dict[str, Any]: """Serialize accumulator state for chunk-level checkpoint/resume.""" return { "run_id": self.run_id, "initial_capital": self.initial_capital, "n_days": self.n_days, "days_with_trades": self.days_with_trades, "days_with_activity": self.days_with_activity, "start_date": self.start_date, "end_date": self.end_date, "total_trades": self.total_trades, "stop_loss_exits": self.stop_loss_exits, "idle_sleeve_entry_days": self.idle_sleeve_entry_days, "idle_sleeve_positions_opened": self.idle_sleeve_positions_opened, "win_count": self.win_count, "loss_count": self.loss_count, "sum_win_pct": self.sum_win_pct, "sum_loss_pct": self.sum_loss_pct, "gross_profit": self.gross_profit, "gross_loss": self.gross_loss, "hold_minutes_sum": self.hold_minutes_sum, "hold_minutes_count": self.hold_minutes_count, "daily_returns": list(self.daily_returns), "equity": self.equity, "max_equity": self.max_equity, "max_drawdown": self.max_drawdown, } @classmethod def from_snapshot( cls, config: IntradayConfig, snapshot: dict[str, Any], *, run_id: str = "", ) -> "IntradayMetricsAccumulator": """Restore a previously serialized accumulator state.""" accumulator = cls(config, run_id=run_id or snapshot.get("run_id", "")) accumulator.initial_capital = float(snapshot.get("initial_capital", accumulator.initial_capital)) accumulator.n_days = int(snapshot.get("n_days", 0)) accumulator.days_with_trades = int(snapshot.get("days_with_trades", 0)) accumulator.days_with_activity = int(snapshot.get("days_with_activity", 0)) accumulator.start_date = snapshot.get("start_date", "") or "" accumulator.end_date = snapshot.get("end_date", "") or "" accumulator.total_trades = int(snapshot.get("total_trades", 0)) accumulator.stop_loss_exits = int(snapshot.get("stop_loss_exits", 0)) accumulator.idle_sleeve_entry_days = int(snapshot.get("idle_sleeve_entry_days", 0)) accumulator.idle_sleeve_positions_opened = int( snapshot.get("idle_sleeve_positions_opened", 0) ) accumulator.win_count = int(snapshot.get("win_count", 0)) accumulator.loss_count = int(snapshot.get("loss_count", 0)) accumulator.sum_win_pct = float(snapshot.get("sum_win_pct", 0.0)) accumulator.sum_loss_pct = float(snapshot.get("sum_loss_pct", 0.0)) accumulator.gross_profit = float(snapshot.get("gross_profit", 0.0)) accumulator.gross_loss = float(snapshot.get("gross_loss", 0.0)) accumulator.hold_minutes_sum = float(snapshot.get("hold_minutes_sum", 0.0)) accumulator.hold_minutes_count = int(snapshot.get("hold_minutes_count", 0)) accumulator.daily_returns = [ float(value) for value in snapshot.get("daily_returns", []) ] accumulator.equity = float(snapshot.get("equity", accumulator.initial_capital)) accumulator.max_equity = float(snapshot.get("max_equity", accumulator.initial_capital)) accumulator.max_drawdown = float(snapshot.get("max_drawdown", 0.0)) return accumulator def finalize(self) -> IntradayMetrics: if self.total_trades == 0: return IntradayMetrics( run_id=self.run_id, params_hash=_hash_strategy(self.active_strategy), start_date=self.start_date, end_date=self.end_date, trading_days=self.n_days, days_with_trades=self.days_with_trades, days_with_activity=self.days_with_activity, total_trades=0, stop_loss_exits=0, idle_sleeve_entry_days=self.idle_sleeve_entry_days, idle_sleeve_positions_opened=self.idle_sleeve_positions_opened, total_return_pct=0.0 if self.n_days > 0 else None, annualized_return_pct=0.0 if self.n_days > 0 else None, avg_daily_return_pct=round(statistics.mean(self.daily_returns), 6) if self.daily_returns else None, max_drawdown_pct=0.0 if self.n_days > 0 else None, initial_capital=self.initial_capital, final_equity=round(self.equity, 2), ) win_rate = self.win_count / self.total_trades if self.total_trades else None avg_win_pct = self.sum_win_pct / self.win_count if self.win_count > 0 else None avg_loss_pct = self.sum_loss_pct / self.loss_count if self.loss_count > 0 else None profit_factor = ( self.gross_profit / self.gross_loss if self.gross_loss > 0 else None ) expectancy_pct = ( (win_rate * avg_win_pct + (1 - win_rate) * avg_loss_pct) if win_rate is not None and avg_win_pct is not None and avg_loss_pct is not None else None ) total_return_pct = ( (self.equity - self.initial_capital) / self.initial_capital if self.initial_capital > 0 else None ) annualized = ( total_return_pct * (252 / self.n_days) if total_return_pct is not None and self.n_days > 0 else None ) avg_daily = statistics.mean(self.daily_returns) if self.daily_returns else None sharpe = sortino = calmar = None if len(self.daily_returns) >= 5: try: mean_r = statistics.mean(self.daily_returns) std_r = statistics.stdev(self.daily_returns) if std_r > 0: sharpe = (mean_r / std_r) * math.sqrt(252) down_devs = [r for r in self.daily_returns if r < 0] if down_devs: downside_std = math.sqrt( sum(r ** 2 for r in down_devs) / len(self.daily_returns) ) if downside_std > 0: sortino = (mean_r / downside_std) * math.sqrt(252) except Exception: pass if annualized is not None and self.max_drawdown < 0: calmar = annualized / abs(self.max_drawdown) stop_pct = self.stop_loss_exits / self.total_trades if self.total_trades else None loss_stats = _loss_containment_stats(self.daily_returns, include_score=True) return IntradayMetrics( run_id=self.run_id, params_hash=_hash_strategy(self.active_strategy), start_date=self.start_date, end_date=self.end_date, trading_days=self.n_days, days_with_trades=self.days_with_trades, days_with_activity=self.days_with_activity, total_trades=self.total_trades, stop_loss_exits=self.stop_loss_exits, idle_sleeve_entry_days=self.idle_sleeve_entry_days, idle_sleeve_positions_opened=self.idle_sleeve_positions_opened, win_rate=round(win_rate, 4) if win_rate is not None else None, avg_win_pct=round(avg_win_pct, 4) if avg_win_pct is not None else None, avg_loss_pct=round(avg_loss_pct, 4) if avg_loss_pct is not None else None, profit_factor=round(profit_factor, 4) if profit_factor is not None else None, expectancy_pct=round(expectancy_pct, 4) if expectancy_pct is not None else None, total_return_pct=round(total_return_pct, 4) if total_return_pct is not None else None, annualized_return_pct=round(annualized, 4) if annualized is not None else None, avg_daily_return_pct=round(avg_daily, 6) if avg_daily is not None else None, max_drawdown_pct=round(self.max_drawdown, 4), sharpe_ratio=round(sharpe, 4) if sharpe is not None else None, sortino_ratio=round(sortino, 4) if sortino is not None else None, calmar_ratio=round(calmar, 4) if calmar is not None else None, loss_day_rate=loss_stats["loss_day_rate"], avg_loss_day_pct=loss_stats["avg_loss_day_pct"], tail_loss_20_pct=loss_stats["tail_loss_20_pct"], worst_day_return_pct=loss_stats["worst_day_return_pct"], loss_containment_score=loss_stats["loss_containment_score"], avg_hold_minutes=( round(self.hold_minutes_sum / self.hold_minutes_count, 1) if self.hold_minutes_count > 0 else None ), stop_loss_exit_pct=round(stop_pct, 4) if stop_pct is not None else None, initial_capital=self.initial_capital, final_equity=round(self.equity, 2), ) def _loss_containment_stats( daily_returns: list[float], *, include_score: bool, ) -> dict[str, float | None]: if not daily_returns: return { "loss_day_rate": None, "avg_loss_day_pct": None, "tail_loss_20_pct": None, "worst_day_return_pct": None, "loss_containment_score": None, } loss_days = sorted(r for r in daily_returns if r < 0) loss_day_rate = len(loss_days) / len(daily_returns) worst_day = min(daily_returns) avg_loss_day = statistics.mean(loss_days) if loss_days else None tail_loss = None if loss_days: tail_n = max(1, math.ceil(len(loss_days) * 0.2)) tail_loss = statistics.mean(loss_days[:tail_n]) score = None if include_score: if not loss_days: score = 100.0 else: avg_abs = abs(avg_loss_day or 0.0) * 100.0 tail_abs = abs(tail_loss or 0.0) * 100.0 worst_abs = abs(worst_day) * 100.0 score = max(0.0, min(100.0, 100.0 - avg_abs * 12.0 - tail_abs * 6.0 - worst_abs * 2.0)) return { "loss_day_rate": round(loss_day_rate, 4), "avg_loss_day_pct": None if avg_loss_day is None else round(avg_loss_day, 4), "tail_loss_20_pct": None if tail_loss is None else round(tail_loss, 4), "worst_day_return_pct": round(worst_day, 4), "loss_containment_score": None if score is None else round(score, 2), } def compute_metrics( day_results: list[DayResult], config: IntradayConfig, run_id: str = "", ) -> IntradayMetrics: """Compute all performance metrics from simulation results.""" all_trades: list[IntradayTrade] = [t for r in day_results for t in r.trades] # Include ALL days (0% for no-trade days) — idle capital dilutes Sharpe correctly daily_returns = [r.daily_return_pct for r in day_results] initial_capital = _get_initial_capital(config) n_days = len(day_results) days_with_trades = sum(1 for r in day_results if r.trades) idle_sleeve_entry_days = sum(1 for r in day_results if _idle_sleeve_open_count(r) > 0) idle_sleeve_positions_opened = sum(_idle_sleeve_open_count(r) for r in day_results) days_with_activity = sum(1 for r in day_results if _day_activity_count(r) > 0) dates = sorted(r.date for r in day_results) start_date = dates[0] if dates else "" end_date = dates[-1] if dates else "" is_orb = getattr(config, "strategy_mode", "momentum") == "orb" active_strategy = config.orb_strategy if (is_orb and config.orb_strategy) else config.strategy if not all_trades: return IntradayMetrics( run_id=run_id or str(uuid.uuid4())[:8], params_hash=_hash_strategy(active_strategy), start_date=start_date, end_date=end_date, trading_days=n_days, days_with_trades=days_with_trades, days_with_activity=days_with_activity, total_trades=0, stop_loss_exits=0, idle_sleeve_entry_days=idle_sleeve_entry_days, idle_sleeve_positions_opened=idle_sleeve_positions_opened, total_return_pct=0.0 if n_days > 0 else None, annualized_return_pct=0.0 if n_days > 0 else None, avg_daily_return_pct=round(statistics.mean(daily_returns), 6) if daily_returns else None, max_drawdown_pct=0.0 if n_days > 0 else None, initial_capital=initial_capital, final_equity=initial_capital, ) # Build equity curve equity = initial_capital equity_curve: list[float] = [equity] for r in day_results: equity += r.daily_pnl equity_curve.append(equity) # Win/loss wins = [t for t in all_trades if t.pnl > 0] losses = [t for t in all_trades if t.pnl <= 0] win_rate = len(wins) / len(all_trades) if all_trades else None avg_win_pct = statistics.mean(t.pnl_pct for t in wins) if wins else None avg_loss_pct = statistics.mean(t.pnl_pct for t in losses) if losses else None gross_profit = sum(t.pnl for t in wins) gross_loss = abs(sum(t.pnl for t in losses)) profit_factor = gross_profit / gross_loss if gross_loss > 0 else None expectancy_pct = ( (win_rate * avg_win_pct + (1 - win_rate) * avg_loss_pct) if win_rate is not None and avg_win_pct is not None and avg_loss_pct is not None else None ) # Returns total_return_pct = (equity_curve[-1] - equity_curve[0]) / equity_curve[0] annualized = total_return_pct * (252 / n_days) if n_days > 0 else None avg_daily = statistics.mean(daily_returns) if daily_returns else None # Max drawdown max_eq = equity_curve[0] max_dd = 0.0 for eq in equity_curve: max_eq = max(max_eq, eq) dd = (eq - max_eq) / max_eq max_dd = min(max_dd, dd) # Sharpe / Sortino (annualized, assuming 252 trading days) sharpe = sortino = calmar = None if len(daily_returns) >= 5: try: mean_r = statistics.mean(daily_returns) std_r = statistics.stdev(daily_returns) if std_r > 0: sharpe = (mean_r / std_r) * math.sqrt(252) down_devs = [r for r in daily_returns if r < 0] if down_devs: downside_std = math.sqrt( sum(r ** 2 for r in down_devs) / len(daily_returns) ) if downside_std > 0: sortino = (mean_r / downside_std) * math.sqrt(252) except Exception: pass if annualized is not None and max_dd < 0: calmar = annualized / abs(max_dd) # Intraday-specific stop_exits = [t for t in all_trades if t.exit_reason == "stop_loss"] stop_pct = len(stop_exits) / len(all_trades) if all_trades else None loss_stats = _loss_containment_stats(daily_returns, include_score=True) # Average hold time (in minutes) hold_minutes: list[float] = [] for t in all_trades: try: from zoneinfo import ZoneInfo _ET = ZoneInfo("America/New_York") entry = datetime.fromisoformat(t.entry_time.replace("Z", "+00:00")) exit_ = datetime.fromisoformat(t.exit_time.replace("Z", "+00:00")) hold_minutes.append((exit_ - entry).total_seconds() / 60) except Exception: pass return IntradayMetrics( run_id=run_id or str(uuid.uuid4())[:8], params_hash=_hash_strategy(active_strategy), start_date=start_date, end_date=end_date, trading_days=n_days, days_with_trades=days_with_trades, days_with_activity=days_with_activity, total_trades=len(all_trades), stop_loss_exits=len(stop_exits), idle_sleeve_entry_days=idle_sleeve_entry_days, idle_sleeve_positions_opened=idle_sleeve_positions_opened, win_rate=round(win_rate, 4) if win_rate is not None else None, avg_win_pct=round(avg_win_pct, 4) if avg_win_pct is not None else None, avg_loss_pct=round(avg_loss_pct, 4) if avg_loss_pct is not None else None, profit_factor=round(profit_factor, 4) if profit_factor is not None else None, expectancy_pct=round(expectancy_pct, 4) if expectancy_pct is not None else None, total_return_pct=round(total_return_pct, 4), annualized_return_pct=round(annualized, 4) if annualized is not None else None, avg_daily_return_pct=round(avg_daily, 6) if avg_daily is not None else None, max_drawdown_pct=round(max_dd, 4), sharpe_ratio=round(sharpe, 4) if sharpe is not None else None, sortino_ratio=round(sortino, 4) if sortino is not None else None, calmar_ratio=round(calmar, 4) if calmar is not None else None, loss_day_rate=loss_stats["loss_day_rate"], avg_loss_day_pct=loss_stats["avg_loss_day_pct"], tail_loss_20_pct=loss_stats["tail_loss_20_pct"], worst_day_return_pct=loss_stats["worst_day_return_pct"], loss_containment_score=loss_stats["loss_containment_score"], avg_hold_minutes=round(statistics.mean(hold_minutes), 1) if hold_minutes else None, stop_loss_exit_pct=round(stop_pct, 4) if stop_pct is not None else None, initial_capital=initial_capital, final_equity=round(equity_curve[-1], 2), ) def _hash_strategy(strategy: Any) -> str: s = json.dumps(strategy.model_dump(), sort_keys=True, default=str) return hashlib.md5(s.encode()).hexdigest()[:8] # ── Reporting ────────────────────────────────────────────────────────────── def _describe_momentum_stop(strategy) -> str: if strategy.atr_stop_multiplier is not None: base = f"{strategy.atr_stop_multiplier:.2f}xATR" elif strategy.opening_range_stop_multiplier is not None: base = f"{strategy.opening_range_stop_multiplier:.2f}xOR" elif strategy.stop_loss_pct is not None: base = f"{strategy.stop_loss_pct:.3f}" else: base = "none" if strategy.trailing_stop_pct is None: return base trail = f"trail {strategy.trailing_stop_pct:.3f}" if strategy.trailing_activation_gain_pct is not None: trail += f" @+{strategy.trailing_activation_gain_pct*100:.1f}%" return f"{base} + {trail}" def format_summary(metrics: IntradayMetrics, config: IntradayConfig) -> str: """Format summary table for terminal output using rich.""" from rich.console import Console from rich.table import Table from io import StringIO buf = StringIO() console = Console(file=buf, width=80) # Header console.print() is_orb = getattr(config, "strategy_mode", "momentum") == "orb" title = "Opening Range Breakout (ORB) Results" if is_orb else "Morning Momentum Backtest Results" console.print( f"[bold cyan]{title}[/bold cyan] " f"[dim]{metrics.start_date} → {metrics.end_date}[/dim]" ) if is_orb and config.orb_strategy: p = config.orb_strategy min_rvol_label = "off" if p.min_rvol is None else f"{p.min_rvol:.1f}x" console.print( f"[dim]Universe: {config.universe.source} | " f"ORB: {p.orb_minutes}min | " f"Stop: {p.atr_stop_multiplier*100:.0f}%×ATR | " f"Risk: {p.risk_per_trade_pct*100:.2f}%/trade | " f"MinRVOL: {min_rvol_label} | " f"Exit: -{p.exit_minutes_before_close}min[/dim]" ) else: console.print( f"[dim]Universe: {config.universe.source} | " f"Entry: +{config.strategy.entry_minutes_after_open}min | " f"Exit: -{config.strategy.exit_minutes_before_close}min | " f"Stop: {_describe_momentum_stop(config.strategy)} | " f"Top N: {config.strategy.top_n}[/dim]" ) console.print() t = Table(show_header=True, header_style="bold") t.add_column("Metric", style="cyan") t.add_column("Value", justify="right") def _pct(v: float | None, decimals: int = 2) -> str: if v is None: return "—" return f"{v*100:+.{decimals}f}%" def _f(v: float | None, decimals: int = 2) -> str: if v is None: return "—" return f"{v:.{decimals}f}" activity_days = getattr(metrics, "days_with_activity", 0) or metrics.days_with_trades t.add_row( "Period", f"{metrics.trading_days} days ({metrics.days_with_trades} exit days, {activity_days} activity days)", ) t.add_row("Total trades", str(metrics.total_trades)) if getattr(metrics, "idle_sleeve_positions_opened", 0): t.add_row( "Idle entries", f"{metrics.idle_sleeve_positions_opened} on {metrics.idle_sleeve_entry_days} days", ) t.add_row("Stop-loss exits", str(metrics.stop_loss_exits)) t.add_section() t.add_row("Total return", _pct(metrics.total_return_pct)) t.add_row("Annualized return", _pct(metrics.annualized_return_pct)) t.add_row("Final equity", f"${metrics.final_equity:,.2f}") t.add_section() t.add_row("Win rate", _pct(metrics.win_rate, 1)) t.add_row("Avg winner", _pct(metrics.avg_win_pct)) t.add_row("Avg loser", _pct(metrics.avg_loss_pct)) t.add_row("Profit factor", _f(metrics.profit_factor)) t.add_row("Expectancy", _pct(metrics.expectancy_pct)) t.add_section() t.add_row("Max drawdown", _pct(metrics.max_drawdown_pct)) t.add_row("Sharpe ratio", _f(metrics.sharpe_ratio)) t.add_row("Sortino ratio", _f(metrics.sortino_ratio)) t.add_row("Calmar ratio", _f(metrics.calmar_ratio)) t.add_row("Avg losing day", _pct(metrics.avg_loss_day_pct)) t.add_row("Tail loss (20%)", _pct(metrics.tail_loss_20_pct)) t.add_row("Loss containment", _f(metrics.loss_containment_score)) t.add_section() t.add_row("Avg hold (min)", _f(metrics.avg_hold_minutes, 0)) t.add_row("Stop-loss rate", _pct(metrics.stop_loss_exit_pct, 1)) console.print(t) return buf.getvalue() def format_daily_breakdown(day_results: list[DayResult]) -> str: """Format per-day P&L breakdown table.""" from rich.console import Console from rich.table import Table from io import StringIO activity_days = [r for r in day_results if _day_activity_count(r) > 0] if not activity_days: return "No trades.\n" buf = StringIO() console = Console(file=buf, width=120) t = Table(show_header=True, header_style="bold", title="Daily Breakdown") t.add_column("Date", style="cyan") t.add_column("#", justify="right") t.add_column("P&L", justify="right") t.add_column("Return", justify="right") t.add_column("Tickers") for r in activity_days: closed_tickers = " ".join( f"{tr.ticker}([green]+{tr.pnl_pct*100:.1f}%[/green])" if tr.pnl > 0 else f"{tr.ticker}([red]{tr.pnl_pct*100:.1f}%[/red])" for tr in r.trades ) pending_open_count = max( 0, _idle_sleeve_open_count(r) - _same_day_idle_trade_count(r), ) opened_symbols = _idle_sleeve_open_symbols(r)[:pending_open_count] opened_labels = _idle_sleeve_open_labels(r)[:pending_open_count] opened_tickers = " ".join( f"{symbol}(open:{opened_labels[idx] if idx < len(opened_labels) else 'idle'})" for idx, symbol in enumerate(opened_symbols) ) tickers_str = closed_tickers if opened_tickers: tickers_str = f"{tickers_str} " if tickers_str else "" tickers_str += f"[dim]Opened: {opened_tickers}[/dim]" pnl_str = f"[green]${r.daily_pnl:+,.2f}[/green]" if r.daily_pnl >= 0 else f"[red]${r.daily_pnl:+,.2f}[/red]" ret_str = f"[green]{r.daily_return_pct*100:+.2f}%[/green]" if r.daily_return_pct >= 0 else f"[red]{r.daily_return_pct*100:+.2f}%[/red]" t.add_row(r.date, str(_day_activity_count(r)), pnl_str, ret_str, tickers_str) console.print(t) return buf.getvalue() def format_top_trades(day_results: list[DayResult], n: int = 5) -> str: """Format best and worst N trades.""" from rich.console import Console from rich.table import Table from io import StringIO all_trades = [t for r in day_results for t in r.trades] if not all_trades: return "" buf = StringIO() console = Console(file=buf, width=100) sorted_trades = sorted(all_trades, key=lambda t: t.pnl_pct, reverse=True) for label, trades in [("Top Winners", sorted_trades[:n]), ("Top Losers", sorted_trades[-n:])]: t = Table(show_header=True, header_style="bold", title=label) t.add_column("Date") t.add_column("Ticker") t.add_column("Entry", justify="right") t.add_column("Exit", justify="right") t.add_column("Return", justify="right") t.add_column("P&L", justify="right") t.add_column("Morn Gain", justify="right") t.add_column("Exit Reason") for tr in trades: color = "green" if tr.pnl >= 0 else "red" t.add_row( tr.date, tr.ticker, f"${tr.entry_price:.2f}", f"${tr.exit_price:.2f}", f"[{color}]{tr.pnl_pct*100:+.2f}%[/{color}]", f"[{color}]${tr.pnl:+.2f}[/{color}]", f"{tr.morning_gain_pct*100:+.2f}%", tr.exit_reason, ) console.print(t) return buf.getvalue() def format_sweep_comparison(sweep_results: list[SweepResult], top_n: int = 20) -> str: """Format sweep results as a ranked comparison table (sorted by Sharpe). Automatically detects ORB vs momentum by inspecting params keys. """ from rich.console import Console from rich.table import Table from io import StringIO buf = StringIO() console = Console(file=buf, width=160) # Sort by the configured sweep objective when present; otherwise preserve # legacy Sharpe-first display. ranked = sorted( sweep_results, key=lambda r: ( r.objective_score if getattr(r, "objective_score", None) is not None else (r.metrics.sharpe_ratio or -999), r.metrics.sharpe_ratio or -999, r.metrics.total_return_pct or -999, ), reverse=True, )[:top_n] # Detect strategy from first result's params is_orb = bool(ranked) and "atr_stop_multiplier" in ranked[0].params has_objective = any(getattr(sr, "objective_score", None) is not None for sr in ranked) title_suffix = "Objective" if has_objective else "Sharpe" t = Table(show_header=True, header_style="bold", title=f"Sweep Results — Top {top_n} by {title_suffix}") t.add_column("#", justify="right") if is_orb: t.add_column("ATR\nMult", justify="right") t.add_column("Min\nRVOL", justify="right") t.add_column("MaxCand", justify="right") t.add_column("BE\n@R", justify="right") t.add_column("Trail\n@R", justify="right") t.add_column("Timeout\n(min)", justify="right") t.add_column("Risk\n%", justify="right") else: t.add_column("Entry\n(min)", justify="right") t.add_column("Exit\n(min)", justify="right") t.add_column("Stop\n(%)", justify="right") t.add_column("Max\nGain%", justify="right") t.add_column("Min\nGain%", justify="right") t.add_column("Cooldown\n(days)", justify="right") t.add_column("MinVol\n(K)", justify="right") t.add_column("Top\nN", justify="right") t.add_column("Return", justify="right") t.add_column("Ann\nReturn", justify="right") t.add_column("Sharpe", justify="right") t.add_column("Max\nDD%", justify="right") t.add_column("Win\n%", justify="right") t.add_column("PF", justify="right") t.add_column("Trades", justify="right") if has_objective: t.add_column("Obj", justify="right") for i, sr in enumerate(ranked, 1): m = sr.metrics p = sr.params sharpe_str = f"{m.sharpe_ratio:.2f}" if m.sharpe_ratio is not None else "—" color = "green" if (m.total_return_pct or 0) >= 0 else "red" if is_orb: param_cells = [ f"{p.get('atr_stop_multiplier', 0)*100:.0f}%", f"{p.get('min_rvol', 0):.1f}x", str(p.get("max_candidates", "")), f"{p.get('breakeven_at_r', '')}R", f"{p.get('trailing_at_r', '')}R", str(p.get("order_timeout_minutes", "")), f"{p.get('risk_per_trade_pct', 0)*100:.2f}%", ] else: if p.get("atr_stop_multiplier") is not None: stop = f"{p.get('atr_stop_multiplier'):.2f}xATR" elif p.get("opening_range_stop_multiplier") is not None: stop = f"{p.get('opening_range_stop_multiplier'):.2f}xOR" elif p.get("stop_loss_pct") is not None: stop = f"{p.get('stop_loss_pct', 0)*100:.0f}" else: stop = "none" param_cells = [ str(p.get("entry_minutes_after_open", "")), str(p.get("exit_minutes_before_close", "")), stop, f"{p.get('max_morning_gain_pct', 0)*100:.0f}" if p.get("max_morning_gain_pct") is not None else "none", f"{p.get('min_morning_gain_pct', 0)*100:.1f}", str(p.get("ticker_cooldown_days", 0)), f"{p.get('min_entry_volume', 0)//1000:.0f}" if p.get("min_entry_volume") else "—", str(p.get("top_n", "")), ] row = [ str(i), *param_cells, f"[{color}]{(m.total_return_pct or 0)*100:+.1f}%[/{color}]", f"[{color}]{(m.annualized_return_pct or 0)*100:+.1f}%[/{color}]", sharpe_str, f"{(m.max_drawdown_pct or 0)*100:.1f}%", f"{(m.win_rate or 0)*100:.1f}%", f"{m.profit_factor:.2f}" if m.profit_factor else "—", str(m.total_trades), ] if has_objective: row.append( f"{sr.objective_score:.3f}" if getattr(sr, "objective_score", None) is not None else "—" ) t.add_row(*row) console.print(t) return buf.getvalue() def write_results( metrics: IntradayMetrics, day_results: list[DayResult], config: IntradayConfig, output_dir: str, data_provenance: dict[str, Any] | None = None, ) -> Path: """Write full results to a JSON file in the output directory.""" out = Path(output_dir) out.mkdir(parents=True, exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") filename = out / f"intraday_{ts}_{metrics.run_id}.json" all_trades = [t.model_dump() for r in day_results for t in r.trades] # Aggregate skip breakdown and filter stats across all days skip_breakdown: dict[str, int] = {"traded": 0} agg_filter_stats: dict[str, int] = {} for r in day_results: if r.skip_reason: skip_breakdown[r.skip_reason] = skip_breakdown.get(r.skip_reason, 0) + 1 elif r.trades: skip_breakdown["traded"] += 1 elif _idle_sleeve_open_count(r) > 0: skip_breakdown["opened_idle_sleeve"] = ( skip_breakdown.get("opened_idle_sleeve", 0) + 1 ) else: skip_breakdown["traded_no_fill"] = skip_breakdown.get("traded_no_fill", 0) + 1 if r.candidate_filter_stats: for k, v in r.candidate_filter_stats.items(): agg_filter_stats[k] = agg_filter_stats.get(k, 0) + v payload = { "run_id": metrics.run_id, "generated_at": datetime.now().isoformat(), "config": config.model_dump(), "data_provenance": data_provenance or {}, "metrics": metrics.model_dump(), "skip_breakdown": skip_breakdown, "aggregate_filter_stats": agg_filter_stats, "trades": all_trades, "daily_summary": [ { "date": r.date, "daily_pnl": r.daily_pnl, "daily_return_pct": r.daily_return_pct, "candidates_found": r.candidates_found, "trades": len(r.trades), "idle_sleeve_positions_opened": _idle_sleeve_open_count(r), "idle_sleeve_symbols_opened": _idle_sleeve_open_symbols(r), "idle_sleeve_labels_opened": _idle_sleeve_open_labels(r), "activity_count": _day_activity_count(r), "skip_reason": r.skip_reason, "rolling_loss_synthetic_pnl": r.rolling_loss_synthetic_pnl, "candidate_filter_stats": r.candidate_filter_stats, "entry_diagnostics": r.entry_diagnostics, "regime_scaler": r.regime_scaler, "regime_gap_pct": r.regime_gap_pct, "breadth_scaler": r.breadth_scaler, "breadth_ratio": r.breadth_ratio, "breadth_positive_count": r.breadth_positive_count, "breadth_total_count": r.breadth_total_count, "market_orb_quality_scaler": r.market_orb_quality_scaler, "market_orb_quality_close_location": r.market_orb_quality_close_location, "market_orb_quality_return_pct": r.market_orb_quality_return_pct, "market_orb_quality_secondary_close_location": r.market_orb_quality_secondary_close_location, "market_orb_quality_secondary_return_pct": r.market_orb_quality_secondary_return_pct, "market_thrust_breadth_override_active": ( r.market_thrust_breadth_override_active ), "market_thrust_opening_breadth_override_active": ( r.market_thrust_opening_breadth_override_active ), "market_thrust_opening_breadth_positive_ratio": ( r.market_thrust_opening_breadth_positive_ratio ), "market_thrust_opening_breadth_avg_return_pct": ( r.market_thrust_opening_breadth_avg_return_pct ), "market_thrust_opening_breadth_strong_close_location_ratio": ( r.market_thrust_opening_breadth_strong_close_location_ratio ), "market_thrust_opening_breadth_total_count": ( r.market_thrust_opening_breadth_total_count ), "market_orb_quality_divergence_active": r.market_orb_quality_divergence_active, "market_orb_quality_primary_weak_secondary_strong_active": ( r.market_orb_quality_primary_weak_secondary_strong_active ), "market_orb_quality_primary_weak_secondary_strong_max_trades_active": ( r.market_orb_quality_primary_weak_secondary_strong_max_trades_active ), "market_orb_quality_primary_lag_secondary_lead_active": ( r.market_orb_quality_primary_lag_secondary_lead_active ), "market_orb_quality_primary_lag_secondary_lead_max_trades_active": ( r.market_orb_quality_primary_lag_secondary_lead_max_trades_active ), "market_orb_quality_joint_weak_active": r.market_orb_quality_joint_weak_active, "market_orb_quality_joint_weak_max_trades_active": ( r.market_orb_quality_joint_weak_max_trades_active ), "market_orb_quality_joint_panic_active": r.market_orb_quality_joint_panic_active, "market_orb_quality_joint_panic_max_trades_active": ( r.market_orb_quality_joint_panic_max_trades_active ), "sector_scaler": r.sector_scaler, "tail_risk_scaler": r.tail_risk_scaler, "conditional_confirmation_active": r.conditional_confirmation_active, "is_soft_day": r.is_soft_day, "soft_day_reason": r.soft_day_reason, "event_day_liquid_active": r.event_day_liquid_active, "event_day_liquid_event_count": r.event_day_liquid_event_count, "event_day_liquid_total_event_entry_dollar_volume": r.event_day_liquid_total_event_entry_dollar_volume, } for r in day_results ], } filename.write_text(json.dumps(payload, indent=2, default=str)) return filename