Merge branch 'dev_awg' of http://git.jiayuepowertech.com:9022/anweiguo/algorithm_platform into dev_awg

data_processing/data_operation/data_handler.py

@@ -226,6 +226,7 @@ class DataHandler(object):
         # 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.applymap(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)
         pre_data = data[features]
         scaled_features = feature_scaler.transform(data[features])

data_processing/data_operation/pre_prod_ftp.py

@@ -7,7 +7,19 @@
 
 """
 要实现的功能：
-1. 获取场站-配置接口，根据 / 算法工程师 / 模型 版本 日期 / 站名
+1. 获取场站-配置接口，根据：
+类型（超短期 短期 中期）
+    算法工程师 / 模型 版本
+        场站编码
+获取所有当前配置的场站模型
+2. 根据场站模型配置和时刻，形成：
+类型（超短期 短期 中期）
+    场站编码集合
+其中，通过算法工程师 / 模型 版本从测试FTP上获取场站编码集合的步骤如下：
+（1）遍历算法工程师下的 模型 版本 zip文件，根据当前时刻，从FTP上下载， 解压
+（2）根据 模型 版本 下的 场站编码 依次获取，并组装成这个算法工程师的场站编码集合
+（3）合并所有算法工程是的场站编码集合，形成类型下的场站编码集合
+3. 压缩成类型（超短期 短期 中期）三个zip文件，上传生产FTP
 """
 
 if __name__ == "__main__":

models_processing/model_tf/losses.py

@@ -10,7 +10,7 @@ from typeguard import typechecked
 tf.compat.v1.set_random_seed(1234)
 
 
-class Rmse(tf.keras.losses.Loss):
+class MSE(tf.keras.losses.Loss):
     """
     自定义损失函数模板
     功能：实现名称设置、参数保存、张量运算分离
@@ -36,6 +36,39 @@ class Rmse(tf.keras.losses.Loss):
         return loss
 
 
+class RMSE(Loss):
+    """Root Mean Squared Error 损失函数（兼容单变量/多变量回归）"""
+
+    def __init__(self,
+                 name,
+                 reduction="sum_over_batch_size",  # 默认自动选择 'sum_over_batch_size' (等效于 mean)
+                 **kwargs):
+        super().__init__(name=name, reduction=reduction)
+
+    def call(self, y_true, y_pred):
+        """
+        计算逻辑：
+        1. 计算预测值与真实值的平方差
+        2. 沿最后一个维度（特征维度）取均值
+        3. 对每个样本的损失取平方根
+        """
+        # 确保输入张量形状兼容（如 squeeze 冗余维度）
+        y_pred = tf.convert_to_tensor(y_pred)
+        y_true = tf.cast(y_true, y_pred.dtype)
+
+        # 计算均方误差（沿最后一个轴取平均）
+        mse_per_sample = tf.reduce_mean(tf.square(y_pred - y_true), axis=-1)
+
+        # 对每个样本的 MSE 取平方根得到 RMSE
+        rmse_per_sample = tf.sqrt(mse_per_sample)
+
+        return rmse_per_sample
+
+    def get_config(self):
+        """支持序列化配置（用于模型保存/加载）"""
+        base_config = super().get_config()
+        return base_config
+
 class SouthLoss(Loss):
     """
     南网新规则损失函数（支持完整序列化）
@@ -46,8 +79,8 @@ class SouthLoss(Loss):
         reduction (str): 损失归约方式，默认为 'sum_over_batch_size'
 
     示例:
-        >>> loss = SouthLoss(cap=0.5)
-        >>> model.compile(loss=loss, optimizer='adam')
+        # >>> loss = SouthLoss(cap=0.5)
+        # >>> model.compile(loss=loss, optimizer='adam')
     """
 
     @typechecked
@@ -137,7 +170,7 @@ class SouthLoss(Loss):
 
 
 region_loss_d = {
-    'northeast': lambda region: Rmse(region),
+    'northeast': lambda region: RMSE(region),
     'south': lambda cap, region: SouthLoss(cap, region)
 }
 
@@ -153,4 +186,37 @@ def region_loss(opt):
         else:
             return handler(opt.region)
     else:
-        raise TypeError("无效的损失函数")
+        raise TypeError("无效的损失函数")
+
+if __name__ == '__main__':
+    # 测试数据
+    y_true = tf.constant([[1.0], [2.0], [3.0]], dtype=tf.float32)
+    y_pred = tf.constant([[1.5], [2.5], [3.5]], dtype=tf.float32)
+
+    # 标准 MSE
+    mse = tf.keras.losses.MeanSquaredError()(y_true, y_pred).numpy()
+
+    # 自定义损失（权重=1时等效于MSE）
+    custom_mse = MSE(name='test')(y_true, y_pred).numpy()
+
+    print("标准 MSE:", mse)  # 输出: 0.25
+    print("自定义 MSE:", custom_mse)  # 应输出: 0.25
+    assert abs(mse - custom_mse) < 1e-6
+
+    # 定义变量和优化器
+    y_pred_var = tf.Variable([[1.5], [2.5], [3.5]], dtype=tf.float32)
+    optimizer = tf.keras.optimizers.Adam()
+
+    with tf.GradientTape() as tape:
+        loss = MSE(name='test')(y_true, y_pred_var)
+    grads = tape.gradient(loss, y_pred_var)
+
+    # 理论梯度公式：2*(y_pred - y_true)/N （N=3）
+    expected_grads = 2 * (y_pred_var - y_true) / 3.0
+    print("实际梯度:\n", grads.numpy())
+    print("理论梯度:\n", expected_grads.numpy())
+
+    with tf.GradientTape() as tape:
+        loss = tf.keras.losses.MeanSquaredError()(y_true, y_pred_var)
+    grads = tape.gradient(loss, y_pred_var)
+    print("实际梯度1:\n", grads.numpy())

models_processing/model_tf/tf_bp.py

@@ -31,8 +31,8 @@ class BPHandler(object):
         """
         try:
             with model_lock:
-                loss = region_loss(self.opt)
-                self.model = get_h5_model_from_mongo(args, {type(loss).__name__: loss})
+                # loss = region_loss(self.opt)
+                self.model = get_h5_model_from_mongo(args)
         except Exception as e:
             self.logger.info("加载模型权重失败:{}".format(e.args))
 
@@ -54,7 +54,7 @@ class BPHandler(object):
     @staticmethod
     def get_keras_model(opt):
         # 自定义损失函数（需确保正确性）
-        loss = region_loss(opt)
+        # loss = region_loss(opt)
 
         # 网络结构
         model = Sequential([
@@ -81,7 +81,7 @@ class BPHandler(object):
             epsilon=1e-6
         )
 
-        model.compile(loss=loss, optimizer=adam)
+        model.compile(loss=tf.keras.losses.MeanSquaredError(), optimizer=adam)
         return model
 
     def train_init(self):
@@ -100,7 +100,6 @@ class BPHandler(object):
 
     def training(self, train_and_valid_data):
         model = self.train_init()
-        # tf.reset_default_graph() # 清除默认图
         train_x, train_y, valid_x, valid_y = train_and_valid_data
         print("----------", np.array(train_x[0]).shape)
         print("++++++++++", np.array(train_x[1]).shape)

models_processing/model_tf/tf_bp_pre.py

@@ -65,6 +65,9 @@ def model_prediction_bp():
             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['cdq'] = args.get('cdq', 1)
             pre_data['dq'] = args.get('dq', 1)

models_processing/model_tf/tf_cnn_pre.py

@@ -65,6 +65,9 @@ def model_prediction_bp():
             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['cdq'] = args.get('cdq', 1)
             pre_data['dq'] = args.get('dq', 1)

models_processing/model_tf/tf_lstm_pre.py

@@ -64,6 +64,9 @@ def model_prediction_bp():
             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['cdq'] = args.get('cdq', 1)
             pre_data['dq'] = args.get('dq', 1)