API reference¶

The calculations behind the notebook, the docs and the app all live here.

Data¶

quantile_compass.fetch_data ¶

Fetch daily historical price data for the core portfolio risk factors and a few supplementary risk-factor series, and align them on their common overlapping date range.

Run directly to (re)populate data/raw/market_data.csv:

python -m quantile_compass.fetch_data

fetch_series ¶

fetch_series(ticker: str) -> pd.Series

Fetch max-available daily closing prices for one ticker.

Source code in src/quantile_compass/fetch_data.py

def fetch_series(ticker: str) -> pd.Series:
    """Fetch max-available daily closing prices for one ticker."""
    df = yf.Ticker(ticker).history(period="max", interval="1d", auto_adjust=True)
    if df.empty:
        raise RuntimeError(f"No data returned for {ticker!r}")
    close = df["Close"].copy()
    close.index = pd.to_datetime(close.index).tz_localize(None).normalize()
    close.index.name = "Date"
    return close

fetch_all ¶

fetch_all(tickers: dict[str, str]) -> pd.DataFrame

Fetch every ticker in tickers and report each one's own availability.

Source code in src/quantile_compass/fetch_data.py

def fetch_all(tickers: dict[str, str]) -> pd.DataFrame:
    """Fetch every ticker in `tickers` and report each one's own availability."""
    series = {}
    for ticker, name in tickers.items():
        s = fetch_series(ticker)
        print(
            f"  {name:10s} ({ticker:9s}): {s.index.min().date()} -> {s.index.max().date()}  ({len(s)} rows)"
        )
        series[name] = s
    return pd.DataFrame(series)

quantile_compass.data ¶

Load and clean the fetched market data.

load_market_data ¶

load_market_data(path: str | Path = DEFAULT_DATA_PATH) -> pd.DataFrame

Load the raw fetched CSV, indexed by date.

Source code in src/quantile_compass/data.py

def load_market_data(path: str | Path = DEFAULT_DATA_PATH) -> pd.DataFrame:
    """Load the raw fetched CSV, indexed by date."""
    df = pd.read_csv(path, index_col=0, parse_dates=True)
    df.index.name = "Date"
    return df.sort_index()

clean_bad_ticks ¶

clean_bad_ticks(df: DataFrame, threshold: float = 0.5) -> pd.DataFrame

Detect and interpolate single-day "V-shaped" price spikes: a day whose return exceeds threshold in magnitude and is (mostly) reversed by the very next day's return in the opposite direction. Real market moves of that size persist; a value that fully round-trips in one day is a bad tick, not a crisis.

Returns a new DataFrame; flagged points are replaced by linear interpolation between their (good) neighbors.

Source code in src/quantile_compass/data.py

def clean_bad_ticks(df: pd.DataFrame, threshold: float = 0.5) -> pd.DataFrame:
    """
    Detect and interpolate single-day "V-shaped" price spikes: a day whose
    return exceeds `threshold` in magnitude and is (mostly) reversed by the
    very next day's return in the opposite direction. Real market moves of
    that size persist; a value that fully round-trips in one day is a bad
    tick, not a crisis.

    Returns a new DataFrame; flagged points are replaced by linear
    interpolation between their (good) neighbors.
    """
    out = df.copy()
    for col in CLEANABLE_COLUMNS:
        if col not in out.columns:
            continue
        s = out[col]
        ret = s.pct_change()
        next_ret = ret.shift(-1)
        # flag point t where |ret[t]| is extreme and next_ret[t] roughly
        # reverses it (opposite sign, similar magnitude)
        is_spike = ret.abs() > threshold
        reverts = (np.sign(ret) != np.sign(next_ret)) & (next_ret.abs() > threshold * 0.5)
        flagged = is_spike & reverts
        if flagged.any():
            s = s.mask(flagged)
            s = s.interpolate(method="linear")
            out[col] = s
    return out

prepare_dataset ¶

prepare_dataset(path: str | Path = DEFAULT_DATA_PATH, start: str | None = DENSE_DATA_START) -> pd.DataFrame

Load, trim to the dense period, and clean the dataset - the standard entry point for the rest of the package.

Pass start=None to keep the full fetched history including its sparse early section.

Source code in src/quantile_compass/data.py

def prepare_dataset(
    path: str | Path = DEFAULT_DATA_PATH,
    start: str | None = DENSE_DATA_START,
) -> pd.DataFrame:
    """
    Load, trim to the dense period, and clean the dataset - the standard entry
    point for the rest of the package.

    Pass ``start=None`` to keep the full fetched history including its sparse
    early section.
    """
    df = load_market_data(path)
    if start is not None:
        df = df.loc[start:]
    return clean_bad_ticks(df)

Returns¶

quantile_compass.returns ¶

Log returns and the equity/forex decomposition of the portfolio.

PortfolioSpec `dataclass` ¶

Weights and betas of the equity portfolio.

Attributes:

Name	Type	Description
`w_us`	`float`	Share of the portfolio held in US equity.
`w_de`	`float`	Share held in German equity.
`beta_us`	`float`	Market beta of the US holding.
`beta_de`	`float`	Market beta of the German holding.

Source code in src/quantile_compass/returns.py

@dataclass(frozen=True)
class PortfolioSpec:
    """
    Weights and betas of the equity portfolio.

    Attributes:
        w_us: Share of the portfolio held in US equity.
        w_de: Share held in German equity.
        beta_us: Market beta of the US holding.
        beta_de: Market beta of the German holding.
    """

    w_us: float = 0.6
    w_de: float = 0.4
    beta_us: float = 1.6
    beta_de: float = 1.3

    def __post_init__(self) -> None:
        total = self.w_us + self.w_de
        if not np.isclose(total, 1.0):
            raise ValueError(f"weights must sum to 1, got {total}")

compute_log_returns ¶

compute_log_returns(prices: DataFrame | Series, max_gap_days: int = DEFAULT_MAX_GAP_DAYS) -> pd.DataFrame | pd.Series

Continuously-compounded (log) returns.

Returns spanning more than max_gap_days calendar days are set to NaN: when the data source is missing a stretch of days, the change across that hole is not a one-day return and must not be treated as one.

Source code in src/quantile_compass/returns.py

def compute_log_returns(
    prices: pd.DataFrame | pd.Series,
    max_gap_days: int = DEFAULT_MAX_GAP_DAYS,
) -> pd.DataFrame | pd.Series:
    """
    Continuously-compounded (log) returns.

    Returns spanning more than `max_gap_days` calendar days are set to NaN:
    when the data source is missing a stretch of days, the change across that
    hole is not a one-day return and must not be treated as one.
    """
    log_ret = np.log(prices / prices.shift(1))
    spacing = prices.index.to_series().diff().dt.days
    too_wide = spacing > max_gap_days
    if isinstance(log_ret, pd.DataFrame):
        log_ret.loc[too_wide, :] = np.nan
    else:
        log_ret.loc[too_wide] = np.nan
    return log_ret

decompose_portfolio_returns ¶

decompose_portfolio_returns(prices: DataFrame, spec: PortfolioSpec | None = None, max_gap_days: int = DEFAULT_MAX_GAP_DAYS) -> pd.DataFrame

Split the portfolio's ruble return into its equity and forex components.

The ruble value of a foreign holding is the local-currency price times the exchange rate, so in logs the return separates cleanly into an equity part and a currency part:

r_equity = w_de * beta_de * r_DAX + w_us * beta_us * r_NASDAQ
r_forex  = w_de * r_EURRUB      + w_us * r_USDRUB
r_total  = r_equity + r_forex

Parameters:

Name	Type	Description	Default
`prices`	`DataFrame`	Price data containing NASDAQ100, DAX, USD_RUB, EUR_RUB columns.	required
`spec`	`PortfolioSpec \| None`	Portfolio weights/betas; defaults to 60/40 US/German with betas 1.6/1.3.	`None`
`max_gap_days`	`int`	Returns spanning wider gaps than this are dropped.	`DEFAULT_MAX_GAP_DAYS`

Returns:

Type	Description
`DataFrame`	DataFrame with `equity`, `forex` and `total` return columns, NaN rows removed.

Source code in src/quantile_compass/returns.py

def decompose_portfolio_returns(
    prices: pd.DataFrame,
    spec: PortfolioSpec | None = None,
    max_gap_days: int = DEFAULT_MAX_GAP_DAYS,
) -> pd.DataFrame:
    """
    Split the portfolio's ruble return into its equity and forex components.

    The ruble value of a foreign holding is the local-currency price times the
    exchange rate, so in logs the return separates cleanly into an equity part
    and a currency part:

        r_equity = w_de * beta_de * r_DAX + w_us * beta_us * r_NASDAQ
        r_forex  = w_de * r_EURRUB      + w_us * r_USDRUB
        r_total  = r_equity + r_forex

    Args:
        prices: Price data containing NASDAQ100, DAX, USD_RUB, EUR_RUB columns.
        spec: Portfolio weights/betas; defaults to 60/40 US/German with betas 1.6/1.3.
        max_gap_days: Returns spanning wider gaps than this are dropped.

    Returns:
        DataFrame with `equity`, `forex` and `total` return columns, NaN rows removed.
    """
    spec = spec or PortfolioSpec()
    required = {"NASDAQ100", "DAX", "USD_RUB", "EUR_RUB"}
    missing = required - set(prices.columns)
    if missing:
        raise KeyError(f"missing required price columns: {sorted(missing)}")

    r = compute_log_returns(prices[sorted(required)], max_gap_days=max_gap_days)

    equity = spec.w_de * spec.beta_de * r["DAX"] + spec.w_us * spec.beta_us * r["NASDAQ100"]
    forex = spec.w_de * r["EUR_RUB"] + spec.w_us * r["USD_RUB"]

    out = pd.DataFrame({"equity": equity, "forex": forex})
    out["total"] = out["equity"] + out["forex"]
    return out.dropna(how="any")

Volatility¶

quantile_compass.volatility ¶

EWMA volatility, covariance and correlation estimation (RiskMetrics style).

ewma_variance ¶

ewma_variance(returns: Series, lam: float = DEFAULT_LAMBDA, burn_in: int = DEFAULT_BURN_IN, mask_burn_in: bool = True) -> pd.Series

EWMA variance estimate.

.. math:: \hat\sigma_t^2 = (1-\lambda) r_{t-1}^2 + \lambda \hat\sigma_{t-1}^2

The recursion is seeded from the first burn_in observations rather than the whole sample, so that from burn_in onward the estimate at time t depends only on returns before t - i.e. it is usable as a genuine one-day-ahead forecast. Seeding on the full sample (as textbook treatments often do) leaks future information into every earlier estimate.

Parameters:

Name	Type	Description	Default
`returns`	`Series`	Return series.	required
`lam`	`float`	Decay factor, strictly between 0 and 1.	`DEFAULT_LAMBDA`
`burn_in`	`int`	Number of observations used to seed the recursion.	`DEFAULT_BURN_IN`
`mask_burn_in`	`bool`	If True, the seeding window is returned as NaN, since those estimates are informed by their own window.	`True`

Source code in src/quantile_compass/volatility.py

def ewma_variance(
    returns: pd.Series,
    lam: float = DEFAULT_LAMBDA,
    burn_in: int = DEFAULT_BURN_IN,
    mask_burn_in: bool = True,
) -> pd.Series:
    r"""
    EWMA variance estimate.

    .. math::
        \hat\sigma_t^2 = (1-\lambda) r_{t-1}^2 + \lambda \hat\sigma_{t-1}^2

    The recursion is seeded from the first `burn_in` observations rather than
    the whole sample, so that from `burn_in` onward the estimate at time *t*
    depends only on returns before *t* - i.e. it is usable as a genuine
    one-day-ahead forecast. Seeding on the full sample (as textbook treatments
    often do) leaks future information into every earlier estimate.

    Args:
        returns: Return series.
        lam: Decay factor, strictly between 0 and 1.
        burn_in: Number of observations used to seed the recursion.
        mask_burn_in: If True, the seeding window is returned as NaN, since
            those estimates are informed by their own window.
    """
    _validate_lambda(lam)
    r = returns.to_numpy(dtype=float)
    var, k = _ewma_recursion(r**2, lam, burn_in)
    if mask_burn_in and len(var):
        var[:k] = np.nan
    return pd.Series(var, index=returns.index, name=f"{returns.name}_var")

ewma_volatility ¶

ewma_volatility(returns: Series, lam: float = DEFAULT_LAMBDA, burn_in: int = DEFAULT_BURN_IN, mask_burn_in: bool = True) -> pd.Series

EWMA standard deviation - the square root of :func:ewma_variance.

Source code in src/quantile_compass/volatility.py

def ewma_volatility(
    returns: pd.Series,
    lam: float = DEFAULT_LAMBDA,
    burn_in: int = DEFAULT_BURN_IN,
    mask_burn_in: bool = True,
) -> pd.Series:
    """EWMA standard deviation - the square root of :func:`ewma_variance`."""
    var = ewma_variance(returns, lam=lam, burn_in=burn_in, mask_burn_in=mask_burn_in)
    return np.sqrt(var).rename(f"{returns.name}_vol")

ewma_covariance ¶

ewma_covariance(a: Series, b: Series, lam: float = DEFAULT_LAMBDA, burn_in: int = DEFAULT_BURN_IN, mask_burn_in: bool = True) -> pd.Series

EWMA covariance between two contemporaneous return series.

.. math:: \hat\sigma_{ab,t} = (1-\lambda) r_{a,t-1} r_{b,t-1} + \lambda \hat\sigma_{ab,t-1}

Seeded the same way as :func:ewma_variance, for the same reason.

Source code in src/quantile_compass/volatility.py

def ewma_covariance(
    a: pd.Series,
    b: pd.Series,
    lam: float = DEFAULT_LAMBDA,
    burn_in: int = DEFAULT_BURN_IN,
    mask_burn_in: bool = True,
) -> pd.Series:
    r"""
    EWMA covariance between two contemporaneous return series.

    .. math::
        \hat\sigma_{ab,t} = (1-\lambda) r_{a,t-1} r_{b,t-1} + \lambda \hat\sigma_{ab,t-1}

    Seeded the same way as :func:`ewma_variance`, for the same reason.
    """
    _validate_lambda(lam)
    if not a.index.equals(b.index):
        raise ValueError("series must share an index")
    ra, rb = a.to_numpy(dtype=float), b.to_numpy(dtype=float)
    cov, k = _ewma_recursion(ra * rb, lam, burn_in)
    if mask_burn_in and len(cov):
        cov[:k] = np.nan
    return pd.Series(cov, index=a.index, name="covariance")

ewma_correlation ¶

ewma_correlation(a: Series, b: Series, lam: float = DEFAULT_LAMBDA, burn_in: int = DEFAULT_BURN_IN, mask_burn_in: bool = True) -> pd.Series

EWMA correlation, i.e. covariance normalised by the two EWMA volatilities.

Source code in src/quantile_compass/volatility.py

def ewma_correlation(
    a: pd.Series,
    b: pd.Series,
    lam: float = DEFAULT_LAMBDA,
    burn_in: int = DEFAULT_BURN_IN,
    mask_burn_in: bool = True,
) -> pd.Series:
    """EWMA correlation, i.e. covariance normalised by the two EWMA volatilities."""
    kw = dict(lam=lam, burn_in=burn_in, mask_burn_in=mask_burn_in)
    cov = ewma_covariance(a, b, **kw)
    corr = cov / (ewma_volatility(a, **kw) * ewma_volatility(b, **kw))
    return corr.rename("correlation")

annualize_volatility ¶

annualize_volatility(daily_vol: Series | float, trading_days: int = TRADING_DAYS)

Scale a daily volatility to an annual one by the square root of time.

Source code in src/quantile_compass/volatility.py

def annualize_volatility(daily_vol: pd.Series | float, trading_days: int = TRADING_DAYS):
    """Scale a daily volatility to an annual one by the square root of time."""
    return daily_vol * np.sqrt(trading_days)

covariance_matrix ¶

covariance_matrix(equity: Series, forex: Series, lam: float = DEFAULT_LAMBDA, annualize: bool = True, trading_days: int = TRADING_DAYS, burn_in: int = DEFAULT_BURN_IN, asof: Timestamp | str | None = None) -> pd.DataFrame

Build the 2x2 equity/forex covariance matrix as of a given date (default: the final observation).

Returns a labelled DataFrame so it reads clearly in a notebook.

Source code in src/quantile_compass/volatility.py

def covariance_matrix(
    equity: pd.Series,
    forex: pd.Series,
    lam: float = DEFAULT_LAMBDA,
    annualize: bool = True,
    trading_days: int = TRADING_DAYS,
    burn_in: int = DEFAULT_BURN_IN,
    asof: pd.Timestamp | str | None = None,
) -> pd.DataFrame:
    """
    Build the 2x2 equity/forex covariance matrix as of a given date
    (default: the final observation).

    Returns a labelled DataFrame so it reads clearly in a notebook.
    """
    kw = dict(lam=lam, burn_in=burn_in)
    var_e_s = ewma_variance(equity, **kw)
    var_f_s = ewma_variance(forex, **kw)
    cov_s = ewma_covariance(equity, forex, **kw)
    if asof is None:
        var_e, var_f, cov_ef = var_e_s.iloc[-1], var_f_s.iloc[-1], cov_s.iloc[-1]
    else:
        var_e, var_f, cov_ef = var_e_s.loc[asof], var_f_s.loc[asof], cov_s.loc[asof]

    if not np.isfinite([var_e, var_f, cov_ef]).all():
        raise ValueError(
            f"EWMA estimates are undefined at the requested date: the series has "
            f"{len(equity)} observations and burn_in={burn_in}, so no estimate is "
            f"available outside the seeding window. Supply more data or lower burn_in."
        )
    scale = trading_days if annualize else 1.0
    matrix = np.array([[var_e, cov_ef], [cov_ef, var_f]]) * scale
    return pd.DataFrame(matrix, index=["equity", "forex"], columns=["equity", "forex"])

Value-at-Risk¶

quantile_compass.var ¶

Value-at-Risk estimators: parametric, historical, and age-weighted historical.

parametric_var ¶

parametric_var(sensitivities: ndarray | list[float], cov_matrix: DataFrame | ndarray, alpha: float = DEFAULT_ALPHA, horizon_days: int = 1, trading_days: int = TRADING_DAYS, annualized_cov: bool = True) -> float

Normal parametric VaR for a linear portfolio of risk factors.

.. math:: VaR_{h,\alpha} = \Phi^{-1}(1-\alpha)\,\sqrt{\theta' \Omega_h \theta}

Parameters:

Name	Type	Description	Default
`sensitivities`	`ndarray \| list[float]`	Exposure vector :math:`\theta` - e.g. `[1, 1]` for the combined portfolio, `[1, 0]` for equity stand-alone.	required
`cov_matrix`	`DataFrame \| ndarray`	Risk-factor covariance matrix.	required
`alpha`	`float`	Significance level (0.01 = 99% confidence).	`DEFAULT_ALPHA`
`horizon_days`	`int`	Risk horizon in trading days.	`1`
`trading_days`	`int`	Trading days per year, used to de-annualize.	`TRADING_DAYS`
`annualized_cov`	`bool`	Whether `cov_matrix` is annualized.	`True`

Returns:

Type	Description
`float`	VaR as a positive fraction of portfolio value (0.04 = a 4% loss).

Source code in src/quantile_compass/var.py

def parametric_var(
    sensitivities: np.ndarray | list[float],
    cov_matrix: pd.DataFrame | np.ndarray,
    alpha: float = DEFAULT_ALPHA,
    horizon_days: int = 1,
    trading_days: int = TRADING_DAYS,
    annualized_cov: bool = True,
) -> float:
    r"""
    Normal parametric VaR for a linear portfolio of risk factors.

    .. math::
        VaR_{h,\alpha} = \Phi^{-1}(1-\alpha)\,\sqrt{\theta' \Omega_h \theta}

    Args:
        sensitivities: Exposure vector :math:`\theta` - e.g. ``[1, 1]`` for the
            combined portfolio, ``[1, 0]`` for equity stand-alone.
        cov_matrix: Risk-factor covariance matrix.
        alpha: Significance level (0.01 = 99% confidence).
        horizon_days: Risk horizon in trading days.
        trading_days: Trading days per year, used to de-annualize.
        annualized_cov: Whether `cov_matrix` is annualized.

    Returns:
        VaR as a positive fraction of portfolio value (0.04 = a 4% loss).
    """
    _validate_alpha(alpha)
    if horizon_days <= 0:
        raise ValueError(f"horizon_days must be positive, got {horizon_days}")
    theta = np.asarray(sensitivities, dtype=float)
    omega = np.asarray(cov_matrix, dtype=float)
    variance = float(theta @ omega @ theta)
    if variance < 0:
        raise ValueError("covariance matrix produced a negative variance")
    scale = np.sqrt(horizon_days / trading_days) if annualized_cov else np.sqrt(horizon_days)
    return float(scale * norm.ppf(1 - alpha) * np.sqrt(variance))

historical_var ¶

historical_var(returns: Series, alpha: float = DEFAULT_ALPHA) -> float

Historical-simulation VaR: the empirical alpha-quantile of realised returns, sign-flipped so a loss is reported positive. Makes no distributional assumption.

Source code in src/quantile_compass/var.py

def historical_var(returns: pd.Series, alpha: float = DEFAULT_ALPHA) -> float:
    """
    Historical-simulation VaR: the empirical alpha-quantile of realised
    returns, sign-flipped so a loss is reported positive. Makes no
    distributional assumption.
    """
    _validate_alpha(alpha)
    if returns.empty:
        raise ValueError("cannot compute VaR on an empty series")
    return float(-returns.quantile(alpha))

age_weighted_historical_var ¶

age_weighted_historical_var(returns: Series, alpha: float = DEFAULT_ALPHA, lam: float = DEFAULT_LAMBDA) -> float

Age-weighted ("hybrid") historical VaR after Boudoukh, Richardson and Whitelaw (1998).

Each observation gets an exponentially decaying probability weight, most recent first (:math:\omega_T = 1-\lambda, :math:\omega_{i-1} = \lambda\omega_i). Returns are then sorted and the quantile is read off the cumulative weighted distribution, so recent market conditions dominate the tail without discarding older history outright.

Source code in src/quantile_compass/var.py

def age_weighted_historical_var(
    returns: pd.Series,
    alpha: float = DEFAULT_ALPHA,
    lam: float = DEFAULT_LAMBDA,
) -> float:
    r"""
    Age-weighted ("hybrid") historical VaR after Boudoukh, Richardson and
    Whitelaw (1998).

    Each observation gets an exponentially decaying probability weight, most
    recent first (:math:`\omega_T = 1-\lambda`, :math:`\omega_{i-1} = \lambda\omega_i`).
    Returns are then sorted and the quantile is read off the cumulative
    weighted distribution, so recent market conditions dominate the tail
    without discarding older history outright.
    """
    _validate_alpha(alpha)
    if returns.empty:
        raise ValueError("cannot compute VaR on an empty series")

    n = len(returns)
    # weights increase towards the most recent observation
    ages = np.arange(n - 1, -1, -1)
    weights = (1 - lam) * lam**ages
    weights /= weights.sum()

    order = np.argsort(returns.to_numpy())
    sorted_returns = returns.to_numpy()[order]
    cumulative = np.cumsum(weights[order])

    idx = int(np.searchsorted(cumulative, alpha))
    idx = min(idx, n - 1)
    return float(-sorted_returns[idx])

scale_var_horizon ¶

scale_var_horizon(var_1day: float, horizon_days: int) -> float

Square-root-of-time scaling of a 1-day VaR (assumes i.i.d. returns).

Source code in src/quantile_compass/var.py

def scale_var_horizon(var_1day: float, horizon_days: int) -> float:
    """Square-root-of-time scaling of a 1-day VaR (assumes i.i.d. returns)."""
    if horizon_days <= 0:
        raise ValueError(f"horizon_days must be positive, got {horizon_days}")
    return var_1day * np.sqrt(horizon_days)

rolling_parametric_var ¶

rolling_parametric_var(equity: Series, forex: Series, alpha: float = DEFAULT_ALPHA, horizon_days: int = 1, lam: float = DEFAULT_LAMBDA, burn_in: int = DEFAULT_BURN_IN) -> pd.Series

Parametric VaR recomputed at every date from that date's EWMA covariance.

The combined portfolio's variance at time t is

.. math:: \sigma^2_t = \sigma^2_{E,t} + 2\sigma_{EF,t} + \sigma^2_{F,t}

(the quadratic form with exposures :math:\theta = [1, 1]), and the VaR is :math:\Phi^{-1}(1-\alpha)\sqrt{h}\,\sigma_t.

Because the EWMA estimates carry no look-ahead, the value at t is what the model would have forecast on the morning of t - which is what makes the series usable for backtesting.

Returns:

Type	Description
`Series`	VaR as a positive fraction of portfolio value, NaN through the burn-in.

Source code in src/quantile_compass/var.py

def rolling_parametric_var(
    equity: pd.Series,
    forex: pd.Series,
    alpha: float = DEFAULT_ALPHA,
    horizon_days: int = 1,
    lam: float = DEFAULT_LAMBDA,
    burn_in: int = DEFAULT_BURN_IN,
) -> pd.Series:
    r"""
    Parametric VaR recomputed at every date from that date's EWMA covariance.

    The combined portfolio's variance at time *t* is

    .. math::
        \sigma^2_t = \sigma^2_{E,t} + 2\sigma_{EF,t} + \sigma^2_{F,t}

    (the quadratic form with exposures :math:`\theta = [1, 1]`), and the VaR is
    :math:`\Phi^{-1}(1-\alpha)\sqrt{h}\,\sigma_t`.

    Because the EWMA estimates carry no look-ahead, the value at *t* is what the
    model would have forecast on the morning of *t* - which is what makes the
    series usable for backtesting.

    Returns:
        VaR as a positive fraction of portfolio value, NaN through the burn-in.
    """
    _validate_alpha(alpha)
    if horizon_days <= 0:
        raise ValueError(f"horizon_days must be positive, got {horizon_days}")

    kw = dict(lam=lam, burn_in=burn_in)
    var_e = ewma_variance(equity, **kw)
    var_f = ewma_variance(forex, **kw)
    cov_ef = ewma_covariance(equity, forex, **kw)

    total_var = var_e + 2 * cov_ef + var_f
    total_var = total_var.clip(lower=0)  # numerical guard
    var = norm.ppf(1 - alpha) * np.sqrt(horizon_days) * np.sqrt(total_var)
    return var.rename("var")

count_var_breaches ¶

count_var_breaches(returns: Series, var_series: Series) -> dict[str, float]

Backtest a VaR series: how often did the realised loss exceed the forecast?

A well-calibrated 99% VaR should be breached on about 1% of days. Far fewer means the model is too conservative and ties up capital; far more means it understates the risk.

Returns:

Type	Description
`dict[str, float]`	Dict with the number of observations compared, the breach count, the
`dict[str, float]`	realised breach rate and the rate the confidence level implies.

Source code in src/quantile_compass/var.py

def count_var_breaches(returns: pd.Series, var_series: pd.Series) -> dict[str, float]:
    """
    Backtest a VaR series: how often did the realised loss exceed the forecast?

    A well-calibrated 99% VaR should be breached on about 1% of days. Far fewer
    means the model is too conservative and ties up capital; far more means it
    understates the risk.

    Returns:
        Dict with the number of observations compared, the breach count, the
        realised breach rate and the rate the confidence level implies.
    """
    aligned = pd.concat([returns.rename("r"), var_series.rename("v")], axis=1).dropna()
    if aligned.empty:
        return {"observations": 0, "breaches": 0, "breach_rate": float("nan")}
    breaches = (aligned["r"] < -aligned["v"]).sum()
    return {
        "observations": int(len(aligned)),
        "breaches": int(breaches),
        "breach_rate": float(breaches / len(aligned)),
    }

Plotting¶

quantile_compass.viz ¶

Reusable plotting helpers shared by the notebook, the docs and the app.

shade_crises ¶

shade_crises(ax: Axes, windows: dict[str, tuple[str, str]] | None = None) -> None

Shade the crisis windows on a time-axis chart.

Source code in src/quantile_compass/viz.py

def shade_crises(ax: plt.Axes, windows: dict[str, tuple[str, str]] | None = None) -> None:
    """Shade the crisis windows on a time-axis chart."""
    for start, end in (windows or CRISIS_WINDOWS).values():
        ax.axvspan(
            pd.Timestamp(start), pd.Timestamp(end), color=PALETTE["crisis"], alpha=0.18, lw=0
        )

plot_price_levels ¶

plot_price_levels(prices: DataFrame, columns: list[str], ax: Axes | None = None) -> plt.Axes

Plot price series rebased to 100 at the start of the sample.

Source code in src/quantile_compass/viz.py

def plot_price_levels(
    prices: pd.DataFrame, columns: list[str], ax: plt.Axes | None = None
) -> plt.Axes:
    """Plot price series rebased to 100 at the start of the sample."""
    ax = ax or plt.subplots(figsize=(10, 4.5), constrained_layout=True)[1]
    for i, col in enumerate(columns):
        rebased = prices[col] / prices[col].iloc[0] * 100
        color = list(PALETTE.values())[i % len(PALETTE)]
        ax.plot(rebased.index, rebased, lw=1.3, label=col, color=color)
    shade_crises(ax)
    _format_time_axis(ax, "Index level (start = 100)")
    ax.legend(frameon=False, loc="upper left")
    return ax

plot_volatility ¶

plot_volatility(vol_equity: Series, vol_forex: Series, ax: Axes | None = None, as_percent: bool = True) -> plt.Axes

Plot the two EWMA volatility series with crisis windows shaded.

Source code in src/quantile_compass/viz.py

def plot_volatility(
    vol_equity: pd.Series,
    vol_forex: pd.Series,
    ax: plt.Axes | None = None,
    as_percent: bool = True,
) -> plt.Axes:
    """Plot the two EWMA volatility series with crisis windows shaded."""
    ax = ax or plt.subplots(figsize=(10, 4.5), constrained_layout=True)[1]
    scale = 100 if as_percent else 1
    ax.plot(vol_equity.index, vol_equity * scale, lw=1.2, color=PALETTE["equity"], label="Equity")
    ax.plot(vol_forex.index, vol_forex * scale, lw=1.2, color=PALETTE["forex"], label="Forex")
    shade_crises(ax)
    _format_time_axis(ax, "EWMA annualised volatility (%)" if as_percent else "EWMA volatility")
    ax.legend(frameon=False, loc="upper left")
    return ax

plot_correlation ¶

plot_correlation(correlation: Series, ax: Axes | None = None) -> plt.Axes

Plot the EWMA equity/forex correlation through time.

Source code in src/quantile_compass/viz.py

def plot_correlation(correlation: pd.Series, ax: plt.Axes | None = None) -> plt.Axes:
    """Plot the EWMA equity/forex correlation through time."""
    ax = ax or plt.subplots(figsize=(10, 4.5), constrained_layout=True)[1]
    ax.plot(correlation.index, correlation, lw=1.2, color=PALETTE["accent"])
    ax.axhline(0, color=PALETTE["muted"], lw=0.9, ls="--")
    shade_crises(ax)
    _format_time_axis(ax, "EWMA equity-forex correlation")
    ax.set_ylim(-1, 1)
    return ax

plot_return_distribution ¶

plot_return_distribution(returns: Series, ax: Axes | None = None, bins: int = 100) -> plt.Axes

Histogram of portfolio returns.

Source code in src/quantile_compass/viz.py

def plot_return_distribution(
    returns: pd.Series, ax: plt.Axes | None = None, bins: int = 100
) -> plt.Axes:
    """Histogram of portfolio returns."""
    ax = ax or plt.subplots(figsize=(10, 4.5), constrained_layout=True)[1]
    ax.hist(returns * 100, bins=bins, color=PALETTE["total"], alpha=0.85)
    ax.set_xlabel("Daily portfolio return (%)")
    ax.set_ylabel("Frequency")
    ax.grid(alpha=0.25, lw=0.6)
    for spine in ("top", "right"):
        ax.spines[spine].set_visible(False)
    return ax

plot_var_comparison ¶

plot_var_comparison(var_by_method: dict[str, float], ax: Axes | None = None) -> plt.Axes

Bar chart comparing VaR estimates across methods.

Source code in src/quantile_compass/viz.py

def plot_var_comparison(var_by_method: dict[str, float], ax: plt.Axes | None = None) -> plt.Axes:
    """Bar chart comparing VaR estimates across methods."""
    ax = ax or plt.subplots(figsize=(7, 4.2), constrained_layout=True)[1]
    names = list(var_by_method)
    values = [var_by_method[n] * 100 for n in names]
    colors = [PALETTE["equity"], PALETTE["forex"], PALETTE["accent"]][: len(names)]
    bars = ax.bar(names, values, color=colors, width=0.55)
    ax.bar_label(bars, fmt="%.2f%%", padding=3)
    ax.set_ylabel("1-day 99% VaR (%)")
    ax.grid(alpha=0.25, axis="y", lw=0.6)
    for spine in ("top", "right"):
        ax.spines[spine].set_visible(False)
    ax.set_ylim(0, max(values) * 1.2)
    return ax

API reference¶

Data¶

quantile_compass.fetch_data ¶

fetch_series ¶

fetch_all ¶

quantile_compass.data ¶

load_market_data ¶

clean_bad_ticks ¶

prepare_dataset ¶

Returns¶

quantile_compass.returns ¶

PortfolioSpec dataclass ¶

compute_log_returns ¶

decompose_portfolio_returns ¶

Volatility¶

quantile_compass.volatility ¶

ewma_variance ¶

ewma_volatility ¶

ewma_covariance ¶

ewma_correlation ¶

annualize_volatility ¶

covariance_matrix ¶

Value-at-Risk¶

quantile_compass.var ¶

parametric_var ¶

historical_var ¶

age_weighted_historical_var ¶

scale_var_horizon ¶

rolling_parametric_var ¶

count_var_breaches ¶

Plotting¶

quantile_compass.viz ¶

shade_crises ¶

plot_price_levels ¶

plot_volatility ¶

plot_correlation ¶

plot_return_distribution ¶

plot_var_comparison ¶

PortfolioSpec `dataclass` ¶