#!/usr/bin/env python # -*- coding:utf-8 -*- # @FileName :cdq_coe_gen.py # @Time :2025/3/28 16:20 # @Author :David # @Company: shenyang JY import os, requests, json, time, traceback import pandas as pd import numpy as np from common.database_dml_koi import get_data_from_mongo from pymongo import MongoClient from flask import Flask,request,jsonify, g from datetime import datetime # from common.logs import Log # logger = Log('post-processing').logger from logging import getLogger logger = getLogger('xx') current_path = os.path.dirname(__file__) API_URL = "http://ds2:18080/accuracyAndBiasByJSON" app = Flask('cdq_coe_gen——service') @app.before_request def update_config(): # ------------ 整理参数，整合请求参数 ------------ g.coe = {} def iterate_coe(pre_data, point, col_power, col_pre, coe): """ 更新16个点系数 """ T = 'T' + str(point + 1) 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) 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') acc, acc_score = calculate_acc(API_URL, req_new) if acc > best_acc: best_acc = acc best_score1 = acc_score best_coe_m = i / 170 best_coe_n = j / 170 if acc_score < best_score: best_score = acc_score best_acc1 = acc best_score_m = i / 170 best_score_n = j / 170 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)) coe[T]['score_m'] = round(best_score_m, 3) coe[T]['score_n'] = round(best_score_n, 3) coe[T]['acc_m'] = round(best_coe_m, 3) coe[T]['acc_n'] = round(best_coe_n, 3) logger.info("系数轮询后，最终调整的系数为：{}".format(coe)) def prepare_request_body(df, col_power, col_pre): """ 准备请求体，动态保留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') # 构造请求体（固定部分+动态数据部分） request_body = { "stationCode": "J00600", "realPowerColumn": col_power, "ablePowerColumn": col_power, "predictPowerColumn": col_pre, "inStalledCapacityName": 153, "computTypeEnum": "E2", "computMeasEnum": "E2", "openCapacityName": 153, "onGridEnergy": 0, "price": 0, "fault": -99, "colTime": "dateTime", #时间列名（可选，要与上面'dateTime一致'） # "computPowersEnum": "E4" # 计算功率类型（可选） "data": data # MongoDB数据 } return request_body def calculate_acc(api_url, request_body): """ 调用API接口 """ headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } try: response = requests.post( api_url, data=json.dumps(request_body, ensure_ascii=False), headers=headers ) result = response.json() 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") except requests.exceptions.RequestException as e: print(f"API调用失败: {e}") print("333333333") return None def history_error(data, col_power, col_pre): 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) 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) return data @app.route('/cdq_coe_gen', methods=['POST']) def get_station_cdq_coe(): # 获取程序开始时间 start_time = time.time() result = {} success = 0 args = {} coe = g.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') for point in range(0, 16, 1): iterate_coe(pre_data, point, 'realPower', 'dq', coe) success = 1 except Exception as e: my_exception = traceback.format_exc() my_exception.replace("\n", "\t") result['msg'] = my_exception logger.info("调系数出错：{}".format(my_exception)) end_time = time.time() result['success'] = success result['args'] = args result['coe'] = coe 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)) return result if __name__ == "__main__": # args = { # 'mongodb_database': 'ldw_ftp', # 'mongodb_read_table': 'j00600', # # 'timeBegin': '2025-01-01 00:00:00', # # 'timeEnd': '2025-01-03 23:45:00' # } # data = get_data_from_mongo(args).sort_values(by='dateTime', ascending=True) # pre_data = history_error(data, col_power='realPower', col_pre='dq') # for point in range(0, 16, 1): # iterate_coe(pre_data, point, 'realPower', 'dq') # 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 # 连接超时时间（秒） )