<jupyter_start><jupyter_text>Mutual Information (MI) for Utility FactorImplemented Mutual Information as a metric for Utility Factor.<jupyter_code>import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import datetime
import seaborn as sns
import matplotlib.colors as colors
import klcalculator
from random import choices
import os

from math import sqrt
from functools import reduce

os.listdir()<jupyter_output><empty_output><jupyter_text>Load Data  Synthetic employee survey<jupyter_code>data = pd.read_csv('Synthetic_PMES_data.csv')
def lookup_vals(myval, mydict):
    try:
        return mydict[myval]
    except KeyError:
        return None
    
def create_sparse_dict(df):
    labels = df.value_counts().sort_values()
    lookup_dict = {}
    c = 1 # starting from 1 because we will 0-pad
    for i, v in labels.items():
        lookup_dict[i] = c
        c += 1
    df = pd.DataFrame(df.apply(lookup_vals, args=(lookup_dict,)))
    return df, lookup_dict

def date_to_int(dt):
    return 1000*(dt - datetime.datetime(1970,1,1)).total_seconds()
categorical_cols = [col for col in data.columns if 
                    (data[col].dtype != np.dtype('float64') and data[col].dtype != np.dtype('int64'))]
dicts = []
for col in categorical_cols:
    data[col], my_dict = create_sparse_dict(data[col])
    dicts.append(my_dict)<jupyter_output><empty_output><jupyter_text>Inmates<jupyter_code>data2 = pd.read_csv('Inmate_Admissions.csv')
data2 = data2.fillna(0)
data2.dtypes
data2.columns
data2.ADMITTED_DT = data2.ADMITTED_DT.astype('datetime64[ns]')
data2.DISCHARGED_DT = data2.DISCHARGED_DT.astype('datetime64[ns]')

data2.RACE, race_dict = create_sparse_dict(data2.RACE)
data2.GENDER, gender_dict = create_sparse_dict(data2.GENDER)
data2.INMATE_STATUS_CODE, status_dict = create_sparse_dict(data2.INMATE_STATUS_CODE)
data2.TOP_CHARGE, charge_dict = create_sparse_dict(data2.TOP_CHARGE)
data2.ADMITTED_DT = data2.ADMITTED_DT.apply(date_to_int)
data2.DISCHARGED_DT = data2.DISCHARGED_DT.apply(date_to_int)<jupyter_output><empty_output><jupyter_text>Multi-column generalization<jupyter_code>from collections import Counter
from itertools import combinations
import math
import pandas as pd
from typing import Mapping, Sequence, Any, Tuple, NewType, cast, Optional

distribution = NewType('distribution', Mapping[Any, float])


def get_sample_df() -> pd.DataFrame:
    """
    Define a sample dataframe that we can re-use in docstring tests.
    """
    columns = ['gender', 'age', 'eyes']
    data = [[0, 30, 1], [0, 33, 1], [1, 30, 2], [2, 30, 1], [2, 35, 3], [1, 35, 2], [1, 32, 2], [0, 30, 1]]
    return pd.DataFrame(data, columns=columns)

def get_rounded_df() -> pd.DataFrame:
    columns = ['gender', 'age', 'eyes']
    data = [[0, 30, 1], [0, 35, 1], [1, 30, 2], [1, 30, 1], [1, 35, 2], [1, 35, 2], [1, 30, 2], [0, 30, 1]]
    return pd.DataFrame(data, columns=columns)
    
def calculate_kl(posterior: distribution,
                 prior: distribution,
                 ) -> float:
    """Calculate D_KL(P || Q) (the KL-divergence) in bits.

    D_KL(P || Q) is the `information gained when one revises one's
    beliefs from the prior probability distribution Q to the posterior
    probability distribution P`. (Wikipedia, Kullback-Leibler divergence)

    `p` and `q` are both dictionaries mapping some hashable to a number.
    It is assumed that they are both normalised: their values should add
    up to 0. `q` must not have any 0 values unless the corresponding `p`
    value is also 0.

    >>> calculate_kl(posterior={0: 1, 1: 0}, prior={0: 0.5, 1: 0.5})
    1.0
    >>> calculate_kl(posterior={0: 0.25, 1: 0.75}, prior={0: 0.5, 1: 0.5})
    0.18872187554086717
    """
    return sum(pk * math.log2(pk / prior.get(k, 0))
               for k, pk in posterior.items()
               if pk > 0)

def calculate_distribution_from_values(values: Sequence) -> distribution:
    """
    >>> assert calculate_distribution_from_values([1, 2, 3, 2, 3, 2, 2, 3]) == {1: 0.125, 2: 0.5, 3: 0.375}
    >>> assert calculate_distribution_from_values([(1, 2), (2, 2), (3, 2), (3, 2), (2, 2)]) == \
    {(1, 2): 0.2, (3, 2): 0.4, (2, 2): 0.4}
    """
    counts = Counter(values)
    total_counts = sum(counts.values())
    return cast(distribution, {v: c / total_counts for v, c in counts.items()})

def as_sequence_of_tuples(df: pd.DataFrame) -> Sequence[Tuple]:
    """
    >>> as_sequence_of_tuples(get_sample_df())
    ((0, 30, 1), (0, 35, 1), (1, 30, 2), (1, 30, 1), (1, 35, 2), (1, 35, 2), (1, 30, 2), (0, 30, 1))
    """
    return tuple(tuple(row) for row in df.values)

def calculate_mutual_info(x: pd.Series,
                          y: pd.Series,
                         ) -> float:
    joint_df = pd.DataFrame({'x': x, 'y': y})
    pxy = calculate_distribution_from_values(as_sequence_of_tuples(joint_df))
    px = calculate_distribution_from_values(as_sequence_of_tuples(pd.DataFrame(x)))
    py = calculate_distribution_from_values(as_sequence_of_tuples(pd.DataFrame(y)))
    px_times_py = {(kx[0], ky[0]): x_value * y_value  # Quick and dirty
                   for kx, x_value in px.items()
                   for ky, y_value in py.items()}
    return calculate_kl(prior=px_times_py, posterior=pxy)
x = get_sample_df()['age']
y = get_sample_df()['eyes']
joint_df = pd.DataFrame({'x': x, 'y': y})
pxy = calculate_distribution_from_values(as_sequence_of_tuples(joint_df))
pxy
calculate_mutual_info(x, y)
px = calculate_distribution_from_values(as_sequence_of_tuples(pd.DataFrame(x)))
py = calculate_distribution_from_values(as_sequence_of_tuples(pd.DataFrame(y)))
{(kx[0], ky[0]): x_value * y_value
               for kx, x_value in px.items()
               for ky, y_value in py.items()}

calculate_distribution_from_values(as_sequence_of_tuples(pd.DataFrame(pd.DataFrame({'x': x, 'y': y}))))
def calculate_mutual_infos(original_data: pd.DataFrame,
                           modified_data: pd.DataFrame,
                           ) -> np.ndarray:
    """
    >>> calculate_mutual_infos(get_sample_df(), get_rounded_df())
    """
    n = len(original_data.columns)
    assert len(modified_data.columns) == n
    matMI = np.zeros((n, n))
    for ix, ci in enumerate(original_data):
        for jx, cj in enumerate(modified_data):
            matMI[ix, jx] = calculate_mutual_info(original_data[ci], modified_data[cj])
    return matMI
def round_column(df: pd.DataFrame, 
                 col_name: str,
                 base: int ) -> pd.DataFrame:

    def myround(x, base=5):
        try:
            return base * round(x/base)
        except ValueError:
            return np.nan

    rounded_col = [myround(x, base) for x in df[col_name]]
    rounded_df = df.copy()
    rounded_df[col_name] = rounded_col
    
    return rounded_df

def round_column_quartile(df: pd.DataFrame, 
                          col_name: str,
                          buckets: int ) -> pd.DataFrame:

    rounded_col = pd.qcut(df[col_name], buckets, labels=False, duplicates='drop')
    rounded_df = df.copy()
    rounded_df[col_name] = rounded_col
    
    return rounded_df
    
workforceprofile = data[['ATSI_Status', 'Age_Group', 'Current_Role_Years_Employed',
       'Disability_Status', 'Education', 'Employment_Status', 'Gender',
       'Gross_Salary', 'LGBTI_Status']]
workforceprofile = workforceprofile[:1000].fillna(0)
workforceprofile.describe()
workforceprofile.Gross_Salary.unique()
adjustedworkforceprofile = round_column(workforceprofile, 'Age_Group', 5)
adjustedworkforceprofile = round_column(adjustedworkforceprofile, 'Current_Role_Years_Employed', 2)
adjustedworkforceprofile = round_column(adjustedworkforceprofile, 'Gross_Salary', 2)<jupyter_output><empty_output><jupyter_text>Apply to a real dataset<jupyter_code>#%pdb
a_prime = calculate_mutual_infos(workforceprofile, adjustedworkforceprofile)
a = calculate_mutual_infos(workforceprofile, workforceprofile)
def display_risks(header, risks, rows=20, minr=0, maxr=10):
    """Return an object that prints the attribute-wise risks in Jupyter.

    We colour-code the risks for nice visuals.

    `header` is a list that specifies the table headers.

    `risks` is a list of lists of floats.

    We print at most `rows` (default 20) rows.

    `minr` (default 0) and `maxr` (default 10) specify the cutoff points
    for our colour coding. Anything above `maxr` will be coloured deep
    red for very risky. `minr` is similar.

    The code below is magic and I don't remember how it works.
    """
    df = pd.DataFrame(risks, columns=header)[:rows]
    cm = colors.ListedColormap(
        sns.color_palette("RdYlGn", 256).as_hex()[::-1])
    def risk_to_color(risk):
        color_i = round((min(risk, maxr) - minr) * 255 / (maxr - minr))
        color = cm.colors[color_i]
        return f'background-color: {color}'
    df.index = header
    s = df.style.applymap(risk_to_color)
    return s

display_risks(workforceprofile.columns, a, maxr=1)
display_risks(workforceprofile.columns, a_prime, maxr=1)
info_loss = a-a_prime
display_risks(workforceprofile.columns, info_loss, maxr=1)
my_risks = klcalculator.find_risks_for_records(workforceprofile.values)
adjusted_risks = klcalculator.find_risks_for_records(adjustedworkforceprofile.values)
klcalculator.display_risks(workforceprofile.columns, my_risks)
before = klcalculator.percentile(klcalculator.find_individual_risks(my_risks), 95)
after = klcalculator.percentile(klcalculator.find_individual_risks(adjusted_risks), 95)
print(f'before: {before}')
print(f'after: {after}')
def factors(n):
    step = 2 if n%2 else 1
    f = list(set(reduce(list.__add__,
                ([i, n//i] for i in range(1, int(sqrt(n))+1, step) if n % i == 0))))
    if len(f) == 2:
        f.append(int(n/2))
        f.sort()
    return f

adjustment_targets = ['Age_Group','Current_Role_Years_Employed', 'Gross_Salary']
adjustment_ranges = [len(workforceprofile[x].unique()) for x in adjustment_targets]
adjusted_dataframes = []

for idx, adj in enumerate(adjustment_targets):
    for block_count in factors(adjustment_ranges[idx]):
        adjusted_dataframes.append(round_column(workforceprofile, adj, block_count))
MMIs = []
PIFs = []
MIs = []
base_info = calculate_mutual_infos(workforceprofile, workforceprofile)

for adj in adjusted_dataframes:
    PIFs.append(klcalculator.percentile(klcalculator.find_individual_risks(klcalculator.find_risks_for_records(adj.values)), 95))
    adjusted_info = calculate_mutual_infos(workforceprofile, adj)
    info_loss = base_info - adjusted_info
    MIs.append(max([info_loss[x,x] for x in range(info_loss.shape[1])]))
    col_tots = max([np.sum(info_loss[:,x]) for x in range(info_loss.shape[1])])
    MMIs.append(col_tots)
age_adjustment = list(zip(PIFs[:6], MMIs[:6]))
years_adjustment = list(zip(PIFs[6:9], MMIs[6:9]))
sal_adjustment = list(zip(PIFs[9:], MMIs[9:]))
plt.plot(MMIs[:6], PIFs[:6], '-o')
plt.plot(MMIs[6:9], PIFs[6:9], '-o')
plt.plot(MMIs[9:], PIFs[9:], '-o')
plt.xlabel('MMI Loss')
plt.ylabel('PIF')
plt.legend(['age', 'years in job', 'salary'], ncol=2, loc='lower left');
plt.plot(MIs[:6], PIFs[:6], '-o')
plt.plot(MIs[6:9], PIFs[6:9], '-o')
plt.plot(MIs[9:], PIFs[9:], '-o')
plt.xlabel('MI Loss')
plt.ylabel('PIF')
plt.legend(['age', 'years in job', 'salary'], ncol=2, loc='lower left');
inmate_dataset = data2[['ADMITTED_DT', 'DISCHARGED_DT', 'RACE',
       'GENDER', 'INMATE_STATUS_CODE', 'TOP_CHARGE']][:1000]
inmate_dataset.shape
inmate_adjustment_targets = ['ADMITTED_DT','DISCHARGED_DT']
inmate_adjusted_dataframes = []

for idx, adj in enumerate(inmate_adjustment_targets):
    for block_count in range(1,20):
        inmate_adjusted_dataframes.append(round_column_quartile(inmate_dataset, adj, block_count))
IMMIs = []
IPIFs = []
IMIs = []
Ibase_info = calculate_mutual_infos(inmate_dataset, inmate_dataset)

for adj in inmate_adjusted_dataframes:
    IPIFs.append(klcalculator.percentile(klcalculator.find_individual_risks(klcalculator.find_risks_for_records(adj.values)), 95))
    adjusted_info = calculate_mutual_infos(inmate_dataset, adj)
    info_loss = Ibase_info - adjusted_info
    IMIs.append(max([info_loss[x,x] for x in range(info_loss.shape[1])]))
    col_tots = max([np.sum(info_loss[:,x]) for x in range(info_loss.shape[1])])
    IMMIs.append(col_tots)
len(IMMIs)
plt.plot(IMIs[:19], IPIFs[:19], '-o')
plt.plot(IMIs[19:], IPIFs[19:], '-o')
plt.xlabel('MI Loss')
plt.ylabel('PIF')
plt.legend(['admitdate', 'dischdate'], ncol=2, loc='lower left');
b = calculate_mutual_infos(inmate_dataset, inmate_dataset)
adjusted_inmate = round_column_quartile(inmate_dataset, 'ADMITTED_DT', 10)
adjusted_inmate = round_column_quartile(adjusted_inmate, 'DISCHARGED_DT', 10)
b_prime = calculate_mutual_infos(inmate_dataset, adjusted_inmate)
display_risks(inmate_dataset.columns, b, maxr=1)
display_risks(adjusted_inmate.columns, b_prime, maxr=1)
Iinfo_loss = b-b_prime
display_risks(adjusted_inmate.columns, Iinfo_loss, maxr=1)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13601976223560291, 'fraction_numerical': 0.025937934228809634, 'mean_word_length': 4.364389233954451, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '381032', 'license': 'MIT', 'n_tokens_mistral': 4996, 'n_tokens_neox': 4629, 'n_words': 1015, 'path': 'Scripts/Round 3/Good FIghters/generalized_MI.ipynb', 'repo_name': 'marcportlock/Directed_Ideation_2019', 'size': '12954'}