checkpoint/python_version/metrics.py

"""
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}")