Пример применения RL с нуля в финтехе

Источник иллюстрации
Источник иллюстрации
Источник иллюстрации
Иллюстрация автора
class Environment:
'''
Рабочая среда робота, внутри которого будет
происходить дальнейшее обучение
'''
def __init__(self, length = 100, normalize = True, noise = True, data = []):
self.length = length

if len(data) == 0:
# Если данные не поданы, формируем их сами
# на основе синуса размером length
self.data = np.sin(np.arange(length) / 30.0)
else:
# Иначе подугружаем существующие
self.data = np.array(data).flatten()


if noise:
# Подаем шум для данных от 0.1 до 1
self.data += np.random.normal(0, 0.1, size = self.data.shape)

if normalize:
# Нормализация после данных после шума
self.data = (self.data - self.data.min()) / (self.data.max() - self.data.min())

def get_state(self, time, lookback, diff = True):
"""
Возвращаем производные отдельного окна в нашей выборке
и убираем нули в начале
"""
window = self.data[time-lookback:time]
if diff: window = np.diff(window, prepend = window.flatten()[0])
return window

def get_reward(self, action, action_time, reward_time, coef = 100):
"""
Основная логика получения награды
0 => long 1 => hold 2 => short
"""
if action == 0:
action = -1
# print(23, self.data)
# Вытаскиваем текущую цену
price_now = self.data[action_time]
# Вытаскиваем следующую цену
price_reward = self.data[reward_time]
# Получаем разницу в проценте
price_diff = (price_reward - price_now) / price_now
# Прибавляем к портфелю следующее число:
# Дельта изменения валюты * покупку/продажу/холд * коэф. закупки
reward = np.sign(price_diff) * action * coef
# print(12121, reward)
return reward
# Создадим тестовую среду
lin_env = Environment(normalize=True, noise=True)
# Отобразим все производные отдельного окна с 95 по 100 выборку
lin_env.get_state(100, 5, True)
Иллюстрация автора
Источник иллюстрации
# Обратите внимание на то, что в RL часто используются совсем простые нейронные сети
class Net(nn.Module):
"""Строим простую модель нейронки"""
def __init__(self, state_shape, action_shape):
super(Net, self).__init__()
self.fc1 = nn.Linear(state_shape, 10)
self.fc2 = nn.Linear(10, action_shape)

def forward(self, x):
x = F.relu(self.fc1(x))
x = self.fc2(x)
return x
------Net(
(fc1): Linear(in_features=10, out_features=10, bias=True)
(fc2): Linear(in_features=10, out_features=3, bias=True)
)
class BuyHoldSellAgent:
'''
Агент для покупки продажи
'''
def __init__(self, state_shape = 10, action_shape = 2, experience_size = 100):
self.state_shape = state_shape
self.action_shape = action_shape
self.experience_size = experience_size
self.experience = collections.deque(maxlen=self.experience_size)

# Создадим экземпляр модели
self.model = Net(state_shape, action_shape)

# Создадим функцию ошибки
self.criterion = nn.MSELoss()
# Добавим оптимизатор
self.optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)

def save_experience(self, current_state, action, reward, next_state):
"""Метод для сохранения предудыщих данных эксперимента"""
self.experience.append({
'state_i': current_state,
'action_i': action,
'reward_i': reward,
'state_i_1': next_state
})

def replay_experience(self, gamma, sample_size):
"""Метод для оптимизации данных тренировки"""
# Создаем фиксированную выборку из добавленных событий
indices_sampled = np.random.choice(
len(self.experience),
sample_size,
replace=False
)
# Проходимся только по тем элементам, которые были добавлены в выборку

current_states = []
actions = []
rewards = []
next_states = []
for i in indices_sampled:
state_i, action_i, reward_i, state_i_1 = self.experience[i]['state_i'], self.experience[i]['action_i'], self.experience[i]['reward_i'], self.experience[i]['state_i_1']
current_states.append(state_i)
actions.append(action_i)
rewards.append(reward_i)
next_states.append(state_i_1)

current_states = np.array(current_states).squeeze()
next_states = np.array(next_states).squeeze()

# Получаем прогноз по следующему состоянию
next_q_values = self.model(torch.from_numpy(next_states).float()).detach().numpy()

# Получаем прогноз по текущему состоянию
current_q_values = self.model(torch.from_numpy(current_states).float()).detach().numpy()
# Уравнение Бэллмена
# Суть в том, что мы берем максимально возможную награду
# из действия из будущего шага (q_value_i_1) , умножаем ее на гамму
# (коэф. значимости будущих наград), прибавляем к текущей награде
for i in range(len(indices_sampled)):
# и заносим в Q таблицу для обучения
current_q_values[i, actions[i]] = rewards[i] + gamma * next_q_values[i, :].max()

outputs = self.model(torch.from_numpy(np.expand_dims(current_states.reshape(-1, WINDOW_SHAPE), 0)).float())[0]
loss = self.criterion(outputs, torch.Tensor(current_q_values))
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()

def get_value_action_value(self, state):
"""Метод для прогноза сигнала"""
state = np.expand_dims(state, axis = 0)
pred = self.model(torch.from_numpy(state).float())
return pred.flatten()
# Количество эпох обучения
epochs = 20
# Коэф. значимости награды на шаг вперед
gamma = 0.9
# Количество эпох обучения
epsilon = 0.95
# Размер датасета
DATASET_LENGTH = 250
# Размер окна из которого будут браться предыдущие данные
WINDOW_SHAPE = 5
# Шаг предыдущих данных
REWARD_TIME = 1
# Число доступных действий
ACTIONS_SHAPE = 2
# Размер выборки
SAMPLE_SIZE = 30
# Объявляем новую среду с агентом, данные будут генерироваться автоматически
environment = Environment(DATASET_LENGTH, True, False)
agent = BuyHoldSellAgent(WINDOW_SHAPE, ACTIONS_SHAPE)
action_to_backtest_action = {
1: 1, # покупаем
0: -1, # продаем
}

Алгоритм, обученный на простой синусоиде по-прежнему справляется с задачей! Теперь есть некоторые запутанные моменты, но в среднем модель все еще узнает, где находятся долгосрочные тренды нашей зашумленной функции синуса.

!pip install yfinance
clear_output()
import yfinance as yf
df = yf.download(tickers='TSLA')
df = df[-500:]
df = df.reset_index(drop=True)
print(df.head())

--

--

С 2020 года занимаюсь изучением применения нейронных сетей в трейдинге. Мой блог о совершенных ошибках и полезном опыте.

Love podcasts or audiobooks? Learn on the go with our new app.

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store
Ilia Konushok

Ilia Konushok

С 2020 года занимаюсь изучением применения нейронных сетей в трейдинге. Мой блог о совершенных ошибках и полезном опыте.