Monte Carlo Engine : How to find the optimised wager for next bet, following a recent loss. (Python recipe) by alexander baker

ActiveState Code (http://code.activestate.com/recipes/578869/)

'''
    monte carlo tool for simple strategy, more generally for use with any
    strategy that I want to back test...

    Engine takes a class instance, derived from a base class wih two methods
    
    initialise()
    onsimulation()
    aftersimulation()
    ontrial()
    finalise()


'''

import unittest, datetime
import numpy as np
from BackTest import MonteCarloModel, MonteCarloEngine, Simulation 
import matplotlib.pyplot as plt
from random import randint 
from scipy.optimize import fmin_powell
from hashlib import md5
from time import localtime

def Root2(r,verbose=True):
    '''
        simple wrapper routine for solver. returns the energy level for a given R=r.
        the solver can use this method to minimise the energy by varying r as necessary.
    '''
    if r <= 0:
        return 1e99
    else:
        e = MonteCarloEngine(moduleName='MonteCarloHeadsTailsExample', className='SimpleHeadTailModel')
        losers = e.start(args={'wager_multiplier':r})
        if verbose: print 'solving for r: ', r
        return losers              
        

class SimpleHeadTailModel(MonteCarloModel):
    '''
      model wager
      previous bet success or not, impact to profit.
        
    '''
    
    def toss(self):
        '''
            1 - win
            -1 - loss
            Each toss determines whether the position is successful or not. This way no need to keep track of a decision
            and an associated variable. Simply do I win or not.
        '''
        coin_toss = randint(1,2)
        if coin_toss == 1:
            return 1
        else:
            return -1
            
    def initialise(self, context):
                
        self.name = 'My Simple Heads And Tails Model'
        # start with 10 USD bet
        self.wager= 100
        self.wager_initial= 100
        self.starting_pot = 1000
        
        self.previous_value = 1 # default to 1 on first round
        self.simulations = 100    # MC simulation trials        
        self.trials = 100   # subintervals

        self.r = np.zeros(shape=(self.simulations, self.trials), dtype=float) # matrix to hold all results
        self.pnl = np.zeros(shape=(self.simulations, self.trials), dtype=float) # matrix to hold all results
        print "simulations %d, trials %d starting pot %d " % (self.simulations, self.trials, self.starting_pot)
        
        # Tell the engine where to associate the data to security.        
        context[self.name] = Simulation(self.simulations, self.trials, self.toss)
    
        self.fig = plt.figure()
        self.ax = self.fig.add_subplot(211)
        self.ax1 = self.fig.add_subplot(212)
        self.ax.autoscale_view(True,True,True)        
                        
    
    def onsimulation(self, model, simulation, engine):        
        self.r[simulation,0] = 0
        # assume starting pot here
        self.pnl[simulation,0] = self.starting_pot
        
    def aftersimulation(self, model, simulation, engine):
        self.ax.plot(np.arange(0, self.trials, 1), self.r[simulation])        
        self.ax1.plot(np.arange(0, self.trials, 1), self.pnl[simulation])
        
    def reset_wager(self):
        self.wager = self.wager_initial
        
    def ontrial(self, model, simulation, trial, value, engine, args):
        '''
            want to test some strategies for betting
            set wager for each bet
            if previous bet

            value : float
                sample from model
            
        '''
                     
	   
        # if we lost last time then double up		
		
        if self.previous_value == -1:
		if args.has_key('args'):
                	self.wager = (self.wager*float(args['args']['wager_multiplier'][0]))
		else:
			self.wager = (self.wager*0.1)      
        
        # keep track of coin toss paths
        self.r[simulation,trial] = self.r[simulation,trial-1] + value
        if args.has_key('args'):
            self.r0 = float(args['args']['wager_multiplier'][0])
        else:
			self.r0 =0.1
        # if we won, add the wager
        # else subtract the wager
        
        if self.pnl[simulation,trial-1] > 0:        
            if value == 1 :
                self.pnl[simulation,trial] = self.pnl[simulation,trial-1] + self.wager
            else:
                self.pnl[simulation,trial] = self.pnl[simulation,trial-1] - self.wager
        else:
            # no bet to be made here
            self.pnl[simulation,trial] = self.pnl[simulation,trial-1]
             
        # always reset wager    
        self.reset_wager()
        
        # keep track of the previous value for next time around
        self.previous_value = value
        
	def add_prefix(self, filename):
		from hashlib import md5
		from time import localtime
		return "%s_%s"%(md5(str(localtime())).hexdigest(), filename)	
		
    def finalise(self, model, engine):  
		'''
			returns the value that we are trying to minimise, here the number of losers.
		'''
        
	# what is our survivability

		number_of_losers = len([f for f in self.pnl if f[len(f)-1]<=0])
		number_of_survivors = len([f for f in self.pnl if f[len(f)-1]>0])
		number_of_participants = len(self.pnl)
		print "participants [%d] survivors [%2.1f%%] losers [%2.1f%%] weight [%2.6f] "% (number_of_participants, float(number_of_survivors)/float(number_of_participants)*100, float(number_of_losers)/float(number_of_participants)*100, self.r0)

		plt.title('Simulations %d Steps %d' % (int(self.simulations), int(self.trials)))
		plt.xlabel('steps')
		plt.ylabel('profit and loss')					
		plt.savefig("%s_%s"%(md5(str(localtime())).hexdigest(), 'model'))
		return float(number_of_losers)/float(number_of_participants)*100
        
		
class TestNode(unittest.TestCase):
    def setUp(self):
        pass    
   
    def test_engine(self):
        '''
            example of how to launch the MontoCarloTestEngine
            this is modelled on the quantopian style interface.
        '''		
        e = MonteCarloEngine(moduleName='MonteCarloHeadsTailsExample', className='SimpleHeadTailModel')
        e.start()
    
    def test_minimise(self):
                
        print '#################################'
        print '# Test Equilibrium Loss Wager'
        print '#################################'
        
        wager_multiplier=fmin_powell(Root2, x0=1., maxiter=20)
        print "highest survivability following loss, multiply wager by %2.4f %% "%(wager_multiplier*100)

if __name__ == '__main__':
    unittest.main()


==================================================================
== BackTest module
==================================================================


'''
	back testing tool for prediction strategy, more generally for use with any
	strategy that I want to back test...

	Engine takes a class instance, derived from a base class wih twomethods

	
	initialise()
	ondata()


'''

import unittest, time, datetime
from pandas import DataFrame
import numpy as np
from pylab import show
import random
from abc import ABCMeta, abstractmethod
import pandas as pd

#
# 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, args):
		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, **args):
			#print 'starting...', self.obj, self.context, args		
			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, args)
																
						# 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'):
				return self.obj.finalise(model, self)
	
'''
class ExampleModel(BackTestModel):

	def initialise(self, context):
		context['ARM.L'] = 'Book1.csv'
		print 'setting ARM.L' 
		
	def ondata(self, sid, data):
		print sid, data
'''
class Engine(Component):
	'''
		responsible for handling instances of the back test models
	'''
	def __init__(self, moduleName, className):
		
		self.context = {}
		self.data = {}
		self.orders = {}
		self.positions = {}
		self.pnl = {}
		self.risk = {}

		self.obj = self.__generate__(moduleName, className)
		#print isinstance(self.obj, BackTestModel), hasattr(self.obj, 'initialise')
		if isinstance(self.obj, BackTestModel):
			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	
		for k in self.context.keys():
			#print k
			t = time.clock()
			myfilename = self.context[k]
		      	data = DataFrame.from_csv(myfilename,header=0,index_col=0,parse_dates=True)	
			print 'load data', time.clock()-t						
			self.data[k] = data
			self.positions[k] = 0
			self.pnl[k] = 0
			self.risk[k] = 0
	
	def order(self, sid, value):
		# queue order to be processed
		self.orders[sid] = (value, False)
	
	@property
	def position(self):
		return self.positions
	
	def __generate__(self, module_name, class_name):
		module = __import__(module_name)
		class_ = getattr(module, class_name)
		instance = class_()
		print instance 
		return instance
	
	def post_ondata(self, sid, index, value):
		# process orders
		for k in self.context.keys():
			if hasattr(self, 'orders'):
				if len(self.orders) == 0:
					print 'no orders'
					break			
				
			m_size, m_processed = self.orders[k]
			# check we have not processed this order already.
			if not m_processed:
				if not (self.positions[k]+ m_size <= 0):				
					self.positions[k] += m_size
					print 'ordering %d of %s, total %d' % (m_size, k, self.positions[k])
					self.orders[k] = None
				else:
					print 'no short selling'				
		# handle pnl and risk
		# tick() charts
		

	def start(self):
		print 'starting...', self.obj, self.context		
		if not self.obj is None:	
			for k in self.context:
				for i, (index, value) in enumerate(self.data[k]['value'].iteritems()):					
					index2 = datetime.datetime(pd.to_datetime(index).year
											, pd.to_datetime(index).month
											, pd.to_datetime(index).day)					
					self.obj.ondata(k
								, index2
								, value
								, self.data[k]['value'][0:i]
								, self)					
					self.post_ondata(k, index2, value)