""" 光伏出力优化模块 该模块通过调整光伏出力曲线的系数，在给定的系统参数条件下 最小化与电网交换的电量，提高系统的自平衡能力。 作者: iFlow CLI 创建日期: 2025-12-26 """ import numpy as np from typing import List, Dict, Tuple, Optional from dataclasses import dataclass from storage_optimization import SystemParameters, optimize_storage_capacity, calculate_energy_balance @dataclass class SolarOptimizationResult: """光伏优化结果类""" optimal_solar_coefficient: float # 最优光伏系数 original_solar_output: List[float] # 原始光伏出力曲线 optimized_solar_output: List[float] # 优化后光伏出力曲线 min_grid_exchange: float # 最小电网交换电量 grid_purchase: float # 购电量 grid_feed_in: float # 上网电量 storage_result: Dict # 储能优化结果 optimization_history: List[Dict] # 优化历史记录 def calculate_grid_exchange_metric( solar_output: List[float], wind_output: List[float], thermal_output: List[float], load_demand: List[float], params: SystemParameters ) -> Dict[str, float]: """ 计算电网交换电量指标 Args: solar_output: 光伏出力曲线 (MW) wind_output: 风电出力曲线 (MW) thermal_output: 火电出力曲线 (MW) load_demand: 负荷曲线 (MW) params: 系统参数配置 Returns: 包含电网交换指标的字典 """ # 计算最优储能容量 storage_result = optimize_storage_capacity( solar_output, wind_output, thermal_output, load_demand, params ) # 计算电网交换电量 grid_feed_in = storage_result['grid_feed_in'] # 分离购电和上网电量 total_purchase = sum(-x for x in grid_feed_in if x < 0) # 购电量（正值） total_feed_in = sum(x for x in grid_feed_in if x > 0) # 上网电量（正值） # 计算总交换电量（购电 + 上网） total_exchange = total_purchase + total_feed_in return { 'total_exchange': total_exchange, 'grid_purchase': total_purchase, 'grid_feed_in': total_feed_in, 'storage_capacity': storage_result['required_storage_capacity'], 'storage_result': storage_result } def optimize_solar_output( original_solar_output: List[float], wind_output: List[float], thermal_output: List[float], load_demand: List[float], params: SystemParameters, coefficient_range: Tuple[float, float] = (0.1, 2.0), tolerance: float = 0.01, max_iterations: int = 50 ) -> SolarOptimizationResult: """ 优化光伏出力系数以最小化电网交换电量 Args: original_solar_output: 原始光伏出力曲线 (MW) wind_output: 风电出力曲线 (MW) thermal_output: 火电出力曲线 (MW) load_demand: 负荷曲线 (MW) params: 系统参数配置 coefficient_range: 光伏系数搜索范围 (最小值, 最大值) tolerance: 收敛容差 max_iterations: 最大迭代次数 Returns: 光伏优化结果 """ print("开始光伏出力优化...") # 初始化优化历史 optimization_history = [] # 使用黄金分割法进行一维优化 phi = (1 + np.sqrt(5)) / 2 # 黄金比例 resphi = 2 - phi a, b = coefficient_range c = b - resphi * (b - a) d = a + resphi * (b - a) # 计算初始点的目标函数值 fc = calculate_grid_exchange_metric( [x * c for x in original_solar_output], wind_output, thermal_output, load_demand, params ) fd = calculate_grid_exchange_metric( [x * d for x in original_solar_output], wind_output, thermal_output, load_demand, params ) # 记录初始点 optimization_history.append({ 'coefficient': c, 'total_exchange': fc['total_exchange'], 'grid_purchase': fc['grid_purchase'], 'grid_feed_in': fc['grid_feed_in'], 'storage_capacity': fc['storage_capacity'] }) optimization_history.append({ 'coefficient': d, 'total_exchange': fd['total_exchange'], 'grid_purchase': fd['grid_purchase'], 'grid_feed_in': fd['grid_feed_in'], 'storage_capacity': fd['storage_capacity'] }) # 黄金分割搜索 for iteration in range(max_iterations): if abs(fc['total_exchange'] - fd['total_exchange']) < tolerance: break if fc['total_exchange'] < fd['total_exchange']: b = d d = c fd = fc c = b - resphi * (b - a) fc = calculate_grid_exchange_metric( [x * c for x in original_solar_output], wind_output, thermal_output, load_demand, params ) optimization_history.append({ 'coefficient': c, 'total_exchange': fc['total_exchange'], 'grid_purchase': fc['grid_purchase'], 'grid_feed_in': fc['grid_feed_in'], 'storage_capacity': fc['storage_capacity'] }) else: a = c c = d fc = fd d = a + resphi * (b - a) fd = calculate_grid_exchange_metric( [x * d for x in original_solar_output], wind_output, thermal_output, load_demand, params ) optimization_history.append({ 'coefficient': d, 'total_exchange': fd['total_exchange'], 'grid_purchase': fd['grid_purchase'], 'grid_feed_in': fd['grid_feed_in'], 'storage_capacity': fd['storage_capacity'] }) # 确定最优系数 if fc['total_exchange'] < fd['total_exchange']: optimal_coefficient = c best_result = fc else: optimal_coefficient = d best_result = fd # 生成优化后的光伏出力曲线 optimized_solar_output = [x * optimal_coefficient for x in original_solar_output] # 重新计算完整的最优储能配置 final_storage_result = optimize_storage_capacity( optimized_solar_output, wind_output, thermal_output, load_demand, params ) print(f"优化完成！最优光伏系数: {optimal_coefficient:.3f}") print(f"最小电网交换电量: {best_result['total_exchange']:.2f} MWh") return SolarOptimizationResult( optimal_solar_coefficient=optimal_coefficient, original_solar_output=original_solar_output, optimized_solar_output=optimized_solar_output, min_grid_exchange=best_result['total_exchange'], grid_purchase=best_result['grid_purchase'], grid_feed_in=best_result['grid_feed_in'], storage_result=final_storage_result, optimization_history=optimization_history ) def export_optimization_results(result: SolarOptimizationResult, filename: str = None): """ 导出光伏优化结果到Excel文件 Args: result: 光伏优化结果 filename: 输出文件名，如果为None则自动生成 """ import pandas as pd from datetime import datetime if filename is None: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"solar_optimization_results_{timestamp}.xlsx" print(f"正在导出光伏优化结果到Excel文件: {filename}") hours = list(range(1, len(result.original_solar_output) + 1)) # 创建主要数据DataFrame data_df = pd.DataFrame({ '小时': hours, '原始光伏出力(MW)': result.original_solar_output, '优化后光伏出力(MW)': result.optimized_solar_output, '出力变化(MW)': [result.optimized_solar_output[i] - result.original_solar_output[i] for i in range(len(result.original_solar_output))], '变化比例(%)': [(result.optimized_solar_output[i] / result.original_solar_output[i] - 1) * 100 if result.original_solar_output[i] > 0 else 0 for i in range(len(result.original_solar_output))] }) # 创建优化结果摘要DataFrame summary_df = pd.DataFrame({ '指标': [ '最优光伏系数', '最小电网交换电量', '购电量', '上网电量', '所需储能容量', '优化后弃风率', '优化后弃光率', '优化后上网电量比例' ], '数值': [ f"{result.optimal_solar_coefficient:.3f}", f"{result.min_grid_exchange:.2f} MWh", f"{result.grid_purchase:.2f} MWh", f"{result.grid_feed_in:.2f} MWh", f"{result.storage_result['required_storage_capacity']:.2f} MWh", f"{result.storage_result['total_curtailment_wind_ratio']:.3f}", f"{result.storage_result['total_curtailment_solar_ratio']:.3f}", f"{result.storage_result['total_grid_feed_in_ratio']:.3f}" ] }) # 创建优化历史DataFrame history_df = pd.DataFrame(result.optimization_history) history_df.columns = ['光伏系数', '电网交换电量(MWh)', '购电量(MWh)', '上网电量(MWh)', '储能容量(MWh)'] # 写入Excel文件 with pd.ExcelWriter(filename, engine='openpyxl') as writer: # 写入主要数据 data_df.to_excel(writer, sheet_name='出力曲线对比', index=False) # 写入优化结果摘要 summary_df.to_excel(writer, sheet_name='优化结果摘要', index=False) # 写入优化历史 history_df.to_excel(writer, sheet_name='优化历史', index=False) # 创建说明工作表 description_df = pd.DataFrame({ '项目': [ '文件说明', '生成时间', '优化目标', '优化方法', '数据长度', '注意事项' ], '内容': [ '光伏出力优化结果 - 通过调整光伏系数最小化电网交换电量', datetime.now().strftime("%Y-%m-%d %H:%M:%S"), '最小化与电网交换的总电量（购电 + 上网）', '黄金分割一维优化算法', f"{len(result.original_solar_output)} 小时", '优化结果在给定的系统参数约束下得出' ] }) description_df.to_excel(writer, sheet_name='说明', index=False) print(f"光伏优化结果已成功导出到: {filename}") return filename def main(): """主函数，提供示例使用""" # 示例数据 original_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 wind_output = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0] * 4 thermal_output = [5.0] * 24 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] # 系统参数 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, rated_thermal_capacity=100.0, rated_solar_capacity=100.0, rated_wind_capacity=100.0, available_thermal_energy=2400.0, available_solar_energy=600.0, available_wind_energy=1200.0 ) # 执行光伏优化 result = optimize_solar_output( original_solar_output, wind_output, thermal_output, load_demand, params ) # 打印结果 print("\n=== 光伏出力优化结果 ===") print(f"最优光伏系数: {result.optimal_solar_coefficient:.3f}") print(f"最小电网交换电量: {result.min_grid_exchange:.2f} MWh") print(f"其中购电量: {result.grid_purchase:.2f} MWh") print(f"其中上网电量: {result.grid_feed_in:.2f} MWh") print(f"所需储能容量: {result.storage_result['required_storage_capacity']:.2f} MWh") print(f"优化后弃风率: {result.storage_result['total_curtailment_wind_ratio']:.3f}") print(f"优化后弃光率: {result.storage_result['total_curtailment_solar_ratio']:.3f}") print(f"优化后上网电量比例: {result.storage_result['total_grid_feed_in_ratio']:.3f}") # 导出结果 export_optimization_results(result) return result if __name__ == "__main__": main()