06171256

models_processing/model_tf/losses_cash.py

@@ -73,7 +73,9 @@ class SouthLossCash(Loss):
         ############### 新增惩罚项部分 ###############
         # 计算负偏差惩罚（预测值低于真实值时进行惩罚）
         negative_bias = tf.maximum(diff, 0.0)  # 获取负偏差部分的绝对值
-        penalty = self.penalty_coeff * tf.reduce_mean(negative_bias, axis=-1)
+        # 统一惩罚项 和 准确率损失函数量纲
+        negative_bias_ = tf.square(negative_bias / safe_base)
+        penalty = self.penalty_coeff * tf.reduce_mean(negative_bias_, axis=-1)
 
         # 组合最终损失（可以尝试不同权重）
         total_loss = base_loss + penalty

models_processing/model_tf/tf_transformer.py

@@ -37,26 +37,6 @@ class TransformerHandler(object):
             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_transformer_model(opt, time_series=1):
         time_steps = 48
         input_features = 21

models_processing/model_tf/tf_transformer_pre.py

@@ -0,0 +1,143 @@
+#!/usr/bin/env python
+# -*- coding:utf-8 -*-
+# @FileName  :tf_lstm_pre.py
+# @Time      :2025/2/13 10:52
+# @Author    :David
+# @Company: shenyang JY
+import json, copy
+import numpy as np
+from flask import Flask, request, g
+import logging, argparse, traceback
+from common.database_dml import *
+from common.processing_data_common import missing_features, str_to_list
+from data_processing.data_operation.data_handler import DataHandler
+from threading import Lock
+import time, yaml
+from copy import deepcopy
+model_lock = Lock()
+from itertools import chain
+from common.logs import Log
+from common.data_utils import deep_update
+from tf_transformer import TransformerHandler
+# logger = Log('tf_bp').logger()
+logger = Log('tf_ts').logger
+np.random.seed(42)  # NumPy随机种子
+# tf.set_random_seed(42)  # TensorFlow随机种子
+app = Flask('tf_lstm_pre——service')
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+with open(os.path.join(current_dir, 'lstm.yaml'), 'r', encoding='utf-8') as f:
+    global_config = yaml.safe_load(f)  # 只读的全局配置
+
+@app.before_request
+def update_config():
+    # ------------ 整理参数，整合请求参数 ------------
+    # 深拷贝全局配置 + 合并请求参数
+    current_config = deepcopy(global_config)
+    request_args = request.values.to_dict()
+    # features参数规则：1.有传入，解析，覆盖 2. 无传入，不覆盖，原始值
+    request_args['features'] = request_args['features'].split(',') if 'features' in request_args else current_config['features']
+    request_args['time_series'] = request_args.get('time_series', 1)
+    current_config = deep_update(current_config, request_args)
+
+    # 存储到请求上下文
+    g.opt = argparse.Namespace(**current_config)
+    g.dh = DataHandler(logger, current_config)  # 每个请求独立实例
+    g.trans = TransformerHandler(logger, current_config)
+
+@app.route('/tf_lstm_predict', methods=['POST'])
+def model_prediction_lstm():
+    # 获取程序开始时间
+    start_time = time.time()
+    result = {}
+    success = 0
+    dh = g.dh
+    trans = g.trans
+    args = deepcopy(g.opt.__dict__)
+    logger.info("Program starts execution!")
+    try:
+        pre_data = get_data_from_mongo(args)
+        if args.get('algorithm_test', 0):
+            field_mapping = {'clearsky_ghi': 'clearskyGhi', 'dni_calcd': 'dniCalcd','surface_pressure': 'surfacePressure'}
+            pre_data = pre_data.rename(columns=field_mapping)
+        feature_scaler, target_scaler = get_scaler_model_from_mongo(args)
+        trans.opt.cap = round(target_scaler.transform(np.array([[float(args['cap'])]]))[0, 0], 2)
+        trans.get_model(args)
+        dh.opt.features = json.loads(trans.model_params)['Model']['features'].split(',')
+        scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler, time_series=args['time_series'], lstm_type=1)
+        res = list(chain.from_iterable(target_scaler.inverse_transform(trans.predict(scaled_pre_x))))
+        pre_data['farm_id'] = args.get('farm_id', 'null')
+        if int(args.get('algorithm_test', 0)):
+            pre_data[args['model_name']] = res[:len(pre_data)]
+            pre_data.rename(columns={args['col_time']: 'dateTime'}, inplace=True)
+            pre_data = pre_data[['dateTime', 'farm_id', args['target'], args['model_name'], 'dq']]
+            pre_data = pre_data.melt(id_vars=['dateTime', 'farm_id', args['target']], var_name='model', value_name='power_forecast')
+            res_cols = ['dateTime', 'power_forecast', 'farm_id', args['target'], 'model']
+            if 'howLongAgo' in args:
+                pre_data['howLongAgo'] = int(args['howLongAgo'])
+                res_cols += ['howLongAgo']
+        else:
+            pre_data['power_forecast'] = res[:len(pre_data)]
+            pre_data.rename(columns={args['col_time']: 'date_time'}, inplace=True)
+            res_cols = ['date_time', 'power_forecast', 'farm_id']
+        pre_data = pre_data[res_cols]
+
+        pre_data.loc[:, 'power_forecast'] = pre_data.loc[:, 'power_forecast'].apply(lambda x: float(f"{x:.2f}"))
+        pre_data.loc[pre_data['power_forecast'] > float(args['cap']), 'power_forecast'] = float(args['cap'])
+        pre_data.loc[pre_data['power_forecast'] < 0, 'power_forecast'] = 0
+
+        insert_data_into_mongo(pre_data, args)
+        success = 1
+    except Exception as e:
+        my_exception = traceback.format_exc()
+        my_exception.replace("\n", "\t")
+        result['msg'] = my_exception
+    end_time = time.time()
+
+    result['success'] = success
+    result['args'] = args
+    result['start_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))
+    result['end_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))
+    print("Program execution ends!")
+    return result
+
+
+if __name__ == "__main__":
+    print("Program starts execution!")
+    from waitress import serve
+    serve(app, host="0.0.0.0", port=10114,
+          threads=8,  # 指定线程数（默认4，根据硬件调整）
+          channel_timeout=600  # 连接超时时间（秒）
+          )
+    print("server start!")
+
+    # ------------------------测试代码------------------------
+    # args_dict = {"mongodb_database": 'david_test', 'scaler_table': 'j00083_scaler', 'model_name': 'bp1.0.test',
+    #              'model_table': 'j00083_model', 'mongodb_read_table': 'j00083_test', 'col_time': 'date_time', 'mongodb_write_table': 'j00083_rs',
+    #              'features': 'speed10,direction10,speed30,direction30,speed50,direction50,speed70,direction70,speed90,direction90,speed110,direction110,speed150,direction150,speed170,direction170'}
+    # args_dict['features'] = args_dict['features'].split(',')
+    # arguments.update(args_dict)
+    # dh = DataHandler(logger, arguments)
+    # ts = TSHandler(logger)
+    # opt = argparse.Namespace(**arguments)
+    #
+    # opt.Model['input_size'] = len(opt.features)
+    # pre_data = get_data_from_mongo(args_dict)
+    # feature_scaler, target_scaler = get_scaler_model_from_mongo(arguments)
+    # pre_x = dh.pre_data_handler(pre_data, feature_scaler, opt)
+    # ts.get_model(arguments)
+    # result = ts.predict(pre_x)
+    # result1 = list(chain.from_iterable(target_scaler.inverse_transform([result.flatten()])))
+    # pre_data['power_forecast'] = result1[:len(pre_data)]
+    # pre_data['farm_id'] = 'J00083'
+    # pre_data['cdq'] = 1
+    # pre_data['dq'] = 1
+    # pre_data['zq'] = 1
+    # pre_data.rename(columns={arguments['col_time']: 'date_time'}, inplace=True)
+    # pre_data = pre_data[['date_time', 'power_forecast', 'farm_id', 'cdq', 'dq', 'zq']]
+    #
+    # pre_data['power_forecast'] = pre_data['power_forecast'].round(2)
+    # pre_data.loc[pre_data['power_forecast'] > opt.cap, 'power_forecast'] = opt.cap
+    # pre_data.loc[pre_data['power_forecast'] < 0, 'power_forecast'] = 0
+    #
+    # insert_data_into_mongo(pre_data, arguments)

models_processing/model_tf/tf_transformer_train.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python
+# -*- coding:utf-8 -*-
+# @FileName  :tf_lstm_train.py
+# @Time      :2025/2/13 10:52
+# @Author    :David
+# @Company: shenyang JY
+import json, copy
+import numpy as np
+from flask import Flask, request, jsonify, g
+import traceback, uuid
+import logging, argparse
+from data_processing.data_operation.data_handler import DataHandler
+import time, yaml, threading
+from copy import deepcopy
+from models_processing.model_tf.tf_transformer import TransformerHandler
+from common.database_dml import *
+from common.logs import Log
+from common.data_utils import deep_update
+logger = Log('tf_ts').logger
+np.random.seed(42)  # NumPy随机种子
+app = Flask('tf_lstm_train——service')
+
+current_dir = os.path.dirname(os.path.abspath(__file__))
+with open(os.path.join(current_dir, 'lstm.yaml'), 'r', encoding='utf-8') as f:
+    global_config = yaml.safe_load(f)  # 只读的全局配置
+
+@app.before_request
+def update_config():
+    # ------------ 整理参数，整合请求参数 ------------
+    # 深拷贝全局配置 + 合并请求参数
+    current_config = deepcopy(global_config)
+    request_args = request.values.to_dict()
+    # features参数规则：1.有传入，解析，覆盖 2. 无传入，不覆盖，原始值
+    request_args['features'] = request_args['features'].split(',') if 'features' in request_args else current_config['features']
+    request_args['time_series'] = request_args.get('time_series', 1)
+    current_config = deep_update(current_config, request_args)
+
+    # 存储到请求上下文
+    g.opt = argparse.Namespace(**current_config)
+    g.dh = DataHandler(logger, current_config)  # 每个请求独立实例
+    g.trans = TransformerHandler(logger, current_config)
+
+
+@app.route('/tf_lstm_training', methods=['POST'])
+def model_training_lstm():
+    # 获取程序开始时间
+    start_time = time.time()
+    result = {}
+    success = 0
+    dh = g.dh
+    trans = g.trans
+    args = deepcopy(g.opt.__dict__)
+    logger.info("Program starts execution!")
+    try:
+        # ------------ 获取数据，预处理训练数据 ------------
+        train_data = get_data_from_mongo(args)
+        train_x, train_y, valid_x, valid_y, scaled_train_bytes, scaled_target_bytes, scaled_cap = dh.train_data_handler(train_data, time_series=args['time_series'])
+        trans.opt.cap = round(scaled_cap, 2)
+        trans.opt.Model['input_size'] = len(dh.opt.features)
+        # ------------ 训练模型，保存模型 ------------
+        # 1. 如果是加强训练模式，先加载预训练模型特征参数，再预处理训练数据
+        # 2. 如果是普通模式，先预处理训练数据，再根据训练数据特征加载模型
+        model = trans.train_init() if trans.opt.Model['add_train'] else trans.get_transformer_model(trans.opt, time_series=args['time_series'], lstm_type=1)
+        if trans.opt.Model['add_train']:
+            if model:
+                feas = json.loads(trans.model_params)['features']
+                if set(feas).issubset(set(dh.opt.features)):
+                    dh.opt.features = list(feas)
+                    train_x, train_y, valid_x, valid_y, scaled_train_bytes, scaled_target_bytes, scaled_cap = dh.train_data_handler(train_data, time_series=args['time_series'])
+                else:
+                    model = trans.get_transformer_model(trans.opt, time_series=args['time_series'], lstm_type=1)
+                    logger.info("训练数据特征，不满足，加强训练模型特征")
+            else:
+                model = trans.get_transformer_model(trans.opt, time_series=args['time_series'], lstm_type=1)
+        ts_model = trans.training(model, [train_x, train_y, valid_x, valid_y])
+        args['Model']['features'] = ','.join(dh.opt.features)
+        args['params'] = json.dumps(args)
+        args['descr'] = '测试'
+        args['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
+
+        insert_trained_model_into_mongo(ts_model, args)
+        insert_scaler_model_into_mongo(scaled_train_bytes, scaled_target_bytes, args)
+        success = 1
+    except Exception as e:
+        my_exception = traceback.format_exc()
+        my_exception.replace("\n", "\t")
+        result['msg'] = my_exception
+    end_time = time.time()
+    result['success'] = success
+    result['args'] = args
+    result['start_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))
+    result['end_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))
+    print("Program execution ends!")
+    return result
+
+
+if __name__ == "__main__":
+    print("Program starts execution!")
+    from waitress import serve
+    serve(app, host="0.0.0.0", port=10115,
+          threads=8,  # 指定线程数（默认4，根据硬件调整）
+          channel_timeout=600  # 连接超时时间（秒）
+          )
+    print("server start!")
+    # args_dict = {"mongodb_database": 'realtimeDq', 'scaler_table': 'j00600_scaler', 'model_name': 'lstm1',
+    # 'model_table': 'j00600_model', 'mongodb_read_table': 'j00600', 'col_time': 'dateTime',
+    # 'features': 'speed10,direction10,speed30,direction30,speed50,direction50,speed70,direction70,speed90,direction90,speed110,direction110,speed150,direction150,speed170,direction170'}
+    # args_dict['features'] = args_dict['features'].split(',')
+    # args.update(args_dict)
+    # dh = DataHandler(logger, args)
+    # ts = TSHandler(logger, args)
+    # opt = argparse.Namespace(**args)
+    # opt.Model['input_size'] = len(opt.features)
+    # train_data = get_data_from_mongo(args_dict)
+    # train_x, train_y, valid_x, valid_y, scaled_train_bytes, scaled_target_bytes = dh.train_data_handler(train_data)
+    # ts_model = ts.training([train_x, train_y, valid_x, valid_y])
+    #
+    # args_dict['gen_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
+    # args_dict['params'] = args
+    # args_dict['descr'] = '测试'
+    # insert_trained_model_into_mongo(ts_model, args_dict)
+    # insert_scaler_model_into_mongo(scaled_train_bytes, scaled_target_bytes, args_dict)