""" ML-3m-trader Metrics & Evaluation ==================================== Computes comprehensive trading performance metrics and generates a formatted report. """ import os from datetime import timedelta import numpy as np import pandas as pd import config as cfg def compute_metrics( trades: list, equity_curve: np.ndarray, starting_balance: float = cfg.STARTING_BALANCE, ) -> dict: """ Compute all performance metrics from backtest results. Parameters ---------- trades : list of dict Trade records from backtester. equity_curve : np.ndarray Balance at each bar. starting_balance : float Initial account balance. Returns ------- dict All computed metrics. """ if not trades: return {"error": "No trades executed."} df_trades = pd.DataFrame(trades) # Basic counts total_trades = len(df_trades) wins = df_trades[df_trades["result"] == "TP"] losses = df_trades[df_trades["result"] == "SL"] n_wins = len(wins) n_losses = len(losses) win_rate = n_wins / total_trades if total_trades > 0 else 0.0 # PnL pnl_array = df_trades["pnl"].values gross_profit = pnl_array[pnl_array > 0].sum() if (pnl_array > 0).any() else 0.0 gross_loss = abs(pnl_array[pnl_array < 0].sum()) if (pnl_array < 0).any() else 0.0 net_profit = pnl_array.sum() # Average win/loss as percentage of balance at entry avg_win_pct = 0.0 avg_loss_pct = 0.0 if n_wins > 0: # Each trade's PnL as % of balance before trade win_pcts = [] for _, t in wins.iterrows(): bal_before = t["balance_after"] - t["pnl"] if bal_before > 0: win_pcts.append(t["pnl"] / bal_before * 100.0) avg_win_pct = np.mean(win_pcts) if win_pcts else 0.0 if n_losses > 0: loss_pcts = [] for _, t in losses.iterrows(): bal_before = t["balance_after"] - t["pnl"] if bal_before > 0: loss_pcts.append(abs(t["pnl"]) / bal_before * 100.0) avg_loss_pct = np.mean(loss_pcts) if loss_pcts else 0.0 # Profit factor profit_factor = gross_profit / gross_loss if gross_loss > 0 else float("inf") # Expectancy expectancy = net_profit / total_trades if total_trades > 0 else 0.0 # Equity curve metrics final_balance = equity_curve[-1] total_return = (final_balance - starting_balance) / starting_balance * 100.0 # Max drawdown running_max = np.maximum.accumulate(equity_curve) drawdowns = (equity_curve - running_max) / running_max max_drawdown = abs(drawdowns.min()) * 100.0 # percentage # Sharpe Ratio (annualized from bar-level returns) eq_returns = np.diff(equity_curve) / equity_curve[:-1] eq_returns = eq_returns[np.isfinite(eq_returns)] bars_per_day = (6.5 * 60) / cfg.TIMEFRAME_MINUTES annual_factor = np.sqrt(252 * bars_per_day) if len(eq_returns) > 1 and np.std(eq_returns) > 0: sharpe = (np.mean(eq_returns) / np.std(eq_returns)) * annual_factor else: sharpe = 0.0 # Sortino Ratio downside = eq_returns[eq_returns < 0] if len(downside) > 1 and np.std(downside) > 0: sortino = (np.mean(eq_returns) / np.std(downside)) * annual_factor else: sortino = 0.0 # Calmar Ratio # Annualized return / max drawdown total_bars = len(equity_curve) years = total_bars / (bars_per_day * 252) if bars_per_day > 0 else 1.0 annualized_return = ((final_balance / starting_balance) ** (1 / max(years, 0.01)) - 1) * 100.0 calmar = annualized_return / max_drawdown if max_drawdown > 0 else float("inf") # Trade duration durations = (df_trades["exit_bar"] - df_trades["entry_bar"]).values avg_duration_bars = np.mean(durations) if len(durations) > 0 else 0 avg_duration_min = avg_duration_bars * cfg.TIMEFRAME_MINUTES # Intraday PnL stats daily_pnl = {} if "time_entry" in df_trades.columns: for _, t in df_trades.iterrows(): if t["time_entry"] is not None: day = pd.Timestamp(t["time_entry"]).date() daily_pnl.setdefault(day, 0.0) daily_pnl[day] += t["pnl"] daily_pnl_values = np.array(list(daily_pnl.values())) if daily_pnl else np.array([0.0]) daily_mean = np.mean(daily_pnl_values) daily_std = np.std(daily_pnl_values) if len(daily_pnl_values) > 1 else 0.0 daily_min = np.min(daily_pnl_values) daily_max = np.max(daily_pnl_values) # Buy/Sell breakdown buy_trades = df_trades[df_trades["direction"] == "BUY"] sell_trades = df_trades[df_trades["direction"] == "SELL"] metrics = { "total_trades": total_trades, "winning_trades": n_wins, "losing_trades": n_losses, "win_rate": win_rate, "avg_win_pct": avg_win_pct, "avg_loss_pct": avg_loss_pct, "gross_profit": gross_profit, "gross_loss": gross_loss, "net_profit": net_profit, "profit_factor": profit_factor, "expectancy": expectancy, "starting_balance": starting_balance, "final_balance": final_balance, "total_return_pct": total_return, "max_drawdown_pct": max_drawdown, "sharpe_ratio": sharpe, "sortino_ratio": sortino, "calmar_ratio": calmar, "avg_trade_duration_bars": avg_duration_bars, "avg_trade_duration_min": avg_duration_min, "total_buy_trades": len(buy_trades), "total_sell_trades": len(sell_trades), "daily_pnl_mean": daily_mean, "daily_pnl_std": daily_std, "daily_pnl_min": daily_min, "daily_pnl_max": daily_max, "trading_days": len(daily_pnl), } return metrics def format_report(metrics: dict) -> str: """Format metrics into a readable console report.""" if "error" in metrics: return f"\n[REPORT] {metrics['error']}\n" sep = "=" * 60 lines = [ "", sep, " ML-3m-trader BACKTEST REPORT", sep, "", " ACCOUNT", f" Starting Balance : ${metrics['starting_balance']:>12,.2f}", f" Final Balance : ${metrics['final_balance']:>12,.2f}", f" Net Profit : ${metrics['net_profit']:>12,.2f}", f" Total Return : {metrics['total_return_pct']:>11.2f}%", "", " TRADE STATISTICS", f" Total Trades : {metrics['total_trades']:>12,}", f" Winning Trades : {metrics['winning_trades']:>12,}", f" Losing Trades : {metrics['losing_trades']:>12,}", f" Win Rate : {metrics['win_rate'] * 100:>11.2f}%", f" Buy Trades : {metrics['total_buy_trades']:>12,}", f" Sell Trades : {metrics['total_sell_trades']:>12,}", "", " PROFIT / LOSS", f" Avg Win : {metrics['avg_win_pct']:>11.4f}%", f" Avg Loss : {metrics['avg_loss_pct']:>11.4f}%", f" Gross Profit : ${metrics['gross_profit']:>12,.2f}", f" Gross Loss : ${metrics['gross_loss']:>12,.2f}", f" Profit Factor : {metrics['profit_factor']:>12.4f}", f" Expectancy (per trade): ${metrics['expectancy']:>12,.2f}", "", " RISK METRICS", f" Max Drawdown : {metrics['max_drawdown_pct']:>11.2f}%", f" Sharpe Ratio : {metrics['sharpe_ratio']:>12.4f}", f" Sortino Ratio : {metrics['sortino_ratio']:>12.4f}", f" Calmar Ratio : {metrics['calmar_ratio']:>12.4f}", "", " DURATION", f" Avg Duration (bars) : {metrics['avg_trade_duration_bars']:>12.1f}", f" Avg Duration (min) : {metrics['avg_trade_duration_min']:>12.1f}", "", " INTRADAY PNL", f" Trading Days : {metrics['trading_days']:>12,}", f" Daily Mean PnL : ${metrics['daily_pnl_mean']:>12,.2f}", f" Daily Std PnL : ${metrics['daily_pnl_std']:>12,.2f}", f" Daily Min PnL : ${metrics['daily_pnl_min']:>12,.2f}", f" Daily Max PnL : ${metrics['daily_pnl_max']:>12,.2f}", "", sep, ] return "\n".join(lines) def save_report(report: str, filename: str = "report.txt"): """Save the report to the results directory.""" os.makedirs(cfg.RESULTS_DIR, exist_ok=True) path = os.path.join(cfg.RESULTS_DIR, filename) with open(path, "w", encoding="utf-8") as f: f.write(report) print(f"[INFO] Report saved to {path}") def save_trades_csv(trades: list, filename: str = "trades.csv"): """Save individual trade records to CSV.""" os.makedirs(cfg.RESULTS_DIR, exist_ok=True) path = os.path.join(cfg.RESULTS_DIR, filename) pd.DataFrame(trades).to_csv(path, index=False) print(f"[INFO] Trade log saved to {path}")