Unit tests and exception handling.

examples/compare_option_pricers.py

@@ -1,3 +1,4 @@
+
 from time import time
 
 import numpy as np

examples/convergence order_cir.py

@@ -1,3 +1,4 @@
+
 import numpy as np
 
 from sde import SDE

examples/simulate_cir.py

@@ -9,14 +9,14 @@
 
 """
 We begin with the definition of both drift and diffusion functions and define the CIR process.
-The order in the function interface does not matter.
+The order in the function interface does matter - the interface needs to be f(x, t, arguments).
 """
 
 def cir_drift(x, a, b):
     return a * (b - x)
 
 
-def cir_diffusion(c, x):
+def cir_diffusion(x, c):
     return np.sqrt(x) * c
 
 cir_process = SDE(cir_drift, cir_diffusion, timerange=[0,2])

requirements.txt

@@ -1,2 +1,6 @@
+
+# This file contains all packages needed to run SDEtools.
+
 numpy
-scipy
+scipy
+pytest

sde.py

@@ -18,43 +18,10 @@ def __init__(self, drift, diffusion, timerange=None, startvalue=1):
         self._diffusion = diffusion
         self._timerange = timerange
         self._startvalue = startvalue
-        self.build_information()
-        self._driving_process = self
-
-    def build_information(self):
-        drift_parameter = list(signature(self.drift).parameters)
-        diffusion_parameter = list(signature(self.diffusion).parameters)
-
         self._information = dict()
-        self._information['drift'] = dict()
-        self._information['diffusion'] = dict()
-        if "x" in drift_parameter:
-            self._information['drift']['spatial'] = True
-            drift_parameter.remove("x")
-        else:
-            self._information['drift']['spatial'] = False
-
-        if "t" in drift_parameter:
-            self._information['drift']['time'] = True
-            drift_parameter.remove("t")
-        else:
-            self._information['drift']['time']  = False
-
-        if "x" in diffusion_parameter:
-            self._information['diffusion']['spatial'] = True
-            diffusion_parameter.remove("x")
-        else:
-            self._information['diffusion']['spatial'] = False
-
-        if "t" in diffusion_parameter:
-            self._information['diffusion']['time'] = True
-            diffusion_parameter.remove("t")
-        else:
-            self._information['diffusion']['time'] = False
-
-        self._information['drift']['parameter'] = drift_parameter
-        self._information['diffusion']['parameter'] = diffusion_parameter
-        self._information['parameter'] = [drift_parameter, diffusion_parameter]
+        self._information['drift'] = build_information(drift)
+        self._information['diffusion'] = build_information(diffusion)
+        self._driving_process = self
 
     @property
     def drift(self):
@@ -76,3 +43,26 @@ def startvalue(self):
     def information (self):
         return self._information
 
+
+def build_information(func):
+    try:
+        parameter = list(signature(func).parameters)
+    except TypeError:
+        print("ERROR: No proper function was given, the information set can't be built.")
+        raise
+    else:
+        information = dict()
+        if "x" in parameter:
+            information['spatial'] = True
+            parameter.remove("x")
+        else:
+            information['spatial'] = False
+
+        if "t" in parameter:
+            information['time'] = True
+            parameter.remove("t")
+        else:
+            information['time'] = False
+
+        information['parameter'] = parameter
+        return information

simulation/scheme.py

@@ -1,12 +1,16 @@
 
 """
-This file contains several simulation methods for SDEs.
+A generic base class to implement a numerical scheme.
+Once inherited, the specific scheme has to implement the method 'propagation' to represent
+a functional implementation.
 """
 
 from collections import deque
 
 import numpy as np
 
+from sde import build_information
+
 
 class Scheme:
     def __init__(self, sde, parameter, steps, **kwargs):
@@ -20,34 +24,54 @@ def __init__(self, sde, parameter, steps, **kwargs):
 
         if 'derivatives' in kwargs:
             if 'diffusion_x' in kwargs['derivatives']:
-                self.diffusion_x = self.map_to_parameter_set(kwargs['derivatives']['diffusion_x'], parameter, sde.information['diffusion'])
+                self.diffusion_x = self.map_to_parameter_set(kwargs['derivatives']['diffusion_x'], parameter,
+                                                             build_information(kwargs['derivatives']['diffusion_x']))
+                kwargs['derivatives'].pop('diffusion_x')
+            if 'diffusion_xx' in kwargs['derivatives']:
+                self.diffusion_xx = self.map_to_parameter_set(kwargs['derivatives']['diffusion_xx'], parameter,
+                                                              build_information(kwargs['derivatives']['diffusion_xx']))
+                kwargs['derivatives'].pop('diffusion_xx')
+            if 'drift_x' in kwargs['derivatives']:
+                self.diffusion_x = self.map_to_parameter_set(kwargs['derivatives']['drift_x'], parameter,
+                                                             build_information(kwargs['derivatives']['drift_x']))
+                kwargs['derivatives'].pop('drift_x')
+            if 'drift_xx' in kwargs['derivatives']:
+                self.diffusion_x = self.map_to_parameter_set(kwargs['derivatives']['drift_xx'], parameter,
+                                                             build_information(kwargs['derivatives']['drift_xx']))
+                kwargs['derivatives'].pop('drift_xx')
+
         if 'path' in kwargs:
             self.driving_stochastic_differential = deque(kwargs['path'])
         else:
             if 'strong' in self.__module__:
                 self.driving_stochastic_differential = deque(np.random.standard_normal(steps) * np.sqrt(self.h))
-            else:
+            elif 'weak' in self.__module__:
                 if 'Order_10' in self.__module__:
                     self.driving_stochastic_differential = deque(
                         (2 * np.random.randint(0, 2, steps) - 1) * np.sqrt(self.h))
                 elif 'Order_20' in self.__module__:
                     s = np.sqrt(3 * self.h)
                     self.driving_stochastic_differential = (
                     deque([0 if l in (1, 2, 3, 4) else s if l == 5 else -s for l in np.random.randint(0, 6, steps)]))
+            else:
+                raise TypeError('The proposed scheme is neither weak nor strong; no convergence order can be set.')
+
+        if len(self.driving_stochastic_differential) != steps:
+            raise ValueError('The resolution of the driving stochastic differential does not match the discretization.')
 
         self.dW = []
 
     def map_to_parameter_set(self, func, parameter, information):
-        func_parameter = {key: parameter[key] for key in information['parameter']}
+        func_parameter = tuple(parameter[key] for key in information['parameter'])
         if information['spatial']:
             if information['time']:
-                return lambda x, t: func(x=x, t=t, **func_parameter)
+                return lambda x, t: func(x, t, *func_parameter)
             else:
-                return lambda x, t: func(x=x, **func_parameter)
+                return lambda x, t: func(x, *func_parameter)
         elif information['time']:
-            return lambda x, t: func(t=t, **func_parameter)
+            return lambda x, t: func(t, *func_parameter)
         else:
-            return lambda x, t: func(**func_parameter)
+            return lambda x, t: func(*func_parameter)
 
     def propagation(self, x, t):
         pass

simulation/strong/implicit/taylor.py

@@ -26,7 +26,7 @@ def __init__(self, sde, parameter, steps, alpha=1):
     def state_equation(self, x, dW):
         explixit_diffusion = self.diffusion(self.x, self.t)
         return x - (
-        self.x + (self.alpha * self.drift(x, self.t + self.h) + (1 - self.alpha) * self.drift(self.x, self.t)) *
+        self.x + (self.alpha * self.drift(x, self.t + self.h) + (1 - self.alpha) * self.drift(self.x, self.t)) * \
         self.h + explixit_diffusion * dW + 0.5 * self.diffusion_x(self.x, self.t) * explixit_diffusion * (
         dW ** 2 - self.h))
 

tests/test_basic_scheme.py

@@ -0,0 +1,30 @@
+import pytest
+
+from sde import SDE
+from simulation.scheme import Scheme
+
+
+def test_missing_parameter():
+    with pytest.raises(KeyError):
+        sample_sde = SDE(lambda x: x, lambda c: c)
+        Scheme(sample_sde, parameter={}, steps=10)
+
+
+def test_path_size_consistency():
+    with pytest.raises(ValueError):
+        sample_sde = SDE(lambda x: x, lambda x: x)
+        dW = list(range(5))
+        Scheme(sample_sde, parameter={}, steps=10, path=dW)
+
+
+def test_scheme_derivation():
+    with pytest.raises(TypeError):
+        class SomeSpecialScheme(Scheme):
+            def __init__(self, sde, parameter, steps, **kwargs):
+                super().__init__(sde, parameter, steps, **kwargs)
+
+            def propagation(self, x, t):
+                self.x = x * t
+
+        sample_sde = SDE(lambda x: x, lambda x: x)
+        SomeSpecialScheme(sample_sde, parameter={}, steps=10)

tests/test_scheme_implementations.py

@@ -0,0 +1,23 @@
+import numpy as np
+
+from sde import SDE
+from simulation.strong.explicit.taylor import Order_05
+
+steps_used = 50
+sample_differential_brownian_motion = np.array([0.03329902, 0.20850244, 0.12094308, -0.14159548, 0.02973983,
+                                                0.06103259, -0.00915205, 0.01928274, 0.09207789, -0.13199381,
+                                                0.17663064, 0.1333172, -0.01288733, -0.31281056, -0.05924482,
+                                                -0.01702982, 0.18025385, -0.17514341, 0.03477228, 0.31712905,
+                                                -0.25351569, -0.19384718, -0.29929325, 0.20444405, 0.08353272,
+                                                0.09427778, 0.05516237, -0.18329133, -0.18365494, -0.13901742,
+                                                -0.15492822, 0.0384501, -0.0544241, -0.15041881, -0.07649629,
+                                                0.07692755, -0.12122493, 0.18393892, 0.12113368, 0.10871338,
+                                                -0.1328373, -0.05468304, 0.08074539, 0.52846189, -0.00426639,
+                                                0.04982364, 0.16280621, -0.03664431, 0.22651330, -0.08565257])
+
+
+def test_if_path_is_handed_through_correctly():
+    sample_sde = SDE(lambda x: x, lambda x: x)
+    euler_instance = Order_05(sample_sde, parameter={}, steps=steps_used, path=sample_differential_brownian_motion)
+    for _ in euler_instance: pass
+    assert all(euler_instance.return_path() == sample_differential_brownian_motion)

tests/test_sde.py

@@ -0,0 +1,11 @@
+import pytest
+
+from sde import SDE
+
+
+def test_if_construction_fails_without_function():
+    with pytest.raises(TypeError):
+        def test_drift(x):
+            return x
+
+        SDE(test_drift, 2)