当前位置：首页 > news >正文

多时间尺度的配电网深度强化学习无功优化策略的Python示例代码框架

news 来源：原创 2025/5/20 12:54:17

以下是一个简单的多时间尺度的配电网深度强化学习无功优化策略的Python示例代码框架，用于帮助你理解如何使用深度强化学习（以深度Q网络 DQN 为例）来处理配电网的无功优化问题。在实际应用中，你可能需要根据具体的配电网模型和需求进行大量的修改和扩展。

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt


# 定义深度Q网络模型
class DQN(nn.Module):
    def __init__(self, state_size, action_size):
        super(DQN, self).__init__()
        self.fc1 = nn.Linear(state_size, 128)
        self.fc2 = nn.Linear(128, 128)
        self.fc3 = nn.Linear(128, action_size)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x


# 定义环境类，这里简化为配电网无功优化的环境抽象
class DistributionNetworkEnv:
    def __init__(self, num_buses, time_steps):
        self.num_buses = num_buses
        self.time_steps = time_steps
        self.current_time_step = 0
        self.state = np.zeros((self.num_buses,))  # 示例状态，可根据实际定义

    def step(self, action):
        # 这里需要根据实际的配电网模型计算奖励、下一个状态和是否结束
        reward = 0
        next_state = self.state  # 示例，需实际计算
        done = self.current_time_step >= self.time_steps - 1
        self.current_time_step += 1
        return next_state, reward, done

    def reset(self):
        self.current_time_step = 0
        self.state = np.zeros((self.num_buses,))
        return self.state


# 训练函数
def train_dqn(env, dqn, target_dqn, memory, batch_size, gamma, tau, episodes):
    optimizer = optim.Adam(dqn.parameters(), lr=0.001)
    loss_fn = nn.MSELoss()

    for episode in range(episodes):
        state = env.reset()
        state = torch.FloatTensor(state).unsqueeze(0)
        done = False

        while not done:
            # 选择动作
            q_values = dqn(state)
            action = torch.argmax(q_values, dim=1).item()

            # 执行动作，获取下一个状态、奖励和是否结束
            next_state, reward, done = env.step(action)
            next_state = torch.FloatTensor(next_state).unsqueeze(0)
            reward = torch.tensor([reward], dtype=torch.float32).unsqueeze(0)
            done_tensor = torch.tensor([done], dtype=torch.float32).unsqueeze(0)

            # 存储经验
            memory.append((state, action, reward, next_state, done_tensor))

            if len(memory) > batch_size:
                # 从记忆中采样
                batch = np.random.choice(len(memory), batch_size, replace=False)
                state_batch, action_batch, reward_batch, next_state_batch, done_batch = zip(*[memory[i] for i in batch])

                state_batch = torch.cat(state_batch)
                action_batch = torch.tensor(action_batch, dtype=torch.long)
                reward_batch = torch.cat(reward_batch)
                next_state_batch = torch.cat(next_state_batch)
                done_batch = torch.cat(done_batch)

                # 计算目标Q值
                with torch.no_grad():
                    target_q_values = target_dqn(next_state_batch)
                    max_target_q = torch.max(target_q_values, dim=1)[0].unsqueeze(1)
                    target_q = reward_batch + (1 - done_batch) * gamma * max_target_q

                # 计算当前Q值
                current_q = dqn(state_batch).gather(1, action_batch.unsqueeze(1)).squeeze(1)

                # 计算损失并更新网络
                loss = loss_fn(current_q, target_q)
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()

            # 更新目标网络
            for target_param, param in zip(target_dqn.parameters(), dqn.parameters()):
                target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)

            state = next_state

        print(f"Episode {episode + 1} completed")

    return dqn


# 主函数
def main():
    num_buses = 10  # 示例配电网的节点数
    time_steps = 100  # 示例时间尺度
    state_size = num_buses
    action_size = 5  # 示例动作数量
    batch_size = 32
    gamma = 0.99
    tau = 0.001
    episodes = 100

    env = DistributionNetworkEnv(num_buses, time_steps)
    dqn = DQN(state_size, action_size)
    target_dqn = DQN(state_size, action_size)
    memory = []

    trained_dqn = train_dqn(env, dqn, target_dqn, memory, batch_size, gamma, tau, episodes)

    # 这里可以添加对训练好的模型进行评估和可视化的代码


if __name__ == "__main__":
    main()