move markov into own directory
This commit was merged in pull request #1.
This commit is contained in:
175
temp/markov.py
175
temp/markov.py
@@ -1,175 +0,0 @@
|
||||
import numpy as np
|
||||
import itertools
|
||||
import functools
|
||||
|
||||
""" Define our data """
|
||||
# The Statespace
|
||||
states = np.array(['L', 'w', 'W'])
|
||||
|
||||
# Possible sequences of events
|
||||
transitionName = np.array([['LL', 'Lw', 'LW'],
|
||||
['wL', 'ww', 'wW'],
|
||||
['WL', 'Ww', 'WW']])
|
||||
|
||||
# Probabilities Matrix (transition matrix)
|
||||
transitionMatrix = np.array([[0.8, 0.15, 0.05],
|
||||
[0.8, 0.15, 0.05],
|
||||
[0.8, 0.15, 0.05]])
|
||||
|
||||
# Starting state
|
||||
startingState = 'L'
|
||||
# Steps to run
|
||||
stepTime = 2
|
||||
# End state you want to find probabilites of
|
||||
endState = 'L'
|
||||
|
||||
|
||||
""" Set our parameters """
|
||||
# Should we seed the results?
|
||||
setSeed = False
|
||||
seedNum = 27
|
||||
|
||||
|
||||
""" Simulation parameters """
|
||||
# Should we simulate more than once?
|
||||
setSim = True
|
||||
simNum = 10
|
||||
|
||||
|
||||
# A class that implements the Markov chain to forecast the state/mood:
|
||||
class markov(object):
|
||||
"""simulates a markov chain given its states, current state and
|
||||
transition matrix.
|
||||
|
||||
Parameters:
|
||||
states: 1-d array containing all the possible states
|
||||
transitionName: 2-d array containing a list
|
||||
of the all possible state directions
|
||||
transitionMatrix: 2-d array containing all
|
||||
the probabilites of moving to each state
|
||||
currentState: a string indicating the starting state
|
||||
steps: an integer determining how many steps (or times) to simulate"""
|
||||
|
||||
def __init__(self, states: np.array, transitionName: np.array,
|
||||
transitionMatrix: np.array, currentState: str,
|
||||
steps: int):
|
||||
super(markov, self).__init__()
|
||||
self.states = states
|
||||
self.list = list
|
||||
self.transitionName = transitionName
|
||||
self.transitionMatrix = transitionMatrix
|
||||
self.currentState = currentState
|
||||
self.steps = steps
|
||||
|
||||
@staticmethod
|
||||
def setSeed(num: int):
|
||||
return np.random.seed(num)
|
||||
|
||||
@functools.lru_cache(maxsize=128)
|
||||
def forecast(self):
|
||||
print(f'Start state: {self.currentState}')
|
||||
# Shall store the sequence of states taken
|
||||
self.stateList = [self.currentState]
|
||||
i = 0
|
||||
# To calculate the probability of the stateList
|
||||
self.prob = 1
|
||||
while i != self.steps:
|
||||
if self.currentState == 'L':
|
||||
self.change = np.random.choice(self.transitionName[0],
|
||||
replace=True,
|
||||
p=transitionMatrix[0])
|
||||
if self.change == 'LL':
|
||||
self.prob = self.prob * 0.8
|
||||
self.stateList.append('L')
|
||||
pass
|
||||
elif self.change == 'Lw':
|
||||
self.prob = self.prob * 0.15
|
||||
self.currentState = 'w'
|
||||
self.stateList.append('w')
|
||||
else:
|
||||
self.prob = self.prob * 0.05
|
||||
self.currentState = "W"
|
||||
self.stateList.append("W")
|
||||
elif self.currentState == "w":
|
||||
self.change = np.random.choice(self.transitionName[1],
|
||||
replace=True,
|
||||
p=transitionMatrix[1])
|
||||
if self.change == "ww":
|
||||
self.prob = self.prob * 0.15
|
||||
self.stateList.append("w")
|
||||
pass
|
||||
elif self.change == "wL":
|
||||
self.prob = self.prob * 0.8
|
||||
self.currentState = "L"
|
||||
self.stateList.append("L")
|
||||
else:
|
||||
self.prob = self.prob * 0.05
|
||||
self.currentState = "W"
|
||||
self.stateList.append("W")
|
||||
elif self.currentState == "W":
|
||||
self.change = np.random.choice(self.transitionName[2],
|
||||
replace=True,
|
||||
p=transitionMatrix[2])
|
||||
if self.change == "WW":
|
||||
self.prob = self.prob * 0.05
|
||||
self.stateList.append("W")
|
||||
pass
|
||||
elif self.change == "WL":
|
||||
self.prob = self.prob * 0.8
|
||||
self.currentState = "L"
|
||||
self.stateList.append("L")
|
||||
else:
|
||||
self.prob = self.prob * 0.15
|
||||
self.currentState = "w"
|
||||
self.stateList.append("w")
|
||||
i += 1
|
||||
print(f'Possible states: {self.stateList}')
|
||||
print(f'End state after {self.steps} steps: {self.currentState}')
|
||||
print(f'Probability of all the possible sequence of states:'
|
||||
f' {self.prob}')
|
||||
return self.stateList
|
||||
|
||||
|
||||
def main(*args, **kwargs):
|
||||
try:
|
||||
simNum = kwargs['simNum']
|
||||
except KeyError:
|
||||
pass
|
||||
sumTotal = 0
|
||||
# Check validity of transitionMatrix
|
||||
for i in range(len(transitionMatrix)):
|
||||
sumTotal += sum(transitionMatrix[i])
|
||||
if i != len(states) and i == len(transitionMatrix):
|
||||
raise ValueError('Probabilities should add to 1')
|
||||
# Set the seed so we can repeat with the same results
|
||||
if setSeed:
|
||||
markov.setSeed(seedNum)
|
||||
# Save our simulations:
|
||||
list_state = []
|
||||
count = 0
|
||||
# Simulate Multiple Times
|
||||
if setSim:
|
||||
for _ in itertools.repeat(None, simNum):
|
||||
markovChain = markov(states, transitionName,
|
||||
transitionMatrix, startingState,
|
||||
stepTime)
|
||||
list_state.append(markovChain.forecast())
|
||||
else:
|
||||
for _ in range(1, 2):
|
||||
list_state.append(markov(states, transitionName,
|
||||
transitionMatrix, startingState,
|
||||
stepTime).forecast())
|
||||
for list in list_state:
|
||||
if(list[-1] == f'{endState!s}'):
|
||||
print(True, list)
|
||||
count += 1
|
||||
else:
|
||||
print(False, list)
|
||||
if setSim is False:
|
||||
simNum = 1
|
||||
print(f'\nThe probability of starting in {startingState} and finishing'
|
||||
f' in {endState} after {stepTime} steps is {(count / simNum):.2%}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main(simNum=simNum)
|
||||
Reference in New Issue
Block a user