algorithm_platform/models_processing/model_koi/losses.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# time: 2023/5/8 13:15
# file: loss.py.py
# author: David
# company: shenyang JY
import tensorflow as tf
from tensorflow.keras.losses import Loss
from typeguard import typechecked
tf.compat.v1.set_random_seed(1234)


class Rmse(tf.keras.losses.Loss):
    """
    自定义损失函数模板
    功能：实现名称设置、参数保存、张量运算分离
    """

    def __init__(self,
                 name,  # 设置损失名称
                 reduction='mean',
                 **kwargs):
        super().__init__(name=name, reduction=reduction)

        # 可添加自定义参数（自动序列化）
        self.param = kwargs.get('param', 1.0)

    def call(self, y_true, y_pred):
        """核心计算逻辑（分离张量运算和非张量运算）"""
        # 非张量运算（预处理）
        coefficient = tf.constant(self.param, dtype=y_pred.dtype)

        # 张量运算（保持计算图兼容性）
        squared_difference = tf.square(y_pred - y_true)
        loss = tf.reduce_mean(squared_difference, axis=-1) * coefficient
        return loss


class SouthLoss(Loss):
    """
    南网新规则损失函数（支持完整序列化）

    参数:
        cap (float): 归一化后的装机容量（需在 [0,1] 范围内）
        name (str): 损失函数名称，默认为 'south_loss'
        reduction (str): 损失归约方式，默认为 'sum_over_batch_size'

    示例:
        >>> loss = SouthLoss(cap=0.5)
        >>> model.compile(loss=loss, optimizer='adam')
    """

    @typechecked
    def __init__(self,
                 cap: float,
                 name: str = "south_loss",
                 reduction: str = "sum_over_batch_size"):
        # 参数校验
        if not 0 <= cap <= 1:
            raise ValueError("cap 必须为归一化后的值且位于 [0,1] 区间")

        super().__init__(name=name, reduction=reduction)

        # 内部处理缩放逻辑（保持原始 cap 的序列化）
        self._raw_cap = cap  # 保存原始值用于序列化
        self.cap = tf.constant(cap * 0.2, dtype=tf.float32)  # 实际计算值

    def get_config(self):
        """获取序列化配置（保存原始 cap 值）"""
        config = super().get_config()
        config.update({
            "cap": self._raw_cap,  # 保存未缩放前的原始值
            "name": self.name,
            "reduction": self.reduction
        })
        return config

    @classmethod
    def from_config(cls, config):
        """反序列化时重建实例"""
        return cls(
            cap=config["cap"],
            name=config["name"],
            reduction=config["reduction"]
        )

    def call(self, y_true, y_pred):
        """计算损失值（带数值稳定化）"""
        # 确保数据类型一致
        y_true = tf.cast(y_true, tf.float32)
        y_pred = tf.cast(y_pred, tf.float32)

        # 数值稳定化处理
        diff = y_true - y_pred
        delta = y_true - self.cap

        # 使用稳定化的 sigmoid 计算
        logistic_input = tf.clip_by_value(10000.0 * delta, -50.0, 50.0)  # 防止梯度爆炸
        logistic_values = tf.sigmoid(logistic_input)

        # 计算基值
        base = logistic_values * y_true + (1 - logistic_values) * self.cap

        # 避免除零错误
        safe_base = tf.where(tf.equal(base, 0.0), 1e-7, base)

        # 计算损失
        loss = tf.reduce_mean(tf.square(diff / safe_base), axis=-1)
        return loss

    def call2(self, y_true, y_predict):
        y_true = y_true * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
        y_predict = y_predict * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
        y_true = y_true[:, 15]
        y_predict = y_predict[:, 15]
        diff = y_true - y_predict
        logistic_values = tf.sigmoid(10000 * (y_true - self.cap))
        base = logistic_values * y_true + (1 - logistic_values) * self.cap
        loss = K.square(diff / base)

        mask_logical = tf.logical_and(tf.greater(y_true, self.cap01), tf.greater(y_predict, self.cap01))
        count = tf.reduce_sum(tf.cast(mask_logical, tf.float32), axis=-1)
        safe_count = tf.maximum(count, 1)
        # reduce_sum_loss = tf.reduce_sum(loss, axis=-1)
        mean_loss = loss / safe_count
        return mean_loss

    def call1(self, y_true, y_predict):
        y_true = y_true * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
        y_predict = y_predict * self.opt.std['C_REAL_VALUE'] + self.opt.mean['C_REAL_VALUE']
        base = tf.where(y_true > self.cap, y_true, tf.ones_like(y_true)*self.cap)
        loss = (y_true - y_predict) / base
        squared_loss = tf.square(loss)
        mean_squared_loss = tf.reduce_mean(squared_loss, axis=[1])
        return  mean_squared_loss



region_loss_d = {
    'northeast': lambda region: Rmse(region),
    'south': lambda cap, region: SouthLoss(cap, region)
}


# 根据地区调用对应逻辑
def region_loss(opt):
    handler = region_loss_d.get(opt.region, opt.region)
    # 判断处理类型并执行
    if callable(handler):
        # 如果是lambda或函数，直接调用
        if opt.region == "south":  # 需要额外参数的地区
            return handler(opt.cap, opt.region)
        else:
            return handler(opt.region)
    else:
        raise TypeError("无效的损失函数")