如何解决Battery Storage Pyomo：在 365 小时的时间范围内优化和迭代年度数据

我有名为“HOEP”的年度电价数据。使用我的 pyomo 模型，我想确定电池全年的行为，但有 365 小时的时间范围（能量输入 = Ein 和能量输出 = Eout）。换句话说，我想让我的算法在前 365 小时内运行，然后在下一个 365 小时时间范围内再次运行，初始电池状态等于前一个时间范围内的最后一个小时。

我尝试将我的年度数据分成多个块（一年 365 小时的 24 个块）。使用 df_list = np.vsplit(dfa,24)，我创建了一个块列表并将它们转换为 24 个不同的数据帧。然后，在我的模型循环数据之前，我使用 for idx,df in enumerate([df0,df1,df2]),（这里只有 3 个块用于测试）。但是，当我查看结果时，似乎该模型仅针对 enumerate([df0,df2]) 的最后一个参数 df2 进行了优化。

有人知道为什么它不适用于 3 个块吗？或者我怎么能以不同的方式做到这一点？

预先感谢您的帮助！

这是我现在可以使用的代码的编辑版本，但我知道这可能不是最pythonic的方法。

import numpy as np
import pandas as pd
from typing import List
from itertools import chain
from pyomo.environ import *

output = []

for idx,df2]):

    model = ConcreteModel()
    
    # Variables of the model
    model.T = Set(initialize=df.hour.tolist(),ordered=True)
    model.Rmax = Param(initialize=1,within=Any) 
    model.Smax = Param(initialize=5,within=Any)
    model.Dmax = Param(initialize=5,within=Any) 

    model.Ein = Var(model.T,domain=NonNegativeReals)  
    model.Eout = Var(model.T,domain=NonNegativeReals) 
    model.Z = Var(model.T,domain=NonNegativeReals)  
    model.L = Var(model.T,domain=NonNegativeReals) 
    model.nes = Var(model.T)

    # Constraints
    def storage_state(model,t):
        if t == model.T.first():
            return model.Z[t] == 0
        else:
            return (model.Z[t] == (model.Z[t-1] + (model.Ein[t]) - (model.Eout[t])))                         
    model.charge_state = Constraint(model.T,rule=storage_state)

    def discharge_constraint(model,t):
        return model.Eout[t] <= model.Rmax
    model.discharge = Constraint(model.T,rule=discharge_constraint)
    
    def charge_constraint(model,t):
        return model.Ein[t] <= model.Rmax
    model.charge = Constraint(model.T,rule=charge_constraint)

    def positive_charge(model,t): 
        return model.Eout[t] <= model.Z[t] 
    model.positive_charge = Constraint(model.T,rule=positive_charge)

    def max_SOC(model,t): 
        return model.Z[t] <= model.Smax
    model.max_SOC = Constraint(model.T,rule=max_SOC)

    def demand_constraint(model,t):
        return (model.L[t] == (df.loc[t,'MktDemand'] + (model.Ein[t]) - (model.Eout[t]))) 
    model.demand_constraint = Constraint(model.T,rule=demand_constraint)

    def discharge_limit(model,t):    
        max_t = model.T.last()
        if t < max_t - 24:
            return sum(model.Eout[i] for i in range(t,t+24)) <= model.Dmax
        else:
            return Constraint.Skip
    model.limit_disch_out = Constraint(model.T,rule=discharge_limit)

    def charge_limit(model,t):    
        max_t = model.T.last()
        if t < max_t - 24:
            return sum(model.Ein[i] for i in range(t,t+24)) <= model.Dmax
        else:
            return Constraint.Skip
    model.limit_charg_out = Constraint(model.T,rule=charge_limit)

    def Net_energy_sold(model,t):
        return model.nes[t] == ((model.Eout[t] - model.Ein[t]) / model.Rmax * 100)                       
    model.net_energy = Constraint(model.T,rule=Net_energy_sold)

    # Objective function and optimization
    income = sum(df.loc[t,'HOEP'] * model.Eout[t] for t in model.T)
    expenses = sum(df.loc[t,'HOEP'] * model.Ein[t] for t in model.T)
    profits = (income - expenses) 
    model.objective = Objective(expr=profits,sense=maximize)

    # Solve model
    solver = SolverFactory('glpk')
    solver.solve(model) 
    
    # Extract model output in list
    Date = list(df['Date'])
    output.append([Date,model.Ein.get_values().values(),model.Eout.get_values().values(),model.Z.get_values().values(),model.nes.get_values().values(),model.L.get_values().values()])
    
df_results = pd.DataFrame(output)
df_results.rename(columns = {0: 'Date',1: 'Ein',2:'Eout',3:'Z',4:'nes',5:'Load'},inplace = True)
df_results

# Present final results in dataframe
d = ein = eout = z = l = nes = []

for i in list(df_results.index):
    d = d + list(df_results.loc[i,'Date'])
    ein = ein + list(df_results.loc[i,'Ein'])
    eout = eout + list(df_results.loc[i,'Eout'])
    z = z + list(df_results.loc[i,'Z'])
    nes = nes + list(df_results.loc[i,'nes'])
    l = l + list(df_results.loc[i,'Load'])

results = pd.DataFrame(zip(d,ein,eout,z,nes,l),columns = ['Date','Ein','Eout','SOC','nes','Load'])
results 


# Returned dataframe

            Date  Ein  Eout   SOC    nes      Load
0     2019-01-01  0.0  0.00  0.00    0.0  16231.00
1     2019-01-01  0.0  0.00  0.00    0.0  16051.00
2     2019-01-01  1.0  0.00  1.00 -100.0  15806.00
3     2019-01-01  1.0  0.00  2.00 -100.0  15581.00

...

解决方法

为什么它不起作用

（免责声明：这是我看到的一个问题，可能还有其他问题）。

在 for 循环的每次迭代中，list_of_series 都是从头开始定义的，因此之前迭代中获得的所有结果都丢失了。

我还会检查 df.hour 是“一年中的小时”或“从数据开始的小时”而不是“一天中的小时”（如果是后者，这也会导致错误） .

解决问题

（显然有几种解决方案）在 for 循环的每次迭代中，将 list_of_series 变成 pd.DataFrame，并将数据帧附加到列表中。在 for 循环结束时（一旦您在每个数据块上运行了模型），连接数据帧列表。

from typing import List

...

# find a better name,variable names shouldn't specify the type
list_of_dataframes: List[pd.DataFrame] = []
for ...:  # for each chunk of data
    ...  # create model,solve
    list_of_series = ...
    list_of_dataframes.append(pd.DataFrame(list_of_series))

results = pd.concat(list_of_dataframes,axis=0) # use `ignore_index=True` if needed

一些提示

将代码分解为函数。创建一个定义模型的函数。这突出了输入和输出是什么，使 for 循环更具可读性，允许您在其他上下文中使用它并可能对其进行测试。
（自以为是）将您的“数据”设置为模型的参数，而不是直接使用它们来构建约束和目标函数。这使您可以在模型中将每条数据提取到一个位置，在模型中创建内部一致性，并允许您纯粹基于优化模型提取结果。
将 I/O（读取/写入文件）与其余代码分开。如果您的数据源更改了格式或文件类型，您将能够在不更改任何其余代码的情况下进行更改。

def main(input_data: pd.DataFrame) -> pd.DataFrame:
    # group by week,month,or any applicable resolution
    # this assumes the index is a `pd.DatetimeIndex`
    # `MS` is "Month Start" - watch out with weeks because `freq="w"` starts
    # on Mondays,and your data might start on a different weekday.
    # If you want split into chunks of some number of days,# use e.g. `freq="14d"`
    grouped = df.groupby(pd.Grouper(freq="MS"))
    
    results_list: List[pd.DataFrame] = []
    for month,group in grouped:
        model = create_model(group)
        optimization_results = SolverFactory('glpk').solve(model)
        results_list.append(extract_results(model)) # pass `group` if needed
    results = pd.concat(results_list,axis=0,ignore_index=True)
    return results


def create_model(df: pd.DataFrame) -> ConcreteModel:
    # NOTE: instead of hard-coding parameters such as battery capacity,# pass them as inputs to the function.
    model = ConcreteModel()
    ...

    return model


def extract_results(model: ConcreteModel) -> pd.DataFrame:
    ...


def load_data(filename) -> pd.DataFrame:
    ...


if __name__ == "__main__":
     input_data = load_data(...)
     results = main(input_data)
     results.to_csv(...)