make sample abstract

Author: lsbardel

quantflow/sp/base.py

@@ -19,14 +19,14 @@ class StochasticProcess(BaseModel, ABC):
     Base class for stochastic processes in continuous time
     """
 
+    @abstractmethod
     def sample(self, n: int, t: float = 1, steps: int = 0) -> np.ndarray:
         """Generate random paths from the process
 
         :param n: Number of paths
         :param t: time horizon
         :param steps: number of time steps to arrive at horizon
         """
-        raise NotImplementedError
 
     def sample_dt(self, t: float, steps: int = 0) -> Tuple[int, float]:
         """Time delta for sampling paths

quantflow/sp/heston.py

@@ -48,8 +48,6 @@ def characteristic(self, t: float, u: Vector) -> Vector:
         a = theta_kappa * (2 * np.log(c) + (gamma - kappa) * t) / eta2
         return np.exp(-a - b * self.variance_process.rate)
 
-    def pdf(self, t: float, n: Vector) -> Vector:
-        raise NotImplementedError
-
-    def cdf(self, t: float, n: Vector) -> Vector:
+    def sample(self, n: int, t: float = 1, steps: int = 0) -> np.ndarray:
+        # TODO: implement
         raise NotImplementedError

quantflow/sp/weiner.py

@@ -3,6 +3,7 @@
 from functools import lru_cache
 
 import numpy as np
+from numpy.random import normal
 from pydantic import Field
 from scipy.stats import norm
 
@@ -32,6 +33,14 @@ def characteristic(self, t: float, u: Vector) -> Vector:
         su = self.sigma * u
         return np.exp(-0.5 * su * su * t)
 
+    def sample(self, n: int, t: float = 1, steps: int = 0) -> np.ndarray:
+        time_steps, dt = self.sample_dt(t, steps)
+        sdt = self.sigma * np.sqrt(dt)
+        paths = np.zeros((time_steps + 1, n))
+        for t in range(time_steps):
+            paths[t + 1, :] = paths[t, :] + normal(scale=sdt, size=n)
+        return paths
+
 
 class WeinerMarginal(StochasticProcess1DMarginal[WeinerProcess]):
     def mean(self) -> float:

tests/test_weiner.py

@@ -15,3 +15,11 @@ def test_characteristic(weiner: WeinerProcess) -> None:
     assert marginal.mean_from_characteristic() == 0
     assert marginal.std() == 0.5
     assert marginal.variance_from_characteristic() == pytest.approx(0.25)
+
+
+def test_sampling(weiner: WeinerProcess) -> None:
+    paths = weiner.paths(1000, t=1, steps=1000)
+    mean = paths.mean()
+    assert mean[0] == 0
+    std = paths.std()
+    assert std[0] == 0