<jupyter_start><jupyter_text>02: Probabilistic Regression Model with TFP<jupyter_code>import matplotlib.pyplot as plt
import numpy as np
from sklearn.model_selection import train_test_split
import tensorflow_probability as tfp
import tensorflow as tf
from tensorflow.keras.layers import Input, Dense, Concatenate
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
%matplotlib inline<jupyter_output><empty_output><jupyter_text>Make Dataset  Design standard deviation $\sigma$<jupyter_code>x_1 = np.linspace(1, 8, 100)
x_2 = np.repeat(8, 30)
x_3 = np.linspace(8, 1, 150)
x_4 = np.repeat(1, 60)
x_5 = np.linspace(1, 8, 120)
x_6 = np.linspace(8, 1, 200)
x_7 = np.repeat(1, 50)
x = np.concatenate([x_1, x_2, x_3, x_4, x_5, x_6, x_7])
plt.plot(x)
plt.xlabel("index")
plt.ylabel("$\sigma$")
plt.show()
x.shape<jupyter_output><empty_output><jupyter_text>Generate samples based on $\sigma$<jupyter_code>noise = np.random.normal(0, x, size=len(x))
plt.plot(noise)
plt.xlabel("indx")
plt.ylabel("noise")
plt.show()
x_sample = np.sort(np.random.uniform(-3, 3, size=len(x)))
y_sample = 3.3 * x_sample + 3.0 + noise
plt.scatter(x_sample, y_sample)
plt.xlabel("x")
plt.ylabel("y")
plt.show()<jupyter_output><empty_output><jupyter_text>Split Train / Validation / Test dataset<jupyter_code>x_train, x_test, y_train, y_test = train_test_split(x_sample, y_sample, test_size=0.2)
x_train, x_val, y_train, y_val = train_test_split(x_train, y_train, test_size=0.2)
order_idx_train = x_train.argsort(axis=0)
x_train = x_train[order_idx_train, np.newaxis]
y_train = y_train[order_idx_train, np.newaxis]
x_train.shape
order_idx_val = x_val.argsort(axis=0)
x_val = x_val[order_idx_val, np.newaxis]
y_val = y_val[order_idx_val, np.newaxis]
x_val.shape
order_idx_test = x_test.argsort(axis=0)
x_test = x_test[order_idx_test, np.newaxis]
y_test = y_test[order_idx_test, np.newaxis]
x_test.shape<jupyter_output><empty_output><jupyter_text>Visualize Train, Validation, Test<jupyter_code>plt.scatter(x_train, y_train)
plt.scatter(x_val, y_val)
plt.scatter(x_test, y_test)<jupyter_output><empty_output><jupyter_text>Fitting and evaluating a linear regression model with constant variance<jupyter_code>def NLL(y, distribution):
    return -distribution.log_prob(y)

def my_dist(params):
    return tfp.distributions.Normal(loc=params, scale=1)

inputs = Input(shape=(1,))
params = Dense(1)(inputs)
dist = tfp.layers.DistributionLambda(my_dist)(params)
model_sd_1 = Model(inputs=inputs, outputs=dist)
model_sd_1.compile(Adam(), loss=NLL)
model_sd_1.summary()
history = model_sd_1.fit(x_train, y_train, epochs=2000, verbose=0, validation_data=(x_val, y_val))<jupyter_output><empty_output><jupyter_text>Let's check the training found the correct slope and intercept.<jupyter_code>model_sd_1.get_weights()
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.legend(['loss', 'val_loss'])
plt.ylabel('NLL')
plt.xlabel('Epochs')
plt.show()
x_pred = np.arange(-3, 3, 0.1)
x_pred.shape
preds = model_sd_1(x_pred).mean()
sigma = 1 # predefined
plt.figure(figsize=(14, 5))
plt.subplot(1,2,1)
plt.scatter(x_train, y_train)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Train Data ($\sigma=1$)")

plt.subplot(1,2,2)
plt.scatter(x_val, y_val)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Validation Data ($\sigma=1$)")
plt.show()<jupyter_output><empty_output><jupyter_text>Performance (NLL)<jupyter_code>model_sd_1.evaluate(x_train, y_train)
model_sd_1.evaluate(x_val, y_val)<jupyter_output>4/4 [==============================] - 0s 2ms/step - loss: 8.5542<jupyter_text>How about using $\sigma=5$?<jupyter_code>def NLL(y, distribution):
    return -distribution.log_prob(y)

def my_dist(params):
    return tfp.distributions.Normal(loc=params, scale=5)

inputs = Input(shape=(1,))
params = Dense(1)(inputs)
dist = tfp.layers.DistributionLambda(my_dist)(params)
model_sd_5 = Model(inputs=inputs, outputs=dist)
model_sd_5.compile(Adam(), loss=NLL)
history = model_sd_5.fit(x_train, y_train, epochs=2000, verbose=0, validation_data=(x_val, y_val))
model_sd_5.get_weights()
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.legend(['loss', 'val_loss'])
plt.ylabel('NLL')
plt.xlabel('Epochs')
plt.show()
preds = model_sd_5(x_pred).mean()
sigma = 5 # predefined
plt.figure(figsize=(14, 5))
plt.subplot(1,2,1)
plt.scatter(x_train, y_train)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Train Data ($\sigma=5$)")

plt.subplot(1,2,2)
plt.scatter(x_val, y_val)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Validation Data ($\sigma=5$)")
plt.show()
model_sd_5.evaluate(x_train, y_train)
model_sd_5.evaluate(x_val, y_val)<jupyter_output>4/4 [==============================] - 0s 1ms/step - loss: 2.8337<jupyter_text>Fitting and evaluating a linear regression model with a nonconstant variance<jupyter_code>def NLL(y, distribution):
    return -distribution.log_prob(y)

def my_dist(params):
    return tfp.distributions.Normal(loc=params[:,0:1], scale=1e-3 + tf.math.softplus(0.05 * params[:,1:2]))

inputs = Input(shape=(1,))
params = Dense(2)(inputs)
dist = tfp.layers.DistributionLambda(my_dist)(params)
model_monotonic_sd = Model(inputs=inputs, outputs=dist)
model_monotonic_sd.compile(Adam(), loss=NLL)
model_monotonic_sd.summary()
history = model_monotonic_sd.fit(x_train, y_train, epochs=2000, verbose=0, validation_data=(x_val,y_val))
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.legend(['loss', 'val_loss'])
plt.ylabel('NLL')
plt.xlabel('Epochs')
plt.show()
print(model_monotonic_sd.evaluate(x_train,y_train, verbose=0))
print(model_monotonic_sd.evaluate(x_val,y_val, verbose=0))
preds = model_monotonic_sd(x_pred).mean()
sigma = model_monotonic_sd(x_pred).stddev()
plt.figure(figsize=(14, 5))
plt.subplot(1,2,1)
plt.scatter(x_train, y_train)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Train Data")

plt.subplot(1,2,2)
plt.scatter(x_val, y_val)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Validation Data")
plt.show()<jupyter_output><empty_output><jupyter_text>Give more flexibility for standard deviation<jupyter_code>def NLL(y, distr):
    return -distr.log_prob(y) 

def my_dist(params):
    return tfp.distributions.Normal(loc=params[:,0:1], scale=1e-3 + tf.math.softplus(0.05 * params[:,1:2]))

inputs = Input(shape=(1,))
out1 = Dense(1)(inputs)
hidden1 = Dense(30,activation="relu")(inputs)
hidden2 = Dense(20,activation="relu")(hidden1)
hidden3 = Dense(20,activation="relu")(hidden2)
out2 = Dense(1)(hidden3)
params = Concatenate()([out1,out2])
dist = tfp.layers.DistributionLambda(my_dist)(params) 

model_flex_sd = Model(inputs=inputs, outputs=dist)
model_flex_sd.compile(Adam(learning_rate=0.01), loss=NLL)
model_flex_sd.summary()
history = model_flex_sd.fit(x_train, y_train, epochs=2000, verbose=0, validation_data=(x_val,y_val))
preds = model_flex_sd(x_pred).mean()
sigma = model_flex_sd(x_pred).stddev()
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.legend(['loss', 'val_loss'])
plt.ylabel('NLL')
plt.xlabel('Epochs')
plt.show()
print(model_flex_sd.evaluate(x_train,y_train, verbose=0))
print(model_flex_sd.evaluate(x_val,y_val, verbose=0))
preds = model_flex_sd(x_pred).mean()
sigma = model_flex_sd(x_pred).stddev()
plt.figure(figsize=(14, 5))
plt.subplot(1,2,1)
plt.scatter(x_train, y_train)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Train Data")

plt.subplot(1,2,2)
plt.scatter(x_val, y_val)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Validation Data")
plt.show()<jupyter_output><empty_output><jupyter_text>Test the model on test data **Make sure that training and test data shape is a matrix, not a vector.**<jupyter_code>model_flex_sd.evaluate(x_test, y_test, verbose=0)
plt.figure(figsize=(14, 5))
plt.scatter(x_test, y_test)
plt.plot(x_pred, preds, color="red", linewidth=4)
plt.plot(x_pred, preds + 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.plot(x_pred, preds - 2*sigma, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Test Data")
plt.show()
-np.mean(model_flex_sd(x_test).log_prob(y_test))
NLL_test = np.mean(-np.log(1/(np.sqrt(2*np.pi*np.square(model_flex_sd(x_test).stddev()))))+
                 ((np.square(y_test-model_flex_sd(x_test).mean())/(2*np.square(model_flex_sd(x_test).stddev())))))
print(NLL_test)<jupyter_output>2.6366794<jupyter_text>Test it on outside the observed data region (extrapolation)<jupyter_code>x_extp = np.arange(-10, 10, 0.1)
preds_extp = model_flex_sd(x_extp).mean()
sigma_extp = model_flex_sd(x_extp).stddev()

plt.scatter(x_test, y_test)
plt.plot(x_extp, preds_extp, color="red", linewidth=4)
plt.plot(x_extp, preds_extp + 2*sigma_extp, color="orange", linestyle="--", linewidth=4)
plt.plot(x_extp, preds_extp - 2*sigma_extp, color="orange", linestyle="--", linewidth=4)
plt.xlabel("x")
plt.ylabel("y")
plt.title("Test Data (with extrapolation)")<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup