backtester.py

import pandas as pd
import numpy as np


def run_backtest(
    df,
    risk_per_trade,
    stop_loss_pct,
    take_profit_pct,
    initial_capital,
    start_date,
    end_date,
    max_trades_per_day,
    commission_amount=2.0,
    include_high_spike=False,
):
    """
    Runs the backtest logic on the provided dataframe with given parameters.
    """
    # Filter by date
    start_ts = pd.Timestamp(start_date)
    end_ts = pd.Timestamp(end_date)

    mask = (df["datetime"] >= start_ts) & (df["datetime"] <= end_ts)
    sub_df = df[mask].copy()

    if sub_df.empty:
        return pd.DataFrame()

    dates = sorted(sub_df["date"].unique())

    trades = []
    capital_net = initial_capital
    capital_gross = initial_capital
    total_comm_accum = 0

    # Market‐session price columns
    ms_columns = [
        "marketsession_1min",
        "marketsession_3min",
        "marketsession_5min",
        "marketsession_10min",
        "marketsession_15min",
        "marketsession_30min",
        "marketsession_60min",
        "marketsession_120min",
        # "marketsession_high",
    ]
    
    if include_high_spike:
        ms_columns.append("marketsession_high")

    for current_date in dates:
        countertradesperday = 0
        day_df = sub_df[sub_df["date"] == current_date]

        for _, row in day_df.iterrows():
            if row.get("Ticker") == "QMMM":  # specific exclusion from user script
                continue

            entry_price = row["premarket_close"]
            current_risk_amt = capital_net * risk_per_trade
            size = current_risk_amt  # size is dollar amount? user script: size = capital_net * RISK_PER_TRADE

            # User script:
            # stop_price = entry_price * (1 + STOP_LOSS_PCT)
            # target_price = entry_price * (1 - TAKE_PROFIT_PCT)
            # Short setup
            stop_price = entry_price * (1 + stop_loss_pct)
            target_price = entry_price * (1 - take_profit_pct)

            exit_price = None
            exit_type = None

            for col in ms_columns:
                if col not in row or pd.isna(row[col]):
                    continue

                price = row[col]
                # Stop-loss
                if price >= stop_price:
                    exit_price = stop_price
                    exit_type = "stop"
                    break
                # Take-profit
                if price <= target_price:
                    exit_price = target_price
                    exit_type = "target"
                    break

            if exit_price is None:
                exit_price = row["marketsession_close"]
                exit_type = "close"

            # Pnl for short
            pnl_gross = (entry_price - exit_price) / entry_price * size

            # Commission logic as requested/defined
            comission_entry = commission_amount * size / entry_price / 200
            comission_exit = comission_entry  # using user's approximation

            total_comm = comission_entry + comission_exit
            pnl_net = pnl_gross - total_comm

            # Update capitals
            capital_net += pnl_net
            capital_gross += pnl_gross
            total_comm_accum += total_comm

            pnl_perc = (
                pnl_net / (capital_net - pnl_net) * 100
                if (capital_net - pnl_net) != 0
                else 0
            )

            if capital_net < initial_capital / 2:
                # Stop out logic
                break

            trades.append(
                {
                    "date": current_date,
                    "ticker": row.get("Ticker"),
                    "entry_price": entry_price,
                    "exit_price": exit_price,
                    "exit_type": exit_type,
                    "size": size,
                    "pnl": pnl_net,
                    "pnl_gross": pnl_gross,
                    "pnl_perc": pnl_perc,
                    "capital_net": capital_net,
                    "capital_gross": capital_gross,
                    "comm": total_comm,
                    "cumulative_comm": total_comm_accum,
                }
            )

            countertradesperday += 1
            if countertradesperday >= max_trades_per_day:
                break

        if capital_net < initial_capital / 2:
            break

    return pd.DataFrame(trades)


def analyze_day_trading(trades_df):
    """
    Analyze day trading performance based on trade logs.
    Returns results dict and enriched df.
    """
    if trades_df.empty:
        return {}, trades_df

    df = trades_df.copy()

    # Calculate additional metrics
    df["is_win"] = df["pnl"] > 0
    df["cumulative_pnl"] = df["pnl"].cumsum()
    df["cumulative_pnl_gross"] = df["pnl_gross"].cumsum()

    df["running_max"] = df["cumulative_pnl"].cummax()
    df["drawdown"] = df["running_max"] - df["cumulative_pnl"]
    df["drawdown_pct"] = (df["pnl_gross"] / df["capital_gross"]) * 100

    # Return per trade
    # Note: user used pnl_perc which is pnl/running_capital*100.
    df["return"] = df["pnl_perc"] / 100

    total_trades = len(df)
    profitable_trades = sum(df["is_win"])
    losing_trades = total_trades - profitable_trades
    win_rate = profitable_trades / total_trades if total_trades > 0 else 0

    total_pnl = df["pnl"].sum()
    avg_pnl = df["pnl"].mean()
    max_pnl = df["pnl"].max()
    min_pnl = df["pnl"].min()

    avg_pnl_perc = df["pnl_perc"].mean()

    avg_win = df.loc[df["is_win"], "pnl"].mean() if profitable_trades > 0 else 0
    avg_loss = df.loc[~df["is_win"], "pnl"].mean() if losing_trades > 0 else 0

    risk_reward_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else float("inf")

    max_drawdown = df["drawdown"].max()
    max_drawdown_perc = (
        max_drawdown / df["running_max"].max() * 100
        if df["running_max"].max() > 0
        else 0
    )

    mean_return = df["return"].mean()
    std_return = df["return"].std()
    sharpe_ratio = (mean_return * 252**0.5) / std_return if std_return > 0 else 0

    expectancy = (win_rate * avg_win) + ((1 - win_rate) * avg_loss)

    total_profit = df.loc[df["is_win"], "pnl"].sum() if profitable_trades > 0 else 0
    total_loss = abs(df.loc[~df["is_win"], "pnl"].sum()) if losing_trades > 0 else 1
    profit_factor = total_profit / total_loss if total_loss > 0 else float("inf")

    by_date = df.groupby("date")["pnl"].sum().reset_index()
    by_ticker = df.groupby("ticker").agg(
        {"pnl": ["sum", "mean", "count"], "is_win": "mean"}
    )

    # New Metrics Calculation
    initial_capital_inferred = (
        df.iloc[0]["capital_net"] - df.iloc[0]["pnl"] if not df.empty else 0
    )
    return_on_initial_capital = (
        (total_pnl / initial_capital_inferred * 100)
        if initial_capital_inferred != 0
        else 0
    )

    total_commissions = df["comm"].sum() if not df.empty else 0
    commission_impact_pct = (
        (total_commissions / total_pnl * 100) if total_pnl != 0 else 0
    )

    results = {
        "total_trades": total_trades,
        "profitable_trades": profitable_trades,
        "losing_trades": losing_trades,
        "win_rate": win_rate,
        "total_pnl": total_pnl,
        "return_on_init_cap_pct": return_on_initial_capital,
        "total_commissions": total_commissions,
        "comm_to_pnl_pct": commission_impact_pct,
        "avg_pnl": avg_pnl,
        "max_pnl": max_pnl,
        "min_pnl": min_pnl,
        "avg_pnl_perc": avg_pnl_perc,
        "avg_win": avg_win,
        "avg_loss": avg_loss,
        "risk_reward_ratio": risk_reward_ratio,
        "max_drawdown": max_drawdown,
        "max_drawdown_perc": max_drawdown_perc,
        "sharpe_ratio": sharpe_ratio,
        "expectancy": expectancy,
        "profit_factor": profit_factor,
        # 'by_date': by_date, # Keep dataframes out of strict dict if not needed for simple display, or keep them.
        # 'by_ticker': by_ticker
    }

    return results, df