Metrics API

PerformanceMetrics

Static methods for calculating various performance metrics.

All methods work with both Series (single asset) and DataFrame (multiple assets).

Examples:

>>> data = loader.fetch_data()
>>> annual_return = PerformanceMetrics.calculate_annual_return(data)
>>> sharpe = PerformanceMetrics.calculate_sharpe_ratio(data, risk_free_rate=0.02)

Source code in portfolio_analysis/metrics/performance.py

class PerformanceMetrics:
    """
    Static methods for calculating various performance metrics.

    All methods work with both Series (single asset) and DataFrame (multiple assets).

    Examples
    --------
    >>> data = loader.fetch_data()
    >>> annual_return = PerformanceMetrics.calculate_annual_return(data)
    >>> sharpe = PerformanceMetrics.calculate_sharpe_ratio(data, risk_free_rate=0.02)
    """

    TRADING_DAYS = TRADING_DAYS_PER_YEAR

    @staticmethod
    def calculate_annual_return(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate annualized return from price data.

        Uses year-end prices to calculate annual returns, then averages.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Annualized return(s)
        """
        annual_return = data.resample("Y").last().pct_change().mean()
        return annual_return

    @staticmethod
    def calculate_cagr(data: Union[pd.Series, pd.DataFrame]) -> Union[float, pd.Series]:
        """
        Calculate Compound Annual Growth Rate (CAGR).

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            CAGR value(s)
        """
        first_value = data.iloc[0]
        last_value = data.iloc[-1]
        years = (data.index[-1] - data.index[0]).days / DAYS_PER_YEAR

        cagr = (last_value / first_value) ** (1 / years) - 1
        return cagr

    @staticmethod
    def calculate_annual_volatility(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate annualized volatility (standard deviation of returns).

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Annualized volatility
        """
        daily_returns = data.pct_change().dropna()
        annual_volatility = daily_returns.std() * np.sqrt(
            PerformanceMetrics.TRADING_DAYS
        )
        return annual_volatility

    @staticmethod
    def calculate_sharpe_ratio(
        data: Union[pd.Series, pd.DataFrame],
        risk_free_rate: float = DEFAULT_RISK_FREE_RATE,
    ) -> Union[float, pd.Series]:
        """
        Calculate Sharpe ratio.

        Sharpe Ratio = (Return - Risk Free Rate) / Volatility

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index
        risk_free_rate : float, default 0.02
            Annual risk-free rate

        Returns
        -------
        float or pd.Series
            Sharpe ratio(s)
        """
        annual_return = PerformanceMetrics.calculate_annual_return(data)
        annual_volatility = PerformanceMetrics.calculate_annual_volatility(data)
        sharpe_ratio = (annual_return - risk_free_rate) / annual_volatility
        return sharpe_ratio

    @staticmethod
    def calculate_sortino_ratio(
        data: Union[pd.Series, pd.DataFrame],
        risk_free_rate: float = DEFAULT_RISK_FREE_RATE,
    ) -> Union[float, pd.Series]:
        """
        Calculate Sortino ratio (uses downside deviation instead of volatility).

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index
        risk_free_rate : float, default 0.02
            Annual risk-free rate

        Returns
        -------
        float or pd.Series
            Sortino ratio(s)
        """
        returns = data.pct_change().dropna()

        # Downside deviation (only negative returns, positive returns set to 0)
        # Using .where() for both Series and DataFrame ensures consistent behavior:
        # positive returns are replaced with 0, preserving the sample size
        downside_returns = returns.where(returns < 0, 0)

        downside_deviation = downside_returns.std() * np.sqrt(
            PerformanceMetrics.TRADING_DAYS
        )
        annual_return = PerformanceMetrics.calculate_annual_return(data)

        sortino_ratio = (annual_return - risk_free_rate) / downside_deviation
        return sortino_ratio

    @staticmethod
    def calculate_max_drawdown(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate maximum drawdown.

        Maximum drawdown is the largest peak-to-trough decline.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Maximum drawdown (negative value)
        """
        cumulative_returns = (1 + data.pct_change()).cumprod()
        peak = cumulative_returns.expanding(min_periods=1).max()
        drawdown = (cumulative_returns / peak) - 1
        max_drawdown = drawdown.min()
        return max_drawdown

    @staticmethod
    def calculate_var(
        data: Union[pd.Series, pd.DataFrame],
        confidence_level: float = DEFAULT_CONFIDENCE_LEVEL,
    ) -> Union[float, pd.Series]:
        """
        Calculate Value at Risk (VaR) using historical method.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index
        confidence_level : float, default 0.95
            Confidence level (e.g., 0.95 for 95%)

        Returns
        -------
        float or pd.Series
            VaR value (typically negative)
        """
        returns = data.pct_change().dropna()
        var = np.percentile(returns, (1 - confidence_level) * 100, axis=0)

        if isinstance(returns, pd.DataFrame):
            return pd.Series(var, index=returns.columns)
        return var

    @staticmethod
    def calculate_calmar_ratio(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate Calmar ratio (CAGR / Max Drawdown).

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Calmar ratio(s)
        """
        cagr = PerformanceMetrics.calculate_cagr(data)
        max_dd = PerformanceMetrics.calculate_max_drawdown(data)

        # Max drawdown is negative, so we negate it
        return cagr / abs(max_dd)

    @staticmethod
    def calculate_cvar(
        data: Union[pd.Series, pd.DataFrame],
        confidence_level: float = DEFAULT_CONFIDENCE_LEVEL,
    ) -> Union[float, pd.Series]:
        """
        Calculate Conditional Value at Risk (CVaR), also known as Expected Shortfall.

        CVaR represents the expected loss given that the loss exceeds VaR.
        It is a more conservative risk measure than VaR as it considers
        the tail of the distribution.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index
        confidence_level : float, default 0.95
            Confidence level (e.g., 0.95 for 95%)

        Returns
        -------
        float or pd.Series
            CVaR value (typically negative)
        """
        returns = data.pct_change().dropna()
        var = PerformanceMetrics.calculate_var(data, confidence_level)

        if isinstance(returns, pd.DataFrame):
            cvar_values = {}
            for col in returns.columns:
                col_returns = returns[col]
                col_var = var[col]
                cvar_values[col] = col_returns[col_returns <= col_var].mean()
            return pd.Series(cvar_values)
        else:
            return returns[returns <= var].mean()

    @staticmethod
    def calculate_omega_ratio(
        data: Union[pd.Series, pd.DataFrame],
        threshold: float = 0.0,
    ) -> Union[float, pd.Series]:
        """
        Calculate Omega ratio.

        The Omega ratio compares the probability-weighted gains above a threshold
        to the probability-weighted losses below it. Higher values indicate better
        risk-adjusted performance.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index
        threshold : float, default 0.0
            Daily return threshold (0.0 for break-even)

        Returns
        -------
        float or pd.Series
            Omega ratio(s)
        """
        returns = data.pct_change().dropna()

        if isinstance(returns, pd.DataFrame):
            omega_values = {}
            for col in returns.columns:
                col_returns = returns[col]
                gains = col_returns[col_returns > threshold] - threshold
                losses = threshold - col_returns[col_returns <= threshold]
                if losses.sum() == 0:
                    omega_values[col] = np.inf
                else:
                    omega_values[col] = gains.sum() / losses.sum()
            return pd.Series(omega_values)
        else:
            gains = returns[returns > threshold] - threshold
            losses = threshold - returns[returns <= threshold]
            if losses.sum() == 0:
                return np.inf
            return gains.sum() / losses.sum()

    @staticmethod
    def calculate_ulcer_index(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate Ulcer Index.

        The Ulcer Index measures downside volatility based on drawdowns.
        It penalizes deep and prolonged drawdowns more heavily than
        standard deviation. Lower values indicate less risk.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Ulcer Index value(s)
        """
        cumulative_returns = (1 + data.pct_change()).cumprod()
        peak = cumulative_returns.expanding(min_periods=1).max()
        drawdown = (cumulative_returns / peak) - 1

        # Ulcer Index is the quadratic mean of drawdowns
        ulcer_index = np.sqrt((drawdown**2).mean())
        return ulcer_index

    @staticmethod
    def calculate_recovery_factor(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate Recovery Factor.

        Recovery Factor = Total Return / |Max Drawdown|

        A higher recovery factor indicates the portfolio generates more
        return per unit of maximum drawdown risk.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Recovery factor(s)
        """
        total_return = (data.iloc[-1] / data.iloc[0]) - 1
        max_dd = PerformanceMetrics.calculate_max_drawdown(data)

        return total_return / abs(max_dd)

    @staticmethod
    def calculate_win_rate(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate Win Rate (percentage of positive return periods).

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Win rate as a decimal (e.g., 0.55 for 55% win rate)
        """
        returns = data.pct_change().dropna()
        win_rate = (returns > 0).mean()
        return win_rate

    @staticmethod
    def calculate_profit_factor(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate Profit Factor.

        Profit Factor = Sum of Gains / |Sum of Losses|

        A value greater than 1 indicates profitable trading.
        Higher values indicate more profitable strategies.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Profit factor(s)
        """
        returns = data.pct_change().dropna()

        if isinstance(returns, pd.DataFrame):
            profit_factors = {}
            for col in returns.columns:
                col_returns = returns[col]
                gains = col_returns[col_returns > 0].sum()
                losses = abs(col_returns[col_returns < 0].sum())
                if losses == 0:
                    profit_factors[col] = np.inf if gains > 0 else 0.0
                else:
                    profit_factors[col] = gains / losses
            return pd.Series(profit_factors)
        else:
            gains = returns[returns > 0].sum()
            losses = abs(returns[returns < 0].sum())
            if losses == 0:
                return np.inf if gains > 0 else 0.0
            return gains / losses

    @staticmethod
    def calculate_payoff_ratio(
        data: Union[pd.Series, pd.DataFrame],
    ) -> Union[float, pd.Series]:
        """
        Calculate Payoff Ratio (Average Win / Average Loss).

        Also known as the Risk/Reward ratio. Higher values indicate
        that winning trades are larger than losing trades on average.

        Parameters
        ----------
        data : pd.Series or pd.DataFrame
            Price data with datetime index

        Returns
        -------
        float or pd.Series
            Payoff ratio(s)
        """
        returns = data.pct_change().dropna()

        if isinstance(returns, pd.DataFrame):
            payoff_ratios = {}
            for col in returns.columns:
                col_returns = returns[col]
                avg_win = col_returns[col_returns > 0].mean()
                avg_loss = abs(col_returns[col_returns < 0].mean())
                if avg_loss == 0 or np.isnan(avg_loss):
                    payoff_ratios[col] = np.inf if avg_win > 0 else 0.0
                elif np.isnan(avg_win):
                    payoff_ratios[col] = 0.0
                else:
                    payoff_ratios[col] = avg_win / avg_loss
            return pd.Series(payoff_ratios)
        else:
            avg_win = returns[returns > 0].mean()
            avg_loss = abs(returns[returns < 0].mean())
            if avg_loss == 0 or np.isnan(avg_loss):
                return np.inf if avg_win > 0 else 0.0
            if np.isnan(avg_win):
                return 0.0
            return avg_win / avg_loss

    @staticmethod
    def calculate_herfindahl_index(weights: list[float]) -> float:
        """
        Calculate Herfindahl-Hirschman Index (HHI) for portfolio concentration.

        HHI = sum(w_i^2) for all weights
        - HHI = 1.0 means full concentration in one asset
        - HHI = 1/n means equal weighting across n assets

        Parameters
        ----------
        weights : list of float
            Portfolio weights

        Returns
        -------
        float
            HHI value between 0 and 1
        """
        weights = np.array(weights)
        return float(np.sum(weights**2))

    @staticmethod
    def calculate_effective_n(weights: list[float]) -> float:
        """
        Calculate Effective N (effective number of assets).

        Effective N = 1 / HHI

        Represents how many equal-weighted assets would produce
        the same concentration level.

        Parameters
        ----------
        weights : list of float
            Portfolio weights

        Returns
        -------
        float
            Effective number of assets
        """
        hhi = PerformanceMetrics.calculate_herfindahl_index(weights)
        if hhi == 0:
            return float("inf")
        return 1.0 / hhi

calculate_annual_return(data) staticmethod

Calculate annualized return from price data.

Uses year-end prices to calculate annual returns, then averages.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Annualized return(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_annual_return(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate annualized return from price data.

    Uses year-end prices to calculate annual returns, then averages.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Annualized return(s)
    """
    annual_return = data.resample("Y").last().pct_change().mean()
    return annual_return

calculate_cagr(data) staticmethod

Calculate Compound Annual Growth Rate (CAGR).

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	CAGR value(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_cagr(data: Union[pd.Series, pd.DataFrame]) -> Union[float, pd.Series]:
    """
    Calculate Compound Annual Growth Rate (CAGR).

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        CAGR value(s)
    """
    first_value = data.iloc[0]
    last_value = data.iloc[-1]
    years = (data.index[-1] - data.index[0]).days / DAYS_PER_YEAR

    cagr = (last_value / first_value) ** (1 / years) - 1
    return cagr

calculate_annual_volatility(data) staticmethod

Calculate annualized volatility (standard deviation of returns).

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Annualized volatility

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_annual_volatility(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate annualized volatility (standard deviation of returns).

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Annualized volatility
    """
    daily_returns = data.pct_change().dropna()
    annual_volatility = daily_returns.std() * np.sqrt(
        PerformanceMetrics.TRADING_DAYS
    )
    return annual_volatility

calculate_sharpe_ratio(data, risk_free_rate=DEFAULT_RISK_FREE_RATE) staticmethod

Calculate Sharpe ratio.

Sharpe Ratio = (Return - Risk Free Rate) / Volatility

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required
risk_free_rate	float	Annual risk-free rate	0.02

Returns:

Type	Description
float or Series	Sharpe ratio(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_sharpe_ratio(
    data: Union[pd.Series, pd.DataFrame],
    risk_free_rate: float = DEFAULT_RISK_FREE_RATE,
) -> Union[float, pd.Series]:
    """
    Calculate Sharpe ratio.

    Sharpe Ratio = (Return - Risk Free Rate) / Volatility

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index
    risk_free_rate : float, default 0.02
        Annual risk-free rate

    Returns
    -------
    float or pd.Series
        Sharpe ratio(s)
    """
    annual_return = PerformanceMetrics.calculate_annual_return(data)
    annual_volatility = PerformanceMetrics.calculate_annual_volatility(data)
    sharpe_ratio = (annual_return - risk_free_rate) / annual_volatility
    return sharpe_ratio

calculate_sortino_ratio(data, risk_free_rate=DEFAULT_RISK_FREE_RATE) staticmethod

Calculate Sortino ratio (uses downside deviation instead of volatility).

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required
risk_free_rate	float	Annual risk-free rate	0.02

Returns:

Type	Description
float or Series	Sortino ratio(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_sortino_ratio(
    data: Union[pd.Series, pd.DataFrame],
    risk_free_rate: float = DEFAULT_RISK_FREE_RATE,
) -> Union[float, pd.Series]:
    """
    Calculate Sortino ratio (uses downside deviation instead of volatility).

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index
    risk_free_rate : float, default 0.02
        Annual risk-free rate

    Returns
    -------
    float or pd.Series
        Sortino ratio(s)
    """
    returns = data.pct_change().dropna()

    # Downside deviation (only negative returns, positive returns set to 0)
    # Using .where() for both Series and DataFrame ensures consistent behavior:
    # positive returns are replaced with 0, preserving the sample size
    downside_returns = returns.where(returns < 0, 0)

    downside_deviation = downside_returns.std() * np.sqrt(
        PerformanceMetrics.TRADING_DAYS
    )
    annual_return = PerformanceMetrics.calculate_annual_return(data)

    sortino_ratio = (annual_return - risk_free_rate) / downside_deviation
    return sortino_ratio

calculate_max_drawdown(data) staticmethod

Calculate maximum drawdown.

Maximum drawdown is the largest peak-to-trough decline.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Maximum drawdown (negative value)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_max_drawdown(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate maximum drawdown.

    Maximum drawdown is the largest peak-to-trough decline.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Maximum drawdown (negative value)
    """
    cumulative_returns = (1 + data.pct_change()).cumprod()
    peak = cumulative_returns.expanding(min_periods=1).max()
    drawdown = (cumulative_returns / peak) - 1
    max_drawdown = drawdown.min()
    return max_drawdown

calculate_var(data, confidence_level=DEFAULT_CONFIDENCE_LEVEL) staticmethod

Calculate Value at Risk (VaR) using historical method.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required
confidence_level	float	Confidence level (e.g., 0.95 for 95%)	0.95

Returns:

Type	Description
float or Series	VaR value (typically negative)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_var(
    data: Union[pd.Series, pd.DataFrame],
    confidence_level: float = DEFAULT_CONFIDENCE_LEVEL,
) -> Union[float, pd.Series]:
    """
    Calculate Value at Risk (VaR) using historical method.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index
    confidence_level : float, default 0.95
        Confidence level (e.g., 0.95 for 95%)

    Returns
    -------
    float or pd.Series
        VaR value (typically negative)
    """
    returns = data.pct_change().dropna()
    var = np.percentile(returns, (1 - confidence_level) * 100, axis=0)

    if isinstance(returns, pd.DataFrame):
        return pd.Series(var, index=returns.columns)
    return var

calculate_calmar_ratio(data) staticmethod

Calculate Calmar ratio (CAGR / Max Drawdown).

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Calmar ratio(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_calmar_ratio(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate Calmar ratio (CAGR / Max Drawdown).

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Calmar ratio(s)
    """
    cagr = PerformanceMetrics.calculate_cagr(data)
    max_dd = PerformanceMetrics.calculate_max_drawdown(data)

    # Max drawdown is negative, so we negate it
    return cagr / abs(max_dd)

calculate_cvar(data, confidence_level=DEFAULT_CONFIDENCE_LEVEL) staticmethod

Calculate Conditional Value at Risk (CVaR), also known as Expected Shortfall.

CVaR represents the expected loss given that the loss exceeds VaR. It is a more conservative risk measure than VaR as it considers the tail of the distribution.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required
confidence_level	float	Confidence level (e.g., 0.95 for 95%)	0.95

Returns:

Type	Description
float or Series	CVaR value (typically negative)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_cvar(
    data: Union[pd.Series, pd.DataFrame],
    confidence_level: float = DEFAULT_CONFIDENCE_LEVEL,
) -> Union[float, pd.Series]:
    """
    Calculate Conditional Value at Risk (CVaR), also known as Expected Shortfall.

    CVaR represents the expected loss given that the loss exceeds VaR.
    It is a more conservative risk measure than VaR as it considers
    the tail of the distribution.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index
    confidence_level : float, default 0.95
        Confidence level (e.g., 0.95 for 95%)

    Returns
    -------
    float or pd.Series
        CVaR value (typically negative)
    """
    returns = data.pct_change().dropna()
    var = PerformanceMetrics.calculate_var(data, confidence_level)

    if isinstance(returns, pd.DataFrame):
        cvar_values = {}
        for col in returns.columns:
            col_returns = returns[col]
            col_var = var[col]
            cvar_values[col] = col_returns[col_returns <= col_var].mean()
        return pd.Series(cvar_values)
    else:
        return returns[returns <= var].mean()

calculate_omega_ratio(data, threshold=0.0) staticmethod

Calculate Omega ratio.

The Omega ratio compares the probability-weighted gains above a threshold to the probability-weighted losses below it. Higher values indicate better risk-adjusted performance.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required
threshold	float	Daily return threshold (0.0 for break-even)	0.0

Returns:

Type	Description
float or Series	Omega ratio(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_omega_ratio(
    data: Union[pd.Series, pd.DataFrame],
    threshold: float = 0.0,
) -> Union[float, pd.Series]:
    """
    Calculate Omega ratio.

    The Omega ratio compares the probability-weighted gains above a threshold
    to the probability-weighted losses below it. Higher values indicate better
    risk-adjusted performance.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index
    threshold : float, default 0.0
        Daily return threshold (0.0 for break-even)

    Returns
    -------
    float or pd.Series
        Omega ratio(s)
    """
    returns = data.pct_change().dropna()

    if isinstance(returns, pd.DataFrame):
        omega_values = {}
        for col in returns.columns:
            col_returns = returns[col]
            gains = col_returns[col_returns > threshold] - threshold
            losses = threshold - col_returns[col_returns <= threshold]
            if losses.sum() == 0:
                omega_values[col] = np.inf
            else:
                omega_values[col] = gains.sum() / losses.sum()
        return pd.Series(omega_values)
    else:
        gains = returns[returns > threshold] - threshold
        losses = threshold - returns[returns <= threshold]
        if losses.sum() == 0:
            return np.inf
        return gains.sum() / losses.sum()

calculate_ulcer_index(data) staticmethod

Calculate Ulcer Index.

The Ulcer Index measures downside volatility based on drawdowns. It penalizes deep and prolonged drawdowns more heavily than standard deviation. Lower values indicate less risk.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Ulcer Index value(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_ulcer_index(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate Ulcer Index.

    The Ulcer Index measures downside volatility based on drawdowns.
    It penalizes deep and prolonged drawdowns more heavily than
    standard deviation. Lower values indicate less risk.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Ulcer Index value(s)
    """
    cumulative_returns = (1 + data.pct_change()).cumprod()
    peak = cumulative_returns.expanding(min_periods=1).max()
    drawdown = (cumulative_returns / peak) - 1

    # Ulcer Index is the quadratic mean of drawdowns
    ulcer_index = np.sqrt((drawdown**2).mean())
    return ulcer_index

calculate_recovery_factor(data) staticmethod

Calculate Recovery Factor.

Recovery Factor = Total Return / |Max Drawdown|

A higher recovery factor indicates the portfolio generates more return per unit of maximum drawdown risk.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Recovery factor(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_recovery_factor(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate Recovery Factor.

    Recovery Factor = Total Return / |Max Drawdown|

    A higher recovery factor indicates the portfolio generates more
    return per unit of maximum drawdown risk.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Recovery factor(s)
    """
    total_return = (data.iloc[-1] / data.iloc[0]) - 1
    max_dd = PerformanceMetrics.calculate_max_drawdown(data)

    return total_return / abs(max_dd)

calculate_win_rate(data) staticmethod

Calculate Win Rate (percentage of positive return periods).

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Win rate as a decimal (e.g., 0.55 for 55% win rate)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_win_rate(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate Win Rate (percentage of positive return periods).

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Win rate as a decimal (e.g., 0.55 for 55% win rate)
    """
    returns = data.pct_change().dropna()
    win_rate = (returns > 0).mean()
    return win_rate

calculate_profit_factor(data) staticmethod

Calculate Profit Factor.

Profit Factor = Sum of Gains / |Sum of Losses|

A value greater than 1 indicates profitable trading. Higher values indicate more profitable strategies.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Profit factor(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_profit_factor(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate Profit Factor.

    Profit Factor = Sum of Gains / |Sum of Losses|

    A value greater than 1 indicates profitable trading.
    Higher values indicate more profitable strategies.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Profit factor(s)
    """
    returns = data.pct_change().dropna()

    if isinstance(returns, pd.DataFrame):
        profit_factors = {}
        for col in returns.columns:
            col_returns = returns[col]
            gains = col_returns[col_returns > 0].sum()
            losses = abs(col_returns[col_returns < 0].sum())
            if losses == 0:
                profit_factors[col] = np.inf if gains > 0 else 0.0
            else:
                profit_factors[col] = gains / losses
        return pd.Series(profit_factors)
    else:
        gains = returns[returns > 0].sum()
        losses = abs(returns[returns < 0].sum())
        if losses == 0:
            return np.inf if gains > 0 else 0.0
        return gains / losses

calculate_payoff_ratio(data) staticmethod

Calculate Payoff Ratio (Average Win / Average Loss).

Also known as the Risk/Reward ratio. Higher values indicate that winning trades are larger than losing trades on average.

Parameters:

Name	Type	Description	Default
data	Series or DataFrame	Price data with datetime index	required

Returns:

Type	Description
float or Series	Payoff ratio(s)

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_payoff_ratio(
    data: Union[pd.Series, pd.DataFrame],
) -> Union[float, pd.Series]:
    """
    Calculate Payoff Ratio (Average Win / Average Loss).

    Also known as the Risk/Reward ratio. Higher values indicate
    that winning trades are larger than losing trades on average.

    Parameters
    ----------
    data : pd.Series or pd.DataFrame
        Price data with datetime index

    Returns
    -------
    float or pd.Series
        Payoff ratio(s)
    """
    returns = data.pct_change().dropna()

    if isinstance(returns, pd.DataFrame):
        payoff_ratios = {}
        for col in returns.columns:
            col_returns = returns[col]
            avg_win = col_returns[col_returns > 0].mean()
            avg_loss = abs(col_returns[col_returns < 0].mean())
            if avg_loss == 0 or np.isnan(avg_loss):
                payoff_ratios[col] = np.inf if avg_win > 0 else 0.0
            elif np.isnan(avg_win):
                payoff_ratios[col] = 0.0
            else:
                payoff_ratios[col] = avg_win / avg_loss
        return pd.Series(payoff_ratios)
    else:
        avg_win = returns[returns > 0].mean()
        avg_loss = abs(returns[returns < 0].mean())
        if avg_loss == 0 or np.isnan(avg_loss):
            return np.inf if avg_win > 0 else 0.0
        if np.isnan(avg_win):
            return 0.0
        return avg_win / avg_loss

calculate_herfindahl_index(weights) staticmethod

Calculate Herfindahl-Hirschman Index (HHI) for portfolio concentration.

HHI = sum(w_i^2) for all weights - HHI = 1.0 means full concentration in one asset - HHI = 1/n means equal weighting across n assets

Parameters:

Name	Type	Description	Default
weights	list of float	Portfolio weights	required

Returns:

Type	Description
float	HHI value between 0 and 1

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_herfindahl_index(weights: list[float]) -> float:
    """
    Calculate Herfindahl-Hirschman Index (HHI) for portfolio concentration.

    HHI = sum(w_i^2) for all weights
    - HHI = 1.0 means full concentration in one asset
    - HHI = 1/n means equal weighting across n assets

    Parameters
    ----------
    weights : list of float
        Portfolio weights

    Returns
    -------
    float
        HHI value between 0 and 1
    """
    weights = np.array(weights)
    return float(np.sum(weights**2))

calculate_effective_n(weights) staticmethod

Calculate Effective N (effective number of assets).

Effective N = 1 / HHI

Represents how many equal-weighted assets would produce the same concentration level.

Parameters:

Name	Type	Description	Default
weights	list of float	Portfolio weights	required

Returns:

Type	Description
float	Effective number of assets

Source code in portfolio_analysis/metrics/performance.py

@staticmethod
def calculate_effective_n(weights: list[float]) -> float:
    """
    Calculate Effective N (effective number of assets).

    Effective N = 1 / HHI

    Represents how many equal-weighted assets would produce
    the same concentration level.

    Parameters
    ----------
    weights : list of float
        Portfolio weights

    Returns
    -------
    float
        Effective number of assets
    """
    hhi = PerformanceMetrics.calculate_herfindahl_index(weights)
    if hhi == 0:
        return float("inf")
    return 1.0 / hhi

BenchmarkComparison

Compare portfolio performance against market benchmarks.

Calculates alpha, beta, tracking error, information ratio, and generates comparison reports and visualizations.

Parameters:

Name	Type	Description	Default
portfolio_data	DataFrame	Historical price data for portfolio assets	required
weights	array - like	Portfolio weights (must sum to 1.0)	required
benchmark_ticker	str	Ticker symbol for benchmark	'SPY'
risk_free_rate	float	Annual risk-free rate for calculations	0.02

Examples:

>>> comparison = BenchmarkComparison(data, weights, benchmark_ticker='SPY')
>>> comparison.generate_report()
>>> comparison.plot_cumulative_returns()

Source code in portfolio_analysis/metrics/benchmark.py

class BenchmarkComparison:
    """
    Compare portfolio performance against market benchmarks.

    Calculates alpha, beta, tracking error, information ratio, and
    generates comparison reports and visualizations.

    Parameters
    ----------
    portfolio_data : pd.DataFrame
        Historical price data for portfolio assets
    weights : array-like
        Portfolio weights (must sum to 1.0)
    benchmark_ticker : str, default 'SPY'
        Ticker symbol for benchmark
    risk_free_rate : float, default 0.02
        Annual risk-free rate for calculations

    Examples
    --------
    >>> comparison = BenchmarkComparison(data, weights, benchmark_ticker='SPY')
    >>> comparison.generate_report()
    >>> comparison.plot_cumulative_returns()
    """

    BENCHMARKS = {
        "SPY": "S&P 500 (US Large Cap)",
        "VTI": "Total US Stock Market",
        "BND": "Total US Bond Market",
        "VT": "Total World Stock Market",
        "QQQ": "NASDAQ 100",
        "IWM": "Russell 2000 (US Small Cap)",
        "EFA": "Developed Markets ex-US",
        "AGG": "US Aggregate Bond",
    }

    def __init__(
        self,
        portfolio_data: pd.DataFrame,
        weights: list[float],
        benchmark_ticker: str = "SPY",
        risk_free_rate: float = DEFAULT_RISK_FREE_RATE,
    ):
        self.portfolio_data = portfolio_data
        self.weights = np.array(weights)
        self.benchmark_ticker = benchmark_ticker
        self.risk_free_rate = risk_free_rate

        # Validate weights
        self._validate_weights()

        # Calculate portfolio returns
        returns = portfolio_data.pct_change().dropna()
        self.portfolio_returns = returns.dot(self.weights)

    def _validate_weights(self) -> None:
        """Validate that weights are valid for portfolio calculations."""
        if len(self.weights) != len(self.portfolio_data.columns):
            raise ValidationError(
                f"Number of weights ({len(self.weights)}) must match "
                f"number of assets ({len(self.portfolio_data.columns)})"
            )

        weight_sum = np.sum(self.weights)
        if not np.isclose(weight_sum, 1.0, atol=WEIGHT_SUM_TOLERANCE):
            raise ValidationError(f"Weights must sum to 1.0, got {weight_sum:.6f}")

        # Fetch benchmark data
        self.benchmark_data = self._fetch_benchmark()
        self.benchmark_returns = self.benchmark_data.pct_change().dropna()

        # Align dates
        common_dates = self.portfolio_returns.index.intersection(
            self.benchmark_returns.index
        )
        self.portfolio_returns = self.portfolio_returns.loc[common_dates]
        self.benchmark_returns = self.benchmark_returns.loc[common_dates]

    def _fetch_benchmark(self) -> pd.Series:
        """Fetch benchmark price data."""
        start_date = self.portfolio_data.index.min()
        end_date = self.portfolio_data.index.max()

        raw_data = yf.download(
            self.benchmark_ticker, start=start_date, end=end_date, progress=False
        )

        # Handle yfinance column format changes across versions
        # Check for MultiIndex columns (yfinance >= 0.2.40)
        if isinstance(raw_data.columns, pd.MultiIndex):
            price_types = raw_data.columns.get_level_values(0).unique()
            if "Adj Close" in price_types:
                benchmark = raw_data["Adj Close"]
            elif "Close" in price_types:
                benchmark = raw_data["Close"]
            else:
                raise ValueError(
                    f"No Close or Adj Close column found. Available: {price_types.tolist()}"
                )
        else:
            if "Adj Close" in raw_data.columns:
                benchmark = raw_data["Adj Close"]
            elif "Close" in raw_data.columns:
                benchmark = raw_data["Close"]
            else:
                raise ValueError(
                    f"No Close or Adj Close column found. Available: {raw_data.columns.tolist()}"
                )

        # Ensure we return a Series, not a DataFrame
        if isinstance(benchmark, pd.DataFrame):
            benchmark = benchmark.squeeze()

        return benchmark

    def calculate_beta(self) -> float:
        """Calculate portfolio beta relative to benchmark."""
        covariance = np.cov(self.portfolio_returns, self.benchmark_returns)[0, 1]
        benchmark_variance = np.var(self.benchmark_returns)
        return covariance / benchmark_variance

    def calculate_alpha(self, annualized: bool = True) -> float:
        """Calculate Jensen's alpha (CAPM alpha)."""
        beta = self.calculate_beta()

        portfolio_mean = self.portfolio_returns.mean()
        benchmark_mean = self.benchmark_returns.mean()
        rf_daily = self.risk_free_rate / TRADING_DAYS_PER_YEAR

        alpha = portfolio_mean - (rf_daily + beta * (benchmark_mean - rf_daily))

        if annualized:
            alpha = alpha * TRADING_DAYS_PER_YEAR

        return alpha

    def calculate_tracking_error(self, annualized: bool = True) -> float:
        """Calculate tracking error (active risk)."""
        active_returns = self.portfolio_returns - self.benchmark_returns
        tracking_error = active_returns.std()

        if annualized:
            tracking_error = tracking_error * np.sqrt(TRADING_DAYS_PER_YEAR)

        return tracking_error

    def calculate_information_ratio(self) -> float:
        """Calculate information ratio."""
        active_return = (
            self.portfolio_returns.mean() - self.benchmark_returns.mean()
        ) * TRADING_DAYS_PER_YEAR
        tracking_error = self.calculate_tracking_error(annualized=True)

        if tracking_error == 0:
            return np.inf if active_return > 0 else -np.inf

        return active_return / tracking_error

    def calculate_correlation(self) -> float:
        """Calculate correlation with benchmark."""
        return np.corrcoef(self.portfolio_returns, self.benchmark_returns)[0, 1]

    def calculate_r_squared(self) -> float:
        """Calculate R-squared."""
        return self.calculate_correlation() ** 2

    def calculate_up_capture(self) -> float:
        """Calculate upside capture ratio."""
        up_mask = self.benchmark_returns > 0
        if up_mask.sum() == 0:
            return np.nan

        portfolio_up = self.portfolio_returns[up_mask].mean()
        benchmark_up = self.benchmark_returns[up_mask].mean()

        return (portfolio_up / benchmark_up) * 100

    def calculate_down_capture(self) -> float:
        """Calculate downside capture ratio."""
        down_mask = self.benchmark_returns < 0
        if down_mask.sum() == 0:
            return np.nan

        portfolio_down = self.portfolio_returns[down_mask].mean()
        benchmark_down = self.benchmark_returns[down_mask].mean()

        return (portfolio_down / benchmark_down) * 100

    def get_metrics(self) -> dict:
        """Get all benchmark comparison metrics as a dictionary."""
        return {
            "beta": self.calculate_beta(),
            "alpha": self.calculate_alpha(),
            "tracking_error": self.calculate_tracking_error(),
            "information_ratio": self.calculate_information_ratio(),
            "correlation": self.calculate_correlation(),
            "r_squared": self.calculate_r_squared(),
            "up_capture": self.calculate_up_capture(),
            "down_capture": self.calculate_down_capture(),
            "portfolio_return": self.portfolio_returns.mean() * TRADING_DAYS_PER_YEAR,
            "benchmark_return": self.benchmark_returns.mean() * TRADING_DAYS_PER_YEAR,
            "portfolio_volatility": self.portfolio_returns.std()
            * np.sqrt(TRADING_DAYS_PER_YEAR),
            "benchmark_volatility": self.benchmark_returns.std()
            * np.sqrt(TRADING_DAYS_PER_YEAR),
        }

    def generate_report(self) -> None:
        """Print a comprehensive comparison report."""
        metrics = self.get_metrics()

        print("\n" + "=" * 60)
        print("BENCHMARK COMPARISON REPORT")
        print("=" * 60)
        print(f"Benchmark: {self.benchmark_ticker}", end="")
        if self.benchmark_ticker in self.BENCHMARKS:
            print(f" ({self.BENCHMARKS[self.benchmark_ticker]})")
        else:
            print()
        print(
            f"Period: {self.portfolio_returns.index.min().strftime('%Y-%m-%d')} to "
            f"{self.portfolio_returns.index.max().strftime('%Y-%m-%d')}"
        )
        print("-" * 60)

        print("\nAnnualized Returns:")
        print(f"  Portfolio:  {metrics['portfolio_return']*100:.2f}%")
        print(f"  Benchmark:  {metrics['benchmark_return']*100:.2f}%")
        print(
            f"  Difference: {(metrics['portfolio_return']-metrics['benchmark_return'])*100:+.2f}%"
        )

        print("\nAnnualized Volatility:")
        print(f"  Portfolio:  {metrics['portfolio_volatility']*100:.2f}%")
        print(f"  Benchmark:  {metrics['benchmark_volatility']*100:.2f}%")

        print("\nRisk Metrics:")
        print(f"  Beta:              {metrics['beta']:.3f}")
        print(f"  Alpha (annual):    {metrics['alpha']*100:.2f}%")
        print(f"  R-squared:         {metrics['r_squared']:.3f}")
        print(f"  Correlation:       {metrics['correlation']:.3f}")

        print("\nPerformance Metrics:")
        print(f"  Tracking Error:    {metrics['tracking_error']*100:.2f}%")
        print(f"  Information Ratio: {metrics['information_ratio']:.3f}")
        print(f"  Up Capture:        {metrics['up_capture']:.1f}%")
        print(f"  Down Capture:      {metrics['down_capture']:.1f}%")
        print("=" * 60)

    def plot_cumulative_returns(
        self, initial_value: float = 10000, show: bool = True
    ) -> plt.Figure:
        """
        Plot cumulative returns comparison.

        Parameters
        ----------
        initial_value : float, default 10000
            Starting portfolio value for visualization
        show : bool, default True
            Whether to display the plot. Set to False for automated/server contexts.

        Returns
        -------
        plt.Figure
            The matplotlib figure object
        """
        portfolio_cum = (1 + self.portfolio_returns).cumprod() * initial_value
        benchmark_cum = (1 + self.benchmark_returns).cumprod() * initial_value

        fig, ax = plt.subplots(figsize=(12, 6))

        ax.plot(
            portfolio_cum.index,
            portfolio_cum.values,
            label="Portfolio",
            linewidth=2,
            color="blue",
        )
        ax.plot(
            benchmark_cum.index,
            benchmark_cum.values,
            label=f"Benchmark ({self.benchmark_ticker})",
            linewidth=2,
            color="orange",
        )

        ax.set_title("Cumulative Returns: Portfolio vs Benchmark")
        ax.set_xlabel("Date")
        ax.set_ylabel(f"Value (starting from ${initial_value:,.0f})")
        ax.legend()
        ax.grid(True, alpha=0.3)
        fig.tight_layout()

        if show:
            plt.show()

        return fig

    def plot_rolling_metrics(
        self, window: int = TRADING_DAYS_PER_YEAR, show: bool = True
    ) -> plt.Figure:
        """
        Plot rolling alpha and beta.

        Parameters
        ----------
        window : int, default 252
            Rolling window size in trading days
        show : bool, default True
            Whether to display the plot. Set to False for automated/server contexts.

        Returns
        -------
        plt.Figure
            The matplotlib figure object
        """
        fig, axes = plt.subplots(2, 1, figsize=(12, 8), sharex=True)

        rolling_beta = []
        rolling_alpha = []
        dates = []

        for i in range(window, len(self.portfolio_returns)):
            port_window = self.portfolio_returns.iloc[i - window : i]
            bench_window = self.benchmark_returns.iloc[i - window : i]

            cov = np.cov(port_window, bench_window)[0, 1]
            var = np.var(bench_window)
            beta = cov / var

            rf_daily = self.risk_free_rate / TRADING_DAYS_PER_YEAR
            alpha = (
                port_window.mean()
                - (rf_daily + beta * (bench_window.mean() - rf_daily))
            ) * TRADING_DAYS_PER_YEAR

            rolling_beta.append(beta)
            rolling_alpha.append(alpha)
            dates.append(self.portfolio_returns.index[i])

        axes[0].plot(dates, rolling_beta, linewidth=1.5, color="blue")
        axes[0].axhline(
            y=1.0, color="red", linestyle="--", linewidth=1, label="Beta = 1.0"
        )
        axes[0].set_ylabel("Beta")
        axes[0].set_title(f"Rolling {window}-Day Beta and Alpha")
        axes[0].legend()
        axes[0].grid(True, alpha=0.3)

        axes[1].plot(
            dates, [a * 100 for a in rolling_alpha], linewidth=1.5, color="green"
        )
        axes[1].axhline(
            y=0, color="red", linestyle="--", linewidth=1, label="Alpha = 0"
        )
        axes[1].set_ylabel("Alpha (%)")
        axes[1].set_xlabel("Date")
        axes[1].legend()
        axes[1].grid(True, alpha=0.3)

        fig.tight_layout()

        if show:
            plt.show()

        return fig

calculate_beta()

Calculate portfolio beta relative to benchmark.

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_beta(self) -> float:
    """Calculate portfolio beta relative to benchmark."""
    covariance = np.cov(self.portfolio_returns, self.benchmark_returns)[0, 1]
    benchmark_variance = np.var(self.benchmark_returns)
    return covariance / benchmark_variance

calculate_alpha(annualized=True)

Calculate Jensen's alpha (CAPM alpha).

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_alpha(self, annualized: bool = True) -> float:
    """Calculate Jensen's alpha (CAPM alpha)."""
    beta = self.calculate_beta()

    portfolio_mean = self.portfolio_returns.mean()
    benchmark_mean = self.benchmark_returns.mean()
    rf_daily = self.risk_free_rate / TRADING_DAYS_PER_YEAR

    alpha = portfolio_mean - (rf_daily + beta * (benchmark_mean - rf_daily))

    if annualized:
        alpha = alpha * TRADING_DAYS_PER_YEAR

    return alpha

calculate_tracking_error(annualized=True)

Calculate tracking error (active risk).

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_tracking_error(self, annualized: bool = True) -> float:
    """Calculate tracking error (active risk)."""
    active_returns = self.portfolio_returns - self.benchmark_returns
    tracking_error = active_returns.std()

    if annualized:
        tracking_error = tracking_error * np.sqrt(TRADING_DAYS_PER_YEAR)

    return tracking_error

calculate_information_ratio()

Calculate information ratio.

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_information_ratio(self) -> float:
    """Calculate information ratio."""
    active_return = (
        self.portfolio_returns.mean() - self.benchmark_returns.mean()
    ) * TRADING_DAYS_PER_YEAR
    tracking_error = self.calculate_tracking_error(annualized=True)

    if tracking_error == 0:
        return np.inf if active_return > 0 else -np.inf

    return active_return / tracking_error

calculate_correlation()

Calculate correlation with benchmark.

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_correlation(self) -> float:
    """Calculate correlation with benchmark."""
    return np.corrcoef(self.portfolio_returns, self.benchmark_returns)[0, 1]

calculate_r_squared()

Calculate R-squared.

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_r_squared(self) -> float:
    """Calculate R-squared."""
    return self.calculate_correlation() ** 2

calculate_up_capture()

Calculate upside capture ratio.

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_up_capture(self) -> float:
    """Calculate upside capture ratio."""
    up_mask = self.benchmark_returns > 0
    if up_mask.sum() == 0:
        return np.nan

    portfolio_up = self.portfolio_returns[up_mask].mean()
    benchmark_up = self.benchmark_returns[up_mask].mean()

    return (portfolio_up / benchmark_up) * 100

calculate_down_capture()

Calculate downside capture ratio.

Source code in portfolio_analysis/metrics/benchmark.py

def calculate_down_capture(self) -> float:
    """Calculate downside capture ratio."""
    down_mask = self.benchmark_returns < 0
    if down_mask.sum() == 0:
        return np.nan

    portfolio_down = self.portfolio_returns[down_mask].mean()
    benchmark_down = self.benchmark_returns[down_mask].mean()

    return (portfolio_down / benchmark_down) * 100

get_metrics()

Get all benchmark comparison metrics as a dictionary.

Source code in portfolio_analysis/metrics/benchmark.py

def get_metrics(self) -> dict:
    """Get all benchmark comparison metrics as a dictionary."""
    return {
        "beta": self.calculate_beta(),
        "alpha": self.calculate_alpha(),
        "tracking_error": self.calculate_tracking_error(),
        "information_ratio": self.calculate_information_ratio(),
        "correlation": self.calculate_correlation(),
        "r_squared": self.calculate_r_squared(),
        "up_capture": self.calculate_up_capture(),
        "down_capture": self.calculate_down_capture(),
        "portfolio_return": self.portfolio_returns.mean() * TRADING_DAYS_PER_YEAR,
        "benchmark_return": self.benchmark_returns.mean() * TRADING_DAYS_PER_YEAR,
        "portfolio_volatility": self.portfolio_returns.std()
        * np.sqrt(TRADING_DAYS_PER_YEAR),
        "benchmark_volatility": self.benchmark_returns.std()
        * np.sqrt(TRADING_DAYS_PER_YEAR),
    }

generate_report()

Print a comprehensive comparison report.

Source code in portfolio_analysis/metrics/benchmark.py

def generate_report(self) -> None:
    """Print a comprehensive comparison report."""
    metrics = self.get_metrics()

    print("\n" + "=" * 60)
    print("BENCHMARK COMPARISON REPORT")
    print("=" * 60)
    print(f"Benchmark: {self.benchmark_ticker}", end="")
    if self.benchmark_ticker in self.BENCHMARKS:
        print(f" ({self.BENCHMARKS[self.benchmark_ticker]})")
    else:
        print()
    print(
        f"Period: {self.portfolio_returns.index.min().strftime('%Y-%m-%d')} to "
        f"{self.portfolio_returns.index.max().strftime('%Y-%m-%d')}"
    )
    print("-" * 60)

    print("\nAnnualized Returns:")
    print(f"  Portfolio:  {metrics['portfolio_return']*100:.2f}%")
    print(f"  Benchmark:  {metrics['benchmark_return']*100:.2f}%")
    print(
        f"  Difference: {(metrics['portfolio_return']-metrics['benchmark_return'])*100:+.2f}%"
    )

    print("\nAnnualized Volatility:")
    print(f"  Portfolio:  {metrics['portfolio_volatility']*100:.2f}%")
    print(f"  Benchmark:  {metrics['benchmark_volatility']*100:.2f}%")

    print("\nRisk Metrics:")
    print(f"  Beta:              {metrics['beta']:.3f}")
    print(f"  Alpha (annual):    {metrics['alpha']*100:.2f}%")
    print(f"  R-squared:         {metrics['r_squared']:.3f}")
    print(f"  Correlation:       {metrics['correlation']:.3f}")

    print("\nPerformance Metrics:")
    print(f"  Tracking Error:    {metrics['tracking_error']*100:.2f}%")
    print(f"  Information Ratio: {metrics['information_ratio']:.3f}")
    print(f"  Up Capture:        {metrics['up_capture']:.1f}%")
    print(f"  Down Capture:      {metrics['down_capture']:.1f}%")
    print("=" * 60)

plot_cumulative_returns(initial_value=10000, show=True)

Plot cumulative returns comparison.

Parameters:

Name	Type	Description	Default
initial_value	float	Starting portfolio value for visualization	10000
show	bool	Whether to display the plot. Set to False for automated/server contexts.	True

Returns:

Type	Description
Figure	The matplotlib figure object

Source code in portfolio_analysis/metrics/benchmark.py

def plot_cumulative_returns(
    self, initial_value: float = 10000, show: bool = True
) -> plt.Figure:
    """
    Plot cumulative returns comparison.

    Parameters
    ----------
    initial_value : float, default 10000
        Starting portfolio value for visualization
    show : bool, default True
        Whether to display the plot. Set to False for automated/server contexts.

    Returns
    -------
    plt.Figure
        The matplotlib figure object
    """
    portfolio_cum = (1 + self.portfolio_returns).cumprod() * initial_value
    benchmark_cum = (1 + self.benchmark_returns).cumprod() * initial_value

    fig, ax = plt.subplots(figsize=(12, 6))

    ax.plot(
        portfolio_cum.index,
        portfolio_cum.values,
        label="Portfolio",
        linewidth=2,
        color="blue",
    )
    ax.plot(
        benchmark_cum.index,
        benchmark_cum.values,
        label=f"Benchmark ({self.benchmark_ticker})",
        linewidth=2,
        color="orange",
    )

    ax.set_title("Cumulative Returns: Portfolio vs Benchmark")
    ax.set_xlabel("Date")
    ax.set_ylabel(f"Value (starting from ${initial_value:,.0f})")
    ax.legend()
    ax.grid(True, alpha=0.3)
    fig.tight_layout()

    if show:
        plt.show()

    return fig

plot_rolling_metrics(window=TRADING_DAYS_PER_YEAR, show=True)

Plot rolling alpha and beta.

Parameters:

Name	Type	Description	Default
window	int	Rolling window size in trading days	252
show	bool	Whether to display the plot. Set to False for automated/server contexts.	True

Returns:

Type	Description
Figure	The matplotlib figure object

Source code in portfolio_analysis/metrics/benchmark.py

def plot_rolling_metrics(
    self, window: int = TRADING_DAYS_PER_YEAR, show: bool = True
) -> plt.Figure:
    """
    Plot rolling alpha and beta.

    Parameters
    ----------
    window : int, default 252
        Rolling window size in trading days
    show : bool, default True
        Whether to display the plot. Set to False for automated/server contexts.

    Returns
    -------
    plt.Figure
        The matplotlib figure object
    """
    fig, axes = plt.subplots(2, 1, figsize=(12, 8), sharex=True)

    rolling_beta = []
    rolling_alpha = []
    dates = []

    for i in range(window, len(self.portfolio_returns)):
        port_window = self.portfolio_returns.iloc[i - window : i]
        bench_window = self.benchmark_returns.iloc[i - window : i]

        cov = np.cov(port_window, bench_window)[0, 1]
        var = np.var(bench_window)
        beta = cov / var

        rf_daily = self.risk_free_rate / TRADING_DAYS_PER_YEAR
        alpha = (
            port_window.mean()
            - (rf_daily + beta * (bench_window.mean() - rf_daily))
        ) * TRADING_DAYS_PER_YEAR

        rolling_beta.append(beta)
        rolling_alpha.append(alpha)
        dates.append(self.portfolio_returns.index[i])

    axes[0].plot(dates, rolling_beta, linewidth=1.5, color="blue")
    axes[0].axhline(
        y=1.0, color="red", linestyle="--", linewidth=1, label="Beta = 1.0"
    )
    axes[0].set_ylabel("Beta")
    axes[0].set_title(f"Rolling {window}-Day Beta and Alpha")
    axes[0].legend()
    axes[0].grid(True, alpha=0.3)

    axes[1].plot(
        dates, [a * 100 for a in rolling_alpha], linewidth=1.5, color="green"
    )
    axes[1].axhline(
        y=0, color="red", linestyle="--", linewidth=1, label="Alpha = 0"
    )
    axes[1].set_ylabel("Alpha (%)")
    axes[1].set_xlabel("Date")
    axes[1].legend()
    axes[1].grid(True, alpha=0.3)

    fig.tight_layout()

    if show:
        plt.show()

    return fig