beacon.backtest

Portfolio simulation: BacktestEngine consumes a target weight schedule and simulates trading with configurable transaction costs, returning a BacktestResult. See Backtest for the narrative version.

backtest

The init.py for the 'backtest' module.

This module provides an engine for backtesting index methodologies and ETF tracking strategies.

TradeInstruction dataclass

TradeInstruction(
    asset_id: str,
    side: str,
    quantity: float,
    price: float,
    cost: float,
)

A single trade for a portfolio to record.

Produced by whatever decides trades — the backtest engine sizes, prices and costs an order — and consumed by :meth:Portfolio.apply, which does the accounting. It lives here rather than in the backtest layer because the portfolio is the layer that accepts one, and a ledger's input type belongs with the ledger (BN-151; previously in backtest/engine.py, where it forced the codebase's one circular-import workaround).

Attributes:

Name	Type	Description
asset_id	str	Asset identifier.
side	str	"SELL" or "BUY".
quantity	float	Number of units to trade.
price	float	Execution price per unit.
cost	float	Transaction cost in currency terms.

BacktestAssetView

BacktestAssetView(
    asset_id: str,
    data_fetcher: DataFetcher,
    portfolio: Portfolio,
    index_book: object | None = None,
)

Bases: AssetView

AssetView with backtest context for a specific asset.

Parameters:

Name	Type	Description	Default
asset_id	str	The identifier used to look up data in the DataFetcher.	required
data_fetcher	DataFetcher	The data provider instance.	required
portfolio	Portfolio	The run's books — positions, weights, transactions.	required
index_book	object | None	The tracked index's book, when the run tracked one. Target weights come from its source snapshots: what each rebalance decided, which is the comparison slippage is about.	None

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def __init__(self,
             asset_id: str,
             data_fetcher: DataFetcher,
             portfolio: Portfolio,
             index_book: "object | None" = None):
    super().__init__(asset_id, data_fetcher)
    self._portfolio = portfolio
    self._index_book = index_book

trades

trades() -> pd.DataFrame

This asset's transactions.

Returns:

Type	Description
DataFrame	pd.DataFrame: DataFrame with columns: date, type, quantity, price,
DataFrame	cost. Empty DataFrame if no trades exist for this asset.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def trades(self) -> pd.DataFrame:
    """This asset's transactions.

    Returns:
        pd.DataFrame: DataFrame with columns: date, type, quantity, price,
        cost. Empty DataFrame if no trades exist for this asset.
    """
    asset_txns = [t for t in self._portfolio.transactions
                  if t.asset_id == self._asset_id]

    if not asset_txns:
        return pd.DataFrame(columns=["date", "type", "quantity", "price",
                                     "cost"])

    rows = [
        {
            "date": t.transaction_date,
            "type": t.transaction_type,
            "quantity": t.quantity,
            "price": t.price,
            "cost": t.transaction_cost,
        }
        for t in asset_txns
    ]
    return pd.DataFrame(rows)

total_cost

total_cost() -> float

Sum of all transaction costs for this asset.

Returns:

Name	Type	Description
float	float	Total transaction costs incurred for this asset.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def total_cost(self) -> float:
    """Sum of all transaction costs for this asset.

    Returns:
        float: Total transaction costs incurred for this asset.
    """
    return sum(t.transaction_cost for t in self._portfolio.transactions
               if t.asset_id == self._asset_id)

holding_periods

holding_periods() -> list[dict[str, pd.Timestamp]]

Continuous periods when this asset was held.

Read from the positions panel — the record of quantities — rather than inferred from weights. Quantity above zero is the fact of holding; a weight of 0.0000 is a rounding statement about size.

Returns:

Type	Description
list[dict[str, Timestamp]]	list of dict: Each dict has "start" and "end" keys with
list[dict[str, Timestamp]]	Timestamps. An open position at the end of the run has "end"
list[dict[str, Timestamp]]	set to the last recorded date.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def holding_periods(self) -> list[dict[str, pd.Timestamp]]:
    """Continuous periods when this asset was held.

    Read from the positions panel — the record of quantities — rather
    than inferred from weights. Quantity above zero is the fact of
    holding; a weight of 0.0000 is a rounding statement about size.

    Returns:
        list of dict: Each dict has ``"start"`` and ``"end"`` keys with
        Timestamps. An open position at the end of the run has ``"end"``
        set to the last recorded date.
    """
    quantities = self._panel()["QUANTITY"]

    if quantities.empty:
        return []

    # Reindexed over the run's recorded calendar: the panel only carries
    # rows for dates the asset was held, so without the calendar a gap --
    # sold out, later re-bought -- would be invisible and two periods
    # would read as one.
    calendar = self._calendar()
    if not calendar.empty:
        quantities = quantities.reindex(calendar).fillna(0.0)

    held = quantities > 0
    periods = []
    in_period = False
    start = None
    prev_date = None

    for date, is_held in held.items():
        if is_held and not in_period:
            start = date
            in_period = True
        elif not is_held and in_period:
            periods.append({"start": start, "end": prev_date})
            in_period = False
        prev_date = date

    if in_period:
        periods.append({"start": start, "end": prev_date})

    return periods

weight_series

weight_series() -> pd.Series

Time series of this asset's portfolio weight.

Returns:

Type	Description
Series	pd.Series: Weight at each date where the asset was held. Dates
Series	where the asset had zero or no weight are excluded.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def weight_series(self) -> pd.Series:
    """Time series of this asset's portfolio weight.

    Returns:
        pd.Series: Weight at each date where the asset was held. Dates
        where the asset had zero or no weight are excluded.
    """
    weights = self._panel()["WEIGHT"]

    if weights.empty:
        return pd.Series(dtype=float)

    series = weights.dropna().astype(float)
    return series[series > 0]

target_weight_series

target_weight_series() -> pd.Series

Time series of this asset's target index weight.

Read from the rebalance snapshots — what each rebalance decided — rather than the index's daily panel: the target the portfolio traded to is the snapshot, and the daily drift between rebalances is the index's business, not the portfolio's instruction.

Returns:

Type	Description
Series	pd.Series: Target weight at each rebalance date. Rebalance dates
Series	where the asset was not a constituent are excluded. Empty
Series	Series if the run tracked no index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def target_weight_series(self) -> pd.Series:
    """Time series of this asset's target index weight.

    Read from the rebalance snapshots — what each rebalance decided —
    rather than the index's daily panel: the target the portfolio traded
    to is the snapshot, and the daily drift between rebalances is the
    index's business, not the portfolio's instruction.

    Returns:
        pd.Series: Target weight at each rebalance date. Rebalance dates
        where the asset was not a constituent are excluded. Empty
        Series if the run tracked no index.
    """
    snapshots = self._target_snapshots()

    if not snapshots:
        return pd.Series(dtype=float)

    data = {}
    for rebal_date in sorted(snapshots):
        weights = snapshots[rebal_date]
        if self._asset_id in weights:
            data[rebal_date] = weights[self._asset_id]
    return pd.Series(data, dtype=float)

slippage_vs_target

slippage_vs_target() -> pd.Series

Difference between actual and target weights over time.

For each date the asset was held, finds the applicable target weight (most recent rebalance on or before that date) and computes actual - target.

Returns:

Type	Description
Series	pd.Series: Slippage series indexed by date. Positive values mean
Series	the asset is overweight vs target. Empty Series if the run
Series	tracked no index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def slippage_vs_target(self) -> pd.Series:
    """Difference between actual and target weights over time.

    For each date the asset was held, finds the applicable target weight
    (most recent rebalance on or before that date) and computes
    actual - target.

    Returns:
        pd.Series: Slippage series indexed by date. Positive values mean
        the asset is overweight vs target. Empty Series if the run
        tracked no index.
    """
    snapshots = self._target_snapshots()

    if not snapshots:
        return pd.Series(dtype=float)

    actual = self.weight_series()
    if actual.empty:
        return pd.Series(dtype=float)

    sorted_rebal_dates = sorted(snapshots.keys())

    def _target_on_date(date: pd.Timestamp) -> float:
        applicable = [d for d in sorted_rebal_dates if d <= date]
        if not applicable:
            return 0.0
        latest = applicable[-1]
        return snapshots[latest].get(self._asset_id, 0.0)

    target = actual.index.to_series().apply(_target_on_date)
    target.index = actual.index
    return actual - target

weight_on_date

weight_on_date(date: Timestamp) -> float | None

This asset's portfolio weight on a specific date.

Parameters:

Name	Type	Description	Default
date	Timestamp	The query date.	required

Returns:

Type	Description
float | None	float or None: The weight, or None if the asset was not held on
float | None	that date.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/asset_view.py

def weight_on_date(self,
                   date: pd.Timestamp) -> float | None:
    """This asset's portfolio weight on a specific date.

    Args:
        date: The query date.

    Returns:
        float or None: The weight, or None if the asset was not held on
        that date.
    """
    weights = self._panel()["WEIGHT"]

    if weights.empty:
        return None

    if date not in weights.index:
        # The books were written that day and this asset has no row: it
        # was not held, and falling back to an earlier weight would
        # report a position that had already been sold.
        calendar = self._calendar()
        if date in calendar:
            return None

        # Off the run calendar (a weekend, a holiday): the position in
        # force is the last recorded one, as it always was.
        applicable = weights.index[weights.index <= date]
        if applicable.empty:
            return None
        date = applicable[-1]

    value = weights.loc[date]
    if pd.isna(value) or value == 0:
        return None
    return float(value)

BacktestEngine

BacktestEngine(
    start_date: str,
    end_date: str,
    initial_capital: float,
    data_provider: DataFetcher,
    index_result: IndexResult,
    price_column: str = "CLOSE",
    currency: str = "USD",
    transaction_cost_bps: float = 0.0,
    modifiers: list[BacktestModifier] | None = None,
    benchmark: IndexResult | Series | None = None,
    target_index: IndexResult | None = None,
    calendar: str | None = None,
)

Bases: PricingMixin

Simulates portfolio execution against a target weight schedule.

The engine consumes target weights from an IndexResult — the sole schedule source since BN-165, when the raw weight-dict mode was removed — and simulates trading over a date range using prices from a DataFetcher.

Parameters:

Name	Type	Description	Default
start_date	str	The start date of the backtest (YYYY-MM-DD).	required
end_date	str	The end date of the backtest (YYYY-MM-DD).	required
initial_capital	float	The starting capital for the backtest.	required
data_provider	DataFetcher	Data source for market prices.	required
index_result	IndexResult	The IndexResult whose weight_snapshots provide the rebalance schedule and target weights.	required
price_column	str	Column name to read from market data. Defaults to "CLOSE".	'CLOSE'
transaction_cost_bps	float	Transaction cost in basis points applied to each trade's notional value. Defaults to 0 (no cost).	0.0
modifiers	list[BacktestModifier] | None	Optional hooks that can skip rebalances or adjust trades.	None
benchmark	IndexResult | Series | None	The benchmark of record, stored on the result so every reader quotes excess return against the same comparator.	None
target_index	IndexResult | None	The calculated index the traded schedule was derived from, when it differs from the schedule itself — the derived-index shape (BN-167): index_result is an optimised calculation and this is its parent, and they land in index.optimised and index.target respectively. Omitted on a plain run, whose own calculation fills the target book.	None
calendar	str | None	The exchange MIC the traded index schedules on, which since BN-180 every definition carries. It decides which days the run steps onto at all (BN-186) and how a missing bar on one of them is read (BN-183): on a day the calendar says was closed the market was shut and the previous session's price is what the position was worth, while on a day it says was open the data is missing something and the carried price is recorded as a gap. None falls back to the data's own sessions — a day the store has bars for is treated as open — which is all a caller assembling an engine by hand can offer.	None

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/engine.py

def __init__(self,
             start_date: str,
             end_date: str,
             initial_capital: float,
             data_provider: DataFetcher,
             index_result: IndexResult,
             price_column: str = "CLOSE",
             currency: str = "USD",
             transaction_cost_bps: float = 0.0,
             modifiers: list[BacktestModifier] | None = None,
             benchmark: IndexResult | pd.Series | None = None,
             target_index: IndexResult | None = None,
             calendar: str | None = None):
    self.start_date: pd.Timestamp = pd.Timestamp(start_date)
    self.end_date: pd.Timestamp = pd.Timestamp(end_date)
    self.initial_capital: float = initial_capital
    self.data_provider: DataFetcher = data_provider
    self.index_result: IndexResult = index_result

    # The comparators of record (decision 13). The engine trades on
    # neither; it stores them so the run states what it was measured
    # against, and every reader quotes the same numbers.
    self.benchmark: IndexResult | pd.Series | None = benchmark
    self.target_index: IndexResult | None = target_index
    self.price_column: str = price_column
    self.currency: str = currency.upper()
    self.calendar: str | None = calendar

    # Listing currency per identifier, resolved lazily and once. Prices
    # are quoted where the company lists; a portfolio has one currency.
    self._currencies: dict[str, str] = {}

    # The last bar each name actually printed, so a miss is answered from
    # the session before it rather than by refetching a whole history.
    self._last_bars: dict[str, tuple[pd.Timestamp, float]] = {}

    # What the run has to report about its own pricing (BN-183). The set
    # is the dedupe: a rebalance day prices each name several times --
    # the mark, the sell test, the buy test, the re-mark -- and one
    # missing bar is one gap however many readers met it.
    self._price_gaps: list[PriceGap] = []
    self._gaps_seen: set[tuple[str, pd.Timestamp]] = set()
    self._rebalance_pricing: list[RebalancePricing] = []

    # Filled by `run`. A name past its last listed date has no price
    # because it no longer exists, which is neither a holiday nor a gap.
    self._delistings: dict[str, pd.Timestamp] = {}

    self.transaction_cost_bps: float = transaction_cost_bps
    self.modifiers: list[BacktestModifier] = modifiers or []

    # The internal schedule representation: rebalance date -> weights.
    self._weight_schedule: dict[pd.Timestamp, dict[str, float]] = (
        index_result.weight_snapshots)

run

run() -> BacktestResult

Execute the backtest and return a :class:BacktestResult.

Returns:

Type	Description
BacktestResult	BacktestResult

Raises:

Type	Description
CalculationError	If the dataset has no column to price positions from, checked before the first trade rather than discovered at it (BN-217).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/engine.py

def run(self) -> BacktestResult:
    """Execute the backtest and return a :class:`BacktestResult`.

    Returns:
        BacktestResult

    Raises:
        CalculationError: If the dataset has no column to price positions
            from, checked before the first trade rather than discovered at
            it (BN-217).
    """
    require_price_column(self.data_provider, self.price_column,
                         "The backtest")

    logger.info(
        f"Starting backtest from {self.start_date.date()} to "
        f"{self.end_date.date()} with capital {self.initial_capital:.2f}"
    )

    # What this run reports about its own pricing, cleared rather than
    # carried: a second `run()` on the same engine is a second run, and
    # inheriting the first one's gaps would report them twice.
    self._price_gaps.clear()
    self._gaps_seen.clear()
    self._rebalance_pricing.clear()

    # The traded index's own sessions (BN-186), from the calendar BN-183
    # already wired in for the price read -- one source of truth, so the
    # day the engine steps onto is the day it can price. Without a
    # calendar this is still Monday to Friday, which is all a caller
    # assembling an engine by hand has told it.
    trading_days = sessions(self.start_date, self.end_date,
                            self.calendar).union(self._scheduled_days())
    if trading_days.empty:
        logger.warning("No trading days in the specified date range.")
        portfolio = Portfolio(portfolio_id="backtest_portfolio",
                              initial_cash=self.initial_capital,
                              inception=self.start_date,
                              source=self.data_provider)
        portfolio.freeze()
        return self._build_result(portfolio, [])

    # Day zero is the EVE of the first trading day, not the start date:
    # the start date is usually itself a trading day, and history keeps
    # the last write per date -- an inception row dated the first trading
    # day would be overwritten by that day's close mark, and the record
    # of what the run started with would be gone (decision 11).
    eve = trading_days[0] - pd.tseries.offsets.BDay(1)
    portfolio = Portfolio(portfolio_id="backtest_portfolio",
                          initial_cash=self.initial_capital,
                          inception=eve,
                          source=self.data_provider)

    unfilled: list[UnfilledOrder] = []

    # Held on the engine as well as passed down: the price read consults
    # it to decline carrying a delisted name forward (BN-183), and
    # disposal reads it to settle the holding. One mapping, two readers.
    delistings = self._delisting_dates()
    self._delistings = delistings

    for idx, date in enumerate(trading_days):
        # 1. Update prices for existing holdings
        self._update_portfolio_prices(portfolio, date)

        # 1b. Settle anything that stopped being listed. This has to
        # happen before the rebalance, because a delisted holding cannot
        # be sold by the ordinary path -- that path needs a price, and
        # there is not one.
        self._dispose_delisted(portfolio, date, delistings)

        # 2. Check for rebalance
        target_w = self._get_target_weights_for_date(date)

        if target_w is not None:
            target_w = self._drop_stale(target_w, date)

        if target_w is not None:
            # Step 1 prepares the day's page from what is *held*, and on
            # the first day nothing is, so `_warm_holdings` returns without
            # building one and every opening purchase went to the full frame
            # -- 200 trips in a 200-name run (BN-216). Warming from the names
            # being traded makes the first rebalance hit the page like every
            # other.
            #
            # Not, as first claimed, a problem for names *entering* at later
            # rebalances: since BN-213 a page holds every instrument on its
            # session, not only the ones asked for, so a new entrant is
            # already on it. Measured with 50 new names a quarter for twelve
            # quarters: 100 trips before this change, all of them the
            # opening purchase. Held-plus-target is kept anyway because it
            # states what the rebalance needs, and stays correct if a page
            # is ever narrowed back to the names requested.
            self._warm_holdings(sorted(set(portfolio.holdings) | set(target_w)),
                                date)

            unfilled.extend(self._rebalance(portfolio, target_w, date))
            # Re-price after rebalance
            self._update_portfolio_prices(portfolio, date)

        # 3. End-of-day state is already in the books: the dated mark
        # in step 1 (and the re-mark after a rebalance) wrote the day's
        # position, cash and NAV rows. Nothing to flatten here.
        nav = portfolio.get_total_value()

        # Progress logging
        n = len(trading_days)
        if n > 10 and idx % (n // 10) == 0:
            logger.info(
                f"Backtest progress: {(idx + 1) / n * 100:.0f}% "
                f"({date.date()}, NAV={nav:.2f})"
            )

    logger.info(f"Backtest finished. Final NAV: {portfolio.get_total_value():.2f}")

    # The run is over, so its books are closed: the portfolio is now the
    # record of this backtest, and a later write would restate it.
    portfolio.freeze()

    return self._build_result(portfolio, unfilled)

Backtest

Backtest(
    initial_capital: float,
    transaction_cost_bps: float = 0.0,
    price_column: str = "CLOSE",
    currency: str = "USD",
    modifiers: list[BacktestModifier] | None = None,
    benchmark: IndexResult | Series | None = None,
    data_provider: DataFetcher | None = None,
    cache: IndexResultCache | None = None,
)

One-call backtests: assumptions on the object, the index per run.

The constructor mirrors :class:BacktestEngine's parameters — what stays fixed across runs — and :meth:run takes the definition and window, so a parameter sweep is one object per assumption set over one shared (cached) calculation::

bt = Backtest(initial_capital=1_000_000, transaction_cost_bps=5.0)
result = bt.run(definition, start="2023-01-03", end="2023-12-29")

Parameters:

Name	Type	Description	Default
initial_capital	float	The starting capital for each run.	required
transaction_cost_bps	float	Transaction cost in basis points applied to each trade's notional value. Defaults to 0 (no cost).	0.0
price_column	str	Market-data column both the calculator and the engine read. Defaults to "CLOSE".	'CLOSE'
currency	str	The simulated book's currency. Defaults to "USD".	'USD'
modifiers	list[BacktestModifier] | None	Optional hooks that can skip rebalances or adjust trades.	None
benchmark	IndexResult | Series | None	The benchmark of record, stored on every result this object produces.	None
data_provider	DataFetcher | None	Data source for both the calculation and the simulation. None resolves the process's ambient source (:func:beacon.sources.resolve) at each run — resolution is per run, not at construction, so beacon.use() after construction is honoured.	None
cache	IndexResultCache | None	Where calculated IndexResults are kept between runs. None uses the default location when platformdirs is available and degrades to no caching when it is not.	None

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/main.py

def __init__(self,
             initial_capital: float,
             transaction_cost_bps: float = 0.0,
             price_column: str = "CLOSE",
             currency: str = "USD",
             modifiers: list[BacktestModifier] | None = None,
             benchmark: IndexResult | pd.Series | None = None,
             data_provider: DataFetcher | None = None,
             cache: IndexResultCache | None = None):
    self.initial_capital: float = initial_capital
    self.transaction_cost_bps: float = transaction_cost_bps
    self.price_column: str = price_column
    self.currency: str = currency
    self.modifiers: list[BacktestModifier] | None = modifiers
    self.benchmark: IndexResult | pd.Series | None = benchmark
    self.data_provider: DataFetcher | None = data_provider
    self.cache: IndexResultCache | None = (cache if cache is not None
                                           else _default_cache())

    logger.info("Backtest initialised: capital %.2f, cost %.1f bps, "
                "cache %s.",
                initial_capital, transaction_cost_bps,
                "off" if self.cache is None else f"at {self.cache.root}")

run

run(
    definition: AnyIndexDefinition,
    start: str | None = None,
    end: str | None = None,
    optimised: bool = False,
    optimisation_config: OptimisationConfig | None = None,
) -> BacktestResult

Calculate (or reuse) the index, then simulate tracking it.

Parameters:

Name	Type	Description	Default
definition	AnyIndexDefinition	The index to calculate and track — a plain :class:IndexDefinition, or a stored :class:~beacon.index.derived.OptimisedIndexDefinition, which always fills both index books: its parent's calculation as index.target and its own as index.optimised.	required
start	str | None	First date (YYYY-MM-DD). Defaults to the definition's base date.	None
end	str | None	Last date (YYYY-MM-DD). Required.	None
optimised	bool	Ad-hoc optimisation of definition: build an ephemeral derived index over it from optimisation_config — same solve, same chained levels as a stored one — and trade that. Requires the config; needs scipy only on this path.	False
optimisation_config	OptimisationConfig | None	What the ad-hoc derivation is asked to do (objective, constraints, reserved risk model). Only meaningful — and only allowed — with optimised=True.	None

Returns:

Name	Type	Description
BacktestResult	BacktestResult	The engine's result — portfolio kept whole, books
	BacktestResult	filled, data bound to the run's own source.

Raises:

Type	Description
ValueError	If end is not provided, or optimised and optimisation_config contradict each other (a flag with no config, or a config with no flag).
CalculationError	If a calculation comes back empty (see :func:_rejecting_empty), the objective is unknown, or an optimisation is infeasible.
DataSourceError	If no data source is bound and the process has no ambient one.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/main.py

def run(self,
        definition: AnyIndexDefinition,
        start: str | None = None,
        end: str | None = None,
        optimised: bool = False,
        optimisation_config: OptimisationConfig | None = None) -> BacktestResult:
    """Calculate (or reuse) the index, then simulate tracking it.

    Args:
        definition: The index to calculate and track — a plain
            :class:`IndexDefinition`, or a stored
            :class:`~beacon.index.derived.OptimisedIndexDefinition`,
            which always fills both index books: its parent's calculation
            as ``index.target`` and its own as ``index.optimised``.
        start: First date (YYYY-MM-DD). Defaults to the definition's
            base date.
        end: Last date (YYYY-MM-DD). Required.
        optimised: Ad-hoc optimisation of *definition*: build an
            ephemeral derived index over it from *optimisation_config* —
            same solve, same chained levels as a stored one — and trade
            that. Requires the config; needs scipy only on this path.
        optimisation_config: What the ad-hoc derivation is asked to do
            (objective, constraints, reserved risk model). Only
            meaningful — and only allowed — with ``optimised=True``.

    Returns:
        BacktestResult: The engine's result — portfolio kept whole, books
        filled, data bound to the run's own source.

    Raises:
        ValueError: If *end* is not provided, or *optimised* and
            *optimisation_config* contradict each other (a flag with no
            config, or a config with no flag).
        CalculationError: If a calculation comes back empty (see
            :func:`_rejecting_empty`), the objective is unknown, or an
            optimisation is infeasible.
        DataSourceError: If no data source is bound and the process has
            no ambient one.
    """
    if end is None:
        raise ValueError("end must be provided.")
    if optimised and optimisation_config is None:
        raise ValueError(
            "optimised=True needs an optimisation_config saying what to "
            "solve for.")
    if not optimised and optimisation_config is not None:
        raise ValueError(
            "an optimisation_config was given but optimised is False; "
            "pass optimised=True to use it, or drop the config.")

    fetcher = (self.data_provider if self.data_provider is not None
               else sources.resolve())

    logger.info("Backtest run for '%s' from %s to %s.",
                definition.index_id, start or definition.base_date.date(), end)

    derived = self._derived_definition(definition, optimised,
                                       optimisation_config)

    if derived is None:
        index_result = self._calculated(definition, fetcher, start, end)
        target_index = None
    else:
        # The parent's calculation is cache-assisted exactly as a plain
        # run's is — and it is the same entry, so an optimised run over a
        # warm definition solves without recalculating the parent. The
        # derived calculation caches too, when the whole chain keys.
        target_index = self._calculated(derived.source, fetcher, start, end)
        index_result = self._calculated(derived, fetcher, start, end,
                                        parent_result=target_index)

    engine = BacktestEngine(
        start_date=start if start is not None else str(definition.base_date.date()),
        end_date=end,
        initial_capital=self.initial_capital,
        data_provider=fetcher,
        index_result=index_result,
        price_column=self.price_column,
        currency=self.currency,
        transaction_cost_bps=self.transaction_cost_bps,
        modifiers=self.modifiers,
        benchmark=self.benchmark,
        target_index=target_index,
        # The definition's own, resolved through a derivation to its
        # parent's: the engine needs it to tell a market holiday apart
        # from a hole in the data when a bar is missing (BN-183).
        calendar=definition.calendar)

    return engine.run()

BacktestResult dataclass

BacktestResult(
    portfolio: Portfolio,
    index: IndexBooks = IndexBooks(),
    benchmark: Book | None = None,
    unfilled: list[UnfilledOrder] = list(),
    price_gaps: list[PriceGap] = list(),
    rebalance_pricing: list[RebalancePricing] = list(),
    _data_fetcher: DataFetcher | None = None,
)

The record of one backtest run.

Parameters:

Name	Type	Description	Default
portfolio	Portfolio	The books — positions, weights, cash, NAV, transactions — kept whole and frozen by the engine on completion.	required
index	IndexBooks	The run's calculated indices, as an :class:IndexBooks container — always present, its books None when the run calculated none. index.target is the index aimed at, index.optimised the solved calculation when one exists, and index.tracked the book the engine traded toward.	IndexBooks()
benchmark	Book | None	The benchmark of record, when one was given to the engine.	None
unfilled	list[UnfilledOrder]	Buys the simulation could not execute in full. Empty for a run where every rebalance leg filled, so a non-empty list is itself the signal that the portfolio drifted off target for a reason other than price movement.	list()
price_gaps	list[PriceGap]	Days a name had no bar on a session its calendar says was open, and was therefore marked at a carried-forward price (BN-183). Empty for a run with complete data — a market holiday is not a gap, since nothing is missing on a day nothing traded.	list()
rebalance_pricing	list[RebalancePricing]	What each rebalance priced from, in date order. date and priced_from differ only where the schedule landed on a day the market was shut, so the run can state which session its trades were struck at rather than leaving it inferable.	list()

trading_nav property

trading_nav: Series

NAV over the simulated days, with the day-zero row excluded.

The portfolio's own nav opens with initial capital on the eve of the first trading day (decision 11) — the record of what the run started with. Every metric derives from this series instead, which matches the NAV the engine produced before the redesign exactly: the eve row is a starting fact, not a day the simulation traded.

total_unfilled_value property

total_unfilled_value: float

Total notional that went unfilled across the run.

with_data

with_data(data_fetcher: DataFetcher) -> BacktestResult

Bind a DataFetcher for asset-level queries. Returns self for chaining.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def with_data(self,
              data_fetcher: DataFetcher) -> 'BacktestResult':
    """Bind a DataFetcher for asset-level queries. Returns self for chaining."""
    self._data_fetcher = data_fetcher
    return self

asset

asset(asset_id: str) -> BacktestAssetView

Return a BacktestAssetView for an asset the run ever held.

Parameters:

Name	Type	Description	Default
asset_id	str	Identifier of the asset.	required

Returns:

Type	Description
BacktestAssetView	BacktestAssetView

Raises:

Type	Description
RuntimeError	If no DataFetcher has been bound via :meth:with_data.
KeyError	If the run's books never held asset_id. Membership is judged from the positions panel — the record of holdings — rather than from a weight column, so a position too small to round to a visible weight still counts as held.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def asset(self,
          asset_id: str) -> BacktestAssetView:
    """Return a BacktestAssetView for an asset the run ever held.

    Args:
        asset_id: Identifier of the asset.

    Returns:
        BacktestAssetView

    Raises:
        RuntimeError: If no DataFetcher has been bound via
            :meth:`with_data`.
        KeyError: If the run's books never held *asset_id*. Membership is
            judged from the positions panel — the record of holdings —
            rather than from a weight column, so a position too small to
            round to a visible weight still counts as held.
    """
    if self._data_fetcher is None:
        raise RuntimeError(
            "No DataFetcher bound. Call .with_data(fetcher) first."
        )

    positions = self.portfolio.positions

    if positions.empty or asset_id not in set(positions["ASSET_ID"]):
        raise KeyError(
            f"Asset '{asset_id}' does not appear in this backtest's books."
        )

    return BacktestAssetView(asset_id=asset_id,
                             data_fetcher=self._data_fetcher,
                             portfolio=self.portfolio,
                             index_book=self.index.tracked)

against

against(other: Comparable) -> RelativeMetrics

Compare this run's NAV against any comparator, after the fact.

The exploratory half of decision 13: the run-time benchmark is a fact about the run, this is a question asked later — so it computes and returns, and stores nothing. Ask against ten comparators and the result is byte-for-byte what it was.

Parameters:

Name	Type	Description	Default
other	Comparable	Another result, a book, an index result, or a bare level series.	required

Returns:

Name	Type	Description
RelativeMetrics	RelativeMetrics	Excess return, tracking error, beta and
	RelativeMetrics	correlation over the common window, as analysis.relative
	RelativeMetrics	computes them.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def against(self,
            other: Comparable) -> RelativeMetrics:
    """Compare this run's NAV against any comparator, after the fact.

    The exploratory half of decision 13: the run-time benchmark is a fact
    about the run, this is a question asked later — so it computes and
    returns, and **stores nothing**. Ask against ten comparators and the
    result is byte-for-byte what it was.

    Args:
        other: Another result, a book, an index result, or a bare level
            series.

    Returns:
        RelativeMetrics: Excess return, tracking error, beta and
        correlation over the common window, as `analysis.relative`
        computes them.
    """
    return relative_metrics(self.trading_nav, _levels_of(other))

get_returns

get_returns() -> pd.Series

Derive a return series from portfolio NAV.

Returns:

Type	Description
Series	pd.Series: Percentage returns (first entry is dropped).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def get_returns(self) -> pd.Series:
    """Derive a return series from portfolio NAV.

    Returns:
        pd.Series: Percentage returns (first entry is dropped).
    """
    nav = self.trading_nav

    if nav.empty:
        return pd.Series(dtype=float)

    return nav.pct_change().dropna()

get_tracking_error

get_tracking_error() -> float | None

Calculate annualised tracking error against the tracked index.

Tracking error is the annualised standard deviation of the difference between portfolio returns and index returns.

Returns:

Type	Description
float | None	float or None: Annualised tracking error, or None if the run
float | None	tracked no index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def get_tracking_error(self) -> float | None:
    """Calculate annualised tracking error against the tracked index.

    Tracking error is the annualised standard deviation of the
    difference between portfolio returns and index returns.

    Returns:
        float or None: Annualised tracking error, or None if the run
        tracked no index.
    """
    tracked = self.index.tracked
    if tracked is None:
        return None

    port_returns = self.get_returns()
    index_returns = tracked.returns

    # Align on common dates
    aligned = pd.DataFrame({
        "port": port_returns,
        "index": index_returns,
    }).dropna()

    if aligned.empty:
        return None

    active_returns = aligned["port"] - aligned["index"]
    return float(active_returns.std() * np.sqrt(252))

get_tracking_difference

get_tracking_difference() -> float | None

Calculate cumulative tracking difference against the tracked index.

Tracking difference is the difference between the cumulative portfolio return and the cumulative index return over the full backtest period.

Returns:

Type	Description
float | None	float or None: Tracking difference, or None if the run tracked
float | None	no index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def get_tracking_difference(self) -> float | None:
    """Calculate cumulative tracking difference against the tracked index.

    Tracking difference is the difference between the cumulative
    portfolio return and the cumulative index return over the
    full backtest period.

    Returns:
        float or None: Tracking difference, or None if the run tracked
        no index.
    """
    tracked = self.index.tracked
    if tracked is None:
        return None

    port_returns = self.get_returns()
    index_returns = tracked.returns

    if port_returns.empty or index_returns.empty:
        return None

    port_cumulative = (1 + port_returns).prod() - 1
    index_cumulative = (1 + index_returns).prod() - 1
    return float(port_cumulative - index_cumulative)

summary

summary() -> dict[str, float | None]

Calculate key performance metrics for the backtest.

Returns:

Name	Type	Description
dict	dict[str, float | None]	Dictionary containing: total_return, annualised_return,
	dict[str, float | None]	volatility, sharpe_ratio, max_drawdown, and optionally
	dict[str, float | None]	tracking_error and tracking_difference.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/result.py

def summary(self) -> dict[str, float | None]:
    """Calculate key performance metrics for the backtest.

    Returns:
        dict: Dictionary containing: total_return, annualised_return,
        volatility, sharpe_ratio, max_drawdown, and optionally
        tracking_error and tracking_difference.
    """
    returns = self.get_returns()
    n_periods = len(returns)
    nav = self.trading_nav
    initial = self.portfolio.initial_capital

    # Total return
    total_return = (0.0 if nav.empty or initial == 0
                    else float(nav.iloc[-1] / initial - 1))

    # Annualised return
    if n_periods > 0:
        years = n_periods / 252.0
        annualised_return = float((1 + total_return) ** (1 / years) - 1) if years > 0 else 0.0
    else:
        annualised_return = 0.0

    # Volatility (annualised)
    volatility = float(returns.std() * np.sqrt(252)) if n_periods > 1 else 0.0

    # Sharpe ratio (assumes risk-free rate = 0)
    sharpe_ratio = float(annualised_return / volatility) if volatility > 0 else 0.0

    # Max drawdown
    if not nav.empty:
        cumulative_max = nav.cummax()
        drawdown = (nav - cumulative_max) / cumulative_max
        max_drawdown = float(drawdown.min())
    else:
        max_drawdown = 0.0

    result: dict[str, float | None] = {
        "total_return": total_return,
        "annualised_return": annualised_return,
        "volatility": volatility,
        "sharpe_ratio": sharpe_ratio,
        "max_drawdown": max_drawdown,
    }

    # Tracking metrics (only if the run tracked an index)
    te = self.get_tracking_error()
    td = self.get_tracking_difference()
    if te is not None:
        result["tracking_error"] = te
    if td is not None:
        result["tracking_difference"] = td

    return result

PriceGap dataclass

PriceGap(
    date: Timestamp, asset_id: str, priced_from: Timestamp
)

A day a name should have traded on and had no bar (BN-183).

The engine prices from the last session on or before the date it is marking. Two different things can put it there, and only one of them is a fault: a day the index's calendar says was closed is a market that was shut, and the previous session's price is what the position was genuinely worth through it — no gap is recorded, because nothing is missing. A day the calendar says was open is data that is missing something, and the price carried forward is a stale quote.

Carrying it forward is standard practice and beats refusing: a backtest over five hundred names must not die because one of them had one bad day. Doing it silently is not — the mark is not what that day's market said, so it is published here rather than absorbed, the way unfilled publishes the legs a rebalance could not fill.

Attributes:

Name	Type	Description
date	Timestamp	The simulated day whose bar was missing.
asset_id	str	The name with no bar.
priced_from	Timestamp	The session the carried price actually came from, always earlier than date.

RebalancePricing dataclass

RebalancePricing(date: Timestamp, priced_from: Timestamp)

What one rebalance priced from (BN-183).

A rebalance scheduled on a day the market was shut still trades — it prices from the session in force through the closure — and a record that only carried the scheduled date left a reader unable to tell the two cases apart. date and priced_from are equal for the ordinary rebalance, which is what makes an unequal pair worth reading.

Attributes:

Name	Type	Description
date	Timestamp	The rebalance date from the weight schedule.
priced_from	Timestamp	The session its prices were read from.

UnfilledOrder dataclass

UnfilledOrder(
    date: Timestamp,
    asset_id: str,
    requested_quantity: float,
    filled_quantity: float,
    price: float,
    shortfall_value: float,
)

A buy the simulation could not execute in full.

Recorded on the result rather than only logged: a partially filled rebalance leaves the portfolio off its target weights, and a caller comparing tracking error against expectations needs to know that happened rather than reading it as a modelling result.

Attributes:

Name	Type	Description
date	Timestamp	The rebalance date.
asset_id	str	Asset that could not be fully bought.
requested_quantity	float	Quantity the rebalance asked for.
filled_quantity	float	Quantity actually bought; 0.0 when nothing was.
price	float	Execution price used.
shortfall_value	float	Notional value that went unfilled, at price.

BacktestModifier

Bases: ABC

Abstract base class for modifiers that alter rebalance behaviour.

A modifier can veto a rebalance entirely via :meth:should_skip_rebalance or adjust the trade list via :meth:adjust_trades.

should_skip_rebalance abstractmethod

should_skip_rebalance(
    date: Timestamp,
    portfolio: Portfolio,
    target_weights: dict[str, float],
) -> bool

Return True to skip the scheduled rebalance on date.

Parameters:

Name	Type	Description	Default
date	Timestamp	The rebalance date.	required
portfolio	Portfolio	Current portfolio state (prices already updated).	required
target_weights	dict[str, float]	Target weights for this rebalance.	required

Returns:

Type	Description
bool	bool

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/rules.py

@abstractmethod
def should_skip_rebalance(self,
                          date: pd.Timestamp,
                          portfolio: Portfolio,
                          target_weights: dict[str, float]) -> bool:
    """Return ``True`` to skip the scheduled rebalance on *date*.

    Args:
        date: The rebalance date.
        portfolio: Current portfolio state (prices already updated).
        target_weights: Target weights for this rebalance.

    Returns:
        bool
    """

adjust_trades abstractmethod

adjust_trades(
    trades: list[TradeInstruction],
    date: Timestamp,
    portfolio: Portfolio,
) -> list[TradeInstruction]

Optionally modify the trade list before execution.

Parameters:

Name	Type	Description	Default
trades	list[TradeInstruction]	The trades generated by :meth:BacktestEngine._generate_trades.	required
date	Timestamp	The rebalance date.	required
portfolio	Portfolio	Current portfolio state.	required

Returns:

Type	Description
list[TradeInstruction]	list of TradeInstruction: The (possibly modified) trade list.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/rules.py

@abstractmethod
def adjust_trades(self,
                  trades: list[TradeInstruction],
                  date: pd.Timestamp,
                  portfolio: Portfolio) -> list[TradeInstruction]:
    """Optionally modify the trade list before execution.

    Args:
        trades: The trades generated by
            :meth:`BacktestEngine._generate_trades`.
        date: The rebalance date.
        portfolio: Current portfolio state.

    Returns:
        list of TradeInstruction: The (possibly modified) trade list.
    """

DriftThresholdModifier

DriftThresholdModifier(threshold: float)

Bases: BacktestModifier

Only rebalance when max weight drift exceeds a threshold.

Parameters:

Name	Type	Description	Default
threshold	float	Maximum tolerable absolute drift between current and target weights. If every asset's drift is within this threshold the rebalance is skipped. For example, 0.05 means 5%.	required

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/rules.py

def __init__(self,
             threshold: float):
    if threshold < 0:
        raise ValueError("threshold must be non-negative.")
    self.threshold: float = threshold

adjust_trades

adjust_trades(
    trades: list[TradeInstruction],
    date: Timestamp,
    portfolio: Portfolio,
) -> list[TradeInstruction]

Pass-through — no trade adjustment.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/backtest/rules.py

def adjust_trades(self,
                  trades: list[TradeInstruction],
                  date: pd.Timestamp,
                  portfolio: Portfolio) -> list[TradeInstruction]:
    """Pass-through — no trade adjustment."""
    return trades