Skip to content

Instantly share code, notes, and snippets.

@pmgflores

pmgflores/TransferE.ipynb Secret

Created April 2, 2021 11:51
Show Gist options
  • Save pmgflores/a966774e918922ee69e01998c413e837 to your computer and use it in GitHub Desktop.
Save pmgflores/a966774e918922ee69e01998c413e837 to your computer and use it in GitHub Desktop.
Download ZIP
Display the source blob
Display the rendered blob
Raw
TransferE.ipynb
{
"cells": [
{
"cell_type": "markdown",
"metadata": {
"collapsed": false
},
"source": [
"### Transfer Entropy analysis code\n",
"\n",
"In the first part I upgrade dit package for Python. This package has defined the functions to compute time-delayed mutual information and transfer entropy using time series as inputs."
]
},
{
"cell_type": "code",
"execution_count": 75,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
"# Install the upgrade from git\n",
"#!pip install --user --upgrade git+git://github.com/dit/dit.git#egg=dit"
]
},
{
"cell_type": "markdown",
"metadata": {
"collapsed": false
},
"source": [
"**Step 1**\n",
"\n",
"To work with this code correctly we need to import all the packages needed in Python. It is important to notice that this package is designed for the use of discrete variables. dit stands for discrete information theory."
]
},
{
"cell_type": "code",
"execution_count": 76,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
"# Import the packages numpy, dit, and collections to run the code\n",
"import numpy as np\n",
"\n",
"from collections import namedtuple\n",
"\n",
"import dit\n",
"from dit.inference import binned, dist_from_timeseries\n",
"from dit.multivariate import total_correlation as I, intrinsic_total_correlation as IMI\n",
"\n",
"dit.ditParams['repr.print'] = True"
]
},
{
"cell_type": "markdown",
"metadata": {
"collapsed": false
},
"source": [
"**Step 2**\n",
"\n",
"Here we create a function to obtain the time-delayed mutual information and transfer entropy at the same time"
]
},
{
"cell_type": "code",
"execution_count": 77,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
"#Definition of the tuple to save the time-delayed mutual information and transfer entropy\n",
"TimeDelayed = namedtuple('TimeDelayed', ['timedelayedMU', 'transferE'])\n",
"\n",
"#Function to compute the time-delayed mutual information and transfer entropy\n",
"def get_timedelayed(ts, length):\n",
"\n",
" d = dist_from_timeseries(ts, history_length=length)\n",
"\n",
" i_past, j_past, i_pres, j_pres = [0], [1], [2], [3]\n",
"\n",
" timedelayedMU = I(d, [i_past, j_pres])\n",
" transferE = I(d, [i_past, j_pres], j_past)\n",
"\n",
" return TimeDelayed(timedelayedMU, transferE)"
]
},
{
"cell_type": "markdown",
"metadata": {
"collapsed": false
},
"source": [
"**Step 3**\n",
"\n",
"We create three different simulations to calculate the time-delayed mutual information and the transfer entropy showing that is possible to make these measures similar or different between them"
]
},
{
"cell_type": "code",
"execution_count": 78,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
"#Function for the first simulation. If we think about time series X and Y in this case the present of Y is always equal to the past of X\n",
"def get_exemplar_1(length):\n",
"\n",
" xs = np.random.randint(2, size=length)\n",
" ys = np.roll(xs, 1)\n",
" ts = np.vstack([xs, ys]).T\n",
" return ts"
]
},
{
"cell_type": "code",
"execution_count": 79,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"array([[1, 1],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 1],\n",
" [1, 1],\n",
" [0, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 1],\n",
" [0, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 1],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 1],\n",
" [1, 1],\n",
" [0, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0]])"
]
},
"execution_count": 79,
"metadata": {
},
"output_type": "execute_result"
}
],
"source": [
"#Print simulation 1\n",
"example1= get_exemplar_1(100)\n",
"example1"
]
},
{
"cell_type": "code",
"execution_count": 80,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"TimeDelayed(timedelayedMU=0.9833761901392236, transferE=0.983304093937157)"
]
},
"execution_count": 80,
"metadata": {
},
"output_type": "execute_result"
}
],
"source": [
"#Compute mutual information and transfer entropy for case 1. The result is very similar\n",
"case1= get_timedelayed(example1,1)\n",
"case1"
]
},
{
"cell_type": "code",
"execution_count": 81,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
"#Function for the second simulation. In this case the time series X and Y are always equal on time.\n",
"def get_exemplar_2(length, alphabet=2):\n",
"\n",
" xs = np.arange(length) % alphabet\n",
" ts = np.vstack([xs, xs]).T\n",
" return ts"
]
},
{
"cell_type": "code",
"execution_count": 82,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"array([[0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 1]])"
]
},
"execution_count": 82,
"metadata": {
},
"output_type": "execute_result"
}
],
"source": [
"#Print simulation 1\n",
"example2 = get_exemplar_2(100)\n",
"example2"
]
},
{
"cell_type": "code",
"execution_count": 83,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"TimeDelayed(timedelayedMU=0.999926399368686, transferE=0.0)"
]
},
"execution_count": 83,
"metadata": {
},
"output_type": "execute_result"
}
],
"source": [
"#Compute mutual information and transfer entropy for case 1. We can see that conditioning on the past make transfer entropy equals to zero.\n",
"case2= get_timedelayed(example2,1)\n",
"case2"
]
},
{
"cell_type": "code",
"execution_count": 84,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
"#Function for the third simulation. In this case the time series X and Y use the exclusive OR operator in the present to produce Y in the future.\n",
"def get_exemplar_3(length):\n",
"\n",
" xs = np.random.randint(2, size=length)\n",
" ys = [np.random.randint(2)]\n",
" for x in xs[:-1]:\n",
" ys.append(x ^ ys[-1])\n",
" ts = np.vstack([xs, ys]).T\n",
" return ts"
]
},
{
"cell_type": "code",
"execution_count": 85,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"array([[0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1],\n",
" [0, 0],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 0],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [1, 0],\n",
" [0, 1],\n",
" [0, 1],\n",
" [1, 1],\n",
" [0, 0],\n",
" [0, 0],\n",
" [1, 0],\n",
" [1, 1]])"
]
},
"execution_count": 85,
"metadata": {
},
"output_type": "execute_result"
}
],
"source": [
"#Print simulatioon 3\n",
"example3 = get_exemplar_3(100)\n",
"example3"
]
},
{
"cell_type": "code",
"execution_count": 86,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"TimeDelayed(timedelayedMU=0.03979308630427081, transferE=0.9376985634739845)"
]
},
"execution_count": 86,
"metadata": {
},
"output_type": "execute_result"
}
],
"source": [
"#Compute mutual information and transfer entropy for case 3. We can see that conditioning on the past make transfer entropy near to one while the time-delayed mutual information almost zero.\n",
"case3= get_timedelayed(syn,3)\n",
"case3"
]
},
{
"cell_type": "code",
"execution_count": 0,
"metadata": {
"collapsed": false
},
"outputs": [
],
"source": [
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3",
"resource_dir": "/srv/conda/envs/notebook/share/jupyter/kernels/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.6"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment