Execution#

Execution-layer components that drive a backtest day-to-day: portfolio state tracking, order routing through cost models, and structured output reporting.

Portfolio State Manager#

Portfolio state management module.

This module provides the PortfolioStateManager class which encapsulates portfolio state and provides controlled methods for updating positions, cash, and calculating values.

Key components: - AvailableCapital: Immutable result dataclass for capital calculations - PortfolioStateManager: Stateful service managing portfolio holdings

The manager follows the principle of encapsulation, exposing controlled methods for state mutation while protecting invariants.

class AvailableCapital(capital_to_invest: float, cash_reserve: float, total_available: float)#

Bases: object

Result of available capital calculation.

This is an immutable entity representing the breakdown of capital available for investment. Returned by the PortfolioStateManager.get_available_capital() method.

Variables:

  • capital_to_invest (float) – Amount available for investment after cash reserve deduction. This is what can actually be allocated to positions.

  • cash_reserve (float) – Amount held back as cash reserve. Calculated as total_available * cash_reserve_pct.

  • total_available (float) – Total capital before reserve allocation. On first rebalance: initial_capital. On subsequent rebalances: current_cash + portfolio_value.

Examples

>>> capital = AvailableCapital(
...     capital_to_invest=98000.0,
...     cash_reserve=2000.0,
...     total_available=100000.0,
... )
>>> capital.capital_to_invest
98000.0
>>> capital.cash_reserve
2000.0

Notes

The relationship is: capital_to_invest + cash_reserve = total_available

class PortfolioStateManager(tickers: list[str], initial_capital: float, cash_reserve_pct: float = 0.0)#

Bases: object

Manages portfolio state with controlled mutation methods.

This class encapsulates portfolio holdings, cash position, and commission tracking. It provides methods for updating state and calculating derived values like portfolio value and weights.

The manager maintains the following state: - Holdings: shares per ticker - Cash: current cash position - Commissions: cumulative commissions paid - Initialization flag: whether first rebalance has occurred

Variables:

  • holdings (dict[str, float]) – Current shares held by ticker (read-only property).

  • cash (float) – Current cash position.

  • initial_capital (float) – Starting capital amount (read-only).

  • cash_reserve_pct (float) – Percentage of total value to maintain as cash (read-only).

  • total_commissions (float) – Cumulative commissions paid (read-only property).

  • is_initialized (bool) – Whether first rebalancing has occurred (read-only property).

Examples

Creating and using a manager:

>>> manager = PortfolioStateManager(
...     tickers=['AAPL', 'MSFT'],
...     initial_capital=100000.0,
...     cash_reserve_pct=0.02,
... )
>>> manager.update_position('AAPL', delta_shares=100.0)
>>> manager.get_shares('AAPL')
100.0

Getting available capital:

>>> def get_price(ticker):
...     return {'AAPL': 150.0, 'MSFT': 300.0}[ticker]
>>> capital = manager.get_available_capital(get_price)
>>> capital.capital_to_invest
98000.0

Notes

The manager distinguishes between first rebalance (uses initial_capital) and subsequent rebalances (uses current_cash + portfolio_value). This matches the original backtester behavior.

add_commission(amount: float) → None#

Add commission to cumulative total.

Parameters:

amount (float) – Commission amount to add.

Examples

>>> manager.add_commission(5.50)
>>> manager.total_commissions
5.5
calculate_portfolio_value(price_getter: Callable[[str], float]) → float#

Calculate total value of holdings (excluding cash).

Parameters:

price_getter (Callable[[str], float]) – Function that returns price for a ticker. Should raise ValueError or KeyError if unavailable.

Returns:

Sum of (shares * price) for all positive positions.

Return type:

float

Examples

>>> def get_price(ticker):
...     return {'AAPL': 150.0, 'MSFT': 300.0}[ticker]
>>> manager.calculate_portfolio_value(get_price)
30000.0
calculate_total_value(price_getter: Callable[[str], float]) → float#

Calculate total portfolio value including cash.

Parameters:

price_getter (Callable[[str], float]) – Function that returns price for a ticker.

Returns:

Portfolio value plus cash position.

Return type:

float

Examples

>>> manager.calculate_total_value(get_price)
35000.0  # 30000 holdings + 5000 cash
calculate_weights(price_getter: Callable[[str], float]) → dict[str, float]#

Calculate current portfolio weights including cash.

Parameters:

price_getter (Callable[[str], float]) – Function that returns price for a ticker.

Returns:

Weights by ticker, including ‘CASH_RESERVE’. Weights sum to 1.0 and are rounded to 12 decimals.

Return type:

dict[str, float]

Examples

>>> weights = manager.calculate_weights(get_price)
>>> weights['AAPL']
0.75
>>> weights['CASH_RESERVE']
0.25
property cash: float#

Get current cash position.

Returns:

Cash amount in currency units.

Return type:

float

property cash_reserve_pct: float#

Get the cash reserve percentage.

Returns:

Percentage of portfolio to maintain as cash (0.0 to 1.0).

Return type:

float

get_active_positions() → dict[str, float]#

Get only positions with positive share counts.

Returns:

Dictionary of tickers with shares > 0.

Return type:

dict[str, float]

Examples

>>> manager.get_active_positions()
{'AAPL': 100.0, 'MSFT': 50.0}
get_available_capital(price_getter: Callable[[str], float]) → AvailableCapital#

Calculate available capital for investment.

Handles the distinction between first rebalancing (uses initial capital) and subsequent rebalancing (uses current portfolio value).

Parameters:

price_getter (Callable[[str], float]) – Function that returns price for a ticker.

Returns:

Capital breakdown with amounts.

Return type:

AvailableCapital

Examples

First rebalancing (not initialized):

>>> capital = manager.get_available_capital(get_price)
>>> capital.capital_to_invest
98000.0
>>> capital.cash_reserve
2000.0

Notes

This matches the original backtester logic:

if rebalance_date == self.first_date:
    total_available = self.initial_capital_float
else:
    total_available = self.previous_cash + portfolio_value
get_shares(ticker: str) → float#

Get the current share count for a ticker.

Returns the number of shares currently held in the portfolio for the requested ticker. The method is safe to call for any symbol: unknown tickers return 0.0 rather than raising, which lets callers treat missing positions and zeroed-out positions uniformly.

Parameters:

ticker (str) – Ticker symbol to query. Case-sensitive; must match the casing used when the position was created.

Returns:

Number of shares held. Returns 0.0 when the ticker has no position recorded.

Return type:

float

Examples

Querying a held ticker:

>>> manager.get_shares('AAPL')
100.0

Querying an unknown ticker:

>>> manager.get_shares('UNKNOWN')
0.0

Notes

Fractional shares are supported (the return type is float). Negative values are not normally possible during a backtest but are not validated here.

See also

update_position

Adjust a position by a delta.

set_position

Set the absolute share count for a ticker.

property holdings: dict[str, float]#

Get current holdings.

Returns:

Reference to holdings dict (mutable for internal use).

Return type:

dict[str, float]

Notes

Returns direct reference for performance. Use get_shares() for safe read access or get_active_positions() for filtered view.

property initial_capital: float#

Get the initial capital amount.

Returns:

Starting capital in currency units.

Return type:

float

property is_initialized: bool#

Check if portfolio has been initialized.

Returns:

True if first rebalancing has occurred.

Return type:

bool

mark_initialized() → None#

Mark portfolio as initialized after first rebalancing.

This changes the behavior of get_available_capital() from using initial_capital to using current_cash + portfolio_value.

Examples

>>> manager.mark_initialized()
>>> manager.is_initialized
True
set_position(ticker: str, shares: float) → None#

Set absolute position for a ticker.

Parameters:

  • ticker (str) – Ticker symbol to set.

  • shares (float) – Absolute number of shares to hold.

Examples

>>> manager.set_position('AAPL', shares=100.0)
>>> manager.get_shares('AAPL')
100.0
property total_commissions: float#

Get cumulative commissions paid.

Returns:

Total commissions in currency units.

Return type:

float

update_position(ticker: str, delta_shares: float) → None#

Adjust a position by adding or removing shares.

Applies delta_shares to the current share count for the ticker, creating the position if it did not exist. This is the method used by the trade broker after every fill: buys produce a positive delta, sells a negative one. Cash adjustments are handled separately by the broker, not by this method.

Parameters:

  • ticker (str) – Ticker symbol to update. Case-sensitive.

  • delta_shares (float) – Change in share count. Positive adds shares (e.g. a buy); negative removes shares (e.g. a sell). Fractional values are allowed.

Examples

Buy followed by partial sell:

>>> manager.update_position('AAPL', delta_shares=50.0)
>>> manager.update_position('AAPL', delta_shares=-30.0)
>>> manager.get_shares('AAPL')
20.0

First-time entry:

>>> manager.update_position('MSFT', delta_shares=10.0)
>>> manager.get_shares('MSFT')
10.0

Notes

This method does NOT touch cash, commissions, or transaction history. The trade broker is responsible for keeping cash and share counts in sync. Calling this method directly outside of the broker can desynchronise portfolio value calculations.

See also

set_position

Set an absolute share count instead of a delta.

get_shares

Inspect the current share count for a ticker.

Trade Broker#

Trade execution broker module.

This module provides the ExecutionBroker class for calculating and executing trades during portfolio rebalancing. It supports pluggable commission and slippage models while preserving the original calculation logic for shares.

The broker encapsulates all trade execution logic, keeping it separate from portfolio state management. This separation enables independent testing and potential future enhancements like multi-broker support.

IMPORTANT: The share calculation logic in this module matches the original backtester exactly. Changes to this logic would affect backtest results.

class ExecutionBroker(commission_model: CommissionModelInterface, slippage_model: SlippageModelInterface)#

Bases: object

Executes trades for portfolio rebalancing with cost models.

This broker handles the mechanics of determining required trades to move from current positions to target weights, applying commission and slippage models.

The broker is stateless - it does not maintain any position or account information. All state is passed in and results are returned.

Variables:

Examples

Creating a broker:

>>> from kaxanuk.backtest_engine.backtest.models import (
...     PerShareCommission,
...     NoSlippage,
... )
>>> broker = ExecutionBroker(
...     commission_model=PerShareCommission(rate=0.01),
...     slippage_model=NoSlippage(),
... )

Executing rebalance:

>>> result = broker.execute_rebalance(
...     date=pd.Timestamp('2024-01-15'),
...     current_holdings={'AAPL': 100, 'MSFT': 50},
...     target_weights={'AAPL': 0.6, 'MSFT': 0.4},
...     capital=100000.0,
...     price_getter=lambda t: PriceData(150.0, 150.0),
... )

Notes

The broker preserves the original calculation logic for shares: - Shares are calculated using math.floor - Commission is calculated on shares derived from VWAP price - Execution uses the execution price

# For sells
shares = math.floor(abs(diff) / execution_price)
shares_commission = math.floor(abs(diff) / vwap_price)

# For buys
shares = math.floor(diff / execution_price)
shares_commission = math.floor(diff / vwap_price)

# Liquidation uses current_shares directly
if weight == 0:
    shares = current_shares
property commission_model: CommissionModelInterface#

Get the commission model.

execute_rebalance(date: Timestamp, current_holdings: dict[str, float], target_weights: dict[str, float], capital: float, price_getter: Callable[[str], PriceData]) → RebalanceResult#

Calculate and execute the trades needed to reach target weights.

Compares the current holdings against the target weights and produces buy / sell orders such that, after execution at the provided prices, the portfolio reaches the target exposure for each ticker. Slippage is applied to the execution price by the configured slippage model and commissions are computed by the configured commission model; both are recorded on each resulting TradeOrder.

The method is pure with respect to the broker state — it does not mutate any internal cache. The caller is expected to apply the returned orders to the portfolio state manager and update cash separately.

Parameters:

  • date (pandas.Timestamp) – Date of the rebalancing operation. Used only to stamp the resulting orders.

  • current_holdings (dict[str, float]) – Current shares by ticker. Tickers absent from this mapping are treated as zero positions.

  • target_weights (dict[str, float]) – Target weights by ticker. Weights should sum to approximately 1.0 (or 1.0 minus the cash reserve). A weight of 0.0 on a currently held ticker triggers a full liquidation.

  • capital (float) – Available capital for investment after the cash reserve has been deducted. Used to translate weights into target dollar exposures.

  • price_getter (Callable[[str], PriceData]) – Function returning PriceData (execution price + VWAP) for a ticker. May raise ValueError or KeyError if data is unavailable; those errors are caught and the ticker is skipped with a warning.

Returns:

Complete result with buy_orders, sell_orders, total_commissions, total_buy_value, total_sale_value and status.

Return type:

RebalanceResult

Examples

>>> result = broker.execute_rebalance(
...     date=pd.Timestamp('2024-01-15'),
...     current_holdings={'AAPL': 100, 'MSFT': 50},
...     target_weights={'AAPL': 0.6, 'MSFT': 0.4},
...     capital=100000.0,
...     price_getter=get_prices,
... )
>>> result.status
<RebalanceStatus.SUCCESS: 'success'>
>>> len(result.buy_orders) + len(result.sell_orders)
2

Notes

Share calculation matches the original backtester logic exactly:

# For sells
shares = math.floor(abs(diff) / execution_price)
shares_commission = math.floor(abs(diff) / vwap_price)

# For buys
shares = math.floor(diff / execution_price)
shares_commission = math.floor(diff / vwap_price)

# Liquidation uses current_shares directly
if weight == 0:
    shares = current_shares

Sell orders are executed first to free up capital; buy orders then consume the available capital. Commissions are computed on every order regardless of side. Tickers whose price cannot be resolved are skipped (logged at WARNING level) but do not cause the whole rebalance to fail.

See also

CommissionModelInterface.calculate

Applied to every order.

SlippageModelInterface.apply

Applied to execution prices.

RebalanceResult

Structure of the returned object.

property slippage_model: SlippageModelInterface#

Get the slippage model.

Reporting Engine#

Backtest reporting engine module.

This module provides the ReportingEngine class for converting raw backtest records into structured DataFrames and comprehensive reports.

Key components: - BacktestReport: Immutable dataclass containing all report outputs - ReportingEngine: Stateless service for generating reports

The reporting engine transforms lists of DailyRecord and RebalanceResult objects into pandas DataFrames suitable for analysis, visualization, and export to Excel.

class BacktestReport(register_df: DataFrame, total_commissions: float, commission_list: tuple[float, ...], final_value: float, initial_value: float, years: float, start_date: date | None, end_date: date | None, commission_df: DataFrame, shares_df: DataFrame, orders_df: DataFrame, daily_weights_df: DataFrame, portfolio_df: DataFrame, benchmark: Table | None = None)#

Bases: object

Complete backtest results in structured format.

This immutable dataclass contains all outputs from a backtest execution, organized into DataFrames and summary statistics. It serves as the final deliverable from a backtest run.

Variables:

  • register_df (pd.DataFrame) – Time series of portfolio values indexed by date. Contains columns: Portfolio_Value, Total_Portfolio_Value, Returns. The Returns column is calculated as percentage change of Total_Portfolio_Value.

  • total_commissions (float) – Sum of all commissions paid during the entire backtest period. Rounded to 2 decimal places.

  • commission_list (tuple[float, ...]) – Tuple of commission amounts per rebalancing event. Each value is rounded to 2 decimal places. Immutable tuple ensures report integrity.

  • final_value (float) – Portfolio total value at end of backtest. Includes both holdings value and cash position.

  • initial_value (float) – Portfolio value at start of backtest (initial capital). Rounded to 2 decimal places.

  • years (float) – Duration of backtest in years. Calculated using trading days (typically 252 per year).

  • commission_df (pd.DataFrame) – Commission payments pivoted by ticker and date. Rows are dates, columns are tickers. Values represent commission paid for each ticker on each date.

  • shares_df (pd.DataFrame) – Share holdings pivoted by ticker and date. Rows are dates, columns are tickers. Values represent number of shares held.

  • orders_df (pd.DataFrame) – Executed orders pivoted by ticker and date. Rows are dates, columns are tickers. Positive values are buys, negative values are sells.

  • start_date (datetime.date | None) – Start date of the backtest period. None if no explicit start date was configured.

  • end_date (datetime.date | None) – End date of the backtest period. None if no explicit end date was configured.

  • daily_weights_df (pd.DataFrame) – Daily portfolio weights indexed by date. Columns are tickers plus ‘CASH_RESERVE’. Values sum to approximately 1.0 for each date.

  • portfolio_df (pd.DataFrame) – Original portfolio definition DataFrame. Contains the input portfolio weights schedule.

  • benchmark (pa.Table | None) – Benchmark time series data if provided during backtest. None if no benchmark was specified.

Examples

Accessing basic metrics:

>>> report.final_value
125000.0
>>> report.total_commissions
450.50

Converting to dictionary for legacy compatibility:

>>> results = report.to_dict()
>>> results['final_total_portfolio_value']
125000.0

Notes

This class is frozen (immutable) to ensure report integrity after generation. All DataFrames should be treated as read-only.

to_dict() → BacktestResultDict#

Convert report to dictionary format.

Provides backward compatibility with legacy code that expects dictionary output format from the backtester. Keys match the original PyArrowBacktester return format.

Returns:

Dictionary with all report components using legacy keys: - Register_df: Time series DataFrame - Total_commissions: Total commissions paid - lista_comisiones: List of commissions per rebalance - final_total_portfolio_value: Final portfolio value - initial_portfolio_value: Initial capital - years: Backtest duration - Commission_df: Commission DataFrame - Shares_df: Shares DataFrame - orders_df: Orders DataFrame - Daily_Weights: Weights DataFrame - portfolio_df: Portfolio definition - benchmark: Benchmark data

Return type:

BacktestResultDict

Examples

>>> results = report.to_dict()
>>> results['final_total_portfolio_value']
125000.0
>>> results['lista_comisiones']
[10.50, 8.25, 12.00]
type CommissionRecord = dict[str, date | Timestamp | str | float]#

Record for a single commission event: date, ticker, commission.

type OrderRecord = dict[str, date | Timestamp | str | float]#

Record for a single order event: date, ticker, shares, commission.

class ReportingEngine#

Bases: object

Generates structured reports from backtest records.

This stateless service class handles all data aggregation and DataFrame construction from raw backtest tracking records. It transforms lists of record objects into organized DataFrames suitable for analysis and visualization.

The engine is stateless - it does not maintain any internal state between calls. All data is passed in and results are returned.

Examples

Basic usage:

>>> engine = ReportingEngine()
>>> report = engine.generate_report(
...     daily_records=records,
...     rebalance_results=results,
...     initial_capital=100000.0,
...     years=2.5,
...     portfolio_df=portfolio,
... )
>>> report.final_value
125000.0

Accessing generated DataFrames:

>>> report.register_df.head()
             Portfolio_Value  Total_Portfolio_Value  Returns
date
2024-01-02         98500.0              100000.0      NaN
2024-01-03         99200.0              100700.0   0.0070

Notes

This class is intentionally stateless. All methods either operate on passed parameters or are static methods for pure transformations.

generate_report(daily_records: list[DailyRecord], rebalance_results: list[RebalanceResult], initial_capital: float, years: float, portfolio_df: DataFrame, benchmark: Table | None = None, shares_records: list[SharesRecord] | None = None, start_date: date | None = None, end_date: date | None = None) → BacktestReport#

Build a complete backtest report from raw tracking records.

Main entry point of the reporting layer. Takes the streams of daily snapshots and rebalance results produced during execution and turns them into a single BacktestReport dataclass with DataFrames suitable for downstream analysis, Excel export, or serialization to dict for the public BacktestResultDict.

The method is intentionally stateless: every call produces a fresh report, the engine instance is not modified, and inputs are never mutated.

Parameters:

  • daily_records (list[DailyRecord]) – Daily portfolio status records. Each record contains the date, portfolio value, total value, cash position and weights for that day. Must be non-empty for a meaningful report.

  • rebalance_results (list[RebalanceResult]) – One entry per rebalance event. Each contains the buy/sell orders, commissions paid and execution status.

  • initial_capital (float) – Starting portfolio capital in currency units. Used for return-based metrics.

  • years (float) – Duration of the backtest in years (trading-day based, typically trading_days / 252). Used to annualize return statistics.

  • portfolio_df (pandas.DataFrame) – Original portfolio definition. Preserved on the report as-is for reference / round-tripping.

  • benchmark (pyarrow.Table | None, optional) – Optional benchmark time series. Default is None.

  • shares_records (list[SharesRecord] | None, optional) – Share holding records captured at each rebalance event. When None, the shares_df on the report is empty. Default is None.

  • start_date (datetime.date | None, optional) – Start date of the backtest period, preserved on the report. Default is None.

  • end_date (datetime.date | None, optional) – End date of the backtest period, preserved on the report. Default is None.

Returns:

Immutable dataclass with every structured DataFrame plus the metadata fields (start/end dates, initial/final value, years, total commissions).

Return type:

BacktestReport

Examples

Typical use from inside PyArrowBacktester.run:

>>> engine = ReportingEngine()
>>> report = engine.generate_report(
...     daily_records=daily_records,
...     rebalance_results=rebalance_results,
...     initial_capital=100000.0,
...     years=2.5,
...     portfolio_df=portfolio_df,
...     benchmark=benchmark_table,
...     shares_records=shares_records,
... )
>>> report.final_value
125000.0
>>> result_dict = report.to_dict()  # for the public API

Notes

The returned BacktestReport is frozen, so callers can pass it around safely without defensive copies. To customize a report (e.g. inject extra columns), build a new report from modified inputs instead of mutating the returned one.

See also

BacktestReport

Structure of the returned report.

PyArrowBacktester.run

Primary caller of this method.