Неделя 12: Применение RL в реальных задачах и продвинутые методы
1. Теоретическая часть
1.1. Продвинутые методы RL
Методы на основе политик (Policy-Based)
A2C/A3C (Advantage Actor-Critic): Параллельное обучение нескольких агентов
PPO (Proximal Policy Optimization): Стабильное обучение с ограничениями на обновление политики
SAC (Soft Actor-Critic): Максимизация награды и энтропии
Гибридные методы
DDPG (Deep Deterministic Policy Gradient): Для непрерывных пространств действий
TD3 (Twin Delayed DDPG): Улучшенная версия DDPG
1.2. Применение RL в реальных задачах
Область
Пример применения
Финтех
Алгоритмический трейдинг
Робототехника
Обучение роботов ходьбе
Игры
AlphaStar (Starcraft II), OpenAI Five (Dota 2)
Рекомендательные системы
Персонализация контента
2.1. Реализация PPO на PyTorch
python
Copy
import torch
import torch.optim as optim
from torch.distributions import Categorical
class PPONetwork(nn.Module):
def __init__(self, input_dim, output_dim):
super(PPONetwork, self).__init__()
self.actor = nn.Sequential(
nn.Linear(input_dim, 64),
nn.Tanh(),
nn.Linear(64, 64),
nn.Tanh(),
nn.Linear(64, output_dim),
nn.Softmax(dim=-1)
)
self.critic = nn.Sequential(
nn.Linear(input_dim, 64),
nn.Tanh(),
nn.Linear(64, 64),
nn.Tanh(),
nn.Linear(64, 1)
)
def forward(self, x):
return self.actor(x), self.critic(x)
# Параметры обучения
gamma = 0.99
clip_epsilon = 0.2
ppo_epochs = 4
batch_size = 64
# Обучение
def train_ppo(env, model, optimizer, num_steps):
states = []
actions = []
rewards = []
old_log_probs = []
values = []
state = env.reset()
for _ in range(num_steps):
state = torch.FloatTensor(state)
probs, value = model(state)
dist = Categorical(probs)
action = dist.sample()
next_state, reward, done, _ = env.step(action.item())
states.append(state)
actions.append(action)
rewards.append(reward)
old_log_probs.append(dist.log_prob(action))
values.append(value)
state = next_state
if done:
state = env.reset()
# Расчет преимуществ
returns = []
advantages = []
R = 0
for r in reversed(rewards):
R = r + gamma * R
returns.insert(0, R)
returns = torch.tensor(returns)
values = torch.stack(values).squeeze()
advantages = returns - values
# Обновление политики
for _ in range(ppo_epochs):
for idx in range(0, num_steps, batch_size):
batch_states = torch.stack(states[idx:idx+batch_size])
batch_actions = torch.stack(actions[idx:idx+batch_size])
batch_old_log_probs = torch.stack(old_log_probs[idx:idx+batch_size])
batch_advantages = advantages[idx:idx+batch_size]
probs, _ = model(batch_states)
dist = Categorical(probs)
new_log_probs = dist.log_prob(batch_actions)
ratio = (new_log_probs - batch_old_log_probs).exp()
surr1 = ratio * batch_advantages
surr2 = torch.clamp(ratio, 1.0 - clip_epsilon, 1.0 + clip_epsilon) * batch_advantages
actor_loss = -torch.min(surr1, surr2).mean()
critic_loss = F.mse_loss(returns[idx:idx+batch_size], values[idx:idx+batch_size])
loss = actor_loss + 0.5 * critic_loss
optimizer.zero_grad()
loss.backward()
optimizer.step()
2.2. DDPG для непрерывных действий
python
Copy
class DDPG:
def __init__(self, state_dim, action_dim, max_action):
self.actor = Actor(state_dim, action_dim, max_action).to(device)
self.actor_target = Actor(state_dim, action_dim, max_action).to(device)
self.actor_target.load_state_dict(self.actor.state_dict())
self.critic = Critic(state_dim, action_dim).to(device)
self.critic_target = Critic(state_dim, action_dim).to(device)
self.critic_target.load_state_dict(self.critic.state_dict())
self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=1e-4)
self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=1e-3)
self.replay_buffer = ReplayBuffer(max_size=1e6)
def select_action(self, state, noise=None):
state = torch.FloatTensor(state.reshape(1, -1)).to(device)
action = self.actor(state).cpu().data.numpy().flatten()
if noise is not None:
action += noise.sample()
return np.clip(action, -self.max_action, self.max_action)
def train(self, batch_size=64, gamma=0.99, tau=0.005):
# Обновление критика
states, actions, rewards, next_states, dones = self.replay_buffer.sample(batch_size)
next_actions = self.actor_target(next_states)
target_Q = self.critic_target(next_states, next_actions)
target_Q = rewards + (1 - dones) * gamma * target_Q
current_Q = self.critic(states, actions)
critic_loss = F.mse_loss(current_Q, target_Q.detach())
self.critic_optimizer.zero_grad()
critic_loss.backward()
self.critic_optimizer.step()
# Обновление актера
actor_loss = -self.critic(states, self.actor(states)).mean()
self.actor_optimizer.zero_grad()
actor_loss.backward()
self.actor_optimizer.step()
# Мягкое обновление target сетей
for param, target_param in zip(self.actor.parameters(), self.actor_target.parameters()):
target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)
for param, target_param in zip(self.critic.parameters(), self.critic_target.parameters()):
target_param.data.copy_(tau * param.data + (1 - tau) * target_param.data)
2.3. RLlib для распределенного RL
python
Copy
from ray import tune
from ray.rllib.agents.ppo import PPOTrainer
# Конфигурация обучения
config = {
"env": "CartPole-v1",
"framework": "torch",
"num_workers": 4,
"num_gpus": 0,
"lr": 1e-3,
"gamma": 0.99,
"clip_param": 0.2,
}
# Запуск обучения
analysis = tune.run(
PPOTrainer,
config=config,
stop={"episode_reward_mean": 200},
checkpoint_at_end=True
)
3. Домашнее задание
Задача 1: Применение PPO к LunarLander
1. Обучите агента на среде LunarLander-v2 с помощью PPO.
2. Визуализируйте процесс обучения (график наград).
Решение:
python
Copy
env = gym.make('LunarLander-v2')
model = PPONetwork(env.observation_space.shape[0], env.action_space.n)
optimizer = optim.Adam(model.parameters(), lr=3e-4)
rewards = []
for episode in range(1000):
state = env.reset()
episode_reward = 0
while True:
state = torch.FloatTensor(state)
probs, _ = model(state)
dist = Categorical(probs)
action = dist.sample().item()
next_state, reward, done, _ = env.step(action)
episode_reward += reward
# ... (обучение как в 2.1)
if done:
rewards.append(episode_reward)
break
if episode % 50 == 0:
plt.plot(rewards)
plt.title('Training Progress')
plt.show()
Задача 2: Сравнение DDPG и TD3
1. Реализуйте TD3 (Twin Delayed DDPG) для Pendulum-v1.
2. Сравните результаты с DDPG по стабильности обучения.
Решение:
python
Copy
class TD3:
def __init__(self, state_dim, action_dim, max_action):
# Два критика вместо одного
self.critic1 = Critic(state_dim, action_dim).to(device)
self.critic2 = Critic(state_dim, action_dim).to(device)
# ... (аналогично DDPG, но с обновлением через минимум двух критиков)
def train(self):
# Обновление с задержкой и добавлением шума
target_actions = self.actor_target(next_states) + torch.clamp(
torch.randn_like(next_actions) * 0.2, -0.5, 0.5)
target_actions = torch.clamp(target_actions, -self.max_action, self.max_action)
target_Q1 = self.critic1_target(next_states, target_actions)
target_Q2 = self.critic2_target(next_states, target_actions)
target_Q = torch.min(target_Q1, target_Q2)
Задача 3: RL для алгоритмического трейдинга
1. Создайте среду для торговли акциями на основе исторических данных.
2. Обучите агента (PPO или DQN) максимизировать прибыль.
Решение:
python
Copy
class TradingEnv:
def __init__(self, data):
self.data = data # DataFrame с ценами
self.current_step = 0
self.position = 0 # 0: нет позиции, 1: лонг, -1: шорт
self.balance = 10000
def step(self, action):
# action: 0-удерживать, 1-купить, 2-продать
price = self.data.iloc[self.current_step]['close']
if action == 1 and self.position == 0:
self.position = 1
self.entry_price = price
elif action == 2 and self.position == 1:
self.balance += (price - self.entry_price) * 100
self.position = 0
self.current_step += 1
done = self.current_step == len(self.data) - 1
reward = self.balance - 10000 # Прибыль
return self._get_state(), reward, done, {}
def _get_state(self):
return np.array([
self.data.iloc[self.current_step]['open'],
self.data.iloc[self.current_step]['high'],
self.data.iloc[self.current_step]['low'],
self.data.iloc[self.current_step]['close'],
self.position
])
Итоги недели
✅ Изучили продвинутые методы RL (PPO, DDPG, TD3).
✅ Реализовали распределенное обучение с RLlib.
✅ Применили RL к реальным задачам (трейдинг).
✅ Сравнили разные алгоритмы на сложных средах.
Следующая тема: Развертывание ML-моделей в production. 🚀