Обучение с подкреплением на Python: от теории к созданию умных алгоритмов

import numpy as np
import gymnasium as gym

# Создаем среду
env = gym.make('CartPole-v1')

# Инициализируем Q-таблицу нулями
# Дискретизируем непрерывное пространство состояний
def discretize_state(state):
# Упрощенная дискретизация для примера
return tuple(np.round(state, 1))

Q = {} # Q-таблица как словарь

# Параметры обучения
alpha = 0.1 # Скорость обучения
gamma = 0.99 # Коэффициент дисконтирования
epsilon = 0.1 # Вероятность случайного действия

# Цикл обучения
for episode in range(1000):
state, _ = env.reset()
state = discretize_state(state)
done = False

while not done:
# Выбор действия (epsilon-жадная политика)
if state not in Q:
Q[state] = np.zeros(env.action_space.n)

if np.random.random() < epsilon:
action = env.action_space.sample() # Исследование
else:
action = np.argmax(Q[state]) # Использование

# Выполнение действия
next_state, reward, done, _, _ = env.step(action)
next_state = discretize_state(next_state)

# Обновление Q-значения (формула Q-learning)
if next_state not in Q:
Q[next_state] = np.zeros(env.action_space.n)

Q[state][action] += alpha * (reward + gamma * np.max(Q[next_state]) – Q[state][action])

state = next_state

import numpy as np
import gymnasium as gym

env = gym.make('CliffWalking-v0')
Q = np.zeros([env.observation_space.n, env.action_space.n])

# Гиперпараметры
alpha = 0.1
gamma = 0.99
epsilon = 0.1
episodes = 2000

# Обучение с использованием SARSA
for episode in range(episodes):
state, _ = env.reset()

# Выбираем действие из состояния (epsilon-greedy)
if np.random.random() < epsilon:
action = env.action_space.sample()
else:
action = np.argmax(Q[state, :])

done = False
while not done:
# Выполняем действие
next_state, reward, done, _, _ = env.step(action)

# Выбираем следующее действие (epsilon-greedy)
if np.random.random() < epsilon:
next_action = env.action_space.sample()
else:
next_action = np.argmax(Q[next_state, :])

# Обновляем Q-значение по формуле SARSA
Q[state, action] += alpha * (reward + gamma * Q[next_state, next_action] – Q[state, action])

# Переходим к новому состоянию и действию
state = next_state
action = next_action

# Постепенное уменьшение epsilon для меньшего исследования в конце
epsilon = max(0.01, epsilon * 0.995)

# Тестирование обученной политики
state, _ = env.reset()
total_reward = 0
done = False

while not done:
action = np.argmax(Q[state, :])
state, reward, done, _, _ = env.step(action)
total_reward += reward

print(f"Итоговая награда: {total_reward}")

import numpy as np
import gymnasium as gym
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd

# Создаем среду FrozenLake
env = gym.make('FrozenLake-v1', is_slippery=False)

# Инициализируем Q-таблицу нулями
Q = np.zeros([env.observation_space.n, env.action_space.n])

# Параметры обучения
alpha = 0.8 # Скорость обучения
gamma = 0.95 # Коэффициент дисконтирования
epsilon = 1.0 # Начальная вероятность исследования
min_epsilon = 0.01 # Минимальная вероятность исследования
epsilon_decay = 0.001 # Скорость уменьшения epsilon

# Метрики для отслеживания прогресса обучения
rewards_per_episode = []
steps_per_episode = []
success_rate = []
success_window = 100 # Оценка успешности на последних 100 эпизодах

# Обучение агента
num_episodes = 10000
for episode in range(num_episodes):
# Инициализация среды
state, _ = env.reset()
done = False
total_reward = 0
step_count = 0

while not done:
step_count += 1

# Выбор действия (epsilon-жадная политика)
if np.random.random() < epsilon:
action = env.action_space.sample() # Исследование
else:
action = np.argmax(Q[state, :]) # Использование

# Выполнение действия
next_state, reward, done, _, _ = env.step(action)
total_reward += reward

# Обновление Q-значения
old_value = Q[state, action]
next_max = np.max(Q[next_state, :])

# Формула Q-learning
Q[state, action] = old_value + alpha * (reward + gamma * next_max – old_value)

state = next_state

# Сохранение метрик
rewards_per_episode.append(total_reward)
steps_per_episode.append(step_count)

# Расчет скользящей средней успешных эпизодов (награда > 0)
if len(rewards_per_episode) > success_window:
success_rate.append(sum(1 for r in rewards_per_episode[-success_window:] if r > 0) / success_window)
else:
success_rate.append(sum(1 for r in rewards_per_episode if r > 0) / len(rewards_per_episode))

# Уменьшение epsilon для сокращения исследования
epsilon = max(min_epsilon, epsilon – epsilon_decay)

# Визуализация Q-таблицы
def visualize_q_table(q_table, env_size=4):
"""Визуализирует Q-таблицу как тепловую карту для каждого действия"""
actions = ['Left', 'Down', 'Right', 'Up']
fig, axes = plt.subplots(1, len(actions), figsize=(20, 5))

for i, action_name in enumerate(actions):
q_values = q_table[:, i].reshape(env_size, env_size)
sns.heatmap(q_values, annot=True, cmap='viridis', ax=axes[i], cbar=True)
axes[i].set_title(f'Action: {action_name}')

plt.tight_layout()
plt.show()

visualize_q_table(Q)

# Тестирование политики
def test_policy(env, q_table, n_episodes=100):
"""Тестирует обученную политику на нескольких эпизодах"""
successes = 0
total_steps = 0

for _ in range(n_episodes):
state, _ = env.reset()
done = False
steps = 0

while not done and steps < 100: # Ограничение на количество шагов
steps += 1
action = np.argmax(q_table[state, :])
state, reward, done, _, _ = env.step(action)
total_steps += 1

if reward > 0:
successes += 1
break

success_rate = successes / n_episodes
avg_steps = total_steps / n_episodes

return success_rate, avg_steps

success_rate, avg_steps = test_policy(env, Q)
print(f"Успешность: {success_rate * 100:.2f}%, Среднее количество шагов: {avg_steps:.2f}")

import numpy as np
import gymnasium as gym
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
from tensorflow.keras.optimizers import Adam
import random
from collections import deque

# Класс агента DQN
class DQNAgent:
def __init__(self, state_size, action_size):
self.state_size = state_size
self.action_size = action_size
self.memory = deque(maxlen=2000) # Буфер опыта
self.gamma = 0.95 # Коэффициент дисконтирования
self.epsilon = 1.0 # Начальное значение epsilon для исследования
self.epsilon_min = 0.01 # Минимальное значение epsilon
self.epsilon_decay = 0.995 # Скорость уменьшения epsilon
self.learning_rate = 0.001 # Скорость обучения
self.model = self._build_model() # Основная Q-сеть
self.target_model = self._build_model() # Целевая Q-сеть
self.update_target_model() # Копирование весов в целевую сеть

def _build_model(self):
"""Создает нейронную сеть для аппроксимации Q-функции"""
model = Sequential()
model.add(Dense(24, input_dim=self.state_size, activation='relu'))
model.add(Dense(24, activation='relu'))
model.add(Dense(self.action_size, activation='linear'))
model.compile(loss='mse', optimizer=Adam(learning_rate=self.learning_rate))
return model

def update_target_model(self):
"""Копирует веса из основной модели в целевую"""
self.target_model.set_weights(self.model.get_weights())

def remember(self, state, action, reward, next_state, done):
"""Добавляет переход в память опыта"""
self.memory.append((state, action, reward, next_state, done))

def act(self, state, training=True):
"""Выбирает действие с использованием epsilon-greedy политики"""
if training and np.random.rand() <= self.epsilon:
return random.randrange(self.action_size)

act_values = self.model.predict(state, verbose=0)
return np.argmax(act_values[0])

def replay(self, batch_size):
"""Обучение на мини-батче из буфера опыта"""
if len(self.memory) < batch_size:
return

minibatch = random.sample(self.memory, batch_size)

states = np.array([t[0][0] for t in minibatch])
actions = np.array([t[1] for t in minibatch])
rewards = np.array([t[2] for t in minibatch])
next_states = np.array([t[3][0] for t in minibatch])
dones = np.array([t[4] for t in minibatch])

# Предсказания для текущих состояний и следующих состояний
targets = self.model.predict(states, verbose=0)
next_targets = self.target_model.predict(next_states, verbose=0)

# Обновление целевых Q-значений по формуле DQN
for i in range(batch_size):
if dones[i]:
targets[i, actions[i]] = rewards[i]
else:
targets[i, actions[i]] = rewards[i] + self.gamma * np.max(next_targets[i])

# Обучение модели
self.model.fit(states, targets, epochs=1, verbose=0)

# Уменьшение epsilon
if self.epsilon > self.epsilon_min:
self.epsilon *= self.epsilon_decay

# Функция для обучения и тестирования агента
def train_dqn(env_name='CartPole-v1', episodes=1000):
env = gym.make(env_name)
state_size = env.observation_space.shape[0]
action_size = env.action_space.n

agent = DQNAgent(state_size, action_size)
batch_size = 32
target_update_freq = 10 # Частота обновления целевой сети

scores = [] # Награды за эпизоды

for e in range(episodes):
state, _ = env.reset()
state = np.reshape(state, [1, state_size])
total_reward = 0

for time in range(500): # Ограничение на количество шагов
# Выбор и выполнение действия
action = agent.act(state)
next_state, reward, done, _, _ = env.step(action)
next_state = np.reshape(next_state, [1, state_size])

# Модификация награды для ускорения обучения
reward = reward if not done else -10

# Сохранение опыта в буфер
agent.remember(state, action, reward, next_state, done)

state = next_state
total_reward += reward

# Обучение на мини-батче
if len(agent.memory) > batch_size:
agent.replay(batch_size)

if done:
break

# Обновление целевой модели
if e % target_update_freq == 0:
agent.update_target_model()

scores.append(total_reward)

# Вывод прогресса обучения
print(f"Episode: {e+1}/{episodes}, Score: {total_reward}, Epsilon: {agent.epsilon:.2f}")

# Если средний результат за последние 100 эпизодов >= 195, считаем задачу решенной
if len(scores) > 100 and np.mean(scores[-100:]) >= 195:
print("CartPole решена! Средний результат за 100 эпизодов >= 195")
break

return agent, scores

# Запуск обучения
agent, scores = train_dqn()

# Тестирование обученного агента
def test_agent(agent, env_name='CartPole-v1', episodes=10):
env = gym.make(env_name, render_mode='human')
state_size = env.observation_space.shape[0]

for e in range(episodes):
state, _ = env.reset()
state = np.reshape(state, [1, state_size])
total_reward = 0
done = False

while not done:
action = agent.act(state, training=False)
next_state, reward, done, _, _ = env.step(action)
next_state = np.reshape(next_state, [1, state_size])
state = next_state
total_reward += reward

if done:
print(f"Тестовый эпизод {e+1}: Счет = {total_reward}")
break

# Тестирование
test_agent(agent)

import numpy as np
import pandas as pd
import gymnasium as gym
from gymnasium import spaces
import matplotlib.pyplot as plt
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv
from stable_baselines3.common.evaluation import evaluate_policy

# Создаем среду для торговли
class TradingEnv(gym.Env):
"""Среда для торговли на финансовом рынке с RL-агентом"""

metadata = {'render.modes': ['human']}

def __init__(self, df, initial_balance=10000, transaction_fee_percent=0.001):
super(TradingEnv, self).__init__()

# Данные цен
self.df = df
self.reward_range = (-np.inf, np.inf)
self.initial_balance = initial_balance
self.transaction_fee_percent = transaction_fee_percent

# Действия: 0=держать, 1=купить, 2=продать
self.action_space = spaces.Discrete(3)

# Наблюдаемое пространство: [баланс, активы, открытая позиция, OHLCV-данные]
self.observation_space = spaces.Box(low=0, high=1, shape=(10,), dtype=np.float32)

# Индекс текущего шага
self.current_step = 0
self.balance = initial_balance
self.shares_held = 0
self.net_worth = initial_balance
self.cost_basis = 0

# История действий для отображения
self.trades = []

def _next_observation(self):
"""Возвращает текущее наблюдение из среды"""
frame = np.array([
self.balance / self.initial_balance,
self.shares_held / 100, # Нормализация
self.cost_basis / self.df.iloc[self.current_step]['Close'],
self.df.iloc[self.current_step]['Open'] / self.df.iloc[self.current_step]['Close'],
self.df.iloc[self.current_step]['High'] / self.df.iloc[self.current_step]['Close'],
self.df.iloc[self.current_step]['Low'] / self.df.iloc[self.current_step]['Close'],
self.df.iloc[self.current_step]['Volume'] / 1000000, # Нормализация объема
self.df.iloc[self.current_step]['EMA_10'] / self.df.iloc[self.current_step]['Close'],
self.df.iloc[self.current_step]['EMA_30'] / self.df.iloc[self.current_step]['Close'],
self.df.iloc[self.current_step]['RSI'] / 100 # RSI нормализован от 0 до 1
])

return frame

def _take_action(self, action):
"""Выполняет действие в среде и возвращает вознаграждение"""
current_price = self.df.iloc[self.current_step]['Close']

if action == 1: # Покупка
# Используем весь доступный баланс для покупки акций
max_shares = self.balance / current_price
shares_to_buy = int(max_shares)
if shares_to_buy > 0:
# Стоимость с учетом комиссии
cost = shares_to_buy * current_price * (1 + self.transaction_fee_percent)

# Обновляем баланс и активы
self.balance -= cost
self.cost_basis = current_price
self.shares_held += shares_to_buy

# Сохраняем сделку
self.trades.append({
'step': self.current_step,
'price': current_price,
'type': 'buy',
'shares': shares_to_buy,
'cost': cost
})

elif action == 2: # Продажа
if self.shares_held > 0:
# Получаем выручку от продажи с учетом комиссии
proceeds = self.shares_held * current_price * (1 – self.transaction_fee_percent)

# Обновляем баланс и активы
self.balance += proceeds
self.shares_held = 0
self.cost_basis = 0

# Сохраняем сделку
self.trades.append({
'step': self.current_step,
'price': current_price,
'type': 'sell',
'shares': self.shares_held,
'proceeds': proceeds
})

# Рассчитываем чистую стоимость портфеля
self.net_worth = self.balance + self.shares_held * current_price

# Вознаграждение = изменение чистой стоимости
reward = self.net_worth – self.initial_balance
return reward

def step(self, action):
"""Выполняет шаг в среде"""
# Выполняем действие
reward = self._take_action(action)

# Переходим к следующему шагу
self.current_step += 1

# Проверяем, закончился ли эпизод
done = self.current_step >= len(self.df) – 1

# Получаем новое наблюдение
obs = self._next_observation()

return obs, reward, done, {}

def reset(self):
"""Сбрасывает среду в начальное состояние"""
self.balance = self.initial_balance
self.shares_held = 0
self.net_worth = self.initial_balance
self.cost_basis = 0
self.current_step = 0
self.trades = []

return self._next_observation()

def render(self, mode='human'):
"""Визуализирует состояние среды"""
if mode == 'human':
profit = self.net_worth – self.initial_balance
print(f'Шаг: {self.current_step}')
print(f'Баланс: {self.balance}')
print(f'Акции: {self.shares_held}')
print(f'Чистая стоимость: {self.net_worth}')
print(f'Прибыль: {profit}')
return profit

# Подготовка данных
def prepare_data():
"""Загружает и подготавливает данные для обучения"""
# Для примера используем генерированные данные
# В реальном проекте можно использовать API для получения рыночных данных

# Генерируем временной ряд с случайным трендом и волатильностью
np.random.seed(42)
dates = pd.date_range(start='2020-01-01', end='2021-01-01')
n_steps = len(dates)

# Генерируем случайные цены
price = 100
prices = [price]

for _ in range(1, n_steps):
change_percent = np.random.normal(0, 0.01)
price = price * (1 + change_percent)
prices.append(price)

# Создаем DataFrame с OHLCV-данными
df = pd.DataFrame({
'Date': dates,
'Open': prices,
'High': [p * (1 + np.random.uniform(0, 0.02)) for p in prices],
'Low': [p * (1 – np.random.uniform(0, 0.02)) for p in prices],
'Close': [p * (1 + np.random.normal(0, 0.005)) for p in prices],
'Volume': [np.random.randint(100000, 10000000) for _ in range(n_steps)]
})

# Рассчитываем технические индикаторы
df['EMA_10'] = df['Close'].ewm(span=10, adjust=False).mean()
df['EMA_30'] = df['Close'].ewm(span=30, adjust=False).mean()

# Рассчитываем RSI
delta = df['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['RSI'] = 100 – (100 / (1 + rs))

# Заполняем NaN
df = df.fillna(0)

# Разделяем на обучающий и тестовый наборы
train_size = int(len(df) * 0.8)
train_df = df[:train_size]
test_df = df[train_size:]

return train_df, test_df

# Подготовка данных
train_df, test_df = prepare_data()

# Создание векторизованной среды
def make_env(df):
def _init():
env = TradingEnv(df)
return env
return _init

# Создаем и обучаем модель
env = DummyVecEnv([make_env(train_df)])
model = PPO('MlpPolicy', env, verbose=1, learning_rate=0.0001)
model.learn(total_timesteps=50000)

# Оцениваем модель
mean_reward, _ = evaluate_policy(model, env, n_eval_episodes=10)
print(f'Средняя награда: {mean_reward}')

# Тестируем модель на тестовом наборе
test_env = DummyVecEnv([make_env(test_df)])
obs = test_env.reset()
done = False

# Отслеживаем чистую стоимость во время теста
net_worths = [10000] # Начальная стоимость

while not done:
action, _ = model.predict(obs)
obs, reward, done, info = test_env.step(action)

# Получаем текущую чистую стоимость
net_worth = test_env.envs[0].net_worth
net_worths.append(net_worth)

if done[0]:
break

# Визуализируем результаты
plt.figure(figsize=(12, 6))
plt.plot(net_worths)
plt.title('Изменение чистой стоимости портфеля')
plt.xlabel('Шаги')
plt.ylabel('Стоимость портфеля')
plt.grid(True)
plt.show()

# Визуализируем сделки на графике цены
plt.figure(figsize=(15, 7))
plt.plot(test_df['Close'].values, label='Цена закрытия')

# Отмечаем точки покупки и продажи
trades = test_env.envs[0].trades
for trade in trades:
step = trade['step']
price = trade['price']
if trade['type'] == 'buy':
plt.scatter(step, price, color='green', marker='^', s=100)
else: # sell
plt.scatter(step, price, color='red', marker='v', s=100)

plt.title('Торговые сигналы модели')
plt.xlabel('Шаги')
plt.ylabel('Цена')
plt.legend()
plt.grid(True)
plt.show()