multi_energy_complementarity/solar_optimization.py

"""
光伏出力优化模块

该模块通过调整光伏出力曲线的系数，在给定的系统参数条件下
最小化与电网交换的电量，提高系统的自平衡能力。

作者: 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()