File size: 4,396 Bytes
34393ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | import os
import sys
import pandas as pd
import numpy as np
import PATH
import torch
import argparse
#
from models import reader
from models import train_val
from models.popen import Auto_popen
from models import max_activation_patch as MAP
#
from sklearn.linear_model import LassoCV, RidgeCV, ElasticNetCV, LogisticRegressionCV
import warnings
warnings.filterwarnings('ignore')
#
parser = argparse.ArgumentParser('the script to evlauate the effect of ')
parser.add_argument("-c", "--config", type=str, required=True, help='the model config file: xxx.ini')
parser.add_argument("-s", "--set", type=int, default=2, help='train - 0 ,val - 1, test - 2 ')
parser.add_argument("-p", "--n_max_act", type=int, default=500, help='the number of seq')
parser.add_argument("-k", "--kfold_cv", type=int, default=1, help='the repeat')
parser.add_argument("-t", "--task", type=str, default='regression', help='either regression or classification')
parser.add_argument("-d", "--device", type=str, default='cpu', help='the device to use to extract featmap, digit or cpu')
args = parser.parse_args()
config_path = args.config
save_path = config_path.replace(".ini", "_coef")
config = Auto_popen(config_path)
config.batch_size = 256
# config.kfold_cv = 'train_val'
all_task = config.cycle_set
task_channel_effect = {}
task_performance = {}
# path check
if os.path.exists(config_path) and not os.path.exists(save_path):
os.mkdir(save_path)
for task in all_task:
# .... format featmap as data ....
print(f"\n\nevaluating for task: {task}")
# re-instance the map for each task
map_task = MAP.Maximum_activation_patch(popen=config, which_layer=4,
n_patch=args.n_max_act,
kfold_index=args.kfold_cv,
device_string=args.device)
# extract feature map and rl decision chain
featmap = map_task.extract_feature_map(task=task, which_set=args.set)
cum_rl_trend = map_task.cumulative_rl_decision(task=task, which_set=args.set)
# truncate the featmap and rl trend according to sequence length
max_seq_len = map_task.df[config.seq_col].apply(len).max()
to_stay = max_seq_len // np.product(map_task.strides) +1
trunc_start = featmap.shape[2] - to_stay
featmap = featmap[:,:,trunc_start:]
cum_rl_trend = cum_rl_trend[:,trunc_start:]
# construct input for linear regression
n_sample,n_channel,n_posi = featmap.shape
X = featmap.reshape(n_sample,-1)
Y = map_task.Y_ls.flatten()
# .... regression ....
if args.task == 'regression':
L1 = LassoCV(alphas=np.linspace(2e-3, 0.1, 49))
L2 = RidgeCV(alphas=np.linspace(0.001, 0.101, 20))
elastic = ElasticNetCV(alphas=np.linspace(2e-3, 0.1, 49), n_jobs=10)
models = [L1, L2, elastic]
model_names = ['Lasso', 'Ridge', 'Elastic']
else:
LR = LogisticRegressionCV(n_jobs=10)
# L1 = LogisticRegressionCV(n_jobs=10,penalty='l1', solver='saga')
# elastic = LogisticRegressionCV(n_jobs=10,penalty='elasticnet', solver='saga', l1_ratios=np.linspace(0.0, 0.5, 10))
models = [LR]
model_names = ['Logistic']
for model,name in zip(models, model_names):
print(f"\nregressing {name}..")
model.fit(X,Y)
r2 = model.score(X,Y) # will be
sparsity = np.sum(model.coef_==0) / X.shape[1] *100
try:
alpha = model.alpha_
except:
alpha = 0.0
print(f"{name} with optimal alpha {alpha:.5f}, r2/acc {r2:.3f} , zero coeff {sparsity:.1f}%")
# save df
effect = model.coef_.reshape(n_channel,-1)
fullcoef_df = pd.DataFrame(effect, columns=[f"{name}_posi_"+str(trunc_start+i) for i in range(to_stay)])
fullcoef_df.to_csv( os.path.join(save_path , f"{task}_{name}_coef.csv"), index=False)
task_channel_effect[f"{task}_{name}"] = effect.mean(axis=1)
task_performance[f"{task}_{name}"] = [alpha, r2, sparsity, model.coef_.max(), model.coef_.min()]
all_effect = pd.DataFrame(task_channel_effect)
all_effect.to_csv(os.path.join(save_path, "all_task_mean_effect.csv"), index=False)
report_df = pd.DataFrame(task_performance)
report_df.index = ['optim_alpha','r2', 'zero_pctg', 'max_coef', 'min_coef']
report_df.to_csv(os.path.join(save_path, "regression_report.csv"), index=False)
|