ConcordantPercentageMetric.py

class ConcordantPercentage(keras.metrics.Metric):
    def __init__(self, name='concordant_pct', baseline=1, **kwargs):
        baseline_suffix = '' if baseline == 1 else f'_{baseline}'
        _name = f'{name}{baseline_suffix}'
        super().__init__(name=_name, **kwargs)
        self.baseline = baseline
        self.concordant_samples = self.add_weight(name='concordant_samples', shape=(1,), initializer='zeros', dtype=tf.int32)
        self.num_samples = self.add_weight(name='num_samples', shape=(1,), initializer='zeros', dtype=tf.int32)

    def update_state(self, y_true, y_pred, **kwargs):
        bs = tf.shape(y_true)[0]

        cc_positive = tf.logical_and(y_true >= 0, y_pred >= 0)
        cc_negative = tf.logical_and(y_true < 0, y_pred < 0)

        err = tf.abs(y_true - y_pred)
        b = self.baseline

        cc_positive = tf.logical_and(cc_positive, err <= b)
        cc_negative = tf.logical_and(cc_negative, err <= b)
        
        cc_total = tf.logical_or(cc_positive, cc_negative)
        
        cc_samples = tf.reduce_sum(tf.cast(cc_total, dtype=tf.int32))

        self.concordant_samples.assign_add(tf.expand_dims(cc_samples, -1))
        self.num_samples.assign_add(tf.expand_dims(tf.cast(bs, dtype=tf.int32), -1))

    def result(self):
        return tf.cast(self.concordant_samples, dtype=tf.float32) / tf.cast(self.num_samples, dtype=tf.float32)

    def reset_states(self):
        self.concordant_samples.assign([0])
        self.num_samples.assign([0])