Use pydantic

notebooks/cir.md

@@ -5,7 +5,7 @@ jupytext:
     extension: .md
     format_name: myst
     format_version: 0.13
-    jupytext_version: 1.13.8
+    jupytext_version: 1.14.7
 kernelspec:
   display_name: Python 3 (ipykernel)
   language: python
@@ -37,7 +37,7 @@ The model has a close-form solution for the mean and the variance
 
 ```{code-cell} ipython3
 from quantflow.sp.cir import CIR
-pr = CIR(1, kappa=0.8, sigma=0.8, theta=1.2)
+pr = CIR(kappa=0.8, sigma=0.8, theta=1.2)
 pr.is_positive
 ```
 

notebooks/dsp.md

@@ -5,14 +5,33 @@ jupytext:
     extension: .md
     format_name: myst
     format_version: 0.13
-    jupytext_version: 1.13.8
+    jupytext_version: 1.14.7
 kernelspec:
   display_name: Python 3 (ipykernel)
   language: python
   name: python3
 ---
 
-# Doubly Stochastic Poisson Process
+# Poisson Processes
+
+## Poisson Process
+
++++
+
+## Compound Poisson Process
+
+The compound poisson process is a jump process, where the arrival of jumps follows the same dynamic as the Poisson process but the size of jumps is no longer constant and equal to 1, instead it follows a given distribution.
+
+The library includes the Exponential Poisson Process, a compound Poisson process where the jump sizes are sampled from an exponential distribution.
+
+```{code-cell} ipython3
+from quantflow.sp.poisson import ExponentialPoissonProcess
+
+p = ExponentialPoissonProcess(rate=1, decay=1)
+p
+```
+
+## Doubly Stochastic Poisson Process
 
 
 The aim is to identify a stochastic process for simulating the goal arrival which fulfills the following properties
@@ -22,7 +41,7 @@ The aim is to identify a stochastic process for simulating the goal arrival whic
 * Capture the inherent randomness of the goal intensity
 * Intuitive
 
-Before we dive into the details of the DSP process, lets take a quick tour of what Lévy processes are, how a time chage can open the doors to a vast array of models and why they are important in the context of DSP.
+Before we dive into the details of the DSP process, lets take a quick tour of what Lévy processes are, how a time chage can open the doors to a vast array of models and why they are important in the context of DSP.of DSP.
 
 +++
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "quantflow"
-version = "0.1.1"
+version = "0.2.0"
 description = "quantitative analysis"
 authors = ["Luca <[email protected]>"]
 license = "BSD-3-Clause"
@@ -11,13 +11,14 @@ numpy = "^1.22.3"
 scipy = "^1.10.1"
 pandas = "^2.0.1"
 aiohttp = {version = "^3.8.1", optional = true}
+pydantic = "^2.0.2"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.3.0"
 pytest-cov = "^4.0.0"
-mypy = "^1.3.0"
+mypy = "^1.4.0"
 ghp-import = "^2.0.2"
-ruff = "^0.0.269"
+ruff = "^0.0.277"
 
 [tool.poetry.extras]
 data = ["aiohttp"]
@@ -26,20 +27,20 @@ data = ["aiohttp"]
 optional = true
 
 [tool.poetry.group.book.dependencies]
-jupyter-book = "^0.13.1"
+jupyter-book = "^0.15.1"
 nbconvert = "^6.4.5"
 jupytext = "^1.13.8"
 plotly = "^5.7.0"
-ipython = "^8.7.0"
-jsonschema = "<4.0"
-jupyterlab = "^3.5.2"
-jupyter-server = "<2.0"
+jupyterlab = "^4.0.2"
 
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"
 
+[tool.jupytext]
+formats = "ipynb,myst"
+
 [tool.pytest.ini_options]
 testpaths = [
     "tests"

quantflow/sp/base.py

@@ -4,25 +4,21 @@
 from typing import Generic, Tuple, TypeVar, cast
 
 import numpy as np
+from pydantic import BaseModel, Field
 from scipy.optimize import Bounds
 
-from quantflow.utils.marginal import Marginal1D
-from quantflow.utils.param import Param, Parameters, default_bounds
+from quantflow.utils.marginal import Marginal1D, default_bounds
 from quantflow.utils.paths import Paths
 from quantflow.utils.types import Vector
 
 Im = 1j
 
 
-class StochasticProcess(ABC):
+class StochasticProcess(BaseModel, ABC):
     """
     Base class for stochastic processes in continuous time
     """
 
-    @property
-    def parameters(self) -> Parameters:
-        return Parameters()
-
     def sample(self, n: int, t: float = 1, steps: int = 0) -> np.ndarray:
         """Generate random paths from the process
 
@@ -54,12 +50,6 @@ def paths(self, n: int, t: float = 1, steps: int = 0) -> Paths:
         """
         return Paths(t, self.sample(n, t, steps))
 
-    def __repr__(self) -> str:
-        return f"{type(self).__name__} {self.parameters}"
-
-    def __str__(self) -> str:
-        return self.__repr__()
-
 
 class StochasticProcess1D(StochasticProcess):
     """
@@ -280,13 +270,8 @@ class IntensityProcess(StochasticProcess1D):
     as stochastic intensity
     """
 
-    def __init__(self, rate: float, kappa: float) -> None:
-        self.rate = Param(
-            "rate", rate, bounds=(0, None), description="Instantaneous initial rate"
-        )
-        self.kappa = Param(
-            "kappa", kappa, bounds=(0, None), description="Mean reversion speed"
-        )
+    rate: float = Field(default=1.0, gt=0, description="Instantaneous initial rate")
+    kappa: float = Field(default=1.0, gt=0, description="Mean reversion speed")
 
     @abstractmethod
     def cumulative_characteristic(self, t: float, u: Vector) -> Vector:

quantflow/sp/cir.py

@@ -1,8 +1,8 @@
 import numpy as np
 from numpy.random import normal
+from pydantic import Field
 from scipy import special
 
-from quantflow.utils.param import Bounds, Param, Parameters
 from quantflow.utils.types import Vector
 
 from .base import Im, IntensityProcess
@@ -28,34 +28,20 @@ class CIR(IntensityProcess):
     :param sigma: Volatility parameter :math:`\sigma`
     :param theta: Long term mean rate :math:`\theta`
     """
-
-    def __init__(
-        self, rate: float, kappa: float, sigma: float, theta: float = 0
-    ) -> None:
-        super().__init__(rate, kappa)
-        self.sigma = Param("sigma", sigma, bounds=(0, None), description="Volatility")
-        self.theta = Param(
-            "theta", theta or self.rate.value, bounds=(0, None), description="Mean rate"
-        )
-
-    @property
-    def parameters(self) -> Parameters:
-        return Parameters(self.rate, self.kappa, self.sigma, self.theta)
+    sigma: float = Field(default=1.0, gt=0, description="Volatility")
+    theta: float = Field(default=1.0, gt=0, description="Mean rate")
 
     @property
     def is_positive(self) -> bool:
-        return (
-            self.kappa.value * self.theta.value
-            >= 0.5 * self.sigma.value * self.sigma.value
-        )
+        return self.kappa * self.theta >= 0.5 * self.sigma * self.sigma
 
     def pdf(self, t: float, x: Vector) -> Vector:
-        k = self.kappa.value
-        s2 = self.sigma.value * self.sigma.value
+        k = self.kappa
+        s2 = self.sigma * self.sigma
         ekt = np.exp(-k * t)
         c = 2 * k / (1 - ekt) / s2
-        q = 2 * k * self.theta.value / s2 - 1
-        u = c * ekt * self.rate.value
+        q = 2 * k * self.theta / s2 - 1
+        u = c * ekt * self.rate
         v = c * x
         return (
             c
@@ -65,27 +51,27 @@ def pdf(self, t: float, x: Vector) -> Vector:
         )
 
     def mean(self, t: float) -> float:
-        ekt = np.exp(-self.kappa.value * t)
-        return self.rate.value * ekt + self.theta.value * (1 - ekt)
+        ekt = np.exp(-self.kappa * t)
+        return self.rate * ekt + self.theta * (1 - ekt)
 
     def std(self, t: float) -> float:
-        kappa = self.kappa.value
+        kappa = self.kappa
         ekt = np.exp(-kappa * t)
         return np.sqrt(
-            self.sigma.value
-            * self.sigma.value
+            self.sigma
+            * self.sigma
             * (1 - ekt)
-            * (self.rate.value * ekt + 0.5 * self.theta.value * (1 - ekt))
+            * (self.rate * ekt + 0.5 * self.theta * (1 - ekt))
             / kappa
         )
 
     def sample(self, n: int, t: float = 1, steps: int = 0) -> np.ndarray:
         size, dt = self.sample_dt(t, steps)
-        kappa = self.kappa.value
-        theta = self.theta.value
-        sdt = self.sigma.value * np.sqrt(dt)
+        kappa = self.kappa
+        theta = self.theta
+        sdt = self.sigma * np.sqrt(dt)
         paths = np.zeros((size + 1, n))
-        paths[0, :] = self.rate.value
+        paths[0, :] = self.rate
         for p in range(n):
             w = normal(scale=sdt, size=size)
             for i in range(size):
@@ -96,29 +82,26 @@ def sample(self, n: int, t: float = 1, steps: int = 0) -> np.ndarray:
 
     def characteristic(self, t: float, u: Vector) -> Vector:
         iu = Im * u
-        sigma = self.sigma.value
-        kappa = self.kappa.value
+        sigma = self.sigma
+        kappa = self.kappa
         kt = kappa * t
         ekt = np.exp(kt)
         sigma2 = sigma * sigma
         s2u = iu * sigma2
         c = s2u + (2 * kappa - s2u) * ekt
         b = 2 * kappa * iu / c
-        a = 2 * kappa * self.theta.value * (kt + np.log(2 * kappa / c)) / sigma2
-        return np.exp(a + b * self.rate.value)
+        a = 2 * kappa * self.theta * (kt + np.log(2 * kappa / c)) / sigma2
+        return np.exp(a + b * self.rate)
 
     def cumulative_characteristic(self, t: float, u: Vector) -> Vector:
         iu = Im * u
-        sigma = self.sigma.value
-        kappa = self.kappa.value
+        sigma = self.sigma
+        kappa = self.kappa
         sigma2 = sigma * sigma
         gamma = np.sqrt(kappa * kappa - 2 * iu * sigma2)
         egt = np.exp(gamma * t)
         c = (gamma + kappa) * (1 - egt) - 2 * gamma
         d = 2 * gamma * np.exp(0.5 * (gamma + kappa) * t)
-        a = 2 * self.theta.value * kappa * np.log(-d / c) / sigma2
+        a = 2 * self.theta * kappa * np.log(-d / c) / sigma2
         b = 2 * iu * (1 - egt) / c
-        return np.exp(a + b * self.rate.value)
-
-    def domain_range(self) -> Bounds:
-        return Bounds(0, np.inf)
+        return np.exp(a + b * self.rate)

quantflow/sp/dsp.py

@@ -1,14 +1,15 @@
 from typing import List
 
 import numpy as np
+from pydantic import Field
 
 from ..utils.types import Vector, as_array
-from .base import Im
-from .cir import IntensityProcess, Parameters
+from .base import CountingProcess1D, Im
+from .cir import IntensityProcess
 from .poisson import PoissonProcess
 
 
-class DSP(PoissonProcess):
+class DSP(CountingProcess1D):
     r"""
     Doubly Stochastic Poisson process.
 
@@ -17,17 +18,10 @@ class DSP(PoissonProcess):
 
     :param intensity: the stochastic intensity of the Poisson
     """
-
-    def __init__(self, intensity: IntensityProcess):
-        super().__init__(1)
-        self.intensity = intensity
-
-    def __repr__(self) -> str:
-        return f"{type(self).__name__} {self.intensity}"
-
-    @property
-    def parameters(self) -> Parameters:
-        return self.intensity.parameters
+    intensity: IntensityProcess = Field(
+        default_factory=IntensityProcess, description="intensity process"
+    )
+    poisson: PoissonProcess = Field(default_factory=PoissonProcess, exclude=True)
 
     def pdf(self, t: float, n: Vector = 0) -> Vector:
         """PDF of the number of events at time t.
@@ -50,10 +44,13 @@ def cdf(self, t: float, n: Vector) -> Vector:
         """
         return np.cumsum(self.pdf(t, n))
 
+    def characteristic_exponent(self, u: Vector) -> Vector:
+        return self.poisson.characteristic_exponent(u)
+
     def characteristic(self, t: float, u: Vector) -> Vector:
         phi = self.characteristic_exponent(u)
         return self.intensity.cumulative_characteristic(t, -Im * phi)
 
     def arrivals(self, t: float = 1) -> List[float]:
         paths = self.intensity.paths(1, t).integrate()
-        return super().arrivals(paths.data[-1, 0])
+        return self.poisson.arrivals(paths.data[-1, 0])