Welcome, guest | Sign In | My Account | Store | Cart

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
  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
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)