Merge branch 'dev_david' of anweiguo/algorithm_platform into dev_awg

data_processing/data_operation/data_handler.py

@@ -234,6 +234,8 @@ class DataHandler(object):
         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()
         if self.opt.Model['predict_data_fill']:
             pre_data = self.fill_pre_data(pre_data)

models_processing/model_tf/bp.yaml

@@ -36,7 +36,6 @@ full_field: true
 history_hours: 1
 new_field: true
 features:
-- time
 - temperature10
 - temperature190
 - direction160

models_processing/model_tf/cnn.yaml

@@ -36,7 +36,6 @@ full_field: true
 history_hours: 1
 new_field: true
 features:
-- time
 - temperature10
 - temperature190
 - direction160

models_processing/model_tf/test.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python
+# -*- coding:utf-8 -*-
+# @FileName  :test.py
+# @Time      :2025/3/25 09:16
+# @Author    :David
+# @Company: shenyang JY
+
+{"features": ["temperature190", "temperature10", "direction160", "direction40", "temperature110",
+              "speed60", "direction80", "mcc", "temperature150", "speed20", "speed110",
+              "globalr", "solarZenith", "speed190", "direction120", "direction200",
+              "temperature90", "speed150", "temperature50", "direction30",
+              "temperature160", "direction170", "temperature20",
+              "direction70", "direction130", "temperature200", "speed70", "temperature120",
+              "speed30", "speed100", "speed80", "speed180", "dniCalcd", "speed140",
+              "temperature60", "temperature170", "temperature30", "direction20",
+              "humidity2", "direction180", "direction60", "direction140", "hcc", "speed40",
+              "clearskyGhi", "temperature130", "lcc", "speed90", "tcc", "temperature2",
+              "speed170", "direction100", "temperature70", "speed130", "direction190", "temperature40",
+              "direction10", "temperature180", "direction150", "direction50", "speed50", "direction90",
+              "temperature100", "speed10", "temperature140", "speed120", "speed200", "radiation", "tpr",
+              "surfacePressure", "direction110", "speed160", "temperature80"]}
+if __name__ == "__main__":
+    run_code = 0

models_processing/model_tf/tf_bp_pre.py

@@ -60,12 +60,12 @@ def model_prediction_bp():
         bp.opt.cap = round(target_scaler.transform(np.array([[float(args['cap'])]]))[0, 0], 2)
         # ------------ 获取模型，预测结果------------
         bp.get_model(args)
-        dh.opt.features = json.loads(bp.model_params).get('Model').get('features', ','.join(bp.opt.features)).split(',')
+        dh.opt.features = json.loads(bp.model_params)['Model']['features'].split(',')
         scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler, bp_data=True)
 
         res = list(chain.from_iterable(target_scaler.inverse_transform(bp.predict(scaled_pre_x))))
         pre_data['farm_id'] = args.get('farm_id', 'null')
-        if args.get('algorithm_test', 0):
+        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']]
@@ -80,7 +80,7 @@ def model_prediction_bp():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
 
         pre_data = pre_data[res_cols]
-        pre_data['power_forecast'] = pre_data['power_forecast'].round(2)
+        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
         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_bp_train.py

@@ -57,7 +57,7 @@ def model_training_bp():
         model = bp.train_init() if bp.opt.Model['add_train'] else bp.get_keras_model(bp.opt)
         if bp.opt.Model['add_train']:
             if model:
-                feas = json.loads(bp.model_params).get('features', args['features'])
+                feas = json.loads(bp.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)

models_processing/model_tf/tf_cnn_pre.py

@@ -60,13 +60,13 @@ def model_prediction_bp():
         cnn.opt.cap = round(target_scaler.transform(np.array([[float(args['cap'])]]))[0, 0], 2)
 
         cnn.get_model(args)
-        dh.opt.features = json.loads(cnn.model_params).get('Model').get('features', ','.join(cnn.opt.features)).split(',')
+        dh.opt.features = json.loads(cnn.model_params)['Model']['features'].split(',')
         scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler)
         logger.info("---------cap归一化：{}".format(cnn.opt.cap))
 
         res = list(chain.from_iterable(target_scaler.inverse_transform(cnn.predict(scaled_pre_x))))
         pre_data['farm_id'] = args.get('farm_id', 'null')
-        if args.get('algorithm_test', 0):
+        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']]
@@ -81,7 +81,7 @@ def model_prediction_bp():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data['power_forecast'] = pre_data['power_forecast'].round(2)
+        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
         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_train.py

@@ -59,7 +59,7 @@ def model_training_bp():
         model = cnn.train_init() if cnn.opt.Model['add_train'] else cnn.get_keras_model(cnn.opt)
         if cnn.opt.Model['add_train']:
             if model:
-                feas = json.loads(cnn.model_params).get('features', args['features'])
+                feas = json.loads(cnn.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)

models_processing/model_tf/tf_lstm_pre.py

@@ -59,11 +59,11 @@ def model_prediction_bp():
         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).get('Model').get('features', ','.join(ts.opt.features)).split(',')
+        dh.opt.features = json.loads(ts.model_params)['Model']['features'].split(',')
         scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler)
         res = list(chain.from_iterable(target_scaler.inverse_transform(ts.predict(scaled_pre_x))))
         pre_data['farm_id'] = args.get('farm_id', 'null')
-        if args.get('algorithm_test', 0):
+        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']]
@@ -78,7 +78,7 @@ def model_prediction_bp():
             res_cols = ['date_time', 'power_forecast', 'farm_id']
         pre_data = pre_data[res_cols]
 
-        pre_data['power_forecast'] = pre_data['power_forecast'].round(2)
+        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
         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_train.py

@@ -56,7 +56,7 @@ def model_training_bp():
         model = ts.train_init() if ts.opt.Model['add_train'] else ts.get_keras_model(ts.opt)
         if ts.opt.Model['add_train']:
             if model:
-                feas = json.loads(ts.model_params).get('features', args['features'])
+                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)

models_processing/model_tf/tf_test_pre.py

@@ -60,12 +60,12 @@ def model_prediction_test():
         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).get('Model').get('features', ','.join(ts.opt.features)).split(',')
+        dh.opt.features = json.loads(ts.model_params)['Model']['features'].split(',')
         scaled_pre_x, pre_data = dh.pre_data_handler(pre_data, feature_scaler)
 
         res = list(chain.from_iterable(target_scaler.inverse_transform(ts.predict(scaled_pre_x))))
         pre_data['farm_id'] = args.get('farm_id', 'null')
-        if args.get('algorithm_test', 0):
+        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']]
@@ -79,7 +79,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['power_forecast'] = pre_data['power_forecast'].round(2)
+        pre_data.loc[:, 'power_forecast'] = pre_data['power_forecast'].round(2)
         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)

models_processing/model_tf/tf_test_train.py

@@ -56,7 +56,7 @@ def model_training_test():
         model = ts.train_init() if ts.opt.Model['add_train'] else ts.get_keras_model(ts.opt)
         if ts.opt.Model['add_train']:
             if model:
-                feas = json.loads(ts.model_params).get('features', args['features'])
+                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(