backtest.py

"""
backtest.py — walk-forward historical simulation for CryptoNAV Testnet.

Implemented in pure pandas/numpy rather than backtesting.py or vectorbt:
"simpler is better" — zero extra dependencies (vectorbt drags numba,
backtesting.py drags bokeh, both flaky on HF Spaces), full control over
the no-lookahead rules, and the engine is ~150 lines you can audit.

Public API (as requested):
    generate_walkforward_forecast(df, retrain_every, min_train)
        -> df with a 'forecast' column (probability up, 0-100) + meta dict
    run_backtest(df, forecast_col, price_col, **params)
        -> (metrics: dict, equity_curve: DataFrame, trades: DataFrame)

No-lookahead guarantees:
  * walk-forward: the model that scores candle t was trained ONLY on
    candles strictly before its block start
  * execution: a signal read at candle t's close executes at candle
    t+1's OPEN — you never trade a price the signal already saw
"""

import numpy as np
import pandas as pd

import engine

try:
    from xgboost import XGBClassifier
    HAVE_XGB = True
except ImportError:
    HAVE_XGB = False


# ===================================================================
# 1. walk-forward forecast generation
# ===================================================================
def generate_walkforward_forecast(
    df: pd.DataFrame,
    retrain_every: int = 168,     # retrain cadence in candles (1h bars: weekly)
    min_train: int = 300,         # candles required before the first model
) -> tuple[pd.DataFrame, dict]:
    """
    Adds a 'forecast' column: P(up over next 4 candles) in percent.

    Real walk-forward: for each block of `retrain_every` candles, an
    XGBoost model is trained on everything BEFORE the block, then scores
    the block. Candles before `min_train` get NaN (skipped by the
    backtester). If xgboost is unavailable, falls back to a 5-day price
    momentum dummy forecast and flags it so the UI can warn the user.
    """
    feat = engine.build_features(df).copy()
    fwd = feat["Close"].shift(-engine.FORECAST_CANDLES) / feat["Close"] - 1
    feat["_y"] = (fwd * 100 > engine.FEE_BUFFER_PCT).astype(int)
    feat["forecast"] = np.nan

    meta = {"mode": "walkforward-xgb", "retrains": 0, "fallback": False}

    if not HAVE_XGB:
        meta.update(mode="momentum-fallback", fallback=True)
        bars_5d = _bars_per_day(df) * 5
        mom = feat["Close"].pct_change(max(2, bars_5d))
        # squash momentum into a 0-100 pseudo-probability
        feat["forecast"] = 50 + 35 * np.tanh(mom * 25)
        return feat, meta

    cols = engine.FEATURES
    n = len(feat)
    start = max(min_train, 60)
    for block_start in range(start, n, retrain_every):
        # training set: strictly before this block, with matured labels only
        train = feat.iloc[:block_start].dropna(subset=cols)
        train = train.iloc[: -engine.FORECAST_CANDLES]  # drop unlabeled tail
        if len(train) < 150:
            continue
        model = XGBClassifier(
            n_estimators=150, max_depth=4, learning_rate=0.08,
            subsample=0.9, colsample_bytree=0.8,
            eval_metric="logloss", n_jobs=2,
        )
        model.fit(train[cols], train["_y"])
        meta["retrains"] += 1

        block = feat.iloc[block_start: block_start + retrain_every]
        X = block[cols].fillna(0)
        feat.loc[block.index, "forecast"] = model.predict_proba(X)[:, 1] * 100

    if feat["forecast"].notna().sum() == 0:   # not enough history for WF
        meta.update(mode="momentum-fallback", fallback=True)
        bars_5d = _bars_per_day(df) * 5
        mom = feat["Close"].pct_change(max(2, bars_5d))
        feat["forecast"] = 50 + 35 * np.tanh(mom * 25)

    return feat, meta


def _bars_per_day(df) -> int:
    if len(df) < 3:
        return 24
    sec = (df.index[1] - df.index[0]).total_seconds() or 3600
    return max(1, int(round(86400 / sec)))


# ===================================================================
# 2. the backtest engine
# ===================================================================
def run_backtest(
    df: pd.DataFrame,
    forecast_col: str = "forecast",
    price_col: str = "Close",
    initial_capital: float = 10_000.0,
    position_pct: float = 0.20,       # fraction of equity per entry
    fee_pct: float = 0.001,           # per side, includes slippage
    buy_threshold: float = 60.0,
    sell_threshold: float = 40.0,
):
    """
    Long-only signal-following simulation.
      forecast > buy_threshold  and flat  -> BUY  next bar's open
      forecast < sell_threshold and long  -> SELL next bar's open
    NaN forecasts are skipped (no decision that bar).
    Returns (metrics dict, equity_curve DataFrame, trades DataFrame).
    """
    d = df.dropna(subset=[price_col]).copy()
    if len(d) < 10:
        return _empty_result(initial_capital)

    has_open = "Open" in d.columns
    cash = initial_capital
    qty = 0.0
    entry_price = entry_cost = 0.0
    entry_time = None
    trades, eq_times, eq_vals, pos_flags = [], [], [], []

    n = len(d)
    for i in range(n):
        price = float(d[price_col].iloc[i])

        # mark-to-market equity at this bar
        eq_times.append(d.index[i])
        eq_vals.append(cash + qty * price)
        pos_flags.append(1 if qty > 0 else 0)

        if i >= n - 1:
            break  # last bar: nothing can execute "next bar"

        f = d[forecast_col].iloc[i]
        if pd.isna(f):
            continue                       # NaN forecast: skip this bar

        # execution price = NEXT bar's open (fallback: next close)
        nxt = float(d["Open"].iloc[i + 1]) if has_open \
            else float(d[price_col].iloc[i + 1])

        if qty == 0 and f > buy_threshold:
            stake = (cash + qty * price) * position_pct
            stake = min(stake, cash)
            if stake <= 0:
                continue
            qty = stake * (1 - fee_pct) / nxt
            cash -= stake
            entry_price, entry_cost, entry_time = nxt, stake, d.index[i + 1]

        elif qty > 0 and f < sell_threshold:
            proceeds = qty * nxt * (1 - fee_pct)
            pnl = proceeds - entry_cost
            trades.append({
                "Entry Time": entry_time, "Exit Time": d.index[i + 1],
                "Entry": round(entry_price, 4), "Exit": round(nxt, 4),
                "PnL": round(pnl, 2),
                "PnL %": round(pnl / entry_cost * 100, 2),
                "Result": "WIN" if pnl >= 0 else "LOSS",
            })
            cash += proceeds
            qty = 0.0

    # close any open position at the final bar for accounting
    if qty > 0:
        last = float(d[price_col].iloc[-1])
        proceeds = qty * last * (1 - fee_pct)
        pnl = proceeds - entry_cost
        trades.append({
            "Entry Time": entry_time, "Exit Time": d.index[-1],
            "Entry": round(entry_price, 4), "Exit": round(last, 4),
            "PnL": round(pnl, 2),
            "PnL %": round(pnl / entry_cost * 100, 2),
            "Result": ("WIN" if pnl >= 0 else "LOSS") + " (open at end)",
        })
        cash += proceeds
        eq_vals[-1] = cash

    equity = pd.DataFrame({"equity": eq_vals, "in_position": pos_flags},
                          index=pd.Index(eq_times, name="time"))
    # buy & hold benchmark on the same window, same fees both sides
    p0, p1 = float(d[price_col].iloc[0]), float(d[price_col].iloc[-1])
    equity["buy_hold"] = (
        initial_capital * (1 - fee_pct)
        * d[price_col].values[: len(equity)] / p0
    )
    bh_final = initial_capital * (1 - fee_pct) * (p1 / p0) * (1 - fee_pct)

    trades_df = pd.DataFrame(trades)
    metrics = _metrics(equity["equity"], trades_df, initial_capital,
                       bh_final, _bars_per_day(d))
    return metrics, equity, trades_df


# ===================================================================
# 3. performance metrics
# ===================================================================
def _metrics(eq: pd.Series, trades: pd.DataFrame, cap0: float,
             bh_final: float, bars_per_day: int) -> dict:
    final = float(eq.iloc[-1])
    total_ret = (final / cap0 - 1) * 100
    n_bars = len(eq)
    years = max(n_bars / (bars_per_day * 365), 1e-9)
    ann_ret = ((final / cap0) ** (1 / years) - 1) * 100 if final > 0 else -100.0

    rets = eq.pct_change().dropna()
    sharpe = 0.0
    if len(rets) > 2 and rets.std() > 0:
        sharpe = float(rets.mean() / rets.std()
                       * np.sqrt(bars_per_day * 365))

    dd = (eq / eq.cummax() - 1).min() * 100

    n_tr = len(trades)
    if n_tr:
        wins = trades[trades["PnL"] >= 0]["PnL"]
        losses = trades[trades["PnL"] < 0]["PnL"]
        win_rate = len(wins) / n_tr * 100
        gross_win, gross_loss = wins.sum(), -losses.sum()
        pf = (gross_win / gross_loss) if gross_loss > 0 else float("inf")
    else:
        win_rate, pf = 0.0, 0.0

    return {
        "Total Return %": round(total_ret, 2),
        "Annualized Return %": round(ann_ret, 2),
        "Sharpe Ratio": round(sharpe, 2),
        "Max Drawdown %": round(float(dd), 2),
        "Win Rate %": round(win_rate, 1),
        "Profit Factor": (round(pf, 2) if pf != float("inf") else "∞"),
        "Trades": n_tr,
        "Final Equity": round(final, 2),
        "Buy & Hold Return %": round((bh_final / cap0 - 1) * 100, 2),
        "Strategy vs B&H (pp)": round(total_ret - (bh_final / cap0 - 1) * 100, 2),
    }


def _empty_result(cap0):
    return (
        {"Total Return %": 0, "Annualized Return %": 0, "Sharpe Ratio": 0,
         "Max Drawdown %": 0, "Win Rate %": 0, "Profit Factor": 0,
         "Trades": 0, "Final Equity": cap0, "Buy & Hold Return %": 0,
         "Strategy vs B&H (pp)": 0},
        pd.DataFrame(columns=["equity", "in_position", "buy_hold"]),
        pd.DataFrame(),
    )