Введение: Почему распределенный бэктестинг критичен для FinRL?
В мире количественных финансов и алгоритмической торговли, Reinforcement Learning (RL) агенты показывают многообещающие результаты. Однако, разработка и тестирование таких агентов в реалистичных финансовых средах, таких как FinRL, требует огромных вычислительных ресурсов. Бэктестинг одной стратегии может занимать часы, а оптимизация гиперпараметров или тестирование на множестве активов — дни или даже недели. Это замедляет итерационный процесс разработки и ограничивает возможности исследования.
Традиционные подходы к параллелизации, такие как использование модуля multiprocessing в Python, могут помочь, но они часто ограничены одним узлом и не предоставляют гибкости для работы с распределенными кластерами. Здесь на помощь приходит Ray — мощный фреймворк для построения и запуска распределенных приложений Python. Ray позволяет легко масштабировать вычисления от одной машины до крупного кластера, делая его идеальным инструментом для ускорения бэктестинга RL-агентов в FinRL.
Что такое Ray и как он помогает FinRL?
Ray — это универсальный фреймворк с открытым исходным кодом, который предоставляет простой API для параллельных и распределенных вычислений. Он абстрагирует сложности распределенных систем, позволяя разработчикам писать код, который масштабируется без значительных изменений. Ключевые концепции Ray включают:
Tasks: Функции, которые могут быть выполнены удаленно и асинхронно.Actors: Объекты, которые могут быть созданы удаленно и иметь состояние, позволяя инкапсулировать логику и данные.Objects: Неизменяемые данные, которые могут быть переданы между задачами и акторами без копирования.
Для FinRL, Ray предоставляет следующие преимущества:
- Масштабируемость: Легкое распределение множества бэктестов по ядрам CPU, GPU или узлам кластера.
- Ускорение: Сокращение времени, необходимого для тестирования и оптимизации стратегий.
- Гибкость: Возможность запускать различные конфигурации агентов или сред параллельно.
- Устойчивость к сбоям: Ray может обрабатывать сбои узлов, что важно для длительных распределенных вычислений.
Подготовка среды
Прежде чем приступить к коду, убедитесь, что у вас установлены необходимые библиотеки. Если вы еще не работали с FinRL, рекомендуем ознакомиться с ее документацией.
pip install ray[rllib] finrl pandas numpy stable-baselines3 gym
Обратите внимание, что ray[rllib] устанавливает компоненты Ray для Reinforcement Learning, которые могут быть полезны, хотя в данном примере мы будем использовать stable-baselines3 для агентов.
Исходный код
В этом примере мы покажем, как создать простую функцию бэктестинга для FinRL, а затем распределить ее выполнение с помощью Ray. Мы будем симулировать процесс обучения и оценки RL-агента для разных наборов гиперпараметров или различных временных интервалов. Для простоты, мы используем упрощенную модель среды и агента, чтобы сосредоточиться на интеграции с Ray.
import ray
import pandas as pd
import numpy as np
import time
import os
# Импортируем необходимые компоненты FinRL (упрощенно для примера)
# В реальном проекте здесь будут полноценные импорты из finrl.meta.env_stock_trading, finrl.agents.stablebaselines3.models и т.д.
from stable_baselines3 import PPO
from stable_baselines3.common.env_util import make_vec_env
from gym.spaces import Box, Discrete
import gym
# --- Упрощенная среда FinRL для демонстрации ---
class MockFinRLEnv(gym.Env):
metadata = {'render_modes': ['human'], 'render_fps': 30}
def __init__(self, initial_amount=100000, stock_dim=10, turbulence_threshold=100,
hmax=100, iteration=0, render_mode=None):
super().__init__()
self.initial_amount = initial_amount
self.stock_dim = stock_dim
self.turbulence_threshold = turbulence_threshold
self.hmax = hmax
self.iteration = iteration
self.render_mode = render_mode
# Action space: Buy/Sell/Hold for each stock
self.action_space = Box(low=-1, high=1, shape=(self.stock_dim,), dtype=np.float32)
# State space: Balance, stock prices, stock holdings, turbulence
# Simplified for demonstration
self.state_space = Box(low=0, high=np.inf, shape=(1 + self.stock_dim * 2 + 1,), dtype=np.float32)
self.current_step = 0
self.max_steps = 100 # Simulate a fixed number of steps for an episode
self.balance = initial_amount
self.stock_holdings = np.zeros(self.stock_dim)
self.data = self._generate_mock_data()
self.terminal = False
def _generate_mock_data(self):
# Simulate some financial data
dates = pd.date_range(start='2020-01-01', periods=self.max_steps + 1, freq='D')
data = pd.DataFrame({
'date': dates,
'open': np.random.rand(len(dates), self.stock_dim) * 100 + 50,
'high': np.random.rand(len(dates), self.stock_dim) * 100 + 60,
'low': np.random.rand(len(dates), self.stock_dim) * 100 + 40,
'close': np.random.rand(len(dates), self.stock_dim) * 100 + 50,
'volume': np.random.randint(100000, 1000000, size=(len(dates), self.stock_dim))
})
data['turbulence'] = np.random.rand(len(dates)) * 200 # Simulate turbulence
return data
def _get_state(self):
current_prices = self.data.iloc[self.current_step][['close']].values.flatten()
turbulence = self.data.iloc[self.current_step]['turbulence']
return np.concatenate([[self.balance], current_prices, self.stock_holdings, [turbulence]])
def step(self, actions):
self.current_step += 1
if self.current_step >= self.max_steps:
self.terminal = True
# Simulate trading logic (simplified)
current_prices = self.data.iloc[self.current_step][['close']].values.flatten()
for i in range(self.stock_dim):
# Buy/Sell based on action
if actions[i] > 0: # Buy
num_shares = int(self.hmax * actions[i])
cost = num_shares * current_prices[i]
if self.balance >= cost:
self.balance -= cost
self.stock_holdings[i] += num_shares
elif actions[i] < 0: # Sell
num_shares = int(self.hmax * abs(actions[i]))
if self.stock_holdings[i] >= num_shares:
self.balance += num_shares * current_prices[i]
self.stock_holdings[i] -= num_shares
# Calculate reward (simplified: change in portfolio value)
portfolio_value = self.balance + np.sum(self.stock_holdings * current_prices)
reward = (portfolio_value - self.initial_amount) / self.initial_amount # Percentage change
# Simulate done and info
done = self.terminal
info = {'portfolio_value': portfolio_value, 'balance': self.balance, 'stock_holdings': self.stock_holdings}
return self._get_state(), reward, done, info
def reset(self, seed=None, options=None):
super().reset(seed=seed)
self.current_step = 0
self.balance = self.initial_amount
self.stock_holdings = np.zeros(self.stock_dim)
self.terminal = False
return self._get_state(), {}
def render(self):
if self.render_mode == 'human':
print(f"Step: {self.current_step}, Balance: {self.balance:.2f}, Holdings: {self.stock_holdings}")
def close(self):
pass
# --- Функция для выполнения одного бэктеста с FinRL и Ray ---
@ray.remote
def run_finrl_backtest(config_id, initial_amount, stock_dim, turbulence_threshold,
total_timesteps, learning_rate, gamma, seed):
start_time = time.time()
print(f"[Config {config_id}] Starting backtest with initial_amount={initial_amount}, lr={learning_rate}, gamma={gamma}, seed={seed}")
# 1. Создание среды FinRL (используем нашу MockFinRLEnv)
env = MockFinRLEnv(
initial_amount=initial_amount,
stock_dim=stock_dim,
turbulence_threshold=turbulence_threshold,
iteration=config_id # Просто для демонстрации
)
# Оборачиваем среду для Stable Baselines3
vec_env = make_vec_env(lambda: env, n_envs=1, seed=seed)
# 2. Инициализация и обучение агента (PPO из Stable Baselines3)
try:
model = PPO(
"MlpPolicy",
vec_env,
verbose=0,
learning_rate=learning_rate,
gamma=gamma,
seed=seed
)
model.learn(total_timesteps=total_timesteps)
except Exception as e:
print(f"[Config {config_id}] Error during agent training: {e}")
return {
'config_id': config_id,
'status': 'failed_training',
'duration': time.time() - start_time,
'final_portfolio_value': initial_amount,
'error': str(e)
}
# 3. Оценка агента (простой прогон по среде)
obs, _ = vec_env.reset()
portfolio_values = [initial_amount]
current_balance = initial_amount
current_holdings = np.zeros(stock_dim)
for _ in range(env.max_steps):
actions, _states = model.predict(obs, deterministic=True)
obs, rewards, dones, infos = vec_env.step(actions)
# Извлекаем информацию из infos (для make_vec_env, infos - это список словарей)
if isinstance(infos, list) and len(infos) > 0:
current_balance = infos[0].get('balance', current_balance)
current_holdings = infos[0].get('stock_holdings', current_holdings)
portfolio_values.append(infos[0].get('portfolio_value', portfolio_values[-1]))
else:
portfolio_values.append(portfolio_values[-1]) # Если инфо нет, сохраняем последнее значение
if dones[0]: # Если среда завершилась
break
final_portfolio_value = portfolio_values[-1]
duration = time.time() - start_time
print(f"[Config {config_id}] Finished in {duration:.2f}s. Final portfolio value: {final_portfolio_value:.2f}")
return {
'config_id': config_id,
'status': 'success',
'initial_amount': initial_amount,
'stock_dim': stock_dim,
'turbulence_threshold': turbulence_threshold,
'total_timesteps': total_timesteps,
'learning_rate': learning_rate,
'gamma': gamma,
'seed': seed,
'duration': duration,
'final_portfolio_value': final_portfolio_value
}
# --- Основной блок выполнения ---
if __name__ == '__main__':
# Инициализация Ray. Можно указать num_cpus для ограничения использования ядер
# или address='auto' для подключения к существующему кластеру.
ray.init(num_cpus=os.cpu_count() // 2, ignore_reinit_error=True) # Используем половину ядер для примера
print(f"Ray initialized with {ray.cluster_resources().get('CPU', 0)} CPUs.")
# Параметры для бэктестинга
common_params = {
'initial_amount': 100000,
'stock_dim': 5,
'turbulence_threshold': 100,
'total_timesteps': 5000 # Уменьшено для быстрого примера
}
# Различные конфигурации для параллельного запуска
configs = [
{'learning_rate': 1e-4, 'gamma': 0.99, 'seed': 42},
{'learning_rate': 5e-5, 'gamma': 0.98, 'seed': 101},
{'learning_rate': 1e-3, 'gamma': 0.995, 'seed': 202},
{'learning_rate': 2e-4, 'gamma': 0.99, 'seed': 303},
{'learning_rate': 1e-4, 'gamma': 0.98, 'seed': 404},
{'learning_rate': 5e-4, 'gamma': 0.99, 'seed': 505},
]
# Запускаем бэктесты параллельно с Ray
results_refs = []
for i, config in enumerate(configs):
full_config = {**common_params, **config, 'config_id': i}
results_refs.append(run_finrl_backtest.remote(**full_config))
# Собираем результаты
print("\nWaiting for all backtests to complete...")
results = ray.get(results_refs)
print("\n--- Backtest Results ---")
for res in results:
if res['status'] == 'success':
print(f"Config {res['config_id']}: Final Value = {res['final_portfolio_value']:.2f}, Duration = {res['duration']:.2f}s, LR = {res['learning_rate']}, Gamma = {res['gamma']}")
else:
print(f"Config {res['config_id']}: Failed ({res['error']}), Duration = {res['duration']:.2f}s")
# Дополнительный анализ результатов (например, сортировка по доходности)
successful_results = [r for r in results if r['status'] == 'success']
if successful_results:
df_results = pd.DataFrame(successful_results)
df_results = df_results.sort_values(by='final_portfolio_value', ascending=False)
print("\n--- Top 3 Configurations by Final Portfolio Value ---")
print(df_results[['config_id', 'final_portfolio_value', 'learning_rate', 'gamma', 'duration']].head(3).to_string(index=False))
# Завершение работы Ray
ray.shutdown()
print("\nRay shutdown.")
Разбор параметров
ray.init(num_cpus=..., ignore_reinit_error=True): Инициализирует Ray.num_cpusпозволяет указать количество ядер CPU, которые Ray может использовать.ignore_reinit_error=Trueполезен при работе в интерактивных средах (например, Jupyter), чтобы избежать ошибок при повторной инициализации. Для кластера можно использоватьaddress='auto'.@ray.remote: Декоратор, который превращает обычную функцию (run_finrl_backtest) в удаленную задачу Ray. Это означает, что Ray может выполнять эту функцию на любом доступном рабочем процессе в кластере.run_finrl_backtest.remote(...): Вызов удаленной функции. Вместо немедленного выполнения, эта функция возвращаетObjectRef— ссылку на будущий результат. Вызовы.remote()являются асинхронными.ray.get(results_refs): Блокирующая операция, которая ожидает завершения всех задач, на которые ссылаютсяObjectRefв спискеresults_refs, и возвращает их фактические результаты.MockFinRLEnv: Упрощенная среда Gym, имитирующая ключевые аспекты среды FinRL. В реальном проекте вы бы использовали полноценные среды изfinrl.meta.env_stock_trading, такие какStockTradingEnv.stable_baselines3.PPO: Алгоритм Reinforcement Learning, используемый для обучения агента.learning_rateиgamma— это примеры гиперпараметров, которые часто оптимизируются.make_vec_env: Утилита из Stable Baselines3 для создания векторизованных сред, что часто требуется для обучения агентов.total_timesteps: Количество шагов обучения, которое агент выполнит в среде.seed: Зерно для генератора случайных чисел, используемое для воспроизводимости результатов. Крайне важно для сравнительного анализа различных конфигураций.
Как запустить
Для запуска примера выполните следующие шаги:
-
Установите зависимости: Убедитесь, что у вас установлены все необходимые библиотеки, как указано в разделе «Подготовка среды».
pip install ray[rllib] finrl pandas numpy stable-baselines3 gym -
Сохраните код: Скопируйте предоставленный исходный код в файл с именем, например,
ray_finrl_backtest.py. -
Запустите скрипт: Откройте терминал или командную строку, перейдите в директорию, где вы сохранили файл, и выполните команду:
python ray_finrl_backtest.py -
Наблюдайте за выполнением: Вы увидите, как Ray инициализируется и запускает несколько бэктестов параллельно. Результаты будут выведены в консоль после завершения всех задач.
Вы можете экспериментировать с количеством конфигураций в списке configs и параметром num_cpus в ray.init(), чтобы увидеть, как это влияет на скорость выполнения. Для более глубокого анализа данных, полученных в результате бэктестинга, можно применить методы, аналогичные тем, что используются в анализе скользящей корреляции или кластеризации данных, чтобы выявить наиболее эффективные стратегии или паттерны поведения агентов.
Заключение
Интеграция Ray с FinRL открывает новые горизонты для количественных аналитиков и исследователей в области Reinforcement Learning. Возможность эффективно распределять вычислительную нагрузку позволяет значительно ускорить процесс итерации, тестирования и оптимизации торговых стратегий. Это не только экономит время, но и позволяет исследовать более широкий спектр гиперпараметров и архитектур агентов, что в конечном итоге приводит к созданию более надежных и прибыльных систем.
