algorithm_platform/post_processing/cdq_coe_gen.py

#!/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 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
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 = {'T'+str(x):{} for x in range(1, 17)}


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, 999, 0

    pre_data = history_error(pre_data, config['col_power'], config['col_pre'], int(coe[T]['hour']//0.25))
    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, config, '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 iterate_coe(pre_data, point, config, coe):
    """使用贝叶斯优化进行系数寻优"""
    T = 'T' + str(point + 1)
    col_pre = config['col_pre']

    # 初始化最优解存储变量
    best_acc, best_score = 0, 999
    best_coe_m, best_coe_n = 0, 0
    best_score_m, best_score_n = 0, 0

    # 历史数据处理（保持原逻辑）
    pre_data = history_error(pre_data, config['col_power'], config['col_pre'], int(coe[T]['hour'] // 0.25))
    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)
        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=10, n_iter=20)
    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("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_score, his_fix_score))

    # 更新系数表（保持原逻辑）
    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)
        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'] = predict
    # 排除不需要的字段（如果有）
    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": config['stationCode'],
        "realPowerColumn": config['col_power'],
        "ablePowerColumn": config['col_power'],
        "predictPowerColumn": predict,
        "inStalledCapacityName": config['inStalledCapacityName'],
        "computTypeEnum": "E2",
        "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数据
    }

    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])
            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, 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) - 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)
    print("!!!", his_window, numbers, len(pad_data_error))
    print("???", len(data), len(pad_data_error))
    data['his_fix'] = data[col_pre] + pad_data_error
    data = data.iloc[his_window:, :].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=args['col_time'], ascending=True)
        for point in range(0, 16, 1):
            iterate_his_coe(data, point, args, coe)
            iterate_coe(data, point, args, 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  # 连接超时时间（秒）
          )
    print("server start!")