beacon.analysis

Performance and risk analytics for indices, ETFs, and portfolios: ETF tracking metrics (beacon.analysis.etf), performance attribution, and the scalar risk measures covered on the Risk Model concept page.

analysis

The init.py for the 'analysis' module.

This module provides tools for analyzing the performance and risk characteristics of indices, ETFs, and portfolios.

Attribution

Kept for the original portfolio-versus-benchmark helper.

simple_performance_attribution

simple_performance_attribution(
    portfolio_returns: Series, benchmark_returns: Series
) -> dict[str, float]

Total return difference between a portfolio and a benchmark.

Parameters:

Name	Type	Description	Default
portfolio_returns	Series	Portfolio periodic returns.	required
benchmark_returns	Series	Benchmark periodic returns, same length.	required

Returns:

Name	Type	Description
dict	dict[str, float]	total_portfolio_return, total_benchmark_return and
	dict[str, float]	active_return.

Raises:

Type	Description
TypeError	If either input is not a Series.
ValueError	If the lengths differ or the inputs are empty.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def simple_performance_attribution(self,
                                   portfolio_returns: pd.Series,
                                   benchmark_returns: pd.Series) -> dict[str, float]:
    """Total return difference between a portfolio and a benchmark.

    Args:
        portfolio_returns: Portfolio periodic returns.
        benchmark_returns: Benchmark periodic returns, same length.

    Returns:
        dict: total_portfolio_return, total_benchmark_return and
        active_return.

    Raises:
        TypeError: If either input is not a Series.
        ValueError: If the lengths differ or the inputs are empty.
    """
    if (not isinstance(portfolio_returns, pd.Series)
            or not isinstance(benchmark_returns, pd.Series)):
        raise TypeError("portfolio_returns and benchmark_returns must be pandas Series.")
    if len(portfolio_returns) != len(benchmark_returns):
        raise ValueError(
            "Portfolio returns and benchmark returns Series must be of the same length.")
    if portfolio_returns.empty:
        raise ValueError("Input Series cannot be empty.")

    total_portfolio_return = (1 + portfolio_returns).prod() - 1
    total_benchmark_return = (1 + benchmark_returns).prod() - 1

    return {
        "total_portfolio_return": float(total_portfolio_return),
        "total_benchmark_return": float(total_benchmark_return),
        "active_return": float(total_portfolio_return - total_benchmark_return),
    }

AttributionResult dataclass

AttributionResult(
    start: str,
    end: str,
    periods: int,
    total_return: float,
    contributions: list[Contribution],
    residual: float,
    cap_drag: float | None = None,
    cost_drag: float | None = None,
    _weights: DataFrame | None = None,
)

A decomposition of one return into per-constituent contributions.

Attributes:

Name	Type	Description
start	str	First date of the window, ISO 8601.
end	str	Last date, ISO 8601.
periods	int	Return periods decomposed.
total_return	float	The return being explained.
contributions	list[Contribution]	Per constituent, largest first. Sums to total_return up to residual.
residual	float	total_return minus the sum of contributions. Reported always, expected to be at machine epsilon after linking. It is never folded into a constituent.
cap_drag	float | None	Capped return minus uncapped return, when the index applies a cap. Negative when capping cost the index. None when uncapped.
cost_drag	float | None	Portfolio return minus its gross return, when a backtest is supplied. Negative by construction — costs only subtract.

explained property

explained: float

Sum of the contributions.

reconciles

reconciles(tolerance: float = 1e-09) -> bool

Whether the contributions account for the total return.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def reconciles(self,
               tolerance: float = 1e-9) -> bool:
    """Whether the contributions account for the total return."""
    return abs(self.residual) <= tolerance

to_frame

to_frame() -> pd.DataFrame

Contributions as a DataFrame, largest first.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def to_frame(self) -> pd.DataFrame:
    """Contributions as a DataFrame, largest first."""
    return pd.DataFrame([
        {"asset_id": item.asset_id,
         "contribution": item.contribution,
         "average_weight": item.average_weight,
         "total_return": item.total_return}
        for item in self.contributions
    ])

Contribution dataclass

Contribution(
    asset_id: str,
    contribution: float,
    average_weight: float,
    total_return: float,
)

One constituent's share of the total return.

Attributes:

Name	Type	Description
asset_id	str	The constituent.
contribution	float	Its linked contribution. These sum to the total return.
average_weight	float	Mean weight across the window, for context — a large contribution from a small average weight is a different story from the same contribution from a large one.
total_return	float	The constituent's own return over the window.

ConcentrationMetrics dataclass

ConcentrationMetrics(
    assets: int,
    herfindahl_index: float,
    effective_assets: float,
    largest_weight: float,
    largest_asset_id: str | None,
)

How concentrated a set of weights is.

Attributes:

Name	Type	Description
assets	int	Number of weighted positions.
herfindahl_index	float	Sum of squared weights. For weights summing to 1 this runs from 1/n (perfectly equal) to 1 (everything in one name).
effective_assets	float	1 / herfindahl_index — the number of equally weighted positions that would be this concentrated. Reads more naturally than the index itself: "this 100-stock index behaves like 23 equal positions".
largest_weight	float	The biggest single weight.
largest_asset_id	str | None	Which asset holds it, or None when there are no positions.

DriftMetrics dataclass

DriftMetrics(
    per_asset: dict[str, float],
    max_absolute: float,
    max_absolute_asset_id: str | None,
    total_absolute: float,
    turnover: float,
)

How far current weights have moved from their targets.

Attributes:

Name	Type	Description
per_asset	dict[str, float]	Current minus target for every asset in either set. Positive means overweight.
max_absolute	float	Largest absolute drift across assets.
max_absolute_asset_id	str | None	Which asset drifted most, or None when there is nothing to compare.
total_absolute	float	Sum of absolute drifts.
turnover	float	Half of total_absolute — the one-way trading needed to return to target, since every overweight funds an underweight.

RiskMetricsCalculator

RiskMetricsCalculator()

A class to calculate various risk metrics. This class can be expanded to hold state or more complex configurations if needed.

Initializes the RiskMetricsCalculator.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def __init__(self) -> None:
    """Initializes the RiskMetricsCalculator."""

calculate_volatility

calculate_volatility(
    price_series: Series, window: int = 252
) -> float

Calculates annualized volatility from a price series.

Parameters:

Name	Type	Description	Default
price_series	Series	A pandas Series of prices.	required
window	int	The number of trading periods in a year (e.g., 252 for daily).	252

Returns:

Type	Description
float	The annualized volatility as a float.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def calculate_volatility(self,
                         price_series: pd.Series,
                         window: int = 252) -> float:
    """
    Calculates annualized volatility from a price series.

    Args:
        price_series: A pandas Series of prices.
        window: The number of trading periods in a year (e.g., 252 for daily).

    Returns:
        The annualized volatility as a float.
    """
    return calculate_volatility(price_series, window)

calculate_sharpe_ratio

calculate_sharpe_ratio(
    returns: Series,
    risk_free_rate: float,
    periods_per_year: int = 252,
) -> float

Calculates the annualized Sharpe Ratio.

Parameters:

Name	Type	Description	Default
returns	Series	A pandas Series of periodic returns.	required
risk_free_rate	float	The annualized risk-free rate.	required
periods_per_year	int	The number of return periods in a year (e.g., 252 for daily, 12 for monthly).	252

Returns:

Type	Description
float	The annualized Sharpe Ratio as a float.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def calculate_sharpe_ratio(self,
                           returns: pd.Series,
                           risk_free_rate: float,
                           periods_per_year: int = 252) -> float:
    """
    Calculates the annualized Sharpe Ratio.

    Args:
        returns: A pandas Series of periodic returns.
        risk_free_rate: The annualized risk-free rate.
        periods_per_year: The number of return periods in a year (e.g., 252 for daily,
                          12 for monthly).

    Returns:
        The annualized Sharpe Ratio as a float.
    """
    return calculate_sharpe_ratio(returns, risk_free_rate, periods_per_year)

calculate_max_drawdown

calculate_max_drawdown(price_series: Series) -> float

Calculates the maximum drawdown from a price series.

Parameters:

Name	Type	Description	Default
price_series	Series	A pandas Series of prices.	required

Returns:

Type	Description
float	The maximum drawdown as a float (e.g., 0.2 for a 20% drawdown).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def calculate_max_drawdown(self,
                           price_series: pd.Series) -> float:
    """
    Calculates the maximum drawdown from a price series.

    Args:
        price_series: A pandas Series of prices.

    Returns:
        The maximum drawdown as a float (e.g., 0.2 for a 20% drawdown).
    """
    return calculate_max_drawdown(price_series)

attribute

attribute(
    period_returns: Series,
    weights: DataFrame,
    asset_returns: DataFrame,
    cap_drag: float | None = None,
    cost_drag: float | None = None,
) -> AttributionResult

Decompose a return series into per-constituent contributions.

Parameters:

Name	Type	Description	Default
period_returns	Series	The return being explained, per period.	required
weights	DataFrame	Weights per period, constituents on the columns. Aligned to period_returns; the weight used for a period is the one held at its start.	required
asset_returns	DataFrame	Constituent returns per period.	required
cap_drag	float | None	Optional capped-minus-uncapped return.	None
cost_drag	float | None	Optional cost effect on the portfolio return.	None

Returns:

Name	Type	Description
AttributionResult	AttributionResult	The decomposition, with contributions summing to the
	AttributionResult	compounded total return and a residual reported separately.

Raises:

Type	Description
CalculationError	If the inputs cannot be aligned, or a period wipes out the index.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def attribute(period_returns: pd.Series,
              weights: pd.DataFrame,
              asset_returns: pd.DataFrame,
              cap_drag: float | None = None,
              cost_drag: float | None = None) -> AttributionResult:
    """Decompose a return series into per-constituent contributions.

    Args:
        period_returns: The return being explained, per period.
        weights: Weights per period, constituents on the columns. Aligned to
            *period_returns*; the weight used for a period is the one held at
            its start.
        asset_returns: Constituent returns per period.
        cap_drag: Optional capped-minus-uncapped return.
        cost_drag: Optional cost effect on the portfolio return.

    Returns:
        AttributionResult: The decomposition, with contributions summing to the
        compounded total return and a residual reported separately.

    Raises:
        CalculationError: If the inputs cannot be aligned, or a period wipes
            out the index.
    """
    common = period_returns.index.intersection(weights.index).intersection(
        asset_returns.index).sort_values()

    if len(common) == 0:
        raise CalculationError(
            "Attribution",
            "the return, weight and constituent-return series share no dates.")

    returns = period_returns.loc[common]
    assets = sorted(set(weights.columns) & set(asset_returns.columns))

    # The weight that earns a period's return is the one held at its start,
    # hence the shift. Using the end-of-period weight would credit a
    # constituent for a move it was not yet holding.
    lagged = weights[assets].reindex(common).shift(1)
    contributions = lagged * asset_returns[assets].reindex(common)
    contributions = contributions.dropna(how="all")

    linked = link_contributions(contributions, returns.loc[contributions.index])
    total = float((1.0 + returns.loc[contributions.index]).prod() - 1.0)

    rows = [
        Contribution(asset_id=asset_id,
                     contribution=float(linked.get(asset_id, 0.0)),
                     average_weight=float(lagged[asset_id].mean()),
                     total_return=float(
                         (1.0 + asset_returns[asset_id].reindex(
                             contributions.index).fillna(0.0)).prod() - 1.0))
        for asset_id in assets
    ]
    rows.sort(key=lambda item: item.contribution, reverse=True)

    return AttributionResult(
        start=contributions.index[0].isoformat(),
        end=contributions.index[-1].isoformat(),
        periods=len(contributions),
        total_return=total,
        contributions=rows,
        residual=total - float(linked.sum()),
        cap_drag=cap_drag,
        cost_drag=cost_drag,
        _weights=lagged)

cap_drag

cap_drag(
    capped_weights: dict[Timestamp, dict[str, float]],
    uncapped_weights: dict[Timestamp, dict[str, float]],
    prices: DataFrame,
) -> float

What capping cost, or gained, over the window.

The capped index's return minus the return of the same methodology left uncapped. Negative when the cap held back a name that went on to outperform, which is the usual case and the reason the number is worth reporting.

Both paths are built by drifting their own snapshots forward, so the comparison isolates the effect of the cap rather than of any other difference.

Parameters:

Name	Type	Description	Default
capped_weights	dict[Timestamp, dict[str, float]]	Rebalance date -> capped weights.	required
uncapped_weights	dict[Timestamp, dict[str, float]]	Rebalance date -> weights before capping.	required
prices	DataFrame	Constituent prices over the window.	required

Returns:

Name	Type	Description
float	float	Capped total return minus uncapped total return.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def cap_drag(capped_weights: dict[pd.Timestamp, dict[str, float]],
             uncapped_weights: dict[pd.Timestamp, dict[str, float]],
             prices: pd.DataFrame) -> float:
    """What capping cost, or gained, over the window.

    The capped index's return minus the return of the same methodology left
    uncapped. Negative when the cap held back a name that went on to
    outperform, which is the usual case and the reason the number is worth
    reporting.

    Both paths are built by drifting their own snapshots forward, so the
    comparison isolates the effect of the cap rather than of any other
    difference.

    Args:
        capped_weights: Rebalance date -> capped weights.
        uncapped_weights: Rebalance date -> weights before capping.
        prices: Constituent prices over the window.

    Returns:
        float: Capped total return minus uncapped total return.
    """
    asset_returns = prices.pct_change()

    capped_return = _path_return(capped_weights, prices, asset_returns)
    uncapped_return = _path_return(uncapped_weights, prices, asset_returns)

    return capped_return - uncapped_return

carino_factor

carino_factor(period_return: float) -> float

The Carino coefficient for one period.

ln(1+R)/R, with the removable singularity at R = 0 filled in with its limit of 1.

Parameters:

Name	Type	Description	Default
period_return	float	The period's total return.	required

Returns:

Name	Type	Description
float	float	The coefficient.

Raises:

Type	Description
CalculationError	If the return is -100% or worse, where the logarithm is undefined. An index that goes to zero cannot have its return attributed, and silently substituting a number would hide that.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def carino_factor(period_return: float) -> float:
    """The Carino coefficient for one period.

    ``ln(1+R)/R``, with the removable singularity at R = 0 filled in with its
    limit of 1.

    Args:
        period_return: The period's total return.

    Returns:
        float: The coefficient.

    Raises:
        CalculationError: If the return is -100% or worse, where the logarithm
            is undefined. An index that goes to zero cannot have its return
            attributed, and silently substituting a number would hide that.
    """
    if period_return <= -1.0:
        raise CalculationError(
            "Attribution",
            f"a period return of {period_return:.4%} wipes out the index; the "
            "linking coefficient is undefined there.")

    if abs(period_return) < NEGLIGIBLE_RETURN:
        return 1.0

    return math.log1p(period_return) / period_return

cost_drag

cost_drag(
    total_costs: float, initial_capital: float
) -> float

Direct effect of transaction costs on a portfolio's return.

Costs paid as a fraction of starting capital, negated so it reads as a drag. This is the direct effect only: it excludes the compounding of the capital that was spent rather than invested, which is second-order but not zero over a long window. Reporting the direct figure keeps the number explainable — it is exactly the money that left the portfolio — and a caller wanting the full effect can difference a zero-cost run instead.

Parameters:

Name	Type	Description	Default
total_costs	float	Sum of transaction costs paid.	required
initial_capital	float	Capital the portfolio started with.	required

Returns:

Name	Type	Description
float	float	A non-positive drag.

Raises:

Type	Description
CalculationError	If initial_capital is not positive.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def cost_drag(total_costs: float,
              initial_capital: float) -> float:
    """Direct effect of transaction costs on a portfolio's return.

    Costs paid as a fraction of starting capital, negated so it reads as a
    drag. This is the **direct** effect only: it excludes the compounding of
    the capital that was spent rather than invested, which is second-order but
    not zero over a long window. Reporting the direct figure keeps the number
    explainable — it is exactly the money that left the portfolio — and a
    caller wanting the full effect can difference a zero-cost run instead.

    Args:
        total_costs: Sum of transaction costs paid.
        initial_capital: Capital the portfolio started with.

    Returns:
        float: A non-positive drag.

    Raises:
        CalculationError: If *initial_capital* is not positive.
    """
    if initial_capital <= 0:
        raise CalculationError(
            "CostDrag", f"initial_capital must be positive, got {initial_capital}.")

    return -abs(total_costs) / initial_capital

drifted_weights

drifted_weights(
    snapshots: dict[Timestamp, dict[str, float]],
    prices: DataFrame,
) -> pd.DataFrame

Reconstruct the index's daily weights from its rebalance snapshots.

Weights are set at a rebalance and then drift with relative performance until the next one, because the index holds fixed units in between. Given the weight at a rebalance and prices since, the drifted weight is

w_i,t ∝ w_i,rebalance × (p_i,t / p_i,rebalance)

normalised across constituents. The unit scale cancels, so nothing beyond the snapshot and prices is needed.

Parameters:

Name	Type	Description	Default
snapshots	dict[Timestamp, dict[str, float]]	Rebalance date -> weights on that date.	required
prices	DataFrame	Dates on the index, constituents on the columns.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: Weights for every date at or after the first rebalance,
DataFrame	each row summing to 1.

Raises:

Type	Description
CalculationError	If there are no snapshots to start from.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def drifted_weights(snapshots: dict[pd.Timestamp, dict[str, float]],
                    prices: pd.DataFrame) -> pd.DataFrame:
    """Reconstruct the index's daily weights from its rebalance snapshots.

    Weights are set at a rebalance and then drift with relative performance
    until the next one, because the index holds fixed units in between. Given
    the weight at a rebalance and prices since, the drifted weight is

        w_i,t ∝ w_i,rebalance × (p_i,t / p_i,rebalance)

    normalised across constituents. The unit scale cancels, so nothing beyond
    the snapshot and prices is needed.

    Args:
        snapshots: Rebalance date -> weights on that date.
        prices: Dates on the index, constituents on the columns.

    Returns:
        pd.DataFrame: Weights for every date at or after the first rebalance,
        each row summing to 1.

    Raises:
        CalculationError: If there are no snapshots to start from.
    """
    if not snapshots:
        raise CalculationError(
            "Attribution", "the index has no weight snapshots to attribute from.")

    rebalances = sorted(snapshots)
    rows: dict[pd.Timestamp, dict[str, float]] = {}

    for date in prices.index:
        active = [r for r in rebalances if r <= date]
        if not active:
            continue

        rows[date] = _weights_on(snapshots[active[-1]], prices, active[-1], date)

    return pd.DataFrame.from_dict(rows, orient="index").fillna(0.0)

link_contributions

link_contributions(
    contributions: DataFrame, period_returns: Series
) -> pd.Series

Scale per-period contributions so they sum to the compounded return.

Parameters:

Name	Type	Description	Default
contributions	DataFrame	Periods on the index, constituents on the columns. Each row must sum to that period's return.	required
period_returns	Series	The total return of each period.	required

Returns:

Type	Description
Series	pd.Series: One linked contribution per constituent. Their sum equals
Series	the compounded total return exactly.

Raises:

Type	Description
CalculationError	If any period return is -100% or worse.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def link_contributions(contributions: pd.DataFrame,
                       period_returns: pd.Series) -> pd.Series:
    """Scale per-period contributions so they sum to the compounded return.

    Args:
        contributions: Periods on the index, constituents on the columns. Each
            row must sum to that period's return.
        period_returns: The total return of each period.

    Returns:
        pd.Series: One linked contribution per constituent. Their sum equals
        the compounded total return exactly.

    Raises:
        CalculationError: If any period return is -100% or worse.
    """
    if contributions.empty:
        return pd.Series(dtype=float)

    total = float((1.0 + period_returns).prod() - 1.0)

    factors = period_returns.map(carino_factor)
    total_factor = carino_factor(total)

    scaled = contributions.mul(factors / total_factor, axis=0)

    return scaled.sum(axis=0)

simple_performance_attribution

simple_performance_attribution(
    portfolio_returns: Series, benchmark_returns: Series
) -> dict[str, float]

Total return difference between a portfolio and a benchmark.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/attribution.py

def simple_performance_attribution(portfolio_returns: pd.Series,
                                   benchmark_returns: pd.Series) -> dict[str, float]:
    """Total return difference between a portfolio and a benchmark."""
    return Attribution().simple_performance_attribution(portfolio_returns,
                                                        benchmark_returns)

drift_from_target

drift_from_target(
    current: dict[str, float], target: dict[str, float]
) -> DriftMetrics

Compare held weights against their targets.

Every asset appearing in either mapping is included, treating absence as a zero weight — a position that has been fully sold, or one the target wants but the portfolio does not hold, is precisely the drift worth seeing.

Parameters:

Name	Type	Description	Default
current	dict[str, float]	Held weights.	required
target	dict[str, float]	Target weights.	required

Returns:

Name	Type	Description
DriftMetrics	DriftMetrics	The comparison. Empty inputs give zeros and a None asset
	DriftMetrics	rather than raising.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/concentration.py

def drift_from_target(current: dict[str, float],
                      target: dict[str, float]) -> DriftMetrics:
    """Compare held weights against their targets.

    Every asset appearing in either mapping is included, treating absence as a
    zero weight — a position that has been fully sold, or one the target wants
    but the portfolio does not hold, is precisely the drift worth seeing.

    Args:
        current: Held weights.
        target: Target weights.

    Returns:
        DriftMetrics: The comparison. Empty inputs give zeros and a None asset
        rather than raising.
    """
    assets = set(current) | set(target)

    if not assets:
        return DriftMetrics(per_asset={},
                            max_absolute=0.0,
                            max_absolute_asset_id=None,
                            total_absolute=0.0,
                            turnover=0.0)

    per_asset = {
        asset_id: float(current.get(asset_id, 0.0) - target.get(asset_id, 0.0))
        for asset_id in sorted(assets)
    }

    worst = _worst_drifter(per_asset)
    total = float(sum(abs(value) for value in per_asset.values()))

    return DriftMetrics(per_asset=per_asset,
                        max_absolute=abs(per_asset[worst]),
                        max_absolute_asset_id=worst,
                        total_absolute=total,
                        turnover=total / 2.0)

drift_history

drift_history(
    weight_history: dict[str, dict[str, float]],
    target: dict[str, float],
) -> dict[str, DriftMetrics]

Drift at each of several snapshots against one set of targets.

Parameters:

Name	Type	Description	Default
weight_history	dict[str, dict[str, float]]	Mapping of snapshot label (typically an ISO date) to the weights held at that point.	required
target	dict[str, float]	The target weights to compare each snapshot against.	required

Returns:

Name	Type	Description
dict	dict[str, DriftMetrics]	Snapshot label -> DriftMetrics, in the order the labels sort.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/concentration.py

def drift_history(weight_history: dict[str, dict[str, float]],
                  target: dict[str, float]) -> dict[str, DriftMetrics]:
    """Drift at each of several snapshots against one set of targets.

    Args:
        weight_history: Mapping of snapshot label (typically an ISO date) to
            the weights held at that point.
        target: The target weights to compare each snapshot against.

    Returns:
        dict: Snapshot label -> DriftMetrics, in the order the labels sort.
    """
    return {label: drift_from_target(weights, target)
            for label, weights in sorted(weight_history.items())}

effective_number_of_assets

effective_number_of_assets(
    weights: dict[str, float],
) -> float

Number of equally weighted positions with the same concentration.

Parameters:

Name	Type	Description	Default
weights	dict[str, float]	Mapping of asset id to weight.	required

Returns:

Name	Type	Description
float	float	1 / HHI. 0.0 when there are no positions, or when every
	float	weight is zero — neither has a meaningful effective count, and
	float	returning 0.0 keeps callers from having to guard against a division by
	float	zero they cannot act on.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/concentration.py

def effective_number_of_assets(weights: dict[str, float]) -> float:
    """Number of equally weighted positions with the same concentration.

    Args:
        weights: Mapping of asset id to weight.

    Returns:
        float: ``1 / HHI``. 0.0 when there are no positions, or when every
        weight is zero — neither has a meaningful effective count, and
        returning 0.0 keeps callers from having to guard against a division by
        zero they cannot act on.
    """
    index = herfindahl_index(weights)

    if index <= 0.0:
        return 0.0

    return 1.0 / index

herfindahl_index

herfindahl_index(weights: dict[str, float]) -> float

Sum of squared weights.

Parameters:

Name	Type	Description	Default
weights	dict[str, float]	Mapping of asset id to weight, expected to sum to 1.	required

Returns:

Name	Type	Description
float	float	The index. 0.0 for no positions. Note that squaring makes this
	float	blind to sign, so a short position concentrates the measure exactly as
	float	a long one of the same size would.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/concentration.py

def herfindahl_index(weights: dict[str, float]) -> float:
    """Sum of squared weights.

    Args:
        weights: Mapping of asset id to weight, expected to sum to 1.

    Returns:
        float: The index. 0.0 for no positions. Note that squaring makes this
        blind to sign, so a short position concentrates the measure exactly as
        a long one of the same size would.
    """
    if not weights:
        return 0.0

    _warn_if_not_fully_invested(weights)

    return float(sum(weight * weight for weight in weights.values()))

top_n_weight

top_n_weight(
    weights: dict[str, float], count: int
) -> float

Combined weight of the count largest positions.

The measure a concentration limit is usually written against — "no more than 40% in the top five" — and not derivable from the Herfindahl index.

Parameters:

Name	Type	Description	Default
weights	dict[str, float]	Mapping of asset id to weight.	required
count	int	How many of the largest positions to sum. Larger than the number of positions sums all of them.	required

Returns:

Name	Type	Description
float	float	The combined weight.

Raises:

Type	Description
CalculationError	If count is not positive.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/concentration.py

def top_n_weight(weights: dict[str, float],
                 count: int) -> float:
    """Combined weight of the *count* largest positions.

    The measure a concentration limit is usually written against — "no more
    than 40% in the top five" — and not derivable from the Herfindahl index.

    Args:
        weights: Mapping of asset id to weight.
        count: How many of the largest positions to sum. Larger than the number
            of positions sums all of them.

    Returns:
        float: The combined weight.

    Raises:
        CalculationError: If *count* is not positive.
    """
    if count <= 0:
        raise CalculationError("TopNWeight", f"count must be positive, got {count}.")

    largest = sorted(weights.values(), reverse=True)[:count]

    return float(sum(largest))

average_daily_volume

average_daily_volume(
    market: DataFrame,
    as_of: Timestamp,
    months: int = TRAILING_MONTHS,
    column: str = VOLUME_COLUMN,
) -> pd.Series

Mean daily volume over the trailing window, per identifier.

Parameters:

Name	Type	Description	Default
market	DataFrame	Market data MultiIndexed by (IDENTIFIER, DATE), as DataFetcher.fetch_market_data returns for a list of identifiers.	required
as_of	Timestamp	End of the window, inclusive.	required
months	int	Length of the window in calendar months.	TRAILING_MONTHS
column	str	Volume column to average.	VOLUME_COLUMN

Returns:

Type	Description
Series	pd.Series: Indexed by identifier. Empty when the frame is empty or
Series	carries no volume column — an absent column is a property of the
Series	dataset, not a failure of the request, so it produces no answer rather
Series	than an error.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/liquidity.py

def average_daily_volume(market: pd.DataFrame,
                         as_of: pd.Timestamp,
                         months: int = TRAILING_MONTHS,
                         column: str = VOLUME_COLUMN) -> pd.Series:
    """Mean daily volume over the trailing window, per identifier.

    Args:
        market: Market data MultiIndexed by ``(IDENTIFIER, DATE)``, as
            ``DataFetcher.fetch_market_data`` returns for a list of
            identifiers.
        as_of: End of the window, inclusive.
        months: Length of the window in calendar months.
        column: Volume column to average.

    Returns:
        pd.Series: Indexed by identifier. Empty when the frame is empty or
        carries no volume column — an absent column is a property of the
        dataset, not a failure of the request, so it produces no answer rather
        than an error.
    """
    if market.empty or column not in market.columns:
        return pd.Series(dtype="float64")

    dates = market.index.get_level_values("DATE")
    start = pd.Timestamp(as_of) - pd.DateOffset(months=months)

    window = market.loc[(dates > start) & (dates <= pd.Timestamp(as_of))]
    if window.empty:
        logger.warning(
            "No volume in the %d month(s) to %s, so no ADV could be computed.",
            months, pd.Timestamp(as_of).date())

        return pd.Series(dtype="float64")

    return window[column].groupby(level="IDENTIFIER").mean()

calculate_max_drawdown

calculate_max_drawdown(price_series: Series) -> float

Calculates the maximum drawdown from a price series.

Parameters:

Name	Type	Description	Default
price_series	Series	A pandas Series of prices.	required

Returns:

Type	Description
float	The maximum drawdown as a float (e.g., 0.2 for a 20% drawdown).

Raises:

Type	Description
ValueError	If price_series is empty or contains non-numeric data.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def calculate_max_drawdown(price_series: pd.Series) -> float:
    """
    Calculates the maximum drawdown from a price series.

    Args:
        price_series: A pandas Series of prices.

    Returns:
        The maximum drawdown as a float (e.g., 0.2 for a 20% drawdown).

    Raises:
        ValueError: If price_series is empty or contains non-numeric data.
    """
    if not isinstance(price_series, pd.Series):
        raise TypeError("price_series must be a pandas Series.")
    if price_series.empty:
        raise ValueError("Price series cannot be empty.")
    if not pd.api.types.is_numeric_dtype(price_series):
        raise ValueError("Price series must contain numeric data.")

    cumulative_max = price_series.cummax()
    drawdown = (price_series - cumulative_max) / cumulative_max
    max_drawdown = drawdown.min()
    return float(max_drawdown) if not pd.isna(max_drawdown) else 0.0

calculate_sharpe_ratio

calculate_sharpe_ratio(
    returns: Series,
    risk_free_rate: float,
    periods_per_year: int = 252,
) -> float

Calculates the annualized Sharpe Ratio.

Parameters:

Name	Type	Description	Default
returns	Series	A pandas Series of periodic returns.	required
risk_free_rate	float	The annualized risk-free rate.	required
periods_per_year	int	The number of return periods in a year (e.g., 252 for daily, 12 for monthly).	252

Returns:

Type	Description
float	The annualized Sharpe Ratio as a float.

Raises:

Type	Description
ValueError	If inputs are invalid.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def calculate_sharpe_ratio(returns: pd.Series,
                           risk_free_rate: float,
                           periods_per_year: int = 252) -> float:
    """
    Calculates the annualized Sharpe Ratio.

    Args:
        returns: A pandas Series of periodic returns.
        risk_free_rate: The annualized risk-free rate.
        periods_per_year: The number of return periods in a year (e.g., 252 for daily,
                          12 for monthly).

    Returns:
        The annualized Sharpe Ratio as a float.

    Raises:
        ValueError: If inputs are invalid.
    """
    if not isinstance(returns, pd.Series):
        raise TypeError("returns must be a pandas Series.")
    if returns.empty:
        raise ValueError("Returns series cannot be empty.")
    if not pd.api.types.is_numeric_dtype(returns):
        raise ValueError("Returns series must contain numeric data.")
    if not isinstance(risk_free_rate, (int, float)):
        raise TypeError("risk_free_rate must be a number.")
    if periods_per_year <= 0:
        raise ValueError("periods_per_year must be a positive integer.")

    excess_returns = returns - (risk_free_rate / periods_per_year)
    mean_excess_return = excess_returns.mean()
    std_dev_excess_return = excess_returns.std()

    if std_dev_excess_return == 0: # Avoid division by zero
        return np.nan if mean_excess_return == 0 else np.inf * np.sign(mean_excess_return)

    sharpe_ratio = (mean_excess_return / std_dev_excess_return) * np.sqrt(periods_per_year)
    return float(sharpe_ratio)

calculate_volatility

calculate_volatility(
    price_series: Series, window: int = 252
) -> float

Calculates annualized volatility from a price series.

Parameters:

Name	Type	Description	Default
price_series	Series	A pandas Series of prices.	required
window	int	The number of trading periods in a year (e.g., 252 for daily).	252

Returns:

Type	Description
float	The annualized volatility as a float.

Raises:

Type	Description
ValueError	If price_series is empty or contains non-numeric data.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/analysis/risk.py

def calculate_volatility(price_series: pd.Series,
                         window: int = 252) -> float:
    """
    Calculates annualized volatility from a price series.

    Args:
        price_series: A pandas Series of prices.
        window: The number of trading periods in a year (e.g., 252 for daily).

    Returns:
        The annualized volatility as a float.

    Raises:
        ValueError: If price_series is empty or contains non-numeric data.
    """
    if not isinstance(price_series, pd.Series):
        raise TypeError("price_series must be a pandas Series.")
    if price_series.empty:
        raise ValueError("Price series cannot be empty.")
    if not pd.api.types.is_numeric_dtype(price_series):
        raise ValueError("Price series must contain numeric data.")
    if window <= 0:
        raise ValueError("Window must be a positive integer.")

    returns = price_series.pct_change().dropna()
    annualized_volatility = returns.std() * np.sqrt(window)
    return float(annualized_volatility)