Skip to content

Correlated geometric Brownian motions

Two assets whose prices follow geometric Brownian motions driven by correlated Wiener processes.

  • Name: correlated_geometric_brownian_motion
  • Field: economics/finance
  • Formalism: stochastic differential equations
  • References: Black and Scholes 1973; Merton 1973

The correlated_geometric_brownian_motion example

Example

Factory

@register_model(
    field='economics/finance',
    tags=('asset', 'sde', 'correlated noise'),
    references=('Black and Scholes 1973', 'Merton 1973'),
    defaults={'until': 1.0, 'dt': 0.01, 'solver': EulerMaruyamaSolver(dt=1e-3)},
)
def correlated_geometric_brownian_motion(
    prices: tuple[float, float] = (100.0, 100.0),
    *,
    mu: tuple[float, float] = (0.05, 0.08),
    sigma: tuple[float, float] = (0.2, 0.3),
    rho: float = 0.7,
) -> System:
    """Return two assets whose prices follow geometric Brownian motions driven by correlated Wiener processes.

    A `WienerProcessSpec` of dimension 2 with the covariance `[[1, rho], [rho, 1]]` drives the two prices, so their log
    returns over a short time have the correlation `rho`.

    Args:
        prices:
            The initial prices of the two assets.

        mu:
            The drifts of the two assets.

        sigma:
            The volatilities of the two assets.

        rho:
            The correlation of the two Wiener processes, from -1 to 1.

    Returns:
        system:
            The model, with the assets `first` and `second`, whose motions share the Wiener processes `market`.

    Examples:
        >>> from skcomplex.models.economics.finance import correlated_geometric_brownian_motion
        >>> correlated_geometric_brownian_motion(rho=0.5).interactions[1].process.covariance
        [[1.0, 0.5], [0.5, 1.0]]
    """
    labels = ('first', 'second')
    market = WienerProcessSpec('market', dimension=2, covariance=[[1.0, rho], [rho, 1.0]])
    return System(
        elements=[AssetElement(label, price=price) for label, price in zip(labels, prices, strict=True)],
        interactions=[
            GeometricBrownianMotionInteraction(
                f'{label}_motion',
                target=label,
                process=market,
                index=index,
                mu=drift,
                sigma=volatility,
            )
            for index, (label, drift, volatility) in enumerate(zip(labels, mu, sigma, strict=True))
        ],
    )

Components

class AssetElement(Element):
    """Traded asset described by its price.

    Args:
        label:
            The label of the asset.

        visual:
            How the element is drawn, or `None` for `default_visual_`.

        price:
            The initial price.

    Examples:
        >>> from skcomplex.models.economics.finance import AssetElement
        >>> AssetElement('stock', price=100.0).price
        100.0
    """

    def __init__(
        self,
        label: str,
        visual: VisualSpec | None = None,
        *,
        price: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 1.0,
    ) -> None:
        super().__init__(label, visual=visual)
        self.price = price

class GeometricBrownianMotionInteraction(RateInteraction):
    """Geometric Brownian motion, `dS = mu * S * dt + sigma * S * dW`, of the price of an asset.

    The Wiener process `dW` is independent by default, or process `index` of a `WienerProcessSpec` that correlates it
    with the motions of other assets.

    Args:
        label:
            The label of the interaction.

        target:
            The path of the asset.

        mu:
            The drift.

        sigma:
            The volatility.

        process:
            The Wiener processes, one of which drives the price, or `None` for an independent process.

        index:
            The process of `process` that drives the price, or `None` without processes.

    Examples:
        >>> from skcomplex.models.economics.finance import GeometricBrownianMotionInteraction
        >>> from skcomplex.specs import WienerProcessSpec
        >>> GeometricBrownianMotionInteraction('gbm', target='stock', sigma=0.3).sigma
        0.3
        >>> market = WienerProcessSpec('market', dimension=2)
        >>> GeometricBrownianMotionInteraction('gbm', target='stock', process=market, index=1).check().index
        1
    """

    def __init__(
        self,
        label: str,
        *,
        target: Annotated[str, RoleSpec()],
        mu: Annotated[float, ParameterSpec()] = 0.05,
        sigma: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.2,
        process: Annotated[WienerProcessSpec | None, RoleSpec(noise=True)] = None,
        index: int | None = None,
    ) -> None:
        super().__init__(label, target=target, state='price', process=process, index=index)
        self.mu = mu
        self.sigma = sigma

    def _get_rate(self, namespace: SimpleNamespace) -> object:
        """Return the drift `mu * S`.

        Args:
            namespace:
                The namespace of the symbols the rate can use.

        Returns:
            rate:
                The drift.
        """
        return namespace.mu * namespace.target.price

    def _get_noise(self, namespace: SimpleNamespace) -> object:
        """Return the noise amplitude `sigma * S`.

        Args:
            namespace:
                The namespace of the symbols the noise can use.

        Returns:
            noise:
                The noise amplitude.
        """
        return namespace.sigma * namespace.target.price

Equations

first.price: diff(price_first(t), t) == mu_first_motion*price_first(t)
second.price: diff(price_second(t), t) == mu_second_motion*price_second(t)