Merge branch 'dev_david' of anweiguo/algorithm_platform into dev_awg

common/data_cleaning.py

@@ -31,7 +31,7 @@ def data_column_cleaning(data, logger, clean_value=[-99.0, -99]):
     for val in clean_value:
         data1 = data1.replace(val, np.nan)
     # nan 列超过80% 删除
-    data1 = data1.dropna(axis=1, thresh=len(data) * 0.8)
+    data1 = data1.dropna(axis=1, thresh=len(data) * 0.5)
     # 删除取值全部相同的列
     data1 = data1.loc[:, (data1 != data1.iloc[0]).any()]
     data = data[data1.columns.tolist()]

data_processing/data_operation/custom_data_handler.py

@@ -262,3 +262,237 @@ class CustomDataHandler(object):
         pre_data.loc[:, features] = scaled_pre_data
         pre_x = self.get_predict_data([pre_data], time_series)
         return pre_x, data
+
+class MultiNwpDataHandler(object):
+    def __init__(self, logger, args):
+        self.logger = logger
+        self.opt = argparse.Namespace(**args)
+
+    def get_train_data(self, dfs, col_time, target, time_series=1):
+        train_x, valid_x, train_y, valid_y = [], [], [], []
+        for i, df in enumerate(dfs, start=1):
+            if len(df) < self.opt.Model["time_step"]:
+                self.logger.info("特征处理-训练数据-不满足time_step")
+
+            datax, datay = self.get_timestep_features(df, col_time, target, is_train=True, time_series=time_series)
+            if len(datax) < 10:
+                self.logger.info("特征处理-训练数据-无法进行最小分割")
+                continue
+            tx, vx, ty, vy = self.train_valid_split(datax, datay, valid_rate=self.opt.Model["valid_data_rate"], shuffle=self.opt.Model['shuffle_train_data'])
+            train_x.extend(tx)
+            valid_x.extend(vx)
+            train_y.extend(ty)
+            valid_y.extend(vy)
+
+        train_x = [np.array([x[0].values for x in train_x]), np.array([x[1].values for x in train_x])]
+        valid_x = [np.array([x[0].values for x in valid_x]), np.array([x[1].values for x in valid_x])]
+        train_y = np.concatenate(np.array([[y.iloc[:, 1].values for y in train_y]]), axis=0)
+        valid_y = np.concatenate(np.array([[y.iloc[:, 1].values for y in valid_y]]), axis=0)
+
+        return train_x, valid_x, train_y, valid_y
+
+    def get_predict_data(self, dfs, time_series=1):
+        test_x = []
+        for i, df in enumerate(dfs, start=1):
+            if len(df) < self.opt.Model["time_step"]*time_series:
+                self.logger.info("特征处理-预测数据-不满足time_step")
+                continue
+            datax = self.get_predict_features(df, time_series)
+            test_x.append(datax)
+        test_x = np.concatenate(test_x, axis=0)
+        return test_x
+
+    def get_predict_features(self, norm_data, time_series=1):
+        """
+        均分数据，获取预测数据集
+        """
+        time_step = self.opt.Model["time_step"]
+        feature_data = norm_data.loc[:, self.opt.features].reset_index(drop=True)
+        time_step *= int(time_series)
+        time_step_loc = time_step - 1
+        iters = int(len(feature_data)) // time_step
+        end = int(len(feature_data)) % time_step
+        features_x = np.array([feature_data.loc[i*time_step:i*time_step + time_step_loc, self.opt.features1].reset_index(drop=True) for i in range(iters)])
+        features_x1 = [feature_data.loc[i*time_step:i*time_step + time_step_loc, self.opt.features2].reset_index(drop=True) for i in range(iters)]
+
+        if end > 0:
+            df = feature_data.tail(end)
+            df_repeated = pd.concat([df] + [pd.DataFrame([df.iloc[-1]]* (time_step-end))]).reset_index(drop=True)
+            features_x = np.concatenate((features_x, np.expand_dims(df_repeated, 0)), axis=0)
+        return features_x
+
+    def get_timestep_features(self, norm_data, col_time, target, is_train, time_series):
+        """
+        步长分割数据，分区建模
+        """
+        time_step = self.opt.Model["time_step"]
+        feature_data = norm_data.reset_index(drop=True)
+        time_step_loc = time_step*time_series - 1
+        train_num = int(len(feature_data))
+        label_features_power = [col_time, target] if is_train is True else [col_time, target]
+        nwp_cs = self.opt.features1
+        nwp_cs1 = self.opt.features2
+        nwp = [feature_data.loc[i:i + time_step_loc, nwp_cs].reset_index(drop=True) for i in range(train_num - time_step*time_series + 1)]
+        nwp1 = [feature_data.loc[i:i + time_step_loc, nwp_cs1].reset_index(drop=True) for i in range(train_num - time_step*time_series + 1)]
+
+        labels_power = [feature_data.loc[i:i + time_step_loc, label_features_power].reset_index(drop=True) for i in range(train_num - time_step*time_series + 1)]
+        features_x, features_y = [], []
+        for i, row in enumerate(zip(nwp, nwp1, labels_power)):
+            features_x.append([row[0], row[1]])
+            features_y.append(row[2])
+        return features_x, features_y
+
+    def fill_train_data(self, unite, col_time):
+        """
+        补值
+        """
+        unite[col_time] = pd.to_datetime(unite[col_time])
+        unite['time_diff'] = unite[col_time].diff()
+        dt_short = pd.Timedelta(minutes=15)
+        dt_long = pd.Timedelta(minutes=15 * self.opt.Model['how_long_fill'])
+        data_train = self.missing_time_splite(unite, dt_short, dt_long, col_time)
+        miss_points = unite[(unite['time_diff'] > dt_short) & (unite['time_diff'] < dt_long)]
+        miss_number = miss_points['time_diff'].dt.total_seconds().sum(axis=0) / (15 * 60) - len(miss_points)
+        self.logger.info("再次测算，需要插值的总点数为：{}".format(miss_number))
+        if miss_number > 0 and self.opt.Model["train_data_fill"]:
+            data_train = self.data_fill(data_train, col_time)
+        return data_train
+
+    def fill_pre_data(self, unite):
+        unite = unite.interpolate(method='linear')  # nwp先进行线性填充
+        unite = unite.ffill().bfill() # 再对超过采样边缘无法填充的点进行二次填充
+        return unite
+
+    def missing_time_splite(self, df, dt_short, dt_long, col_time):
+        df.reset_index(drop=True, inplace=True)
+        n_long, n_short, n_points = 0, 0, 0
+        start_index = 0
+        dfs = []
+        for i in range(1, len(df)):
+            if df['time_diff'][i] >= dt_long:
+                df_long = df.iloc[start_index:i, :-1]
+                dfs.append(df_long)
+                start_index = i
+                n_long += 1
+            if df['time_diff'][i] > dt_short:
+                self.logger.info(f"{df[col_time][i-1]} ~ {df[col_time][i]}")
+                points = df['time_diff'].dt.total_seconds()[i]/(60*15)-1
+                self.logger.info("缺失点数：{}".format(points))
+                if df['time_diff'][i] < dt_long:
+                    n_short += 1
+                    n_points += points
+                    self.logger.info("需要补值的点数：{}".format(points))
+        dfs.append(df.iloc[start_index:, :-1])
+        self.logger.info(f"数据总数：{len(df)}, 时序缺失的间隔：{n_short}, 其中，较长的时间间隔：{n_long}")
+        self.logger.info("需要补值的总点数：{}".format(n_points))
+        return dfs
+
+    def data_fill(self, dfs, col_time, test=False):
+        dfs_fill, inserts = [], 0
+        for i, df in enumerate(dfs):
+            df = rm_duplicated(df, self.logger)
+            df1 = df.set_index(col_time, inplace=False)
+            dff = df1.resample('15T').interpolate(method='linear')  # 采用线性补值，其他补值方法需要进一步对比
+            dff.reset_index(inplace=True)
+            points = len(dff) - len(df1)
+            dfs_fill.append(dff)
+            self.logger.info("{} ~ {} 有 {} 个点, 填补 {} 个点.".format(dff.iloc[0, 0], dff.iloc[-1, 0], len(dff), points))
+            inserts += points
+        name = "预测数据" if test is True else "训练集"
+        self.logger.info("{}分成了{}段，实际一共补值{}点".format(name, len(dfs_fill), inserts))
+        return dfs_fill
+
+    def train_valid_split(self, datax, datay, valid_rate, shuffle):
+        shuffle_index = np.random.permutation(len(datax))
+        indexs = shuffle_index.tolist() if shuffle else np.arange(0, len(datax)).tolist()
+        valid_size = int(len(datax) * valid_rate)
+        valid_index = indexs[-valid_size:]
+        train_index = indexs[:-valid_size]
+        tx, vx, ty, vy = [], [], [], []
+        for i, data in enumerate(zip(datax, datay)):
+            if i in train_index:
+                tx.append(data[0])
+                ty.append(data[1])
+            elif i in valid_index:
+                vx.append(data[0])
+                vy.append(data[1])
+        return tx, vx, ty, vy
+
+    def train_data_handler(self, data, time_series=1):
+        """
+        训练数据预处理：
+        清洗+补值+归一化
+        Args:
+            data: 从mongo中加载的数据
+            opt：参数命名空间
+        return:
+            x_train
+            x_valid
+            y_train
+            y_valid
+        """
+        col_time, features1, features2, target = self.opt.col_time, self.opt.features1, self.opt.features2, self.opt.target
+        # 清洗限电记录
+        if 'is_limit' in data.columns:
+            data = data[data['is_limit'] == False]
+        # 筛选特征，数值化，排序
+        train_data = data[[col_time] + features1 + features2 + [target]]
+        train_data = train_data.applymap(lambda x: float(x.to_decimal()) if isinstance(x, Decimal128) else float(x) if isinstance(x, numbers.Number) else x)
+        train_data = train_data.sort_values(by=col_time)
+        # 清洗特征平均缺失率大于20%的天
+        # train_data = missing_features(train_data, features, col_time)
+        # 对清洗完限电的数据进行特征预处理：
+        # 1.空值异常值清洗
+        train_data_cleaned = cleaning(train_data, '训练集', self.logger, features1 + features2 + [target], col_time)
+        self.opt.features = [x for x in train_data_cleaned.columns.tolist() if x not in [target, col_time] and x in features1+features2]
+        # 2. 标准化
+        # 创建特征和目标的标准化器
+        train_scaler = MinMaxScaler(feature_range=(0, 1))
+        target_scaler = MinMaxScaler(feature_range=(0, 1))
+        # 标准化特征和目标
+        scaled_train_data = train_scaler.fit_transform(train_data_cleaned[self.opt.features])
+        scaled_target = target_scaler.fit_transform(train_data_cleaned[[target]])
+        scaled_cap = target_scaler.transform(np.array([[float(self.opt.cap)]]))[0,0]
+        train_data_cleaned[self.opt.features] = scaled_train_data
+        train_data_cleaned[[target]] = scaled_target
+        # 3.缺值补值
+        train_datas = self.fill_train_data(train_data_cleaned, col_time)
+        # 保存两个scaler
+        scaled_train_bytes = BytesIO()
+        scaled_target_bytes = BytesIO()
+        joblib.dump(train_scaler, scaled_train_bytes)
+        joblib.dump(target_scaler, scaled_target_bytes)
+        scaled_train_bytes.seek(0)  # Reset pointer to the beginning of the byte stream
+        scaled_target_bytes.seek(0)
+
+        train_x, valid_x, train_y, valid_y = self.get_train_data(train_datas, col_time, target, time_series)
+        return train_x, train_y, valid_x, valid_y, scaled_train_bytes, scaled_target_bytes, scaled_cap
+
+    def pre_data_handler(self, data, feature_scaler, time_series=1):
+        """
+        预测数据简单处理
+        Args:
+            data: 从mongo中加载的数据
+            opt：参数命名空间
+        return:
+            scaled_features: 反归一化的特征
+        """
+        # 清洗限电记录
+        if 'is_limit' in data.columns:
+            data = data[data['is_limit'] == False]
+        # features, time_steps, col_time, model_name, col_reserve = str_to_list(args['features']), int(
+        #     args['time_steps']), args['col_time'], args['model_name'], str_to_list(args['col_reserve'])
+        col_time, features = self.opt.col_time, self.opt.features
+        data = data.map(lambda x: float(x.to_decimal()) if isinstance(x, Decimal128) else float(x) if isinstance(x, numbers.Number) else x)
+        data = data.sort_values(by=col_time).reset_index(drop=True, inplace=False)
+        if not set(features).issubset(set(data.columns.tolist())):
+            raise ValueError("预测数据特征不满足模型特征！")
+        pre_data = data[features].copy()
+        pre_data[self.opt.zone] = 1
+        if self.opt.Model['predict_data_fill']:
+            pre_data = self.fill_pre_data(pre_data)
+        scaled_pre_data = feature_scaler.transform(pre_data)[:, :len(features)]
+        pre_data.drop(columns=self.opt.zone, inplace=True)
+        pre_data.loc[:, features] = scaled_pre_data
+        pre_x = self.get_predict_data([pre_data], time_series)
+        return pre_x, data

models_processing/model_tf/losses.py

@@ -115,8 +115,8 @@ class SouthLoss(Loss):
                  name: str = "south_loss",
                  reduction: str = "sum_over_batch_size"):
         # 参数校验
-        # if not 0 <= cap <= 1:
-        #     raise ValueError("cap 必须为归一化后的值且位于 [0,1] 区间")
+        if not 0 <= cap <= 1:
+            raise ValueError("cap 必须为归一化后的值且位于 [0,1] 区间")
 
         super().__init__(name=name, reduction=reduction)
 

models_processing/model_tf/tf_bp_pre.py

@@ -83,7 +83,7 @@ def model_prediction_bp():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
 
         pre_data = pre_data[res_cols]
-        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
+        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
 

models_processing/model_tf/tf_cnn_pre.py

@@ -84,7 +84,7 @@ def model_prediction_cnn():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
+        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
 

models_processing/model_tf/tf_lstm2_pre.py

@@ -83,7 +83,7 @@ def model_prediction_lstm2():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
+        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
 

models_processing/model_tf/tf_lstm3_pre.py

@@ -87,7 +87,7 @@ def model_prediction_lstm3():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
+        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
 

models_processing/model_tf/tf_lstm_pre.py

@@ -82,7 +82,7 @@ def model_prediction_lstm():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
+        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
 

models_processing/model_tf/tf_lstm_zone_pre.py

@@ -82,7 +82,7 @@ def model_prediction_lstm():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
+        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
 

models_processing/model_tf/tf_multi_nwp_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_koi import *
+from common.processing_data_common import missing_features, str_to_list
+from data_processing.data_operation.custom_data_handler import CustomDataHandler
+from models_processing.model_tf.tf_lstm_zone import TSHandler
+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
+# 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_zone_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)
+    request_args['zone'] = request_args['zone'].split(',')
+    current_config = deep_update(current_config, request_args)
+    # 存储到请求上下文
+    g.opt = argparse.Namespace(**current_config)
+    g.dh = CustomDataHandler(logger, current_config)  # 每个请求独立实例
+    g.ts = TSHandler(logger, current_config)
+
+@app.route('/tf_lstm_zone_predict', methods=['POST'])
+def model_prediction_lstm():
+    # 获取程序开始时间
+    start_time = time.time()
+    result = {}
+    success = 0
+    dh = g.dh
+    ts = g.ts
+    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)
+        ts.opt.cap = round(target_scaler.transform(np.array([[float(args['cap'])]]))[0, 0], 2)
+        ts.get_model(args)
+        dh.opt.features = json.loads(ts.model_params)['Model']['features'].split(',')
+        scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler, time_series=args['time_series'])
+        res = list(chain.from_iterable(target_scaler.inverse_transform(ts.predict(scaled_pre_x)[1])))
+        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=10125,
+          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_multi_nwp_train.py

@@ -0,0 +1,123 @@
+#!/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.custom_data_handler import CustomDataHandler
+import time, yaml, threading
+from copy import deepcopy
+from models_processing.model_tf.tf_lstm_zone import TSHandler
+from common.database_dml_koi 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_zone_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 = CustomDataHandler(logger, current_config)  # 每个请求独立实例
+    g.ts = TSHandler(logger, current_config)
+
+
+@app.route('/tf_lstm_zone_training', methods=['POST'])
+def model_training_lstm():
+    # 获取程序开始时间
+    start_time = time.time()
+    result = {}
+    success = 0
+    dh = g.dh
+    ts = g.ts
+    args = deepcopy(g.opt.__dict__)
+    logger.info("Program starts execution!")
+    try:
+        # ------------ 获取数据，预处理训练数据 ------------
+        train_data = get_data_from_mongo(args)
+        train_data_1 = 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'])
+        ts.opt.cap = round(scaled_cap, 2)
+        ts.opt.Model['input_size'] = len(dh.opt.features)
+        # ------------ 训练模型，保存模型 ------------
+        # 1. 如果是加强训练模式，先加载预训练模型特征参数，再预处理训练数据
+        # 2. 如果是普通模式，先预处理训练数据，再根据训练数据特征加载模型
+        model = ts.train_init() if ts.opt.Model['add_train'] else ts.get_keras_model(ts.opt, time_series=args['time_series'], lstm_type=1)
+        if ts.opt.Model['add_train']:
+            if model:
+                feas = json.loads(ts.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 = ts.get_keras_model(ts.opt, time_series=args['time_series'], lstm_type=1)
+                    logger.info("训练数据特征，不满足，加强训练模型特征")
+            else:
+                model = ts.get_keras_model(ts.opt, time_series=args['time_series'], lstm_type=1)
+        ts_model = ts.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=10124,
+          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)

models_processing/model_tf/tf_test_pre.py

@@ -82,7 +82,7 @@ def model_prediction_test():
             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['power_forecast'].round(2)
+        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)

post_processing/cdq_coe_gen.py

@@ -7,14 +7,13 @@
 import os, requests, json, time, traceback
 import pandas as pd
 import numpy as np
-from common.database_dml import get_data_from_mongo
-from pymongo import MongoClient
-from flask import Flask,request,jsonify, g
+from bayes_opt import BayesianOptimization
+from common.database_dml_koi import get_data_from_mongo
+from flask import Flask, request, g
 from datetime import datetime
-# from common.logs import Log
-# logger = Log('post-processing').logger
-from logging import getLogger
-logger = getLogger('xx')
+from common.logs import Log
+
+logger = Log('post-processing').logger
 current_path = os.path.dirname(__file__)
 API_URL = "http://ds2:18080/accuracyAndBiasByJSON"
 app = Flask('cdq_coe_gen——service')
@@ -23,25 +22,29 @@ app = Flask('cdq_coe_gen——service')
 @app.before_request
 def update_config():
     # ------------ 整理参数，整合请求参数 ------------
-    g.coe = {}
+    g.coe = {'T'+str(x):{} for x in range(1, 17)}
 
 
-def iterate_coe(pre_data, point, col_power, col_pre, coe):
+def iterate_coe_simple(pre_data, point, config, coe):
     """
     更新16个点系数
     """
     T = 'T' + str(point + 1)
+    col_pre = config['col_pre']
     best_acc, best_score1, best_coe_m, best_coe_n = 0, 0, 0, 0
-    best_score, best_acc1, best_score_m, best_score_n = 999, 0, 0, 0
-    req_his_fix = prepare_request_body(pre_data, col_power, 'his_fix')
-    req_dq = prepare_request_body(pre_data, col_power, col_pre)
+    best_score, best_acc1, best_score_m, best_score_n = 999, 0, 999, 0
+
+    pre_data = history_error(pre_data, config['col_power'], config['col_pre'], int(coe[T]['hour']//0.25))
+    pre_data = curve_limited(pre_data, config, 'his_fix')
+    req_his_fix = prepare_request_body(pre_data, config, 'his_fix')
+    req_dq = prepare_request_body(pre_data, config, col_pre)
 
     his_fix_acc, his_fix_score = calculate_acc(API_URL, req_his_fix)
     dq_acc, dq_score = calculate_acc(API_URL, req_dq)
     for i in range(5, 210):
         for j in range(5, 210):
             pre_data["new"] = round(i / 170 * pre_data[col_pre] + j / 170 * pre_data['his_fix'], 3)
-            req_new = prepare_request_body(pre_data, col_power, 'new')
+            req_new = prepare_request_body(pre_data, config, 'new')
             acc, acc_score = calculate_acc(API_URL, req_new)
 
             if acc > best_acc:
@@ -57,18 +60,10 @@ def iterate_coe(pre_data, point, col_power, col_pre, coe):
 
     pre_data["coe-acc"] = round(best_coe_m * pre_data[col_pre] + best_coe_n * pre_data['his_fix'], 3)
     pre_data["coe-ass"] = round(best_score_m * pre_data[col_pre] + best_score_n * pre_data['his_fix'], 3)
-    logger.info(
-        "1.过去{} - {}的短期的准确率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}，历史功率：{:.4f}".format(
-            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_acc, T, best_acc, his_fix_acc))
-    logger.info(
-        "2.过去{} - {}的短期的考核分：{:.4f}，自动确认系数后，{} 超短期的考核分：{:.4f}，历史功率：{:.4f}".format(
-            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_score, T, best_score1, his_fix_score))
-    logger.info(
-        "3.过去{} - {}的短期的准确率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}，历史功率：{:.4f}".format(
-            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_acc, T, best_acc1, his_fix_acc))
-    logger.info(
-        "4.过去{} - {}的短期的考核分：{:.4f}，自动确认系数后，{} 超短期的考核分：{:.4f}，历史功率：{:.4f}".format(
-            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_score, T, best_score, his_fix_score))
+    logger.info("1.过去{} - {}的短期的准确率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}，历史功率：{:.4f}".format(pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_acc, T, best_acc, his_fix_acc))
+    logger.info("2.过去{} - {}的短期的考核分：{:.4f}，自动确认系数后，{} 超短期的考核分：{:.4f}，历史功率：{:.4f}".format(pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_score, T, best_score1, his_fix_score))
+    logger.info("3.过去{} - {}的短期的准确率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}，历史功率：{:.4f}".format(pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_acc, T, best_acc1, his_fix_acc))
+    logger.info("4.过去{} - {}的短期的考核分：{:.4f}，自动确认系数后，{} 超短期的考核分：{:.4f}，历史功率：{:.4f}".format(pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_score, T, best_score, his_fix_score))
 
     coe[T]['score_m'] = round(best_score_m, 3)
     coe[T]['score_n'] = round(best_score_n, 3)
@@ -76,38 +71,124 @@ def iterate_coe(pre_data, point, col_power, col_pre, coe):
     coe[T]['acc_n'] = round(best_coe_n, 3)
     logger.info("系数轮询后，最终调整的系数为：{}".format(coe))
 
-def prepare_request_body(df, col_power, col_pre):
+
+def iterate_coe(pre_data, point, config, coe):
+    """使用贝叶斯优化进行系数寻优"""
+    T = 'T' + str(point + 1)
+    col_pre = config['col_pre']
+    col_time = config['col_time']
+
+    # 历史数据处理（保持原逻辑）
+    pre_data = history_error(pre_data, config['col_power'], config['col_pre'], int(coe[T]['hour'] // 0.25))
+    pre_data = curve_limited(pre_data, config, 'his_fix')
+    req_his_fix = prepare_request_body(pre_data, config, 'his_fix')
+    req_dq = prepare_request_body(pre_data, config, col_pre)
+
+    # 获取基准值（保持原逻辑）
+    his_fix_acc, his_fix_score = calculate_acc(API_URL, req_his_fix)
+    dq_acc, dq_score = calculate_acc(API_URL, req_dq)
+
+    # 定义贝叶斯优化目标函数
+    def evaluate_coefficients(m, n):
+        """评估函数返回准确率和考核分的元组"""
+        local_data = pre_data.copy()
+        local_data["new"] = round(m * local_data[col_pre] + n * local_data['his_fix'], 3)
+        local_data = curve_limited(local_data, config, 'new')
+        req_new = prepare_request_body(local_data, config, 'new')
+        acc, score = calculate_acc(API_URL, req_new)
+        return acc, score
+
+    # 优化准确率
+    def acc_optimizer(m, n):
+        acc, _ = evaluate_coefficients(m, n)
+        return acc
+
+    # 优化考核分
+    def score_optimizer(m, n):
+        _, score = evaluate_coefficients(m, n)
+        return -score  # 取负数因为要最大化负分即最小化原分数
+
+    # 参数空间（保持原参数范围）
+    pbounds = {
+        'm': (5 / 170, 210 / 170),  # 原始范围映射到[0.0294, 1.235]
+        'n': (5 / 170, 210 / 170)
+    }
+
+    # 执行准确率优化
+    acc_bo = BayesianOptimization(f=acc_optimizer, pbounds=pbounds, random_state=42)
+    acc_bo.maximize(init_points=70, n_iter=400) # 增大初始点和迭代次数，捕捉可能的多峰结构
+    best_acc_params = acc_bo.max['params']
+    best_coe_m, best_coe_n = best_acc_params['m'], best_acc_params['n']
+    best_acc = acc_bo.max['target']
+
+    # 执行考核分优化
+    # score_bo = BayesianOptimization(f=score_optimizer, pbounds=pbounds, random_state=42)
+    # score_bo.maximize(init_points=10, n_iter=20)
+    # best_score_params = score_bo.max['params']
+    # best_score_m, best_score_n = best_score_params['m'], best_score_params['n']
+    # best_score = -score_bo.max['target']  # 恢复原始分数
+
+    # 应用最优系数（保持原处理逻辑）
+    pre_data["coe-acc"] = round(best_coe_m * pre_data[col_pre] + best_coe_n * pre_data['his_fix'], 3)
+    # pre_data["coe-ass"] = round(best_score_m * pre_data[col_pre] + best_score_n * pre_data['his_fix'], 3)
+
+    # 记录日志（保持原格式）
+    logger.info("过去{} - {}的短期的准确率：{:.4f}，历史功率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}".format(pre_data[col_time][0], pre_data[col_time].iloc[-1], dq_acc, his_fix_acc, T, best_acc))
+
+    # 更新系数表（保持原逻辑）
+    coe[T].update({
+        # 'score_m': round(best_score_m, 3),
+        # 'score_n': round(best_score_n, 3),
+        'acc_m': round(best_coe_m, 3),
+        'acc_n': round(best_coe_n, 3)
+    })
+    logger.info("贝叶斯优化后，最终调整的系数为：{}".format(coe))
+
+def iterate_his_coe(pre_data, point, config, coe):
+    """
+    更新临近时长Δ
+    """
+    T = 'T' + str(point + 1)
+    best_acc, best_hour = 0, 1
+    for hour in np.arange(0.25, 4.25, 0.25):
+        data = pre_data.copy()
+        his_window = int(hour // 0.25)
+        pre_data_f = history_error(data, config['col_power'], config['col_pre'], his_window)
+        pre_data_f = curve_limited(pre_data_f, config, 'his_fix')
+        req_his_fix = prepare_request_body(pre_data_f, config, 'his_fix')
+        his_fix_acc, his_fix_score = calculate_acc(API_URL, req_his_fix)
+
+        if his_fix_acc > best_acc:
+            best_acc = his_fix_acc
+            best_hour = float(round(hour, 2))
+    coe[T]['hour'] = best_hour
+    logger.info(f"{T} 点的最优临近时长：{best_hour}")
+
+def prepare_request_body(df, config, predict):
     """
     准备请求体，动态保留MongoDB中的所有字段
     """
     data = df.copy()
     # 转换时间格式为字符串
-    if 'dateTime' in data.columns and isinstance(data['dateTime'].iloc[0], datetime):
-        data['dateTime'] = data['dateTime'].dt.strftime('%Y-%m-%d %H:%M:%S')
-    data['model'] = col_pre
-    # 排除不需要的字段（如果有）
-    exclude_fields = ['_id']  # 通常排除MongoDB的默认_id字段
-
-    # 获取所有字段名（排除不需要的字段）
-    available_fields = [col for col in data.columns if col not in exclude_fields]
-
-    # 转换为记录列表（保留所有字段）
-    data = data[available_fields].to_dict('records')
-
+    if config['col_time'] in data.columns and isinstance(data[config['col_time']].iloc[0], datetime):
+        data[config['col_time'] ] = data[config['col_time'] ].dt.strftime('%Y-%m-%d %H:%M:%S')
+    data['model'] = predict
+    # 保留必要的字段
+    data = data[[config['col_time'], config['col_power'], predict, 'model']].to_dict('records')
     # 构造请求体（固定部分+动态数据部分）
     request_body = {
-        "stationCode": "J00600",
-        "realPowerColumn": col_power,
-        "ablePowerColumn": col_power,
-        "predictPowerColumn": col_pre,
-        "inStalledCapacityName": 153,
+        "stationCode": config['stationCode'],
+        "realPowerColumn": config['col_power'],
+        "ablePowerColumn": config['col_power'],
+        "predictPowerColumn": predict,
+        "inStalledCapacityName": config['inStalledCapacityName'],
         "computTypeEnum": "E2",
-        "computMeasEnum": "E2",
-        "openCapacityName": 153,
-        "onGridEnergy": 0,
-        "price": 0,
-        "fault": -99,
-        "colTime": "dateTime",  #时间列名（可选，要与上面'dateTime一致'）
+        "computMeasEnum": config.get('computMeasEnum', 'E2'),
+        "openCapacityName": config['openCapacityName'],
+        "onGridEnergy": config.get('onGridEnergy', 1),
+        "price": config.get('price', 1),
+        "fault": config.get('fault', -99),
+        "colTime": config['col_time'],  #时间列名（可选，要与上面'dateTime一致'）
         # "computPowersEnum": "E4"  # 计算功率类型（可选）
         "data": data  # MongoDB数据
     }
@@ -132,32 +213,35 @@ def calculate_acc(api_url, request_body):
         if response.status_code == 200:
             acc = np.average([res['accuracy'] for res in result])
             # ass = np.average([res['accuracyAssessment'] for res in result])
-            print("111111111")
             return acc, 0
         else:
-            logger.info(f"失败：{result['status']},{result['error']}")
-            print(f"失败：{result['status']},{result['error']}")
-            print("22222222")
+            logger.info(f"{response.status_code}失败：{result['status']},{result['error']}")
     except requests.exceptions.RequestException as e:
-        print(f"API调用失败: {e}")
-        print("333333333")
+        logger.info(f"准确率接口调用失败: {e}")
         return None
 
-def history_error(data, col_power, col_pre):
+def history_error(data, col_power, col_pre, his_window):
     data['error'] =  data[col_power] - data[col_pre]
     data['error'] = data['error'].round(2)
     data.reset_index(drop=True, inplace=True)
     # 用前面5个点的平均error，和象心力相加
-    numbers = len(data) - 5
-    datas = [data.iloc[x: x+5, :].reset_index(drop=True) for x in range(0, numbers)]
-    data_error = [np.mean(d.iloc[0:5, -1]) for d in datas]
-    pad_data_error = np.pad(data_error, (5, 0), mode='constant', constant_values=0)
+    numbers = len(data) - his_window
+    datas = [data.iloc[x: x+his_window, :].reset_index(drop=True) for x in range(0, numbers)]
+    data_error = [np.mean(d.iloc[0:his_window, -1]) for d in datas]
+    pad_data_error = np.pad(data_error, (his_window, 0), mode='constant', constant_values=0)
     data['his_fix'] = data[col_pre] + pad_data_error
-    data = data.iloc[5:, :].reset_index(drop=True)
-    data.loc[data[col_pre] <= 0, ['his_fix']] = 0
-    data['dateTime'] = pd.to_datetime(data['dateTime'])
-    data = data.loc[:, ['dateTime', col_power, col_pre, 'his_fix']]
-    # data.to_csv('J01080原始数据.csv', index=False)
+    data = data.iloc[his_window:, :].reset_index(drop=True)
+    return data
+
+def curve_limited(pre_data, config, predict):
+    """
+    plant_type: 0 风 1 光
+    """
+    data = pre_data.copy()
+    col_time, cap = config['col_time'], float(config['openCapacityName'])
+    data[col_time] = pd.to_datetime(data[col_time])
+    data.loc[data[predict] < 0, [predict]] = 0
+    data.loc[data[predict] > cap, [predict]] = cap
     return data
 
 @app.route('/cdq_coe_gen', methods=['POST'])
@@ -171,10 +255,10 @@ def get_station_cdq_coe():
     try:
         args = request.values.to_dict()
         logger.info(args)
-        data = get_data_from_mongo(args).sort_values(by='dateTime', ascending=True)
-        pre_data = history_error(data, col_power='realPower', col_pre='dq')
+        data = get_data_from_mongo(args).sort_values(by=args['col_time'], ascending=True)
         for point in range(0, 16, 1):
-            iterate_coe(pre_data, point, 'realPower', 'dq', coe)
+            iterate_his_coe(data, point, args, coe)
+            iterate_coe(data, point, args, coe)
         success = 1
     except Exception as e:
         my_exception = traceback.format_exc()
@@ -203,11 +287,8 @@ if __name__ == "__main__":
     # run_code = 0
     print("Program starts execution!")
     from waitress import serve
-
-    serve(
-        app,
-        host="0.0.0.0",
-        port=10123,
-        threads=8,  # 指定线程数（默认4，根据硬件调整）
-        channel_timeout=600  # 连接超时时间（秒）
-    )
+    serve(app, host="0.0.0.0", port=10123,
+          threads=8,  # 指定线程数（默认4，根据硬件调整）
+          channel_timeout=600  # 连接超时时间（秒）
+          )
+    print("server start!")