# Import potrebných balíčkov
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler, KBinsDiscretizer
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, mean_absolute_error
from keras.layers import Dense, Input, Activation
from keras.callbacks import Callback
from keras.models import Model
import matplotlib.pyplot as plt
import seaborn as sns
from scipy.optimize import curve_fit
# Uistíme sa, že máme všetky potrebné dáta
!mkdir -p data
!wget -nc -O data/sigmoid_regression_data.csv https://www.dropbox.com/s/p5q7gzupa2ndw55/sigmoid_regression_data.csv?dl=1
# Pomocný kód
class NEpochLogger(Callback):
"""
Trieda na menej časté zobrazovanie priebehu učenia.
"""
def __init__(self, n_epochs=100):
super(NEpochLogger, self).__init__()
self.n_epochs = n_epochs
def on_epoch_end(self, epoch, logs=None):
logs = logs or {}
if epoch % self.n_epochs == 0:
curr_loss = logs.get('loss')
print("epoch = {}; loss = {}".format(
epoch, curr_loss))
V tomto notebooku budeme aplikovať jednoduchú neurónovú sieť zostavenú pomocou balíčka keras
na regresnú úlohu. Kód bude do veľkej miery podobný kódu pre klasifikáciu. Rozdiely budú hlavne v aktivačnej funkcii poslednej vrstvy siete (bude lineárna) a v chybovej funkcii, ktorú budeme minimalizovať (stredná kvadratická chyba).
Dátová množina sa bude skladať z množiny bodov v 2D priestore – načítame ju z CSV súboru:
df = pd.read_csv("data/sigmoid_regression_data.csv")
df.head()
Dáta rozdelíme na tréningové a testovacie. Stratifikácia sa v prípade spojitých dát nedá aplikovať priamo. Ak ju chceme predsa len použiť, musíme stĺpec, podľa ktorého stratifikujeme, najprv diskretizovať: rozdeliť vzorky do určitého konečného počtu skupín (napr. pomocou KBinsDiscretizer
).
Stratifikáciu v tomto prípade nepoužijeme – nedá sa priamo aplikovať, ak príslušný stĺpec obsahuje spojité dáta. V prípade, že chceme stratifikáciu pri regresii použiť, je to stále možné, ale stĺpec, podľa ktorého stratifikujeme, treba
kbins = KBinsDiscretizer(5, encode='ordinal')
y_stratify = kbins.fit_transform(df[['y']])
df_train, df_test = train_test_split(df, stratify=y_stratify,
test_size=0.3, random_state=4)
Súradnice x
jednotlivých bodov použijeme ako vstupy a súradnice y
ako požadované výstupy:
inputs = ["x"]
output = ["y"]
X_train_raw = df_train[inputs]
X_test_raw = df_test[inputs]
Y_train_raw = df_train[output]
Y_test_raw = df_test[output]
Aby sme získali lepšiu predstavu o tom, ako dáta vyzerajú, môžeme si body vykresliť.
plt.scatter(X_train_raw, Y_train_raw, label="tréningové dáta")
plt.scatter(X_test_raw, Y_test_raw, c='r', label="testovacie dáta")
plt.xlabel('x')
plt.ylabel('y')
plt.grid(ls='--')
plt.legend()
Vstupné a výstupné dáta si môžeme preškálovať do štandardného rozsahu; v tomto jednoduchom prípade by však regresia mala fungovať správne aj bez toho.
input_preproc = StandardScaler()
X_train = input_preproc.fit_transform(X_train_raw)
X_test = input_preproc.transform(X_test_raw)
output_preproc = StandardScaler()
Y_train = output_preproc.fit_transform(Y_train_raw)
Y_test = output_preproc.transform(Y_test_raw)
Určíme počet vstupov a výstupov siete (podľa tvaru dátovej množiny):
num_inputs = X_train.shape[1]
num_outputs = Y_train.shape[1]
Vstupnú vrstvu vytvoríme obdobným spôsobom ako pri klasifikácii.
y = x = Input(shape=(num_inputs, ))
Následne pridáme ďalšie vrstvy. Od použitých aktivačných funkcií bude čiastočne závisieť tvar výslednej regresnej krivky. Pri použití funkcií ReLU krivka napríklad nebude celkom hladká.
y = Dense(10)(y)
y = Activation('relu')(y)
y = Dense(10)(y)
y = Activation('relu')(y)
Výstupná vrstva bude mať lineárnu aktivačnú funkciu – aby vedela produkovať výstupy z ľubovoľného rozsahu:
y = Dense(num_outputs)(y)
Vytvoríme a skompilujeme model. Na regresiu môžeme ako chybovú funkciu použiť strednú kvadratickú chybu (mean squared error; mse
).
model = Model(x, y)
model.compile(loss='mse', optimizer="adam")
Keď sme vytvorili model, môžeme spustiť učenie. Špecifikujeme počet epoch učenia a veľkosť minidávok.
Keďže dátová množina je maličká, môžeme použiť plne dávkové učenie: t.j. veľkosť minidávky nastavíme na veľkosť celej dátovej množiny.
model.fit(X_train, Y_train, nb_epoch=2500, batch_size=len(X_train),
verbose=0, callbacks=[NEpochLogger()])
Ďalej si vytvorený model otestujeme na testovacích dátach. Ako metriky môžeme použiť strednú kvadratickú a strednú absolútnu chybu:
y_test = model.predict(X_test)
y_test_raw = output_preproc.inverse_transform(y_test)
print("MSE: {}".format(mean_squared_error(Y_test_raw, y_test_raw)))
print("MAE: {}".format(mean_absolute_error(Y_test_raw, y_test_raw)))
Aby sme si vytvorili lepšiu predstavu o tom, aké je rozdelenie chýb, môžeme si chyby zobraziť aj v histograme. To nám dá lepšiu predstavu o tom, či robíme na všetkých vzorkách približne rovnaké chyby, či metrikám dominujú veľké chyby na menšom počte atypických vzoriek a pod.
sns.distplot(Y_test_raw - y_test_raw)
Keďže dáta sú len dvojrozmerné, môžeme si navyše vykresliť aj výslednú regresnú krivku:
xx_raw = np.linspace(-20, 20, 100).reshape(-1, 1)
xx = input_preproc.transform(xx_raw)
yy = model.predict(xx)
yy_raw = output_preproc.inverse_transform(yy)
plt.scatter(X_train_raw, Y_train_raw, label="tréningové dáta")
plt.scatter(X_test_raw, Y_test_raw, c='r', label="testovacie dáta")
plt.plot(xx_raw, yy_raw, label="regresná závislosť")
plt.xlabel('x')
plt.ylabel('y')
plt.grid(ls='--')
plt.legend()
Ak sme vyššie použili ako aktivačné funkcie ReLU, krivka bude pravdepodobne trochu kostrbatá – bude sa skladať z viacerých lineárnych úsekov. Pri použití hladkých nelinearít môže vyzerať o niečo lepšie.
scipy
¶Ak si všimneme, že tvar závislosti nápadne pripomína logistickú (sigmoidnú) krivku, môžeme skúsiť body namiesto zložitej neurónovej siete nahradiť ňou. Regresiu môžeme v tom prípade vykonať napríklad pomocou funkcie curve_fit
z balíčka scipy
. Keďže body naozaj pochádzajú zo sigmoidnej krivky (s maličkým šumom), výsledky budú pravdepodobne podstatne lepšie.
def sigmoid(x, a, c):
return 1 / (1 + np.exp(-a*x - c))
popt, _ = curve_fit(sigmoid,
X_train_raw.values.reshape(-1),
Y_train_raw.values.reshape(-1))
popt
Výsledný model opäť otestujeme:
y_test_raw = sigmoid(X_test_raw, *popt)
print("MSE: {}".format(mean_squared_error(Y_test_raw, y_test_raw)))
print("MAE: {}".format(mean_absolute_error(Y_test_raw, y_test_raw)))
sns.distplot(Y_test_raw.values - y_test_raw)
A zobrazíme výslednú regresnú závislosť:
xx = np.arange(-20, 20, 0.05)
yy = sigmoid(xx, *popt)
plt.scatter(X_train_raw, Y_train_raw, label="tréningové dáta")
plt.scatter(X_test_raw, Y_test_raw, c='r', label="testovacie dáta")
plt.plot(xx, yy, label="regresná závislosť")
plt.xlabel('x')
plt.ylabel('y')
plt.grid(ls='--')
plt.legend()
keras
¶Keďže logistická (sigmoidná) krivka sa používa(la) aj ako aktivačná funkcia v neurónových sieťach, môžeme obdobné výsledky dosiahnuť aj pomocou balíčka keras
. Na vstup najprv aplikujeme jednoduchú lineárnu transformáciu pomocou vrstvy Dense
a následne aplikujeme sigmoidnú aktivačnú funkciu.
Hoci používame inú optimalizačnú metódu, výsledky by mali byť podobne dobré ako pri funkcii curve_fit
.
num_inputs = X_train_raw.shape[1]
num_outputs = Y_train_raw.shape[1]
y = x = Input(shape=(num_inputs, ))
y = Dense(num_outputs)(y)
y = Activation('sigmoid')(y)
sig_model = Model(x, y)
sig_model.compile(loss='mse', optimizer="adam")
sig_model.fit(X_train_raw, Y_train_raw, nb_epoch=10000,
batch_size=len(X_train_raw),
verbose=0, callbacks=[NEpochLogger()])
Testovanie:
y_test_raw = sig_model.predict(X_test_raw)
print("MSE: {}".format(mean_squared_error(Y_test_raw, y_test_raw)))
print("MAE: {}".format(mean_absolute_error(Y_test_raw, y_test_raw)))
sns.distplot(Y_test_raw.values - y_test_raw)
xx_raw = np.linspace(-20, 20, 100).reshape(-1, 1)
yy_raw = sig_model.predict(xx_raw)
plt.scatter(X_train_raw, Y_train_raw, label="tréningové dáta")
plt.scatter(X_test_raw, Y_test_raw, c='r', label="testovacie dáta")
plt.plot(xx_raw, yy_raw, label="regresná závislosť")
plt.xlabel('x')
plt.ylabel('y')
plt.grid(ls='--')
plt.legend()