pv_evaluate_tool.py 10 KB
# -*- coding: utf-8 -*-

import pandas as pd
import numpy as np
import copy
import numpy_financial as npf


class PvEvaluateTool(object):
    """光伏测算工具"""
    """
    pv_system={
            "user_type": "工商业", #建筑类型
            "install_space":1200,       #屋顶面积m2
            "area_conversion_ratio":0.7, # #面积折算系数
            "capacity_per_meter": 100, # 单位面积容量
            "self_use_ratio":1.0, #自发自用比例
            "efficiency":0.8,         # 发电效率
            "evaluate_year": 25,  #评估年限
            "first_3year_decay_rate":0.015, #前3年衰减率
            "other_year_decay_rate":0.008, #4-25年衰减率
            "annual_sunshine_hours": 1347.945 #年有效日照小时数
            }

    price={
        "rmb_per_wp":7.5,       #建设单价
        "maintenance_per_wp":0.05, # 运维单价
        "coal_in_grid": 0.43, #脱硫电价        
        "self_use_price_discout":1.0, #自发自用电价折扣
        "spfv_price": 0.6  #测算时段平均电价                            
        }

    total_capacity = 315
    df_load负荷曲线
    df_pv负荷曲线
    """

    def __init__(self, pv_system, price, total_capacity, df_load, df_pv):
        self.pv_system = pv_system
        self.price = price
        self.total_capacity = total_capacity
        self.invest_capacity = self.cal_install_capacity()
        self.invest_charge = self.cal_invest_charge()
        self.maintenance_year_charge = self.cal_maintenance_year_charge()
        self.first_year_kwh = self.cal_first_year_kwh()
        self.curve = self._merge_curve(df_load, df_pv)
        self.evaluate_table = None

        self.static_period = 0
        self.dynamic_period = 0
        self.irr = 0
        self.msg = ''

    def _merge_curve(self, df_load, df_pv):
        pv_kWp = self.invest_capacity / 1000
        df_load1 = copy.deepcopy(df_load)
        df_pv1 = copy.deepcopy(df_pv)
        df_pv1["pv_curve"] = df_pv1["pv_curve"]/1000 * pv_kWp * self.pv_system[
            "efficiency"]
        rst = pd.merge(df_load1[["quarter_time", "load_curve"]],
                       df_pv1[["quarter_time", "pv_curve"]], how="left",
                       on="quarter_time")
        return rst

    def cal_install_capacity(self):
        rst = self.pv_system["install_space"] * self.pv_system[
            "area_conversion_ratio"] * self.pv_system["capacity_per_meter"]
        if rst > self.total_capacity*1000:
            rst = self.total_capacity*1000

        return rst

    def cal_maintenance_year_charge(self):
        rst = self.invest_capacity * self.price["maintenance_per_wp"]
        return rst

    def cal_invest_charge(self):
        rst = self.invest_capacity * self.price["rmb_per_wp"]
        return rst

    def cal_first_year_kwh(self):
        rst = self.invest_capacity * self.pv_system[
            "annual_sunshine_hours"] / 1000
        return rst

    def cal_self_use_benefit(self, kwh):
        rst = self.price["spfv_price"] * self.price[
            "self_use_price_discout"] * kwh * self.pv_system["self_use_ratio"]
        return rst

    def cal_on_grid_benefit(self, kwh):
        rst = self.price["coal_in_grid"] * kwh * (
                    1 - self.pv_system["self_use_ratio"])
        return rst

    def cal_total_benefit(self, kwh):
        rst = self.cal_self_use_benefit(kwh) + self.cal_on_grid_benefit(kwh)
        return rst

    def cal_evaluate_table(self):
        table_title = ["年份", "年衰减率", "剩余容量", "年发电量",
                       "维护成本", "累计总成本", "本年收益", "累计收益"]
        decay_rate = ([0] + [self.pv_system["first_3year_decay_rate"]] * 3
                      + [self.pv_system["other_year_decay_rate"]] * (
                                  self.pv_system["evaluate_year"] - 4))
        bank_interest = [1 / (1 + self.price["bank_interest"])] * \
                        self.pv_system["evaluate_year"]
        charge_output = [- self.invest_charge] + [0] * (
                    self.pv_system["evaluate_year"] - 1)
        data = pd.DataFrame(
            {"年份": range(1, self.pv_system["evaluate_year"] + 1),
             "年衰减率": decay_rate,
             "支出": charge_output, "折现率": bank_interest},
            columns=["年份", "年衰减率", "支出", "折现率"])
        data["剩余容量"] = 1 - data["年衰减率"]
        data["剩余容量"] = data["剩余容量"].cumprod()
        data["折现率"] = data["折现率"].cumprod()
        data["年发电量"] = data["剩余容量"] * self.first_year_kwh
        data["维护成本"] = -self.maintenance_year_charge
        data["累计总成本"] = (data["维护成本"] + data["支出"]).cumsum()
        data["本年收益"] = self.cal_total_benefit(data["年发电量"]) + data["维护成本"]
        data["本年折现收益"] = data["本年收益"] * data["折现率"]
        data["累计收益"] = (data["本年收益"] + data["支出"]).cumsum()
        data["折现累计收益"] = (data["本年折现收益"] + data["支出"]).cumsum()

        static_temp = data["累计收益"][data["累计收益"] > 0]
        if len(static_temp) > 0:
            static_index = list(static_temp.index)[0]
            static_index = static_index - 1 if static_index > 0 else 0
            self.static_period = data.loc[static_index, "年份"] - data.loc[
                static_index, "累计收益"] / (data.loc[
                                             static_index + 1, "本年收益"] + 0.001)
        else:
            self.static_period = -1

        dynamic_temp = data["折现累计收益"][data["折现累计收益"] > 0]
        if len(dynamic_temp) > 0:
            dynamic_index = list(dynamic_temp.index)[0]
            dynamic_index = dynamic_index - 1 if dynamic_index > 0 else 0
            self.dynamic_period = data.loc[dynamic_index, "年份"] - data.loc[
                dynamic_index, "折现累计收益"] / (data.loc[
                                                dynamic_index + 1, "本年折现收益"] + 0.001)
        else:
            self.dynamic_period = -1

        self.irr = npf.irr([- self.invest_charge] + list(data["本年收益"].values))
        return data[table_title]

    @staticmethod
    def fix_decimal_points(v, decimal_num=4):
        """ 将浮点型的值固定小数位, 默认保留4位小数位
        :param v: 浮点型的值
        """
        rlt = v
        fmt = "%.df"
        fmt = fmt.replace('d', str(decimal_num))
        if isinstance(v, float):
            s = fmt % v
            rlt = float(s)
        return rlt

    def output(self):
        self.input_param_process()
        data = self.cal_evaluate_table()
        for key in data.columns:
            data[key] = data[key].apply(lambda x: self.fix_decimal_points(x))
        for key in ["load_curve", "pv_curve"]:
            self.curve[key] = self.curve[key].apply(
                lambda x: self.fix_decimal_points(x))
        self.evaluate_table = data

    def input_param_process(self):
        if self.pv_system["area_conversion_ratio"] <= 0:
            self.pv_system["area_conversion_ratio"] = 0.0
        elif self.pv_system["area_conversion_ratio"] >= 1:
            self.pv_system["area_conversion_ratio"] = 1

        if self.pv_system["efficiency"] <= 0:
            self.pv_system["efficiency"] = 0.0
        elif self.pv_system["efficiency"] >= 1:
            self.pv_system["efficiency"] = 1

        if self.pv_system["self_use_ratio"] <= 0:
            self.pv_system["self_use_ratio"] = 0.0
        elif self.pv_system["self_use_ratio"] >= 1:
            self.pv_system["self_use_ratio"] = 1

        if self.pv_system["evaluate_year"] <= 10:
            self.pv_system["evaluate_year"] = 10

        if self.price["self_use_price_discout"] <= 0:
            self.price["self_use_price_discout"] = 0.0
        elif self.price["self_use_price_discout"] >= 1:
            self.price["self_use_price_discout"] = 1


if __name__ == "__main__":

    pv_system = {
        "user_type": "工商业",  # 建筑类型
        "install_space": 2000,  # 屋顶面积m2
        "area_conversion_ratio": 0.8,  # #面积折算系数
        "capacity_per_meter": 100,  # 单位面积容量
        "self_use_ratio": 1.0,  # 自发自用比例
        "efficiency": 0.8,  # 发电效率
        "evaluate_year": 25,  # 评估年限
        "first_3year_decay_rate": 0.015,  # 前3年衰减率
        "other_year_decay_rate": 0.008,  # 4-25年衰减率
        "annual_sunshine_hours": 989.04  # 年峰值日照小数数
    }

    price = {
        "rmb_per_wp": 4.3,  # 建设单价
        "maintenance_per_wp": 0.05,  # 运维单价
        "coal_in_grid": 0.45,  # 脱硫电价
        "self_use_price_discout": 1.0,  # 自发自用电价折扣
        "spfv_price": 0.48885056077566996,  # 测算时段平均电价
        "bank_interest": 0.085
    }

    total_capacity = 160  # 月
    stat_time = pd.to_datetime(
        pd.date_range("2019-01-01", "2019-01-02", freq="0.25H")[:-1])
    df_load = pd.DataFrame(
        {"quarter_time": stat_time, "load_curve": np.random.random(96)},
        columns=["quarter_time", "load_curve"])
    df_load["quarter_time"] = df_load["quarter_time"].apply(
        lambda x: x.strftime("%H:%M:%S"))

    df_pv = pd.DataFrame(
        {"quarter_time": stat_time, "pv_curve": np.random.random(96)},
        columns=["quarter_time", "pv_curve"])
    df_pv["quarter_time"] = df_pv["quarter_time"].apply(
        lambda x: x.strftime("%H:%M:%S"))
    # print(df_pv)

    obj = PvEvaluateTool(pv_system, price, total_capacity, df_load, df_pv)
    obj.output()

    """光伏测算工具"""
    print(
        "**********************************************************************************************************")
    print("输入参数")
    print()
    print(obj.invest_charge)
    print(obj. invest_charge)   # 获取总投资
    print(obj.static_period)   # 静态回收期
    print(obj.invest_capacity)  # 获取装机容量
    print(obj.first_year_kwh)   # 获取首年发电量

    print(obj.evaluate_table)  # 测算表
    print(obj.curve)  # 计算优化分析结果