[Question] Classification for critical cases

[jmunozmendi]

Hi everyone,

I'm training an Approximate GP with Cholesky Variational Distribution and a Bernoulli Likelihood for binary classification. Can I somehow modify the loss function so the penalty is greater when predicting 1 when the value is 0 than when predicting 0 when the value is 1?

Thanks in advance

[wjmaddox]

Probably the best way to do so is to sub-class the Bernoulli likelihood and to re-adjust the weights of class zero as compared to class one.

[jmunozmendi]

Thanks for the answer. Could you provide a piece of code? I'm not familiar with the concepts you are mentioning.

[wjmaddox]

something like this is going to be how you'd implement it. i just copied over the weighted bernoulli distribution from the bernoulli pytorch source.

note that none of this is tested but it should be okay.

from torch.nn.functional import binary_cross_entropy_with_logits
from torch.distributions import Normal

class WeightedBernoulli(torch.distributions.Bernoulli):
    def __init__(self, weight, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.weight = weight
        
    def log_prob(self, value):
        if self._validate_args:
            self._validate_sample(value)
        logits, value = broadcast_all(self.logits, value)
        # return -binary_cross_entropy_with_logits(logits, value, reduction='none', weight=self.weight)
        ## edited for clarity, this is roughly binary_cross_entropy_with_logits
        prob_one = torch.sigmoid(logits)
        loss = self.weight * value * prob_one.log().clamp(min=-100.) + (1. - self.weight) * (1. - value) * (1. - prob_one).log().clamp(min=-100.)
        return loss
    
class WeightedBernoulliLikelihood(gpytorch.likelihoods.BernoulliLikelihood):
    def __init__(self, weight, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.weight = weight
        
    def forward(self, function_samples, **kwargs):
        output_probs = Normal(0, 1).cdf(function_samples)
        return WeightedBernoulli(probs=output_probs, weight=self.weight)

    def marginal(self, function_dist, **kwargs):
        mean = function_dist.mean
        var = function_dist.variance
        link = mean.div(torch.sqrt(1 + var))
        output_probs = Normal(0, 1).cdf(link)
        return WeightedBernoulli(probs=output_probs, weight=self.weight)

likelihood = WeightedBernoulliLikelihood(weight=torch.ones(1))

y = likelihood(f)

[jmunozmendi]

Thank you, I'll try it out and come back to you. So the weight passed to the WeightedBernoulliLikelihood is the weight for class 0 or class 1?

[wjmaddox]

the above function should be updated to specify that the weight is on class 1.

[jmunozmendi]

Hi, I modified my script to use the WeightedBernoulliLikelihood but I can't seem to make it work. The training process is happening as usual, no matter the weight I give to the distribution, and the code is never accessing the log_prob function so I don't know where the loss is coming from.

[wjmaddox]

Can you share a minimum working example for what's going on?

[jmunozmendi]

import tqdm
import math
import torch
import gpytorch
from matplotlib import pyplot as plt
import numpy as np

import urllib.request
import os
from scipy.io import loadmat
from math import floor

import GPy

from scipy import stats
from torch.nn.functional import binary_cross_entropy_with_logits
from torch.distributions import Normal
from torch.distributions.utils import broadcast_all




# Load labels. 0 = collision. 1 = no collision
# Load labels2. -1 = collision. 1 = no collision
path = open('G:/Mi unidad/Tesis doctoral/Bimanipulation GP/Datos IK/labels.csv')
labelsOG = np.loadtxt(path, delimiter=",",dtype='int')
labels = labelsOG*2 - 1
labels = labelsOG

# Load joint angles left arm
path = open('G:/Mi unidad/Tesis doctoral/Bimanipulation GP/Datos IK/jointAnglesLeft.csv')
jointAnglesLeft = np.loadtxt(path, delimiter=" ",dtype='double')


# Load joint angles right arm
path = open('G:/Mi unidad/Tesis doctoral/Bimanipulation GP/Datos IK/jointAnglesRight.csv')
jointAnglesRight = np.loadtxt(path, delimiter=" ",dtype='double')

# Calculate sin and cos of joint angles left arm
sinLeft = np.sin(np.deg2rad(jointAnglesLeft[:,:3]))
cosLeft = np.cos(np.deg2rad(jointAnglesLeft[:,:3]))

# Calculate sin and cos of joint angles right arm
sinRight = np.sin(np.deg2rad(jointAnglesRight[:,:3]))
cosRight = np.cos(np.deg2rad(jointAnglesRight[:,:3]))



# X data
# X = np.hstack([sinLeft, cosLeft, sinRight, cosRight])
# X = np.hstack([jointAnglesLeft[:,:3], jointAnglesRight[:,:3]])
X = stats.zscore(np.hstack([jointAnglesLeft[:,:3], jointAnglesRight[:,:3]]))
# X = np.hstack([jointAnglesLeft, jointAnglesRight])
# X = stats.zscore(np.hstack([jointAnglesLeft, jointAnglesRight]))
X = torch.Tensor(X)

# Y data
y = labels.reshape((100000,))
y = torch.Tensor(y)


train_n = int(floor(0.75 * len(X)))

numPoints = train_n
numInducing = 5000
batch_size = 4096

train_x = X[:train_n, :].contiguous()
train_y = y[:train_n].contiguous()

test_x = X[train_n:, :].contiguous()
test_y = torch.Tensor(labelsOG[train_n:]).contiguous()

if torch.cuda.is_available():
    train_x, train_y, test_x, test_y = train_x.cuda(), train_y.cuda(), test_x.cuda(), test_y.cuda()
    
    
    
from torch.utils.data import TensorDataset, DataLoader
train_dataset = TensorDataset(train_x, train_y)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

test_dataset = TensorDataset(test_x, test_y)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)   
    
    
from gpytorch.models import ApproximateGP
from gpytorch.variational import CholeskyVariationalDistribution
from gpytorch.variational import VariationalStrategy

class GPModel(ApproximateGP):
    def __init__(self, inducing_points):
        variational_distribution = CholeskyVariationalDistribution(inducing_points.size(0))
        variational_strategy = VariationalStrategy(self, inducing_points, variational_distribution, learn_inducing_locations=True)
        super(GPModel, self).__init__(variational_strategy)
        self.mean_module = gpytorch.means.ZeroMean()
        self.covar_module = gpytorch.kernels.ScaleKernel(gpytorch.kernels.MaternKernel())

    def forward(self, x):
        mean_x = self.mean_module(x)
        covar_x = self.covar_module(x)
        return gpytorch.distributions.MultivariateNormal(mean_x, covar_x)
    
class WeightedBernoulli(torch.distributions.Bernoulli):
    
    
    def __init__(self, weight, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.weight = weight
        
        
    def expected_log_prob(self, value):
        print("calculating loss")
        if self._validate_args:
            self._validate_sample(value)
        logits, value = broadcast_all(self.logits, value)
        # return -binary_cross_entropy_with_logits(logits, value, reduction='none', weight=self.weight)
        ## edited for clarity, this is roughly binary_cross_entropy_with_logits
        prob_one = torch.sigmoid(logits)
        loss = self.weight * value * prob_one.log().clamp(min=-100.) + (1. - self.weight) * (1. - value) * (1. - prob_one).log().clamp(min=-100.)
        return loss
    
class WeightedBernoulliLikelihood(gpytorch.likelihoods.BernoulliLikelihood):
    def __init__(self, weight, *args, **kwargs):        
        super().__init__(*args, **kwargs)
        self.weight = weight
        
    def forward(self, function_samples, **kwargs):
        output_probs = Normal(0, 1).cdf(function_samples)
        return WeightedBernoulli(probs=output_probs, weight=self.weight)

    def marginal(self, function_dist, **kwargs):
        mean = function_dist.mean
        var = function_dist.variance
        link = mean.div(torch.sqrt(1 + var))
        output_probs = Normal(0, 1).cdf(link)
        return WeightedBernoulli(probs=output_probs, weight=self.weight)


inducing_points = train_x[:numInducing, :]
np.save('inducing.npy',inducing_points.cpu().numpy())
model = GPModel(inducing_points=inducing_points)
likelihood = WeightedBernoulliLikelihood(weight=0.1)



if torch.cuda.is_available():
    model = model.cuda()
    likelihood = likelihood.cuda() 
    
    
num_epochs = 10


model.train()
likelihood.train()

optimizer = torch.optim.Adam(model.parameters(), lr=0.1)

# Our loss object. We're using the VariationalELBO
mll = gpytorch.mlls.VariationalELBO(likelihood, model, num_data=train_y.size(0))


epochs_iter = tqdm.notebook.tqdm(range(num_epochs), desc="Epoch")
for i in epochs_iter:
    # Within each iteration, we will go over each minibatch of data
    minibatch_iter = tqdm.notebook.tqdm(train_loader, desc="Minibatch", leave=False)
    for x_batch, y_batch in minibatch_iter:
        optimizer.zero_grad()
        output = model(x_batch)
        loss = -mll(output, y_batch)
        # minibatch_iter.set_postfix(loss=loss.item())
        loss.backward()
        print('Iter %d/%d - Loss: %.3f  ' % (i + 1, num_epochs, loss.item()))
        optimizer.step()
    
    
    
model.eval()
likelihood.eval()
means = torch.tensor([0.])
observed_pred = torch.tensor([0.])
 
with torch.no_grad():
    for x_batch, y_batch in test_loader:
        preds = likelihood(model(x_batch))
        means = torch.cat([means, preds.mean.cpu()])