|
|
"""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
|
|
|
|
|
|
|
|
|
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.start_date = ""
|
|
|
self.end_date = ""
|
|
|
self.total_trades = 0
|
|
|
self.stop_loss_exits = 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
|
|
|
|
|
|
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,
|
|
|
"start_date": self.start_date,
|
|
|
"end_date": self.end_date,
|
|
|
"total_trades": self.total_trades,
|
|
|
"stop_loss_exits": self.stop_loss_exits,
|
|
|
"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.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.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,
|
|
|
total_trades=0,
|
|
|
stop_loss_exits=0,
|
|
|
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,
|
|
|
total_trades=self.total_trades,
|
|
|
stop_loss_exits=self.stop_loss_exits,
|
|
|
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)
|
|
|
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,
|
|
|
total_trades=0,
|
|
|
stop_loss_exits=0,
|
|
|
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,
|
|
|
total_trades=len(all_trades),
|
|
|
stop_loss_exits=len(stop_exits),
|
|
|
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
|
|
|
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: {p.min_rvol:.1f}x | "
|
|
|
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}"
|
|
|
|
|
|
t.add_row("Period", f"{metrics.trading_days} days ({metrics.days_with_trades} with trades)")
|
|
|
t.add_row("Total trades", str(metrics.total_trades))
|
|
|
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
|
|
|
|
|
|
trading_days = [r for r in day_results if r.trades]
|
|
|
if not trading_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 trading_days:
|
|
|
tickers_str = " ".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
|
|
|
)
|
|
|
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(len(r.trades)), 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 Sharpe descending, then total_return
|
|
|
ranked = sorted(
|
|
|
sweep_results,
|
|
|
key=lambda r: (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
|
|
|
|
|
|
t = Table(show_header=True, header_style="bold", title=f"Sweep Results — Top {top_n} by Sharpe")
|
|
|
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")
|
|
|
|
|
|
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", "")),
|
|
|
]
|
|
|
|
|
|
t.add_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),
|
|
|
)
|
|
|
|
|
|
console.print(t)
|
|
|
return buf.getvalue()
|
|
|
|
|
|
|
|
|
def write_results(
|
|
|
metrics: IntradayMetrics,
|
|
|
day_results: list[DayResult],
|
|
|
config: IntradayConfig,
|
|
|
output_dir: str,
|
|
|
) -> 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
|
|
|
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(),
|
|
|
"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),
|
|
|
"skip_reason": r.skip_reason,
|
|
|
"candidate_filter_stats": r.candidate_filter_stats,
|
|
|
"regime_scaler": r.regime_scaler,
|
|
|
"breadth_scaler": r.breadth_scaler,
|
|
|
"sector_scaler": r.sector_scaler,
|
|
|
"tail_risk_scaler": r.tail_risk_scaler,
|
|
|
"is_soft_day": r.is_soft_day,
|
|
|
}
|
|
|
for r in day_results
|
|
|
],
|
|
|
}
|
|
|
|
|
|
filename.write_text(json.dumps(payload, indent=2, default=str))
|
|
|
return filename
|