I have a pretty simple little model in a Pipeline which I want to weight the samples for and do cross-validation on:
from lightgbm import LGBMClassifier
from category_encoders import OneHotEncoder, OrdinalEncoder
from sklearn.pipeline import FeatureUnion, Pipeline
from sklearn.base import TransformerMixin
from sklearn.model_selection import cross_validate
from sklearn.model_selection import TimeSeriesSplit
import pandas as pd
categorical_columns = ['time_zone']
ordinal_columns = ['distance_miles']
date_column = ['date_of_entry']
class ColumnExtractor(TransformerMixin):
def __init__(self, features, cast_as=None):
self.features = features
self.cast_as = cast_as
def get_feature_names_out(self, features):
return self.features
def fit(self, *args, **kwargs):
return self
def transform(self, X, **transform_params):
return X[self.features].astype(self.cast_as)
def exp_decay_weighted_obs(obs, decay_rate=0.01):
total_rows = len(obs)
i = 1
weights = []
while (i <= total_rows):
curr_weight = pow((1 - decay_rate), (total_rows - i))
weights.append(curr_weight)
i = i + 1
return (weights)
#generate weights over all days in the range of our obs
def temporal_exp_decay_weighted_obs(dates, decay_rate=0.01):
first_date = dates[0]
last_date = dates[len(dates) - 1]
days_arr = pd.date_range(start=first_date, end=last_date, closed=None)
weights = exp_decay_weighted_obs(obs=days_arr, decay_rate=decay_rate)
return (dict(zip(days_arr, weights)))
def reshape_weights(rows, weight_index_field, weights):
w = []
for idx, row in rows.iterrows():
w.append(weights[row[weight_index_field].to_pydatetime()])
return w
#some fake data..our obs are one week apart, but could be any dates, sorted asc
x_data = {'time_zone':['ET', 'PT', 'PT', 'MT', 'ET', 'ET'],
'distance_miles':[200, 21, 1900, 18, 56, 100]}
y_train = [0, 1, 1, 0, 1, 0]
X_train = pd.DataFrame(x_data)
X_train['date_of_entry'] = pd.date_range('1/1/2022', periods = 6, freq ='W')
dates_x = X_train['date_of_entry'].unique()
#get weights for each day in our range, exponentially decayed from first date by DAY
weights = temporal_exp_decay_weighted_obs(dates_x, decay_rate=0.01)
weights_x = reshape_weights(X_train, 'date_of_entry', weights)
mlpl = Pipeline([
('features', FeatureUnion([
('categoricals', Pipeline([
('extract', ColumnExtractor(categorical_columns, str)),
('encoder', OneHotEncoder())
])),
('ordinals', Pipeline([
('extract', ColumnExtractor(ordinal_columns, int)),
('encoder', OrdinalEncoder())
]))
])),
('classifier', LGBMClassifier())
])
#we can use our weights in fit..like so
mlpl.fit(X=X_train, y=y_train, classifier__sample_weight=weights_x)
# but how might we calculate the weights again in a TimeseriesSplit?
#the values of the temporal exp decay rates would/should be different
#use fit_param?
scores = cross_validate(mlpl, X_train, y_train, cv=TimeSeriesSplit(n_splits=2), scoring='balanced_accuracy')
So, i have devised a sort of exponential decay based upon the date of the observation ("date of entry" --as you move further back in time, the observations are weighted less based upon the day and the decay rate.
I can calculate this beforehand and supply it to the fit() on the Pipeline with no issue at all-and this is done in the code snippet.
But I want to cross validate using a timeSeries split (to keep the order of the observations), but the weights would NOT be correct for each of the splits because I want them weighted to the date range of the split, not the date range of the entire data set. basically I need to be able to dynamically generate those weights per each call to fit().
How would I best do this? some fit_param magic on the cross-validate() call? some superclass on the LGBMClassifier which recalcs them on the data passed in (but how do I make those date obs NOT be a part of the X data?)? Thoughts?