Backtesting API Reference

tradingbot.utils.backtest.backtest_bot(bot: Bot, initial_capital: float = 10000.0, save_to_db: bool = True, data: Optional[Union[pd.DataFrame, Dict[str, pd.DataFrame]]] = None, slippage_pct: float = 0.0005, commission_pct: float = 0.0, risk_free_rate: float = 0.0, save_results_to_db: bool = True) -> dict

Backtest a trading bot over historical data.

Works for both single-ticker and multi-ticker bots that implement decisionFunction(). Multi-ticker bots use equal-weight position sizing: each ticker targets total_portfolio_value / N.

Parameters:

Name	Type	Description	Default
bot	Bot	Bot instance to backtest (must have decisionFunction implemented).	required
initial_capital	float	Starting capital in USD (default: $10,000).	10000.0
save_to_db	bool	Whether to save fetched data to database (default: True).	True
data	Optional[Union[DataFrame, Dict[str, DataFrame]]]	Optional pre-fetched data. - Single-ticker: pd.DataFrame with timestamp/close/TA columns. - Multi-ticker: dict[str, pd.DataFrame] keyed by ticker symbol. If None, data is fetched automatically for all bot.tickers.	None
slippage_pct	float	One-way slippage as fraction of price (default: 0.05%).	0.0005
commission_pct	float	Commission as fraction of trade value (default: 0.0).	0.0
risk_free_rate	float	Annualized risk-free rate for Sharpe (default: 0.0).	0.0
save_results_to_db	bool	Whether to save best result to database.	True

Returns:

Type	Description
dict	Dictionary with keys: yearly_return, buy_hold_return, sharpe_ratio,
dict	nrtrades, maxdrawdown.

Raises:

Type	Description
NotImplementedError	If bot doesn't implement decisionFunction.
ValueError	If insufficient data is available for backtesting.

Source code in tradingbot/utils/backtest.py

def backtest_bot(
    bot: Bot,
    initial_capital: float = 10000.0,
    save_to_db: bool = True,
    data: Optional[Union[pd.DataFrame, Dict[str, pd.DataFrame]]] = None,
    slippage_pct: float = 0.0005,
    commission_pct: float = 0.0,
    risk_free_rate: float = 0.0,
    save_results_to_db: bool = True,
) -> dict:
    """
    Backtest a trading bot over historical data.

    Works for both single-ticker and multi-ticker bots that implement
    decisionFunction(). Multi-ticker bots use equal-weight position sizing:
    each ticker targets total_portfolio_value / N.

    Args:
        bot: Bot instance to backtest (must have decisionFunction implemented).
        initial_capital: Starting capital in USD (default: $10,000).
        save_to_db: Whether to save fetched data to database (default: True).
        data: Optional pre-fetched data.
              - Single-ticker: pd.DataFrame with timestamp/close/TA columns.
              - Multi-ticker: dict[str, pd.DataFrame] keyed by ticker symbol.
              If None, data is fetched automatically for all bot.tickers.
        slippage_pct: One-way slippage as fraction of price (default: 0.05%).
        commission_pct: Commission as fraction of trade value (default: 0.0).
        risk_free_rate: Annualized risk-free rate for Sharpe (default: 0.0).
        save_results_to_db: Whether to save best result to database.

    Returns:
        Dictionary with keys: yearly_return, buy_hold_return, sharpe_ratio,
        nrtrades, maxdrawdown.

    Raises:
        NotImplementedError: If bot doesn't implement decisionFunction.
        ValueError: If insufficient data is available for backtesting.
    """
    if type(bot).decisionFunction is Bot.decisionFunction:
        raise NotImplementedError(
            "Bot must implement decisionFunction() for backtesting. "
            "Bots that only override makeOneIteration() are not supported."
        )

    tickers = getattr(bot, "tickers", None) or ([bot.symbol] if bot.symbol else [])
    if not tickers:
        raise ValueError("Bot must have tickers or symbol defined for backtesting.")
    N = len(tickers)

    # ------------------------------------------------------------------ #
    #  Multi-ticker path (N > 1)                                          #
    # ------------------------------------------------------------------ #
    if N > 1:
        backtest_period = None
        data_dict: Dict[str, pd.DataFrame] = {}

        if isinstance(data, dict):
            data_dict = data
        elif data is not None:
            raise ValueError(
                "For multi-ticker bots, 'data' must be a dict[str, pd.DataFrame]. "
                "Pass None to fetch automatically."
            )
        else:
            backtest_period = _get_backtest_period(bot.interval)
            for ticker in tickers:
                try:
                    df = bot.getYFDataWithTA(
                        symbol=ticker,
                        interval=bot.interval,
                        period=backtest_period,
                        saveToDB=save_to_db,
                    )
                    data_dict[ticker] = df
                except Exception as e:
                    raise ValueError(f"Failed to fetch data for {ticker}: {e}")

        # Sort and index each DataFrame by timestamp
        indexed: Dict[str, pd.DataFrame] = {}
        for ticker, df in data_dict.items():
            if df.empty or len(df) < 2:
                raise ValueError(f"Insufficient data for ticker {ticker}")
            if "timestamp" in df.columns:
                df = df.sort_values("timestamp").reset_index(drop=True)
                data_dict[ticker] = df
                indexed[ticker] = df.set_index("timestamp")
            else:
                indexed[ticker] = df.sort_index()

        if backtest_period:
            bot.datasettings = (bot.interval, backtest_period)

        # Inner-join on common timestamps
        common_ts = indexed[tickers[0]].index
        for t in tickers[1:]:
            common_ts = common_ts.intersection(indexed[t].index)
        common_ts = sorted(common_ts)
        if len(common_ts) < 2:
            raise ValueError(
                "Insufficient common timestamps across tickers for multi-ticker backtest."
            )

        has_ta_columns = all("trend_adx" in indexed[t].columns for t in tickers)

        portfolio: Dict[str, float] = {"USD": initial_capital}
        portfolio_values: list = []
        portfolio_timestamps: list = []
        nrtrades = 0

        for ts in common_ts:
            rows = {t: indexed[t].loc[ts] for t in tickers}

            # Update bot's datas cache with current slice to prevent look-ahead bias
            # if the bot uses self.datas[ticker] inside decisionFunction.
            bot.datas = {t: indexed[t].loc[:ts] for t in tickers}

            # Validate prices for all tickers
            prices: Dict[str, float] = {}
            valid = True
            for ticker, row in rows.items():
                try:
                    price = float(row["close"])
                    if price <= 0 or not np.isfinite(price):
                        valid = False
                        break
                    prices[ticker] = price
                except (KeyError, ValueError, TypeError):
                    valid = False
                    break
            if not valid:
                continue

            # Skip warmup bars (any ticker with trend_adx == 0 = still warming up)
            if has_ta_columns and any(rows[t]["trend_adx"] == 0.0 for t in tickers):
                continue

            total_value = portfolio.get("USD", 0.0) + sum(
                portfolio.get(t, 0.0) * prices[t] for t in tickers
            )
            target = total_value / N

            for ticker in tickers:
                try:
                    bot._current_ticker = ticker
                    decision = bot.decisionFunction(rows[ticker])
                except Exception as e:
                    logger.warning(f"Error in decisionFunction for {ticker} at {ts}: {e}")
                    decision = 0

                price = prices[ticker]
                holding = portfolio.get(ticker, 0.0)
                holding_value = holding * price

                if decision == 1:
                    shortfall = target - holding_value
                    if shortfall > 0:
                        cash = portfolio.get("USD", 0.0)
                        buy_amount = min(shortfall, cash)
                        if buy_amount > 0:
                            commission_cost = buy_amount * commission_pct
                            available = buy_amount - commission_cost
                            execution_price = price * (1 + slippage_pct)
                            qty = available / execution_price
                            portfolio["USD"] = cash - buy_amount
                            portfolio[ticker] = holding + qty
                            nrtrades += 1
                elif decision == -1 and holding > 0:
                    execution_price = price * (1 - slippage_pct)
                    cash_proceeds = holding * execution_price
                    commission_cost = cash_proceeds * commission_pct
                    net_proceeds = cash_proceeds - commission_cost
                    portfolio["USD"] = portfolio.get("USD", 0.0) + net_proceeds
                    portfolio[ticker] = 0.0
                    nrtrades += 1

            current_total = portfolio.get("USD", 0.0) + sum(
                portfolio.get(t, 0.0) * prices[t] for t in tickers
            )
            portfolio_values.append(current_total)
            portfolio_timestamps.append(ts)

        metrics = _compute_backtest_metrics(portfolio_values, bot.interval, risk_free_rate)

        # Buy-and-hold: equal-weight mean of individual B&H returns across tickers
        bh_returns = []
        for ticker, df in data_dict.items():
            close = df["close"].dropna()
            if len(close) >= 2:
                first = float(close.iloc[0])
                last = float(close.iloc[-1])
                if first > 0 and np.isfinite(first) and np.isfinite(last):
                    bh_returns.append((last - first) / first)
        buy_hold_return = float(np.mean(bh_returns)) if bh_returns else 0.0

        result = {**metrics, "nrtrades": int(nrtrades), "buy_hold_return": buy_hold_return}

        if save_results_to_db:
            _save_backtest_to_db(
                bot=bot,
                symbol_key=",".join(tickers),
                result=result,
                portfolio_values=portfolio_values,
                portfolio_timestamps=portfolio_timestamps,
                data_for_qs=data_dict[tickers[0]],
            )

        return result

    # ------------------------------------------------------------------ #
    #  Single-ticker path (N == 1)                                        #
    # ------------------------------------------------------------------ #
    symbol = tickers[0]
    backtest_period = None

    if data is not None:
        if isinstance(data, dict):
            # Unwrap single-ticker dict (e.g., passed from hyperparameter tuner)
            data = data.get(symbol, next(iter(data.values())))
        if "close" not in data.columns or "timestamp" not in data.columns:
            raise ValueError(
                "Provided data must have 'close' and 'timestamp' columns. "
                "It should also include all TA indicators required by decisionFunction."
            )
    else:
        backtest_period = _get_backtest_period(bot.interval)
        try:
            data = bot.getYFDataWithTA(
                symbol=symbol,
                interval=bot.interval,
                period=backtest_period,
                saveToDB=save_to_db,
            )
        except Exception as e:
            raise ValueError(f"Failed to fetch historical data: {e}")

    if data.empty:
        raise ValueError("No historical data available for backtesting")
    if len(data) < 2:
        raise ValueError("Insufficient data points for backtesting (need at least 2)")

    if "timestamp" in data.columns:
        data = data.sort_values("timestamp").reset_index(drop=True)
    elif data.index.name in ["timestamp", "date", "datetime"]:
        data = data.sort_index()

    if backtest_period:
        bot.datasettings = (bot.interval, backtest_period)

    # trend_adx has ~26-bar warmup; warmup rows have trend_adx == 0.0 after fillna.
    has_ta_columns = "trend_adx" in data.columns

    portfolio = {"USD": initial_capital}
    portfolio_values = []
    portfolio_timestamps = []
    nrtrades = 0

    for idx, row in data.iterrows():
        # Update bot's data cache with current slice to prevent look-ahead bias
        # if the bot uses self.data inside decisionFunction.
        bot.data = data.iloc[:idx+1]

        try:
            current_price = float(row["close"])
        except (KeyError, ValueError, TypeError):
            continue
        if current_price <= 0 or not np.isfinite(current_price):
            continue
        if has_ta_columns and row["trend_adx"] == 0.0:
            continue

        try:
            decision = bot.decisionFunction(row)
        except Exception as e:
            logger.warning(f"Error in decisionFunction at row {idx}: {e}")
            decision = 0

        cash = portfolio.get("USD", 0.0)
        holdings = portfolio.get(symbol, 0.0)

        if decision == 1:
            if cash > 0:
                execution_price = current_price * (1 + slippage_pct)
                commission_cost = cash * commission_pct
                available = cash - commission_cost
                quantity = available / execution_price
                portfolio["USD"] = 0.0
                portfolio[symbol] = holdings + quantity
                nrtrades += 1
        elif decision == -1:
            if holdings > 0:
                execution_price = current_price * (1 - slippage_pct)
                cash_proceeds = holdings * execution_price
                commission_cost = cash_proceeds * commission_pct
                net_proceeds = cash_proceeds - commission_cost
                portfolio["USD"] = cash + net_proceeds
                portfolio[symbol] = 0.0
                nrtrades += 1

        current_cash = portfolio.get("USD", 0.0)
        current_holdings = portfolio.get(symbol, 0.0)
        portfolio_value = current_cash + (current_holdings * current_price)
        portfolio_values.append(portfolio_value)
        portfolio_timestamps.append(row["timestamp"] if "timestamp" in row.index else None)

    metrics = _compute_backtest_metrics(portfolio_values, bot.interval, risk_free_rate)

    close = data["close"].dropna()
    if len(close) < 2:
        buy_hold_return = 0.0
    else:
        first_close = float(close.iloc[0])
        last_close = float(close.iloc[-1])
        if first_close > 0 and np.isfinite(first_close) and np.isfinite(last_close):
            buy_hold_return = float((last_close - first_close) / first_close)
        else:
            buy_hold_return = 0.0

    result = {**metrics, "nrtrades": int(nrtrades), "buy_hold_return": buy_hold_return}

    if save_results_to_db:
        _save_backtest_to_db(
            bot=bot,
            symbol_key=symbol,
            result=result,
            portfolio_values=portfolio_values,
            portfolio_timestamps=portfolio_timestamps,
            data_for_qs=data,
        )

    return result