beacon.derivatives¶

Delta-1 instruments referencing indices, ETFs, and equities: IndexFuture, ETFFuture, and TotalReturnSwap, built on a DerivativeBase ABC, plus pure pricing functions (cost-of-carry, discrete-dividend forward, implied repo, roll return, TRS breakeven spread).

derivatives ¶

The 'derivatives' package models exchange-traded and OTC Delta-1 derivatives that reference beacon indices, ETFs, and equities.

DerivativeBase ¶

DerivativeBase(
    derivative_id: str,
    underlying_id: str,
    underlying_type: str,
    currency: str,
    expiry_date: str,
    notional: float,
)

Bases: ABC

Abstract base for Delta-1 derivative instruments.

Holds the common contract terms (identifiers, currency, expiry, notional) and the ACT/365 time-to-expiry helper. Concrete subclasses implement :meth:fair_value and :meth:mark_to_market.

Initialise the common contract terms.

Parameters:

Name	Type	Description	Default
`derivative_id`	`str`	Unique identifier for this derivative.	required
`underlying_id`	`str`	Identifier of the referenced underlying.	required
`underlying_type`	`str`	One of `INDEX`, `ETF` or `EQUITY` (case-insensitive).	required
`currency`	`str`	Contract currency (e.g. `USD`).	required
`expiry_date`	`str`	Expiry date (YYYY-MM-DD).	required
`notional`	`float`	Contract notional; must be positive.	required

Raises:

Type	Description
`ValueError`	If any required field is empty/invalid, the underlying type is unrecognised, or notional is not positive.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/base.py

def __init__(self,
             derivative_id: str,
             underlying_id: str,
             underlying_type: str,
             currency: str,
             expiry_date: str,
             notional: float):
    """Initialise the common contract terms.

    Args:
        derivative_id: Unique identifier for this derivative.
        underlying_id: Identifier of the referenced underlying.
        underlying_type: One of ``INDEX``, ``ETF`` or ``EQUITY``
            (case-insensitive).
        currency: Contract currency (e.g. ``USD``).
        expiry_date: Expiry date (YYYY-MM-DD).
        notional: Contract notional; must be positive.

    Raises:
        ValueError: If any required field is empty/invalid, the underlying
            type is unrecognised, or *notional* is not positive.
    """
    if not derivative_id:
        raise ValueError("derivative_id cannot be empty.")
    if not underlying_id:
        raise ValueError("underlying_id cannot be empty.")
    if not underlying_type:
        raise ValueError("underlying_type cannot be empty.")
    if not currency:
        raise ValueError("currency cannot be empty.")
    if not expiry_date:
        raise ValueError("expiry_date cannot be empty.")

    underlying_type_norm = underlying_type.upper()
    if underlying_type_norm not in self.VALID_UNDERLYING_TYPES:
        raise ValueError(
            f"underlying_type must be one of "
            f"{sorted(self.VALID_UNDERLYING_TYPES)}, got '{underlying_type}'."
        )

    if notional <= 0:
        raise ValueError(f"notional must be positive, got {notional}.")

    self.derivative_id: str = derivative_id
    self.underlying_id: str = underlying_id
    self.underlying_type: str = underlying_type_norm
    self.currency: str = currency.upper()
    self.expiry_date: pd.Timestamp = pd.Timestamp(expiry_date)
    self.notional: float = notional

    logger.info(
        f"{type(self).__name__} '{self.derivative_id}' created on "
        f"{self.underlying_type} '{self.underlying_id}', "
        f"expiry {self.expiry_date.strftime('%Y-%m-%d')}."
    )

time_to_expiry ¶

time_to_expiry(valuation_date: Timestamp) -> float

Time to expiry in years using the ACT/365 convention.

Parameters:

Name	Type	Description	Default
`valuation_date`	`Timestamp`	The date from which to measure.	required

Returns:

Type	Description
`float`	Years to expiry, clamped to `0.0` once the contract has expired.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/base.py

def time_to_expiry(self,
                   valuation_date: pd.Timestamp) -> float:
    """Time to expiry in years using the ACT/365 convention.

    Args:
        valuation_date: The date from which to measure.

    Returns:
        Years to expiry, clamped to ``0.0`` once the contract has expired.
    """
    seconds = float((self.expiry_date - pd.Timestamp(valuation_date)).total_seconds())
    return max(0.0, seconds / _SECONDS_PER_YEAR)

fair_value `abstractmethod` ¶

fair_value(
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> float

Return the model fair value of the derivative.

Parameters:

Name	Type	Description	Default
`spot_price`	`float`	Current spot/level of the underlying.	required
`valuation_date`	`Timestamp`	The valuation date.	required
`market_data`	`dict[str, Any]`	Additional inputs keyed by name. Most are scalar rates (e.g. `risk_free_rate`, `dividend_yield`), but subclasses also read non-scalar entries such as `discrete_dividends` (a list of `(time, amount)` tuples).	required

Returns:

Type	Description
`float`	The fair value in contract currency.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/base.py

@abstractmethod
def fair_value(self,
               spot_price: float,
               valuation_date: pd.Timestamp,
               market_data: dict[str, Any]) -> float:
    """Return the model fair value of the derivative.

    Args:
        spot_price: Current spot/level of the underlying.
        valuation_date: The valuation date.
        market_data: Additional inputs keyed by name. Most are scalar rates
            (e.g. ``risk_free_rate``, ``dividend_yield``), but subclasses
            also read non-scalar entries such as ``discrete_dividends``
            (a list of ``(time, amount)`` tuples).

    Returns:
        The fair value in contract currency.
    """
    raise NotImplementedError

mark_to_market `abstractmethod` ¶

mark_to_market(
    market_price: float,
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> dict[str, float]

Mark the position to market against an observed market_price.

Parameters:

Name	Type	Description	Default
`market_price`	`float`	Observed traded price of the derivative.	required
`spot_price`	`float`	Current spot/level of the underlying.	required
`valuation_date`	`Timestamp`	The valuation date.	required
`market_data`	`dict[str, Any]`	Additional inputs keyed by name.	required

Returns:

Type	Description
`dict[str, float]`	A dictionary of mark-to-market results (e.g. fair value, PnL,
`dict[str, float]`	basis) in contract currency.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/base.py

@abstractmethod
def mark_to_market(self,
                   market_price: float,
                   spot_price: float,
                   valuation_date: pd.Timestamp,
                   market_data: dict[str, Any]) -> dict[str, float]:
    """Mark the position to market against an observed *market_price*.

    Args:
        market_price: Observed traded price of the derivative.
        spot_price: Current spot/level of the underlying.
        valuation_date: The valuation date.
        market_data: Additional inputs keyed by name.

    Returns:
        A dictionary of mark-to-market results (e.g. fair value, PnL,
        basis) in contract currency.
    """
    raise NotImplementedError

RateCurve `dataclass` ¶

RateCurve(
    tenors: tuple[float, ...], rates: tuple[float, ...]
)

A zero-rate curve defined by pillar points.

Attributes:

Name	Type	Description
`tenors`	`tuple[float, ...]`	Pillar tenors in years, strictly increasing.
`rates`	`tuple[float, ...]`	Continuously compounded zero rate at each pillar.

is_flat `property` ¶

is_flat: bool

Whether every pillar carries the same rate.

flat `classmethod` ¶

flat(rate: float) -> RateCurve

A curve with the same rate at every tenor.

The bridge back to scalar-rate pricing: a flat curve returns exactly the rate it was given, so every existing result is reproduced bit for bit rather than approximately.

Parameters:

Name	Type	Description	Default
`rate`	`float`	The continuously compounded rate.	required

Returns:

Name	Type	Description
`RateCurve`	`RateCurve`	A single-pillar curve.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

@classmethod
def flat(cls,
         rate: float) -> "RateCurve":
    """A curve with the same rate at every tenor.

    The bridge back to scalar-rate pricing: a flat curve returns exactly the
    rate it was given, so every existing result is reproduced bit for bit
    rather than approximately.

    Args:
        rate: The continuously compounded rate.

    Returns:
        RateCurve: A single-pillar curve.
    """
    return cls(tenors=(1.0,), rates=(float(rate),))

from_pillars `classmethod` ¶

from_pillars(pillars: dict[float, float]) -> RateCurve

Build a curve from a {tenor: rate} mapping, sorted by tenor.

Parameters:

Name	Type	Description	Default
`pillars`	`dict[float, float]`	Tenor in years to continuously compounded zero rate.	required

Returns:

Name	Type	Description
`RateCurve`	`RateCurve`	The curve.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

@classmethod
def from_pillars(cls,
                 pillars: dict[float, float]) -> "RateCurve":
    """Build a curve from a ``{tenor: rate}`` mapping, sorted by tenor.

    Args:
        pillars: Tenor in years to continuously compounded zero rate.

    Returns:
        RateCurve: The curve.
    """
    if not pillars:
        raise CalculationError("RateCurve", "a curve needs at least one pillar.")

    ordered = sorted(pillars.items())

    return cls(tenors=tuple(float(tenor) for tenor, _ in ordered),
               rates=tuple(float(rate) for _, rate in ordered))

zero_rate ¶

zero_rate(tenor: float) -> float

The zero rate at tenor, interpolated between pillars.

Parameters:

Name	Type	Description	Default
`tenor`	`float`	Years from the valuation date. Must be non-negative.	required

Returns:

Name	Type	Description
`float`	`float`	The continuously compounded zero rate. Flat beyond the first
	`float`	and last pillar.

Raises:

Type	Description
`CalculationError`	If tenor is negative.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

def zero_rate(self,
              tenor: float) -> float:
    """The zero rate at *tenor*, interpolated between pillars.

    Args:
        tenor: Years from the valuation date. Must be non-negative.

    Returns:
        float: The continuously compounded zero rate. Flat beyond the first
        and last pillar.

    Raises:
        CalculationError: If *tenor* is negative.
    """
    if tenor < 0.0:
        raise CalculationError("RateCurve",
                               f"tenor must be non-negative, got {tenor}.")

    # Single pillar, or every pillar equal: one answer, returned exactly.
    # Not an optimisation — it is what makes a flat curve reproduce a scalar
    # rate without any interpolation arithmetic in between.
    if len(self.tenors) == 1 or self.is_flat:
        return self.rates[0]

    if tenor <= self.tenors[0]:
        return self.rates[0]

    if tenor >= self.tenors[-1]:
        return self.rates[-1]

    return self._interpolate(tenor)

discount_factor ¶

discount_factor(tenor: float) -> float

Present value of one unit paid at tenor.

Parameters:

Name	Type	Description	Default
`tenor`	`float`	Years from the valuation date.	required

Returns:

Name	Type	Description
`float`	`float`	`exp(-z(T) * T)`. Exactly 1.0 at tenor zero.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

def discount_factor(self,
                    tenor: float) -> float:
    """Present value of one unit paid at *tenor*.

    Args:
        tenor: Years from the valuation date.

    Returns:
        float: ``exp(-z(T) * T)``. Exactly 1.0 at tenor zero.
    """
    if tenor == 0.0:
        return 1.0

    return math.exp(-self.zero_rate(tenor) * tenor)

forward_rate ¶

forward_rate(start: float, end: float) -> float

The rate implied between two future dates.

The rate that makes discounting to end the same as discounting to start and then forward at this rate — which is what a financing leg resetting at start should be projected at.

Parameters:

Name	Type	Description	Default
`start`	`float`	Start of the forward period, in years.	required
`end`	`float`	End of the forward period, in years.	required

Returns:

Name	Type	Description
`float`	`float`	Continuously compounded forward rate.

Raises:

Type	Description
`CalculationError`	If end is not after start.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

def forward_rate(self,
                 start: float,
                 end: float) -> float:
    """The rate implied between two future dates.

    The rate that makes discounting to *end* the same as discounting to
    *start* and then forward at this rate — which is what a financing leg
    resetting at *start* should be projected at.

    Args:
        start: Start of the forward period, in years.
        end: End of the forward period, in years.

    Returns:
        float: Continuously compounded forward rate.

    Raises:
        CalculationError: If *end* is not after *start*.
    """
    if end - start <= TENOR_TOLERANCE:
        raise CalculationError(
            "RateCurve",
            f"the forward period must be positive, got {start} to {end}.")

    # (z_end * end - z_start * start) / (end - start), which is the same as
    # -ln(DF_end / DF_start) / (end - start) without the round trip through
    # exp and log.
    return ((self.zero_rate(end) * end - self.zero_rate(start) * start)
            / (end - start))

shifted ¶

shifted(bump: float) -> RateCurve

A copy with every pillar moved by bump.

The parallel shift a DV01 is measured against.

Parameters:

Name	Type	Description	Default
`bump`	`float`	Amount to add to every rate, in decimal. One basis point is `BASIS_POINT`.	required

Returns:

Name	Type	Description
`RateCurve`	`RateCurve`	The shifted curve. The original is unchanged.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

def shifted(self,
            bump: float) -> "RateCurve":
    """A copy with every pillar moved by *bump*.

    The parallel shift a DV01 is measured against.

    Args:
        bump: Amount to add to every rate, in decimal. One basis point is
            ``BASIS_POINT``.

    Returns:
        RateCurve: The shifted curve. The original is unchanged.
    """
    return RateCurve(tenors=self.tenors,
                     rates=tuple(rate + bump for rate in self.rates))

with_pillar_bump ¶

with_pillar_bump(tenor: float, bump: float) -> RateCurve

A copy with one pillar moved, for a key-rate sensitivity.

Parameters:

Name	Type	Description	Default
`tenor`	`float`	The pillar to move. Must be an existing pillar — bumping a tenor that is not there would silently add a pillar and change the curve's shape rather than its level, which is not what a key-rate bump means.	required
`bump`	`float`	Amount to add to that pillar's rate.	required

Returns:

Name	Type	Description
`RateCurve`	`RateCurve`	The bumped curve.

Raises:

Type	Description
`CalculationError`	If tenor is not a pillar.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

def with_pillar_bump(self,
                     tenor: float,
                     bump: float) -> "RateCurve":
    """A copy with one pillar moved, for a key-rate sensitivity.

    Args:
        tenor: The pillar to move. Must be an existing pillar — bumping a
            tenor that is not there would silently add a pillar and change
            the curve's shape rather than its level, which is not what a
            key-rate bump means.
        bump: Amount to add to that pillar's rate.

    Returns:
        RateCurve: The bumped curve.

    Raises:
        CalculationError: If *tenor* is not a pillar.
    """
    for position, pillar in enumerate(self.tenors):
        if abs(pillar - tenor) <= TENOR_TOLERANCE:
            rates = list(self.rates)
            rates[position] += bump

            return RateCurve(tenors=self.tenors, rates=tuple(rates))

    raise CalculationError(
        "RateCurve",
        f"{tenor} is not a pillar on this curve. Available: "
        f"{', '.join(str(pillar) for pillar in self.tenors)}.")

to_dict ¶

to_dict() -> dict[float, float]

The pillars as a {tenor: rate} mapping.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/curves.py

def to_dict(self) -> dict[float, float]:
    """The pillars as a ``{tenor: rate}`` mapping."""
    return dict(zip(self.tenors, self.rates, strict=True))

ETFFuture ¶

ETFFuture(
    derivative_id: str,
    underlying_id: str,
    currency: str,
    expiry_date: str,
    contract_multiplier: float,
    tick_size: float,
    tick_value: float,
)

Bases: IndexFuture

A futures contract on an ETF.

Behaves like :class:IndexFuture but prices with discrete known dividends, which better reflects an ETF's periodic cash distributions than a continuous yield. When discrete dividends are supplied via the market_data key "discrete_dividends" (a list of (time_to_ex_years, amount) tuples for ex-dates within the tenor), fair value uses F = (S - PV(divs)) * exp(r * T). Otherwise it falls back to the continuous cost-of-carry model inherited from :class:IndexFuture.

Initialise an ETF future. See :class:IndexFuture for the args.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def __init__(self,
             derivative_id: str,
             underlying_id: str,
             currency: str,
             expiry_date: str,
             contract_multiplier: float,
             tick_size: float,
             tick_value: float):
    """Initialise an ETF future. See :class:`IndexFuture` for the args."""
    super().__init__(
        derivative_id=derivative_id,
        underlying_id=underlying_id,
        currency=currency,
        expiry_date=expiry_date,
        contract_multiplier=contract_multiplier,
        tick_size=tick_size,
        tick_value=tick_value,
        underlying_type="ETF",
    )

fair_value ¶

fair_value(
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> float

Discrete-dividend fair value, falling back to continuous carry.

If market_data["discrete_dividends"] is present and non-empty, prices with the discrete-dividend model; otherwise defers to the continuous cost-of-carry model of :class:IndexFuture.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def fair_value(self,
               spot_price: float,
               valuation_date: pd.Timestamp,
               market_data: dict[str, Any]) -> float:
    """Discrete-dividend fair value, falling back to continuous carry.

    If ``market_data["discrete_dividends"]`` is present and non-empty, prices
    with the discrete-dividend model; otherwise defers to the continuous
    cost-of-carry model of :class:`IndexFuture`.
    """
    market_data = market_data or {}
    dividends = market_data.get("discrete_dividends")
    if not dividends:
        return super().fair_value(spot_price, valuation_date, market_data)

    t = self.time_to_expiry(valuation_date)
    return discrete_dividend_fair_value(
        spot=spot_price,
        risk_free_rate=market_data.get("risk_free_rate", 0.0),
        time_to_expiry_years=t,
        dividends=dividends,
    )

IndexFuture ¶

IndexFuture(
    derivative_id: str,
    underlying_id: str,
    currency: str,
    expiry_date: str,
    contract_multiplier: float,
    tick_size: float,
    tick_value: float,
    underlying_type: str = "INDEX",
)

Bases: DerivativeBase

A cash-settled futures contract on an equity index.

Prices are quoted in index points; currency amounts are obtained by multiplying by :attr:contract_multiplier. Fair value uses the cost-of-carry model from :mod:beacon.derivatives.pricing.

Market-data inputs (passed via the market_data dict on valuation methods) are read by key:

risk_free_rate — continuous risk-free rate r (default 0)
dividend_yield — continuous dividend yield q (default 0)
borrow_cost — continuous borrow/financing spread c (default 0)

Initialise an index future.

Parameters:

Name	Type	Description	Default
`derivative_id`	`str`	Unique identifier for the contract.	required
`underlying_id`	`str`	Identifier of the referenced index.	required
`currency`	`str`	Contract currency (e.g. `USD`).	required
`expiry_date`	`str`	Expiry date (YYYY-MM-DD).	required
`contract_multiplier`	`float`	Currency value of one index point.	required
`tick_size`	`float`	Minimum price increment, in index points.	required
`tick_value`	`float`	Currency value of one tick.	required
`underlying_type`	`str`	Underlying instrument type; defaults to `INDEX`. Subclasses (e.g. :class:`ETFFuture`) override it.	`'INDEX'`

Raises:

Type	Description
`ValueError`	If any of contract_multiplier, tick_size or tick_value is non-positive (plus the base-class validations).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def __init__(self,
             derivative_id: str,
             underlying_id: str,
             currency: str,
             expiry_date: str,
             contract_multiplier: float,
             tick_size: float,
             tick_value: float,
             underlying_type: str = "INDEX"):
    """Initialise an index future.

    Args:
        derivative_id: Unique identifier for the contract.
        underlying_id: Identifier of the referenced index.
        currency: Contract currency (e.g. ``USD``).
        expiry_date: Expiry date (YYYY-MM-DD).
        contract_multiplier: Currency value of one index point.
        tick_size: Minimum price increment, in index points.
        tick_value: Currency value of one tick.
        underlying_type: Underlying instrument type; defaults to ``INDEX``.
            Subclasses (e.g. :class:`ETFFuture`) override it.

    Raises:
        ValueError: If any of *contract_multiplier*, *tick_size* or
            *tick_value* is non-positive (plus the base-class validations).
    """
    if contract_multiplier <= 0:
        raise ValueError(
            f"contract_multiplier must be positive, got {contract_multiplier}."
        )
    if tick_size <= 0:
        raise ValueError(f"tick_size must be positive, got {tick_size}.")
    if tick_value <= 0:
        raise ValueError(f"tick_value must be positive, got {tick_value}.")

    # The per-point multiplier stands in as the contract notional for the base.
    super().__init__(
        derivative_id=derivative_id,
        underlying_id=underlying_id,
        underlying_type=underlying_type,
        currency=currency,
        expiry_date=expiry_date,
        notional=contract_multiplier,
    )

    self.contract_multiplier: float = contract_multiplier
    self.tick_size: float = tick_size
    self.tick_value: float = tick_value

fair_value ¶

fair_value(
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> float

Cost-of-carry fair value F = S * exp((r - q + c) * T) in points.

Returns spot_price when the contract is at or past expiry (T == 0).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def fair_value(self,
               spot_price: float,
               valuation_date: pd.Timestamp,
               market_data: dict[str, Any]) -> float:
    """Cost-of-carry fair value ``F = S * exp((r - q + c) * T)`` in points.

    Returns *spot_price* when the contract is at or past expiry (``T == 0``).
    """
    market_data = market_data or {}
    t = self.time_to_expiry(valuation_date)
    return cost_of_carry_fair_value(
        spot=spot_price,
        risk_free_rate=market_data.get("risk_free_rate", 0.0),
        dividend_yield=market_data.get("dividend_yield", 0.0),
        time_to_expiry_years=t,
        borrow_cost=market_data.get("borrow_cost", 0.0),
    )

basis ¶

basis(futures_price: float, spot_price: float) -> float

Simple basis: futures_price - spot_price (index points).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def basis(self,
          futures_price: float,
          spot_price: float) -> float:
    """Simple basis: ``futures_price - spot_price`` (index points)."""
    return futures_price - spot_price

annualised_basis ¶

annualised_basis(
    futures_price: float,
    spot_price: float,
    valuation_date: Timestamp,
) -> float

Annualised implied financing rate ln(F / S) / T.

Implemented via :func:implied_repo_rate with zero dividend yield.

Raises:

Type	Description
`ValueError`	At or past expiry (`T == 0`), where the rate is undefined.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def annualised_basis(self,
                     futures_price: float,
                     spot_price: float,
                     valuation_date: pd.Timestamp) -> float:
    """Annualised implied financing rate ``ln(F / S) / T``.

    Implemented via :func:`implied_repo_rate` with zero dividend yield.

    Raises:
        ValueError: At or past expiry (``T == 0``), where the rate is
            undefined.
    """
    t = self.time_to_expiry(valuation_date)
    return implied_repo_rate(
        futures_price=futures_price,
        spot=spot_price,
        dividend_yield=0.0,
        time_to_expiry_years=t,
    )

daily_settlement_pnl ¶

daily_settlement_pnl(
    settle_today: float,
    settle_yesterday: float,
    contracts: float = 1.0,
) -> float

Variation-margin P&L for the day, in contract currency.

(settle_today - settle_yesterday) * contract_multiplier * contracts. Positive contracts is a long position, negative is short.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def daily_settlement_pnl(self,
                         settle_today: float,
                         settle_yesterday: float,
                         contracts: float = 1.0) -> float:
    """Variation-margin P&L for the day, in contract currency.

    ``(settle_today - settle_yesterday) * contract_multiplier * contracts``.
    Positive *contracts* is a long position, negative is short.
    """
    return (settle_today - settle_yesterday) * self.contract_multiplier * contracts

roll_cost ¶

roll_cost(front_price: float, back_price: float) -> float

Cost of rolling from the front to the back contract: back - front.

Positive in contango (back above front), negative in backwardation.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def roll_cost(self,
              front_price: float,
              back_price: float) -> float:
    """Cost of rolling from the front to the back contract: ``back - front``.

    Positive in contango (back above front), negative in backwardation.
    """
    return back_price - front_price

mark_to_market ¶

mark_to_market(
    market_price: float,
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> dict[str, float]

Mark the contract against an observed market_price.

Returns a dict with fair_value (points), basis (market vs spot), theoretical_edge (fair value minus market price), and time_to_expiry (years).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/futures.py

def mark_to_market(self,
                   market_price: float,
                   spot_price: float,
                   valuation_date: pd.Timestamp,
                   market_data: dict[str, Any]) -> dict[str, float]:
    """Mark the contract against an observed *market_price*.

    Returns a dict with ``fair_value`` (points), ``basis`` (market vs spot),
    ``theoretical_edge`` (fair value minus market price), and
    ``time_to_expiry`` (years).
    """
    fv = self.fair_value(spot_price, valuation_date, market_data)
    return {
        "fair_value": fv,
        "basis": self.basis(market_price, spot_price),
        "theoretical_edge": fv - market_price,
        "time_to_expiry": self.time_to_expiry(valuation_date),
    }

TotalReturnSwap ¶

TotalReturnSwap(
    derivative_id: str,
    underlying_id: str,
    currency: str,
    start_date: str,
    end_date: str,
    notional: float,
    spread_bps: float,
    reference_rate: str,
    payment_frequency: str,
    reset_type: str = "UNFUNDED",
)

Bases: DerivativeBase

A total return swap (TRS) on an index or equity basket.

The total-return receiver earns the price return of the underlying and pays a financing leg. For an UNFUNDED swap the financing leg is reference_rate + spread; for a FUNDED swap the principal is posted up front and only the spread accrues.

market_data inputs (read by key on the valuation methods):

initial_price — reference price S_0 at inception/last reset (defaults to spot_price, i.e. zero return)
reference_rate — the floating rate for the current period (default 0)
last_reset_date — start of the current accrual period (defaults to the swap start date)

Initialise a total return swap.

Parameters:

Name	Type	Description	Default
`derivative_id`	`str`	Unique identifier for the swap.	required
`underlying_id`	`str`	Identifier of the referenced index/basket.	required
`currency`	`str`	Contract currency.	required
`start_date`	`str`	Swap start date (YYYY-MM-DD).	required
`end_date`	`str`	Swap maturity date (YYYY-MM-DD); used as the base expiry.	required
`notional`	`float`	Swap notional; must be positive.	required
`spread_bps`	`float`	Financing spread over the reference rate, in basis points.	required
`reference_rate`	`str`	Name/identifier of the floating reference rate (e.g. `SOFR`).	required
`payment_frequency`	`str`	One of `MONTHLY`, `QUARTERLY`, `SEMI-ANNUAL`, `ANNUAL` (case-insensitive).	required
`reset_type`	`str`	`UNFUNDED` (default) or `FUNDED`.	`'UNFUNDED'`

Raises:

Type	Description
`ValueError`	On empty dates, `end_date` not after `start_date`, unrecognised payment_frequency/reset_type, or the base-class validations (including non-positive notional).

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/swaps.py

def __init__(self,
             derivative_id: str,
             underlying_id: str,
             currency: str,
             start_date: str,
             end_date: str,
             notional: float,
             spread_bps: float,
             reference_rate: str,
             payment_frequency: str,
             reset_type: str = "UNFUNDED"):
    """Initialise a total return swap.

    Args:
        derivative_id: Unique identifier for the swap.
        underlying_id: Identifier of the referenced index/basket.
        currency: Contract currency.
        start_date: Swap start date (YYYY-MM-DD).
        end_date: Swap maturity date (YYYY-MM-DD); used as the base expiry.
        notional: Swap notional; must be positive.
        spread_bps: Financing spread over the reference rate, in basis points.
        reference_rate: Name/identifier of the floating reference rate
            (e.g. ``SOFR``).
        payment_frequency: One of ``MONTHLY``, ``QUARTERLY``,
            ``SEMI-ANNUAL``, ``ANNUAL`` (case-insensitive).
        reset_type: ``UNFUNDED`` (default) or ``FUNDED``.

    Raises:
        ValueError: On empty dates, ``end_date`` not after ``start_date``,
            unrecognised *payment_frequency*/*reset_type*, or the base-class
            validations (including non-positive *notional*).
    """
    if not start_date:
        raise ValueError("start_date cannot be empty.")
    if not end_date:
        raise ValueError("end_date cannot be empty.")

    freq = (payment_frequency or "").upper()
    if freq not in self.VALID_PAYMENT_FREQUENCIES:
        raise ValueError(
            f"payment_frequency must be one of "
            f"{sorted(self.VALID_PAYMENT_FREQUENCIES)}, got '{payment_frequency}'."
        )

    reset = (reset_type or "").upper()
    if reset not in self.VALID_RESET_TYPES:
        raise ValueError(
            f"reset_type must be one of {sorted(self.VALID_RESET_TYPES)}, "
            f"got '{reset_type}'."
        )

    # end_date is the contract expiry from the base class's perspective.
    super().__init__(
        derivative_id=derivative_id,
        underlying_id=underlying_id,
        underlying_type="INDEX",
        currency=currency,
        expiry_date=end_date,
        notional=notional,
    )

    self.start_date: pd.Timestamp = pd.Timestamp(start_date)
    self.end_date: pd.Timestamp = pd.Timestamp(end_date)
    if self.end_date <= self.start_date:
        raise ValueError("end_date must be after start_date.")

    self.spread_bps: float = spread_bps
    self.spread: float = spread_bps / 10_000.0
    self.reference_rate: str = reference_rate
    self.payment_frequency: str = freq
    self.reset_type: str = reset

financing_cost ¶

financing_cost(
    valuation_date: Timestamp,
    last_reset_date: Timestamp,
    reference_rate: float,
) -> float

Financing accrued since last_reset_date on an ACT/360 basis.

For an UNFUNDED swap the accrual rate is reference_rate + spread; for a FUNDED swap only the spread accrues.

Parameters:

Name	Type	Description	Default
`valuation_date`	`Timestamp`	The accrual end date.	required
`last_reset_date`	`Timestamp`	Start of the current accrual period.	required
`reference_rate`	`float`	Floating reference rate for the period (decimal).	required

Returns:

Type	Description
`float`	The accrued financing cost in contract currency.

Raises:

Type	Description
`ValueError`	If valuation_date precedes last_reset_date.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/swaps.py

def financing_cost(self,
                   valuation_date: pd.Timestamp,
                   last_reset_date: pd.Timestamp,
                   reference_rate: float) -> float:
    """Financing accrued since *last_reset_date* on an ACT/360 basis.

    For an ``UNFUNDED`` swap the accrual rate is ``reference_rate + spread``;
    for a ``FUNDED`` swap only the ``spread`` accrues.

    Args:
        valuation_date: The accrual end date.
        last_reset_date: Start of the current accrual period.
        reference_rate: Floating reference rate for the period (decimal).

    Returns:
        The accrued financing cost in contract currency.

    Raises:
        ValueError: If *valuation_date* precedes *last_reset_date*.
    """
    days = int((pd.Timestamp(valuation_date) - pd.Timestamp(last_reset_date)).days)
    if days < 0:
        raise ValueError("valuation_date must be on or after last_reset_date.")

    rate = self.spread
    if self.reset_type == "UNFUNDED":
        rate += reference_rate

    day_count_fraction = days / _FINANCING_DAY_COUNT
    return self.notional * rate * day_count_fraction

dv01 ¶

dv01(
    valuation_date: Timestamp,
    last_reset_date: Timestamp,
    reference_rate: float = 0.0,
) -> float

Change in the receiver's value for a one-basis-point rate rise.

Computed by bumping and revaluing rather than by the closed form. The two agree exactly here — financing is linear in the rate — and a test holds them to that. The bump-and-revalue version is the one kept because it stays correct if the financing leg ever stops being linear, and because it is obviously right by inspection.

The sign is negative for a total-return receiver, and that is not a convention choice. The receiver pays financing, so a higher rate makes their position worth less. Reporting DV01 as a positive magnitude is common, but it loses the one piece of information a risk report most needs: which way this position hurts.

A FUNDED swap returns 0.0. Only the spread accrues on one, and the spread does not move with the reference rate — so the position genuinely has no sensitivity to it, rather than a small one.

Parameters:

Name	Type	Description	Default
`valuation_date`	`Timestamp`	The accrual end date.	required
`last_reset_date`	`Timestamp`	Start of the current accrual period.	required
`reference_rate`	`float`	The floating rate the bump is applied to. The answer does not depend on its level, since financing is linear, but it is accepted so the call reads the same as the others.	`0.0`

Returns:

Name	Type	Description
`float`	`float`	Value change per +1bp, in contract currency. Negative for a
	`float`	receiver on an unfunded swap.

Raises:

Type	Description
`ValueError`	If valuation_date precedes last_reset_date.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/swaps.py

def dv01(self,
         valuation_date: pd.Timestamp,
         last_reset_date: pd.Timestamp,
         reference_rate: float = 0.0) -> float:
    """Change in the receiver's value for a one-basis-point rate rise.

    Computed by bumping and revaluing rather than by the closed form. The
    two agree exactly here — financing is linear in the rate — and a test
    holds them to that. The bump-and-revalue version is the one kept
    because it stays correct if the financing leg ever stops being linear,
    and because it is obviously right by inspection.

    **The sign is negative for a total-return receiver**, and that is not a
    convention choice. The receiver *pays* financing, so a higher rate
    makes their position worth less. Reporting DV01 as a positive magnitude
    is common, but it loses the one piece of information a risk report most
    needs: which way this position hurts.

    A ``FUNDED`` swap returns 0.0. Only the spread accrues on one, and the
    spread does not move with the reference rate — so the position genuinely
    has no sensitivity to it, rather than a small one.

    Args:
        valuation_date: The accrual end date.
        last_reset_date: Start of the current accrual period.
        reference_rate: The floating rate the bump is applied to. The
            answer does not depend on its level, since financing is linear,
            but it is accepted so the call reads the same as the others.

    Returns:
        float: Value change per +1bp, in contract currency. Negative for a
        receiver on an unfunded swap.

    Raises:
        ValueError: If *valuation_date* precedes *last_reset_date*.
    """
    base = self.financing_cost(valuation_date, last_reset_date, reference_rate)
    bumped = self.financing_cost(valuation_date, last_reset_date,
                                 reference_rate + _ONE_BASIS_POINT)

    # Financing is a cost to the receiver, so more of it is less value.
    # Subtracting this way round rather than negating the difference keeps
    # a zero-sensitivity funded swap at 0.0 instead of -0.0.
    return base - bumped

financing_duration ¶

financing_duration(
    valuation_date: Timestamp, last_reset_date: Timestamp
) -> float

The accrual year fraction the DV01 scales with, ACT/360.

Exposed because it is the whole of the DV01 story: the sensitivity is notional × 1bp × this, so a reader who wants to check the number by hand needs it rather than having to rederive the day count.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/swaps.py

def financing_duration(self,
                       valuation_date: pd.Timestamp,
                       last_reset_date: pd.Timestamp) -> float:
    """The accrual year fraction the DV01 scales with, ACT/360.

    Exposed because it is the whole of the DV01 story: the sensitivity is
    notional × 1bp × this, so a reader who wants to check the number by hand
    needs it rather than having to rederive the day count.
    """
    days = int((pd.Timestamp(valuation_date) - pd.Timestamp(last_reset_date)).days)
    if days < 0:
        raise ValueError("valuation_date must be on or after last_reset_date.")

    return days / _FINANCING_DAY_COUNT

fair_value ¶

fair_value(
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> float

Total-return-receiver P&L: total return leg minus accrued financing.

receiver_pnl = notional * (S_t / S_0 - 1) - accrued_financing

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/swaps.py

def fair_value(self,
               spot_price: float,
               valuation_date: pd.Timestamp,
               market_data: dict[str, Any]) -> float:
    """Total-return-receiver P&L: total return leg minus accrued financing.

    ``receiver_pnl = notional * (S_t / S_0 - 1) - accrued_financing``
    """
    market_data = market_data or {}
    s0 = float(market_data.get("initial_price", spot_price))
    if s0 <= 0:
        raise ValueError(f"initial_price must be positive, got {s0}.")

    last_reset = market_data.get("last_reset_date", self.start_date)
    reference_rate = market_data.get("reference_rate", 0.0)

    total_return_leg = self.notional * (spot_price / s0 - 1.0)
    financing = self.financing_cost(valuation_date, last_reset, reference_rate)
    return total_return_leg - financing

mark_to_market ¶

mark_to_market(
    market_price: float,
    spot_price: float,
    valuation_date: Timestamp,
    market_data: dict[str, Any],
) -> dict[str, float]

Decompose the swap P&L into its legs.

market_price is unused (a TRS has no separately quoted price); it is accepted to satisfy the :class:DerivativeBase interface.

Returns a dict with total_return_leg, financing_leg, net_mtm and accrued_days.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/swaps.py

def mark_to_market(self,
                   market_price: float,
                   spot_price: float,
                   valuation_date: pd.Timestamp,
                   market_data: dict[str, Any]) -> dict[str, float]:
    """Decompose the swap P&L into its legs.

    *market_price* is unused (a TRS has no separately quoted price); it is
    accepted to satisfy the :class:`DerivativeBase` interface.

    Returns a dict with ``total_return_leg``, ``financing_leg``,
    ``net_mtm`` and ``accrued_days``.
    """
    market_data = market_data or {}
    s0 = market_data.get("initial_price", spot_price)
    if s0 <= 0:
        raise ValueError(f"initial_price must be positive, got {s0}.")

    last_reset = pd.Timestamp(market_data.get("last_reset_date", self.start_date))
    reference_rate = market_data.get("reference_rate", 0.0)

    total_return_leg = self.notional * (spot_price / s0 - 1.0)
    financing_leg = self.financing_cost(valuation_date, last_reset, reference_rate)
    accrued_days = (pd.Timestamp(valuation_date) - last_reset).days

    return {
        "total_return_leg": total_return_leg,
        "financing_leg": financing_leg,
        "net_mtm": total_return_leg - financing_leg,
        "accrued_days": accrued_days,
    }

FuturesQuote `dataclass` ¶

FuturesQuote(
    expiry: Timestamp,
    market_price: float | None = None,
    label: str = "",
)

One expiry and the price the market puts on it.

Attributes:

Name	Type	Description
`expiry`	`Timestamp`	Contract expiry date.
`market_price`	`float \| None`	Traded price. None when only a theoretical value is wanted, in which case basis and implied repo are not reported for this pillar rather than being invented.
`label`	`str`	Optional contract code, for display.

TermStructure `dataclass` ¶

TermStructure(
    underlying: str,
    spot: float,
    valuation_date: Timestamp,
    quotes: list[FuturesQuote],
    curve: RateCurve,
    dividend_yield: float = 0.0,
    borrow_cost: float = 0.0,
    _sorted: list[FuturesQuote] = list(),
)

A strip of futures on one underlying, valued off one curve.

Attributes:

Name	Type	Description
`underlying`	`str`	Identifier of the underlying.
`spot`	`float`	Spot price at valuation_date.
`valuation_date`	`Timestamp`	The date everything is measured from.
`quotes`	`list[FuturesQuote]`	The expiries, in any order; they are sorted on construction.
`curve`	`RateCurve`	Financing curve. A flat curve reproduces scalar-rate pricing exactly.
`dividend_yield`	`float`	Continuous dividend yield on the underlying.
`borrow_cost`	`float`	Continuous borrow spread.

expiries `property` ¶

expiries: list[Timestamp]

Expiry dates, nearest first.

times_to_expiry ¶

times_to_expiry() -> list[float]

Year fractions to each expiry, ACT/365.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def times_to_expiry(self) -> list[float]:
    """Year fractions to each expiry, ACT/365."""
    return [(pd.Timestamp(quote.expiry) - self.valuation_date).days / DAYS_PER_YEAR
            for quote in self._sorted]

financing_rates ¶

financing_rates() -> list[float]

The curve's rate at each expiry.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def financing_rates(self) -> list[float]:
    """The curve's rate at each expiry."""
    return [self.curve.zero_rate(tenor) for tenor in self.times_to_expiry()]

theoretical_prices ¶

theoretical_prices() -> pd.Series

Fair value at each expiry, off the curve.

Returns:

Type	Description
`Series`	pd.Series: Indexed by expiry date.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def theoretical_prices(self) -> pd.Series:
    """Fair value at each expiry, off the curve.

    Returns:
        pd.Series: Indexed by expiry date.
    """
    values = [
        cost_of_carry_fair_value(spot=self.spot,
                                 risk_free_rate=rate,
                                 dividend_yield=self.dividend_yield,
                                 time_to_expiry_years=tenor,
                                 borrow_cost=self.borrow_cost)
        for rate, tenor in zip(self.financing_rates(),
                               self.times_to_expiry(),
                               strict=True)
    ]

    return pd.Series(values, index=self.expiries, name="theoretical")

market_prices ¶

market_prices() -> pd.Series

Quoted prices, NaN where a quote carries none.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def market_prices(self) -> pd.Series:
    """Quoted prices, NaN where a quote carries none."""
    return pd.Series([quote.market_price for quote in self._sorted],
                     index=self.expiries,
                     dtype=float,
                     name="market")

basis ¶

basis() -> pd.Series

Market minus theoretical, per expiry.

Positive means the contract trades rich to the model.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def basis(self) -> pd.Series:
    """Market minus theoretical, per expiry.

    Positive means the contract trades rich to the model.
    """
    return (self.market_prices() - self.theoretical_prices()).rename("basis")

implied_repo ¶

implied_repo() -> pd.Series

The financing rate each quoted price implies.

NaN for expiries with no quote, and for an expiry today — a zero year fraction carries no information about a rate, and dividing by it would manufacture one.

Returns:

Type	Description
`Series`	pd.Series: Continuously compounded rates, indexed by expiry.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def implied_repo(self) -> pd.Series:
    """The financing rate each quoted price implies.

    NaN for expiries with no quote, and for an expiry today — a zero year
    fraction carries no information about a rate, and dividing by it would
    manufacture one.

    Returns:
        pd.Series: Continuously compounded rates, indexed by expiry.
    """
    rates: list[float] = []

    for quote, tenor in zip(self._sorted, self.times_to_expiry(), strict=True):
        if quote.market_price is None or tenor <= 0.0:
            rates.append(float("nan"))
            continue

        rates.append(implied_repo_rate(futures_price=quote.market_price,
                                       spot=self.spot,
                                       dividend_yield=self.dividend_yield,
                                       time_to_expiry_years=tenor))

    return pd.Series(rates, index=self.expiries, name="implied_repo")

to_frame ¶

to_frame() -> pd.DataFrame

Everything the strip says, one row per expiry.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def to_frame(self) -> pd.DataFrame:
    """Everything the strip says, one row per expiry."""
    frame = pd.DataFrame({
        "label": [quote.label for quote in self._sorted],
        "time_to_expiry": self.times_to_expiry(),
        "financing_rate": self.financing_rates(),
        "theoretical": self.theoretical_prices().to_numpy(),
        "market": self.market_prices().to_numpy(),
        "basis": self.basis().to_numpy(),
        "implied_repo": self.implied_repo().to_numpy(),
    }, index=self.expiries)
    frame.index.name = "expiry"

    return frame

cost_of_carry_fair_value ¶

cost_of_carry_fair_value(
    spot: float,
    risk_free_rate: float,
    dividend_yield: float,
    time_to_expiry_years: float,
    borrow_cost: float = 0.0,
) -> float

Fair forward/futures value under continuous cost of carry.

F = S * exp((r - q + c) * T)

Parameters:

Name	Type	Description	Default
`spot`	`float`	Current spot price `S` (must be non-negative).	required
`risk_free_rate`	`float`	Continuously compounded risk-free rate `r`.	required
`dividend_yield`	`float`	Continuous dividend yield `q`.	required
`time_to_expiry_years`	`float`	Time to expiry `T` in years (must be non-negative).	required
`borrow_cost`	`float`	Continuous borrow/financing spread `c` (default 0).	`0.0`

Returns:

Type	Description
`float`	The fair value `F`. Equals spot when `T == 0`.

Raises:

Type	Description
`ValueError`	If spot or time_to_expiry_years is negative.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/pricing.py

def cost_of_carry_fair_value(spot: float,
                             risk_free_rate: float,
                             dividend_yield: float,
                             time_to_expiry_years: float,
                             borrow_cost: float = 0.0) -> float:
    """Fair forward/futures value under continuous cost of carry.

    ``F = S * exp((r - q + c) * T)``

    Args:
        spot: Current spot price ``S`` (must be non-negative).
        risk_free_rate: Continuously compounded risk-free rate ``r``.
        dividend_yield: Continuous dividend yield ``q``.
        time_to_expiry_years: Time to expiry ``T`` in years (must be non-negative).
        borrow_cost: Continuous borrow/financing spread ``c`` (default 0).

    Returns:
        The fair value ``F``. Equals *spot* when ``T == 0``.

    Raises:
        ValueError: If *spot* or *time_to_expiry_years* is negative.
    """
    if spot < 0:
        raise ValueError(f"spot must be non-negative, got {spot}")
    if time_to_expiry_years < 0:
        raise ValueError(
            f"time_to_expiry_years must be non-negative, got {time_to_expiry_years}"
        )

    carry = risk_free_rate - dividend_yield + borrow_cost
    return spot * math.exp(carry * time_to_expiry_years)

discrete_dividend_fair_value ¶

discrete_dividend_fair_value(
    spot: float,
    risk_free_rate: float,
    time_to_expiry_years: float,
    dividends: list[tuple[float, float]],
) -> float

Fair forward/futures value with discrete cash dividends.

F = (S - PV(divs)) * exp(r * T) where each dividend is discounted at the risk-free rate to today: PV = amount * exp(-r * t_ex).

Parameters:

Name	Type	Description	Default
`spot`	`float`	Current spot price `S` (must be non-negative).	required
`risk_free_rate`	`float`	Continuously compounded risk-free rate `r`.	required
`time_to_expiry_years`	`float`	Time to expiry `T` in years (must be non-negative).	required
`dividends`	`list[tuple[float, float]]`	List of `(time_to_ex_years, amount)` tuples. Only dividends with ex-dates on or before expiry (`0 <= t_ex <= T`) are included.	required

Returns:

Type	Description
`float`	The fair value `F`.

Raises:

Type	Description
`ValueError`	If spot or time_to_expiry_years is negative.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/pricing.py

def discrete_dividend_fair_value(spot: float,
                                 risk_free_rate: float,
                                 time_to_expiry_years: float,
                                 dividends: list[tuple[float, float]]) -> float:
    """Fair forward/futures value with discrete cash dividends.

    ``F = (S - PV(divs)) * exp(r * T)`` where each dividend is discounted at the
    risk-free rate to today: ``PV = amount * exp(-r * t_ex)``.

    Args:
        spot: Current spot price ``S`` (must be non-negative).
        risk_free_rate: Continuously compounded risk-free rate ``r``.
        time_to_expiry_years: Time to expiry ``T`` in years (must be non-negative).
        dividends: List of ``(time_to_ex_years, amount)`` tuples. Only dividends
            with ex-dates on or before expiry (``0 <= t_ex <= T``) are included.

    Returns:
        The fair value ``F``.

    Raises:
        ValueError: If *spot* or *time_to_expiry_years* is negative.
    """
    if spot < 0:
        raise ValueError(f"spot must be non-negative, got {spot}")
    if time_to_expiry_years < 0:
        raise ValueError(
            f"time_to_expiry_years must be non-negative, got {time_to_expiry_years}"
        )

    pv_dividends = 0.0
    for t_ex, amount in dividends:
        if 0.0 <= t_ex <= time_to_expiry_years:
            pv_dividends += amount * math.exp(-risk_free_rate * t_ex)

    return (spot - pv_dividends) * math.exp(risk_free_rate * time_to_expiry_years)

futures_roll_return ¶

futures_roll_return(
    front_price: float,
    back_price: float,
    front_expiry: Timestamp,
    back_expiry: Timestamp,
) -> float

Annualised simple roll return from rolling a front contract to a back one.

roll = (front / back - 1) / dt where dt is the year fraction between the two expiries. Positive in backwardation (front above back), negative in contango.

Parameters:

Name	Type	Description	Default
`front_price`	`float`	Price of the near (front) contract (must be positive).	required
`back_price`	`float`	Price of the far (back) contract (must be positive).	required
`front_expiry`	`Timestamp`	Expiry of the front contract.	required
`back_expiry`	`Timestamp`	Expiry of the back contract (must be after front_expiry).	required

Returns:

Type	Description
`float`	The annualised roll return as a decimal.

Raises:

Type	Description
`ValueError`	If either price is non-positive, or back_expiry is not strictly after front_expiry.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/pricing.py

def futures_roll_return(front_price: float,
                        back_price: float,
                        front_expiry: pd.Timestamp,
                        back_expiry: pd.Timestamp) -> float:
    """Annualised simple roll return from rolling a front contract to a back one.

    ``roll = (front / back - 1) / dt`` where ``dt`` is the year fraction between
    the two expiries. Positive in backwardation (front above back), negative in
    contango.

    Args:
        front_price: Price of the near (front) contract (must be positive).
        back_price: Price of the far (back) contract (must be positive).
        front_expiry: Expiry of the front contract.
        back_expiry: Expiry of the back contract (must be after *front_expiry*).

    Returns:
        The annualised roll return as a decimal.

    Raises:
        ValueError: If either price is non-positive, or *back_expiry* is not
            strictly after *front_expiry*.
    """
    if front_price <= 0:
        raise ValueError(f"front_price must be positive, got {front_price}")
    if back_price <= 0:
        raise ValueError(f"back_price must be positive, got {back_price}")

    dt_years = float((back_expiry - front_expiry).total_seconds()) / _SECONDS_PER_YEAR
    if dt_years <= 0:
        raise ValueError("back_expiry must be strictly after front_expiry.")

    return (front_price / back_price - 1.0) / dt_years

implied_repo_rate ¶

implied_repo_rate(
    futures_price: float,
    spot: float,
    dividend_yield: float,
    time_to_expiry_years: float,
) -> float

Continuously compounded financing rate implied by a futures price.

Inverts the cost-of-carry relationship: r_implied = (ln(F / S) + q * T) / T

Parameters:

Name	Type	Description	Default
`futures_price`	`float`	Observed futures price `F` (must be positive).	required
`spot`	`float`	Current spot price `S` (must be positive).	required
`dividend_yield`	`float`	Continuous dividend yield `q`.	required
`time_to_expiry_years`	`float`	Time to expiry `T` in years (must be positive).	required

Returns:

Type	Description
`float`	The implied repo (financing) rate.

Raises:

Type	Description
`ValueError`	If time_to_expiry_years, spot or futures_price is non-positive.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/pricing.py

def implied_repo_rate(futures_price: float,
                      spot: float,
                      dividend_yield: float,
                      time_to_expiry_years: float) -> float:
    """Continuously compounded financing rate implied by a futures price.

    Inverts the cost-of-carry relationship:
    ``r_implied = (ln(F / S) + q * T) / T``

    Args:
        futures_price: Observed futures price ``F`` (must be positive).
        spot: Current spot price ``S`` (must be positive).
        dividend_yield: Continuous dividend yield ``q``.
        time_to_expiry_years: Time to expiry ``T`` in years (must be positive).

    Returns:
        The implied repo (financing) rate.

    Raises:
        ValueError: If *time_to_expiry_years*, *spot* or *futures_price* is
            non-positive.
    """
    if time_to_expiry_years <= 0:
        raise ValueError(
            f"time_to_expiry_years must be positive, got {time_to_expiry_years}"
        )
    if spot <= 0:
        raise ValueError(f"spot must be positive, got {spot}")
    if futures_price <= 0:
        raise ValueError(f"futures_price must be positive, got {futures_price}")

    return (math.log(futures_price / spot) + dividend_yield * time_to_expiry_years) \
        / time_to_expiry_years

trs_breakeven_spread ¶

trs_breakeven_spread(
    futures_price: float,
    spot: float,
    risk_free_rate: float,
    time_to_expiry_years: float,
    dividend_yield: float,
) -> float

Financing spread at which a total return swap matches futures economics.

The futures price embeds an implied financing rate (:func:implied_repo_rate). A TRS financed at r + spread reproduces those economics when the spread equals the gap between the implied financing rate and the risk-free rate:

spread = implied_repo_rate(F, S, q, T) - r

A fairly priced future (financed exactly at r) gives a breakeven spread of zero.

Parameters:

Name	Type	Description	Default
`futures_price`	`float`	Observed futures price `F` (must be positive).	required
`spot`	`float`	Current spot price `S` (must be positive).	required
`risk_free_rate`	`float`	Continuously compounded risk-free rate `r`.	required
`time_to_expiry_years`	`float`	Time to expiry `T` in years (must be positive).	required
`dividend_yield`	`float`	Continuous dividend yield `q`.	required

Returns:

Type	Description
`float`	The breakeven financing spread as a decimal.

Raises:

Type	Description
`ValueError`	If time_to_expiry_years, spot or futures_price is non-positive.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/pricing.py

def trs_breakeven_spread(futures_price: float,
                         spot: float,
                         risk_free_rate: float,
                         time_to_expiry_years: float,
                         dividend_yield: float) -> float:
    """Financing spread at which a total return swap matches futures economics.

    The futures price embeds an implied financing rate (:func:`implied_repo_rate`).
    A TRS financed at ``r + spread`` reproduces those economics when the spread
    equals the gap between the implied financing rate and the risk-free rate:

    ``spread = implied_repo_rate(F, S, q, T) - r``

    A fairly priced future (financed exactly at ``r``) gives a breakeven spread
    of zero.

    Args:
        futures_price: Observed futures price ``F`` (must be positive).
        spot: Current spot price ``S`` (must be positive).
        risk_free_rate: Continuously compounded risk-free rate ``r``.
        time_to_expiry_years: Time to expiry ``T`` in years (must be positive).
        dividend_yield: Continuous dividend yield ``q``.

    Returns:
        The breakeven financing spread as a decimal.

    Raises:
        ValueError: If *time_to_expiry_years*, *spot* or *futures_price* is
            non-positive.
    """
    implied = implied_repo_rate(
        futures_price, spot, dividend_yield, time_to_expiry_years
    )
    return implied - risk_free_rate

sensitivity_grid ¶

sensitivity_grid(
    spot: float,
    tenors: list[float],
    rates: list[float],
    dividend_yield: float = 0.0,
    borrow_cost: float = 0.0,
) -> pd.DataFrame

Fair value across a tenor × rate grid.

What a position is worth if the curve is somewhere else and expiry is further out — the two axes a Delta-1 desk actually moves along, laid out so the shape is visible at once rather than one revaluation at a time.

Parameters:

Name	Type	Description	Default
`spot`	`float`	Spot price of the underlying.	required
`tenors`	`list[float]`	Times to expiry in years, one per row.	required
`rates`	`list[float]`	Continuously compounded financing rates, one per column.	required
`dividend_yield`	`float`	Continuous dividend yield.	`0.0`
`borrow_cost`	`float`	Continuous borrow spread.	`0.0`

Returns:

Type	Description
`DataFrame`	pd.DataFrame: Fair values, tenors on the index and rates on the
`DataFrame`	columns, both labelled with their values.

Raises:

Type	Description
`CalculationError`	If either axis is empty.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/derivatives/term_structure.py

def sensitivity_grid(spot: float,
                     tenors: list[float],
                     rates: list[float],
                     dividend_yield: float = 0.0,
                     borrow_cost: float = 0.0) -> pd.DataFrame:
    """Fair value across a tenor × rate grid.

    What a position is worth if the curve is somewhere else and expiry is
    further out — the two axes a Delta-1 desk actually moves along, laid out so
    the shape is visible at once rather than one revaluation at a time.

    Args:
        spot: Spot price of the underlying.
        tenors: Times to expiry in years, one per row.
        rates: Continuously compounded financing rates, one per column.
        dividend_yield: Continuous dividend yield.
        borrow_cost: Continuous borrow spread.

    Returns:
        pd.DataFrame: Fair values, tenors on the index and rates on the
        columns, both labelled with their values.

    Raises:
        CalculationError: If either axis is empty.
    """
    if not tenors or not rates:
        raise CalculationError("SensitivityGrid",
                               "both tenors and rates must be non-empty.")

    grid = [
        [cost_of_carry_fair_value(spot=spot,
                                  risk_free_rate=rate,
                                  dividend_yield=dividend_yield,
                                  time_to_expiry_years=tenor,
                                  borrow_cost=borrow_cost)
         for rate in rates]
        for tenor in tenors
    ]

    frame = pd.DataFrame(grid, index=list(tenors), columns=list(rates))
    frame.index.name = "time_to_expiry"
    frame.columns.name = "rate"

    return frame

beacon.derivatives¶

derivatives ¶

DerivativeBase ¶

time_to_expiry ¶

fair_value abstractmethod ¶

mark_to_market abstractmethod ¶

RateCurve dataclass ¶

is_flat property ¶

flat classmethod ¶

from_pillars classmethod ¶

zero_rate ¶

discount_factor ¶

forward_rate ¶

shifted ¶

with_pillar_bump ¶

to_dict ¶

ETFFuture ¶

fair_value ¶

IndexFuture ¶

fair_value ¶

basis ¶

annualised_basis ¶

daily_settlement_pnl ¶

roll_cost ¶

mark_to_market ¶

TotalReturnSwap ¶

financing_cost ¶

dv01 ¶

financing_duration ¶

fair_value ¶

mark_to_market ¶

FuturesQuote dataclass ¶

TermStructure dataclass ¶

expiries property ¶

times_to_expiry ¶

financing_rates ¶

theoretical_prices ¶

market_prices ¶

basis ¶

implied_repo ¶

to_frame ¶

cost_of_carry_fair_value ¶

discrete_dividend_fair_value ¶

futures_roll_return ¶

implied_repo_rate ¶

trs_breakeven_spread ¶

sensitivity_grid ¶

fair_value `abstractmethod` ¶

mark_to_market `abstractmethod` ¶

RateCurve `dataclass` ¶

is_flat `property` ¶

flat `classmethod` ¶

from_pillars `classmethod` ¶

FuturesQuote `dataclass` ¶

TermStructure `dataclass` ¶

expiries `property` ¶