### Helper Functions for Q4
# Importing libraries
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import skfda
from scipy.interpolate import splrep
from skfda.preprocessing.dim_reduction.feature_extraction import FPCA
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.metrics import (classification_report,
confusion_matrix)
from sklearn.model_selection import RepeatedStratifiedKFold
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.svm import SVC
# %% Part 1
class_names = ['Greece', 'Italy', 'Portugal', 'Spain']
# Produces the 70-knot B-spline coefficients per data point
def bsplinec(row):
x = np.array(row)
knots = np.linspace(0, row.size, 68)[1:-1]
tck = splrep(np.arange(row.size), x, k=3, t=knots)
return tck[1][:-4]
# Custom Transformer for `bsplinec`
class BSplineC(BaseEstimator, TransformerMixin):
def __init__(self):
pass
def fit(self, X, y=None):
return self
def transform(self, X):
return np.apply_along_axis(bsplinec, 1, X)
# SVM Classification with Cubic B-spline Feature Extraction
def SVM_bspline(X, y, clf=SVC(kernel='rbf', C=16, gamma=0.2)):
x_train, x_test, y_train, y_test =
train_test_split(X, y, test_size=30, shuffle=False)
scaler_pre = StandardScaler()
scaler_post = StandardScaler()
min_max_pre = MinMaxScaler()
min_max_post = MinMaxScaler()
pipe = Pipeline(steps=[('pre_scale', scaler_pre),
('pre_minmax', min_max_pre),
('bspline', BSplineC()),
('post_scale', scaler_post),
('post_minmax', min_max_post),
('clf', clf)])
pipe.fit(x_train, y_train)
y_pred = pipe.predict(x_test)
print("SVM Classification Report:\n" +
classification_report(y_test, y_pred,
target_names=class_names))
return y_pred, confusion_matrix(y_test, y_pred)
# %% Part 2
# Custom Transformer for to convert a dataframe to a skfda.FDataGrid
class DfToFdat(BaseEstimator, TransformerMixin):
def __init__(self):
pass
def fit(self, X, y=None):
return self
def transform(self, X):
X_ = pd.DataFrame(X)
return df_to_fdat(X_)
# Function to convert a dataframe to a FPCA object
def df_to_fdat(df):
grid_pnts = df.columns.tolist()
data_matrix = df.values.tolist()
return skfda.FDataGrid(data_matrix, grid_pnts)
# SVM Classification with fPCA Feature Extraction
def SVM_fPCA(x, y, n_components, c=4, gamma=0.62):
x_train, x_test, y_train, y_test =
train_test_split(x, y, test_size=30, shuffle=False)
scaler_pre = StandardScaler()
scaler_post = StandardScaler()
min_max_post = MinMaxScaler()
pipe = Pipeline(steps=[
('scaler_pre', scaler_pre),
('df_to_fdat', DfToFdat()),
('fPCA', FPCA(n_components=n_components)),
('scaler_post', scaler_post),
('min_max_post', min_max_post),
('SVC', SVC(kernel='rbf', C=c, gamma=gamma))])
pipe.fit(x_train, y_train)
y_test_pred = pipe.predict(x_test)
print("SVM Classification Report:\n" +
classification_report(y_test, y_test_pred,
target_names=class_names))
return y_test_pred, confusion_matrix(y_test, y_test_pred)
# %% CV Grid Search
# Cross-Validation for SVM Classification with fPCA Feature Extraction
def SVM_fPCA_CV(x, y, n_components, k=5, repeats=5, param_grid=None):
rskf = RepeatedStratifiedKFold(n_splits=k,
n_repeats=repeats,
random_state=42)
pipe = Pipeline(steps=[
('df_to_fdat', DfToFdat()),
('fPCA', FPCA(n_components=n_components)),
('SVC', SVC(kernel='rbf', C=16, gamma=0.2))])
scaler_pre = StandardScaler()
scaler_post = StandardScaler()
min_max_post = MinMaxScaler()
pipe = Pipeline(steps=[
('scaler_pre', scaler_pre),
('df_to_fdat', DfToFdat()),
('fPCA', FPCA(n_components=n_components)),
('scaler_post', scaler_post),
('min_max_post', min_max_post),
('SVC', SVC(kernel='rbf'))])
if param_grid is None:
param_grid = {
'SVC__C': [1, 2, 4, 8, 16, 32],
'SVC__gamma': [0.38, 0.44, 0.5, 0.56, 0.62, 0.68, 0.74]
}
search = GridSearchCV(pipe, param_grid, cv=rskf,
n_jobs=-1, verbose=1)
search.fit(x, y)
print("Best parameter (CV score=%0.3f):" % search.best_score_)
print(search.best_params_)
return search
# %% Confusion Matrix Plotting Function
def plt_cm(cmat):
plt.rc('font', size=15)
fig, ax = plt.subplots()
im = ax.imshow(cmat, cmap='RdPu')
ax.figure.colorbar(im, ax=ax)
ax.set(xticks=np.arange(cmat.shape[1]),
yticks=np.arange(cmat.shape[0]))
ax.set_xticklabels(class_names)
ax.set_yticklabels(class_names, rotation=45)
plt.setp(ax.get_xticklabels(), rotation=45, ha="right")
ax.set_xlabel('Predicted')
ax.set_ylabel('Actual')
for i in range(cmat.shape[0]):
for j in range(cmat.shape[1]):
_ = ax.text(j, i, cmat[i, j], ha="center", va="center",
color="w" if cmat[i, j] > cmat.max() / 2.1 else "k")
plt.title('Confusion Matrix')
fig.tight_layout()
plt.show()
return ax
