maneshtrader/web_core/analytics.py

from __future__ import annotations

import pandas as pd

from .constants import TREND_BEAR, TREND_BULL


def _trade_return_pct(direction: str, entry_price: float, exit_price: float) -> float:
    if direction == "LONG":
        return ((exit_price - entry_price) / entry_price) * 100.0
    return ((entry_price - exit_price) / entry_price) * 100.0


def backtest_signals(
    df: pd.DataFrame,
    slippage_bps: float = 0.0,
    fee_bps: float = 0.0,
    stop_loss_pct: float | None = None,
    take_profit_pct: float | None = None,
    max_hold_bars: int = 1,
    min_hold_bars: int = 1,
) -> dict[str, float | int]:
    if len(df) < 4:
        return {"trades": 0, "wins": 0, "losses": 0, "win_rate": 0.0, "avg_pnl_pct": 0.0}

    trend_series = df["trend_state"]
    trend_change = trend_series != trend_series.shift(1)
    signal_idx = df.index[trend_change & trend_series.isin([TREND_BULL, TREND_BEAR])]

    wins = 0
    losses = 0
    pnl_values: list[float] = []
    max_hold = max(1, int(max_hold_bars))
    min_hold = max(1, min(int(min_hold_bars), max_hold))
    total_cost_pct = ((float(slippage_bps) + float(fee_bps)) * 2.0) / 100.0

    for idx in signal_idx:
        pos = df.index.get_loc(idx)
        if pos + 1 >= len(df):
            continue

        entry_close = float(df.iloc[pos]["Close"])
        signal_trend = df.iloc[pos]["trend_state"]
        direction = "LONG" if signal_trend == TREND_BULL else "SHORT"

        exit_price = float(df.iloc[min(pos + max_hold, len(df) - 1)]["Close"])
        for hold in range(min_hold, max_hold + 1):
            exit_pos = min(pos + hold, len(df) - 1)
            candidate_exit = float(df.iloc[exit_pos]["Close"])
            raw_pnl = _trade_return_pct(direction, entry_close, candidate_exit)
            if stop_loss_pct is not None and raw_pnl <= -float(stop_loss_pct):
                exit_price = candidate_exit
                break
            if take_profit_pct is not None and raw_pnl >= float(take_profit_pct):
                exit_price = candidate_exit
                break
            exit_price = candidate_exit

        pnl_pct = _trade_return_pct(direction, entry_close, exit_price) - total_cost_pct
        pnl_values.append(pnl_pct)
        wins += int(pnl_pct > 0)
        losses += int(pnl_pct <= 0)

    trades = wins + losses
    win_rate = (wins / trades * 100.0) if trades else 0.0
    avg_pnl_pct = (sum(pnl_values) / len(pnl_values)) if pnl_values else 0.0
    return {
        "trades": trades,
        "wins": wins,
        "losses": losses,
        "win_rate": round(win_rate, 2),
        "avg_pnl_pct": round(avg_pnl_pct, 2),
    }


def simulate_trend_trades(df: pd.DataFrame, max_examples: int = 20) -> pd.DataFrame:
    if len(df) < 4:
        return pd.DataFrame(
            columns=[
                "direction",
                "entry_timestamp",
                "exit_timestamp",
                "entry_price",
                "exit_price",
                "bars_held",
                "pnl_pct",
                "outcome",
            ]
        )

    trend_series = df["trend_state"]
    trend_change = trend_series != trend_series.shift(1)
    signal_idx = df.index[trend_change & trend_series.isin([TREND_BULL, TREND_BEAR])]

    direction_by_trend = {
        TREND_BULL: "LONG",
        TREND_BEAR: "SHORT",
    }

    open_trade: dict[str, object] | None = None
    closed_trades: list[dict[str, object]] = []

    for idx in signal_idx:
        pos = int(df.index.get_loc(idx))
        trend_now = str(df.iloc[pos]["trend_state"])
        direction = direction_by_trend[trend_now]
        close_price = float(df.iloc[pos]["Close"])

        if open_trade is None:
            open_trade = {
                "direction": direction,
                "entry_timestamp": idx,
                "entry_pos": pos,
                "entry_price": close_price,
            }
            continue

        if open_trade["direction"] == direction:
            continue

        entry_price = float(open_trade["entry_price"])
        if str(open_trade["direction"]) == "LONG":
            pnl_pct = ((close_price - entry_price) / entry_price) * 100.0
        else:
            pnl_pct = ((entry_price - close_price) / entry_price) * 100.0

        closed_trades.append(
            {
                "direction": str(open_trade["direction"]),
                "entry_timestamp": open_trade["entry_timestamp"],
                "exit_timestamp": idx,
                "entry_price": round(entry_price, 4),
                "exit_price": round(close_price, 4),
                "bars_held": pos - int(open_trade["entry_pos"]),
                "pnl_pct": round(pnl_pct, 2),
                "outcome": "Win" if pnl_pct > 0 else ("Loss" if pnl_pct < 0 else "Flat"),
            }
        )

        # Flip into the new direction at the same confirmed reversal bar.
        open_trade = {
            "direction": direction,
            "entry_timestamp": idx,
            "entry_pos": pos,
            "entry_price": close_price,
        }

    if not closed_trades:
        return pd.DataFrame(
            columns=[
                "direction",
                "entry_timestamp",
                "exit_timestamp",
                "entry_price",
                "exit_price",
                "bars_held",
                "pnl_pct",
                "outcome",
            ]
        )

    trades_df = pd.DataFrame(closed_trades)
    if max_examples > 0 and len(trades_df) > max_examples:
        trades_df = trades_df.iloc[-max_examples:].copy()
    return trades_df