# -*- 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) # 计算优化分析结果