Base Backtest#

PyArrow-based Backtesting Engine Module.

This module implements a high-performance backtesting engine using PyArrow for efficient columnar data processing. The backtester simulates portfolio performance with support for rebalancing, pluggable commission and slippage models, and benchmark comparison.

The implementation follows a component-based architecture with: - PortfolioStateManager: Handles portfolio state and holdings - ExecutionBroker: Simulates trade execution with cost models - ReportingEngine: Generates structured output reports

This is the primary backtester implementation for the backtest engine, designed for production-grade performance and reliability.

class PyArrowBacktester(*, configuration: Configuration, market_data_bundle: MarketDataBundlePyarrow, portfolio_dict: dict[Timestamp, dict[str, float]], benchmark: Table | None, portfolio_df: DataFrame | None = None, commission_model: CommissionModelInterface | None = None, slippage_model: SlippageModelInterface | None = None)#

Bases: BaseBacktesterInterface

High-performance PyArrow-based portfolio backtesting engine.

This class implements a comprehensive backtesting system using PyArrow for efficient columnar data processing. It simulates realistic portfolio performance including rebalancing, pluggable commission and slippage models, cash reserves, and benchmark comparison.

The backtester uses a component-based architecture: - PortfolioStateManager: Manages holdings, cash, and portfolio state - ExecutionBroker: Simulates trade execution with cost models - ReportingEngine: Generates structured output reports

Key Features#

  • PyArrow-based data processing for performance

  • Multiple execution price sources (VWAP, adjusted close, etc.)

  • Dynamic portfolio rebalancing with configurable schedules

  • Pluggable commission and slippage models

  • Cash reserve management

  • Comprehensive result logging and reporting

  • Benchmark comparison

ivar configuration:

User-provided backtest configuration.

vartype configuration:

Configuration

ivar market_data_bundle:

Container for market data in PyArrow format.

vartype market_data_bundle:

MarketDataBundlePyarrow

ivar portfolio_manager:

Component managing portfolio state.

vartype portfolio_manager:

PortfolioStateManager

ivar execution_broker:

Component simulating trade execution.

vartype execution_broker:

ExecutionBroker

ivar reporting_engine:

Component generating output reports.

vartype reporting_engine:

ReportingEngine

ivar benchmark:

Benchmark data for comparison.

vartype benchmark:

pa.Table | None

ivar tickers:

List of ticker symbols in the universe (read-only).

vartype tickers:

list[str]

ivar rebalance_dates:

Sorted list of rebalancing dates (read-only).

vartype rebalance_dates:

list[pd.Timestamp]

Examples

Using factory method (recommended):

>>> backtester = PyArrowBacktester.create_from_configuration(
...     configuration=config,
...     input_handlers=[csv_handler],
...     portfolio_handlers=[portfolio_handler],
... )
>>> results = backtester.run()

With custom cost models:

>>> from kaxanuk.backtest_engine.backtest.models import (
...     FixedSlippage,
...     PercentageCommission,
... )
>>> backtester = PyArrowBacktester.create_from_configuration(
...     configuration=config,
...     input_handlers=[csv_handler],
...     portfolio_handlers=[portfolio_handler],
...     commission_model=PercentageCommission(pct=0.001),
...     slippage_model=FixedSlippage(slippage_pct=0.0005),
... )

Notes

Direct instantiation is not recommended. Use the factory method create_from_configuration() to ensure proper data validation and pipeline execution.

See also

BaseBacktesterInterface

Base class defining the backtester contract

PortfolioStateManager

Component for portfolio state management

ExecutionBroker

Component for trade execution

ReportingEngine

Component for report generation

classmethod create_from_configuration(configuration: Configuration, input_handlers: list[InputHandlerInterface], portfolio_handlers: list[PortfolioInputHandlerInterface], commission_model: CommissionModelInterface | None = None, slippage_model: SlippageModelInterface | None = None) → PyArrowBacktester#

Create a fully validated PyArrowBacktester from Configuration.

This is the primary factory method used by orchestrators. It executes the complete data pipeline to load and validate market data and portfolio definitions before constructing the backtester.

Parameters:

  • configuration (Configuration) – Global configuration object containing all backtest parameters.

  • input_handlers (list[InputHandlerInterface]) – Input handlers for reading market data. Typically includes CsvInput or ParquetInput handlers.

  • portfolio_handlers (list[PortfolioInputHandlerInterface]) – Input handlers for reading portfolio definitions. Supports CSV and Excel portfolio formats.

  • commission_model (CommissionModelInterface | None, optional) – Custom commission model for trade execution. If None, uses configuration.commission_cents as per-share rate. Default is None.

  • slippage_model (SlippageModelInterface | None, optional) – Custom slippage model for price impact. If None, uses NoSlippage (zero slippage). Default is None.

Returns:

Fully initialized backtester ready for execution.

Return type:

PyArrowBacktester

Raises:

  • MissingDataError – If required market data is missing for any ticker.

  • MissingPortfolioError – If portfolio definition cannot be read.

  • ConfigurationError – If configuration is invalid.

Examples

Basic usage:

>>> from kaxanuk.backtest_engine.input_handlers import CsvInput
>>> from kaxanuk.backtest_engine.input_handlers import CsvPortfolioInputHandler
>>> backtester = PyArrowBacktester.create_from_configuration(
...     configuration=config,
...     input_handlers=[CsvInput(input_dir="./data")],
...     portfolio_handlers=[CsvPortfolioInputHandler(base_dir="./portfolios")],
... )
classmethod create_from_pipeline_result(configuration: Configuration, pipeline_result: DataPipelineResult, commission_model: CommissionModelInterface | None = None, slippage_model: SlippageModelInterface | None = None) → PyArrowBacktester#

Construct PyArrowBacktester from validated DataPipelineResult.

This factory method is useful when the data pipeline has already been executed separately, allowing for more granular control over the data loading process.

Parameters:

  • configuration (Configuration) – Configuration object for the backtest.

  • pipeline_result (DataPipelineResult) – Validated market data, portfolio, and benchmark from pipeline.

  • commission_model (CommissionModelInterface | None, optional) – Custom commission model or None for default. Default is None.

  • slippage_model (SlippageModelInterface | None, optional) – Custom slippage model or None for default. Default is None.

Returns:

Ready-to-run backtester instance.

Return type:

PyArrowBacktester

Examples

>>> from kaxanuk.backtest_engine.data_processors import execute_data_pipeline
>>> pipeline_result = execute_data_pipeline(
...     configuration=config,
...     input_handlers=handlers,
...     portfolio_handlers=portfolio_handlers,
... )
>>> backtester = PyArrowBacktester.create_from_pipeline_result(
...     configuration=config,
...     pipeline_result=pipeline_result,
... )
property execution_broker: ExecutionBroker#

Get the execution broker component.

Returns:

The component simulating trade execution.

Return type:

ExecutionBroker

property portfolio_manager: PortfolioStateManager#

Get the portfolio state manager component.

Returns:

The component managing portfolio state.

Return type:

PortfolioStateManager

property rebalance_dates: list[Timestamp]#

Get the chronologically sorted list of rebalance dates.

These are the dates on which the backtester will compute new target weights and place trades. They are derived from the portfolio dictionary keys passed at construction time and sorted ascending.

Returns:

Copy of the internal rebalance dates list, sorted ascending. Modifying the returned list does not affect the backtester.

Return type:

list[pandas.Timestamp]

Examples

>>> backtester.rebalance_dates[:2]
[Timestamp('2024-01-02 00:00:00'), Timestamp('2024-04-01 00:00:00')]
>>> len(backtester.rebalance_dates)
4

Notes

Trading days that are not rebalance dates still produce a daily portfolio snapshot in the final report; only the trade execution is skipped on those days.

See also

tickers

Companion property listing the trading universe.

property reporting_engine: ReportingEngine#

Get the reporting engine component.

Returns:

The component generating output reports.

Return type:

ReportingEngine

run() → BacktestResultDict#

Execute the complete backtest simulation.

Iterates through every trading day in the execution price table, performing a rebalance whenever the current date matches one of the scheduled rebalance dates, and recording the portfolio state at the end of each day. This is the main entry point for running a backtest against a single configuration.

The method delegates execution-cost simulation to the configured commission and slippage models, and state tracking to the PortfolioStateManager. After the loop finishes, results are assembled by the ReportingEngine and returned as a plain dict suitable for serialization.

Returns:

Complete backtest results dictionary containing:

  • Register_df: time series DataFrame of portfolio values.

  • Total_commissions: total commissions paid (currency units).

  • lista_comisiones: list of commissions per rebalance event.

  • final_total_portfolio_value: portfolio value on the last day.

  • initial_portfolio_value: starting capital.

  • years: backtest duration in years (trading-day based).

  • Commission_df, Shares_df, orders_df: detail tables.

  • Daily_Weights: DataFrame of holdings weights per day.

  • portfolio_df: original portfolio definition.

  • benchmark: benchmark data, populated only by run_with_benchmark().

Return type:

BacktestResultDict

Raises:

  • ExecutionError – Raised by the execution layer (e.g. broker failures, invalid order states). Propagated unchanged.

  • GeneralBacktestError – Wraps unexpected data/type errors raised by PyArrow or by arithmetic on the price tables.

Examples

>>> results = backtester.run()
>>> print(f"Final value: ${results['final_total_portfolio_value']:,.2f}")
Final value: $125,000.00
>>> print(f"Total commissions: ${results['Total_commissions']:,.2f}")
Total commissions: $450.50
>>> results['Register_df'].head()
            portfolio_value  cash
2024-01-02       100000.00   0.0
2024-01-03       100150.25   0.0

Notes

Execution flow per trading day:

  1. Snapshot all prices for the day via _get_all_prices_for_date.

  2. If the day is a rebalance date, build and execute the trade plan; abort the loop if the rebalance fails.

  3. Mark-to-market the portfolio and record a daily snapshot.

Performance scales linearly with the number of trading days. The rebalance loop is the only quadratic-ish step (proportional to number of tickers times number of rebalance dates).

See also

run_with_benchmark

Same execution plus benchmark data in the result.

create_from_configuration

Recommended factory for building a backtester.

ReportingEngine

Assembles the returned BacktestResultDict.

run_with_benchmark() → BacktestResultDict#

Execute the backtest and attach benchmark data to the result.

Convenience wrapper around run() for the common case where the caller wants to compute relative performance against a benchmark (e.g. SPY). This method does not change the simulation in any way; it just copies the benchmark table that was passed in at construction time into the benchmark key of the returned dictionary.

Returns:

Same structure as run(), with the benchmark key populated when a benchmark table was provided. If no benchmark was supplied at construction time, the key is None.

Return type:

BacktestResultDict

Raises:

  • ExecutionError – Propagated from run().

  • GeneralBacktestError – Propagated from run().

Examples

>>> results = backtester.run_with_benchmark()
>>> if results['benchmark'] is not None:
...     benchmark_returns = results['benchmark']['returns']

Notes

Calling this method twice will execute the backtest twice; results are not cached. To compute multiple metrics against the same run, call this method once and reuse the returned dict.

See also

run

Underlying execution method.

property tickers: list[str]#

Get the list of ticker symbols traded in this backtest.

Returns the deduplicated union of all tickers that appeared in the rebalance schedule. The list reflects the trading universe as seen by the backtester, not the full set of tickers loaded in the market data bundle.

Returns:

Copy of the internal tickers list. Modifying the returned list does not affect the backtester’s state.

Return type:

list[str]

Examples

>>> backtester.tickers
['AAPL', 'GOOG', 'MSFT']
>>> len(backtester.tickers)
3

Notes

The order follows insertion order from the parsed portfolio definition; do not rely on it being sorted alphabetically.

See also

rebalance_dates

Companion property listing the rebalance schedule.