This is a simple Monte Carlo Engine model, based on some ideas from the Quantopian folks. The idea is that the model and the engine are separate. I have integrated a simple example from an earlier post. The idea was to try and build some game theory simulations using simple models that were separate from the Engine.
Python, 241 lines
Download
Copy to clipboard1 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 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 | import unittest, datetime
import numpy as np
from engine.BackTest import MonteCarloModel, MonteCarloEngine, Simulation
import matplotlib.pyplot as plt
from matplotlib.dates import date2num, DateFormatter, DayLocator
from pandas import DataFrame
from numpy.random import standard_normal
from numpy import array, zeros, sqrt, shape
class SimpleMonteCarloModel(MonteCarloModel):
'''
simple example implementation of a model. This will be back tested based on the data
passed to the initialise method.
class ExampleModel(MonteCarloModel):
def initialise(self, context):
context['My Simple Model'] = Simulation(10, 100, standard_normal())
print 'setting My Simple Model'
def ondata(self, model, simulation, trial, value, engine):
print simulation, trial, value, engine
'''
def initialise(self, context):
self.r0 = 0.05 # current UK funding rate
self.theta = 0.10 # 1 % long term interest rate
self.k = 0.3
self.beta = 0.03
## simulate short rate paths
self.n = 1000 # MC simulation trials
self.T = 24. # total time
self.m = 100 # subintervals
self.dt = self.T/self.m # difference in time each subinterval
self.r = np.zeros(shape=(self.n, self.m), dtype=float) # matrix to hold short rate paths
# Tell the engine where to associate the data to security.
context['My Simple Model'] = Simulation(self.n, self.m, standard_normal)
self.fig = plt.figure()
self.ax = self.fig.add_subplot(111)
self.ax.autoscale_view(True,True,True)
def onsimulation(self, model, simulation, engine):
#print 'new simulation,', model, simulation
#plot(np.arange(0, T, dt), r[j])
self.r[simulation,0] = self.r0
def aftersimulation(self, model, simulation, engine):
self.ax.plot(np.arange(0, self.T, self.dt), self.r[simulation])
def finalise(self, model, engine):
self.t = np.arange(0, self.T, self.dt)
self.rT_expected = self.theta + (self.r0-self.theta)*pow(np.e,-self.k*self.t)
self.rT_stdev = sqrt( pow(self.beta,2)/(2*self.k)*(1-pow(np.e,-2*self.k*self.t)))
#print 'expected', self.rT_expected, 'std', self.rT_stdev
self.ax.plot(self.t, self.rT_expected, '-+r')
self.ax.plot(self.t, self.rT_expected+2*self.rT_stdev, '-b')
self.ax.plot(self.t, self.rT_expected-2*self.rT_stdev, '-b')
self.ax.plot(self.t, self.rT_expected+4*self.rT_stdev, '-g')
self.ax.plot(self.t, self.rT_expected-4*self.rT_stdev, '-g')
print shape(self.t), shape(self.r)
plt.title('Simulations %d Steps %d r0 %.2f alpha %.2f beta %.2f sigma %.2f' % (int(self.n), int(self.m), self.r0, self.k, self.theta, self.beta))
plt.xlabel('steps')
plt.ylabel('short term interest rate')
plt.show()
def ontrial(self, model, simulation, trial, value, engine):
'''
called when we have some new data to play into the model.
each call is an event
value : float
sample from model
'''
#print simulation, trial, value
# evaluation the model step.
self.r[simulation,trial] = self.r[simulation,trial-1] + self.k*(self.theta-self.r[simulation,trial-1])*self.dt + self.beta*sqrt(self.dt)*value;
class TestNode(unittest.TestCase):
def setUp(self):
pass
def test_engine(self):
'''
example of how to launch the BackTestEngine
this is modelled on the quantopian style interface.
'''
e = MonteCarloEngine(moduleName='MonteCarloTestExample', className='SimpleMonteCarloModel')
e.start()
if __name__ == '__main__':
unittest.main()
import unittest, time, datetime
from abc import ABCMeta, abstractmethod
#
# Simple python component wrapper
#
class Component(object):
__metaclass__ = ABCMeta
@abstractmethod
def start(self):
raise NotImplementedError("Should implement intialise()!")
class BackTestModel(object):
__metaclass__ = ABCMeta
@abstractmethod
def initialise(self, context):
raise NotImplementedError("Should implement intialise()!")
@abstractmethod
def ondata(self, sid, data):
raise NotImplementedError("Should implement ondata()!")
class MonteCarloModel(object):
__metaclass__ = ABCMeta
@abstractmethod
def initialise(self, context):
raise NotImplementedError("Should implement intialise()!")
@abstractmethod
def ontrial(self, model, simulation, trial, value, engine):
raise NotImplementedError("Should implement ontrial()!")
def onsimulation(self, model, simulation, engine):
raise NotImplementedError("Should implement onsimulation()!")
def aftersimulation(self, model, simulation, engine):
raise NotImplementedError("Should implement aftersimulation()!")
def finalise(self, model, engine):
raise NotImplementedError("Should implement finalise()!")
class Simulation(object):
'''
simple wrapper that describes a simulation
'''
def __init__(self, n, m, func):
'''
n integer
number of simulations
m integer
number of trials per simulation
func class method or function
used to sample the value
'''
self.number_of_simulations = n
self.number_of_trials = m
self.func = func
@property
def sample(self):
return self.func
class MonteCarloEngineException(Exception):
pass
class MonteCarloEngine(Component):
'''
twist on the Engine that will take a different type of context, this time
a Simulation class instance. This will be called
class ExampleModel(MonteCarloModel):
def initialise(self, context):
context['My Simple Model'] = Simulation(10, 100, standard_normal())
print 'setting My Simple Model'
def ondata(self, model, simulation, trial, value, engine):
print simulation, trial, value, engine
'''
def __init__(self, moduleName, className):
self.context = {}
self.obj = self.__generate__(moduleName, className)
print isinstance(self.obj, MonteCarloModel), hasattr(self.obj, 'initialise')
if isinstance(self.obj, MonteCarloModel):
if hasattr(self.obj, 'initialise'):
self.obj.initialise(self.context)
print 'calling initialise'
else:
print 'no initialise'
#
# TODO: load data from somewhere for securities in context
#
print self.context
def __generate__(self, module_name, class_name):
module = __import__(module_name)
class_ = getattr(module, class_name)
instance = class_()
print instance
return instance
def start(self):
print 'starting...', self.obj, self.context
if self.obj is None:
raise MonteCarloEngineException('No engine exists')
for name, model in self.context.items():
for simulation in np.arange(0, model.number_of_simulations): # number of MC simulations
# call to signal new simulation
if hasattr(self.obj, 'onsimulation'):
self.obj.onsimulation(model, simulation, self)
for trial in np.arange(1,model.number_of_trials): #trials per simulation
value = model.sample()
if hasattr(self.obj, 'ontrial'):
self.obj.ontrial(model, simulation, trial, value, self)
# call to signal after simulation
if hasattr(self.obj, 'aftersimulation'):
self.obj.aftersimulation(model, simulation, self)
#self.post_ondata(k, index2, value)
if hasattr(self.obj, 'finalise'):
self.obj.finalise(model, self)
|
