Stochastic Process¶

This page gives an overview of all Stochastic Processes available in the library.

quantflow.sp.base.StochasticProcess `pydantic-model` ¶

Bases: BaseModel, ABC

Base class for stochastic processes in continuous time

sample_from_draws `abstractmethod` ¶

sample_from_draws(draws, *args)

Sample Paths from the process given a set of draws

Source code in quantflow/sp/base.py

@abstractmethod
def sample_from_draws(self, draws: Paths, *args: Paths) -> Paths:
    """Sample [Paths][quantflow.ta.paths.Paths]
    from the process given a set of draws"""

sample `abstractmethod` ¶

sample(n, time_horizon=1, time_steps=100)

Generate random Paths from the process.

PARAMETER	DESCRIPTION
`n`	number of paths TYPE: `int`
`time_horizon`	time horizon TYPE: `float` DEFAULT: `1`
`time_steps`	number of time steps to arrive at horizon TYPE: `int` DEFAULT: `100`

Source code in quantflow/sp/base.py

@abstractmethod
def sample(
    self,
    n: Annotated[int, Doc("number of paths")],
    time_horizon: Annotated[float, Doc("time horizon")] = 1,
    time_steps: Annotated[
        int, Doc("number of time steps to arrive at horizon")
    ] = 100,
) -> Paths:
    """Generate random [Paths][quantflow.ta.paths.Paths] from the process."""

characteristic_exponent `abstractmethod` ¶

characteristic_exponent(t, u)

Characteristic exponent at time t for a given input parameter

Source code in quantflow/sp/base.py

@abstractmethod
def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    """Characteristic exponent at time `t` for a given input parameter"""

characteristic ¶

characteristic(t, u)

Characteristic function at time t for a given input parameter

The characteristic function represents the Fourier transform of the probability density function

\[\begin{equation} \phi = {\mathbb E} \left[e^{i u x_t}\right] \end{equation}\]

PARAMETER	DESCRIPTION
`t`	Time horizon TYPE: `FloatArrayLike`
`u`	Characteristic function input parameter TYPE: `Vector`

Source code in quantflow/sp/base.py

def characteristic(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon")],
    u: Annotated[Vector, Doc("Characteristic function input parameter")],
) -> Vector:
    r"""Characteristic function at time `t` for a given input parameter

    The characteristic function represents the Fourier transform of the
    probability density function

    \begin{equation}
        \phi = {\mathbb E} \left[e^{i u x_t}\right]
    \end{equation}
    """
    return np.exp(-self.characteristic_exponent(t, u))

convexity_correction ¶

convexity_correction(t)

Convexity correction for the process

Source code in quantflow/sp/base.py

def convexity_correction(self, t: FloatArrayLike) -> Vector:
    """Convexity correction for the process"""
    return -self.characteristic_exponent(t, complex(0, -1)).real

analytical_std ¶

analytical_std(t)

Source code in quantflow/sp/base.py

def analytical_std(self, t: FloatArrayLike) -> FloatArrayLike:
    return np.sqrt(self.analytical_variance(t))

analytical_mean ¶

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical mean of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_variance ¶

analytical_variance(t)

Analytical variance of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical variance of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_pdf ¶

analytical_pdf(t, x)

Analytical pdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_pdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical pdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_cdf ¶

analytical_cdf(t, x)

Analytical cdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_cdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical cdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

quantflow.sp.base.StochasticProcess1D `pydantic-model` ¶

Bases: StochasticProcess

Base class for 1D stochastic process in continuous time

marginal ¶

marginal(t)

Source code in quantflow/sp/base.py

def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal:
    return StochasticProcess1DMarginal(process=self, t=t)

domain_range ¶

domain_range()

Source code in quantflow/sp/base.py

def domain_range(self) -> Bounds:
    return default_bounds()

frequency_range ¶

frequency_range(std, max_frequency=None)

Maximum frequency when calculating characteristic functions

Source code in quantflow/sp/base.py

def frequency_range(self, std: float, max_frequency: float | None = None) -> Bounds:
    """Maximum frequency when calculating characteristic functions"""
    if max_frequency is None:
        max_frequency = np.sqrt(40 / std / std)
    return Bounds(0, max_frequency)

support ¶

support(mean, std, points)

Support of the process at time t

Source code in quantflow/sp/base.py

def support(self, mean: float, std: float, points: int) -> FloatArray:
    """Support of the process at time `t`"""
    bounds = self.domain_range()
    start = float(sigfig(bound_from_any(bounds.lb, mean - std)))
    end = float(sigfig(bound_from_any(bounds.ub, mean + std)))
    return np.linspace(start, end, points + 1)

sample_from_draws `abstractmethod` ¶

sample_from_draws(draws, *args)

Sample Paths from the process given a set of draws

Source code in quantflow/sp/base.py

@abstractmethod
def sample_from_draws(self, draws: Paths, *args: Paths) -> Paths:
    """Sample [Paths][quantflow.ta.paths.Paths]
    from the process given a set of draws"""

sample `abstractmethod` ¶

sample(n, time_horizon=1, time_steps=100)

Generate random Paths from the process.

PARAMETER	DESCRIPTION
`n`	number of paths TYPE: `int`
`time_horizon`	time horizon TYPE: `float` DEFAULT: `1`
`time_steps`	number of time steps to arrive at horizon TYPE: `int` DEFAULT: `100`

Source code in quantflow/sp/base.py

@abstractmethod
def sample(
    self,
    n: Annotated[int, Doc("number of paths")],
    time_horizon: Annotated[float, Doc("time horizon")] = 1,
    time_steps: Annotated[
        int, Doc("number of time steps to arrive at horizon")
    ] = 100,
) -> Paths:
    """Generate random [Paths][quantflow.ta.paths.Paths] from the process."""

characteristic_exponent `abstractmethod` ¶

characteristic_exponent(t, u)

Characteristic exponent at time t for a given input parameter

Source code in quantflow/sp/base.py

@abstractmethod
def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    """Characteristic exponent at time `t` for a given input parameter"""

characteristic ¶

characteristic(t, u)

Characteristic function at time t for a given input parameter

The characteristic function represents the Fourier transform of the probability density function

\[\begin{equation} \phi = {\mathbb E} \left[e^{i u x_t}\right] \end{equation}\]

PARAMETER	DESCRIPTION
`t`	Time horizon TYPE: `FloatArrayLike`
`u`	Characteristic function input parameter TYPE: `Vector`

Source code in quantflow/sp/base.py

def characteristic(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon")],
    u: Annotated[Vector, Doc("Characteristic function input parameter")],
) -> Vector:
    r"""Characteristic function at time `t` for a given input parameter

    The characteristic function represents the Fourier transform of the
    probability density function

    \begin{equation}
        \phi = {\mathbb E} \left[e^{i u x_t}\right]
    \end{equation}
    """
    return np.exp(-self.characteristic_exponent(t, u))

convexity_correction ¶

convexity_correction(t)

Convexity correction for the process

Source code in quantflow/sp/base.py

def convexity_correction(self, t: FloatArrayLike) -> Vector:
    """Convexity correction for the process"""
    return -self.characteristic_exponent(t, complex(0, -1)).real

analytical_std ¶

analytical_std(t)

Source code in quantflow/sp/base.py

def analytical_std(self, t: FloatArrayLike) -> FloatArrayLike:
    return np.sqrt(self.analytical_variance(t))

analytical_mean ¶

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical mean of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_variance ¶

analytical_variance(t)

Analytical variance of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical variance of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_pdf ¶

analytical_pdf(t, x)

Analytical pdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_pdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical pdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_cdf ¶

analytical_cdf(t, x)

Analytical cdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_cdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical cdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

quantflow.sp.base.IntensityProcess `pydantic-model` ¶

Bases: StochasticProcess1D

Base class for mean reverting 1D processes which can be used as stochastic intensity

Fields:

rate (float)
kappa (float)

rate `pydantic-field` ¶

rate = 1.0

Instantaneous initial rate :math:r_0

kappa `pydantic-field` ¶

kappa = 1.0

Mean reversion speed :math:\kappa

integrated_log_laplace `abstractmethod` ¶

integrated_log_laplace(t, u)

The log-Laplace transform of the cumulative process:

.. math:: e^{\phi_{t, u}} = {\mathbb E} \left[e^{i u \int_0^t x_s ds}\right]

PARAMETER	DESCRIPTION
`t`	time horizon TYPE: `FloatArrayLike`
`u`	frequency TYPE: `Vector`

Source code in quantflow/sp/base.py

@abstractmethod
def integrated_log_laplace(
    self,
    t: Annotated[FloatArrayLike, Doc("time horizon")],
    u: Annotated[Vector, Doc("frequency")],
) -> Vector:
    r"""The log-Laplace transform of the cumulative process:

    .. math::
        e^{\phi_{t, u}} = {\mathbb E} \left[e^{i u \int_0^t x_s ds}\right]
    """

domain_range ¶

domain_range()

Source code in quantflow/sp/base.py

def domain_range(self) -> Bounds:
    return Bounds(0, np.inf)

ekt ¶

ekt(t)

Source code in quantflow/sp/base.py

def ekt(self, t: FloatArrayLike) -> FloatArrayLike:
    return np.exp(-self.kappa * t)

sample_from_draws `abstractmethod` ¶

sample_from_draws(draws, *args)

Sample Paths from the process given a set of draws

Source code in quantflow/sp/base.py

@abstractmethod
def sample_from_draws(self, draws: Paths, *args: Paths) -> Paths:
    """Sample [Paths][quantflow.ta.paths.Paths]
    from the process given a set of draws"""

sample `abstractmethod` ¶

sample(n, time_horizon=1, time_steps=100)

Generate random Paths from the process.

PARAMETER	DESCRIPTION
`n`	number of paths TYPE: `int`
`time_horizon`	time horizon TYPE: `float` DEFAULT: `1`
`time_steps`	number of time steps to arrive at horizon TYPE: `int` DEFAULT: `100`

Source code in quantflow/sp/base.py

@abstractmethod
def sample(
    self,
    n: Annotated[int, Doc("number of paths")],
    time_horizon: Annotated[float, Doc("time horizon")] = 1,
    time_steps: Annotated[
        int, Doc("number of time steps to arrive at horizon")
    ] = 100,
) -> Paths:
    """Generate random [Paths][quantflow.ta.paths.Paths] from the process."""

characteristic_exponent `abstractmethod` ¶

characteristic_exponent(t, u)

Characteristic exponent at time t for a given input parameter

Source code in quantflow/sp/base.py

@abstractmethod
def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector:
    """Characteristic exponent at time `t` for a given input parameter"""

characteristic ¶

characteristic(t, u)

Characteristic function at time t for a given input parameter

The characteristic function represents the Fourier transform of the probability density function

\[\begin{equation} \phi = {\mathbb E} \left[e^{i u x_t}\right] \end{equation}\]

PARAMETER	DESCRIPTION
`t`	Time horizon TYPE: `FloatArrayLike`
`u`	Characteristic function input parameter TYPE: `Vector`

Source code in quantflow/sp/base.py

def characteristic(
    self,
    t: Annotated[FloatArrayLike, Doc("Time horizon")],
    u: Annotated[Vector, Doc("Characteristic function input parameter")],
) -> Vector:
    r"""Characteristic function at time `t` for a given input parameter

    The characteristic function represents the Fourier transform of the
    probability density function

    \begin{equation}
        \phi = {\mathbb E} \left[e^{i u x_t}\right]
    \end{equation}
    """
    return np.exp(-self.characteristic_exponent(t, u))

convexity_correction ¶

convexity_correction(t)

Convexity correction for the process

Source code in quantflow/sp/base.py

def convexity_correction(self, t: FloatArrayLike) -> Vector:
    """Convexity correction for the process"""
    return -self.characteristic_exponent(t, complex(0, -1)).real

analytical_std ¶

analytical_std(t)

Source code in quantflow/sp/base.py

def analytical_std(self, t: FloatArrayLike) -> FloatArrayLike:
    return np.sqrt(self.analytical_variance(t))

analytical_mean ¶

analytical_mean(t)

Analytical mean of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_mean(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical mean of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_variance ¶

analytical_variance(t)

Analytical variance of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike:
    """Analytical variance of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_pdf ¶

analytical_pdf(t, x)

Analytical pdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_pdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical pdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

analytical_cdf ¶

analytical_cdf(t, x)

Analytical cdf of the process at time t

Implement if available

Source code in quantflow/sp/base.py

def analytical_cdf(self, t: FloatArrayLike, x: FloatArrayLike) -> FloatArrayLike:
    """Analytical cdf of the process at time `t`

    Implement if available
    """
    raise NotImplementedError

marginal ¶

marginal(t)

Source code in quantflow/sp/base.py

def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal:
    return StochasticProcess1DMarginal(process=self, t=t)

frequency_range ¶

frequency_range(std, max_frequency=None)

Maximum frequency when calculating characteristic functions

Source code in quantflow/sp/base.py

def frequency_range(self, std: float, max_frequency: float | None = None) -> Bounds:
    """Maximum frequency when calculating characteristic functions"""
    if max_frequency is None:
        max_frequency = np.sqrt(40 / std / std)
    return Bounds(0, max_frequency)

support ¶

support(mean, std, points)

Support of the process at time t

Source code in quantflow/sp/base.py

def support(self, mean: float, std: float, points: int) -> FloatArray:
    """Support of the process at time `t`"""
    bounds = self.domain_range()
    start = float(sigfig(bound_from_any(bounds.lb, mean - std)))
    end = float(sigfig(bound_from_any(bounds.ub, mean + std)))
    return np.linspace(start, end, points + 1)

Stochastic Process¶

quantflow.sp.base.StochasticProcess pydantic-model ¶

sample_from_draws abstractmethod ¶

sample abstractmethod ¶

characteristic_exponent abstractmethod ¶

characteristic ¶

convexity_correction ¶

analytical_std ¶

analytical_mean ¶

analytical_variance ¶

analytical_pdf ¶

analytical_cdf ¶

quantflow.sp.base.StochasticProcess1D pydantic-model ¶

marginal ¶

domain_range ¶

frequency_range ¶

support ¶

sample_from_draws abstractmethod ¶

sample abstractmethod ¶

characteristic_exponent abstractmethod ¶

characteristic ¶

convexity_correction ¶

analytical_std ¶

analytical_mean ¶

analytical_variance ¶

analytical_pdf ¶

analytical_cdf ¶

quantflow.sp.base.IntensityProcess pydantic-model ¶

rate pydantic-field ¶

kappa pydantic-field ¶

integrated_log_laplace abstractmethod ¶

domain_range ¶

ekt ¶

sample_from_draws abstractmethod ¶

sample abstractmethod ¶

characteristic_exponent abstractmethod ¶

characteristic ¶

convexity_correction ¶

analytical_std ¶

analytical_mean ¶

analytical_variance ¶

analytical_pdf ¶

analytical_cdf ¶

marginal ¶

frequency_range ¶

support ¶

quantflow.sp.base.StochasticProcess `pydantic-model` ¶

sample_from_draws `abstractmethod` ¶

sample `abstractmethod` ¶

characteristic_exponent `abstractmethod` ¶

quantflow.sp.base.StochasticProcess1D `pydantic-model` ¶

sample_from_draws `abstractmethod` ¶

sample `abstractmethod` ¶

characteristic_exponent `abstractmethod` ¶

quantflow.sp.base.IntensityProcess `pydantic-model` ¶

rate `pydantic-field` ¶

kappa `pydantic-field` ¶

integrated_log_laplace `abstractmethod` ¶

sample_from_draws `abstractmethod` ¶

sample `abstractmethod` ¶

characteristic_exponent `abstractmethod` ¶