From 0f5e1b0963d288eab25212275f0aa7a4285bd3f1 Mon Sep 17 00:00:00 2001 From: Luca Date: Sun, 22 Feb 2026 11:35:53 +0000 Subject: [PATCH] Fix docs --- app/poisson_sampling.py | 72 +++++++++++++++++++++ docs/api/sp/heston.md | 18 +----- docs/api/sp/jump_diffusion.md | 12 +--- docs/api/sp/poisson.md | 2 + docs/api/sp/weiner.md | 13 +--- mkdocs.yml | 1 + quantflow/sp/base.py | 38 ++++++----- quantflow/sp/dsp.py | 6 +- quantflow/sp/heston.py | 113 +++++++++++++++++++------------- quantflow/sp/jump_diffusion.py | 57 ++++++++++------- quantflow/sp/poisson.py | 114 ++++++++++++++++++++------------- quantflow/ta/paths.py | 59 +++++++++-------- 12 files changed, 314 insertions(+), 191 deletions(-) create mode 100644 app/poisson_sampling.py diff --git a/app/poisson_sampling.py b/app/poisson_sampling.py new file mode 100644 index 00000000..6c2406b1 --- /dev/null +++ b/app/poisson_sampling.py @@ -0,0 +1,72 @@ +import marimo + +__generated_with = "0.19.7" +app = marimo.App(width="medium") + + +@app.cell +def _(): + import marimo as mo + from app.utils import nav_menu + nav_menu() + return (mo,) + + +@app.cell +def _(mo): + mo.md(r""" + # Poisson Sampling + + Evaluate the MC simulation for The Poisson process against the analytical PDF. + """) + return + + +@app.cell +def _(): + from quantflow.sp.poisson import PoissonProcess + import pandas as pd + + def simulate_poisson(intensity: float, samples: int) -> pd.DataFrame: + pr = PoissonProcess(intensity=intensity) + paths = pr.sample(samples, 1, 1000) + pdf = paths.pdf(delta=1) + pdf["simulation"] = pdf["pdf"] + pdf["analytical"] = pr.marginal(1).pdf(pdf.index) + return pdf + return (simulate_poisson,) + + +@app.cell +def _(mo): + samples = mo.ui.slider(start=100, stop=10000, step=100, value=1000, debounce=True, label="Samples") + intensity = mo.ui.slider(start=2, stop=5, step=0.1, debounce=True, label="Poisson intensity $\lambda$") + + controls = mo.hstack([samples, intensity], justify="start") + controls + return intensity, samples + + +@app.cell +def _(intensity, samples, simulate_poisson): + df = simulate_poisson(intensity=intensity.value, samples=samples.value) + return (df,) + + +@app.cell +def _(df): + import plotly.graph_objects as go + simulation = go.Bar(x=df.index, y=df["simulation"], name="simulation") + analytical = go.Bar(x=df.index, y=df["analytical"], name="analytical") + fig = go.Figure(data=[simulation, analytical]) + fig + return + + +@app.cell +def _(): + return + + +if __name__ == "__main__": + app.run() diff --git a/docs/api/sp/heston.md b/docs/api/sp/heston.md index 12e5fc23..b239d1e4 100644 --- a/docs/api/sp/heston.md +++ b/docs/api/sp/heston.md @@ -1,18 +1,6 @@ -================ -Heston process -================ +# Heston process -.. currentmodule:: quantflow.sp.heston +::: quantflow.sp.heston.Heston -.. autoclass:: Heston - :members: - :member-order: groupwise - :autosummary: - :autosummary-nosignatures: - -.. autoclass:: HestonJ - :members: - :member-order: groupwise - :autosummary: - :autosummary-nosignatures: +::: quantflow.sp.heston.HestonJ diff --git a/docs/api/sp/jump_diffusion.md b/docs/api/sp/jump_diffusion.md index e1293ebc..18c661b0 100644 --- a/docs/api/sp/jump_diffusion.md +++ b/docs/api/sp/jump_diffusion.md @@ -1,16 +1,8 @@ -================ -Jump diffusions -================ +# Jump diffusions Jump-diffusions models are a class of stochastic processes that combine a diffusion process with a jump process. The jump process is a Poisson process that generates jumps in the value of the underlying asset. The jump-diffusion model is a generalization of the Black-Scholes model that allows for the possibility of large, discontinuous jumps in the value of the underlying asset. The most famous jump-diffusion model is the Merton model, which was introduced by Robert Merton in 1976. The Merton model assumes that the underlying asset follows a geometric Brownian motion with jumps that are normally distributed. -.. currentmodule:: quantflow.sp.jump_diffusion - -.. autoclass:: JumpDiffusion - :members: - :member-order: groupwise - :autosummary: - :autosummary-nosignatures: +::: quantflow.sp.jump_diffusion.JumpDiffusion diff --git a/docs/api/sp/poisson.md b/docs/api/sp/poisson.md index 95bbaf81..51fd3850 100644 --- a/docs/api/sp/poisson.md +++ b/docs/api/sp/poisson.md @@ -1,3 +1,5 @@ # Poisson process +::: quantflow.sp.poisson.PoissonBase + ::: quantflow.sp.poisson.PoissonProcess diff --git a/docs/api/sp/weiner.md b/docs/api/sp/weiner.md index d4c213fe..ff63446f 100644 --- a/docs/api/sp/weiner.md +++ b/docs/api/sp/weiner.md @@ -1,12 +1,3 @@ -=============== -Weiner process -=============== - -.. module:: quantflow.sp.weiner - -.. autoclass:: WeinerProcess - :members: - :member-order: groupwise - :autosummary: - :autosummary-nosignatures: +# Weiner process +::: quantflow.sp.weiner.WeinerProcess diff --git a/mkdocs.yml b/mkdocs.yml index bccc5efc..9d6c1408 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -56,6 +56,7 @@ nav: - Home: index.md - Examples: - Gaussian Sampling: examples/gaussian-sampling + - Poisson Sampling: examples/poisson-sampling - Hurst: examples/hurst - Supersmoother: examples/supersmoother - API Reference: diff --git a/quantflow/sp/base.py b/quantflow/sp/base.py index 441ba536..498439c8 100755 --- a/quantflow/sp/base.py +++ b/quantflow/sp/base.py @@ -6,6 +6,7 @@ import numpy as np from pydantic import BaseModel, ConfigDict, Field from scipy.optimize import Bounds +from typing_extensions import Annotated, Doc from quantflow.ta.paths import Paths from quantflow.utils.marginal import Marginal1D, default_bounds @@ -26,29 +27,33 @@ def sample_from_draws(self, draws: Paths, *args: Paths) -> Paths: """Sample :class:`.Paths` from the process given a set of draws""" @abstractmethod - def sample(self, n: int, time_horizon: float = 1, time_steps: int = 100) -> Paths: - """Generate random :class:`.Paths` from the process. - - :param n: number of paths - :param time_horizon: time horizon - :param time_steps: number of time steps to arrive at horizon - """ + 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 :class:`.Paths` from the process.""" @abstractmethod def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector: """Characteristic exponent at time `t` for a given input parameter""" - def characteristic(self, t: FloatArrayLike, u: Vector) -> Vector: + 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 - .. math:: + \begin{equation} \phi = {\mathbb E} \left[e^{i u x_t}\right] - - :param t: time horizon - :param u: characteristic function input parameter + \end{equation} """ return np.exp(-self.characteristic_exponent(t, u)) @@ -174,14 +179,15 @@ class IntensityProcess(StochasticProcess1D): r"""Mean reversion speed :math:`\kappa`""" @abstractmethod - def integrated_log_laplace(self, t: FloatArrayLike, u: Vector) -> Vector: + 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] - - :param t: time horizon - :param u: frequency """ def domain_range(self) -> Bounds: diff --git a/quantflow/sp/dsp.py b/quantflow/sp/dsp.py index a10d6859..2368f8ab 100755 --- a/quantflow/sp/dsp.py +++ b/quantflow/sp/dsp.py @@ -15,12 +15,12 @@ class DSP(PoissonBase): Doubly Stochastic Poisson process. It's a process where the inter-arrival time is exponentially distributed - with rate :math:`\lambda_t` + with rate $\lambda_t$ :param intensity: the stochastic intensity of the Poisson """ - intensity: IntensityProcess = Field( # type ignore + intensity: IntensityProcess = Field( default_factory=CIR, description="intensity process" ) poisson: PoissonProcess = Field(default_factory=PoissonProcess, exclude=True) @@ -39,7 +39,7 @@ def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector: phi = self.poisson.characteristic_exponent(t, u) return -self.intensity.integrated_log_laplace(t, phi) - def arrivals(self, t: float = 1) -> list[float]: + def arrivals(self, t: float = 1) -> FloatArray: paths = self.intensity.sample(1, t, math.ceil(100 * t)).integrate() intensity = paths.data[-1, 0] return poisson_arrivals(intensity, t) diff --git a/quantflow/sp/heston.py b/quantflow/sp/heston.py index 80b445ee..68e48a7b 100644 --- a/quantflow/sp/heston.py +++ b/quantflow/sp/heston.py @@ -4,6 +4,7 @@ import numpy as np from pydantic import Field +from typing_extensions import Annotated, Doc from quantflow.ta.paths import Paths from quantflow.utils.types import FloatArrayLike, Vector @@ -18,53 +19,78 @@ class Heston(StochasticProcess1D): r"""The Heston stochastic volatility model The classical square-root stochastic volatility model of Heston (1993) - can be regarded as a standard Brownian motion :math:`x_t` time changed by a CIR + can be regarded as a standard Brownian motion $x_t$ time changed by a CIR activity rate process. - .. math:: - \begin{align} - d x_t &= d w^1_t \\ - d v_t &= \kappa (\theta - v_t) dt + \nu \sqrt{v_t} dw^2_t \\ - \rho dt &= {\tt E}[dw^1 dw^2] - \end{align} + \begin{align} + d x_t &= d w^1_t \\ + d v_t &= \kappa (\theta - v_t) dt + \nu \sqrt{v_t} dw^2_t \\ + \rho dt &= {\tt E}[dw^1 dw^2] + \end{align} """ - variance_process: CIR = Field(default_factory=CIR, description="Variance process") - """The variance process is a Cox-Ingersoll-Ross (:class:`.CIR`) process""" - rho: float = Field(default=0, ge=-1, le=1, description="Correlation") - """Correlation between the Brownian motions - provides the leverage effect""" + variance_process: CIR = Field( + default_factory=CIR, + description="The variance process is a Cox-Ingersoll-Ross process", + ) + rho: float = Field( + default=0, + ge=-1, + le=1, + description=( + "Correlation between the Brownian motions, it provides " + "the leverage effect and therefore the skewness of the distribution" + ), + ) @classmethod def create( cls, *, - rate: float = 1.0, - vol: float = 0.5, - kappa: float = 1, - sigma: float = 0.8, - rho: float = 0, - theta: float | None = None, + rate: Annotated[float, Doc("Initial rate of the variance process")] = 1.0, + vol: Annotated[ + float, + Doc( + "Volatility of the price process, normalized by the " + "square root of time, as time tends to infinity " + "(the long term standard deviation)" + ), + ] = 0.5, + kappa: Annotated[ + float, + Doc( + "Mean reversion speed for the variance process, the lower the " + "more pronounced the volatility clustering and therefore the fatter " + "the tails of the distribution of the price process" + ), + ] = 1, + sigma: Annotated[ + float, Doc("Volatility of the variance process (a.k.a. the vol of vol)") + ] = 0.8, + rho: Annotated[ + float, + Doc( + "Correlation between the Brownian motions of the " + "variance and price processes" + ), + ] = 0, + theta: Annotated[ + float | None, + Doc( + "Long-term mean of the variance process. " + r"If `None`, it defaults to the variance given by ${\tt var}$" + " the long term variance described above." + ), + ] = None, ) -> Self: r"""Create an Heston model. To understand the parameters lets introduce the following notation: - .. math:: - \begin{align} - {\tt var} &= {\tt vol}^2 \\ - v_0 &= {\tt rate}\cdot{\tt var} - \end{align} - - :param rate: define the initial value of the variance process - :param vol: The standard deviation of the price process, normalized by the - square root of time, as time tends to infinity - (the long term standard deviation) - :param kappa: The mean reversion speed for the variance process - :param sigma: The volatility of the variance process - :param rho: The correlation between the Brownian motions of the - variance and price processes - :param theta: The long-term mean of the variance process, if `None`, it - defaults to the variance given by :math:`{\tt var}` + \begin{align} + {\tt var} &= {\tt vol}^2 \\ + v_0 &= {\tt rate}\cdot{\tt var} + \end{align} """ variance = vol * vol return cls( @@ -118,20 +144,20 @@ class HestonJ(Heston, Generic[D]): stochastic volatility model of Heston (1993) with the addition of jump processes. The jumps are modeled via a compound Poisson process - .. math:: + \begin{align} d x_t &= d w^1_t + d N_t\\ d v_t &= \kappa (\theta - v_t) dt + \nu \sqrt{v_t} dw^2_t \\ \rho dt &= {\tt E}[dw^1 dw^2] + \end{align} This model is generic and therefore allows for different types of jumps distributions **D**. The Bates model is obtained by using the - :class:`.Normal` distribution for the jump sizes. + [Normal][quantflow.utils.distributions.Normal] distribution for the jump sizes. """ - jumps: CompoundPoissonProcess[D] = Field(description="jumps process") - """Jump process driven by a :class:`.CompoundPoissonProcess`""" + jumps: CompoundPoissonProcess[D] = Field(description="Jump process") @classmethod def create( # type: ignore [override] @@ -152,13 +178,12 @@ def create( # type: ignore [override] To understand the parameters lets introduce the following notation: - .. math:: - \begin{align} - {\tt var} &= {\tt vol}^2 \\ - {\tt var}_j &= {\tt var} \cdot {\tt jump\_fraction} \\ - {\tt var}_d &= {\tt var} - {\tt var}_j \\ - v_0 &= {\tt rate}\cdot{\tt var}_d - \end{align} + \begin{align} + {\tt var} &= {\tt vol}^2 \\ + {\tt var}_j &= {\tt var} \cdot {\tt jump\_fraction} \\ + {\tt var}_d &= {\tt var} - {\tt var}_j \\ + v_0 &= {\tt rate}\cdot{\tt var}_d + \end{align} :param jump_distribution: The distribution of jump size (currently only :class:`.Normal` and :class:`.DoubleExponential` are supported) diff --git a/quantflow/sp/jump_diffusion.py b/quantflow/sp/jump_diffusion.py index a0ce533e..9f480aa3 100644 --- a/quantflow/sp/jump_diffusion.py +++ b/quantflow/sp/jump_diffusion.py @@ -4,6 +4,7 @@ import numpy as np from pydantic import Field +from typing_extensions import Annotated, Doc from ..ta.paths import Paths from ..utils.types import FloatArrayLike, Vector @@ -15,20 +16,20 @@ class JumpDiffusion(StochasticProcess1D, Generic[D]): r"""A generic jump-diffusion model - .. math:: + \begin{equation} dx_t = \sigma d w_t + d N_t + \end{equation} - where :math:`w_t` is a Weiner process with standard deviation :math:`\sigma` - and :math:`N_t` is a :class:`.CompoundPoissonProcess` - with intensity :math:`\lambda` and generic jump distribution `D` + where $w_t$ is a [WeinerProcess][quantflow.sp.weiner.WeinerProcess] process + with standard deviation $\sigma$ and $N_t$ is a + [CompoundPoissonProcess][quantflow.sp.poisson.CompoundPoissonProcess] + with intensity $\lambda$ and generic jump distribution $D$ """ diffusion: WeinerProcess = Field( - default_factory=WeinerProcess, description="diffusion" + default_factory=WeinerProcess, description="diffusion process" ) - """The diffusion process is a standard :class:`.WeinerProcess`""" - jumps: CompoundPoissonProcess[D] = Field(description="jump process") - """The jump process is a generic :class:`.CompoundPoissonProcess`""" + jumps: CompoundPoissonProcess[D] = Field(description="The jump process") def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector: return self.diffusion.characteristic_exponent( @@ -56,25 +57,33 @@ def analytical_variance(self, t: FloatArrayLike) -> FloatArrayLike: @classmethod def create( cls, - jump_distribution: type[D], - vol: float = 0.5, - jump_intensity: float = 100, - jump_fraction: float = 0.5, - jump_asymmetry: float = 0.0, + jump_distribution: Annotated[ + type[D], + Doc( + "The distribution of jump sizes. Currently " + "[Normal][quantflow.utils.distributions.Normal] and " + "[DoubleExponential][quantflow.utils.distributions.DoubleExponential] " + "are supported. If the jump distribution is set to the Normal " + "distribution, the model reduces to a Merton jump-diffusion." + ), + ], + vol: Annotated[float, Doc("total standard deviation per unit time")] = 0.5, + jump_intensity: Annotated[ + float, Doc("The expected number of jumps per unit time") + ] = 100, + jump_fraction: Annotated[ + float, Doc("The fraction of variance due to jumps (between 0 and 1)") + ] = 0.5, + jump_asymmetry: Annotated[ + float, + Doc( + "The asymmetry of the jump distribution " + "(0 for symmetric, only used by distributions with asymmetry)" + ), + ] = 0.0, ) -> JumpDiffusion[D]: """Create a jump-diffusion model with a given jump distribution, volatility and jump fraction. - - :param jump_distribution: The distribution of jump sizes (currently only - :class:`.Normal` and :class:`.DoubleExponential` are supported) - :param vol: total annualized standard deviation - :param jump_intensity: The average number of jumps per year - :param jump_fraction: The fraction of variance due to jumps (between 0 and 1) - :param jump_asymmetry: The asymmetry of the jump distribution (0 for symmetric, - only used by distributions with asymmetry) - - If the jump distribution is set to the :class:`.Normal` distribution, the - model reduces to a Merton jump-diffusion model. """ variance = vol * vol if jump_fraction >= 1: diff --git a/quantflow/sp/poisson.py b/quantflow/sp/poisson.py index af2ebbee..9c3c81cd 100755 --- a/quantflow/sp/poisson.py +++ b/quantflow/sp/poisson.py @@ -8,6 +8,7 @@ from scipy.integrate import simpson from scipy.optimize import Bounds from scipy.stats import poisson +from typing_extensions import Annotated, Doc from quantflow.ta.paths import Paths from quantflow.utils.distributions import Distribution1D @@ -22,23 +23,34 @@ class PoissonBase(StochasticProcess1D): @abstractmethod - def sample_jumps(self, n: int) -> np.ndarray: + def sample_jumps(self, n: Annotated[int, Doc("Number of jumps")]) -> np.ndarray: """Generate a list of jump sizes""" @abstractmethod - def arrivals(self, time_horizon: float = 1) -> list[float]: - """Generate a list of jump arrivals times up to time t""" + def arrivals( + self, time_horizon: Annotated[float, Doc("Time horizon")] = 1 + ) -> FloatArray: + """Generate a list of jump arrivals times up to time `time_horizon`""" - def sample(self, n: int, time_horizon: float = 1, time_steps: int = 100) -> Paths: + 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")] = 100, + ) -> Paths: + """Sample a number of paths of the process up to a given time horizon and + with a given number of time steps. + """ dt = time_horizon / time_steps paths = np.zeros((time_steps + 1, n)) for p in range(n): - if arrivals := self.arrivals(time_horizon): - jumps = self.sample_jumps(len(arrivals)) + arrivals = self.arrivals(time_horizon) + if num_arrivals := len(arrivals): + jumps = self.sample_jumps(num_arrivals) i = 1 y = 0.0 for j, arrival in enumerate(arrivals): - while i * dt < arrival: + while i <= time_steps and i * dt < arrival: paths[i, p] = y i += 1 y += jumps[j] @@ -52,22 +64,33 @@ def domain_range(self) -> Bounds: return Bounds(0, np.inf) -def poisson_arrivals(intensity: float, time_horizon: float = 1) -> list[float]: - """Generate a list of jump arrivals times up to time t""" - exp_rate = 1.0 / intensity - arrivals = [] - tt = 0.0 - while tt < time_horizon: - dt = np.random.exponential(scale=exp_rate) - tt += dt - if tt <= time_horizon: - arrivals.append(tt) - return arrivals +def poisson_arrivals(intensity: float, time_horizon: float = 1) -> FloatArray: + r"""Generate a list of jump arrivals times up to time t + + This method conditions on the total number of arrivals $N$, which follows + a Poisson distribution with mean $\lambda T$. + + Given $N$, the arrival times are distributed as the order statistics + of $N$ uniform random variables on $[0, T]$. + """ + n = np.random.poisson(intensity * time_horizon) + return np.sort(np.random.uniform(0, time_horizon, n)) class PoissonProcess(PoissonBase): - intensity: float = Field(default=1.0, ge=0, description="intensity rate") - r"""Intensity rate :math:`\lambda` of the Poisson process""" + r"""A Poisson process is a pure jump process where the number of jumps + in a time interval follows a Poisson distribution + and the jump sizes are always 1. + + The expected number of jumps and the variance in a unit of time + is given by the non-negative `intensity` parameter $\lambda$. + """ + + intensity: float = Field( + default=1.0, + ge=0, + description=r"Intensity rate $\lambda$ of the Poisson process", + ) def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal: return MarginalDiscrete1D(process=self, t=t) @@ -75,7 +98,10 @@ def marginal(self, t: FloatArrayLike) -> StochasticProcess1DMarginal: def characteristic_exponent(self, t: Vector, u: Vector) -> Vector: return t * self.intensity * (1 - np.exp(Im * u)) - def arrivals(self, time_horizon: float = 1) -> list[float]: + def arrivals( + self, time_horizon: Annotated[float, Doc("Time horizon")] = 1 + ) -> FloatArray: + """Generate a list of jump arrivals times up to time `time_horizon`""" return poisson_arrivals(self.intensity, time_horizon) def sample_jumps(self, n: int) -> np.ndarray: @@ -104,13 +130,13 @@ def analytical_cdf(self, t: FloatArrayLike, n: FloatArrayLike) -> FloatArrayLike It's given by - .. math:: - :label: poisson_cdf - + \begin{equation} F\left(n\right)=\frac{\Gamma\left(\left\lfloor n+1\right\rfloor ,\lambda\right)}{\left\lfloor n\right\rfloor !} + \tag{1} + \end{equation} - where :math:`\Gamma` is the upper incomplete gamma function. + where $\Gamma$ is the upper incomplete gamma function. """ return poisson.cdf(n, t * self.intensity) @@ -120,10 +146,10 @@ def analytical_pdf(self, t: FloatArrayLike, n: FloatArrayLike) -> FloatArrayLike It's given by - .. math:: - :label: poisson_pdf - - f\left(n\right)=\frac{\lambda^{n}e^{-\lambda}}{n!} + \begin{equation} + f\left(n\right)=\frac{(\lambda t)^{n}e^{-\lambda t}}{n!} + \tag{2} + \end{equation} """ return poisson.pmf(n, t * self.intensity) @@ -133,11 +159,11 @@ def cdf_jacobian(self, t: FloatArrayLike, n: Vector) -> np.ndarray: It's given by - .. math:: - :label: poisson_cdf_jacobian - - \frac{\partial F}{\partial\lambda}=-\frac{\lambda^{\left\lfloor + \begin{equation} + \frac{\partial F}{\partial\lambda}=-\frac{(\lambda t)^{\left\lfloor n\right\rfloor }e^{-\lambda}}{\left\lfloor n\right\rfloor !} + \tag{3} + \end{equation} """ k = np.floor(n).astype(int) rate = self.intensity @@ -147,28 +173,30 @@ def cdf_jacobian(self, t: FloatArrayLike, n: Vector) -> np.ndarray: class CompoundPoissonProcess(PoissonBase, Generic[D]): """A generic Compound Poisson process.""" - intensity: float = Field(default=1.0, gt=0, description="jump intensity rate") - r"""Intensity rate :math:`\lambda` of the Poisson process - - It determines the number of jumps in the same way as the :class:`.PoissonProcess` - """ + intensity: float = Field( + default=1.0, + gt=0, + description=r"Intensity rate $\lambda$ of the Poisson process", + ) jumps: D = Field(description="Jump size distribution") - """Jump size distribution""" def characteristic_exponent(self, t: FloatArrayLike, u: Vector) -> Vector: r"""The characteristic exponent of the Compound Poisson process, given by - .. math:: + \begin{equation} \phi_{x_t,u} = t\lambda \left(1 - \Phi_{j,u}\right) + \end{equation} - where :math:`\Phi_{j,u}` is the characteristic function + where $\Phi_{j,u}$ is the characteristic function of the jump distribution """ return t * self.intensity * (1 - self.jumps.characteristic(u)) - def arrivals(self, time_horizon: float = 1) -> list[float]: - """Same as Poisson process""" + def arrivals( + self, time_horizon: Annotated[float, Doc("Time horizon")] = 1 + ) -> FloatArray: + """Generate a list of jump arrivals times up to time `time_horizon`""" return poisson_arrivals(self.intensity, time_horizon) def sample_jumps(self, n: int) -> FloatArray: diff --git a/quantflow/ta/paths.py b/quantflow/ta/paths.py index 84f58eed..f00145ab 100755 --- a/quantflow/ta/paths.py +++ b/quantflow/ta/paths.py @@ -22,14 +22,12 @@ class Paths(BaseModel, arbitrary_types_allowed=True): This is the output from a simulation of a stochastic process. """ - t: float = Field(description="time horizon") - """Time horizon - the unit of time is not specified""" - data: FloatArray = Field(description="paths") - """Paths of the stochastic process""" + t: float = Field(description="Time horizon - the unit of time is not specified") + data: FloatArray = Field(description="Paths of the stochastic process") @property def dt(self) -> float: - """Time step""" + """Time step given by the time horizon divided by time steps""" return self.t / self.time_steps @property @@ -124,12 +122,17 @@ def integrate(self) -> Paths: data=cumulative_trapezoid(self.data, dx=self.dt, axis=0, initial=0), ) - def hurst_exponent(self, steps: int | None = None) -> float: - """Estimate the Hurst exponent from all paths - - :param steps: number of lags to consider, if not provided it uses - half of the time steps capped at 100 - """ + def hurst_exponent( + self, + steps: Annotated[ + int | None, + Doc( + "number of lags to consider, if not provided it uses half " + "of the time steps capped at 100" + ), + ] = None, + ) -> float: + """Estimate the Hurst exponent from all paths""" ts = self.time_steps // 2 n = min(steps or ts, 100) lags = [] @@ -140,7 +143,9 @@ def hurst_exponent(self, steps: int | None = None) -> float: lags.extend([lag] * self.samples) return float(np.polyfit(np.log(lags), np.log(tau), 1)[0]) / 2.0 - def cross_section(self, t: float | None = None) -> FloatArray: + def cross_section( + self, t: Annotated[float | None, Doc("time of cross section")] = None + ) -> FloatArray: """Cross section of paths at time t""" index = self.time_steps if t is not None: @@ -149,23 +154,27 @@ def cross_section(self, t: float | None = None) -> FloatArray: def pdf( self, - t: float | None = None, - num_bins: int | None = None, - delta: float | None = None, - symmetric: float | None = None, + t: Annotated[float | None, Doc("time at which to calculate the pdf")] = None, + num_bins: Annotated[int | None, Doc("number of bins to use")] = None, + delta: Annotated[ + float | None, Doc("optional size of bins (cannot be set with num_bins)") + ] = None, + symmetric: Annotated[ + float | None, Doc("An optional value where to center bins") + ] = None, ) -> pd.DataFrame: - """Probability density function of paths - - Calculate a DataFrame with the probability density function of the paths - at a given cross section of time. By default it take the last section. + """Estimate the Probability density function from paths at a given + time horizon. - :param t: time at which to calculate the pdf - :param num_bins: number of bins - :param delta: optional size of bins (cannot be set with num_bins) - :param symmetric: optional center of bins + This method calculates a DataFrame with the probability density function + of the paths at a given cross section of time. + By default it take the last section. """ return bins_pdf( - self.cross_section(t), num_bins=num_bins, delta=delta, symmetric=symmetric + self.cross_section(t), + num_bins=num_bins, + delta=delta, + symmetric=symmetric, ) def plot(self, **kwargs: Any) -> Any: