solar3-tf1/cache/formula.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# time: 2024/9/23 15:27
# file: formula.py
# author: David
# company: shenyang JY
import numpy as np
import pandas as pd
'''
准确率formula.py模块使用手顺：
① Assessment: 考核类 Formular: 准确率计算类
② 初始化并传参：初始化考核类，传递opt命名空间 (包括cap装机容量)、logger日志对象（也可以用print代替）
③ 调用方法：按照考核类标注的方法和参数含义，进行传参和调用，获得日期平均准确率和考核分数
----------
注意：
regulations > solar/wind []列表含义（按顺序）：短期准确率考核线，超短期准确率考核线，短期考核系数，超短期考核系数
西北，东北地区短期采用考核积分电量和偏差带进行考核，这里不进行计算
'''
class Assessment(object):
    def __init__(self, opt, logger):
        self.logger = logger
        self.formula = Formulas(opt)
        self.regulations = {
            "south129":{
                'func': self.formula.calculate_acc,
                'solar': [65, 70, 1 * 0.2, 1 * 0.2],
                'wind': [60, 65, 0.2 * 2, 0.2 * 2],
                'check': True
            },
            "south": {
                'func': self.formula.calculate_acc_south,
                'solar': [65, 70, 1*0.2, 1*0.2],
                'wind': [60, 65, 0.2*2, 0.2*2],
                'check': True
            },
            "east":{
                'func': self.formula.calculate_acc,
                'solar': ['', 97, '', 0.09*1],
                'wind': ['', 96, '', 0.09*1],
                'check': False
            },
            "northeast":{
                'func': self.formula.calculate_acc_northeast,
                'solar': ['', 85, '', 0.02*0.1],
                'wind': ['', 80, '', 0.02*0.1]
            },
            "northwest":{
                'func': self.formula.calculate_acc_northwest,
                'solar': ['', 75, '', 0.015*0.1*0.5],
                'wind': ['', 75, '', 0.015*0.1*0.5]
            }

        }

    def electricity_solar_cdq(self, df, province, predict, label='C_REAL_VALUE', output=True):
        df['C_TIME'] = pd.to_datetime(df['C_TIME'])
        df['C_TIME_DAY'] = df['C_TIME'].dt.strftime("%Y-%m-%d")
        dfs = df.groupby('C_TIME_DAY')
        limit = self.regulations[province]['solar'][1]
        alpha = self.regulations[province]['solar'][3]
        sum_score, sum_acc = 0, 0
        for dt, data in dfs:
            acc = self.regulations[province]['func'](data[label].values, data[predict].values)
            score = (limit - acc) * alpha if acc < limit else 0
            sum_acc += acc
            sum_score += score
            if output:
                self.logger.info("预测名称：{}，日期：{}，区域：{}，超短期的准确率：{:.4f}，考核分数：{:.4f}".format(predict, str(dt), province, acc, score))
        return round(sum_acc/len(dfs), 5), round(sum_score/len(dfs), 5)

    def electricity_wind_cdq(self, df, province, predict, label='C_REAL_VALUE', output=True):
        df['C_TIME'] = pd.to_datetime(df['C_TIME'])
        df['C_TIME_DAY'] = df['C_TIME'].dt.strftime("%Y-%m-%d")
        dfs = df.groupby('C_TIME_DAY')
        limit = self.regulations[province]['solar'][1]
        alpha = self.regulations[province]['solar'][3]
        sum_score, sum_acc = 0, 0
        for dt, data in dfs:
            acc = self.regulations[province]['func'](data[label].values, data[predict].values)
            score = (limit - acc) * alpha if acc < limit else 0
            sum_acc += acc
            sum_score += score
            if output:
                self.logger.info("预测名称：{}，日期：{}，区域：{}，超短期的准确率：{:.4f}，考核分数：{:.4f}".format(predict, str(dt), province, acc, score))
        return round(sum_acc / len(dfs), 2), round(sum_score / len(dfs), 2)


class Formulas(object):
    def __init__(self, opt):
        self.opt = opt

    def calculate_acc(self, label_data, predict_data):
        loss = np.sum((label_data - predict_data) ** 2) / len(label_data)  # mse
        loss_sqrt = np.sqrt(loss)  # rmse
        loss_acc = (1 - loss_sqrt / self.opt.cap) * 100
        return loss_acc


    def calculate_acc_south(self, label_data, predict_data):
        cap = 0.1 * self.opt.cap
        mask = (label_data < cap) & (predict_data < cap)
        label_data = label_data[~mask]
        predict_data = predict_data[~mask]
        if len(predict_data) == 0:
            return 0
        diff = label_data - predict_data
        base = np.where(label_data < self.opt.cap * 0.2, self.opt.cap * 0.2, label_data)
        acc = np.sum((diff / base) ** 2) / len(diff)
        acc = (1 - np.sqrt(acc)) * 100
        return acc


    def calculate_acc_northwest(self, label_data, predict_data):
        cap = 0.03 * self.opt.cap
        mask = (label_data < cap) & (predict_data < cap)
        label_data = label_data[~mask]
        predict_data = predict_data[~mask]
        if len(predict_data) == 0:
            return 0
        diff = np.abs(label_data - predict_data)
        base1 = label_data + predict_data + 1e-9
        base2 = np.sum(diff) + 1e-9
        acc = (1 - 2 * np.sum(np.abs(label_data / base1 - 0.5) * diff / base2)) * 100
        return acc

    def calculate_acc_northeast(self, label_data, predict_data):
        cap = 0.1 * self.opt.cap
        mask = (label_data < cap) & (predict_data < cap)
        label_data = label_data[~mask]
        predict_data = predict_data[~mask]
        if len(predict_data) == 0:
            return 0
        diff = np.abs(predict_data - label_data)
        deviation = diff / np.abs(predict_data + 1e-9)
        acc = np.where(deviation >= 1, 1, deviation)
        acc = 1 - np.mean(acc)
        return acc

if __name__ == '__main__':
    from config import myargparse

    args = myargparse(discription="场站端配置", add_help=False)
    opt = args.parse_args_and_yaml()
    formula = Formulas(opt)
    test = pd.read_csv('./data/测试集16.csv')
    test = test.iloc[:96, :]
    acc = formula.calculate_acc_northeast(test['C_REAL_VALUE'], test['C_FP_VALUE'])
    print(acc)