algorithm_platform/models_processing/model_tf/tf_transformer.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @FileName  :tf_lstm.py
# @Time      :2025/2/12 14:03
# @Author    :David
# @Company: shenyang JY

from tensorflow.keras.layers import Input, Dense, LSTM, concatenate, Conv1D, Conv2D, MaxPooling1D, Reshape, Flatten, LayerNormalization, Dropout
from tensorflow.keras.models import Model, load_model
from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping, TensorBoard, ReduceLROnPlateau
from tensorflow.keras import optimizers, regularizers
from models_processing.model_tf.losses import region_loss
import numpy as np
from common.database_dml_koi import *
from models_processing.model_tf.settings import set_deterministic
from threading import Lock
import argparse
model_lock = Lock()
set_deterministic(42)

class TSHandler(object):
    def __init__(self, logger, args):
        self.logger = logger
        self.opt = argparse.Namespace(**args)
        self.model = None
        self.model_params = None

    def get_model(self, args):
        """
        单例模式+线程锁，防止在异步加载时引发线程安全
        """
        try:
            with model_lock:
                loss = region_loss(self.opt)
                self.model, self.model_params = get_keras_model_from_mongo(args, {type(loss).__name__: loss})
        except Exception as e:
            self.logger.info("加载模型权重失败:{}".format(e.args))

    @staticmethod
    def get_keras_model(opt, time_series=1, lstm_type=1):
        loss = region_loss(opt)
        l1_reg = regularizers.l1(opt.Model['lambda_value_1'])
        l2_reg = regularizers.l2(opt.Model['lambda_value_2'])
        nwp_input = Input(shape=(opt.Model['time_step']*time_series, opt.Model['input_size']), name='nwp')

        con1 = Conv1D(filters=64, kernel_size=5, strides=1, padding='valid', activation='relu', kernel_regularizer=l2_reg)(nwp_input)
        con1_p = MaxPooling1D(pool_size=5, strides=1, padding='valid', data_format='channels_last')(con1)
        nwp_lstm = LSTM(units=opt.Model['hidden_size'], return_sequences=False, kernel_regularizer=l2_reg)(con1_p)
        if lstm_type == 2:
            output = Dense(opt.Model['time_step'], name='cdq_output')(nwp_lstm)
        else:
            output = Dense(opt.Model['time_step']*time_series, name='cdq_output')(nwp_lstm)

        model = Model(nwp_input, output)
        adam = optimizers.Adam(learning_rate=0.001, beta_1=0.9, beta_2=0.999, epsilon=1e-7, amsgrad=True)
        model.compile(loss=loss, optimizer=adam)
        return model

    @staticmethod
    def get_tcn_model(opt, time_series=1):
        # 参数设置
        loss = region_loss(opt)
        time_steps = 48  # 输入时间步长 (16*3)
        output_steps = 16  # 输出时间步长
        hidden_size = opt.Model.get('hidden_size', 64)
        l2_reg = regularizers.l2(opt.Model['lambda_value_2'])
        dropout_rate = opt.Model.get('dropout_rate', 0.2)

        # 输入层
        nwp_input = Input(shape=(opt.Model['time_step']*time_series, opt.Model['input_size']), name='nwp')

        # 初始卷积层 (将通道数扩展到hidden_size)
        x = Conv1D(filters=hidden_size, kernel_size=3, strides=1, padding='causal', activation='relu', kernel_regularizer=l2_reg)(nwp_input)

        # 时序卷积块 (TCN块)
        for d in [1, 2, 4, 8]:  # 扩张系数
            # 扩张因果卷积
            conv = Conv1D(filters=hidden_size, kernel_size=3, strides=1,
                          padding='causal', activation='relu',
                          dilation_rate=d,
                          kernel_regularizer=l2_reg)
            x = conv(x)
            # 残差连接
            skip = Conv1D(filters=hidden_size, kernel_size=1,
                          padding='same')(x)
            # 层归一化
            x = LayerNormalization()(x)
            x = tf.keras.activations.relu(x)
            x = Dropout(dropout_rate)(x)
            x = x + skip  # 残差连接

        # 提取中间16个时间步的表示
        # 这里我们使用全局平均池化或直接切片
        # 方法1: 使用全局平均池化然后上采样
        # x = tf.reduce_mean(x, axis=1, keepdims=True)
        # x = tf.tile(x, [1, output_steps, 1])

        # 方法2: 直接切片中间16个时间步 (更符合你的需求)
        # 由于是因果卷积，中间时间步大致对应输入的中间部分
        start_idx = (time_steps - output_steps) // 2
        x = x[:, start_idx:start_idx + output_steps, :]

        # 输出层
        output = Dense(output_steps, activation=None, name='cdq_output')(x)

        # 创建模型
        model = Model(nwp_input, output)

        # 优化器
        adam = optimizers.Adam(learning_rate=0.001, beta_1=0.9, beta_2=0.999, epsilon=1e-7, amsgrad=True)
        model.compile(loss=loss, optimizer=adam)
        return model

    def train_init(self):
        try:
            # 进行加强训练，支持修模
            loss = region_loss(self.opt)
            base_train_model, self.model_params = get_keras_model_from_mongo(vars(self.opt), {type(loss).__name__: loss})
            base_train_model.summary()
            self.logger.info("已加载加强训练基础模型")
            return base_train_model
        except Exception as e:
            self.logger.info("加载加强训练模型权重失败:{}".format(e.args))
            return False

    def training(self, model, train_and_valid_data):
        model.summary()
        train_x, train_y, valid_x, valid_y = train_and_valid_data
        early_stop = EarlyStopping(monitor='val_loss', patience=self.opt.Model['patience'], mode='auto')
        history = model.fit(train_x, train_y, batch_size=self.opt.Model['batch_size'], epochs=self.opt.Model['epoch'],
                            verbose=2, validation_data=(valid_x, valid_y), callbacks=[early_stop], shuffle=False)
        loss = np.round(history.history['loss'], decimals=5)
        val_loss = np.round(history.history['val_loss'], decimals=5)
        self.logger.info("-----模型训练经过{}轮迭代-----".format(len(loss)))
        self.logger.info("训练集损失函数为：{}".format(loss))
        self.logger.info("验证集损失函数为：{}".format(val_loss))
        return model

    def predict(self, test_x, batch_size=1):
        result = self.model.predict(test_x, batch_size=batch_size)
        self.logger.info("执行预测方法")
        return result



if __name__ == "__main__":
    run_code = 0