-
Notifications
You must be signed in to change notification settings - Fork 79
/
diffusion.py
134 lines (106 loc) · 4.3 KB
/
diffusion.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import numpy as np
from stochastic.processes.noise import GaussianNoise
from stochastic.utils import ensure_single_arg_constant_function
from stochastic.utils.validation import check_numeric
from stochastic.utils.validation import check_numeric_or_single_arg_callable
from stochastic.utils.validation import check_positive_integer
class DiffusionProcess(GaussianNoise):
r"""Generalized diffusion process.
A base process for more specific diffusion processes.
The process :math:`X_t` that satisfies the following
stochastic differential equation with Wiener process :math:`W_t`:
.. math::
dX_t = \theta_t (\mu_t - X_t) dt + \sigma_t X_t^{\gamma_t} dW_t
Realizations are generated using the Euler-Maruyama method.
.. note::
Since the family of diffusion processes have parameters which
generalize to functions of ``t``, parameter attributes will be returned
as callables, even if they are initialized as constants. e.g. a
``speed`` parameter of 1 accessed from an instance attribute will return
a function which accepts a single argument and always returns 1.
:param func speed: the speed of reversion, or :math:`\theta_t` above
:param func mean: the mean of the process, or :math:`\mu_t` above
:param func vol: volatility coefficient of the process, or :math:`\sigma_t`
above
:param func volexp: volatility exponent of the process, or :math:`\gamma_t`
above
:param float t: the right hand endpoint of the time interval :math:`[0,t]`
for the process
:param numpy.random.Generator rng: a custom random number generator
"""
def __init__(self, speed=1, mean=0, vol=1, volexp=0, t=1, rng=None):
super().__init__(t=t, rng=rng)
self.speed = speed
self.mean = mean
self.vol = vol
self.volexp = volexp
def __str__(self):
return "Diffusion process with speed={s}, mean={m}, vol={v}, volexp={e} on [0, {t}]".format(
s=str(self.speed),
m=str(self.mean),
v=str(self.vol),
e=str(self.volexp),
t=str(self.t),
)
def __repr__(self):
return "Diffusion(speed={s}, mean={m}, vol={v}, volexp={e} t={t})".format(
s=str(self.speed),
m=str(self.mean),
v=str(self.vol),
e=str(self.volexp),
t=str(self.t),
)
@property
def speed(self):
"""Speed, or :math:`\theta_t`."""
return self._speed
@speed.setter
def speed(self, value):
check_numeric_or_single_arg_callable(value, "speed")
self._speed = ensure_single_arg_constant_function(value)
@property
def mean(self):
r"""Mean, or :math:`\mu_t`."""
return self._mean
@mean.setter
def mean(self, value):
check_numeric_or_single_arg_callable(value, "mean")
self._mean = ensure_single_arg_constant_function(value)
@property
def vol(self):
r"""Volatility, or :math:`\sigma_t`."""
return self._vol
@vol.setter
def vol(self, value):
check_numeric_or_single_arg_callable(value, "vol")
self._vol = ensure_single_arg_constant_function(value)
@property
def volexp(self):
r"""Volatility exponent, or :math:`\gamma_t`."""
return self._volexp
@volexp.setter
def volexp(self, value):
check_numeric_or_single_arg_callable(value, "volexp")
self._volexp = ensure_single_arg_constant_function(value)
def _sample(self, n, initial=1.0):
"""Generate a realization of a diffusion process using Euler-Maruyama."""
check_positive_integer(n)
check_numeric(initial, "Initial")
delta_t = 1.0 * self.t / n
gns = self._sample_gaussian_noise(n)
s = [initial]
t = 0
for k in range(n):
t += delta_t
initial += (
self._speed(t) * (self._mean(t) - initial) * delta_t
+ self._vol(t) * initial ** self._volexp(initial) * gns[k]
)
s.append(initial)
return np.array(s)
def sample(self, n, initial=1.0):
"""Generate a realization.
:param int n: the number of increments to generate
:param float initial: the initial value of the process
"""
return self._sample(n, initial)