facat/multi_energy_complementarity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修复了储能电量不平衡的问题。

Browse Source
This commit is contained in:
dmy
2025-12-27 12:25:01 +08:00
parent a522132ede
commit 164b9da026
2 changed files with 251 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

263
src/storage_optimization.py
View File

@@ -109,11 +109,12 @@ def calculate_energy_balance(
thermal_output: List[float],
load_demand: List[float],
params: SystemParameters,
storage_capacity: float
storage_capacity: float,
initial_soc: float = 0.0
) -> Dict[str, List[float]]:
"""
计算给定储能容量下的系统电能平衡
Args:
solar_output: 光伏出力曲线 (MW) - 支持24小时或8760小时
wind_output: 风电出力曲线 (MW) - 支持24小时或8760小时
@@ -121,7 +122,8 @@ def calculate_energy_balance(
load_demand: 负荷曲线 (MW) - 支持24小时或8760小时
params: 系统参数配置
storage_capacity: 储能容量 (MWh)
initial_soc: 初始储能状态 (MWh)，默认为0.0
Returns:
包含各种功率曲线的字典
"""
@@ -130,7 +132,7 @@ def calculate_energy_balance(
wind = np.array(wind_output)
thermal = np.array(thermal_output)
load = np.array(load_demand)
# 初始化输出数组
hours = len(solar_output)
storage_soc = np.zeros(hours) # 储能状态 (MWh)
@@ -139,16 +141,19 @@ def calculate_energy_balance(
curtailed_wind = np.zeros(hours) # 弃风量 (MW)
curtailed_solar = np.zeros(hours) # 弃光量 (MW)
grid_feed_in = np.zeros(hours) # 上网电量 (MW)
# 设置初始储能状态
storage_soc[0] = initial_soc
# 计算总发电潜力
total_potential_wind = np.sum(wind)
total_potential_solar = np.sum(solar)
# 判断是否只有一种可再生能源
# 判断是否只有一种可再生能源
has_wind = total_potential_wind > 0
has_solar = total_potential_solar > 0
single_renewable = (has_wind and not has_solar) or (has_solar and not has_wind)
# 计算允许的最大弃风弃光量
if single_renewable:
# 只有一种可再生能源时，弃电量不受限制
@@ -162,32 +167,32 @@ def calculate_energy_balance(
# 有多种可再生能源且上网电量限制不为0时，应用弃风弃光限制
max_curtailed_wind_total = total_potential_wind * params.max_curtailment_wind
max_curtailed_solar_total = total_potential_solar * params.max_curtailment_solar
# 初始化累计弃风弃光量
accumulated_curtailed_wind = 0.0
accumulated_curtailed_solar = 0.0
# 计算总可用发电量上限（不考虑火电）
total_available_energy = params.available_solar_energy + params.available_wind_energy
max_total_grid_feed_in = total_available_energy * params.max_grid_ratio
# 初始化累计上网电量
cumulative_grid_feed_in = 0.0
# 逐小时计算
for hour in range(hours):
# 确保储能状态不为负
storage_soc[hour] = max(0, storage_soc[hour])
# 确保储能状态不为负且不超过容量
storage_soc[hour] = max(0, min(storage_capacity, storage_soc[hour]))
# 可用发电量（未考虑弃风弃光）
available_generation = thermal[hour] + wind[hour] + solar[hour]
# 需求电量（负荷）
demand = load[hour]
# 计算功率平衡
power_surplus = available_generation - demand
if power_surplus > 0:
# 有盈余电力，优先储能
max_charge = min(
@@ -195,30 +200,30 @@ def calculate_energy_balance(
storage_capacity * params.charge_rate, # 充电功率限制
power_surplus # 可用盈余电力
)
# 实际充电功率
actual_charge = min(max_charge, power_surplus)
charge_power[hour] = actual_charge
# 更新储能状态（考虑充电效率）
if hour < hours - 1:
storage_soc[hour + 1] = storage_soc[hour] + actual_charge * params.storage_efficiency
# 剩余电力优先上网，超出上网电量比例限制时才弃风弃光
remaining_surplus = power_surplus - actual_charge
# 计算当前允许的最大上网电量
# 基于总可用发电量和已累计上网电量
remaining_grid_quota = max_total_grid_feed_in - cumulative_grid_feed_in
# 优先上网，但不超过剩余配额
grid_feed_allowed = min(remaining_surplus, max(0, remaining_grid_quota))
grid_feed_in[hour] = grid_feed_allowed
cumulative_grid_feed_in += grid_feed_allowed
# 剩余电力考虑弃风弃光
remaining_surplus -= grid_feed_allowed
# 计算弃风弃光（优先弃光，然后弃风）
if remaining_surplus > 0:
# 在单一可再生能源场景下，弃风弃光不受限制
@@ -228,7 +233,7 @@ def calculate_energy_balance(
curtailed_solar[hour] = min(solar[hour], remaining_surplus)
remaining_surplus -= curtailed_solar[hour]
accumulated_curtailed_solar += curtailed_solar[hour]
# 如果还有剩余，弃风
if remaining_surplus > 0 and wind[hour] > 0:
curtailed_wind[hour] = min(wind[hour], remaining_surplus)
@@ -246,12 +251,12 @@ def calculate_energy_balance(
solar[hour],
max_curtailed_solar_total - accumulated_curtailed_solar
)
if available_solar_curtail > 0:
curtailed_solar[hour] = min(available_solar_curtail, remaining_surplus)
remaining_surplus -= curtailed_solar[hour]
accumulated_curtailed_solar += curtailed_solar[hour]
# 如果还有剩余，弃风
if remaining_surplus > 0:
if max_curtailed_wind_total == float('inf'):
@@ -263,45 +268,45 @@ def calculate_energy_balance(
wind[hour],
max_curtailed_wind_total - accumulated_curtailed_wind
)
if available_wind_curtail > 0:
curtailed_wind[hour] = min(available_wind_curtail, remaining_surplus)
remaining_surplus -= curtailed_wind[hour]
accumulated_curtailed_wind += curtailed_wind[hour]
# 确保电力平衡：如果仍有剩余电力，强制弃掉（安全机制）
if remaining_surplus > 0:
# 记录警告但不影响计算
# 在实际系统中，这种情况不应该发生，但作为安全保护
pass
else:
# 电力不足，优先放电
power_deficit = -power_surplus
grid_feed_in[hour] = 0 # 初始化购电为0
max_discharge = min(
storage_soc[hour], # 储能状态限制
storage_capacity * params.discharge_rate, # 放电功率限制
power_deficit # 缺电功率
)
# 实际放电功率
actual_discharge = min(max_discharge, power_deficit)
discharge_power[hour] = actual_discharge
# 更新储能状态（考虑放电效率）
if hour < hours - 1:
storage_soc[hour + 1] = storage_soc[hour] - actual_discharge / params.storage_efficiency
# 计算剩余缺电，需要从电网购电
remaining_deficit = power_deficit - actual_discharge
# 如果还有缺电，从电网购电
if remaining_deficit > 0:
# 购电功率为负值，表示从电网输入
grid_feed_in[hour] = -remaining_deficit
return {
'storage_profile': storage_soc.tolist(),
'charge_profile': charge_power.tolist(),
@@ -312,6 +317,83 @@ def calculate_energy_balance(
}
def find_periodic_steady_state(
solar_output: List[float],
wind_output: List[float],
thermal_output: List[float],
load_demand: List[float],
params: SystemParameters,
storage_capacity: float,
soc_convergence_threshold: float = 0.001,
max_iterations: int = 100
) -> Dict[str, List[float]]:
"""
通过迭代找到满足周期性平衡的储能初始状态
步骤：
1. 从初始SOC=0开始，运行一次全年仿真，记录最后一小时的SOC值
2. 将这个SOC值作为新的"初始SOC"，再次运行仿真
3. 重复上述过程，直到首尾SOC的差值小于设定的阈值
Args:
solar_output: 光伏出力曲线 (MW) - 支持24小时或8760小时
wind_output: 风电出力曲线 (MW) - 支持24小时或8760小时
thermal_output: 火电出力曲线 (MW) - 支持24小时或8760小时
load_demand: 负荷曲线 (MW) - 支持24小时或8760小时
params: 系统参数配置
storage_capacity: 储能容量 (MWh)
soc_convergence_threshold: SOC收敛阈值（相对于容量的比例），默认0.1%
max_iterations: 最大迭代次数
Returns:
包含各种功率曲线的字典，且满足周期性平衡条件
"""
# 计算收敛阈值的绝对值
absolute_threshold = storage_capacity * soc_convergence_threshold
# 初始SOC从0开始
initial_soc = 0.0
iteration = 0
soc_diff = float('inf')
print(f"正在寻找周期性平衡状态（SOC收敛阈值: {absolute_threshold:.4f} MWh）...")
while iteration < max_iterations and soc_diff > absolute_threshold:
# 运行仿真
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand,
params, storage_capacity, initial_soc
)
# 获取初始和最终的SOC
storage_initial = balance_result['storage_profile'][0]
storage_final = balance_result['storage_profile'][-1]
# 计算SOC差值
soc_diff = abs(storage_final - storage_initial)
# 更新初始SOC（使用最终SOC作为下一次的初始SOC）
initial_soc = storage_final
# 确保SOC在合理范围内
initial_soc = max(0, min(storage_capacity, initial_soc))
iteration += 1
# 输出迭代信息（每10次迭代或最后一次）
if iteration % 10 == 0 or iteration == 1 or soc_diff <= absolute_threshold:
print(f" 迭代 {iteration}: 初始SOC={storage_initial:.4f} MWh, "
f"最终SOC={storage_final:.4f} MWh, 差值={soc_diff:.4f} MWh")
# 输出收敛结果
if soc_diff <= absolute_threshold:
print(f"✓ 周期性平衡收敛成功（迭代{iteration}次，SOC差值={soc_diff:.4f} MWh）")
else:
print(f"⚠ 未达到收敛条件（迭代{iteration}次，SOC差值={soc_diff:.4f} MWh）")
return balance_result
def check_constraints(
solar_output: List[float],
wind_output: List[float],
@@ -321,14 +403,14 @@ def check_constraints(
) -> Dict[str, float]:
"""
检查约束条件是否满足
Args:
solar_output: 光伏出力曲线 (MW) - 支持24小时或8760小时
wind_output: 风电出力曲线 (MW) - 支持24小时或8760小时
thermal_output: 火电出力曲线 (MW) - 支持24小时或8760小时
balance_result: 电能平衡计算结果
params: 系统参数配置
Returns:
包含各约束实际比例的字典
"""
@@ -336,21 +418,21 @@ def check_constraints(
total_wind_potential = sum(wind_output)
total_solar_potential = sum(solar_output)
total_thermal = sum(thermal_output)
total_curtailed_wind = sum(balance_result['curtailed_wind'])
total_curtailed_solar = sum(balance_result['curtailed_solar'])
total_grid_feed_in = sum(balance_result['grid_feed_in'])
# 实际发电量（考虑弃风弃光）
actual_wind_generation = total_wind_potential - total_curtailed_wind
actual_solar_generation = total_solar_potential - total_curtailed_solar
total_generation = total_thermal + actual_wind_generation + actual_solar_generation
# 计算比例
actual_curtailment_wind_ratio = total_curtailed_wind / total_wind_potential if total_wind_potential > 0 else 0
actual_curtailment_solar_ratio = total_curtailed_solar / total_solar_potential if total_solar_potential > 0 else 0
actual_grid_feed_in_ratio = total_grid_feed_in / total_generation if total_generation > 0 else 0
return {
'total_curtailment_wind_ratio': actual_curtailment_wind_ratio,
'total_curtailment_solar_ratio': actual_curtailment_solar_ratio,
@@ -369,7 +451,7 @@ def optimize_storage_capacity(
) -> Dict:
"""
优化储能容量，使用迭代方法寻找满足所有约束的最小储能容量
Args:
solar_output: 光伏出力曲线 (MW) - 支持24小时或8760小时
wind_output: 风电出力曲线 (MW) - 支持24小时或8760小时
@@ -378,39 +460,54 @@ def optimize_storage_capacity(
params: 系统参数配置
max_iterations: 最大迭代次数
tolerance: 收敛容差
Returns:
包含优化结果的字典
"""
# 验证输入
validate_inputs(solar_output, wind_output, thermal_output, load_demand, params)
# 初始化搜索范围
lower_bound = 0.0
theoretical_max = max(sum(solar_output) + sum(wind_output) + sum(thermal_output), sum(load_demand))
# 应用储能容量上限限制
if params.max_storage_capacity is not None:
upper_bound = min(theoretical_max, params.max_storage_capacity)
else:
upper_bound = theoretical_max
# 判断数据类型（24小时或8760小时）
data_length = len(solar_output)
is_yearly_data = data_length == 8760
if is_yearly_data:
print(f"处理8760小时全年数据，启用周期性平衡优化...")
# 二分搜索寻找最小储能容量
best_capacity = upper_bound
best_result = None
solution_found = False # 标记是否找到可行解
for iteration in range(max_iterations):
mid_capacity = (lower_bound + upper_bound) / 2
# 计算当前容量下的平衡
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand, params, mid_capacity
)
# 对于8760小时数据，使用周期性平衡函数
# 对于24小时数据，使用普通平衡函数（初始SOC=0）
if is_yearly_data:
balance_result = find_periodic_steady_state(
solar_output, wind_output, thermal_output, load_demand,
params, mid_capacity
)
else:
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand, params, mid_capacity
)
# 检查约束条件
constraint_results = check_constraints(solar_output, wind_output, thermal_output, balance_result, params)
# 检查是否满足所有约束
# max_grid_ratio只限制上网电量比例，不约束购电
# 只有当grid_feed_in为正时（上网）才需要检查约束
@@ -421,16 +518,16 @@ def optimize_storage_capacity(
else:
# 没有上网电量或为负值（购电），总是满足约束
grid_constraint_satisfied = True
# 判断是否只有一种可再生能源
has_wind = sum(wind_output) > 0
has_solar = sum(solar_output) > 0
single_renewable = (has_wind and not has_solar) or (has_solar and not has_wind)
# 特殊情况：当上网电量限制为0时，所有超额电力都必须被弃掉
# 此时应该允许无限制弃风弃光
grid_quota_zero = params.max_grid_ratio == 0
if single_renewable or grid_quota_zero:
# 只有一种可再生能源时，或上网电量限制为0时，跳过弃风弃光约束检查
constraints_satisfied = grid_constraint_satisfied
@@ -441,12 +538,12 @@ def optimize_storage_capacity(
constraint_results['total_curtailment_solar_ratio'] <= params.max_curtailment_solar and
grid_constraint_satisfied
)
# 检查储能日平衡（周期结束时储能状态应接近初始值）
storage_initial = balance_result['storage_profile'][0]
storage_final = balance_result['storage_profile'][-1]
daily_balance = abs(storage_final - storage_initial) < tolerance
if constraints_satisfied and daily_balance:
# 满足条件，尝试减小容量
best_capacity = mid_capacity
@@ -456,32 +553,44 @@ def optimize_storage_capacity(
else:
# 不满足条件，增大容量
lower_bound = mid_capacity
# 检查收敛
if upper_bound - lower_bound < tolerance:
break
# 处理储能容量上限限制的情况
if not solution_found and params.max_storage_capacity is not None:
print(f"警告：在储能容量上限 {params.max_storage_capacity:.2f} MWh 内无法找到满足所有约束的解")
print("使用最大允许容量进行计算，但某些约束条件可能无法满足")
# 使用最大允许容量计算结果
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand, params, params.max_storage_capacity
)
if is_yearly_data:
balance_result = find_periodic_steady_state(
solar_output, wind_output, thermal_output, load_demand,
params, params.max_storage_capacity
)
else:
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand, params, params.max_storage_capacity
)
constraint_results = check_constraints(solar_output, wind_output, thermal_output, balance_result, params)
best_result = {**balance_result, **constraint_results}
best_capacity = params.max_storage_capacity
elif best_result is None:
# 如果没有找到可行解（且没有容量上限限制），使用最大容量
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand, params, upper_bound
)
if is_yearly_data:
balance_result = find_periodic_steady_state(
solar_output, wind_output, thermal_output, load_demand,
params, upper_bound
)
else:
balance_result = calculate_energy_balance(
solar_output, wind_output, thermal_output, load_demand, params, upper_bound
)
constraint_results = check_constraints(solar_output, wind_output, thermal_output, balance_result, params)
best_result = {**balance_result, **constraint_results}
best_capacity = upper_bound
# 添加能量平衡校验
total_generation = sum(thermal_output) + sum(wind_output) + sum(solar_output)
total_consumption = sum(load_demand)
@@ -490,12 +599,12 @@ def optimize_storage_capacity(
total_charge = sum(best_result['charge_profile'])
total_discharge = sum(best_result['discharge_profile'])
storage_net_change = best_result['storage_profile'][-1] - best_result['storage_profile'][0]
# 能量平衡校验：发电量 + 放电量/效率 = 负荷 + 充电量*效率 + 弃风弃光 + 上网电量
# 考虑储能充放电效率的能量平衡
energy_from_storage = total_discharge / params.storage_efficiency # 储能提供的有效能量
energy_to_storage = total_charge * params.storage_efficiency # 储能消耗的电网能量
# 能量平衡校验：应该接近0，但允许一定误差
# 当total_grid为负时（购电），应该加到左侧（供给侧）
# 当total_grid为正时（上网），应该加到右侧（需求侧）
@@ -512,7 +621,15 @@ def optimize_storage_capacity(
# 储能效率损失可能达到总能量的10%以上
tolerance = max(10.0, total_generation * 0.15)
energy_balance_check = energy_balance_error < tolerance
# 输出周期性平衡信息
if is_yearly_data:
soc_initial_final_diff = abs(best_result['storage_profile'][-1] - best_result['storage_profile'][0])
print(f"\n周期性平衡信息:")
print(f" 初始SOC: {best_result['storage_profile'][0]:.4f} MWh")
print(f" 最终SOC: {best_result['storage_profile'][-1]:.4f} MWh")
print(f" SOC差值: {soc_initial_final_diff:.4f} MWh")
# 返回最终结果
return {
'required_storage_capacity': best_capacity,