TensorFlow2的模型训练组件（2）

损失函数
- 损失函数和正则化项
- Tensorflow2内置的损失函数
- 自定义损失函数
评估指标
- 常用的内置评估指标
- 自定义评估指标
优化器
- 优化器的使用
- - 使用optimizer.apply_gradients
  - 使用optimizer.minimize
  - 使用model.fit
- Tensorflow2内置的优化器
回调函数
- 内置回调函数
- 自定义回调函数
参考资料

损失函数

一般地，监督学习的目标函数由损失函数（loss function）和正则化项组成。（Objective = Loss + Regularization）。对于Keras模型，目标函数中的正则化项一般在各层中指定，例如使用Dense的kernel_regularizer 和 bias_regularizer等参数指定权重使用L1或者L2正则化项，此外还可以用kernel_constraint 和 bias_constraint等参数约束权重的取值范围，这也是一种正则化手段。

损失函数在模型编译时候指定。

对于回归模型，通常使用的损失函数是均方损失函数mean_squared_error。
对于二分类模型，通常使用的是二元交叉熵损失函数binary_crossentropy。
对于多分类模型，如果label是one-hot编码的，则使用类别交叉熵损失函数categorical_crossentropy。如果label是类别序号编码的，则需要使用稀疏类别交叉熵损失函数sparse_categorical_crossentropy。

如果有需要，也可以自定义损失函数，自定义损失函数需要接收两个张量y_true,y_pred作为输入参数，并输出一个标量作为损失函数值。

首先，导入相关的依赖库

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras import layers, models, losses, regularizers, constraints

损失函数和正则化项

tf.keras.backend.clear_session()

model = models.Sequential()
model.add(layers.Dense(64, input_dim=64, kernel_regularizer=regularizers.l2(0.01), activity_regularizer=regularizers.l1(0.01), kernel_constraint=constraints.MaxNorm(max_value=2, axis=0)))
model.add(layers.Dense(10, kernel_regularizer=regularizers.l1_l2(0.01,0.01), activation='sigmoid'))
model.compile(optimizer='rmsprop', loss='binary_crossentropy', metrics=['AUC'])
model.summary()
'''
Model: "sequential"
_________________________________________________________________
 Layer (type)                Output Shape              Param #   
=================================================================
 dense (Dense)               (None, 64)                4160      
                                                                 
 dense_1 (Dense)             (None, 10)                650       
                                                                 
=================================================================
Total params: 4,810
Trainable params: 4,810
Non-trainable params: 0
_________________________________________________________________
'''

Tensorflow2内置的损失函数

Tensorflow2内置的损失函数一般有类的实现和函数的实现两种形式。例如：CategoricalCrossentropy 和 categorical_crossentropy 都是类别交叉熵损失函数，前者是类的实现形式，后者是函数的实现形式。

常用的一些内置损失函数说明如下：

mean_squared_error（均方误差损失，用于回归，简写为mse, 类与函数实现形式分别为MeanSquaredError 和MSE）
mean_absolute_error (平均绝对值误差损失，用于回归，简写为mae, 类与函数实现形式分别为MeanAbsoluteError 和MAE)
mean_absolute_percentage_error (平均百分比误差损失，用于回归，简写为 mape, 类与函数实现形式分别为 MeanAbsolutePercentageError 和 MAPE)
Huber(Huber损失，只有类实现形式，用于回归，介于mse和mae之间，对异常值比较鲁棒，相对mse有一定的优势)
binary_crossentropy(二元交叉熵，用于二分类，类实现形式为 BinaryCrossentropy)
categorical_crossentropy(类别交叉熵，用于多分类，要求label为onehot编码，类实现形式为 CategoricalCrossentropy)
sparse_categorical_crossentropy(稀疏类别交叉熵，用于多分类，要求label为序号编码形式，类实现形式为 SparseCategoricalCrossentropy)
hinge(合页损失函数，用于二分类，最著名的应用是作为支持向量机SVM的损失函数，类实现形式为Hinge)
kld(相对熵损失，也叫KL散度，常用于最大期望算法EM的损失函数，两个概率分布差异的一种信息度量。类与函数实现形式分别为 KLDivergence 或 KLD)
cosine_similarity(余弦相似度，可用于多分类，类实现形式为 CosineSimilarity)

自定义损失函数

自定义损失函数接收两个张量y_true,y_pred作为输入参数，并输出一个标量作为损失函数值。也可以对tf.keras.losses.Loss进行子类化，重写call方法实现损失的计算逻辑，从而得到损失函数的类的实现。

下面是一个Focal Loss的自定义实现示范。Focal Loss是一种对binary_crossentropy的改进损失函数形式。它在样本不均衡和存在较多易分类的样本时相比binary_crossentropy具有明显的优势。它有两个可调参数，alpha参数和gamma参数。其中alpha参数主要用于衰减负样本的权重，gamma参数主要用于衰减容易训练样本的权重。从而让模型更加聚焦在正样本和困难样本上。这就是为什么这个损失函数叫做Focal Loss。
$\begin{cases} -\alpha (1-p)^{\gamma}\log(p) & \text{if y = 1}\\ -(1-\alpha) p^{\gamma}\log(1-p) & \text{if y = 0} \end{cases}$

def focal_loss(gamma=2., alpha=0.75):
    def focal_loss_fixed(y_true, y_pred):
        bce = tf.losses.binary_crossentropy(y_true, y_pred)
        p_t = (y_true * y_pred) + ((1 - y_true) * (1 - y_pred))
        alpha_factor = y_true * alpha + (1 - y_true) * (1 - alpha)
        modulating_factor = tf.pow(1.0 - p_t, gamma)
        loss = tf.reduce_sum(alpha_factor * modulating_factor * bce, axis=-1)
        return loss

    return focal_loss_fixed

class FocalLoss(tf.keras.losses.Loss):

    def __init__(self, gamma=2.0, alpha=0.75, name="focal_loss"):
        self.gamma = gamma
        self.alpha = alpha

    def call(self, y_true, y_pred):
        bce = tf.losses.binary_crossentropy(y_true, y_pred)
        p_t = (y_true * y_pred) + ((1 - y_true) * (1 - y_pred))
        alpha_factor = y_true * self.alpha + (1 - y_true) * (1 - self.alpha)
        modulating_factor = tf.pow(1.0 - p_t, self.gamma)
        loss = tf.reduce_sum(alpha_factor * modulating_factor * bce, axis=-1)
        return loss

评估指标

损失函数除了作为模型训练时候的优化目标，也能够作为模型好坏的一种评价指标。但是通常人们还会从其它角度评估模型的好坏。这就是评估指标（metrics）。通常损失函数都可以作为评估指标，如MAE, MSE, CategoricalCrossentropy等也是常用的评估指标。

虽然损失函数可以一定程度上评价模型好坏，但是评估指标不一定可以作为损失函数，例如AUC, Accuracy, Precision。因为评估指标不要求连续可导，而损失函数通常要求连续可导。

编译模型时，可以通过列表形式指定多个评估指标。如果有需要，也可以自定义评估指标。

自定义评估指标需要接收两个张量y_true, y_pred作为输入参数，并输出一个标量作为评估值。也可以对tf.keras.metrics.Metric进行子类化，重写初始化方法, update_state方法, result方法实现评估指标的计算逻辑，从而得到评估指标的类的实现形式。

由于训练的过程通常是分批次训练的，而评估指标要跑完一个epoch才能够得到整体的指标结果。因此，类形式的评估指标更为常见。即需要编写初始化方法以创建与计算指标结果相关的一些中间变量，编写update_state方法在每个batch后更新相关中间变量的状态，编写result方法输出最终指标结果。

如果编写函数形式的评估指标，则只能取epoch中各个batch计算的评估指标结果的平均值作为整个epoch上的评估指标结果，这个结果通常会偏离整个epoch数据一次计算的结果。

常用的内置评估指标

Tensorflow2中常用的内置评估指标如下：

MeanSquaredError（均方误差，用于回归，可以简写为MSE，函数形式为mse）
MeanAbsoluteError (平均绝对值误差，用于回归，可以简写为MAE，函数形式为mae)
MeanAbsolutePercentageError (平均百分比误差，用于回归，可以简写为MAPE，函数形式为mape)
RootMeanSquaredError (均方根误差，用于回归)
Accuracy (准确率，用于分类，可以用字符串"Accuracy"表示，Accuracy=(TP+TN)/(TP+TN+FP+FN)，要求y_true和y_pred都为类别序号编码)
Precision (精确率，用于二分类，Precision = TP/(TP+FP))
Recall (召回率，用于二分类，Recall = TP/(TP+FN))
TruePositives (真正例，用于二分类)
TrueNegatives (真负例，用于二分类)
FalsePositives (假正例，用于二分类)
FalseNegatives (假负例，用于二分类)
AUC(ROC曲线(TPR vs FPR)下的面积，用于二分类，直观解释为随机抽取一个正样本和一个负样本，正样本的预测值大于负样本的概率)
CategoricalAccuracy（分类准确率，与Accuracy含义相同，要求y_true(label)为onehot编码形式）
SparseCategoricalAccuracy (稀疏分类准确率，与Accuracy含义相同，要求y_true(label)为序号编码形式)
MeanIoU (Intersection-Over-Union，常用于图像分割)
TopKCategoricalAccuracy (多分类TopK准确率，要求y_true(label)为onehot编码形式)
SparseTopKCategoricalAccuracy (稀疏多分类TopK准确率，要求y_true(label)为序号编码形式)
Mean (平均值)
Sum (求和)

自定义评估指标

以金融风控领域常用的KS指标为例，示范自定义评估指标。KS指标适合二分类问题，其计算方式为 $K S = ma x (TPR - FPR)$ 。其中 $TPR = TP / (TP + FN)$ , $FPR = FP / (FP + TN)$ 。TPR曲线实际上就是正样本的累积分布曲线(CDF)，FPR曲线实际上就是负样本的累积分布曲线(CDF)。

KS指标就是正样本和负样本累积分布曲线差值的最大值。
KS曲线
（1）函数形式的自定义评估指标

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras import layers, models, losses, metrics

@tf.function
def ks(y_true, y_pred):
    y_true = tf.reshape(y_true, (-1,))
    y_pred = tf.reshape(y_pred, (-1,))
    length = tf.shape(y_true)[0]
    t = tf.math.top_k(y_pred, k=length, sorted=False)
    y_pred_sorted = tf.gather(y_pred, t.indices)
    y_true_sorted = tf.gather(y_true, t.indices)
    cum_positive_ratio = tf.truediv(
        tf.cumsum(y_true_sorted), tf.reduce_sum(y_true_sorted))
    cum_negative_ratio = tf.truediv(
        tf.cumsum(1 - y_true_sorted), tf.reduce_sum(1 - y_true_sorted))
    ks_value = tf.reduce_max(tf.abs(cum_positive_ratio - cum_negative_ratio))
    return ks_value

y_true = tf.constant([[1], [1], [1], [0], [1], [1], [1], [0], [0], [0], [1], [0], [1], [0]])
y_pred = tf.constant([[0.6], [0.1], [0.4], [0.5], [0.7], [0.7], [0.7],
                      [0.4], [0.4], [0.5], [0.8], [0.3], [0.5], [0.3]])
tf.print(ks(y_true, y_pred)) # 0.625

（2）类形式的自定义评估指标

#类形式的自定义评估指标
class KS(metrics.Metric):

    def __init__(self, name = "ks", **kwargs):
        super(KS,self).__init__(name=name,**kwargs)
        self.true_positives = self.add_weight(
            name = "tp",shape = (101,), initializer = "zeros")
        self.false_positives = self.add_weight(
            name = "fp",shape = (101,), initializer = "zeros")

    @tf.function
    def update_state(self,y_true,y_pred):
        y_true = tf.cast(tf.reshape(y_true,(-1,)),tf.bool)
        y_pred = tf.cast(100*tf.reshape(y_pred,(-1,)),tf.int32)

        for i in tf.range(0,tf.shape(y_true)[0]):
            if y_true[i]:
                self.true_positives[y_pred[i]].assign(
                    self.true_positives[y_pred[i]]+1.0)
            else:
                self.false_positives[y_pred[i]].assign(
                    self.false_positives[y_pred[i]]+1.0)
        return (self.true_positives,self.false_positives)

    @tf.function
    def result(self):
        cum_positive_ratio = tf.truediv(
            tf.cumsum(self.true_positives),tf.reduce_sum(self.true_positives))
        cum_negative_ratio = tf.truediv(
            tf.cumsum(self.false_positives),tf.reduce_sum(self.false_positives))
        ks_value = tf.reduce_max(tf.abs(cum_positive_ratio - cum_negative_ratio)) 
        return ks_value

y_true = tf.constant([[1], [1], [1], [0], [1], [1], [1], [0], [0], [0], [1], [0], [1], [0]])
y_pred = tf.constant([[0.6], [0.1], [0.4], [0.5], [0.7], [0.7],
                      [0.7], [0.4], [0.4], [0.5], [0.8], [0.3], [0.5], [0.3]])

myks = KS()
myks.update_state(y_true, y_pred)
tf.print(myks.result()) # 0.625

优化器

模型优化算法的选择直接关系到最终模型的性能。有时候效果不好，未必是特征的问题或者模型设计的问题，很可能就是优化算法的问题。深度学习优化算法大概经历了SGD -> SGDM -> NAG ->Adagrad -> Adadelta(RMSprop) -> Adam -> Nadam这样的发展历程。

对于一般新手炼丹师，优化器直接使用Adam，并使用其默认参数就OK了。一些爱写论文的炼丹师由于追求评估指标效果，可能会偏爱前期使用Adam优化器快速下降，后期使用SGD并精调优化器参数得到更好的结果。

此外目前也有一些前沿的优化算法，据说效果比Adam更好，例如LazyAdam, Look-ahead, RAdam, Ranger等.

优化器的使用

优化器主要使用apply_gradients方法传入变量和对应梯度从而来对给定变量进行迭代，或者直接使用minimize方法对目标函数进行迭代优化。

更常见的使用是在编译时将优化器传入keras的Model，通过调用model.fit实现对Loss的的迭代优化。

初始化优化器时会创建一个变量optimier.iterations用于记录迭代的次数。因此优化器和tf.Variable一样，一般需要在@tf.function外创建。
导入依赖库

import tensorflow as tf
import numpy as np

求 $f(x)=a\times x**2+b\times x+c$ 的最小值

使用optimizer.apply_gradients

x = tf.Variable(0.0, name="x", dtype=tf.float32)
optimizer = tf.keras.optimizers.SGD(learning_rate=0.01)


@tf.function
def minimizef():
    a = tf.constant(1.0)
    b = tf.constant(-2.0)
    c = tf.constant(1.0)
    while tf.constant(True):
        with tf.GradientTape() as tape:
            y = a * tf.pow(x, 2) + b * x + c
        dy_dx = tape.gradient(y, x)
        optimizer.apply_gradients(grads_and_vars=[(dy_dx, x)])
        #迭代终止条件
        if tf.abs(dy_dx) < tf.constant(0.00001):
            break
        if tf.math.mod(optimizer.iterations, 100) == 0:
            tf.print("step = ", optimizer.iterations)
            tf.print("x = ", x)
            tf.print("")
    y = a * tf.pow(x, 2) + b * x + c
    return y

tf.print('y=', minimizef())
tf.print('x=', x)
'''
step =  300
x =  0.997667611

step =  400
x =  0.999690711

step =  500
x =  0.999959

step =  600
x =  0.999994516

y= 0
x= 0.999995232
'''

使用optimizer.minimize

x = tf.Variable(0.0, name='x', dtype=tf.float32)
optimizer = tf.keras.optimizers.SGD(learning_rate=0.01)

def f():
    a = tf.constant(1.0)
    b = tf.constant(-2.0)
    c = tf.constant(1.0)
    y = a * tf.pow(x, 2)+b*x+c
    return y

@tf.function
def train(epoch=1000):
    for _ in tf.range(epoch):
        optimizer.minimize(f, [x])
    tf.print('epoch = ',optimizer.iterations)
    return f()

train(1000)
tf.print('y=',f())
tf.print('x=',x)
'''
epoch =  1000
y= 0
x= 0.999998569
'''

使用model.fit

tf.keras.backend.clear_session()
class FakeModel(tf.keras.models.Model):
    def __init__(self, a, b, c):
        super(FakeModel, self).__init__()
        self.a = a
        self.b = b
        self.c = c

    def build(self):
        self.x = tf.Variable(0.0, name='x')
        self.built = True
    def call(self, features):
        loss = self.a*(self.x)**2+self.b*(self.x)+self.c
        return tf.ones_like(features)*loss

def myloss(y_true, y_pred):
    return tf.reduce_mean(y_pred)

model = FakeModel(tf.constant(1.0), tf.constant(-2.0), tf.constant(1.0))

model.build()
model.summary()
model.compile(optimizer=tf.keras.optimizers.SGD(learning_rate=0.01), loss=myloss)
history = model.fit(tf.zeros((100,2)),tf.ones(100), batch_size=1, epochs=10) # 迭代1000次

tf.print('x=',model.x)
tf.print('loss=',model(tf.constant(0.0)))
'''
x= 0.999998569
loss= 0
'''

Tensorflow2内置的优化器

从SGD到Nadam的各种优化算法，在tf.keras.optimizers子模块种，都有对应的类的实现：

SGD，默认参数为纯SGD, 设置momentum参数不为0实际上变成SGDM, 考虑了一阶动量, 设置nesterov为True后变成NAG，即Nesterov Accelerated Gradient，在计算梯度时计算的是向前走一步所在位置的梯度。
Adagrad, 考虑了二阶动量，对于不同的参数有不同的学习率，即自适应学习率。缺点是学习率单调下降，可能后期学习速率过慢乃至提前停止学习。
RMSprop, 考虑了二阶动量，对于不同的参数有不同的学习率，即自适应学习率，对Adagrad进行了优化，通过指数平滑只考虑一定窗口内的二阶动量。
Adadelta, 考虑了二阶动量，与RMSprop类似，但是更加复杂一些，自适应性更强。
Adam, 同时考虑了一阶动量和二阶动量，可以看成RMSprop上进一步考虑了一阶动量。
Nadam, 在Adam基础上进一步考虑了Nesterov Acceleration。

回调函数

tf.keras的回调函数实际上是一个类，一般是在model.fit时作为参数指定，用于控制①在训练过程开始或者在训练过程结束，②在每个epoch训练开始或者训练结束，③在每个batch训练开始或者训练结束时执行一些操作，比如收集一些日志信息，改变学习率等超参数，提前终止训练过程等等。

同样地，针对model.evaluate或者model.predict也可以指定callbacks参数，用于控制在评估或预测开始或者结束时，在每个batch开始或者结束时执行一些操作，但这种用法相对不常用。大部分的时候，keras.callbacks子模块中定义的回调函数类已经足够使用了，如果有特定的需要，也可以通过keras.callbacks.Callbacks实施子类化构造自定义的回调函数。

所有回调函数都继承至keras.callbacks.Callbacks基类，拥有params和model这两个属性。其中params是一个dict，记录了训练相关参数（例如verbosity，batch size, number of epochs等）。model即当前关联模型的引用。

此外，对于回调类中的一些方法如on_epoch_begin, on_batch_end，还会有一个输入参数logs，提供有关当前epoch或者batch的一些信息，并能够记录计算结果，如果model.fit指定了多个回调函数类，这些logs变量将在这些回调函数类的同名函数间依顺序传递。

内置回调函数

Tensorflow2的内置回调函数如下：

BaseLogger：收集每个epoch上metrics在各个batch上的平均值，对statefull_metrics参数中的带中间状态的指标直接拿最终值，无需对各个batch平均，指标均值结果将添加到logs变量中。该回调函数被所有模型默认添加，且是第一个被添加的。
History： 将BaseLogger计算的各个epoch的metrics结果记录到history这个dict变量中，并作为model.fit的返回值。该回调函数被所有模型默认添加，在BaseLogger之后被添加。
EarlyStopping：当被监控指标在设定的若干个epoch后没有提升，则提前终止训练。
TensorBoard：为Tensorboard可视化保存日志信息。支持评估指标，计算图，模型参数等的可视化。
ModelCheckpoint：在每个epoch后保存模型。
ReduceLROnPlateau：如果监控指标在设定的若干个epoch后没有提升，则以一定的因子减少学习率。
TerminateOnNaN：如果遇到loss为NaN，提前终止训练。
LearningRateScheduler：学习率控制器。给定学习率lr和epoch的函数关系，根据该函数关系在每个epoch前调整学习率。
CSVLogger：将每个epoch后的logs结果记录到CSV文件中。
ProgbarLogger：将每个epoch后的logs结果打印到标准输出流中。

自定义回调函数

可以使用callbacks.LambdaCallback编写较为简单的回调函数，也可以通过对callbacks.Callback子类化编写更加复杂的回调函数逻辑。
想要更加深度了解tf.Keras的回调函数，可以阅读这一部分的源代码。
导入相关依赖库

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras import layers, models, losses, metrics, callbacks
from tensorflow.keras import backend

通过Callback子类化编写回调函数（也就是LearningRateScheduler的源码）

@keras_export('keras.callbacks.LearningRateScheduler')
class LearningRateScheduler(Callback):
  """Learning rate scheduler.

  At the beginning of every epoch, this callback gets the updated learning rate
  value from `schedule` function provided at `__init__`, with the current epoch
  and current learning rate, and applies the updated learning rate
  on the optimizer.

  Args:
    schedule: a function that takes an epoch index (integer, indexed from 0)
        and current learning rate (float) as inputs and returns a new
        learning rate as output (float).
    verbose: int. 0: quiet, 1: update messages.

  Example:

  >>> # This function keeps the initial learning rate for the first ten epochs
  >>> # and decreases it exponentially after that.
  >>> def scheduler(epoch, lr):
  ...   if epoch < 10:
  ...     return lr
  ...   else:
  ...     return lr * tf.math.exp(-0.1)
  >>>
  >>> model = tf.keras.models.Sequential([tf.keras.layers.Dense(10)])
  >>> model.compile(tf.keras.optimizers.SGD(), loss='mse')
  >>> round(model.optimizer.lr.numpy(), 5)
  0.01

  >>> callback = tf.keras.callbacks.LearningRateScheduler(scheduler)
  >>> history = model.fit(np.arange(100).reshape(5, 20), np.zeros(5),
  ...                     epochs=15, callbacks=[callback], verbose=0)
  >>> round(model.optimizer.lr.numpy(), 5)
  0.00607

  """

  def __init__(self, schedule, verbose=0):
    super(LearningRateScheduler, self).__init__()
    self.schedule = schedule
    self.verbose = verbose

  def on_epoch_begin(self, epoch, logs=None):
    if not hasattr(self.model.optimizer, 'lr'):
      raise ValueError('Optimizer must have a "lr" attribute.')
    try:  # new API
      lr = float(backend.get_value(self.model.optimizer.lr))
      lr = self.schedule(epoch, lr)
    except TypeError:  # Support for old API for backward compatibility
      lr = self.schedule(epoch)
    if not isinstance(lr, (tf.Tensor, float, np.float32, np.float64)):
      raise ValueError('The output of the "schedule" function '
                       f'should be float. Got: {lr}')
    if isinstance(lr, tf.Tensor) and not lr.dtype.is_floating:
      raise ValueError(
          f'The dtype of `lr` Tensor should be float. Got: {lr.dtype}')
    backend.set_value(self.model.optimizer.lr, backend.get_value(lr))
    if self.verbose > 0:
      io_utils.print_msg(
          f'\nEpoch {epoch + 1}: LearningRateScheduler setting learning '
          f'rate to {lr}.')

  def on_epoch_end(self, epoch, logs=None):
    logs = logs or {}
    logs['lr'] = backend.get_value(self.model.optimizer.lr)