基于CVXPY的水库调度优化：如何规避时段间水位超限？

核心问题拆解 #

• 原方案仅验证时段起始/结束水位，忽略时段内水位波动，导致实际运行中出现超限
• 需要将泵运行时长变量X（7个时段的整数运行分钟）拆解为30分钟间隔的运行状态，匹配流出数据的粒度，从而逐段验证水位约束

步骤1：重构30分钟间隔的运行体积矩阵 #

把每个大时段拆分为若干个30分钟子时段，同时处理部分运行的情况（比如X[0]=231分钟，对应7个完整30分钟+21分钟部分运行）：

1. 计算每个大时段包含的完整30分钟子时段数和剩余分钟数：比如480分钟的时段拆为16个完整30分钟子时段；231分钟运行时长拆为7个完整30分钟（210分钟）+1个21分钟的部分子时段
2. 基于泵的单位时间流量（假设为flow_rate，单位：体积/分钟），生成每个子时段的泵入体积：
   • 完整运行子时段：30 * flow_rate
   • 部分运行子时段：剩余分钟数 * flow_rate
   • 停机子时段：0
3. 通过CVXPY表达式将泵入体积数组与X变量关联，确保运行时长的整数约束传递到子时段

步骤2：逐30分钟子时段添加水位约束 #

将水位计算细化到每个30分钟子时段，避免中间超限：

1. 初始化初始水位initial_level
2. 遍历每个子时段，依次计算当前水位：

   current_level = previous_level + pump_in_volume - outflow_volume
   

   其中outflow_volume为对应子时段的已知流出数据
3. 为每个current_level添加上下限约束：min_level ≤ current_level ≤ max_level，确保所有子时段水位都符合要求

代码示例（关键部分） #

1. 拆分大时段为30分钟子时段 #

import cvxpy as cp
import numpy as np

# 原始时段时长（分钟）
period_durations = [480, 360, 120, 180, 90, 120, 90]
# 30分钟子时段长度
sub_period_len = 30
# 泵单位时间流量（体积/分钟）
flow_rate = 0.5  # 替换为实际值

# 计算每个大时段的子时段数量
sub_period_counts = [dur // sub_period_len for dur in period_durations]
total_sub_periods = sum(sub_period_counts)

# 定义原变量X（7个时段的运行时长，整数）
X = cp.Variable(7, integer=True)
# 基础约束：运行时长不超过时段长度、非负
X_constraints = [X[i] <= period_durations[i] for i in range(7)]
X_constraints += [X[i] >= 0 for i in range(7)]

2. 生成30分钟间隔的泵入体积数组 #

# 构建泵入体积的表达式数组
pump_in = []
for i in range(7):
    k = sub_period_counts[i]
    # 计算当前时段内完整运行的30分钟子时段数
    full_runs = cp.floor(X[i] / sub_period_len)
    # 剩余运行分钟数
    remaining = X[i] - full_runs * sub_period_len
    
    # 逐个生成子时段的泵入体积
    for sub_idx in range(k):
        if sub_idx < full_runs:
            pump_in.append(sub_period_len * flow_rate)
        elif sub_idx == full_runs and remaining > 0:
            pump_in.append(remaining * flow_rate)
        else:
            pump_in.append(0)

# 转换为CVXPY表达式数组
pump_in = cp.hstack(pump_in)

3. 逐子时段添加水位约束 #

# 已知的30分钟间隔流出数据（长度需等于total_sub_periods）
outflow = np.loadtxt("outflow_data.csv")  # 替换为实际流出数据路径
# 水位上下限
min_level = 100
max_level = 500
# 初始水位
initial_level = 200

# 构建水位序列并添加约束
current_level = initial_level
for i in range(total_sub_periods):
    current_level = current_level + pump_in[i] - outflow[i]
    X_constraints.append(current_level >= min_level)
    X_constraints.append(current_level <= max_level)

4. 构建目标函数并求解 #

# 能耗目标：假设各时段单位分钟能耗不同
cost_per_min = np.array([0.1, 0.2, 0.15, 0.18, 0.22, 0.16, 0.2])  # 替换为实际能耗系数
objective = cp.Minimize(cp.sum(cost_per_min @ X))

# 构建问题并求解（选择支持整数规划的求解器）
prob = cp.Problem(objective, X_constraints)
prob.solve(solver=cp.GLPK_MI)

# 输出结果
print("最优泵运行时长：", X.value)

关键注意事项 #

• 确保流出数据的长度与总子时段数完全匹配，避免维度错误
• 选择适配的整数规划求解器（如GLPK_MI、CBC、Gurobi），CVXPY对不同求解器的支持存在差异
• 若整数约束导致求解效率低下，可先松弛为连续变量验证逻辑，再添加整数约束

内容的提问来源于stack exchange，提问作者Aidan Donnelly