python/methylnet/model_interpretability.py
def produce_shapley_data(train_pkl, val_pkl, test_pkl, model_pickle, n_workers, batch_size, cuda, n_samples, n_top_features, output_dir, method, shap_sample_batch_size, n_random_representative, interest_col, n_random_representative_test, categorical_encoder, cross_class, feature_selection, top_outputs, vae_interpret, pred_class, results_csv, individual, residual_cutoff, cell_names, dump_shap_data):
    """Explanations (coefficients for CpGs) for every individual prediction.
    Produce SHAPley scores for each CpG for each individualized prediction, and then aggregate them across coarser classes.
    Store CpGs with top SHAPley scores as well for quick access.
    Store in Shapley data object."""
    os.makedirs(output_dir,exist_ok=True)
    if not pred_class:
        pred_class = None
    if not top_outputs or not feature_selection:
        top_outputs = None
    if os.path.exists(categorical_encoder):
        categorical_encoder=pickle.load(open(categorical_encoder,'rb'))
        if categorical_encoder:
            prediction_classes=list(categorical_encoder.categories_[0])
        else:
            prediction_classes = None
    else:
        prediction_classes = None
    cell_names=list(filter(None,cell_names))
    train_methyl_array, val_methyl_array, test_methyl_array=MethylationArray.from_pickle(train_pkl), MethylationArray.from_pickle(val_pkl), MethylationArray.from_pickle(test_pkl)#preprocessed_methyl_array.subset_index(train_test_idx_dict['train']), preprocessed_methyl_array.subset_index(train_test_idx_dict['test'])
    #preprocessed_methyl_array=MethylationArray(*extract_pheno_beta_df_from_pickle_dict(input_dict))
    train_methyl_array = MethylationArrays([train_methyl_array,val_methyl_array]).combine()
    cpgs, train_samples= train_methyl_array.return_cpgs(), train_methyl_array.return_idx()
    cpgs, test_samples= test_methyl_array.return_cpgs(), test_methyl_array.return_idx()
    model = torch.load(model_pickle)
    if os.path.exists(results_csv):
        results_df=pd.read_csv(results_csv)
        if residual_cutoff:
            replace_dict={k:k.replace('_pred','') for k in list(results_df)}
            pred_df = results_df.iloc[:,np.vectorize(lambda x: x.endswith('_pred'))(list(results_df))].rename(replace_dict)
            true_df = results_df.iloc[:,np.vectorize(lambda x: x.endswith('_true'))(list(results_df))].rename(replace_dict)
            residual_df = pred_df-true_df
            cell_names=list(residual_df)
            for col in residual_df:
                residual_df.loc[:,col] = (residual_df[col] - residual_df[col].mean())/residual_df[col].std(ddof=0)
            residual_df = (results_df <= residual_cutoff)
            residual_df = residual_df.all(axis=1)
            test_methyl_array=test_methyl_array.subset_index(np.array(list(residual_df.index)))
        else:
            test_methyl_array=test_methyl_array.subset_index(np.array(list((results_df['y_pred']==results_df['y_true']).index)))
    if individual:
        test_methyl_array=test_methyl_array.subset_index(np.array([individual]))
    if n_random_representative and method != 'gradient':
        train_methyl_array = train_methyl_array.subsample(interest_col, n_samples=n_random_representative, categorical=model.categorical if 'categorical' in dir(model) else False)
    if n_random_representative_test:
        test_methyl_array = test_methyl_array.subsample(interest_col, n_samples=n_random_representative_test, categorical=model.categorical if 'categorical' in dir(model) else False)
    if 'categorical' in dir(model) and model.categorical:
        print("TRAIN:")
        train_methyl_array.categorical_breakdown(interest_col)
        print("TEST:")
        test_methyl_array.categorical_breakdown(interest_col)
    model.eval()
    if cuda:
        model = model.cuda()
    if vae_interpret:
        model.forward = model.mlp
    elif method == 'deep' or method == 'gradient':
            model.forward = model.forward_predict
    t_arr = train_methyl_array.return_raw_beta_array()
    prediction_function=main_prediction_function(n_workers,min(batch_size,t_arr.shape[0]), model, cuda) if method == 'kernel' else model
    # in order for shap to work, prediction_function must work with methylation array beta values
    n_test_results_outputs=main_prediction_function(n_workers, 1, model, cuda)(t_arr[:2, ...]).shape[1]
    # if above does not work, shap will not work
    cpg_explainer = CpGExplainer(prediction_function, cuda)
    cpg_explainer.build_explainer(train_methyl_array, method, batch_size=batch_size)
    if not shap_sample_batch_size:
        shap_sample_batch_size = None
    cpg_explainer.return_shapley_scores(test_methyl_array, n_samples, n_outputs=n_test_results_outputs, shap_sample_batch_size=shap_sample_batch_size, top_outputs=top_outputs)
    if dump_shap_data:
        np.save(file='raw_shap_data.npy',arr=cpg_explainer.shap_values)
    if not residual_cutoff:
        cpg_explainer.classifier_assign_scores_to_shap_data(test_methyl_array, n_top_features, interest_col=interest_col, prediction_classes=prediction_classes)
    else:
        cpg_explainer.regressor_assign_scores_to_shap_data(test_methyl_array, n_top_features, cell_names=cell_names)
    if feature_selection:
        print('FEATURE SELECT')
        feature_selected_methyl_array=cpg_explainer.feature_select(MethylationArrays([train_methyl_array,test_methyl_array]).combine(),n_top_features)
        feature_selected_methyl_array.write_pickle(join(output_dir, 'feature_selected_methyl_array.pkl'))
    else:
        output_all_cpgs=join(output_dir,'all_cpgs.p')
        shapley_output=join(output_dir,'shapley_data.p')
        pickle.dump(train_methyl_array.return_cpgs(),open(output_all_cpgs,'wb'))
        cpg_explainer.shapley_data.to_pickle(shapley_output)
Explanations (coefficients for CpGs) for every individual prediction.
    Produce SHAPley scores for each CpG for each individualized prediction, and then aggregate them across coarser classes.
    Store CpGs with top SHAPley scores as well for quick access.
    Store in Shapley data object.
Explanations (coefficients for CpGs) for every individual prediction.
Produce SHAPley scores for each CpG for each individualized prediction, and then aggregate them across coarser classes.
Store CpGs with top SHAPley scores as well for quick access.
Store in Shapley data object.
Explanations ( coefficients for CpGs ) for every individual prediction . Produce SHAPley scores for each CpG for each individualized prediction and then aggregate them across coarser classes . Store CpGs with top SHAPley scores as well for quick access . Store in Shapley data object .
def produce_shapley_data(train_pkl, val_pkl, test_pkl, model_pickle, n_workers, batch_size, cuda, n_samples, n_top_features, output_dir, method, shap_sample_batch_size, n_random_representative, interest_col, n_random_representative_test, categorical_encoder, cross_class, feature_selection, top_outputs, vae_interpret, pred_class, results_csv, individual, residual_cutoff, cell_names, dump_shap_data):
    os.makedirs(output_dir,exist_ok=True)
    if not pred_class:
        pred_class = None
    if not top_outputs or not feature_selection:
        top_outputs = None
    if os.path.exists(categorical_encoder):
        categorical_encoder=pickle.load(open(categorical_encoder,'rb'))
        if categorical_encoder:
            prediction_classes=list(categorical_encoder.categories_[0])
        else:
            prediction_classes = None
    else:
        prediction_classes = None
    cell_names=list(filter(None,cell_names))
    train_methyl_array, val_methyl_array, test_methyl_array=MethylationArray.from_pickle(train_pkl), MethylationArray.from_pickle(val_pkl), MethylationArray.from_pickle(test_pkl)
    train_methyl_array = MethylationArrays([train_methyl_array,val_methyl_array]).combine()
    cpgs, train_samples= train_methyl_array.return_cpgs(), train_methyl_array.return_idx()
    cpgs, test_samples= test_methyl_array.return_cpgs(), test_methyl_array.return_idx()
    model = torch.load(model_pickle)
    if os.path.exists(results_csv):
        results_df=pd.read_csv(results_csv)
        if residual_cutoff:
            replace_dict={k:k.replace('_pred','') for k in list(results_df)}
            pred_df = results_df.iloc[:,np.vectorize(lambda x: x.endswith('_pred'))(list(results_df))].rename(replace_dict)
            true_df = results_df.iloc[:,np.vectorize(lambda x: x.endswith('_true'))(list(results_df))].rename(replace_dict)
            residual_df = pred_df-true_df
            cell_names=list(residual_df)
            for col in residual_df:
                residual_df.loc[:,col] = (residual_df[col] - residual_df[col].mean())/residual_df[col].std(ddof=0)
            residual_df = (results_df <= residual_cutoff)
            residual_df = residual_df.all(axis=1)
            test_methyl_array=test_methyl_array.subset_index(np.array(list(residual_df.index)))
        else:
            test_methyl_array=test_methyl_array.subset_index(np.array(list((results_df['y_pred']==results_df['y_true']).index)))
    if individual:
        test_methyl_array=test_methyl_array.subset_index(np.array([individual]))
    if n_random_representative and method != 'gradient':
        train_methyl_array = train_methyl_array.subsample(interest_col, n_samples=n_random_representative, categorical=model.categorical if 'categorical' in dir(model) else False)
    if n_random_representative_test:
        test_methyl_array = test_methyl_array.subsample(interest_col, n_samples=n_random_representative_test, categorical=model.categorical if 'categorical' in dir(model) else False)
    if 'categorical' in dir(model) and model.categorical:
        print("TRAIN:")
        train_methyl_array.categorical_breakdown(interest_col)
        print("TEST:")
        test_methyl_array.categorical_breakdown(interest_col)
    model.eval()
    if cuda:
        model = model.cuda()
    if vae_interpret:
        model.forward = model.mlp
    elif method == 'deep' or method == 'gradient':
            model.forward = model.forward_predict
    t_arr = train_methyl_array.return_raw_beta_array()
    prediction_function=main_prediction_function(n_workers,min(batch_size,t_arr.shape[0]), model, cuda) if method == 'kernel' else model
    n_test_results_outputs=main_prediction_function(n_workers, 1, model, cuda)(t_arr[:2, ...]).shape[1]
    cpg_explainer = CpGExplainer(prediction_function, cuda)
    cpg_explainer.build_explainer(train_methyl_array, method, batch_size=batch_size)
    if not shap_sample_batch_size:
        shap_sample_batch_size = None
    cpg_explainer.return_shapley_scores(test_methyl_array, n_samples, n_outputs=n_test_results_outputs, shap_sample_batch_size=shap_sample_batch_size, top_outputs=top_outputs)
    if dump_shap_data:
        np.save(file='raw_shap_data.npy',arr=cpg_explainer.shap_values)
    if not residual_cutoff:
        cpg_explainer.classifier_assign_scores_to_shap_data(test_methyl_array, n_top_features, interest_col=interest_col, prediction_classes=prediction_classes)
    else:
        cpg_explainer.regressor_assign_scores_to_shap_data(test_methyl_array, n_top_features, cell_names=cell_names)
    if feature_selection:
        print('FEATURE SELECT')
        feature_selected_methyl_array=cpg_explainer.feature_select(MethylationArrays([train_methyl_array,test_methyl_array]).combine(),n_top_features)
        feature_selected_methyl_array.write_pickle(join(output_dir, 'feature_selected_methyl_array.pkl'))
    else:
        output_all_cpgs=join(output_dir,'all_cpgs.p')
        shapley_output=join(output_dir,'shapley_data.p')
        pickle.dump(train_methyl_array.return_cpgs(),open(output_all_cpgs,'wb'))
        cpg_explainer.shapley_data.to_pickle(shapley_output)
def produce_shapley_data ( train_pkl , val_pkl , test_pkl , model_pickle , n_workers , batch_size , cuda , n_samples , n_top_features , output_dir , method , shap_sample_batch_size , n_random_representative , interest_col , n_random_representative_test , categorical_encoder , cross_class , feature_selection , top_outputs , vae_interpret , pred_class , results_csv , individual , residual_cutoff , cell_names , dump_shap_data ) : os . makedirs ( output_dir , exist_ok = True ) if not pred_class : pred_class = None if not top_outputs or not feature_selection : top_outputs = None if os . path . exists ( categorical_encoder ) : categorical_encoder = pickle . load ( open ( categorical_encoder , 'rb' ) ) if categorical_encoder : prediction_classes = list ( categorical_encoder . categories_ [ 0 ] ) else : prediction_classes = None else : prediction_classes = None cell_names = list ( filter ( None , cell_names ) ) train_methyl_array , val_methyl_array , test_methyl_array = MethylationArray . from_pickle ( train_pkl ) , MethylationArray . from_pickle ( val_pkl ) , MethylationArray . from_pickle ( test_pkl ) train_methyl_array = MethylationArrays ( [ train_methyl_array , val_methyl_array ] ) . combine ( ) cpgs , train_samples = train_methyl_array . return_cpgs ( ) , train_methyl_array . return_idx ( ) cpgs , test_samples = test_methyl_array . return_cpgs ( ) , test_methyl_array . return_idx ( ) model = torch . load ( model_pickle ) if os . path . exists ( results_csv ) : results_df = pd . read_csv ( results_csv ) if residual_cutoff : replace_dict = { k : k . replace ( '_pred' , '' ) for k in list ( results_df ) } pred_df = results_df . iloc [ : , np . vectorize ( lambda x : x . endswith ( '_pred' ) ) ( list ( results_df ) ) ] . rename ( replace_dict ) true_df = results_df . iloc [ : , np . vectorize ( lambda x : x . endswith ( '_true' ) ) ( list ( results_df ) ) ] . rename ( replace_dict ) residual_df = pred_df - true_df cell_names = list ( residual_df ) for col in residual_df : residual_df . loc [ : , col ] = ( residual_df [ col ] - residual_df [ col ] . mean ( ) ) / residual_df [ col ] . std ( ddof = 0 ) residual_df = ( results_df <= residual_cutoff ) residual_df = residual_df . all ( axis = 1 ) test_methyl_array = test_methyl_array . subset_index ( np . array ( list ( residual_df . index ) ) ) else : test_methyl_array = test_methyl_array . subset_index ( np . array ( list ( ( results_df [ 'y_pred' ] == results_df [ 'y_true' ] ) . index ) ) ) if individual : test_methyl_array = test_methyl_array . subset_index ( np . array ( [ individual ] ) ) if n_random_representative and method != 'gradient' : train_methyl_array = train_methyl_array . subsample ( interest_col , n_samples = n_random_representative , categorical = model . categorical if 'categorical' in dir ( model ) else False ) if n_random_representative_test : test_methyl_array = test_methyl_array . subsample ( interest_col , n_samples = n_random_representative_test , categorical = model . categorical if 'categorical' in dir ( model ) else False ) if 'categorical' in dir ( model ) and model . categorical : print ( "TRAIN:" ) train_methyl_array . categorical_breakdown ( interest_col ) print ( "TEST:" ) test_methyl_array . categorical_breakdown ( interest_col ) model . eval ( ) if cuda : model = model . cuda ( ) if vae_interpret : model . forward = model . mlp elif method == 'deep' or method == 'gradient' : model . forward = model . forward_predict t_arr = train_methyl_array . return_raw_beta_array ( ) prediction_function = main_prediction_function ( n_workers , min ( batch_size , t_arr . shape [ 0 ] ) , model , cuda ) if method == 'kernel' else model n_test_results_outputs = main_prediction_function ( n_workers , 1 , model , cuda ) ( t_arr [ : 2 , ... ] ) . shape [ 1 ] cpg_explainer = CpGExplainer ( prediction_function , cuda ) cpg_explainer . build_explainer ( train_methyl_array , method , batch_size = batch_size ) if not shap_sample_batch_size : shap_sample_batch_size = None cpg_explainer . return_shapley_scores ( test_methyl_array , n_samples , n_outputs = n_test_results_outputs , shap_sample_batch_size = shap_sample_batch_size , top_outputs = top_outputs ) if dump_shap_data : np . save ( file = 'raw_shap_data.npy' , arr = cpg_explainer . shap_values ) if not residual_cutoff : cpg_explainer . classifier_assign_scores_to_shap_data ( test_methyl_array , n_top_features , interest_col = interest_col , prediction_classes = prediction_classes ) else : cpg_explainer . regressor_assign_scores_to_shap_data ( test_methyl_array , n_top_features , cell_names = cell_names ) if feature_selection : print ( 'FEATURE SELECT' ) feature_selected_methyl_array = cpg_explainer . feature_select ( MethylationArrays ( [ train_methyl_array , test_methyl_array ] ) . combine ( ) , n_top_features ) feature_selected_methyl_array . write_pickle ( join ( output_dir , 'feature_selected_methyl_array.pkl' ) ) else : output_all_cpgs = join ( output_dir , 'all_cpgs.p' ) shapley_output = join ( output_dir , 'shapley_data.p' ) pickle . dump ( train_methyl_array . return_cpgs ( ) , open ( output_all_cpgs , 'wb' ) ) cpg_explainer . shapley_data . to_pickle ( shapley_output )
Explanations (coefficients for CpGs) for every individual prediction.
Explanations ( coefficients for CpGs ) for every individual prediction .
"""Explanations (coefficients for CpGs) for every individual prediction.
    Produce SHAPley scores for each CpG for each individualized prediction, and then aggregate them across coarser classes.
    Store CpGs with top SHAPley scores as well for quick access.
    Store in Shapley data object.""" #preprocessed_methyl_array.subset_index(train_test_idx_dict['train']), preprocessed_methyl_array.subset_index(train_test_idx_dict['test']) #preprocessed_methyl_array=MethylationArray(*extract_pheno_beta_df_from_pickle_dict(input_dict)) # in order for shap to work, prediction_function must work with methylation array beta values # if above does not work, shap will not work
train_pkl
val_pkl
test_pkl
model_pickle
n_workers
batch_size
cuda
n_samples
n_top_features
output_dir
method
shap_sample_batch_size
n_random_representative
interest_col
n_random_representative_test
categorical_encoder
cross_class
feature_selection
top_outputs
vae_interpret
pred_class
results_csv
individual
residual_cutoff
cell_names
dump_shap_data