How to implement custom stateful metric in Keras-CodePudding

I am trying to implement a custom stateful metric in Keras. The Keras API website provides a short demo of how to do so. However, the demo only includes class definition, creating an instance, and calculation of the metric from some data. They do not demonstrate how to use it in the model.fit method, which is how I (and most people) would use it. When I use it, I get results that (1) do not change epoch-to-epoch and (2) conflict with built-in metrics, so I know that I am getting incorrect results. Am I using the code provided incorrectly or is the sample code not functional?

I am including a minimal demonstration of the issue below:

# libraries
import numpy as np
import random
import tensorflow as tf
from sklearn import datasets
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
from sklearn.model_selection import train_test_split

# setting psuedorandom seeds for reproducibility
np.random.seed(0)
tf.random.set_seed(0)
random.seed(0)

# loads famous Iris dataset
iris = datasets.load_iris()
x = iris.data
y = iris.target

# converts to binary prediction problem
y = np.where(y == 2, 1, 0)

# train/validation split 
x_train, x_val, y_train, y_val = train_test_split(x, y, test_size = 0.2)

# reformatting 
x_train = tf.constant(x_train)
x_val = tf.constant(x_val)
y_train = tf.constant(y_train)
y_val = tf.constant(y_val)

# stateful metric code from keras website
class BinaryTruePositives(tf.keras.metrics.Metric):

    def __init__(self, name='binary_true_positives', **kwargs):
        super(BinaryTruePositives, self).__init__(name=name, **kwargs)
        self.true_positives = self.add_weight(name='tp', initializer='zeros')

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_true = tf.cast(y_true, tf.bool)
        y_pred = tf.cast(y_pred, tf.bool)

        values = tf.logical_and(tf.equal(y_true, True), tf.equal(y_pred, True))
        values = tf.cast(values, self.dtype)
        if sample_weight is not None:
            sample_weight = tf.cast(sample_weight, self.dtype)
            values = tf.multiply(values, sample_weight)
        self.true_positives.assign_add(tf.reduce_sum(values))

    def result(self):
        return self.true_positives

    def reset_states(self):
        self.true_positives.assign(0)

# demonstration from website (correct value is returned)
m = BinaryTruePositives()
m.update_state([0, 1, 1, 1], [0, 1, 0, 0])
print(m.result())

# instantiates metric
true_pos = BinaryTruePositives()

# defines a very simple model
model = Sequential()
model.add(Dense(1, activation = 'sigmoid'))

# compiles model
model.compile(optimizer = 'adam',
              loss = 'binary_crossentropy',
              metrics = ['Recall', true_pos])

# trains model and prints output
history = model.fit(x_train, y_train,
                    epochs = 30,
                    verbose = 1,
                    validation_data = (x_val, y_val),
                    batch_size = 10
                    )

Notice that recall improves, while true positives remains constant, which is impossible.

CodePudding user response：

I have slightly modified your BinaryTruePositives class, because in your model, y_pred is a continuous variable in (0, 1), not boolean. You need to convert it to a boolean variable using tf.where(y_pred > 0.5, True, False) for example.

class BinaryTruePositives(tf.keras.metrics.Metric):

    def __init__(self, name='binary_true_positives', **kwargs):
        super(BinaryTruePositives, self).__init__(name=name, **kwargs)
        self.true_positives = self.add_weight(name='tp', initializer='zeros')

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_true = tf.cast(y_true, tf.bool)
        y_pred = tf.where(y_pred > 0.5, True, False)

        values = tf.logical_and(tf.equal(y_true, True), tf.equal(y_pred, True))
        values = tf.cast(values, self.dtype)
        if sample_weight is not None:
            sample_weight = tf.cast(sample_weight, self.dtype)
            values = tf.multiply(values, sample_weight)
        self.true_positives.assign_add(tf.reduce_sum(values))

    def result(self):
        return self.true_positives

    def reset_states(self):
        self.true_positives.assign(0)