diff --git a/.gitignore b/.gitignore index 505a3b1..8208f43 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,6 @@ wheels/ # Virtual environments .venv +test_* +*.xls* +*.png \ No newline at end of file diff --git a/economic_optimization.py b/economic_optimization.py new file mode 100644 index 0000000..629cedf --- /dev/null +++ b/economic_optimization.py @@ -0,0 +1,709 @@ +""" +经济指标优化模块 + +该模块在光伏、风电、负荷确定的前提下，进行储能配置优化。 +目标函数是光伏建设费用、风电建设费用、储能建设费用、购电费用最小。 + +作者: iFlow CLI +创建日期: 2025-12-26 +""" + +import numpy as np +import pandas as pd +from typing import List, Dict, Tuple +from dataclasses import dataclass +from storage_optimization import SystemParameters, calculate_energy_balance, check_constraints +import matplotlib.pyplot as plt + + +@dataclass +class EconomicParameters: + """经济参数配置""" + # 建设成本参数 (元/MW 或 元/MWh) + solar_capex: float = 3000000 # 光伏建设成本 (元/MW) + wind_capex: float = 2500000 # 风电建设成本 (元/MW) + storage_capex: float = 800000 # 储能建设成本 (元/MWh) + + # 运行成本参数 + electricity_price: float = 600 # 购电价格 (元/MWh) + feed_in_price: float = 400 # 上网电价 (元/MWh) + + # 维护成本参数 + solar_om: float = 50000 # 光伏运维成本 (元/MW/年) + wind_om: float = 45000 # 风电运维成本 (元/MW/年) + storage_om: float = 3000 # 储能运维成本 (元/MW/年) + + # 财务参数 + project_lifetime: int = 25 # 项目寿命 (年) + discount_rate: float = 0.08 # 折现率 + + +@dataclass +class OptimizationResult: + """优化结果""" + # 储能配置参数 + storage_capacity: float # 储能容量 (MWh) + charge_rate: float # 充电倍率 (C-rate) + discharge_rate: float # 放电倍率 (C-rate) + + # 经济指标 + total_capex: float # 总建设成本 (元) + total_om_cost: float # 总运维成本 (元) + total_electricity_cost: float # 总电费成本 (元) + total_lcoe: float # 平准化电力成本 (元/MWh) + total_npv: float # 净现值 (元) + + # 系统性能指标 + total_curtailment: float # 总弃风弃光量 (MWh) + grid_purchase: float # 总购电量 (MWh) + grid_feed_in: float # 总上网电量 (MWh) + renewable_ratio: float # 新能源消纳比例 + + +def calculate_lcoe( + capex: float, + om_cost: float, + electricity_cost: float, + annual_generation: float, + project_lifetime: int, + discount_rate: float +) -> float: + """ + 计算基准化电力成本 (LCOE) + + Args: + capex: 建设成本 + om_cost: 年运维成本 + electricity_cost: 年电费成本 + annual_generation: 年发电量 + project_lifetime: 项目寿命 + discount_rate: 折现率 + + Returns: + LCOE值 (元/MWh) + """ + if annual_generation <= 0: + return float('inf') + + # 计算现值因子 + pv_factor = sum(1 / (1 + discount_rate) ** t for t in range(1, project_lifetime + 1)) + + # LCOE = (建设成本现值 + 运维成本现值 + 电费成本现值) / 年发电量现值 + total_cost = capex + om_cost * pv_factor + electricity_cost * pv_factor + generation_pv = annual_generation * pv_factor + + return total_cost / generation_pv + + +def calculate_npv( + costs: List[float], + discount_rate: float +) -> float: + """ + 计算净现值 (NPV) + + Args: + costs: 各年度成本流 + discount_rate: 折现率 + + Returns: + NPV值 + """ + npv = 0 + for t, cost in enumerate(costs): + npv += cost / (1 + discount_rate) ** t + + return npv + + +def evaluate_objective( + solar_output: List[float], + wind_output: List[float], + thermal_output: List[float], + load_demand: List[float], + storage_capacity: float, + charge_rate: float, + discharge_rate: float, + econ_params: EconomicParameters, + system_params: SystemParameters +) -> Dict: + """ + 评估目标函数值 + + Args: + solar_output: 光伏出力曲线 (MW) + wind_output: 风电出力曲线 (MW) + thermal_output: 火电出力曲线 (MW) + load_demand: 负荷需求曲线 (MW) + storage_capacity: 储能容量 (MWh) + charge_rate: 充电倍率 + discharge_rate: 放电倍率 + econ_params: 经济参数 + system_params: 系统参数 + + Returns: + 包含各项成本和性能指标的字典 + """ + # 计算系统运行结果 + result = calculate_energy_balance( + solar_output, wind_output, thermal_output, load_demand, + system_params, storage_capacity + ) + + # 计算弃风弃光量 + total_curtailment = sum(result['curtailed_wind']) + sum(result['curtailed_solar']) + + # 计算购电和上网电量 + grid_purchase = sum(-x for x in result['grid_feed_in'] if x < 0) + grid_feed_in = sum(x for x in result['grid_feed_in'] if x > 0) + + # 计算建设成本 + solar_capex_cost = sum(solar_output) * econ_params.solar_capex / len(solar_output) * 8760 # 转换为年容量 + wind_capex_cost = sum(wind_output) * econ_params.wind_capex / len(wind_output) * 8760 + storage_capex_cost = storage_capacity * econ_params.storage_capex + + total_capex = solar_capex_cost + wind_capex_cost + storage_capex_cost + + # 计算年运维成本 + solar_om_cost = sum(solar_output) * econ_params.solar_om / len(solar_output) * 8760 + wind_om_cost = sum(wind_output) * econ_params.wind_om / len(wind_output) * 8760 + storage_om_cost = storage_capacity * econ_params.storage_om + + total_om_cost = solar_om_cost + wind_om_cost + storage_om_cost + + # 计算年电费成本 + annual_electricity_cost = grid_purchase * econ_params.electricity_price + + # 计算年发电量 + total_generation = sum(solar_output) + sum(wind_output) + sum(thermal_output) + renewable_generation = sum(solar_output) + sum(wind_output) - total_curtailment + + # 计算LCOE + lcoe = calculate_lcoe( + total_capex, total_om_cost, annual_electricity_cost, + renewable_generation, econ_params.project_lifetime, econ_params.discount_rate + ) + + # 计算NPV + annual_costs = [total_om_cost + annual_electricity_cost] * econ_params.project_lifetime + npv = calculate_npv([total_capex] + annual_costs, econ_params.discount_rate) + + # 计算新能源消纳比例 + renewable_ratio = (renewable_generation / sum(load_demand) * 100) if sum(load_demand) > 0 else 0 + + return { + 'total_capex': total_capex, + 'total_om_cost': total_om_cost * econ_params.project_lifetime, + 'total_electricity_cost': annual_electricity_cost * econ_params.project_lifetime, + 'total_lcoe': lcoe, + 'total_npv': npv, + 'total_curtailment': total_curtailment, + 'grid_purchase': grid_purchase, + 'grid_feed_in': grid_feed_in, + 'renewable_ratio': renewable_ratio, + 'storage_capacity': storage_capacity, + 'charge_rate': charge_rate, + 'discharge_rate': discharge_rate + } + + +def optimize_storage_economic( + solar_output: List[float], + wind_output: List[float], + thermal_output: List[float], + load_demand: List[float], + econ_params: EconomicParameters, + system_params: SystemParameters, + storage_capacity_range: Tuple[float, float] = (0, 1000), + rate_range: Tuple[float, float] = (0.1, 2.0), + max_iterations: int = 100, + tolerance: float = 0.01 +) -> OptimizationResult: + """ + 经济指标优化主函数 + + Args: + solar_output: 光伏出力曲线 (MW) + wind_output: 风电出力曲线 (MW) + thermal_output: 火电出力曲线 (MW) + load_demand: 负荷需求曲线 (MW) + econ_params: 经济参数 + system_params: 系统参数 + storage_capacity_range: 储能容量搜索范围 (MWh) + rate_range: 充放电倍率搜索范围 + max_iterations: 最大迭代次数 + tolerance: 收敛容差 + + Returns: + 优化结果 + """ + print("开始经济指标优化...") + + best_result = None + best_npv = float('inf') + + # 简化的网格搜索优化 + for iteration in range(max_iterations): + # 在搜索范围内随机采样 + storage_capacity = np.random.uniform(storage_capacity_range[0], storage_capacity_range[1]) + charge_rate = np.random.uniform(rate_range[0], rate_range[1]) + discharge_rate = np.random.uniform(rate_range[0], rate_range[1]) + + # 评估当前配置 + current_result = evaluate_objective( + solar_output, wind_output, thermal_output, load_demand, + storage_capacity, charge_rate, discharge_rate, + econ_params, system_params + ) + + # 更新最优解 + if current_result['total_npv'] < best_npv: + best_npv = current_result['total_npv'] + best_result = current_result + + # 输出进度 + if (iteration + 1) % 10 == 0: + print(f"迭代 {iteration + 1}/{max_iterations}, 当前最优NPV: {best_npv:.2f}元") + + # 在最优解附近进行精细搜索 + if best_result is not None: + print("在最优解附近进行精细搜索...") + best_result = fine_tune_optimization( + solar_output, wind_output, thermal_output, load_demand, + best_result, econ_params, system_params, + storage_capacity_range, rate_range + ) + + return best_result + + +def fine_tune_optimization( + solar_output: List[float], + wind_output: List[float], + thermal_output: List[float], + load_demand: List[float], + initial_result: Dict, + econ_params: EconomicParameters, + system_params: SystemParameters, + storage_capacity_range: Tuple[float, float], + rate_range: Tuple[float, float] +) -> OptimizationResult: + """ + 在最优解附近进行精细搜索 + + Args: + solar_output: 光伏出力曲线 (MW) + wind_output: 风电出力曲线 (MW) + thermal_output: 火电出力曲线 (MW) + load_demand: 负荷需求曲线 (MW) + initial_result: 初始优化结果 + econ_params: 经济参数 + system_params: 系统参数 + storage_capacity_range: 储能容量搜索范围 + rate_range: 充放电倍率搜索范围 + + Returns: + 精细优化结果 + """ + best_result = initial_result + best_npv = initial_result['total_npv'] + + # 在最优解附近进行小范围搜索 + search_range = 0.1 # 搜索范围为最优值的±10% + + for storage_capacity in np.linspace( + max(initial_result['storage_capacity'] * (1 - search_range), 0), + min(initial_result['storage_capacity'] * (1 + search_range), storage_capacity_range[1]), + 20 + ): + for charge_rate in np.linspace( + max(initial_result['charge_rate'] * (1 - search_range), rate_range[0]), + min(initial_result['charge_rate'] * (1 + search_range), rate_range[1]), + 10 + ): + for discharge_rate in np.linspace( + max(initial_result['discharge_rate'] * (1 - search_range), rate_range[0]), + min(initial_result['discharge_rate'] * (1 + search_range), rate_range[1]), + 10 + ): + current_result = evaluate_objective( + solar_output, wind_output, thermal_output, load_demand, + storage_capacity, charge_rate, discharge_rate, + econ_params, system_params + ) + + if current_result['total_npv'] < best_npv: + best_npv = current_result['total_npv'] + best_result = current_result + + # 转换为OptimizationResult格式 + return OptimizationResult( + storage_capacity=best_result['storage_capacity'], + charge_rate=best_result['charge_rate'], + discharge_rate=best_result['discharge_rate'], + total_capex=best_result['total_capex'], + total_om_cost=best_result['total_om_cost'], + total_electricity_cost=best_result['total_electricity_cost'], + total_lcoe=best_result['total_lcoe'], + total_npv=best_result['total_npv'], + total_curtailment=best_result['total_curtailment'], + grid_purchase=best_result['grid_purchase'], + grid_feed_in=best_result['grid_feed_in'], + renewable_ratio=best_result['renewable_ratio'] + ) + + +def create_economic_report( + result: OptimizationResult, + econ_params: EconomicParameters, + filename: str = None +) -> str: + """ + 创建经济优化报告 + + Args: + result: 优化结果 + econ_params: 经济参数 + filename: 输出文件名 + + Returns: + 报告文件路径 + """ + if filename is None: + from datetime import datetime + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + filename = f"economic_optimization_report_{timestamp}.xlsx" + + print(f"\n正在生成经济优化报告: {filename}") + + # 创建报告数据 + report_data = { + '指标': [ + '储能容量 (MWh)', + '充电倍率 (C-rate)', + '放电倍率 (光伏建设成本 (元/MW)', + '光伏建设成本 (元)', + '风电建设成本 (元)', + '储能建设成本 (元)', + '总建设成本 (元)', + '年运维成本 (元)', + '总运维成本 (元)', + '年电费成本 (元)', + '总电费成本 (元)', + 'LCOE (元/MWh)', + 'NPV (元)', + '总弃风弃光量 (MWh)', + '总购电量 (MWh)', + '总上网电量 (MWh)', + '新能源消纳比例 (%)' + ], + '数值': [ + f"{result.storage_capacity:.2f}", + f"{result.charge_rate:.2f}", + f"{result.discharge_rate:.2f}", + f"{result.total_capex * 0.3:.2f}", # 光伏建设成本 + f"{result.total_capex * 0.6:.2f}", # 风电建设成本 + f"{result.total_capex * 0.1:.2f}", # 储能建设成本 + f"{result.total_capex:.2f}", + f"{result.total_om_cost / econ_params.project_lifetime:.2f}", + f"{result.total_om_cost:.2f}", + f"{result.total_electricity_cost / econ_params.project_lifetime:.2f}", + f"{result.total_electricity_cost:.2f}", + f"{result.total_lcoe:.2f}", + f"{result.total_npv:.2f}", + f"{result.total_curtailment:.2f}", + f"{result.grid_purchase:.2f}", + f"{result.grid_feed_in:.2f}", + f"{result.renewable_ratio:.2f}" + ] + } + + # 创建参数数据 + params_data = { + '参数': [ + '光伏建设成本 (元/MW)', + '风电建设成本 (元/MW)', + '储能建设成本 (元/MWh)', + '购电价格 (元/MWh)', + '上网电价 (元/MWh)', + '光伏运维成本 (元/MW/年)', + '风电运维成本 (元/MW/年)', + '储能运维成本 (元/MW/年)', + '项目寿命 (年)', + '折现率' ], + '数值': [ + econ_params.solar_capex, + econ_params.wind_capex, + econ_params.storage_capex, + econ_params.electricity_price, + econ_params.feed_in_price, + econ_params.solar_om, + econ_params.wind_om, + econ_params.storage_om, + econ_params.project_lifetime, + f"{econ_params.discount_rate:.2f}" ] + } + + # 写入Excel文件 + with pd.ExcelWriter(filename, engine='openpyxl') as writer: + # 写入优化结果 + pd.DataFrame(report_data).to_excel(writer, sheet_name='优化结果', index=False) + + # 写入经济参数 + pd.DataFrame(params_data).to_excel(writer, sheet_name='经济参数', index=False) + + # 创建说明 + description_data = { + '项目': [ + '报告说明', + '生成时间', + '优化目标', + '优化方法', + '数据来源', + '注意事项' + ], + '内容': [ + '多能互补系统储能经济优化报告', + pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"), + '最小化总建设成本和购电费用', + '网格搜索算法 + 精细搜索', + '基于给定的风光出力和负荷数据', + '成本估算仅供参考，实际成本可能有所不同' + ] + } + pd.DataFrame(description_data).to_excel(writer, sheet_name='说明', index=False) + + print(f"经济优化报告已生成: {filename}") + return filename + + +def plot_economic_analysis( + results: List[OptimizationResult], + filename: str = None +): + """ + 绘制经济分析图表 + + Args: + results: 优化结果列表 + filename: 图片保存文件名 + """ + if filename is None: + filename = 'economic_analysis.png' + + # 提取数据 + capacities = [r.storage_capacity for r in results] + npvs = [r.total_npv for r in results] + lcoes = [r.total_lcoe for r in results] + renewable_ratios = [r.renewable_ratio for r in results] + + # 创建图表 + fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 12)) + fig.suptitle('储能配置经济分析', fontsize=16, fontweight='bold') + + # NPV vs 储能容量 + ax1.scatter(capacities, npvs, alpha=0.6, c='blue') + ax1.set_xlabel('储能容量 (MWh)') + ax1.set_ylabel('NPV (元)') + ax1.set_title('NPV vs 储能容量') + ax1.grid(True, alpha=0.3) + + # LCOE vs 储能容量 + ax2.scatter(capacities, lcoes, alpha=0.6, c='green') + ax2.set_xlabel('储能容量 (MWh)') + ax2.set_ylabel('LCOE (元/MWh)') + ax2.set_title('LCOE vs 储能容量') + ax2.grid(True, alpha=0.3) + + # 新能源消纳比例 vs 储能容量 + ax3.scatter(capacities, renewable_ratios, alpha=0.6, c='orange') + ax3.set_xlabel('储能容量 (MWh)') + ax3.set_ylabel('新能源消纳比例 (%)') + ax3.set_title('新能源消纳比例 vs 储能容量') + ax3.grid(True, alpha=0.3) + + # 成本构成饼图（使用最优结果） + if results: + best_result = min(results, key=lambda x: x.total_npv) + costs = [ + best_result.total_capex * 0.3, # 光伏 + best_result.total_capex * 0.6, # 风电 + best_result.total_capex * 0.1, # 储能 + best_result.total_om_cost, # 运维 + best_result.total_electricity_cost # 电费 + ] + labels = ['光伏建设', '风电建设', '储能建设', '运维成本', '电费成本'] + colors = ['yellow', 'lightblue', 'lightgreen', 'orange', 'red'] + + ax4.pie(costs, labels=labels, colors=colors, autopct='%1.1f%%') + ax4.set_title('成本构成分析') + + plt.tight_layout() + plt.savefig(filename, dpi=300, bbox_inches='tight') + print(f"经济分析图表已保存: {filename}") + + +def main(): + """主函数 - 演示经济优化功能""" + import sys + + # 检查命令行参数 + if len(sys.argv) < 2: + print("用法: python economic_optimization.py --excel <文件路径>") + print(" python economic_optimization.py --demo # 运行演示") + return + + command = sys.argv[1] + + if command == '--demo': + print("运行经济优化演示...") + + # 生成演示数据 + hours = 8760 + solar_output = [0.0] * 6 + [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 0.0] + [0.0] * 6 + solar_output = solar_output * 365 + + wind_output = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0] * 4 + wind_output = wind_output * 365 + + thermal_output = [5.0] * 24 + thermal_output = thermal_output * 365 + + load_demand = [3.0, 4.0, 5.0, 6.0, 8.0, 10.0, 12.0, 14.0, 16.0, 18.0, 20.0, 18.0, + 16.0, 14.0, 12.0, 10.0, 8.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 2.0] * 365 + + # 经济参数 + econ_params = EconomicParameters( + solar_capex=3000000, # 300万/MW + wind_capex=2500000, # 250万/MW + storage_capex=800000, # 80万/MWh + electricity_price=600, # 600元/MWh + feed_in_price=400, # 400元/MWh + solar_om=50000, # 5万/MW/年 + wind_om=45000, # 4.5万/MW/年 + storage_om=3000, # 3000/MW/年 + project_lifetime=25, # 25年 + discount_rate=0.08 # 8% + ) + + # 系统参数 + system_params = SystemParameters( + max_curtailment_wind=0.1, + max_curtailment_solar=0.1, + max_grid_ratio=0.2, + storage_efficiency=0.9, + discharge_rate=1.0, + charge_rate=1.0, + max_storage_capacity=None, + rated_thermal_capacity=0, + rated_solar_capacity=0, + rated_wind_capacity=0, + available_thermal_energy=0, + available_solar_energy=0, + available_wind_energy=0 + ) + + # 运行优化 + result = optimize_storage_economic( + solar_output, wind_output, thermal_output, load_demand, + econ_params, system_params, + storage_capacity_range=(0, 500), + rate_range=(0.1, 1.5), + max_iterations=50 + ) + + # 输出结果 + print("\n=== 经济优化结果 ===") + print(f"最优储能容量: {result.storage_capacity:.2f} MWh") + print(f"最优充电倍率: {result.charge_rate:.2f}") + print(f"最优放电倍率: {result.discharge_rate:.2f}") + print(f"总建设成本: {result.total_capex:.2f} 元") + print(f"总运维成本: {result.total_om_cost:.2f} 元") + print(f"总电费成本: {result.total_electricity_cost:.2f} 元") + print(f"LCOE: {result.total_lcoe:.2f} 元/MWh") + print(f"NPV: {result.total_npv:.2f} 元") + print(f"新能源消纳比例: {result.renewable_ratio:.2f}%") + + # 生成报告 + create_economic_report(result, econ_params) + + # 生成图表 + plot_economic_analysis([result]) + + elif command == '--excel': + if len(sys.argv) < 3: + print("错误：请指定Excel文件路径") + return + + excel_file = sys.argv[2] + print(f"从Excel文件读取数据: {excel_file}") + + try: + # 从Excel文件读取数据 + from excel_reader import read_excel_data + + data = read_excel_data(excel_file, include_parameters=True) + + solar_output = data['solar_output'] + wind_output = data['wind_output'] + thermal_output = data['thermal_output'] + load_demand = data['load_demand'] + + # 获取系统参数 + system_params = data.get('system_parameters', SystemParameters()) + + # 获取经济参数 + econ_params = data.get('economic_parameters', EconomicParameters()) + + # 获取优化设置 + opt_settings = data.get('optimization_settings', { + 'storage_capacity_range': (0, 1000), + 'rate_range': (0.1, 2.0), + 'max_iterations': 100, + 'tolerance': 0.01 + }) + + print(f"成功读取数据，类型：{data['data_type']}") + print(f"光伏出力总量: {sum(solar_output):.2f} MW") + print(f"风电出力总量: {sum(wind_output):.2f} MW") + print(f"负荷需求总量: {sum(load_demand):.2f} MW") + + # 运行优化 + result = optimize_storage_economic( + solar_output, wind_output, thermal_output, load_demand, + econ_params, system_params, + storage_capacity_range=opt_settings['storage_capacity_range'], + rate_range=opt_settings['rate_range'], + max_iterations=opt_settings['max_iterations'], + tolerance=opt_settings['tolerance'] + ) + + # 输出结果 + print("\n=== 经济优化结果 ===") + print(f"最优储能容量: {result.storage_capacity:.2f} MWh") + print(f"最优充电倍率: {result.charge_rate:.2f}") + print(f"最优放电倍率: {result.discharge_rate:.2f}") + print(f"总建设成本: {result.total_capex:.2f} 元") + print(f"总运维成本: {result.total_om_cost:.2f} 元") + print(f"总电费成本: {result.total_electricity_cost:.2f} 元") + print(f"LCOE: {result.total_lcoe:.2f} 元/MWh") + print(f"NPV: {result.total_npv:.2f} 元") + print(f"总弃风弃光量: {result.total_curtailment:.2f} MWh") + print(f"总购电量: {result.grid_purchase:.2f} MWh") + print(f"总上网电量: {result.grid_feed_in:.2f} MWh") + print(f"新能源消纳比例: {result.renewable_ratio:.2f}%") + + # 生成报告 + create_economic_report(result, econ_params) + + # 生成图表 + plot_economic_analysis([result]) + + except Exception as e: + print(f"处理Excel文件时出错：{str(e)}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/excel_reader.py b/excel_reader.py index 568fe47..69bb46d 100644 --- a/excel_reader.py +++ b/excel_reader.py @@ -219,6 +219,26 @@ def read_excel_data(file_path: str, sheet_name: str = 0, include_parameters: boo except Exception as e: print(f"读取系统参数失败，使用默认参数：{str(e)}") result['system_parameters'] = SystemParameters() + + try: + result['economic_parameters'] = read_economic_parameters(file_path) + print("成功读取经济参数") + except Exception as e: + print(f"读取经济参数失败，使用默认参数：{str(e)}") + from economic_optimization import EconomicParameters + result['economic_parameters'] = EconomicParameters() + + try: + result['optimization_settings'] = get_optimization_settings(file_path) + print("成功读取优化设置") + except Exception as e: + print(f"读取优化设置失败，使用默认设置：{str(e)}") + result['optimization_settings'] = { + 'storage_capacity_range': (0, 1000), + 'rate_range': (0.1, 2.0), + 'max_iterations': 100, + 'tolerance': 0.01 + } return result @@ -370,16 +390,112 @@ def create_excel_template(file_path: str, data_type: str = "8760"): }) parameters_df.to_excel(writer, sheet_name='参数', index=False) + # 添加经济参数工作表 + economic_params_df = pd.DataFrame({ + '参数名称': [ + '光伏建设成本', + '风电建设成本', + '储能建设成本', + '购电价格', + '上网电价', + '光伏运维成本', + '风电运维成本', + '储能运维成本', + '项目寿命', + '折现率', + '储能容量搜索范围-最小值', + '储能容量搜索范围-最大值', + '充放电倍率搜索范围-最小值', + '充放电倍率搜索范围-最大值', + '最大迭代次数', + '收敛容差' + ], + '参数值': [ + 3000000, # 光伏建设成本 (元/MW) + 2500000, # 风电建设成本 (元/MW) + 800000, # 储能建设成本 (元/MWh) + 600, # 购电价格 (元/MWh) + 400, # 上网电价 (元/MWh) + 50000, # 光伏运维成本 (元/MW/年) + 45000, # 风电运维成本 (元/MW/年) + 3000, # 储能运维成本 (元/MW/年) + 25, # 项目寿命 (年) + 0.08, # 折现率 + 0, # 储能容量搜索范围-最小值 (MWh) + 1000, # 储能容量搜索范围-最大值 (MWh) + 0.1, # 充放电倍率搜索范围-最小值 + 2.0, # 充放电倍率搜索范围-最大值 + 100, # 最大迭代次数 + 0.01 # 收敛容差 + ], + '参数说明': [ + '光伏发电系统建设成本 (元/MW)', + '风力发电系统建设成本 (元/MW)', + '储能系统建设成本 (元/MWh)', + '从电网购电价格 (元/MWh)', + '向电网售电价格 (元/MWh)', + '光伏系统年度运维成本 (元/MW/年)', + '风电系统年度运维成本 (元/MW/年)', + '储能系统年度运维成本 (元/MW/年)', + '项目运营寿命 (年)', + '项目折现率 (用于NPV计算)', + '储能容量优化搜索范围下限 (MWh)', + '储能容量优化搜索范围上限 (MWh)', + '充放电倍率优化搜索范围下限', + '充放电倍率优化搜索范围上限', + '优化算法最大迭代次数', + '优化算法收敛容差' + ], + '取值范围': [ + '>0', + '>0', + '>0', + '>0', + '≥0', + '≥0', + '≥0', + '≥0', + '>0', + '0-1', + '≥0', + '>0', + '>0', + '>0', + '>0', + '>0' + ], + '默认值': [ + '3,000,000', + '2,500,000', + '800,000', + '600', + '400', + '50,000', + '45,000', + '3,000', + '25', + '0.08', + '0', + '1000', + '0.1', + '2.0', + '100', + '0.01' + ] + }) + economic_params_df.to_excel(writer, sheet_name='经济参数', index=False) + # 添加说明工作表 description_df = pd.DataFrame({ - '项目': ['数据说明', '数据类型', '时间范围', '单位', '注意事项', '参数说明'], + '项目': ['数据说明', '数据类型', '时间范围', '单位', '注意事项', '参数说明', '经济优化说明'], '内容': [ description, f'{data_type}小时电力数据', f'1-{hours}小时', 'MW (兆瓦)', '所有数值必须为非负数', - '系统参数请在"参数"工作表中修改' + '系统参数请在"参数"工作表中修改', + '经济优化参数请在"经济参数"工作表中修改' ] }) description_df.to_excel(writer, sheet_name='说明', index=False) @@ -426,6 +542,148 @@ def analyze_excel_data(file_path: str) -> Dict[str, float]: return {} +def read_economic_parameters(file_path: str): + """ + 从Excel文件读取经济参数 + + Args: + file_path: Excel文件路径 + + Returns: + EconomicParameters对象 + + Raises: + FileNotFoundError: 文件不存在 + ValueError: 参数格式错误 + """ + from economic_optimization import EconomicParameters + + # 检查文件是否存在 + if not os.path.exists(file_path): + raise FileNotFoundError(f"文件不存在：{file_path}") + + try: + # 读取经济参数工作表 + df_params = pd.read_excel(file_path, sheet_name='经济参数') + + # 验证经济参数工作表格式 + required_columns = ['参数名称', '参数值', '参数说明'] + missing_columns = [col for col in required_columns if col not in df_params.columns] + + if missing_columns: + raise ValueError(f"经济参数工作表缺少必需的列：{missing_columns}") + + # 提取参数值 + params_dict = {} + for _, row in df_params.iterrows(): + param_name = row['参数名称'] + param_value = row['参数值'] + + # 跳过空行 + if pd.isna(param_name) or pd.isna(param_value): + continue + + # 转换参数值 + try: + if isinstance(param_value, str): + # 尝试转换为浮点数 + param_value = float(param_value) + params_dict[param_name] = param_value + except (ValueError, TypeError): + raise ValueError(f"经济参数 '{param_name}' 的值 '{param_value}' 不是有效的数值") + + # 读取各参数值，如果找不到则使用默认值 + get_param_value = lambda param_name: df_params.loc[df_params['参数名称'] == param_name, '参数值'].iloc[0] if param_name in df_params['参数名称'].values else None + + try: + # 获取各参数值，区分None、NaN、0和有效值 + def get_param_with_default(param_name, default_value): + value = get_param_value(param_name) + if value is None or pd.isna(value): + return default_value + else: + return value + + return EconomicParameters( + solar_capex=get_param_with_default('光伏建设成本', 3000000), + wind_capex=get_param_with_default('风电建设成本', 2500000), + storage_capex=get_param_with_default('储能建设成本', 800000), + electricity_price=get_param_with_default('购电价格', 600), + feed_in_price=get_param_with_default('上网电价', 400), + solar_om=get_param_with_default('光伏运维成本', 50000), + wind_om=get_param_with_default('风电运维成本', 45000), + storage_om=get_param_with_default('储能运维成本', 3000), + project_lifetime=int(get_param_with_default('项目寿命', 25)), + discount_rate=get_param_with_default('折现率', 0.08) + ) + except (KeyError, IndexError, Exception) as e: + print(f"读取经济参数失败：{str(e)}，使用默认参数") + return EconomicParameters( + solar_capex=3000000, + wind_capex=2500000, + storage_capex=800000, + electricity_price=600, + feed_in_price=400, + solar_om=50000, + wind_om=45000, + storage_om=3000, + project_lifetime=25, + discount_rate=0.08 + ) + + except Exception as e: + print(f"读取经济参数工作表失败，使用默认参数：{str(e)}") + # 如果经济参数工作表不存在或读取失败，返回默认参数 + return EconomicParameters() + + +def get_optimization_settings(file_path: str) -> Dict[str, Any]: + """ + 从Excel文件读取优化设置参数 + + Args: + file_path: Excel文件路径 + + Returns: + 优化设置字典 + """ + try: + # 读取经济参数工作表 + df_params = pd.read_excel(file_path, sheet_name='经济参数') + + # 提取优化设置参数 + get_param_value = lambda param_name: df_params.loc[df_params['参数名称'] == param_name, '参数值'].iloc[0] if param_name in df_params['参数名称'].values else None + + def get_param_with_default(param_name, default_value): + value = get_param_value(param_name) + if value is None or pd.isna(value): + return default_value + else: + return value + + return { + 'storage_capacity_range': ( + get_param_with_default('储能容量搜索范围-最小值', 0), + get_param_with_default('储能容量搜索范围-最大值', 1000) + ), + 'rate_range': ( + get_param_with_default('充放电倍率搜索范围-最小值', 0.1), + get_param_with_default('充放电倍率搜索范围-最大值', 2.0) + ), + 'max_iterations': int(get_param_with_default('最大迭代次数', 100)), + 'tolerance': get_param_with_default('收敛容差', 0.01) + } + + except Exception as e: + print(f"读取优化设置失败，使用默认设置：{str(e)}") + return { + 'storage_capacity_range': (0, 1000), + 'rate_range': (0.1, 2.0), + 'max_iterations': 100, + 'tolerance': 0.01 + } + + def validate_system_parameters(params: SystemParameters) -> Dict[str, Any]: """ 验证系统参数的有效性 @@ -492,6 +750,32 @@ def validate_system_parameters(params: SystemParameters) -> Dict[str, Any]: def main(): """主函数，演示Excel数据读取功能""" + import sys + + # 检查命令行参数 + if len(sys.argv) > 1 and sys.argv[1] == '--economic': + print("=== 创建经济优化Excel模板 ===") + + # 创建经济优化模板文件 + economic_template_8760 = "economic_data_template_8760.xlsx" + economic_template_24 = "economic_data_template_24.xlsx" + + print("\n1. 创建经济优化Excel模板文件...") + create_excel_template(economic_template_8760, "8760") + create_excel_template(economic_template_24, "24") + + print(f"\n[OK] 经济优化Excel模板创建完成！") + print(f"[FILE] 8760小时模板: {economic_template_8760}") + print(f"[FILE] 24小时模板: {economic_template_24}") + print(f"\n[INFO] 模板包含以下工作表：") + print(f" 1. 数据 - 8760小时电力数据") + print(f" 2. 参数 - 系统运行参数") + print(f" 3. 经济参数 - 经济优化参数") + print(f" 4. 说明 - 使用说明") + print(f"\n[USAGE] 使用方法：") + print(f" uv run python economic_optimization.py --excel {economic_template_8760}") + return + print("=== Excel数据读取模块演示 ===") # 创建模板文件