beacon.index

Index construction and calculation: IndexDefinition captures the static rules, methodology provides eligibility rules and weighting schemes, and IndexCalculator runs the day-by-day calculation. See Methodology for the narrative version.

index

The init.py for the 'index' module.

This module is core for defining index methodologies, selecting constituents, calculating weights, and computing index levels.

IndexAssetView

IndexAssetView(
    asset_id: str,
    data_fetcher: DataFetcher,
    weight_snapshots: dict[Timestamp, dict[str, float]],
    index_levels: Series,
)

Bases: AssetView

AssetView with index weight history and contribution analysis.

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
weight_snapshots	dict[Timestamp, dict[str, float]]	Mapping of rebalance date -> dict of {asset_id: weight} from the parent IndexResult.	required
index_levels	Series	Index level time series from the parent IndexResult.	required

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

def __init__(self,
             asset_id: str,
             data_fetcher: DataFetcher,
             weight_snapshots: dict[pd.Timestamp, dict[str, float]],
             index_levels: pd.Series):
    super().__init__(asset_id, data_fetcher)
    self._weight_snapshots = weight_snapshots
    self._index_levels = index_levels

weight_on_date

weight_on_date(date: Timestamp) -> float | None

Get this asset's index weight on a specific date.

Finds the most recent rebalance on or before date and returns the asset's weight. Returns None if the asset was not a constituent at that point.

Parameters:

Name	Type	Description	Default
date	Timestamp	The query date.	required

Returns:

Type	Description
float | None	float or None

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

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

    Finds the most recent rebalance on or before *date* and returns
    the asset's weight. Returns ``None`` if the asset was not a
    constituent at that point.

    Args:
        date: The query date.

    Returns:
        float or None
    """
    applicable = [d for d in self._weight_snapshots if d <= date]
    if not applicable:
        return None
    latest = max(applicable)
    return self._weight_snapshots[latest].get(self._asset_id)

weight_series

weight_series() -> pd.Series

Return a Series of this asset's weight at each rebalance date.

Returns:

Type	Description
Series	pd.Series: Indexed by rebalance date. Rebalance dates where the
Series	asset was not a constituent are excluded.

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

def weight_series(self) -> pd.Series:
    """Return a Series of this asset's weight at each rebalance date.

    Returns:
        pd.Series: Indexed by rebalance date. Rebalance dates where the
        asset was not a constituent are excluded.
    """
    data = {}
    for rebal_date in sorted(self._weight_snapshots):
        weights = self._weight_snapshots[rebal_date]
        if self._asset_id in weights:
            data[rebal_date] = weights[self._asset_id]
    return pd.Series(data, dtype=float)

contribution

contribution(
    start: str, end: str, price_column: str = "CLOSE"
) -> pd.Series

Calculate this asset's contribution to index returns.

Contribution on day t = weight_{t-1} * asset_return_t.

Parameters:

Name	Type	Description	Default
start	str	Start date (YYYY-MM-DD).	required
end	str	End date (YYYY-MM-DD).	required
price_column	str	Column name for return calculation.	'CLOSE'

Returns:

Type	Description
Series	pd.Series: Contribution series indexed by date.

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

def contribution(self,
                 start: str,
                 end: str,
                 price_column: str = "CLOSE") -> pd.Series:
    """Calculate this asset's contribution to index returns.

    Contribution on day *t* = weight_{t-1} * asset_return_t.

    Args:
        start: Start date (YYYY-MM-DD).
        end: End date (YYYY-MM-DD).
        price_column: Column name for return calculation.

    Returns:
        pd.Series: Contribution series indexed by date.
    """
    asset_returns = self.returns(start, end, price_column=price_column)
    if asset_returns.empty:
        return pd.Series(dtype=float)

    # Build a weight series aligned to the return dates
    # For each return date, look up the weight from the most recent rebalance
    weights = asset_returns.index.to_series().apply(
        self.weight_on_date
    ).shift(1)  # weight_{t-1}

    contribution = weights * asset_returns
    return contribution.dropna()

IndexCalculator

IndexCalculator(
    index_definition: IndexDefinition,
    data_provider: DataFetcher,
    price_column: str = "CLOSE",
)

Bases: MarketValuesMixin, DeletionMixin, TotalReturnMixin, CorporateActionsMixin

Stateless index calculator. Accepts an IndexDefinition and DataFetcher, and provides methods for constituent selection, weighting, index level calculation, and corporate action adjustments. All state is passed through method parameters and return values.

Initializes the IndexCalculator.

Parameters:

Name	Type	Description	Default
index_definition	IndexDefinition	The IndexDefinition object that specifies the index rules.	required
data_provider	DataFetcher	A DataFetcher instance to access market and asset data.	required
price_column	str	Market-data column read as the constituent price when computing market values. Defaults to "CLOSE".	'CLOSE'

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def __init__(self,
             index_definition: IndexDefinition,
             data_provider: DataFetcher,
             price_column: str = "CLOSE"):
    """
    Initializes the IndexCalculator.

    Args:
        index_definition: The IndexDefinition object that specifies the index rules.
        data_provider: A DataFetcher instance to access market and asset data.
        price_column: Market-data column read as the constituent price when
            computing market values. Defaults to ``"CLOSE"``.
    """
    if not index_definition:
        raise ValueError("index_definition must be provided.")
    if not data_provider:
        raise ValueError("data_provider must be provided.")

    self.definition: IndexDefinition = index_definition
    self.data: DataFetcher = data_provider
    self.price_column: str = price_column

    # What the methodology gets to know about the index it is running
    # inside. Rules and schemes have always taken this argument and it was
    # never supplied, so a market-cap weighting could not tell which
    # currency to compare its caps in and added them as though every
    # currency's unit were the same size (BN-188).
    self.context: IndexContext = IndexContext(currency=self.definition.currency)

    logger.info(f"IndexCalculator initialized for index '{self.definition.index_name}'.")

resolve_universe

resolve_universe(date: Timestamp) -> list[Asset]

Resolve the definition's universe identifiers into Asset objects.

The public entry point for universe resolution, for callers outside the calculation loop — the constituent preview, for one. Delegates to the internal implementation, so anything that stubs that also governs this.

Parameters:

Name	Type	Description	Default
date	Timestamp	Point-in-time date for the reference-data lookup.	required

Returns:

Type	Description
list[Asset]	list[Asset]: Assets for every identifier that resolved.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def resolve_universe(self,
                     date: pd.Timestamp) -> list[Asset]:
    """Resolve the definition's universe identifiers into Asset objects.

    The public entry point for universe resolution, for callers outside
    the calculation loop — the constituent preview, for one. Delegates to
    the internal implementation, so anything that stubs that also governs
    this.

    Args:
        date: Point-in-time date for the reference-data lookup.

    Returns:
        list[Asset]: Assets for every identifier that resolved.
    """
    return self._get_universe(date)

select_constituents

select_constituents(
    universe: list[Asset], current_date: Timestamp
) -> list[Asset]

Selects index constituents from a given universe based on eligibility rules.

A thin projection of :meth:select_with_provenance: the survivors, with the record of which rule removed each excluded name discarded. Callers wanting that record — the preview waterfall, anything answering "why is this name missing" — should use the fuller method rather than repeating the walk, which is what BN-102 existed to stop.

Parameters:

Name	Type	Description	Default
universe	list[Asset]	A list of potential Asset objects to consider for inclusion.	required
current_date	Timestamp	The date for which selection is being made.	required

Returns:

Type	Description
list[Asset]	A list of Asset objects that are eligible for the index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def select_constituents(self,
                        universe: list[Asset],
                        current_date: pd.Timestamp) -> list[Asset]:
    """
    Selects index constituents from a given universe based on eligibility rules.

    A thin projection of :meth:`select_with_provenance`: the survivors, with
    the record of which rule removed each excluded name discarded. Callers
    wanting that record — the preview waterfall, anything answering "why is
    this name missing" — should use the fuller method rather than repeating
    the walk, which is what BN-102 existed to stop.

    Args:
        universe: A list of potential Asset objects to consider for inclusion.
        current_date: The date for which selection is being made.

    Returns:
        A list of Asset objects that are eligible for the index.
    """
    return self.select_with_provenance(universe, current_date).survivors

select_with_provenance

select_with_provenance(
    universe: list[Asset], current_date: Timestamp
) -> SelectionResult

Select constituents, keeping the record of how the universe narrowed.

Parameters:

Name	Type	Description	Default
universe	list[Asset]	A list of potential Asset objects to consider for inclusion.	required
current_date	Timestamp	The date for which selection is being made.	required

Returns:

Name	Type	Description
SelectionResult	SelectionResult	Survivors, one step per rule, and the position of
	SelectionResult	the rule that excluded each removed asset.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def select_with_provenance(self,
                           universe: list[Asset],
                           current_date: pd.Timestamp) -> SelectionResult:
    """Select constituents, keeping the record of how the universe narrowed.

    Args:
        universe: A list of potential Asset objects to consider for inclusion.
        current_date: The date for which selection is being made.

    Returns:
        SelectionResult: Survivors, one step per rule, and the position of
        the rule that excluded each removed asset.
    """
    logger.info(
        f"[{current_date.strftime('%Y-%m-%d')}] Selecting constituents for "
        f"'{self.definition.index_name}'. Universe size: {len(universe)}")

    # BN-184, triaged as leave. No universe means no survivors, which is
    # the arithmetic rather than a stand-in for it, and the provenance
    # record says so explicitly: a universe step of zero remaining. The
    # condition is not lost downstream either — `run` refuses a base date
    # with no constituents outright, in `_require_a_base_composition`.
    if not universe:
        logger.warning("Constituent selection called with an empty universe.")

        return SelectionResult(survivors=[],
                               steps=[SelectionStep(position=UNIVERSE_POSITION,
                                                    remaining=0)])

    result = select_with_provenance(universe,
                                    self.definition.eligibility_rules,
                                    current_date,
                                    self.data,
                                    self.context)

    logger.info(
        f"Selected {len(result.survivors)} constituents for "
        f"'{self.definition.index_name}'.")

    return result

calculate_constituent_weights

calculate_constituent_weights(
    constituents: list[Asset], current_date: Timestamp
) -> dict[Asset, float]

Calculates the weights for the given constituents based on the index's weighting scheme.

Parameters:

Name	Type	Description	Default
constituents	list[Asset]	A list of Asset objects that are part of the index.	required
current_date	Timestamp	The date for which weights are calculated.	required

Returns:

Type	Description
dict[Asset, float]	A dictionary mapping each Asset to its float weight. Sum of weights should be 1.0.

Raises:

Type	Description
CalculationError	If the scheme refuses — an unpriced constituent, an unknown share count, a market cap of zero — in which case it propagates exactly as the scheme raised it, remedy and all (BN-196). Also if its weights do not sum to 1: that used to be silently renormalised with a warning, and a scheme's own output rescaled is the scheme not being applied, which is the BN-179 argument exactly (BN-184).
UnexpectedCalculationError	If the scheme raises anything else. A crash, not a decision, and it carries its own published code so a client does not read it as a refusal (BN-194).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def calculate_constituent_weights(self,
                                  constituents: list[Asset],
                                  current_date: pd.Timestamp) -> dict[Asset, float]:
    """
    Calculates the weights for the given constituents based on the index's weighting scheme.

    Args:
        constituents: A list of Asset objects that are part of the index.
        current_date: The date for which weights are calculated.

    Returns:
        A dictionary mapping each Asset to its float weight. Sum of weights should be 1.0.

    Raises:
        CalculationError: If the scheme refuses — an unpriced constituent,
            an unknown share count, a market cap of zero — in which case it
            propagates exactly as the scheme raised it, remedy and all
            (BN-196). Also if its weights do not sum to 1: that used to be
            silently renormalised with a warning, and a scheme's own output
            rescaled is the scheme not being applied, which is the BN-179
            argument exactly (BN-184).
        UnexpectedCalculationError: If the scheme raises anything else. A
            crash, not a decision, and it carries its own published code so
            a client does not read it as a refusal (BN-194).
    """
    date_str = current_date.strftime('%Y-%m-%d')
    if not constituents:
        logger.warning(
            f"[{date_str}] Calculating weights for an empty list of "
            f"constituents for '{self.definition.index_name}'.")
        return {}

    logger.info(
        f"[{date_str}] Calculating weights for {len(constituents)} "
        f"constituents of '{self.definition.index_name}'.")

    try:
        weights = self.definition.weighting_scheme.calculate_weights(
            constituents, current_date, self.data, self.context
        )

    # A scheme's own refusal, passed through as it was raised (BN-196).
    # BN-194 wrapped this on the premise that a guard in *this* file refuses
    # deliberately while a scheme only ever faults — which BN-179, BN-188
    # and BN-191 had already made false: every substitution they removed
    # became a `CalculationError` raised from inside a scheme, each naming
    # its remedy. Wrapping them relabelled the whole class as a crash, so
    # "unpriced constituent" published as "the engine broke" — the exact
    # inversion BN-194 set out to fix. What reaches the `except` below is
    # what that premise actually described: an exception the scheme never
    # meant to raise.
    except CalculationError:
        raise

    except Exception as e:
        logger.error(
            f"Error applying weighting scheme "
            f"{self.definition.weighting_scheme.scheme_name}: {e}")

        # The `WeightingScheme-` prefix survives as a *name* only — nothing
        # may branch on it, which is why the class differs.
        raise UnexpectedCalculationError(
            calculation_name=f"WeightingScheme-{self.definition.weighting_scheme.scheme_name}",
            cause=e) from e

    # A scheme that does not return weights summing to 1 has not produced
    # the weighting it names. Rescaling them here published a *different*
    # allocation under the scheme's name and said so only in a log, which
    # is the same substitution BN-179 removed from the market-cap path.
    # Raised outside the try above so it reaches the caller rather than
    # being re-wrapped as a scheme failure.
    weight_sum = sum(weights.values())

    if weights and abs(weight_sum - 1.0) > 1e-9:
        raise CalculationError(
            calculation_name=(f"WeightingScheme-"
                              f"{self.definition.weighting_scheme.scheme_name}"),
            details=(f"the {len(weights)} weights it returned for "
                     f"{date_str} sum to {weight_sum!r}, not 1. Rescaling "
                     f"them would publish an allocation the scheme did "
                     f"not produce under the scheme's own name."))

    logger.info(f"Weights calculated for '{self.definition.index_name}'.")
    return weights

cap_weights

cap_weights(
    weights: dict[Asset, float],
) -> tuple[dict[Asset, float], CapReport]

Apply the definition's cap, returning the weights and a report.

Capping happens here rather than inside a weighting scheme so that it composes with every scheme, and it returns its report rather than storing one so the calculator stays stateless and run() stays idempotent.

Parameters:

Name	Type	Description	Default
weights	dict[Asset, float]	Normalised weights keyed by Asset.	required

Returns:

Name	Type	Description
tuple	dict[Asset, float]	The capped weights and a CapReport. With no cap configured
	CapReport	the weights are returned unchanged and the report is empty.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def cap_weights(self,
                weights: dict[Asset, float]) -> tuple[dict[Asset, float], CapReport]:
    """Apply the definition's cap, returning the weights and a report.

    Capping happens here rather than inside a weighting scheme so that it
    composes with every scheme, and it returns its report rather than
    storing one so the calculator stays stateless and `run()` stays
    idempotent.

    Args:
        weights: Normalised weights keyed by Asset.

    Returns:
        tuple: The capped weights and a CapReport. With no cap configured
        the weights are returned unchanged and the report is empty.
    """
    cap = self.definition.max_constituent_weight
    if cap is None or not weights:
        return weights, CapReport(cap=cap)

    # apply_cap works on identifiers so its report can name constituents
    # without depending on the asset classes.
    by_id = {asset.asset_id: weight for asset, weight in weights.items()}
    capped, report = apply_cap(by_id, cap)

    return {asset: capped[asset.asset_id] for asset in weights}, report

initialize_divisor

initialize_divisor(
    initial_total_market_value: float,
) -> float

Calculates the initial divisor for the index on its base_date. Divisor = Initial Total Market Value / Base Index Value.

Parameters:

Name	Type	Description	Default
initial_total_market_value	float	The sum of (price * shares * fx_rate * free_float_if_applicable) for all base constituents on the base_date, expressed in index currency.	required

Returns:

Type	Description
float	The initial divisor as a float.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def initialize_divisor(self,
                       initial_total_market_value: float) -> float:
    """
    Calculates the initial divisor for the index on its base_date.
    Divisor = Initial Total Market Value / Base Index Value.

    Args:
        initial_total_market_value: The sum of (price * shares * fx_rate *
            free_float_if_applicable) for all base constituents on the
            base_date, expressed in index currency.

    Returns:
        The initial divisor as a float.
    """
    if initial_total_market_value <= 0:
        logger.error("Initial total market value must be positive to initialize divisor.")
        raise CalculationError(
            "DivisorInitialization",
            f"the base constituents are worth {initial_total_market_value} "
            f"on the base date, so there is no scale to anchor the index "
            f"to. Any divisor chosen here would produce a level series "
            f"that is coherent and means nothing.")
    if self.definition.base_value <= 0:
        logger.error("Base index value must be positive to initialize divisor.")
        raise CalculationError("DivisorInitialization", "Base index value is non-positive.")

    divisor = initial_total_market_value / self.definition.base_value
    logger.info(
        f"Divisor for '{self.definition.index_name}' initialized to: {divisor:.4f} "
        f"(Initial Market Value: {initial_total_market_value:.2f}, "
        f"Base Value: {self.definition.base_value})")
    return divisor

adjust_divisor_for_rebalance staticmethod

adjust_divisor_for_rebalance(
    old_divisor: float,
    old_market_value: float,
    new_market_value: float,
) -> float

Adjust the divisor to maintain index level continuity across a rebalance.

When index composition or weights change, the total market value shifts. To prevent an artificial jump in the index level the divisor is scaled:

new_divisor = old_divisor * (new_market_value / old_market_value)

This guarantees: level_before == level_after.

Parameters:

Name	Type	Description	Default
old_divisor	float	The divisor in effect before the rebalance.	required
old_market_value	float	Aggregate market value under the old composition.	required
new_market_value	float	Aggregate market value under the new composition.	required

Returns:

Type	Description
float	The adjusted divisor.

Raises:

Type	Description
ValueError	If old_divisor, old_market_value or new_market_value is zero or negative.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

@staticmethod
def adjust_divisor_for_rebalance(old_divisor: float,
                                 old_market_value: float,
                                 new_market_value: float) -> float:
    """Adjust the divisor to maintain index level continuity across a rebalance.

    When index composition or weights change, the total market value shifts.
    To prevent an artificial jump in the index level the divisor is scaled:

        new_divisor = old_divisor * (new_market_value / old_market_value)

    This guarantees: level_before == level_after.

    Args:
        old_divisor: The divisor in effect before the rebalance.
        old_market_value: Aggregate market value under the **old** composition.
        new_market_value: Aggregate market value under the **new** composition.

    Returns:
        The adjusted divisor.

    Raises:
        ValueError: If *old_divisor*, *old_market_value* or *new_market_value*
            is zero or negative.
    """
    if old_divisor <= 0:
        raise ValueError(f"old_divisor must be positive, got {old_divisor}")
    if old_market_value <= 0:
        raise ValueError(f"old_market_value must be positive, got {old_market_value}")
    if new_market_value <= 0:
        raise ValueError(f"new_market_value must be positive, got {new_market_value}")

    new_divisor = old_divisor * (new_market_value / old_market_value)

    logger.info(
        f"Divisor adjusted for rebalance: {old_divisor:.6f} -> {new_divisor:.6f} "
        f"(old_mv={old_market_value:.2f}, new_mv={new_market_value:.2f})"
    )
    return new_divisor

run

run(
    start_date: str | None = None,
    end_date: str | None = None,
) -> IndexResult

Run the full index calculation over a date range.

Iterates the index's own trading sessions from start_date to end_date — the definition's calendar, not Monday to Friday (BN-186) — handling three day types:

1. Base date – resolve universe, select constituents, compute weights, initialise divisor, set level = base_value. Rolled forward to the first session when the base date itself was not one.
2. Rebalance date – reconstitute (re-resolve universe, re-select, re-weight) and adjust divisor for continuity.
3. Regular day – compute index level using current constituents and weights.

The method is idempotent: it carries no state between calls.

Parameters:

Name	Type	Description	Default
start_date	str | None	First calculation date (YYYY-MM-DD). Defaults to definition.base_date.	None
end_date	str | None	Last calculation date (YYYY-MM-DD). Required.	None

Returns:

Name	Type	Description
An	IndexResult	class:IndexResult containing index levels, divisor history,
	IndexResult	constituent snapshots, weight snapshots, and the daily weights
	IndexResult	panel — one row per constituent per day, recorded as the loop
	IndexResult	goes, since the state it holds each day is path-dependent and
	IndexResult	cannot be reconstructed from the rebalance snapshots afterwards.

Raises:

Type	Description
ValueError	If end_date is not provided or precedes the base date.
CalculationError	If the dataset lacks a column the definition reads -- checked before any work, rather than discovered at the first read and reported as one company's problem (BN-217).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def run(self,
        start_date: str | None = None,
        end_date: str | None = None) -> IndexResult:
    """Run the full index calculation over a date range.

    Iterates the index's own trading sessions from *start_date* to
    *end_date* — the definition's calendar, not Monday to Friday (BN-186)
    — handling three day types:

    1. **Base date** – resolve universe, select constituents, compute
       weights, initialise divisor, set level = base_value. Rolled forward
       to the first session when the base date itself was not one.
    2. **Rebalance date** – reconstitute (re-resolve universe, re-select,
       re-weight) and adjust divisor for continuity.
    3. **Regular day** – compute index level using current constituents
       and weights.

    The method is idempotent: it carries no state between calls.

    Args:
        start_date: First calculation date (YYYY-MM-DD).  Defaults to
            ``definition.base_date``.
        end_date: Last calculation date (YYYY-MM-DD).  Required.

    Returns:
        An :class:`IndexResult` containing index levels, divisor history,
        constituent snapshots, weight snapshots, and the daily weights
        panel — one row per constituent per day, recorded as the loop
        goes, since the state it holds each day is path-dependent and
        cannot be reconstructed from the rebalance snapshots afterwards.

    Raises:
        ValueError: If *end_date* is not provided or precedes the base date.
        CalculationError: If the dataset lacks a column the definition
            reads -- checked before any work, rather than discovered at the
            first read and reported as one company's problem (BN-217).
    """
    self.require_columns()

    base_date = self.definition.base_date
    pd_start = pd.Timestamp(start_date) if start_date else base_date
    if end_date is None:
        raise ValueError("end_date must be provided.")
    pd_end = pd.Timestamp(end_date)

    if pd_end < base_date:
        raise ValueError(
            f"end_date ({pd_end.strftime('%Y-%m-%d')}) precedes "
            f"base_date ({base_date.strftime('%Y-%m-%d')})."
        )

    # Ensure start is not before base_date
    if pd_start < base_date:
        pd_start = base_date

    # What the calendar can actually speak for, asked before the sessions
    # are built (BN-198). A window the calendar cannot reach used to
    # produce an empty index *successfully*: good data, a real request, a
    # result with no levels in it and a WARNING as the only record.
    coverage = calendar_coverage(pd_start, pd_end, self.definition.calendar)

    if coverage.is_empty and self.definition.calendar is not None:
        raise CalculationError(
            calculation_name="IndexCalculator",
            details=(
                f"cannot schedule {pd_start:%Y-%m-%d} to "
                f"{pd_end:%Y-%m-%d}: {describe_bounds(coverage)}, so it "
                f"can speak for none of that window. An index cannot have "
                f"levels on days its calendar does not know about, and "
                f"publishing an empty one would report that as a result "
                f"rather than as a problem. Move the dates inside those "
                f"bounds, or name a calendar that covers the period."))

    if coverage.is_partial:
        logger.warning(
            "%s narrowed this run: %s. The index is calculated over the "
            "covered range and carries both in its result.",
            self.definition.calendar, coverage.describe_window())

    # The index's own sessions, not Monday to Friday (BN-186). A level is
    # a statement that the market traded and the constituents were worth
    # something; on 25 December neither is true, and iterating business
    # days published one anyway whenever the store happened to carry a bar.
    trading_days = sessions(pd_start, pd_end, self.definition.calendar)

    # Still reachable with a full cover: a window inside the calendar's
    # bounds that holds no session at all, a single weekend being the
    # smallest case. That is a real answer to a narrow question rather
    # than a calendar failing to reach, so it stays an empty result.
    if trading_days.empty:
        logger.warning("No trading days in the requested range.")
        return IndexResult(
            index_id=self.definition.index_id,
            index_levels=pd.Series(dtype=float),
            divisor_history=pd.Series(dtype=float),
            constituent_snapshots={},
            weight_snapshots={},
            calendar_coverage=coverage if coverage.is_partial else None,
        )

    # A base date the market was shut on is initialised on the first
    # session that follows it. Rolling forward is the only direction
    # available -- there is no index before its base date to roll back to
    # -- and refusing would make 1 January, the commonest base date there
    # is, unusable. Only when the run starts at the base date: a run
    # starting later never meets it, which is the behaviour it always had.
    if pd_start <= base_date and base_date not in trading_days:
        logger.warning(
            "Base date %s is not a session on %s; the index is based on "
            "%s, the first session after it.",
            base_date.date(), self.definition.calendar,
            trading_days[0].date())
        base_date = pd.Timestamp(trading_days[0])

    # Pre-compute rebalance dates (excluding base date which is handled separately)
    rebalance_dates_list = self.definition.get_rebalance_dates(
        pd_start.strftime('%Y-%m-%d'),
        pd_end.strftime('%Y-%m-%d'),
    )
    # Announced on one date, in force on another. With no lag the two
    # coincide and this mapping is the identity, which is what keeps every
    # index defined before BN-126 producing identical levels.
    # The panel is built only when a lag actually applies. An index
    # without one does no calendar work at all, which keeps this change
    # free for every index defined before it.
    lag = self.definition.effective_lag_sessions
    panel = (sessions(pd_start, pd_end, self.definition.calendar)
             if lag > 0 else pd.DatetimeIndex([]))
    effective_for = {announced: effective_date(announced, lag, panel)
                     for announced in rebalance_dates_list
                     if announced != base_date}

    # Keyed by the date the composition is *applied*, since that is the day
    # the loop has to act on. Two announcements landing on one effective
    # date would be a schedule shorter than its own lag; the later wins,
    # which is the one a reader would expect to be in force.
    announced_for = {effective: announced
                     for announced, effective in effective_for.items()}
    rebalance_dates = set(announced_for)
    rebalance_dates.discard(base_date)

    # Accumulators
    index_levels: dict[pd.Timestamp, float] = {}
    divisor_values: dict[pd.Timestamp, float] = {}
    constituent_snapshots: dict[pd.Timestamp, list[str]] = {}
    weight_snapshots: dict[pd.Timestamp, dict[str, float]] = {}
    # Only rebalances where the cap actually bound get an entry, so an
    # uncapped index carries an empty mapping rather than noise.
    cap_reports: dict[pd.Timestamp, CapReport] = {}
    # Effective date -> announcement date, populated only where the two
    # differ. An index with no lag carries an empty mapping, so its
    # presence is itself the signal that a lag applies.
    announcements: dict[pd.Timestamp, pd.Timestamp] = {}
    # One record per constituent per day. The loop already holds the true
    # state of the index on every date and used to discard it, keeping
    # only levels, divisors and the rebalance snapshots.
    #
    # Held as dicts until the run ends and converted once, which is the
    # pattern the backtest engine uses. The cost is at the peak rather
    # than at rest: a pending record measures ~240 bytes against ~19 in
    # the frame, so a 6,000-name decade would hold ~3.6 GB here before
    # collapsing to ~286 MB. Chunked conversion is the fix if that ever
    # binds; nothing in this repository runs at that size yet.
    daily_records: list[dict[str, object]] = []

    # Cash distributions, loaded once. A price index skips this entirely,
    # so it costs nothing and reads no action history — which is what keeps
    # every index defined before BN-125 producing identical levels.
    reinvesting = self.definition.return_type in REINVESTING
    distributions = self.cash_distribution_schedule() if reinvesting else {}

    # When each name stops being listed, resolved once. An index over a
    # universe where nothing is ever delisted gets an empty mapping and
    # pays for one `if` per day.
    delistings = self.delisting_schedule()
    previous_date = base_date
    withholding = withholding_for(self.definition.return_type,
                                  self.definition.withholding_tax_rate)

    # Running state. `units` is what the index actually holds: fixed
    # between rebalances, so weights drift with relative performance
    # instead of being reset every day.
    constituents: list[Asset] = []
    weights: dict[Asset, float] = {}
    units: dict[Asset, float] = {}
    divisor: float = 0.0
    level: float = self.definition.base_value

    for date in trading_days:
        # One slice for the day, then every per-name read below is served
        # from it (BN-212). The panel has existed since BN-190 and was
        # wired into selection and weighting, which run at rebalances --
        # roughly forty times over this loop's eight hundred. The daily
        # valuation is the one that runs every session, and it was the one
        # still reading name by name.
        self._warm_holdings(units, date)

        # Today's holdings, valued. Empty on a day the index has no
        # holdings to value, which records no weights.
        values: dict[Asset, float] = {}

        if date == base_date:
            # --- Base date initialisation ---
            constituents_raw = self._get_universe(date)
            constituents = self.select_constituents(constituents_raw, date)

            self._require_a_base_composition(constituents_raw,
                                             constituents, date)

            weights = self.calculate_constituent_weights(constituents, date)
            weights, cap_report = self.cap_weights(weights)
            if cap_report.was_capped:
                cap_reports[date] = cap_report

            # The aggregate the index represents is still the constituents'
            # total market value, which keeps the divisor's magnitude and
            # meaning unchanged. What changes is that the holdings are now
            # units derived from the weights, rather than shares
            # outstanding — so the methodology actually drives the level.
            mv_map = self._get_constituent_market_values(weights, date)
            total_mv = sum(mv_map.values())
            units = self.index_units(weights, total_mv, date)
            values = self.holding_values(units, date)

            # Straight through to `initialize_divisor`, which refuses a
            # non-positive aggregate. A zero market value used to be
            # caught here and turned into a divisor of 1.0, which
            # bypassed that refusal and published a level series scaled
            # by an arbitrary constant: internally coherent, and a
            # measure of nothing (BN-184).
            divisor = self.initialize_divisor(total_mv)

            level = self.definition.base_value

            # Record snapshots
            constituent_snapshots[date] = [a.asset_id for a in constituents]
            weight_snapshots[date] = {a.asset_id: w for a, w in weights.items()}

        elif date in rebalance_dates:
            # --- Rebalance date ---
            # Value the outgoing holdings at today's prices. This is the
            # level the new composition has to start from, which is what
            # the divisor adjustment preserves.
            old_aggregate = self.aggregate_value(units, date)

            # The outgoing holdings are the ones that went ex today, so the
            # reinvestment belongs to them and has to happen before the
            # composition changes. Adjusting the divisor here composes with
            # the continuity adjustment below: that one preserves whatever
            # level is in force, which now includes the distribution.
            if reinvesting:
                paid = distributions.get(date, {})
                divisor = self.reinvest(
                    divisor, old_aggregate,
                    self.distribution_received(
                        units, paid, withholding,
                        self.distribution_rates(
                            paid, units, date,
                            self.definition.currency)))

            # Reconstitute as of the *announcement*: the constituent list
            # and target weights are what was published, even though they
            # are implemented at today's prices. Selecting on the effective
            # date instead would let a name that qualified when the index
            # was announced be dropped by a price move in between, which is
            # not what a published composition means.
            announced_on = announced_for.get(date, date)

            constituents_raw = self._get_universe(announced_on)
            constituents = self.select_constituents(constituents_raw,
                                                    announced_on)
            weights = self.calculate_constituent_weights(constituents,
                                                         announced_on)
            weights, cap_report = self.cap_weights(weights)
            if cap_report.was_capped:
                cap_reports[date] = cap_report

            if announced_on != date:
                announcements[date] = announced_on

            # Rebuild the holdings to the new weights, scaled to the
            # constituents' total market value so the divisor keeps the
            # magnitude it has always had.
            new_mv_map = self._get_constituent_market_values(weights, date)
            new_total_mv = sum(new_mv_map.values())
            units = self.index_units(weights, new_total_mv, date)
            values = self.holding_values(units, date)
            new_aggregate = float(sum(values.values()))

            # Adjust divisor for continuity
            if old_aggregate > 0 and new_aggregate > 0:
                divisor = self.adjust_divisor_for_rebalance(
                    divisor, old_aggregate, new_aggregate
                )
            elif new_aggregate > 0:
                divisor = new_aggregate / level if level > 0 else 1.0

            # Compute level with adjusted divisor
            level = new_aggregate / divisor if divisor > 0 else level

            # Record snapshots
            constituent_snapshots[date] = [a.asset_id for a in constituents]
            weight_snapshots[date] = {a.asset_id: w for a, w in weights.items()}

        else:
            # --- Regular trading day ---
            if not constituents or divisor <= 0:
                # Before base date initialisation or no constituents
                pass
            else:
                # Before anything else: a holding that stopped being
                # listed cannot be valued, and leaving it in would report
                # its whole weight as a loss on the day it went.
                units, divisor, deleted = self.apply_deletions(
                    units, divisor, date, delistings, previous_date)

                if deleted:
                    constituents = [asset for asset in constituents
                                    if asset.asset_id not in set(deleted)]
                    weights = {asset: weight
                               for asset, weight in weights.items()
                               if asset.asset_id not in set(deleted)}

                # Valued once, then used three times over: to reinvest
                # into, to set the level, and to record the day's weights.
                # A price lookup per holding is the run's dominant cost.
                values = self.holding_values(units, date)
                aggregate = float(sum(values.values()))

                if reinvesting:
                    paid = distributions.get(date, {})
                    divisor = self.reinvest(
                        divisor,
                        aggregate,
                        self.distribution_received(
                            units, paid, withholding,
                            self.distribution_rates(
                            paid, units, date,
                            self.definition.currency)))

                level = self.level_from_units(
                    units=units,
                    divisor=divisor,
                    current_date=date,
                    previous_index_level=level,
                    values=values,
                )

        index_levels[date] = level
        divisor_values[date] = divisor
        daily_records.extend(weight_rows(date, units, values))

        # Deletions are valued on the last day the leaver still had a
        # price, so the loop has to remember which day that was.
        previous_date = date

    logger.info(
        f"run() completed for '{self.definition.index_name}': "
        f"{len(trading_days)} trading days, "
        f"{len(constituent_snapshots)} rebalance(s), "
        f"{len(daily_records)} daily weight record(s)."
    )

    return IndexResult(
        index_id=self.definition.index_id,
        index_levels=pd.Series(index_levels),
        divisor_history=pd.Series(divisor_values),
        constituent_snapshots=constituent_snapshots,
        weight_snapshots=weight_snapshots,
        cap_reports=cap_reports,
        announcement_dates=announcements,
        daily_weights=daily_weights_frame(daily_records),
        calendar_coverage=coverage if coverage.is_partial else None,
    ).with_data(self.data)

require_columns

require_columns() -> None

Refuse up front if the dataset cannot support this definition.

Public because the constituent preview runs a definition without calling run, and deserves the same answer: a preview of a market-cap index over a store with no share counts should say so, not fail on the first name it tries to price.

Raises:

Type	Description
CalculationError	Naming each missing column and what needs it.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def require_columns(self) -> None:
    """Refuse up front if the dataset cannot support this definition.

    Public because the constituent preview runs a definition without
    calling `run`, and deserves the same answer: a preview of a
    market-cap index over a store with no share counts should say so,
    not fail on the first name it tries to price.

    Raises:
        CalculationError: Naming each missing column and what needs it.
    """
    require_columns(self.definition, self.data, self.price_column)

run_daily_calculation

run_daily_calculation(
    current_date: Timestamp,
    constituents: list[Asset],
    weights: dict[Asset, float],
    previous_index_level: float,
    previous_divisor: float,
) -> tuple[float, float]

Runs a single day's index calculation process.

Parameters:

Name	Type	Description	Default
current_date	Timestamp	The date for which to perform calculations.	required
constituents	list[Asset]	Current index constituents.	required
weights	dict[Asset, float]	Current constituent weights.	required
previous_index_level	float	Index level from the previous period.	required
previous_divisor	float	Divisor from the previous period.	required

Returns:

Type	Description
tuple[float, float]	Tuple of (new_index_level, new_divisor).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/calculation/calculator.py

def run_daily_calculation(self,
                          current_date: pd.Timestamp,
                          constituents: list[Asset],
                          weights: dict[Asset, float],
                          previous_index_level: float,
                          previous_divisor: float) -> tuple[float, float]:
    """
    Runs a single day's index calculation process.

    Args:
        current_date: The date for which to perform calculations.
        constituents: Current index constituents.
        weights: Current constituent weights.
        previous_index_level: Index level from the previous period.
        previous_divisor: Divisor from the previous period.

    Returns:
        Tuple of (new_index_level, new_divisor).
    """
    divisor = previous_divisor

    if divisor is None or divisor <= 0:
        if current_date == self.definition.base_date:
            if not constituents:
                raise ValueError(
                    "Base date calculation: Constituents not provided. "
                    "Cannot initialize divisor.")
            base_day_values = self._get_constituent_market_values(
                constituents_with_weights=dict.fromkeys(constituents, 0),
                current_date=current_date
            )
            initial_mv = sum(base_day_values.values())
            if initial_mv > 0:
                divisor = self.initialize_divisor(initial_mv)
            else:
                raise ValueError(
                    f"Cannot initialize divisor on base date {current_date} due to "
                    "zero or negative market value.")
        else:
            raise ValueError("Divisor not initialized for index calculation.")

    new_level, final_divisor = self.calculate_index_level(
        current_date=current_date,
        constituents=constituents,
        weights=weights,
        divisor=divisor,
        previous_index_level=previous_index_level,
    )

    return new_level, final_divisor

IndexDefinition

IndexDefinition(
    index_id: str,
    index_name: str,
    base_date: str,
    base_value: float,
    currency: str,
    eligibility_rules: list[EligibilityRuleBase],
    weighting_scheme: WeightingSchemeBase,
    rebalancing_frequency: str,
    calendar: str,
    description: str | None = None,
    universe_identifiers: list[str] | None = None,
    max_constituent_weight: float | None = None,
    rebalance_day_rule: str = DEFAULT_DAY_RULE,
    return_type: str = PRICE,
    withholding_tax_rate: float = 0.0,
    effective_lag_sessions: int = 0,
)

Defines the static characteristics and rules for constructing a financial index.

Initializes an IndexDefinition.

Parameters:

Name	Type	Description	Default
index_id	str	A unique identifier for the index.	required
index_name	str	The common name of the index.	required
base_date	str	The date from which the index calculation begins (YYYY-MM-DD).	required
base_value	float	The initial value of the index on its base_date.	required
currency	str	The currency of the index.	required
eligibility_rules	list[EligibilityRuleBase]	A list of EligibilityRuleBase objects that define criteria for constituent selection.	required
weighting_scheme	WeightingSchemeBase	A WeightingSchemeBase object that defines how constituents are weighted.	required
rebalancing_frequency	str	A string indicating how often the index is rebalanced (e.g., 'QUARTERLY', 'MONTHLY', 'SEMI-ANNUAL', 'ANNUAL'). More complex schedules (e.g. "Third Friday of March, June...") would require a more sophisticated scheduler.	required
calendar	str	Exchange MIC backing trading-day arithmetic, e.g. "XNYS". Required, and deliberately without a default. It had one briefly while BN-180 was being written, and the default was wrong for the same reason the old null was: IndexDefinition(currency="EUR") would silently schedule a European index on New York's holidays, coherently, with nothing on screen to say so. Choosing a calendar for an index that already exists is repair, which is what DEFAULT_CALENDAR and the schema-2 migration are for; choosing one for an index being created is a guess, and the caller is the only one who can make it.	required
description	str | None	Optional textual description of the index.	None
universe_identifiers	list[str] | None	Optional list of string identifiers (e.g., tickers, ISINs) defining the asset universe from which constituents are selected.	None
max_constituent_weight	float | None	Optional cap on any single constituent's weight, as a fraction (0.1 is 10%). Applied after the weighting scheme and iterated until no constituent breaches it. None means uncapped.	None
rebalance_day_rule	str	Which day of a scheduled month the rebalance falls on. Defaults to the first business day, which is what every index defined before BN-121 used.	DEFAULT_DAY_RULE
return_type	str	PRICE, TOTAL_RETURN or NET_TOTAL_RETURN. PRICE is the default and the behaviour of every index defined before BN-125; the other two reinvest cash distributions across the index.	PRICE
withholding_tax_rate	float	Fraction of each distribution withheld, for a net index. Ignored unless the return type is NET_TOTAL_RETURN, so a definition carrying a rate it does not use cannot quietly apply it.	0.0
effective_lag_sessions	int	Sessions between a composition being announced and its weights taking effect. Zero is same-day, which is what every index did before BN-126.	0

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/constructor.py

def __init__(self,
             index_id: str,
             index_name: str,
             base_date: str, # YYYY-MM-DD
             base_value: float,
             currency: str,
             eligibility_rules: list[EligibilityRuleBase],
             weighting_scheme: WeightingSchemeBase,
             rebalancing_frequency: str, # e.g., 'QUARTERLY', 'MONTHLY', 'SEMI-ANNUAL', 'ANNUAL'
             calendar: str,
             description: str | None = None,
             universe_identifiers: list[str] | None = None,
             max_constituent_weight: float | None = None,
             rebalance_day_rule: str = DEFAULT_DAY_RULE,
             return_type: str = PRICE,
             withholding_tax_rate: float = 0.0,
             effective_lag_sessions: int = 0):
    """
    Initializes an IndexDefinition.

    Args:
        index_id: A unique identifier for the index.
        index_name: The common name of the index.
        base_date: The date from which the index calculation begins (YYYY-MM-DD).
        base_value: The initial value of the index on its base_date.
        currency: The currency of the index.
        eligibility_rules: A list of EligibilityRuleBase objects that define
                           criteria for constituent selection.
        weighting_scheme: A WeightingSchemeBase object that defines how
                          constituents are weighted.
        rebalancing_frequency: A string indicating how often the index is rebalanced
                               (e.g., 'QUARTERLY', 'MONTHLY', 'SEMI-ANNUAL', 'ANNUAL').
                               More complex schedules (e.g. "Third Friday of March, June...")
                               would require a more sophisticated scheduler.
        calendar: Exchange MIC backing trading-day arithmetic, e.g.
                              ``"XNYS"``. **Required, and deliberately
                              without a default.** It had one briefly while
                              BN-180 was being written, and the default was
                              wrong for the same reason the old null was:
                              `IndexDefinition(currency="EUR")` would
                              silently schedule a European index on New
                              York's holidays, coherently, with nothing on
                              screen to say so. Choosing a calendar for an
                              index that already exists is repair, which is
                              what `DEFAULT_CALENDAR` and the schema-2
                              migration are for; choosing one for an index
                              being created is a guess, and the caller is
                              the only one who can make it.
        description: Optional textual description of the index.
        universe_identifiers: Optional list of string identifiers (e.g., tickers, ISINs)
                              defining the asset universe from which constituents are selected.
        max_constituent_weight: Optional cap on any single constituent's
                              weight, as a fraction (0.1 is 10%). Applied
                              after the weighting scheme and iterated until
                              no constituent breaches it. None means
                              uncapped.
        rebalance_day_rule: Which day of a scheduled month the rebalance
                              falls on. Defaults to the first business day,
                              which is what every index defined before
                              BN-121 used.
        return_type: PRICE, TOTAL_RETURN or NET_TOTAL_RETURN. PRICE is the
                              default and the behaviour of every index
                              defined before BN-125; the other two reinvest
                              cash distributions across the index.
        withholding_tax_rate: Fraction of each distribution withheld, for a
                              net index. Ignored unless the return type is
                              NET_TOTAL_RETURN, so a definition carrying a
                              rate it does not use cannot quietly apply it.
        effective_lag_sessions: Sessions between a composition being
                              announced and its weights taking effect. Zero
                              is same-day, which is what every index did
                              before BN-126.
    """
    if not index_id:
        raise ValueError("index_id cannot be empty.")
    if not index_name:
        raise ValueError("index_name cannot be empty.")
    if not base_date:
        raise ValueError("base_date cannot be empty.")
    if not calendar:
        raise ValueError(
            "calendar cannot be empty; every index schedules against a "
            f"trading calendar (e.g. '{DEFAULT_CALENDAR}').")
    if rebalance_day_rule not in DAY_RULES:
        raise ValueError(
            f"Unsupported day rule: '{rebalance_day_rule}'. "
            f"Supported: {', '.join(DAY_RULES)}.")
    if return_type not in RETURN_TYPES:
        raise ValueError(
            f"Unsupported return type: '{return_type}'. "
            f"Supported: {', '.join(RETURN_TYPES)}.")
    if not 0.0 <= withholding_tax_rate < 1.0:
        raise ValueError(
            "withholding_tax_rate must be in [0, 1); got "
            f"{withholding_tax_rate}.")
    if effective_lag_sessions < 0:
        raise ValueError(
            "effective_lag_sessions cannot be negative; got "
            f"{effective_lag_sessions}.")
    if base_value <= 0:
        raise ValueError("base_value must be positive.")
    if not currency:
        raise ValueError("currency cannot be empty.")
    if not weighting_scheme:
        raise ValueError("weighting_scheme must be provided.")
    if not rebalancing_frequency:
        raise ValueError("rebalancing_frequency cannot be empty.")

    if not eligibility_rules:
        logger.warning(f"Index '{index_name}' defined with no eligibility rules.")

    if universe_identifiers is not None and not universe_identifiers:
        raise ValueError("universe_identifiers, when provided, must be a non-empty list.")

    if max_constituent_weight is not None and not 0.0 < max_constituent_weight <= 1.0:
        raise ValueError(
            "max_constituent_weight, when provided, must be in (0, 1]; got "
            f"{max_constituent_weight}.")

    self.index_id: str = index_id
    self.index_name: str = index_name
    self.base_date: pd.Timestamp = pd.Timestamp(base_date)
    self.base_value: float = base_value
    self.currency: str = currency.upper()
    self.eligibility_rules: list[EligibilityRuleBase] = eligibility_rules
    self.weighting_scheme: WeightingSchemeBase = weighting_scheme
    self.rebalancing_frequency: str = rebalancing_frequency.upper()
    self.description: str | None = description
    self.universe_identifiers: list[str] | None = universe_identifiers
    self.max_constituent_weight: float | None = max_constituent_weight
    self.rebalance_day_rule: str = rebalance_day_rule
    self.calendar: str = calendar
    self.return_type: str = return_type
    self.withholding_tax_rate: float = withholding_tax_rate
    self.effective_lag_sessions: int = effective_lag_sessions

    logger.info(
        f"IndexDefinition for '{self.index_name}' ({self.index_id}) created successfully.")

get_rebalance_dates

get_rebalance_dates(
    start_date: str, end_date: str
) -> list[pd.Timestamp]

Return all rebalance dates within [start_date, end_date] based on the index's rebalancing frequency, day rule and calendar.

Delegates to beacon.index.schedule, which replaced the first-business- day-of-month assumption this method used to hard-code. Since BN-180 the calendar is always a real one, so a date this returns is always a date the exchange has a session for — an index that named none used to schedule 1 January and 25 December.

Parameters:

Name	Type	Description	Default
start_date	str	Start of the range (YYYY-MM-DD), inclusive.	required
end_date	str	End of the range (YYYY-MM-DD), inclusive.	required

Returns:

Type	Description
list[Timestamp]	A chronologically sorted list of business-day-adjusted rebalance dates.

Raises:

Type	Description
ValueError	If the rebalancing frequency is unsupported.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/constructor.py

def get_rebalance_dates(self,
                        start_date: str,
                        end_date: str) -> list[pd.Timestamp]:
    """
    Return all rebalance dates within [start_date, end_date] based on
    the index's rebalancing frequency, day rule and calendar.

    Delegates to `beacon.index.schedule`, which replaced the first-business-
    day-of-month assumption this method used to hard-code. Since BN-180 the
    calendar is always a real one, so a date this returns is always a date
    the exchange has a session for — an index that named none used to
    schedule 1 January and 25 December.

    Args:
        start_date: Start of the range (YYYY-MM-DD), inclusive.
        end_date: End of the range (YYYY-MM-DD), inclusive.

    Returns:
        A chronologically sorted list of business-day-adjusted rebalance dates.

    Raises:
        ValueError: If the rebalancing frequency is unsupported.
    """
    return rebalance_dates(self.rebalancing_frequency,
                           start_date,
                           end_date,
                           self.calendar,
                           self.rebalance_day_rule)

next_rebalance

next_rebalance(as_of: str) -> pd.Timestamp | None

The first rebalance strictly after a date.

Anchored on the base date, like every other date this class produces, so the answer names a day the index would genuinely rebalance on.

Parameters:

Name	Type	Description	Default
as_of	str	The date being asked from, YYYY-MM-DD.	required

Returns:

Type	Description
Timestamp | None	The date, or None if none falls within the lookahead window.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/constructor.py

def next_rebalance(self,
                   as_of: str) -> pd.Timestamp | None:
    """The first rebalance strictly after a date.

    Anchored on the base date, like every other date this class produces,
    so the answer names a day the index would genuinely rebalance on.

    Args:
        as_of: The date being asked from, YYYY-MM-DD.

    Returns:
        The date, or None if none falls within the lookahead window.
    """
    return next_rebalance(self.rebalancing_frequency,
                          self.base_date,
                          as_of,
                          self.calendar,
                          self.rebalance_day_rule)

OptimisedIndexDefinition

OptimisedIndexDefinition(
    index_id: str,
    index_name: str,
    source: AnyIndexDefinition,
    objective: str = MIN_TRACKING_ERROR,
    constraints: Sequence[Constraint] = (),
    base_date: str | None = None,
    base_value: float | None = None,
    currency: str | None = None,
    description: str | None = None,
    risk_model: RiskModel | None = None,
)

An optimised index: a derivation on a source index, plus identity.

The source stays first-class — referenced, never copied — so editing the parent changes its optimised children at their next calculation, which is what "optimise the index I built" means. Chained optimisation (a source that is itself optimised) falls out of the recursion for free.

Parameters:

Name	Type	Description	Default
index_id	str	A unique identifier for the derived index.	required
index_name	str	The common name of the derived index.	required
source	AnyIndexDefinition	The parent — a plain :class:IndexDefinition, or another :class:OptimisedIndexDefinition for a chain.	required
objective	str	What to minimise. Only "min_tracking_error" exists today; an unknown value fails the calculation loudly, naming the accepted ones.	MIN_TRACKING_ERROR
constraints	Sequence[Constraint]	What the solved weights must satisfy, as :class:~beacon.optimise.constraints.Constraint instances. Empty means the solver's default of full investment alone.	()
base_date	str | None	First calculation date (YYYY-MM-DD). None inherits the source's, which is the usual case: the child lives on the parent's calendar.	None
base_value	float | None	The level the chained path starts at. None inherits the source's.	None
currency	str | None	The derived index's currency. None inherits the source's.	None
description	str | None	Optional textual description.	None
risk_model	RiskModel | None	RESERVED — carried but unused, mirroring :class:~beacon.optimise.config.OptimisationConfig. Setting one makes the calculation uncacheable (it cannot be keyed yet) and changes no result.	None

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/derived.py

def __init__(self,
             index_id: str,
             index_name: str,
             source: AnyIndexDefinition,
             objective: str = MIN_TRACKING_ERROR,
             constraints: Sequence[Constraint] = (),
             base_date: str | None = None,
             base_value: float | None = None,
             currency: str | None = None,
             description: str | None = None,
             risk_model: RiskModel | None = None):
    if not index_id:
        raise ValueError("index_id cannot be empty.")
    if not index_name:
        raise ValueError("index_name cannot be empty.")
    if source is None:
        raise ValueError("source must be provided.")
    if base_value is not None and base_value <= 0:
        raise ValueError("base_value, when provided, must be positive.")

    self.index_id: str = index_id
    self.index_name: str = index_name
    self.source: AnyIndexDefinition = source
    self.objective: str = objective
    self.constraints: tuple[Constraint, ...] = tuple(constraints)
    self.description: str | None = description
    self.risk_model: RiskModel | None = risk_model

    self._base_date: pd.Timestamp | None = (pd.Timestamp(base_date)
                                            if base_date is not None else None)
    self._base_value: float | None = base_value
    self._currency: str | None = currency.upper() if currency is not None else None

    logger.info("OptimisedIndexDefinition '%s' created over source '%s' "
                "with %d constraint(s).",
                index_id, source.index_id, len(self.constraints))

base_date property

base_date: Timestamp

The first calculation date: own when given, else the source's.

base_value property

base_value: float

The starting level: own when given, else the source's.

currency property

currency: str

The index currency: own when given, else the source's.

calendar property

calendar: str

The trading calendar, which is always the source's.

No override, unlike the currency or the base date: the derivation reallocates on exactly the parent's rebalance dates, so a calendar of its own could only disagree with the days it actually has weights for.

universe_identifiers property

universe_identifiers: list[str] | None

The investable universe, which is always the source's.

The derivation holds no universe of its own — it reallocates over exactly the names the parent published — so the answer resolves through the chain to the root definition's.

from_config classmethod

from_config(
    index_id: str,
    index_name: str,
    source: AnyIndexDefinition,
    config: OptimisationConfig,
) -> OptimisedIndexDefinition

The derivation an :class:OptimisationConfig describes.

One vocabulary for ad-hoc and stored runs (owner decision): the config is the stored derivation minus the source, so an ad-hoc Backtest.run builds an ephemeral definition through here and calculates it exactly as a stored one would be.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/derived.py

@classmethod
def from_config(cls,
                index_id: str,
                index_name: str,
                source: AnyIndexDefinition,
                config: OptimisationConfig) -> "OptimisedIndexDefinition":
    """The derivation an :class:`OptimisationConfig` describes.

    One vocabulary for ad-hoc and stored runs (owner decision): the config
    is the stored derivation minus the source, so an ad-hoc `Backtest.run`
    builds an ephemeral definition through here and calculates it exactly
    as a stored one would be.
    """
    return cls(index_id=index_id,
               index_name=index_name,
               source=source,
               objective=config.objective,
               constraints=config.constraints,
               risk_model=config.risk_model)

ExpressionRule

ExpressionRule(
    expression: dict[str, Any],
    on_missing: str = EXCLUDE,
    max_age_days: int | None = MAX_AGE_DAYS,
)

Bases: EligibilityRuleBase

Select instruments that satisfy an expression.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/expression_rules.py

def __init__(self,
             expression: dict[str, Any],
             on_missing: str = EXCLUDE,
             max_age_days: int | None = MAX_AGE_DAYS):
    super().__init__(rule_name="ExpressionRule")

    if on_missing not in ON_MISSING:
        raise InvalidRuleError(
            f"ExpressionRule on_missing '{on_missing}'",
            f"expected one of {', '.join(ON_MISSING)}")

    # Rebuilt eagerly rather than at the first rebalance. A malformed tree
    # is a fact about the rule, and finding out at construction is the
    # difference between a rejected save and a run that dies partway
    # through with thousands of names already priced.
    try:
        self._tree = from_dict(expression)
    except ExpressionError as error:
        raise InvalidRuleError("ExpressionRule expression",
                               str(error)) from error

    self.expression = expression
    self.on_missing = on_missing
    self.max_age_days = max_age_days

tree property

tree: Expression

The rebuilt expression.

required_columns

required_columns() -> frozenset[str]

The market columns the expression reads, derived from its tree.

An expression's needs are whatever it references, so they come from fields_in rather than being written out -- and a derived field is expanded into what it is computed from, because a screen on market_cap needs CLOSE and SHARES_OUTSTANDING, not a column called MARKET_CAP that no store has (BN-217). Reference and feature fields read other tables and add nothing here.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/expression_rules.py

def required_columns(self) -> frozenset[str]:
    """The market columns the expression reads, derived from its tree.

    An expression's needs are whatever it references, so they come from
    `fields_in` rather than being written out -- and a derived field is
    expanded into what it is computed from, because a screen on
    `market_cap` needs CLOSE and SHARES_OUTSTANDING, not a column called
    MARKET_CAP that no store has (BN-217). Reference and feature fields
    read other tables and add nothing here.
    """
    return market_columns_for(fields_in(self._tree))

from_expression classmethod

from_expression(
    expression: Expression,
    on_missing: str = EXCLUDE,
    max_age_days: int | None = MAX_AGE_DAYS,
) -> ExpressionRule

Build from a live expression rather than from its serialised form.

What a user writing Python calls. The stored params are identical either way, which is the point: one representation, two front doors.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/expression_rules.py

@classmethod
def from_expression(cls,
                    expression: Expression,
                    on_missing: str = EXCLUDE,
                    max_age_days: int | None = MAX_AGE_DAYS
                    ) -> "ExpressionRule":
    """Build from a live expression rather than from its serialised form.

    What a user writing Python calls. The stored `params` are identical
    either way, which is the point: one representation, two front doors.
    """
    return cls(expression.to_dict(), on_missing, max_age_days)

is_eligible

is_eligible(
    asset: Asset,
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> bool

Whether the asset passes, as of current_date.

The date is the rebalance date and is passed straight through to the point-in-time reads. A value published after it is invisible.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/expression_rules.py

def is_eligible(self,
                asset: Asset,
                current_date: pd.Timestamp,
                market_data_provider: DataFetcher,
                context: IndexContext | None = None) -> bool:
    """Whether the asset passes, as of `current_date`.

    The date is the rebalance date and is passed straight through to the
    point-in-time reads. A value published after it is invisible.
    """
    return resolve(self._tree, asset.asset_id, current_date,
                   market_data_provider,
                   on_missing=self.on_missing == INCLUDE,
                   max_age_days=self.max_age_days)

FeatureRule

FeatureRule(
    field: str,
    comparison: str = "gt",
    threshold: float = 0.0,
    feature_type: str | None = None,
    on_missing: str = EXCLUDE,
    max_age_days: int | None = MAX_AGE_DAYS,
)

Bases: EligibilityRuleBase

Select instruments whose feature value passes a threshold.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/feature_rules.py

def __init__(self,
             field: str,
             comparison: str = "gt",
             threshold: float = 0.0,
             feature_type: str | None = None,
             on_missing: str = EXCLUDE,
             max_age_days: int | None = MAX_AGE_DAYS):
    super().__init__(rule_name="FeatureRule")

    if comparison not in COMPARISONS:
        raise InvalidRuleError(
            f"FeatureRule comparison '{comparison}'",
            f"expected one of {', '.join(sorted(COMPARISONS))}")

    if on_missing not in ON_MISSING:
        raise InvalidRuleError(
            f"FeatureRule on_missing '{on_missing}'",
            f"expected one of {', '.join(ON_MISSING)}")

    if not field:
        raise InvalidRuleError("FeatureRule field",
                               "a rule must name the datapoint it screens on")

    self.field = field
    self.comparison = comparison
    self.threshold = threshold
    self.feature_type = feature_type
    self.on_missing = on_missing
    self.max_age_days = max_age_days

required_columns

required_columns() -> frozenset[str]

No market columns: a feature is read from the features table.

Stated rather than inherited, so the absence is visibly a decision. Whether the named feature exists is a real question with the same shape as a missing column, but it is asked of a different table, and this check deliberately covers market data only (BN-217).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/feature_rules.py

def required_columns(self) -> frozenset[str]:
    """No market columns: a feature is read from the features table.

    Stated rather than inherited, so the absence is visibly a decision.
    Whether the named feature exists is a real question with the same
    shape as a missing column, but it is asked of a different table, and
    this check deliberately covers market data only (BN-217).
    """
    return frozenset()

is_eligible

is_eligible(
    asset: Asset,
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> bool

Whether the asset passes, as of current_date.

The date is the rebalance date, and it is passed straight through to the point-in-time accessor. A value published after it is invisible.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/feature_rules.py

def is_eligible(self,
                asset: Asset,
                current_date: pd.Timestamp,
                market_data_provider: DataFetcher,
                context: IndexContext | None = None) -> bool:
    """Whether the asset passes, as of `current_date`.

    The date is the rebalance date, and it is passed straight through to
    the point-in-time accessor. A value published after it is invisible.
    """
    value = market_data_provider.fetch_feature(
        asset.asset_id, self.field, current_date,
        self.feature_type, self.max_age_days)

    if value is None:
        logger.debug("FeatureRule: %s has no %s knowable on %s; %sd.",
                     asset.asset_id, self.field,
                     current_date.strftime("%Y-%m-%d"), self.on_missing)

        return self.on_missing == INCLUDE

    return bool(COMPARISONS[self.comparison](value, self.threshold))

EligibilityRuleBase

EligibilityRuleBase(rule_name: str)

Bases: ABC

Abstract base class for an eligibility rule. Eligibility rules determine if an asset can be part of an index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def __init__(self,
             rule_name: str):
    self.rule_name = rule_name

required_columns

required_columns() -> frozenset[str]

The market-data columns this rule reads, declared up front (BN-217).

Checked against the dataset before a run does any work, so a store with no SHARES_OUTSTANDING column is refused on day zero as "this rule needs SHARES_OUTSTANDING and the dataset has none" -- rather than on the first rebalance as "N0 has no SHARES_OUTSTANDING on 2024-01-02", which is true, and sends a reader to inspect one company whose data is fine.

Empty by default rather than abstract, so a rule written outside this package keeps working. It is then simply not checked up front, and fails where it always did -- at the first read -- with the message it always had. Every rule shipped here declares its own.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def required_columns(self) -> frozenset[str]:
    """The market-data columns this rule reads, declared up front (BN-217).

    Checked against the dataset before a run does any work, so a store
    with no SHARES_OUTSTANDING column is refused on day zero as "this rule
    needs SHARES_OUTSTANDING and the dataset has none" -- rather than on
    the first rebalance as "N0 has no SHARES_OUTSTANDING on 2024-01-02",
    which is true, and sends a reader to inspect one company whose data is
    fine.

    Empty by default rather than abstract, so a rule written outside this
    package keeps working. It is then simply not checked up front, and
    fails where it always did -- at the first read -- with the message it
    always had. Every rule shipped here declares its own.
    """
    return frozenset()

prepare

prepare(
    candidates: list[Asset],
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> None

Read in one go whatever this rule is about to read per name.

Called once with the whole candidate set before is_eligible is asked about any of them. It decides nothing and returns nothing: a rule that did no preparation must give exactly the answers it gives now, because this is a hint about how to read rather than about what is eligible. Doing nothing is therefore the right default, and it is the base implementation (BN-190).

It exists because the per-name shape is what made a universe expensive. is_eligible is a predicate over one asset, so a rule reading market data reads it a name at a time, and each read slices a frame whose size is the whole store — the cost of one lookup growing with the universe around it rather than with the row it wants.

Parameters:

Name	Type	Description	Default
candidates	list[Asset]	Everything that reached this rung, in order. A rule that ranks rather than screens would want this set too; that is not what this is for, but it is the same set.	required
current_date	Timestamp	The date selection is being made at.	required
market_data_provider	DataFetcher	The data source the reads will go to.	required
context	IndexContext | None	What the index settles for its rules, as for :meth:is_eligible.	None

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def prepare(self,  # noqa: B027 — an optional hook, not part of the interface
            candidates: list[Asset],
            current_date: pd.Timestamp,
            market_data_provider: DataFetcher,
            context: IndexContext | None = None) -> None:
    """Read in one go whatever this rule is about to read per name.

    Called once with the whole candidate set before `is_eligible` is asked
    about any of them. It decides nothing and returns nothing: a rule that
    did no preparation must give exactly the answers it gives now, because
    this is a hint about *how* to read rather than about *what* is
    eligible. Doing nothing is therefore the right default, and it is the
    base implementation (BN-190).

    It exists because the per-name shape is what made a universe expensive.
    `is_eligible` is a predicate over one asset, so a rule reading market
    data reads it a name at a time, and each read slices a frame whose size
    is the whole store — the cost of one lookup growing with the universe
    around it rather than with the row it wants.

    Args:
        candidates: Everything that reached this rung, in order. A rule
            that ranks rather than screens would want this set too; that is
            not what this is for, but it is the same set.
        current_date: The date selection is being made at.
        market_data_provider: The data source the reads will go to.
        context: What the index settles for its rules, as for
            :meth:`is_eligible`.
    """

is_eligible abstractmethod

is_eligible(
    asset: Asset,
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> bool

Checks if a given asset is eligible based on this rule.

Parameters:

Name	Type	Description	Default
asset	Asset	The asset to check.	required
current_date	Timestamp	The date on which eligibility is being assessed.	required
market_data_provider	DataFetcher	A DataFetcher instance to get necessary market data (e.g., market cap, trading volume).	required
context	IndexContext | None	What the index the rule is running inside reports in and settles. None when the rule is evaluated outside an index, in which case nothing here may assume a currency it was not told.	None

Returns:

Type	Description
bool	True if the asset is eligible, False otherwise.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

@abstractmethod
def is_eligible(self,
                asset: Asset,
                current_date: pd.Timestamp,
                market_data_provider: DataFetcher,
                context: IndexContext | None = None) -> bool:
    """
    Checks if a given asset is eligible based on this rule.

    Args:
        asset: The asset to check.
        current_date: The date on which eligibility is being assessed.
        market_data_provider: A DataFetcher instance to get necessary market data
                              (e.g., market cap, trading volume).
        context: What the index the rule is running inside reports in and
            settles. None when the rule is evaluated outside an index, in
            which case nothing here may assume a currency it was not told.

    Returns:
        True if the asset is eligible, False otherwise.
    """

EqualWeighted

EqualWeighted()

Bases: WeightingSchemeBase

Equal weighting scheme.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def __init__(self) -> None:
    super().__init__(scheme_name="EqualWeighted")

required_columns

required_columns() -> frozenset[str]

Nothing: equal weights are decided without reading the market.

Stated rather than inherited, so the absence is a decision a reader can see. The index still needs a price column to value its holdings daily, but that is the calculator's requirement, not this scheme's.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def required_columns(self) -> frozenset[str]:
    """Nothing: equal weights are decided without reading the market.

    Stated rather than inherited, so the absence is a decision a reader
    can see. The index still needs a price column to value its holdings
    daily, but that is the calculator's requirement, not this scheme's.
    """
    return frozenset()

LiquidityRule

LiquidityRule(
    min_avg_daily_volume: int | None = None,
    min_avg_daily_value: float | None = None,
    lookback_days: int = 60,
)

Bases: EligibilityRuleBase

Eligibility rule based on trading liquidity (e.g., average daily volume or value).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def __init__(self,
             min_avg_daily_volume: int | None = None,
             min_avg_daily_value: float | None = None,
             lookback_days: int = 60):
    super().__init__(rule_name="LiquidityRule")
    self.min_avg_daily_volume = min_avg_daily_volume
    self.min_avg_daily_value = min_avg_daily_value
    self.lookback_days = lookback_days

    if lookback_days <= 0:
        raise ValueError("lookback_days must be positive.")

required_columns

required_columns() -> frozenset[str]

Volume for either threshold, and the close too for a value one.

Declared from the thresholds actually set, because they read different things. And declared at all because of what a missing VOLUME column used to do here: is_eligible treats it as "not liquid enough" and excludes the name, so a store without the column excluded every name, and the run failed as "index holds nothing on its base date" with no mention of volume anywhere. A reader loosened the threshold.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def required_columns(self) -> frozenset[str]:
    """Volume for either threshold, and the close too for a value one.

    Declared from the thresholds actually set, because they read different
    things. And declared at all because of what a missing VOLUME column
    used to do here: `is_eligible` treats it as "not liquid enough" and
    excludes the name, so a store without the column excluded **every**
    name, and the run failed as "index holds nothing on its base date"
    with no mention of volume anywhere. A reader loosened the threshold.
    """
    needed: set[str] = set()

    if self.min_avg_daily_volume is not None:
        needed.add(_VOLUME_COLUMN)

    if self.min_avg_daily_value is not None:
        needed.update({_PRICE_COLUMN, _VOLUME_COLUMN})

    return frozenset(needed)

is_eligible

is_eligible(
    asset: Asset,
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> bool

Whether asset's traded volume and value over the lookback qualify.

No session resolution here, and none needed: this reads a window ending at current_date, so a closed day is already spanned by the days around it rather than being the single day everything hangs on.

Errors are not caught (BN-182). A rule that throws has not said the asset is ineligible, and the two answers must not be spelled the same.

Raises:

Type	Description
CalculationError	If asset is not an equity, so there is no ticker to read volume against (BN-185).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def is_eligible(self,
                asset: Asset,
                current_date: pd.Timestamp,
                market_data_provider: DataFetcher,
                context: IndexContext | None = None) -> bool:
    """Whether *asset*'s traded volume and value over the lookback qualify.

    No session resolution here, and none needed: this reads a *window*
    ending at *current_date*, so a closed day is already spanned by the
    days around it rather than being the single day everything hangs on.

    Errors are not caught (BN-182). A rule that throws has not said the
    asset is ineligible, and the two answers must not be spelled the same.

    Raises:
        CalculationError: If *asset* is not an equity, so there is no
            ticker to read volume against (BN-185).
    """
    equity = require_equity(asset, self.rule_name,
                            "be assessed for liquidity")

    # Fetch more to ensure enough trading days
    start_lookback = (
        current_date - pd.Timedelta(days=self.lookback_days * 2)).strftime('%Y-%m-%d')
    end_lookback = current_date.strftime('%Y-%m-%d')

    price_df = market_data_provider.fetch_market_data(
        equity.ticker, start_lookback, end_lookback)

    if price_df.empty or price_df.shape[0] < (self.lookback_days / 2): # Ensure some data
        logger.warning(
            f"LiquidityRule: Insufficient historical price data for "
            f"{equity.ticker} for period ending {end_lookback}.")
        return False

    # Ensure we have data up to current_date or shortly before
    # (single-identifier market data is indexed by date).
    price_df = price_df[price_df.index <= current_date].tail(self.lookback_days)

    # Heuristic: need at least 80% of lookback days
    if price_df.shape[0] < (self.lookback_days * 0.8):
        logger.warning(
            f"LiquidityRule: Not enough trading days "
            f"({price_df.shape[0]}/{self.lookback_days}) for {equity.ticker} "
            f"for ADV calc.")
        return False

    if self.min_avg_daily_volume is not None:
        if (_VOLUME_COLUMN not in price_df.columns
                or price_df[_VOLUME_COLUMN].isnull().all()):
            logger.warning(f"LiquidityRule: Volume data missing for {equity.ticker}.")
            return False
        avg_daily_volume = price_df[_VOLUME_COLUMN].mean()
        if avg_daily_volume < self.min_avg_daily_volume:
            logger.debug(
                f"LiquidityRule: {equity.ticker} (ADV: {avg_daily_volume:.0f}) below "
                f"min volume {self.min_avg_daily_volume:.0f}")
            return False

    if self.min_avg_daily_value is not None:
        if (_PRICE_COLUMN not in price_df.columns
                or _VOLUME_COLUMN not in price_df.columns
                or price_df[_PRICE_COLUMN].isnull().all()
                or price_df[_VOLUME_COLUMN].isnull().all()):
            logger.warning(
                f"LiquidityRule: Price or Volume data missing for ADTV "
                f"calculation for {equity.ticker}.")
            return False
        avg_daily_value = (price_df[_PRICE_COLUMN] * price_df[_VOLUME_COLUMN]).mean()
        if avg_daily_value < self.min_avg_daily_value:
            logger.debug(
                f"LiquidityRule: {equity.ticker} (ADTV: {avg_daily_value:.2f}) below "
                f"min value {self.min_avg_daily_value:.2f}")
            return False

    logger.debug(f"LiquidityRule: {equity.ticker} is eligible.")

    return True

MarketCapRule

MarketCapRule(
    min_market_cap: float | None = None,
    max_market_cap: float | None = None,
)

Bases: EligibilityRuleBase

Eligibility by market capitalisation, read from a resolved session.

Dates resolve backwards into the data (BN-182). A weekend, a holiday or any date inside the data's coverage that carries no bar is read at the last session on or before it, because that is the universe the index actually held through the closure. Reading the exact date instead excluded every name on a closed day: the universe emptied, the weighting was handed nothing, and an index of no constituents computed a coherent level of zero.

That is the same resolution :class:MarketCapWeighted performs, through the same primitive and by design. Selection running on one calendar and weighting on another is two methodologies under one heading.

The bounds are in the index's currency, and now the arithmetic is too (BN-188). The published help text has always said so while the code compared a name's local number against the bound, so a 5bn floor admitted a name whose yen cap read 5.2bn and excluded a genuinely larger one quoted in a strong currency. The cap is converted at the session's rate before it meets either bound; a missing pair refuses rather than falling back to the local figure.

Outside an index there is no context and so no currency to convert into, and the bounds are then read in the asset's own. That is the only honest answer to "over five billion of what?" when nobody has said — and it is not a fallback inside an index, where the calculator always supplies one.

Past the last bar it refuses rather than excluding. A rule that cannot evaluate has not found the asset ineligible, it has failed, and the two must not share an answer — "not in the index" is a published fact about a name, while "the data does not reach that date" is a fact about the store.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def __init__(self,
             min_market_cap: float | None = None,
             max_market_cap: float | None = None):
    super().__init__(rule_name="MarketCapRule")
    self.min_market_cap = min_market_cap
    self.max_market_cap = max_market_cap

    if (min_market_cap is not None and max_market_cap is not None
            and min_market_cap > max_market_cap):
        raise ValueError("min_market_cap cannot be greater than max_market_cap.")

required_columns

required_columns() -> frozenset[str]

A cap is price times shares, so both, whichever bound is set.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def required_columns(self) -> frozenset[str]:
    """A cap is price times shares, so both, whichever bound is set."""
    return frozenset({_PRICE_COLUMN, _SHARES_COLUMN})

prepare

prepare(
    candidates: list[Asset],
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> None

Read the whole candidate set's session in one slice (BN-190).

Every name this rule is about to be asked about is read on the same session, for the same two columns. Warming that session turns the per-name reads below into dictionary lookups, and — because the weighting scheme then reads the survivors on the same session — makes the second pricing of every surviving name free.

Raises:

Type	Description
CalculationError	If current_date lies outside the data's coverage. That is the same refusal is_eligible makes over the same date, arriving one call earlier; a rule that cannot resolve its session cannot assess anything.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def prepare(self,
            candidates: list[Asset],
            current_date: pd.Timestamp,
            market_data_provider: DataFetcher,
            context: IndexContext | None = None) -> None:
    """Read the whole candidate set's session in one slice (BN-190).

    Every name this rule is about to be asked about is read on the same
    session, for the same two columns. Warming that session turns the
    per-name reads below into dictionary lookups, and — because the
    weighting scheme then reads the survivors on the same session — makes
    the second pricing of every surviving name free.

    Raises:
        CalculationError: If *current_date* lies outside the data's
            coverage. That is the same refusal `is_eligible` makes over the
            same date, arriving one call earlier; a rule that cannot
            resolve its session cannot assess anything.
    """
    if not candidates:
        return

    session = _resolve_session(self.rule_name, "assess eligibility",
                               current_date, market_data_provider)

    market_data_provider.warm_session(_equity_tickers(candidates), session)

is_eligible

is_eligible(
    asset: Asset,
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> bool

Whether asset's market cap at the resolved session clears the bounds.

Raises:

Type	Description
CalculationError	If asset is not an equity, if current_date lies outside the data's coverage, or if the cap cannot be converted into the index currency, so the rule cannot be evaluated at all. Nothing here turns a failure into an exclusion — see the class docstring.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def is_eligible(self,
                asset: Asset,
                current_date: pd.Timestamp,
                market_data_provider: DataFetcher,
                context: IndexContext | None = None) -> bool:
    """Whether *asset*'s market cap at the resolved session clears the bounds.

    Raises:
        CalculationError: If *asset* is not an equity, if *current_date*
            lies outside the data's coverage, or if the cap cannot be
            converted into the index currency, so the rule cannot be
            evaluated at all. Nothing here turns a failure into an
            exclusion — see the class docstring.
    """
    equity = require_equity(asset, self.rule_name,
                            "be assessed for market-cap eligibility")

    session = _resolve_session(self.rule_name, "assess eligibility",
                               current_date, market_data_provider)
    date_str = session.strftime('%Y-%m-%d')

    current_price = market_data_provider.fetch_price(equity.ticker, date_str,
                                                     _PRICE_COLUMN)

    if current_price is None:
        logger.warning(
            f"MarketCapRule: Could not fetch price for {equity.ticker} "
            f"on {date_str}.")
        return False

    # Read on the same session as the price, so the cap is one coherent
    # observation rather than a current share count against an older close.
    shares_outstanding = market_data_provider.fetch_shares_outstanding(
        equity.ticker, date_str)

    if shares_outstanding is None or shares_outstanding <= 0:
        logger.warning(
            f"MarketCapRule: Could not fetch valid shares outstanding for "
            f"{equity.ticker} on {date_str}.")
        return False

    market_cap = current_price * shares_outstanding

    # Into the index's money before it meets a bound stated in that money.
    if context is not None:
        market_cap *= _rate_into(self.rule_name, equity, context.currency,
                                 session, market_data_provider)

    if self.min_market_cap is not None and market_cap < self.min_market_cap:
        logger.debug(
            f"MarketCapRule: {equity.ticker} (MCap: {market_cap:.2f}) below "
            f"min_market_cap {self.min_market_cap:.2f}")
        return False

    if self.max_market_cap is not None and market_cap > self.max_market_cap:
        logger.debug(
            f"MarketCapRule: {equity.ticker} (MCap: {market_cap:.2f}) above "
            f"max_market_cap {self.max_market_cap:.2f}")
        return False

    logger.debug(f"MarketCapRule: {equity.ticker} (MCap: {market_cap:.2f}) is eligible.")

    return True

MarketCapWeighted

MarketCapWeighted(use_free_float: bool = False)

Bases: WeightingSchemeBase

Market capitalization weighting, optionally free-float adjusted.

Every path either weights by real market caps or refuses (BN-179). There is no equal-weight fallback: an index that comes out equal-weighted because the caps could not be read is not a degraded market-cap index, it is a different index published under the same heading, and nothing downstream looks wrong enough for anyone to ask — the levels are right, the weights sum, the backtest tracks.

Dates resolve backwards into the data. A request for a weekend, a holiday, or any date inside the data's coverage that carries no bar reads the last session on or before it, because that is the composition the index actually held through the closure rather than an approximation of one. Past the last bar it refuses, since there the same read would be a stale print presented as the current one. The bound is the data's own coverage, not a day count, which cannot tell those two apart.

Caps are compared in one currency (BN-188). This weighted price x shares in whatever money the name traded in, so a yen name entered the sum as though a thousand billion yen were a thousand billion dollars — a fifteen-fold error on its own weight, and a wrong weight on every other constituent with it. A universe spanning currencies is converted into the index's before the caps are summed, and a missing pair refuses; see :meth:_target_currency for why a universe quoted in one currency needs no conversion at all.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def __init__(self,
             use_free_float: bool = False):
    super().__init__(scheme_name="MarketCapWeighted")
    self.use_free_float = use_free_float

required_columns

required_columns() -> frozenset[str]

Price and shares, and free float only when this scheme uses it.

use_free_float is the declaration: every free-float read in a run is behind it, so a scheme that is not float-adjusted never asks the store for the column and must not be refused for lacking it.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def required_columns(self) -> frozenset[str]:
    """Price and shares, and free float only when this scheme uses it.

    `use_free_float` is the declaration: every free-float read in a run is
    behind it, so a scheme that is not float-adjusted never asks the store
    for the column and must not be refused for lacking it.
    """
    needed = {_PRICE_COLUMN, _SHARES_COLUMN}

    if self.use_free_float:
        needed.add(_FREE_FLOAT_COLUMN)

    return frozenset(needed)

calculate_weights

calculate_weights(
    constituents: list[Asset],
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> dict[Asset, float]

Weights proportional to market cap, or a refusal.

Raises:

Type	Description
CalculationError	If current_date lies outside the data's coverage, if any constituent is unpriceable, unconvertible or is not an equity, or if the caps sum to nothing. Nothing here falls back to another methodology — see the class docstring.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def calculate_weights(self,
                      constituents: list[Asset],
                      current_date: pd.Timestamp,
                      market_data_provider: DataFetcher,
                      context: IndexContext | None = None) -> dict[Asset, float]:
    """Weights proportional to market cap, or a refusal.

    Raises:
        CalculationError: If *current_date* lies outside the data's
            coverage, if any constituent is unpriceable, unconvertible or
            is not an equity, or if the caps sum to nothing. Nothing here
            falls back to another methodology — see the class docstring.
    """
    if not constituents:
        return {}

    session = self._session_for(current_date, market_data_provider)
    to_currency = self._target_currency(constituents, context)

    # The same session a selection rule has just read, over a subset of the
    # names it read it for, so this keeps that panel rather than building
    # another — which is what stops every surviving name being priced twice
    # in one rebalance (BN-190). Where no rule read anything, it is still
    # one slice for the whole constituent list rather than one per name.
    market_data_provider.warm_session(_equity_tickers(constituents), session)

    market_caps: dict[Asset, float] = {}

    for asset in constituents:
        equity = require_equity(asset, self.scheme_name,
                                "be weighted by market cap, having none")

        market_caps[asset] = self._asset_market_cap(
            equity, session, market_data_provider, to_currency)

    total_market_cap = sum(market_caps.values())

    if total_market_cap <= 0:
        raise CalculationError(
            calculation_name=self.scheme_name,
            details=(f"the {len(market_caps)} constituents priced at "
                     f"{session:%Y-%m-%d} have a total market cap of "
                     f"{total_market_cap}, so there is nothing to weight "
                     f"by."))

    return {asset: cap / total_market_cap
            for asset, cap in market_caps.items()}

WeightingSchemeBase

WeightingSchemeBase(scheme_name: str)

Bases: ABC

Abstract base class for a weighting scheme. Weighting schemes determine the proportion of each constituent in an index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def __init__(self,
             scheme_name: str):
    self.scheme_name = scheme_name

required_columns

required_columns() -> frozenset[str]

The market-data columns this scheme reads, declared up front.

The same contract as :meth:EligibilityRuleBase.required_columns, and derived from the scheme's own inputs where they change what it reads: a scheme's parameters are the declaration, so nothing asks a store for a column the configured scheme does not use (BN-217).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

def required_columns(self) -> frozenset[str]:
    """The market-data columns this scheme reads, declared up front.

    The same contract as :meth:`EligibilityRuleBase.required_columns`, and
    derived from the scheme's own inputs where they change what it reads:
    a scheme's parameters *are* the declaration, so nothing asks a store
    for a column the configured scheme does not use (BN-217).
    """
    return frozenset()

calculate_weights abstractmethod

calculate_weights(
    constituents: list[Asset],
    current_date: Timestamp,
    market_data_provider: DataFetcher,
    context: IndexContext | None = None,
) -> dict[Asset, float]

Calculates the weight for each constituent asset.

Parameters:

Name	Type	Description	Default
constituents	list[Asset]	A list of assets that are eligible for the index.	required
current_date	Timestamp	The date for which weights are being calculated.	required
market_data_provider	DataFetcher	A DataFetcher instance.	required
context	IndexContext | None	What the index the scheme is running inside reports in and settles. None when it is invoked outside an index.	None

Returns:

Type	Description
dict[Asset, float]	A dictionary mapping each Asset object to its calculated weight (float).
dict[Asset, float]	The sum of weights should typically be 1.0.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/methodology.py

@abstractmethod
def calculate_weights(self,
                      constituents: list[Asset],
                      current_date: pd.Timestamp,
                      market_data_provider: DataFetcher,
                      context: IndexContext | None = None) -> dict[Asset, float]:
    """
    Calculates the weight for each constituent asset.

    Args:
        constituents: A list of assets that are eligible for the index.
        current_date: The date for which weights are being calculated.
        market_data_provider: A DataFetcher instance.
        context: What the index the scheme is running inside reports in
            and settles. None when it is invoked outside an index.

    Returns:
        A dictionary mapping each Asset object to its calculated weight (float).
        The sum of weights should typically be 1.0.
    """

IndexResult dataclass

IndexResult(
    index_id: str,
    index_levels: Series,
    divisor_history: Series,
    constituent_snapshots: dict[Timestamp, list[str]],
    weight_snapshots: dict[Timestamp, dict[str, float]],
    cap_reports: dict[Timestamp, CapReport] = dict(),
    announcement_dates: dict[Timestamp, Timestamp] = dict(),
    daily_weights: DataFrame = empty_daily_weights(),
    calendar_coverage: CalendarCoverage | None = None,
    _data_fetcher: DataFetcher | None = None,
)

Container holding the output of an index calculation run.

Parameters:

Name	Type	Description	Default
index_id	str	Identifier of the calculated index.	required
index_levels	Series	Time series of index levels indexed by pd.DatetimeIndex.	required
divisor_history	Series	Time series of divisor values indexed by pd.DatetimeIndex.	required
constituent_snapshots	dict[Timestamp, list[str]]	Mapping of rebalance date -> list of asset_id strings.	required
weight_snapshots	dict[Timestamp, dict[str, float]]	Mapping of rebalance date -> dict of {asset_id: weight}.	required
cap_reports	dict[Timestamp, CapReport]	Mapping of rebalance date -> CapReport, for the rebalances where a weight cap actually bound. Empty for an uncapped index, so its presence is itself the signal that capping occurred.	dict()
announcement_dates	dict[Timestamp, Timestamp]	Mapping of effective date -> the date that composition was announced. Snapshots are keyed by the effective date, because that is when the weights are in force and what every consumer — drift, attribution, the backtest engine — needs. The announcement is carried alongside rather than instead, since a client showing "rebalance of 18 Sep, effective 22 Sep" needs both. Empty for an index with no lag, where the two always coincide.	dict()
daily_weights	DataFrame	Long-form panel of what the index held on every calculation day: DATE, IDENTIFIER, AMOUNT (units held) and WEIGHT (that holding's share of the day's aggregate value). Recorded by the calculator as it walks, not derived afterwards — see the note below. Defaults to an empty frame, so a result built by hand or by an older caller is still valid.	empty_daily_weights()

The daily panel is recorded rather than re-derived because the index's daily state is path-dependent. It is not a forward-fill of the rebalance snapshot, and not even "amounts fixed between rebalances, repriced daily": :class:~beacon.index.calculation.deletions.DeletionMixin drops a delisted name mid-period and adjusts the divisor, and :class:~beacon.index.calculation.corporate_actions.CorporateActionsMixin adjusts it on ex-dates. Both change what is held and what each name weighs on a day that is not a rebalance. A path is written down as it happens.

The rebalance snapshots stay what they always were: the record of what a rebalance decided. This panel is the record of what then happened.

capped_assets_on_date

capped_assets_on_date(date: Timestamp) -> dict[str, float]

Return the constituents held at the cap at the given rebalance.

Parameters:

Name	Type	Description	Default
date	Timestamp	A rebalance date.	required

Returns:

Name	Type	Description
dict	dict[str, float]	{asset_id: uncapped_weight} for names the cap bound on
	dict[str, float]	that date. Empty when nothing was capped, or when date is not a
	dict[str, float]	rebalance date.

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

def capped_assets_on_date(self,
                          date: pd.Timestamp) -> dict[str, float]:
    """Return the constituents held at the cap at the given rebalance.

    Args:
        date: A rebalance date.

    Returns:
        dict: ``{asset_id: uncapped_weight}`` for names the cap bound on
        that date. Empty when nothing was capped, or when *date* is not a
        rebalance date.
    """
    report = self.cap_reports.get(date)

    return dict(report.capped) if report is not None else {}

with_data

with_data(data_fetcher: DataFetcher) -> IndexResult

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

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

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

asset

asset(asset_id: str) -> IndexAssetView

Return an IndexAssetView for a constituent.

Parameters:

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

Returns:

Type	Description
IndexAssetView	IndexAssetView

Raises:

Type	Description
RuntimeError	If no DataFetcher has been bound via :meth:with_data.
KeyError	If asset_id is not found in any constituent snapshot.

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

def asset(self,
          asset_id: str) -> IndexAssetView:
    """Return an IndexAssetView for a constituent.

    Args:
        asset_id: Identifier of the constituent asset.

    Returns:
        IndexAssetView

    Raises:
        RuntimeError: If no DataFetcher has been bound via
            :meth:`with_data`.
        KeyError: If *asset_id* is not found in any constituent
            snapshot.
    """
    if self._data_fetcher is None:
        raise RuntimeError(
            "No DataFetcher bound. Call .with_data(fetcher) first."
        )

    all_constituents = set()
    for ids in self.constituent_snapshots.values():
        all_constituents.update(ids)

    if asset_id not in all_constituents:
        raise KeyError(
            f"Asset '{asset_id}' not found in any constituent snapshot."
        )

    return IndexAssetView(
        asset_id=asset_id,
        data_fetcher=self._data_fetcher,
        weight_snapshots=self.weight_snapshots,
        index_levels=self.index_levels,
    )

get_returns

get_returns() -> pd.Series

Derive a return series from index levels.

Returns:

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

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

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

    Returns:
        pd.Series: Percentage returns (first entry is dropped).
    """
    if self.index_levels.empty:
        return pd.Series(dtype=float)
    return self.index_levels.pct_change().dropna()

get_weights_on_date

get_weights_on_date(date: Timestamp) -> dict[str, float]

Get constituent weights effective on a given date.

Locates the most recent rebalance date on or before date.

Parameters:

Name	Type	Description	Default
date	Timestamp	The query date.	required

Returns:

Name	Type	Description
dict	dict[str, float]	Mapping of asset_id to weight. Empty dict if no rebalance
	dict[str, float]	has occurred on or before date.

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

def get_weights_on_date(self,
                        date: pd.Timestamp) -> dict[str, float]:
    """Get constituent weights effective on a given date.

    Locates the most recent rebalance date on or before *date*.

    Args:
        date: The query date.

    Returns:
        dict: Mapping of asset_id to weight. Empty dict if no rebalance
        has occurred on or before *date*.
    """
    applicable_dates = [d for d in self.weight_snapshots if d <= date]
    if not applicable_dates:
        return {}
    latest = max(applicable_dates)
    return self.weight_snapshots[latest]

weights_on

weights_on(date: Timestamp) -> dict[str, float]

Get the recorded constituent weights as of a given date.

Reads the daily panel rather than the rebalance snapshots, so the answer includes everything that happened since the last rebalance: price drift, a deletion, a divisor adjustment. Compare :meth:get_weights_on_date, which answers the different question of what the last rebalance decided.

Falls back to the latest recorded date on or before date — which covers a day the holdings could not be valued at all, since such a day records no rows.

Parameters:

Name	Type	Description	Default
date	Timestamp	The query date.	required

Returns:

Name	Type	Description
dict	dict[str, float]	Mapping of identifier to weight. Empty when nothing was
	dict[str, float]	recorded on or before date, including when no panel was captured
	dict[str, float]	at all.

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

def weights_on(self,
               date: pd.Timestamp) -> dict[str, float]:
    """Get the *recorded* constituent weights as of a given date.

    Reads the daily panel rather than the rebalance snapshots, so the
    answer includes everything that happened since the last rebalance:
    price drift, a deletion, a divisor adjustment. Compare
    :meth:`get_weights_on_date`, which answers the different question of
    what the last rebalance decided.

    Falls back to the latest recorded date on or before *date* — which
    covers a day the holdings could not be valued at all, since such a day
    records no rows.

    Args:
        date: The query date.

    Returns:
        dict: Mapping of identifier to weight. Empty when nothing was
        recorded on or before *date*, including when no panel was captured
        at all.
    """
    panel = self.daily_weights
    if panel.empty:
        return {}

    recorded = panel["DATE"] <= date
    if not recorded.any():
        return {}

    latest = panel.loc[recorded, "DATE"].max()
    rows = panel.loc[panel["DATE"] == latest]

    return {str(identifier): float(weight)
            for identifier, weight in zip(rows["IDENTIFIER"],
                                          rows["WEIGHT"],
                                          strict=True)}

to_dataframe

to_dataframe() -> pd.DataFrame

Flatten index levels and divisor history into a DataFrame.

Returns:

Type	Description
DataFrame	pd.DataFrame: Columns: index_level, divisor.

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

def to_dataframe(self) -> pd.DataFrame:
    """Flatten index levels and divisor history into a DataFrame.

    Returns:
        pd.DataFrame: Columns: ``index_level``, ``divisor``.
    """
    df = pd.DataFrame({
        "index_level": self.index_levels,
        "divisor": self.divisor_history,
    })
    df.index.name = "date"
    return df

calculate_derived_index

calculate_derived_index(
    definition: OptimisedIndexDefinition,
    data_provider: DataFetcher,
    start_date: str | None = None,
    end_date: str | None = None,
    price_column: str = "CLOSE",
    parent_result: IndexResult | None = None,
) -> IndexResult

Calculate an optimised index into a standard :class:IndexResult.

The three-step workflow of the design record: calculate the parent (or accept a pre-supplied calculation — the Backtest integration passes its cached one), solve the parent's published weights at every rebalance under the definition's constraints, then chain the solved weights into the derived index's own daily levels.

Parameters:

Name	Type	Description	Default
definition	OptimisedIndexDefinition	The derivation to calculate.	required
data_provider	DataFetcher	Data source for the parent calculation, prices and FX.	required
start_date	str | None	First date (YYYY-MM-DD). Defaults to the definition's base date. Ignored when parent_result is supplied, whose own window governs.	None
end_date	str | None	Last date (YYYY-MM-DD). Required unless parent_result is supplied.	None
price_column	str	Market-data column read as the price.	'CLOSE'
parent_result	IndexResult | None	The source's calculation, when the caller already has it. None calculates the source here — recursively, when the source is itself optimised.	None

Returns:

Name	Type	Description
IndexResult	IndexResult	Daily levels, divisor history, constituent and weight
	IndexResult	snapshots at exactly the parent's rebalance dates, and the daily
	IndexResult	weights panel — a normal index result, data-bound to data_provider.

Raises:

Type	Description
CalculationError	If the objective is unknown, the parent produced no rebalance snapshots to solve, or a solve is infeasible — the solver's own message names the binding conflict.
ValueError	If no window end is available to calculate the parent.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/index/derived.py

def calculate_derived_index(definition: OptimisedIndexDefinition,
                            data_provider: DataFetcher,
                            start_date: str | None = None,
                            end_date: str | None = None,
                            price_column: str = "CLOSE",
                            parent_result: IndexResult | None = None) -> IndexResult:
    """Calculate an optimised index into a standard :class:`IndexResult`.

    The three-step workflow of the design record: calculate the parent (or
    accept a pre-supplied calculation — the Backtest integration passes its
    cached one), solve the parent's published weights at every rebalance under
    the definition's constraints, then chain the solved weights into the
    derived index's own daily levels.

    Args:
        definition: The derivation to calculate.
        data_provider: Data source for the parent calculation, prices and FX.
        start_date: First date (YYYY-MM-DD). Defaults to the definition's
            base date. Ignored when *parent_result* is supplied, whose own
            window governs.
        end_date: Last date (YYYY-MM-DD). Required unless *parent_result* is
            supplied.
        price_column: Market-data column read as the price.
        parent_result: The source's calculation, when the caller already has
            it. None calculates the source here — recursively, when the
            source is itself optimised.

    Returns:
        IndexResult: Daily levels, divisor history, constituent and weight
        snapshots at exactly the parent's rebalance dates, and the daily
        weights panel — a normal index result, data-bound to *data_provider*.

    Raises:
        CalculationError: If the objective is unknown, the parent produced no
            rebalance snapshots to solve, or a solve is infeasible — the
            solver's own message names the binding conflict.
        ValueError: If no window end is available to calculate the parent.
    """
    check_objective(definition)

    parent = (parent_result if parent_result is not None
              else calculate_source(definition.source, data_provider,
                                    start_date, end_date, price_column))

    if not parent.weight_snapshots:
        raise CalculationError(
            "DerivedIndex",
            f"the source index '{definition.source.index_id}' produced no "
            f"rebalance snapshots, so there are no weights to optimise for "
            f"'{definition.index_id}'.")

    solved = _solved_schedule(definition, parent)

    return chain_levels(definition.index_id,
                        definition.base_value,
                        definition.currency,
                        parent,
                        solved,
                        data_provider,
                        price_column).with_data(data_provider)