turtle_quant_1.backtesting.engine

turtle_quant_1.backtesting.engine

Backtesting engine for strategy evaluation.

BacktestingEngine

BacktestingEngine(
    strategy_engine: BaseStrategyEngine,
    symbols: list[str] = BACKTESTING_SYMBOLS,
    initial_capital: float = 0.0,
    max_history_days: int = MAX_HISTORY_DAYS,
    max_lookback_days: int = BACKTESTING_MAX_LOOKBACK_DAYS,
    max_lookforward_days: int = BACKTESTING_MAX_LOOKFORWARD_DAYS,
    tick_interval: CandleUnit = "4H",
    event_interval: CandleUnit = "30M",
)

Backtesting engine for evaluating trading strategies.

Initialize the backtesting engine.

Parameters:

Name	Type	Description	Default
strategy_engine	BaseStrategyEngine	The strategy engine to use for generating signals.	required
symbols	list[str]	List of symbols to backtest. If None, uses BACKTESTING_SYMBOLS.	BACKTESTING_SYMBOLS
initial_capital	float	Starting capital for the portfolio (default: 0.0).	0.0
max_history_days	int	Maximum days of history to download.	MAX_HISTORY_DAYS
max_lookback_days	int	Days of data to use for strategy signals.	BACKTESTING_MAX_LOOKBACK_DAYS
max_lookforward_days	int	Days of data to simulate trading on.	BACKTESTING_MAX_LOOKFORWARD_DAYS
tick_interval	CandleUnit	Duration of one tick window as a candle-unit string (e.g. "4H"). Passed through to TradingEngine to determine how many candle events are replayed per tick.	'4H'
event_interval	CandleUnit	Duration of one event window as a candle-unit string (e.g. "30M"). Passed through to TradingEngine to determine how many candle events are replayed per event.	'30M'

Source code in src/turtle_quant_1/backtesting/engine.py

def __init__(
    self,
    strategy_engine: BaseStrategyEngine,
    symbols: list[str] = BACKTESTING_SYMBOLS,
    initial_capital: float = 0.0,
    max_history_days: int = MAX_HISTORY_DAYS,
    max_lookback_days: int = BACKTESTING_MAX_LOOKBACK_DAYS,
    max_lookforward_days: int = BACKTESTING_MAX_LOOKFORWARD_DAYS,
    tick_interval: CandleUnit = "4H",
    event_interval: CandleUnit = "30M",
):
    """Initialize the backtesting engine.

    Args:
        strategy_engine: The strategy engine to use for generating signals.
        symbols: List of symbols to backtest. If None, uses BACKTESTING_SYMBOLS.
        initial_capital: Starting capital for the portfolio (default: 0.0).
        max_history_days: Maximum days of history to download.
        max_lookback_days: Days of data to use for strategy signals.
        max_lookforward_days: Days of data to simulate trading on.
        tick_interval: Duration of one tick window as a candle-unit string
            (e.g. "4H"). Passed through to TradingEngine to determine how
            many candle events are replayed per tick.
        event_interval: Duration of one event window as a candle-unit string
            (e.g. "30M"). Passed through to TradingEngine to determine how
            many candle events are replayed per event.
    """
    self.strategy_engine = strategy_engine
    self.symbols = symbols
    self.initial_capital = initial_capital
    self.max_history_days = max_history_days
    self.max_lookback_days = max_lookback_days
    self.max_lookforward_days = max_lookforward_days

    self.trading_engine = TradingEngine(
        strategy_engine=strategy_engine,
        portfolio=Portfolio(initial_capital),
        max_lookback_days=max_lookback_days,
        tick_interval=tick_interval,
        event_interval=event_interval,
    )
    # Convenience shorthand so the rest of this class can use self.portfolio directly.
    self.portfolio = self.trading_engine.portfolio

    # Initialize data management components
    self.data_processor = DataProcessor(symbols=self.symbols)
    self.data_cache: dict[str, pd.DataFrame] = {}
    cache = get_global_data_cache()

    for symbol in self.symbols:
        # Check if data exists in cache
        data = cache.get(symbol)
        if data is None:
            # Load and cache the data
            data = self._load_data_for_symbol(symbol, impute_data=True)
            cache.set(symbol, data)

        self.data_cache[symbol] = data

        # This reduces duplicated calculations across all DataUnitConverter instances,
        # because data conversion is a static utility method.
        DataUnitConverter.preload_global_instance_cache(symbol, data, "15M")
        DataUnitConverter.preload_global_instance_cache(symbol, data, "30M")
        DataUnitConverter.preload_global_instance_cache(symbol, data, "1H")
        DataUnitConverter.preload_global_instance_cache(symbol, data, "2H")
        DataUnitConverter.preload_global_instance_cache(symbol, data, "4H")
        DataUnitConverter.preload_global_instance_cache(symbol, data, "1D")
        DataUnitConverter.preload_global_instance_cache(symbol, data, "1W")

        # This reduces duplicated calculations across all SupResIndicator instances,
        # because the cache is a global singleton.
        # This needs to be done before the BaseStrategyEngine fires off separate processes.
        SupResIndicator.preload_global_instance_cache(symbol, data)

    # For quantstats metrics
    self.portfolio_returns: List[Tuple[datetime, float]] = []

    logger.info(f"Initialized backtesting engine for symbols: {self.symbols}")

get_metrics

get_metrics(benchmark: str) -> Dict

Calculate metrics for the portfolio performance.

NOTE: This is currently a wrapper around quantstats metrics. TODO: Implement our own metrics.

Parameters:

Name	Type	Description	Default
benchmark	str	Benchmark symbol to compare against (default: SPY)	required

Returns:

Type	Description
Dict	Dictionary containing quantstats metrics

Source code in src/turtle_quant_1/backtesting/engine.py

def get_metrics(self, benchmark: str) -> Dict:
    """Calculate metrics for the portfolio performance.

    NOTE: This is currently a wrapper around quantstats metrics.
    TODO: Implement our own metrics.

    Args:
        benchmark: Benchmark symbol to compare against (default: SPY)

    Returns:
        Dictionary containing quantstats metrics
    """
    if len(self.portfolio_returns) < 2:
        logger.warning("Not enough portfolio returns data for metrics")
        logger.warning("Run the backtest first before calculating metrics")
        return {}

    try:
        # Convert portfolio values to returns
        df_returns = pd.DataFrame(
            self.portfolio_returns, columns=["date", "portfolio_value"]
        )

        # Remove duplicate timestamps if any
        df_returns = df_returns.drop_duplicates(subset=["date"], keep="last")

        # Sort by date to ensure proper chronological order
        df_returns = df_returns.sort_values("date")

        # Set index and calculate returns
        df_returns = df_returns.set_index("date")
        df_returns["returns"] = df_returns["portfolio_value"].pct_change().dropna()

        if df_returns["returns"].empty or len(df_returns["returns"]) < 2:
            logger.warning("Insufficient returns data for quantstats metrics")
            return {}

        # Calculate quantstats metrics
        metrics = qs.reports.metrics(
            returns=df_returns["returns"],
            benchmark=benchmark,
            display=False,
        )
        if metrics is None:
            raise ValueError("Quantstats metrics returned None")

        return (
            metrics.to_dict() if hasattr(metrics, "to_dict") else dict(metrics)
        )  # pyrefly: ignore[no-any-return]

    except Exception as e:
        logger.error(f"Failed to calculate quantstats metrics: {e}")
        return {}

run_backtest

run_backtest() -> TestCaseResults

Run the backtesting simulation.

Returns:

Type	Description
TestCaseResults	BacktestingResults with backtesting results.

Source code in src/turtle_quant_1/backtesting/engine.py

def run_backtest(self) -> TestCaseResults:
    """Run the backtesting simulation.

    Returns:
        BacktestingResults with backtesting results.
    """
    logger.info("Starting backtesting simulation...")

    # Find the overall time range across all symbols
    history_start, history_end = self._get_simulation_time_range(self.symbols)

    # Split into lookback and simulation periods
    lookback_end = history_start + timedelta(days=self.max_lookback_days)
    simulation_start = lookback_end
    simulation_end = min(
        history_end,
        simulation_start + timedelta(days=self.max_lookforward_days),
    )

    logger.info(f"Simulation period: {simulation_start} to {simulation_end}")

    # Generate simulation ticks for the entire simulation period
    simulation_ticks = self._generate_simulation_ticks(
        simulation_start, simulation_end
    )

    logger.info(
        f"Generated {len(simulation_ticks)} simulation ticks for simulation"
    )

    if len(simulation_ticks) < 2:
        raise ValueError(
            f"Not enough simulation ticks for meaningful backtesting: {len(simulation_ticks)}"
        )

    # Get current prices for all symbols at this timestamp
    curr_prices = {}

    for timestamp in tqdm(simulation_ticks, desc="Backtesting", unit="tick"):
        # Generate signals for each symbol
        for symbol in self.symbols:
            # Get full historical data up to this point for strategy analysis
            full_data = self.data_cache[symbol]
            # Filter data to only include data up to the current timestamp to prevent data leakage
            # i.e. This caps the newest timestamp.
            curr_data_mask = full_data["datetime"] <= timestamp
            curr_data = full_data[curr_data_mask]

            self.trading_engine.handle_tick(symbol, curr_data)

            curr_prices[symbol] = curr_data.iloc[-1]["Close"]

        # Record portfolio value for returns calculation (once per timestamp)
        if curr_prices:
            portfolio_value = self.portfolio.get_portfolio_value(curr_prices)
            self.portfolio_returns.append((timestamp, portfolio_value))

    # Calculate final results
    final_prices = {}
    for symbol in self.symbols:
        data = self.data_cache[symbol]
        if not data.empty:
            # Get the last available price for each symbol
            final_data = data[data["datetime"] <= simulation_end]
            if not final_data.empty:
                final_prices[symbol] = final_data.iloc[-1]["Close"]

    final_portfolio_value = self.portfolio.get_portfolio_value(final_prices)
    total_return = self.portfolio.get_return_dollars(final_prices)

    # Create portfolio summary and results directly
    portfolio_summary = self.portfolio.get_summary(final_prices)

    # Build per-symbol price history sliced to the simulation window
    price_history: dict[str, pd.DataFrame] = {}
    for symbol in self.symbols:
        data = self.data_cache[symbol]
        if not data.empty:
            mask = (data["datetime"] >= simulation_start) & (
                data["datetime"] <= simulation_end
            )
            price_history[symbol] = data[mask].copy().reset_index(drop=True)

    # Convert portfolio_returns list to a DataFrame
    portfolio_value_history = pd.DataFrame(
        self.portfolio_returns, columns=["datetime", "portfolio_value"]
    )

    # Convert signal list to a DataFrame
    signal_history_list = self.trading_engine.signal_history_list
    signal_history = (
        pd.DataFrame(signal_history_list)
        if signal_history_list
        else pd.DataFrame(
            columns=[
                "datetime",
                "symbol",
                "action",
                "score",
                "strategies",
                "executed",
            ]
        )
    )

    results = TestCaseResults(
        initial_capital=self.initial_capital,
        final_portfolio_value=final_portfolio_value,
        final_holdings=self.portfolio.get_current_holdings(),
        final_cash=self.portfolio.cash,
        total_return_dollars=total_return,
        total_return_percent=self.portfolio.get_return_percent(final_prices),
        portfolio_value_history=portfolio_value_history,
        portfolio_summary=PortfolioSummary(**portfolio_summary),
        price_history=price_history,
        total_signals=self.trading_engine.total_signals,
        total_transactions=self.trading_engine.total_transactions,
        total_simulation_ticks=len(simulation_ticks),
        symbols_traded=self.symbols,
        transaction_history=self.portfolio.get_transaction_history(),
        signal_history=signal_history,
        simulation_start=simulation_start,
        simulation_end=simulation_end,
        metrics=self.get_metrics(benchmark="SPY"),  # TODO: Hard-coded.
    )

    logger.info(
        f"Backtesting complete. Total return: ${total_return:.2f} "
        f"({self.portfolio.get_return_percent(final_prices):.2f}%) | "
        f"Signals generated: {self.trading_engine.total_signals} | "
        f"Transactions: {self.trading_engine.total_transactions}"
    )

    return results

get_global_data_cache

get_global_data_cache()

Get the global data cache instance.

Source code in src/turtle_quant_1/backtesting/engine.py

def get_global_data_cache():
    """Get the global data cache instance."""
    return _global_data_cache