def backtest_bot(
bot: Bot,
initial_capital: float = 10000.0,
save_to_db: bool = True,
data: pd.DataFrame | dict[str, pd.DataFrame] | None = None,
# Defaulted from config.py's constants rather than repeating the literals.
# config.py's ExecutionConfig carries a comment saying its values "MUST mirror
# backtest_bot()'s defaults ... or the live equity curve stops being comparable
# to the backtested one" — an invariant that was enforced only by 0.0005 being
# typed correctly in two files. Now there is one source of truth.
slippage_pct: float = DEFAULT_SLIPPAGE_PCT,
commission_pct: float = DEFAULT_COMMISSION_PCT,
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:
# Benchmark tickers are loaded and kept aligned like any other ticker
# (a strategy may need one as a relative-strength baseline) but are
# excluded from the divisor and never traded — matching
# Bot._multi_ticker_target_weights. getattr, not the property, because
# backtest_bot accepts instances built with a stubbed __init__.
benchmarks = set(getattr(bot, "benchmark_tickers", ()) or ())
tradeable = [t for t in tickers if t not in benchmarks] or tickers
n_trade = len(tradeable)
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}") from 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 tradeable)
target = total_value / n_trade
band = EXECUTION_CONFIG.no_trade_threshold(target)
# Decide for every tradeable ticker before trading any of them, so
# exits can fund entries — mirroring the live path, where
# rebalance_portfolio executes all sells before any buy.
decisions: dict[str, int] = {}
for ticker in tradeable:
try:
bot._current_ticker = ticker
decisions[ticker] = bot.decisionFunction(rows[ticker])
except Exception as e:
logger.warning(f"Error in decisionFunction for {ticker} at {ts}: {e}")
decisions[ticker] = 0
# Phase 1: exits and trims. decision 0 caps at one sleeve but is
# never funded; decision -1 exits fully and ignores the band.
for ticker, decision in decisions.items():
price = prices[ticker]
holding = portfolio.get(ticker, 0.0)
if holding <= 0:
continue
holding_value = holding * price
if decision == -1:
wanted = 0.0
elif decision == 1:
wanted = target
else:
wanted = min(holding_value, target)
excess = holding_value - wanted
if excess <= (0.0 if decision == -1 else band):
continue
qty = min(holding, excess / price)
execution_price = price * (1 - slippage_pct)
cash_proceeds = qty * execution_price
net_proceeds = cash_proceeds - cash_proceeds * commission_pct
portfolio["USD"] = portfolio.get("USD", 0.0) + net_proceeds
portfolio[ticker] = holding - qty
nrtrades += 1
# Phase 2: entries and top-ups, funded by the proceeds above.
for ticker, decision in decisions.items():
if decision != 1:
continue
price = prices[ticker]
holding = portfolio.get(ticker, 0.0)
shortfall = target - holding * price
if shortfall <= band:
continue
cash = portfolio.get("USD", 0.0)
buy_amount = min(shortfall, cash)
if buy_amount <= 0:
continue
commission_cost = buy_amount * commission_pct
available = buy_amount - commission_cost
execution_price = price * (1 + slippage_pct)
portfolio["USD"] = cash - buy_amount
portfolio[ticker] = holding + available / execution_price
nrtrades += 1
current_total = portfolio.get("USD", 0.0) + sum(portfolio.get(t, 0.0) * prices[t] for t in tradeable)
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 the
# TRADEABLE tickers. Including a benchmark here would average SPY into
# the very number SPY is the benchmark for.
bh_returns = []
for _ticker, df in ((t, data_dict[t]) for t in tradeable if t in data_dict):
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 None:
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}") from e
elif isinstance(data, dict):
# Unwrap single-ticker dict (e.g., passed from hyperparameter tuner)
data = cast(pd.DataFrame, data.get(symbol, next(iter(data.values()))))
# Narrow type: assert data is not a dict (helps mypy)
assert not isinstance(data, dict)
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."
)
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)
# Final type narrowing for mypy
assert not isinstance(data, dict), "data should be a DataFrame at this point"
# 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)
# No-trade band, mirroring the live path. This was previously applied on
# the multi-ticker branch only, so a single-ticker bot traded on every
# signal in backtest while the same bot live went through
# PortfolioManager.should_trade and skipped sub-band adjustments. The
# backtested trade count and equity curve therefore did not describe the
# strategy that actually ran.
position_value = holdings * current_price
if decision == 1:
# An entry does not bypass the band: easy to exit, hard to enter.
if cash > 0 and should_trade(cash, cash + position_value):
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
# Selling the whole position is a full exit, which always trades —
# otherwise a position smaller than the band could never be closed.
elif decision == -1 and holdings > 0 and should_trade(position_value, position_value, is_full_exit=True):
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