A Markov chain is said to be irreducible if we can reach any state of the given Markov chain from any other state. In terms of states, state j is said to be accessible from another state i if a system that started at state i has a non-zero probability of getting to the state j.
In more formal terms, state j is said to be accessible from state i if an integer nij≥0 exists such that the following condition is met:
P(Xnij=j|X0=i)=Pnijij>0
import numpy as np
from itertools import combinations
class MarkovChain(object):
def __init__(self, transition_matrix, states):
"""
Initialize the MarkovChain instance.
Parameters
----------
transition_matrix: 2-D array
A 2-D array representing the probabilities of change of
state in the Markov Chain.
states: 1-D array
An array representing the states of the Markov Chain. It
needs to be in the same order as transition_matrix.
"""
self.transition_matrix = np.atleast_2d(transition_matrix)
self.states = states
self.index_dict = {self.states[index]: index for index in
range(len(self.states))}
self.state_dict = {index: self.states[index] for index in
range(len(self.states))}
def next_state(self, current_state):
"""
Returns the state of the random variable at the next time
instance.
Parameters
----------
current_state: str
The current state of the system.
"""
return np.random.choice(
self.states,
p=self.transition_matrix[self.index_dict[current_state], :])
def generate_states(self, current_state, no=10):
"""
Generates the next states of the system.
Parameters
----------
current_state: str
The state of the current random variable.
no: int
The number of future states to generate.
"""
future_states = []
for i in range(no):
next_state = self.next_state(current_state)
future_states.append(next_state)
current_state = next_state
return future_states
def is_accessible(self, i_state, f_state):
"""
Check if state f_state is accessible from i_state.
Parameters
----------
i_state: str
The state from which the accessibility needs to be checked.
f_state: str
The state to which accessibility needs to be checked.
"""
source_index = self.index_dict[i_state]
target_index = self.index_dict[f_state]
reachable_state_indexes = set([source_index])
for state_index in reachable_state_indexes:
if state_index == target_index:
return True
else:
reachable_index_list = list(np.nonzero(
self.transition_matrix[self.index_dict[i_state], :])[0])
reachable_state_indexes.union(set(reachable_index_list))
return False
def is_irreducible(self):
"""
Check if the Markov Chain is irreducible.
"""
for (i, j) in combinations(self.states,2):
if not self.is_accessible(i, j):
return False
return True
transition_irreducible = [[0.5, 0.5, 0, 0],
[0.25, 0, 0.5, 0.25],
[0.25, 0.5, 0, 0.25],
[0, 0, 0.5, 0.5]]
transition_reducible = [[0.5, 0.5, 0, 0],
[0, 1, 0, 0],
[0.25, 0.5, 0, 0],
[0, 0, 0.25, 0.75]]
markov_irreducible = MarkovChain(transition_matrix=transition_irreducible,
states=['A', 'B', 'C', 'D'])
markov_reducible = MarkovChain(transition_matrix=transition_reducible,
states=['A', 'B', 'C', 'D'])
list(np.nonzero(np.atleast_2d(transition_irreducible)[1, :])[0])
[0, 2, 3]
markov_irreducible.is_accessible(i_state='A', f_state='D')
False
markov_irreducible.is_accessible(i_state='B', f_state='D')
False
markov_irreducible.is_irreducible()
False
markov_reducible.is_accessible(i_state='A', f_state='D')
False
markov_reducible.is_accessible(i_state='D', f_state='A')
False
markov_reducible.is_accessible(i_state='C', f_state='D')
False
markov_reducible.is_irreducible()
False
State i is said to have period k if any possible path to return to state i would be a multiple of k steps. Formally, it is defined like this:
k=gcd{n>0:P(Xn=i|X0=i)>0}
Here, gcd means the greatest common divisor (GCD). Basically, k is the GCD of the length/number of steps of all possible paths from state i back to itself. If there are no possible paths from state i back to itself, then the period for it is not defined. We also need to note that k has nothing to do with the number of steps required to return to the starting state
import numpy as np
from itertools import combinations
class MarkovChain(object):
def __init__(self, transition_matrix, states):
"""
Initialize the MarkovChain instance.
Parameters
----------
transition_matrix: 2-D array
A 2-D array representing the probabilities of change of
state in the Markov Chain.
states: 1-D array
An array representing the states of the Markov Chain. It
needs to be in the same order as transition_matrix.
"""
self.transition_matrix = np.atleast_2d(transition_matrix)
self.states = states
self.index_dict = {self.states[index]: index for index in
range(len(self.states))}
self.state_dict = {index: self.states[index] for index in
range(len(self.states))}
def next_state(self, current_state):
"""
Returns the state of the random variable at the next time
instance.
Parameters
----------
current_state: str
The current state of the system.
"""
return np.random.choice(
self.states,
p=self.transition_matrix[self.index_dict[current_state], :])
def generate_states(self, current_state, no=10):
"""
Generates the next states of the system.
Parameters
----------
current_state: str
The state of the current random variable.
no: int
The number of future states to generate.
"""
future_states = []
for i in range(no):
next_state = self.next_state(current_state)
future_states.append(next_state)
current_state = next_state
return future_states
def is_accessible(self, i_state, f_state):
"""
Check if state f_state is accessible from i_state.
Parameters
----------
i_state: str
The state from which the accessibility needs to be checked.
f_state: str
The state to which accessibility needs to be checked.
"""
source_index = self.index_dict[i_state]
target_index = self.index_dict[f_state]
reachable_state_indexes = set([source_index])
for state_index in reachable_state_indexes:
if state_index == target_index:
return True
else:
reachable_index_list = list(np.nonzero(
self.transition_matrix[self.index_dict[i_state], :])[0])
reachable_state_indexes.union(set(reachable_index_list))
return False
def is_irreducible(self):
"""
Check if the Markov Chain is irreducible.
"""
for (i, j) in combinations(self.states,2):
if not self.is_accessible(i, j):
return False
return True
def get_period(self, state):
"""
Returns the period of the state in the Markov Chain.
Parameters
----------
state: str
The state for which the period needs to be computed.
"""
source_index = self.index_dict[i_state]
target_index = self.index_dict[f_state]
reachable_state_indexes = set([source_index])
for state_index in reachable_state_indexes:
if state_index == target_index:
return True
else:
reachable_index_list = list(np.nonzero(
self.transition_matrix[self.index_dict[i_state], :])[0])
reachable_state_indexes.union(set(reachable_index_list))
return False
return gcd([len(i) for i in all_possible_paths])
def is_aperiodic(self):
"""
Checks if the Markov Chain is aperiodic.
"""
periods = [self.get_period(state) for state in self.states]
for period in periods:
if period != 1:
return False
return True
def is_transient(self, state):
"""
Checks if a state is transient or not.
Parameters
----------
state: str
The state for which the transient property needs to be checked.
"""
if all(self.transition_matrix[~self.index_dict[state], self.index_dict[state]] == 0):
return True
else:
return False
transition_periodic = [[0, 1, 0, 0, 0],
[0, 0, 1, 0, 0],
[0.5, 0, 0, 0.5, 0],
[0, 0, 0, 0, 1],
[0, 0, 1, 0, 0]]
transition_aperiodic = [[0, 1, 0, 0, 0],
[0, 0, 1, 0, 0],
[0.5, 0.25, 0, 0.25, 0],
[0, 0, 0, 0, 1],
[0, 0, 0.5, 0.5, 0]]
markov_periodic = MarkovChain(transition_matrix=transition_periodic,
states=['A', 'B', 'C', 'D', 'E'])
markov_aperiodic = MarkovChain(transition_matrix=transition_aperiodic,
states=['A', 'B', 'C', 'D', 'E'])
markov_periodic.get_period('A')