Markov-Theory.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Properties of Markov Chains"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Irreducible: \n",
    "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. \n",
    "\n",
    "In more formal terms, state j is said to be accessible from state i if an integer $n_{ij} ≥ 0$ exists such that the following condition is met:\n",
    "\n",
    "$\\large{P(X_{n_{ij}} = j|X_{0} = i) = P_{ij}^{n_{ij}} > 0}$\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from itertools import combinations\n",
    "\n",
    "class MarkovChain(object):\n",
    "    def __init__(self, transition_matrix, states):\n",
    "        \"\"\"\n",
    "        Initialize the MarkovChain instance.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        transition_matrix: 2-D array\n",
    "            A 2-D array representing the probabilities of change of \n",
    "            state in the Markov Chain.\n",
    "\n",
    "        states: 1-D array \n",
    "            An array representing the states of the Markov Chain. It\n",
    "            needs to be in the same order as transition_matrix.\n",
    "        \"\"\"\n",
    "        self.transition_matrix = np.atleast_2d(transition_matrix)\n",
    "        self.states = states\n",
    "        self.index_dict = {self.states[index]: index for index in \n",
    "                           range(len(self.states))}\n",
    "        self.state_dict = {index: self.states[index] for index in\n",
    "                           range(len(self.states))}\n",
    "\n",
    "    def next_state(self, current_state):\n",
    "        \"\"\"\n",
    "        Returns the state of the random variable at the next time \n",
    "        instance.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        current_state: str\n",
    "            The current state of the system.\n",
    "        \"\"\"\n",
    "        return np.random.choice(\n",
    "                    self.states, \n",
    "                    p=self.transition_matrix[self.index_dict[current_state], :])\n",
    "\n",
    "    def generate_states(self, current_state, no=10):\n",
    "        \"\"\"\n",
    "        Generates the next states of the system.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        current_state: str\n",
    "            The state of the current random variable.\n",
    "\n",
    "        no: int\n",
    "            The number of future states to generate.\n",
    "        \"\"\"\n",
    "        future_states = []\n",
    "        for i in range(no):\n",
    "            next_state = self.next_state(current_state)\n",
    "            future_states.append(next_state)\n",
    "            current_state = next_state\n",
    "        return future_states\n",
    "\n",
    "\n",
    "    def is_accessible(self, i_state, f_state):\n",
    "        \"\"\"\n",
    "        Check if state f_state is accessible from i_state.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        i_state: str\n",
    "            The state from which the accessibility needs to be checked.\n",
    "\n",
    "        f_state: str\n",
    "            The state to which accessibility needs to be checked.\n",
    "        \"\"\"\n",
    "        \n",
    "        source_index = self.index_dict[i_state]\n",
    "        target_index = self.index_dict[f_state]\n",
    "        reachable_state_indexes = set([source_index])\n",
    "        \n",
    "        for state_index in reachable_state_indexes:\n",
    "            if state_index == target_index:\n",
    "                return True\n",
    "            else:\n",
    "                reachable_index_list = list(np.nonzero(\n",
    "                  self.transition_matrix[self.index_dict[i_state], :])[0])\n",
    "                reachable_state_indexes.union(set(reachable_index_list))                 \n",
    "        return False\n",
    "    \n",
    "\n",
    "    def is_irreducible(self):\n",
    "        \"\"\"\n",
    "        Check if the Markov Chain is irreducible.\n",
    "        \"\"\"\n",
    "        for (i, j) in combinations(self.states,2):\n",
    "            if not self.is_accessible(i, j):\n",
    "                return False\n",
    "        return True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [],
   "source": [
    "transition_irreducible = [[0.5, 0.5, 0, 0],\n",
    "                              [0.25, 0, 0.5, 0.25],\n",
    "                              [0.25, 0.5, 0, 0.25],\n",
    "                              [0, 0, 0.5, 0.5]]\n",
    "transition_reducible = [[0.5, 0.5, 0, 0],\n",
    "                            [0, 1, 0, 0],\n",
    "                            [0.25, 0.5, 0, 0],\n",
    "                            [0, 0, 0.25, 0.75]]\n",
    "\n",
    "markov_irreducible = MarkovChain(transition_matrix=transition_irreducible,\n",
    "                                     states=['A', 'B', 'C', 'D'])\n",
    "markov_reducible = MarkovChain(transition_matrix=transition_reducible,\n",
    "                                   states=['A', 'B', 'C', 'D'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[0, 2, 3]"
      ]
     },
     "execution_count": 148,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "list(np.nonzero(np.atleast_2d(transition_irreducible)[1, :])[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 149,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_irreducible.is_accessible(i_state='A', f_state='D')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 150,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_irreducible.is_accessible(i_state='B', f_state='D')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 151,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_irreducible.is_irreducible()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 152,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 152,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_reducible.is_accessible(i_state='A', f_state='D')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 153,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 153,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_reducible.is_accessible(i_state='D', f_state='A')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 154,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_reducible.is_accessible(i_state='C', f_state='D')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 155,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 155,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "markov_reducible.is_irreducible()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Periodicity"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "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:\n",
    "\n",
    "$\\large{k = gcd \\{n>0:P(X_n = i|X_0 = i)>0 \\} }$"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "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"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 165,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from itertools import combinations\n",
    "\n",
    "class MarkovChain(object):\n",
    "    def __init__(self, transition_matrix, states):\n",
    "        \"\"\"\n",
    "        Initialize the MarkovChain instance.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        transition_matrix: 2-D array\n",
    "            A 2-D array representing the probabilities of change of \n",
    "            state in the Markov Chain.\n",
    "\n",
    "        states: 1-D array \n",
    "            An array representing the states of the Markov Chain. It\n",
    "            needs to be in the same order as transition_matrix.\n",
    "        \"\"\"\n",
    "        self.transition_matrix = np.atleast_2d(transition_matrix)\n",
    "        self.states = states\n",
    "        self.index_dict = {self.states[index]: index for index in \n",
    "                           range(len(self.states))}\n",
    "        self.state_dict = {index: self.states[index] for index in\n",
    "                           range(len(self.states))}\n",
    "\n",
    "    def next_state(self, current_state):\n",
    "        \"\"\"\n",
    "        Returns the state of the random variable at the next time \n",
    "        instance.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        current_state: str\n",
    "            The current state of the system.\n",
    "        \"\"\"\n",
    "        return np.random.choice(\n",
    "                    self.states, \n",
    "                    p=self.transition_matrix[self.index_dict[current_state], :])\n",
    "\n",
    "    def generate_states(self, current_state, no=10):\n",
    "        \"\"\"\n",
    "        Generates the next states of the system.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        current_state: str\n",
    "            The state of the current random variable.\n",
    "\n",
    "        no: int\n",
    "            The number of future states to generate.\n",
    "        \"\"\"\n",
    "        future_states = []\n",
    "        for i in range(no):\n",
    "            next_state = self.next_state(current_state)\n",
    "            future_states.append(next_state)\n",
    "            current_state = next_state\n",
    "        return future_states\n",
    "\n",
    "\n",
    "    def is_accessible(self, i_state, f_state):\n",
    "        \"\"\"\n",
    "        Check if state f_state is accessible from i_state.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        i_state: str\n",
    "            The state from which the accessibility needs to be checked.\n",
    "\n",
    "        f_state: str\n",
    "            The state to which accessibility needs to be checked.\n",
    "        \"\"\"\n",
    "        \n",
    "        source_index = self.index_dict[i_state]\n",
    "        target_index = self.index_dict[f_state]\n",
    "        reachable_state_indexes = set([source_index])\n",
    "        \n",
    "        for state_index in reachable_state_indexes:\n",
    "            if state_index == target_index:\n",
    "                return True\n",
    "            else:\n",
    "                reachable_index_list = list(np.nonzero(\n",
    "                  self.transition_matrix[self.index_dict[i_state], :])[0])\n",
    "                reachable_state_indexes.union(set(reachable_index_list))                 \n",
    "        return False\n",
    "    \n",
    "\n",
    "    def is_irreducible(self):\n",
    "        \"\"\"\n",
    "        Check if the Markov Chain is irreducible.\n",
    "        \"\"\"\n",
    "        for (i, j) in combinations(self.states,2):\n",
    "            if not self.is_accessible(i, j):\n",
    "                return False\n",
    "        return True\n",
    "    \n",
    "    def get_period(self, state):\n",
    "        \"\"\"\n",
    "        Returns the period of the state in the Markov Chain.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        state: str\n",
    "            The state for which the period needs to be computed.\n",
    "        \"\"\"\n",
    "        source_index = self.index_dict[i_state]\n",
    "        target_index = self.index_dict[f_state]\n",
    "        reachable_state_indexes = set([source_index])\n",
    "        \n",
    "        for state_index in reachable_state_indexes:\n",
    "            if state_index == target_index:\n",
    "                return True\n",
    "            else:\n",
    "                reachable_index_list = list(np.nonzero(\n",
    "                  self.transition_matrix[self.index_dict[i_state], :])[0])\n",
    "                reachable_state_indexes.union(set(reachable_index_list))                 \n",
    "        return False\n",
    "    \n",
    "        return gcd([len(i) for i in all_possible_paths])\n",
    "    \n",
    "    \n",
    "    def is_aperiodic(self):\n",
    "        \"\"\"\n",
    "        Checks if the Markov Chain is aperiodic. \n",
    "        \"\"\"\n",
    "        periods = [self.get_period(state) for state in self.states]\n",
    "        for period in periods:\n",
    "            if period != 1:\n",
    "                return False\n",
    "        return True\n",
    "\n",
    "    def is_transient(self, state):\n",
    "        \"\"\"\n",
    "        Checks if a state is transient or not.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        state: str\n",
    "            The state for which the transient property needs to be checked.\n",
    "        \"\"\"\n",
    "        if all(self.transition_matrix[~self.index_dict[state], self.index_dict[state]] == 0):\n",
    "            return True\n",
    "        else:\n",
    "            return False\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 166,
   "metadata": {},
   "outputs": [],
   "source": [
    "transition_periodic = [[0, 1, 0, 0, 0],\n",
    "                           [0, 0, 1, 0, 0],\n",
    "                           [0.5, 0, 0, 0.5, 0],\n",
    "                           [0, 0, 0, 0, 1],\n",
    "                           [0, 0, 1, 0, 0]]\n",
    "transition_aperiodic = [[0, 1, 0, 0, 0],\n",
    "                            [0, 0, 1, 0, 0],\n",
    "                            [0.5, 0.25, 0, 0.25, 0],\n",
    "                            [0, 0, 0, 0, 1],\n",
    "                            [0, 0, 0.5, 0.5, 0]]\n",
    "\n",
    "markov_periodic = MarkovChain(transition_matrix=transition_periodic,\n",
    "                                  states=['A', 'B', 'C', 'D', 'E'])\n",
    "markov_aperiodic = MarkovChain(transition_matrix=transition_aperiodic,\n",
    "                                   states=['A', 'B', 'C', 'D', 'E'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "markov_periodic.get_period('A')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}