-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathoptimizer.py
More file actions
123 lines (105 loc) · 6.06 KB
/
Copy pathoptimizer.py
File metadata and controls
123 lines (105 loc) · 6.06 KB
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
114
115
116
117
118
119
120
121
122
import warnings
warnings.simplefilter(action='ignore', category=FutureWarning)
import pandas as pd
from sklearn.metrics import mean_absolute_error, r2_score, matthews_corrcoef
import numpy as np
import optuna
import xgboost
import utils
import time
class Optimizer:
def __init__(self, oe, share_fake, share_real, classification, int_prediction, real_val, fake_train, real_train, dataset):
self.current_best_val_result = None
self.oe = oe
self.share_fake = share_fake
self.share_real = share_real
self.classification = classification
self.int_prediction = int_prediction
self.real_val = real_val
self.fake_train = fake_train
self.real_train = real_train
self.dataset = dataset
def run_optimization(self, real_Xy_test: pd.DataFrame = None, fake_trainval:pd.DataFrame = None):
study = utils.create_new_study("classification" if self.classification else "regression")
study.optimize(lambda trial: self.objective(trial=trial), n_trials=30)
final_model = xgboost.XGBClassifier(**study.best_params) if self.classification else xgboost.XGBRegressor(**study.best_params)
if self.oe is None:
fake_trainval_share = fake_trainval.sample(frac=self.share_fake)
real_train_val = pd.concat([self.real_train, self.real_val])
augmented_trainval = pd.concat([real_train_val, fake_trainval_share]).astype(float)
augmented_trainval = augmented_trainval.sample(frac=1)
else:
year_list = fake_trainval.index.year.unique().tolist()
searchfor = year_list[int(len(year_list) - (self.share_fake * 10)):]
fake_trainval_share = fake_trainval.loc[fake_trainval.index.year.isin(searchfor), :]
real_train_val = pd.concat([self.real_train, self.real_val])
augmented_trainval = pd.concat([real_train_val, fake_trainval_share])
augmented_trainval = augmented_trainval.sort_index()
augmented_trainval = augmented_trainval[real_Xy_test.columns]
start_process_time = time.process_time()
start_realclock_time = time.time()
if self.oe is not None:
final_model.fit(augmented_trainval.drop(["total_turnover"], axis=1), augmented_trainval.loc[:, ["total_turnover"]])
predictions = final_model.predict(real_Xy_test.drop(["total_turnover"], axis=1))
else:
final_model.fit(augmented_trainval.iloc[:,:-1], augmented_trainval.iloc[:,-1:])
predictions = np.rint(final_model.predict(real_Xy_test.iloc[:,:-1])) if self.int_prediction else final_model.predict(real_Xy_test.iloc[:,:-1])
process_time_s = time.process_time() - start_process_time
real_time_s = time.time() - start_realclock_time
if self.oe is not None:
return r2_score(real_Xy_test.loc[:, ["total_turnover"]], predictions), process_time_s, real_time_s
else:
return matthews_corrcoef(real_Xy_test.iloc[:,-1:], predictions) if self.classification else r2_score(real_Xy_test.iloc[:,-1:], predictions), process_time_s, real_time_s
def objective(self, trial: optuna.trial.Trial = None):
max_depth = trial.suggest_int("max_depth", 2, 10)
n_estimators = trial.suggest_int("n_estimators", 50, 1000, step=25)
gamma = trial.suggest_float("gamma", 0.001, 1, log=True)
reg_lambda = trial.suggest_float("reg_lambda", 0.1, 100, log=True)
reg_alpha = trial.suggest_float("reg_alpha", 0.1, 100, log=True)
learning_rate = trial.suggest_float("learning_rate", 0.025, 0.5, step=0.025)
subsample = trial.suggest_float("subsample", 0.005, 1.0, step=0.005)
colsample_bytree = trial.suggest_float("colsample_bytree", 0.1, 1.0, step=0.1)
# load the unfitted model to prevent information leak between folds
model = xgboost.XGBClassifier(
random_state=42,
verbosity=0,
tree_method="auto",
max_depth=max_depth,
n_estimators=n_estimators,
gamma=gamma,
reg_lambda=reg_lambda,
reg_alpha=reg_alpha,
learning_rate=learning_rate,
subsample=subsample,
colsample_bytree=colsample_bytree
) if self.classification else xgboost.XGBRegressor(
random_state=42,
verbosity=0,
tree_method="auto",
max_depth=max_depth,
n_estimators=n_estimators,
gamma=gamma,
reg_lambda=reg_lambda,
reg_alpha=reg_alpha,
learning_rate=learning_rate,
subsample=subsample,
colsample_bytree=colsample_bytree,
)
if hasattr(self.real_train, 'name'):
year_list = self.fake_train.index.year.unique().tolist()
searchfor = year_list[int(len(year_list)-(self.share_fake*10)):]
fake_train_share = self.fake_train.loc[self.fake_train.index.year.isin(searchfor), :]
augmented_train = pd.concat([self.real_train, fake_train_share])
augmented_train = augmented_train.sort_index()
augmented_train = augmented_train[self.real_val.columns]
else:
fake_train_share = self.fake_train.sample(frac=self.share_fake)
augmented_train = pd.concat([self.real_train, fake_train_share])
augmented_train = augmented_train.sample(frac=1)
model.fit(augmented_train.drop(["total_turnover"], axis=1), augmented_train.loc[:, ["total_turnover"]]) if hasattr(self.real_train, 'name') else model.fit(augmented_train.iloc[:,:-1], augmented_train.iloc[:,-1:])
if self.int_prediction:
prediction = np.rint(model.predict(self.real_val.iloc[:,:-1]))
else:
prediction = model.predict(self.real_val.drop(["total_turnover"], axis=1)) if hasattr(self.real_train, 'name') else model.predict(self.real_val.iloc[:,:-1])
current_val_result = matthews_corrcoef(y_true=self.real_val.iloc[:,-1:], y_pred=prediction) if self.classification else mean_absolute_error(y_true=self.real_val.iloc[:,-1:], y_pred=prediction)
return current_val_result