algorithm_platform/models_processing/model_tf/losses_cash.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @FileName  :losses_cash.py
# @Time      :2025/6/11 15:31
# @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 SouthLossCash(Loss):
    """南网新规则损失函数（支持完整序列化）"""

    @typechecked
    def __init__(self,
                 cap: float,
                 name: str = "south_cash_loss",
                 penalty_coeff: float = 1.0,  # 新增惩罚系数参数
                 reduction: str = "sum_over_batch_size"):
        # if not 0 <= cap <= 1:
        #     raise ValueError("cap 必须为归一化后的值且位于 [0,1] 区间")

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

        # 序列化相关参数
        self._raw_cap = cap
        self.penalty_coeff = penalty_coeff  # 惩罚项系数

        # 实际计算值（注意：这里的0.2是原代码中的缩放逻辑，保持不动）
        self.cap = tf.constant(cap * 0.2, dtype=tf.float32)

    def get_config(self):
        config = super().get_config()
        config.update({
            "cap": self._raw_cap,
            "penalty_coeff": self.penalty_coeff,  # 序列化新参数
            "name": self.name,
            "reduction": self.reduction
        })
        return config

    @classmethod
    def from_config(cls, config):
        return cls(
            cap=config["cap"],
            penalty_coeff=config.get("penalty_coeff", 1.0),  # 带默认值以兼容旧配置
            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

        # 基值计算逻辑（保持原样）
        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)

        # 原基础损失
        base_loss = tf.reduce_mean(tf.square(diff / safe_base), axis=-1)

        ############### 新增惩罚项部分 ###############
        # 计算负偏差惩罚（预测值低于真实值时进行惩罚）
        negative_bias = tf.maximum(diff, 0.0)  # 获取负偏差部分的绝对值
        penalty = self.penalty_coeff * tf.reduce_mean(negative_bias, axis=-1)

        # 组合最终损失（可以尝试不同权重）
        total_loss = base_loss + penalty

        return total_loss
if __name__ == "__main__":
    run_code = 0