import torch
import torch.autograd.functional as F
import torch.distributions as dist
import numpy as np
import matplotlib.pyplot as plt
import pandas as pd
%matplotlib inline
# Retina display
%config InlineBackend.figure_format = 'retina'
Parameters
from tueplots import bundles
plt.rcParams.update(bundles.beamer_moml())#plt.rcParams.update(bundles.icml2022())
# Also add despine to the bundle using rcParams
'axes.spines.right'] = False
plt.rcParams['axes.spines.top'] = False
plt.rcParams[
# Increase font size to match Beamer template
'font.size'] = 16
plt.rcParams[# Make background transparent
'figure.facecolor'] = 'none' plt.rcParams[
# Prior distribution
= torch.tensor([0.5, 0.5])
P
# Transition matrix
= torch.tensor([[0.9, 0.1],
A 0.2, 0.8]])
[
# States
= {0: 'Sunny', 1: 'Rainy'} STATES
Sampling from discrete state Markov chain
# Generate data from the Markov chain
def generate_data(n_samples=10, seed=0):
torch.manual_seed(seed)= torch.zeros(n_samples, dtype=torch.long)
data 0] = dist.Categorical(P).sample()
data[for i in range(1, n_samples):
= dist.Categorical(A[data[i-1]]).sample()
data[i] return data
def map_state(x_arr):
return [STATES[x.item()] for x in x_arr]
10, 0)
generate_data(10, 0)) map_state(generate_data(
['Rainy',
'Rainy',
'Rainy',
'Rainy',
'Rainy',
'Sunny',
'Sunny',
'Sunny',
'Sunny',
'Sunny']
10, 1)) map_state(generate_data(
['Sunny',
'Sunny',
'Sunny',
'Sunny',
'Rainy',
'Rainy',
'Rainy',
'Sunny',
'Sunny',
'Sunny']
Stationary distribution
= []
PS = [] PR
# P(Sunny) at t=0
0].item())
PS.append(P[ PS
[0.5]
# P (Rainy) at t=0
1].item())
PR.append(P[ PR
[0.5]
def PS_t(t):
return PS[t-1] * A[0, 0].item() + PR[t-1] * A[1, 0].item()
def PR_t(t):
return PS[t-1] * A[0, 1].item() + PR[t-1] * A[1, 1].item()
for t in range(1, 20):
PS.append(PS_t(t)) PR.append(PR_t(t))
PS, PR
([0.5,
0.5499999895691872,
0.5849999801814558,
0.609499971553684,
0.6266499634787454,
0.6386549558053933,
0.647058448423374,
0.6529408912524431,
0.657058599234283,
0.6599409928265687,
0.6619586663486209,
0.6633710358232277,
0.6643596924658253,
0.6650517501268584,
0.6655361885013855,
0.6658752933757711,
0.6661126648003464,
0.6662788228102565,
0.6663951314300427,
0.6664765454768411],
[0.5,
0.45000000670552254,
0.4150000105053187,
0.39050001224130404,
0.37335001251176025,
0.36134501174174294,
0.352941510233172,
0.34705905820045785,
0.3429413407958347,
0.34005893762736905,
0.33804125442175925,
0.33662887518843054,
0.33564020873449596,
0.3349481412252955,
0.33446369297681955,
0.33412457821043834,
0.33388719688123464,
0.3337210289578531,
0.33360471041840567,
0.333523286447613])
='P(Sunny)')
plt.plot(PS, label='P(Rainy)')
plt.plot(PR, label'Time')
plt.xlabel('Probability')
plt.ylabel( plt.legend()
<matplotlib.legend.Legend at 0x7f1af7260910>
@A, torch.matrix_power(A, 2) A
(tensor([[0.8300, 0.1700],
[0.3400, 0.6600]]),
tensor([[0.8300, 0.1700],
[0.3400, 0.6600]]))
Vectorised version
# Distribution of states at t = 0
P
tensor([0.5000, 0.5000])
# P(Sunny) at t = 1
print("--"*20)
print("P(Sunny) at t = 1")
print(P[0] * A[0, 0] + P[1] * A[1, 0])
# P(Rainy) at t = 1
print("--"*20)
print("P(Rain) at t = 1")
print(P[0] * A[0, 1] + P[1] * A[1, 1])
# Distribution of states at t = 1
print("--"*20)
print("Distribution of states at t = 1")
print(P@A)
----------------------------------------
P(Sunny) at t = 1
tensor(0.5500)
----------------------------------------
P(Rain) at t = 1
tensor(0.4500)
----------------------------------------
Distribution of states at t = 1
tensor([0.5500, 0.4500])
# Distribution of states at t = 2
print("--"*20)
print("Distribution of states at t = 2")
print(P@A@A)
# Using torch.matrix_power
print("--"*20)
print("Distribution of states at t = 2")
print(P@torch.matrix_power(A, 2))
----------------------------------------
Distribution of states at t = 2
tensor([0.5850, 0.4150])
----------------------------------------
Distribution of states at t = 2
tensor([0.5850, 0.4150])
# convergence
@torch.matrix_power(A, 100), P@torch.matrix_power(A, 99) P
(tensor([0.6667, 0.3333]), tensor([0.6667, 0.3333]))
PS
[0.5,
0.5499999895691872,
0.5849999801814558,
0.609499971553684,
0.6266499634787454,
0.6386549558053933,
0.647058448423374,
0.6529408912524431,
0.657058599234283,
0.6599409928265687,
0.6619586663486209,
0.6633710358232277,
0.6643596924658253,
0.6650517501268584,
0.6655361885013855,
0.6658752933757711,
0.6661126648003464,
0.6662788228102565,
0.6663951314300427,
0.6664765454768411]
### What if we started with different initial distribution?
= torch.tensor([0.1, 0.9])
P2 = torch.tensor([0.9, 0.1])
P3 = torch.tensor([0.999999, 0.000001])
P4
@torch.matrix_power(A, 100), P3@torch.matrix_power(A, 100), P4@torch.matrix_power(A, 100) P2
(tensor([0.6507, 0.3493]), tensor([0.6733, 0.3267]), tensor([0.6761, 0.3239]))
### Checking for convergence of Markov chain using iterative method
= 1e-6
eps = P
PS for i in range(100):
= PS@A
PS_new if torch.all(torch.abs(PS_new - PS) < eps):
print("Converged at iteration", i)
break
= PS_new PS
Converged at iteration 31
### Checking for convergence
def check_convergence(P, A, iter=100, eps=1e-6):
for i in range(iter):
= P@A
P_new if torch.all(torch.abs(P_new - P) < eps):
print("Converged at iteration", i)
break
= P_new
P return P
check_convergence(P, A)
Converged at iteration 31
tensor([0.6667, 0.3333])
check_convergence(P2, A)
Converged at iteration 34
tensor([0.6667, 0.3333])
check_convergence(P3, A)
Converged at iteration 32
tensor([0.6667, 0.3333])
check_convergence(P4, A)
Converged at iteration 33
tensor([0.6667, 0.3333])
Homogeneous Markov chain
The transition matrix is the same for all time steps.
Irreducible Markov chain
A Markov chain is irreducible if it is possible to get to any state from any state.
Aperiodic Markov chain
A Markov chain is aperiodic if there are no cycles in the state transition graph.
### Markov chain with aperiodic transition matrix
= torch.tensor([[0.9, 0.1],
A 0.2, 0.8]])
[
### Continuous space Markov chain
def transition(a, b):
return dist.Normal(a, 0.1).log_prob(b).exp()