How to correct this custom loss function for keras

Question

I want to write a custom loss function that would penalize underestimation of positive target values with weights. It would work like mean square error, with the only difference that square errors in said case would get multiplied with a weight greater than 1.

I wrote it like this:

def wmse(ground_truth, predictions):
    square_errors = np.square(np.subtract(ground_truth, predictions))
    weights = np.ones_like(square_errors)
    weights[np.logical_and(predictions < ground_truth, np.sign(ground_truth) > 0)] =  100
    weighted_mse = np.mean(np.multiply(square_errors, weights))
    return weighted_mse

However, when I supply it to my Sequential model in keras with tensorflow as backend:

model.compile(loss=wmse,optimizer='rmsprop')

I get the following error:

 raise TypeError("Using a `tf.Tensor` as a Python `bool` is not allowed. 
TypeError: Using a `tf.Tensor` as a Python `bool` is not allowed. Use `if t is not None:` instead of `if t:` to test if a tensor is defined, and use TensorFlow ops such as tf.cond to execute subgraphs conditioned on the value of a tensor.

The traceback points to this line in wmse:

weights[np.logical_and(predictions < ground_truth, np.sign(ground_truth) > 0)] =  100

I have never worked with keras nor tensorflow until now, so I'd appreciate if someone helped me to adapt this loss function to keras/tensorflow framework. I tried to replace np.logical_and with tensorflow.logical_and, but to no avail, the error is still there.

Answer 1

As @nuric mentioned, you have to implement your loss using only Keras / Tensorflow operations with derivatives, as these frameworks won't be able to back-propagate through other operations (like numpy ones).

A Keras only implementation could look like this:

from keras import backend as K

def wmse(ground_truth, predictions):
    square_errors = (ground_truth - predictions) ** 2
    weights = K.ones_like(square_errors)
    mask = K.less(predictions, ground_truth) & K.greater(K.sign(ground_truth), 0)
    weights =  K.switch(mask, weights * 100, weights)
    weighted_mse = K.mean(square_errors * weights)
    return weighted_mse

gt = K.constant([-2, 2, 1, -1, 3], dtype="int32")
pred = K.constant([-2, 1, 1, -1, 1], dtype="int32")
weights, loss = wmse(gt, pred)

sess = K.get_session()
print(loss.eval(session=sess))
# 100