rolling

rolling_metric / expanding_metric are generic engines - they roll any PortPy metric function (or your own) over a window. The rest are convenience wrappers over the same engine.

rolling

Rolling and expanding-window metric engines.

rolling_metric / expanding_metric are the generic engines the more specific helpers below are built on - use them directly to roll any PortPy metric function (or your own) over a window.

rolling_metric

rolling_metric(
    returns: Series,
    func: Callable[..., float],
    window: int,
    min_periods: int | None = None,
    **kwargs: Any,
) -> pd.Series

Apply any Series -> float metric function over a rolling window.

Parameters:

Name	Type	Description	Default
returns	Series	Per-period returns.	required
func	Callable[..., float]	A callable taking a pd.Series window (plus **kwargs) and returning a float - any PortPy metric function works directly (pass as_result=False, the default).	required
window	int	Window size, in periods.	required
min_periods	int | None	Minimum observations required to produce a value. Defaults to window (no partial windows).	None
**kwargs	Any	Additional keyword arguments passed through to func.	{}

Returns:

Type	Description
Series	A Series of func evaluated on each rolling window, aligned to returns.

Source code in src/portpy/metrics/rolling.py

def rolling_metric(
    returns: pd.Series,
    func: Callable[..., float],
    window: int,
    min_periods: int | None = None,
    **kwargs: Any,
) -> pd.Series:
    """
    Apply any Series -> float metric function over a rolling window.

    Args:
        returns: Per-period returns.
        func: A callable taking a pd.Series window (plus **kwargs) and
            returning a float - any PortPy metric function works directly
            (pass as_result=False, the default).
        window: Window size, in periods.
        min_periods: Minimum observations required to produce a value.
            Defaults to `window` (no partial windows).
        **kwargs: Additional keyword arguments passed through to `func`.

    Returns:
        A Series of `func` evaluated on each rolling window, aligned to `returns`.
    """
    return returns.rolling(window, min_periods=min_periods or window).apply(
        lambda x: func(pd.Series(x, copy=False), **kwargs), raw=False
    )

rolling_sharpe

rolling_sharpe(
    returns: Series,
    window: int,
    rf: float = DEFAULT_RISK_FREE_RATE,
    periods_per_year: int = TRADING_DAYS_PER_YEAR,
) -> pd.Series

Sharpe ratio recomputed over a trailing window.

Shows whether risk-adjusted performance is stable or decaying.

Parameters:

Name	Type	Description	Default
returns	Series	Per-period returns.	required
window	int	Window size, in periods.	required
rf	float	Risk-free rate, as an annual rate. Defaults to DEFAULT_RISK_FREE_RATE.	DEFAULT_RISK_FREE_RATE
periods_per_year	int	Number of periods in a year, used to annualize.	TRADING_DAYS_PER_YEAR

Returns:

Type	Description
Series	A Series of rolling Sharpe ratios, named "rolling_sharpe".

Source code in src/portpy/metrics/rolling.py

def rolling_sharpe(
    returns: pd.Series,
    window: int,
    rf: float = DEFAULT_RISK_FREE_RATE,
    periods_per_year: int = TRADING_DAYS_PER_YEAR,
) -> pd.Series:
    """
    Sharpe ratio recomputed over a trailing window.

    Shows whether risk-adjusted performance is stable or decaying.

    Args:
        returns: Per-period returns.
        window: Window size, in periods.
        rf: Risk-free rate, as an annual rate. Defaults to DEFAULT_RISK_FREE_RATE.
        periods_per_year: Number of periods in a year, used to annualize.

    Returns:
        A Series of rolling Sharpe ratios, named "rolling_sharpe".
    """
    from portpy.metrics.performance import sharpe_ratio

    return rolling_metric(
        returns, sharpe_ratio, window, rf=rf, periods_per_year=periods_per_year, annualized=True
    ).rename("rolling_sharpe")

rolling_volatility

rolling_volatility(
    returns: Series,
    window: int,
    annualized: bool = True,
    periods_per_year: int = TRADING_DAYS_PER_YEAR,
) -> pd.Series

Standard deviation recomputed over a trailing window; reveals volatility clustering and regime shifts.

Parameters:

Name	Type	Description	Default
returns	Series	Per-period returns.	required
window	int	Window size, in periods.	required
annualized	bool	If True (default), scale by sqrt(periods_per_year).	True
periods_per_year	int	Number of periods in a year, used to annualize.	TRADING_DAYS_PER_YEAR

Returns:

Type	Description
Series	A Series of rolling volatility, named "rolling_volatility".

Source code in src/portpy/metrics/rolling.py

def rolling_volatility(
    returns: pd.Series,
    window: int,
    annualized: bool = True,
    periods_per_year: int = TRADING_DAYS_PER_YEAR,
) -> pd.Series:
    """
    Standard deviation recomputed over a trailing window; reveals volatility
    clustering and regime shifts.

    Args:
        returns: Per-period returns.
        window: Window size, in periods.
        annualized: If True (default), scale by sqrt(periods_per_year).
        periods_per_year: Number of periods in a year, used to annualize.

    Returns:
        A Series of rolling volatility, named "rolling_volatility".
    """
    vol = returns.rolling(window, min_periods=window).std(ddof=1)
    if annualized:
        vol = vol * np.sqrt(periods_per_year)
    return vol.rename("rolling_volatility")

rolling_beta

rolling_beta(
    returns: Series, benchmark: Series, window: int
) -> pd.Series

Beta recomputed over a trailing window; reveals whether market sensitivity is stable over time.

Parameters:

Name	Type	Description	Default
returns	Series	Per-period returns.	required
benchmark	Series	Per-period benchmark returns.	required
window	int	Window size, in periods.	required

Returns:

Type	Description
Series	A Series of rolling beta, named "rolling_beta".

Source code in src/portpy/metrics/rolling.py

def rolling_beta(returns: pd.Series, benchmark: pd.Series, window: int) -> pd.Series:
    """
    Beta recomputed over a trailing window; reveals whether market sensitivity
    is stable over time.

    Args:
        returns: Per-period returns.
        benchmark: Per-period benchmark returns.
        window: Window size, in periods.

    Returns:
        A Series of rolling beta, named "rolling_beta".
    """
    r, b = align_pair(returns, benchmark)
    cov = r.rolling(window, min_periods=window).cov(b)
    var = b.rolling(window, min_periods=window).var(ddof=1)
    return (cov / var).rename("rolling_beta")

rolling_correlation

rolling_correlation(
    returns_a: Series, returns_b: Series, window: int
) -> pd.Series

Pearson correlation recomputed over a trailing window.

Parameters:

Name	Type	Description	Default
returns_a	Series	First per-period return series.	required
returns_b	Series	Second per-period return series.	required
window	int	Window size, in periods.	required

Returns:

Type	Description
Series	A Series of rolling correlation, named "rolling_correlation".

Source code in src/portpy/metrics/rolling.py

def rolling_correlation(returns_a: pd.Series, returns_b: pd.Series, window: int) -> pd.Series:
    """
    Pearson correlation recomputed over a trailing window.

    Args:
        returns_a: First per-period return series.
        returns_b: Second per-period return series.
        window: Window size, in periods.

    Returns:
        A Series of rolling correlation, named "rolling_correlation".
    """
    a, b = align_pair(returns_a, returns_b, "series_a", "series_b")
    return a.rolling(window, min_periods=window).corr(b).rename("rolling_correlation")

expanding_metric

expanding_metric(
    returns: Series,
    func: Callable[..., float],
    min_periods: int = 2,
    **kwargs: Any,
) -> pd.Series

Apply any Series -> float metric function over an expanding (growing) window.

Parameters:

Name	Type	Description	Default
returns	Series	Per-period returns.	required
func	Callable[..., float]	A callable taking a pd.Series window (plus **kwargs) and returning a float.	required
min_periods	int	Minimum observations required to produce a value. Defaults to 2.	2
**kwargs	Any	Additional keyword arguments passed through to func.	{}

Returns:

Type	Description
Series	A Series of func evaluated on each expanding window, aligned to returns.

Source code in src/portpy/metrics/rolling.py

def expanding_metric(
    returns: pd.Series,
    func: Callable[..., float],
    min_periods: int = 2,
    **kwargs: Any,
) -> pd.Series:
    """
    Apply any Series -> float metric function over an expanding (growing) window.

    Args:
        returns: Per-period returns.
        func: A callable taking a pd.Series window (plus **kwargs) and
            returning a float.
        min_periods: Minimum observations required to produce a value.
            Defaults to 2.
        **kwargs: Additional keyword arguments passed through to `func`.

    Returns:
        A Series of `func` evaluated on each expanding window, aligned to `returns`.
    """
    return returns.expanding(min_periods=min_periods).apply(
        lambda x: func(pd.Series(x, copy=False), **kwargs), raw=False
    )