Разработка кастомной Gym-среды для торговой стратегии

Разработка кастомной Gym-среды для торговой стратегии

Готовые торговые Gym-среды покрывают базовые сценарии. Для специфических стратегий — пары trading, опционные стратегии, высокочастотная торговля — нужна кастомная среда с точным моделированием исполнения ордеров, ликвидности, рыночного влияния.

Структура кастомной Gym-среды

import gymnasium as gym
from gymnasium import spaces
import numpy as np

class CustomTradingEnv(gym.Env):
    metadata = {'render_modes': ['human', 'rgb_array']}

    def __init__(self, df, config):
        super().__init__()
        self.df = df
        self.config = config

        # observation space: OHLCV + indicators + portfolio state
        n_features = config['n_features']
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf,
            shape=(n_features,), dtype=np.float32
        )

        # action space: position size [-1, 1] per asset
        n_assets = config['n_assets']
        self.action_space = spaces.Box(
            low=-1.0, high=1.0,
            shape=(n_assets,), dtype=np.float32
        )

        self._reset_portfolio()

    def reset(self, seed=None, options=None):
        super().reset(seed=seed)
        self._reset_portfolio()
        self.current_step = self.config['window_size']
        obs = self._get_observation()
        return obs, {}

    def step(self, action):
        # 1. Исполнение действия (с реалистичным моделированием)
        executed_action = self._execute_order(action)

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

        # 3. Обновление портфеля по новым ценам
        self._update_portfolio()

        # 4. Расчёт reward
        reward = self._compute_reward()

        # 5. Наблюдение
        obs = self._get_observation()

        # 6. Условие завершения
        terminated = self.current_step >= len(self.df) - 1
        truncated = self.portfolio_value < self.config['min_capital']

        info = {
            'portfolio_value': self.portfolio_value,
            'positions': self.positions.copy(),
            'total_trades': self.total_trades
        }

        return obs, reward, terminated, truncated, info

Реалистичное моделирование исполнения

Market impact: Крупные ордера двигают рынок — нельзя игнорировать при бэктестинге реальных объёмов.

def _execute_order(self, target_weights):
    current_prices = self.df.iloc[self.current_step][['open', 'high', 'low', 'close']]

    # проскальзывание: зависит от размера ордера и spread
    order_size = np.abs(target_weights - self.current_weights)
    slippage = order_size * self.config['slippage_factor']
    execution_price = current_prices['open'] * (1 + slippage)

    # комиссия
    trade_value = np.abs(order_size) * execution_price * self.portfolio_value
    commission = trade_value * self.config['commission_rate']

    self.portfolio_value -= commission.sum()
    self.current_weights = target_weights.copy()
    self.total_trades += (order_size > 0.01).sum()

    return target_weights

Order book simulation (для HFT):

class LOBSimulator:
    """Level-2 order book simulation"""
    def __init__(self, spread_bps=5, depth_levels=10):
        self.spread_bps = spread_bps
        self.depth_levels = depth_levels

    def get_fill_price(self, mid_price, order_size_usd):
        # fill price зависит от глубины стакана
        spread = mid_price * self.spread_bps / 10000
        market_impact = np.sqrt(order_size_usd / 1e6) * spread
        return mid_price + spread/2 + market_impact

Reward Engineering

Выбор reward — самая важная часть кастомной среды:

def _compute_reward(self):
    daily_return = (self.portfolio_value / self.prev_portfolio_value) - 1

    # вариант 1: простой return
    reward = daily_return

    # вариант 2: Sharpe-adjusted (rolling window)
    self.returns_history.append(daily_return)
    if len(self.returns_history) >= 20:
        sharpe = np.mean(self.returns_history[-20:]) / (np.std(self.returns_history[-20:]) + 1e-8)
        reward = daily_return * (1 + sharpe)

    # вариант 3: penalized drawdown
    current_dd = (self.peak_value - self.portfolio_value) / self.peak_value
    reward = daily_return - self.config['dd_penalty'] * current_dd

    # штраф за чрезмерную торговлю
    reward -= self.config['turnover_penalty'] * self.daily_turnover

    return float(reward)

Observation Engineering

def _get_observation(self):
    window = self.df.iloc[self.current_step - self.window_size:self.current_step]

    features = []
    # ценовые returns (нормализованы)
    returns = window['close'].pct_change().fillna(0).values
    features.extend(returns[-self.window_size:])

    # технические индикаторы
    features.extend([
        window['rsi'].iloc[-1] / 100,
        window['macd_norm'].iloc[-1],
        window['bb_position'].iloc[-1]  # (price - lower) / (upper - lower)
    ])

    # portfolio state
    features.extend(self.current_weights)
    features.append(self.portfolio_value / self.initial_capital - 1)

    return np.array(features, dtype=np.float32)

Тестирование среды

Проверка корректности через gymnasium check_env:

from gymnasium.utils.env_checker import check_env

env = CustomTradingEnv(df_train, config)
check_env(env)  # проверяет типы, shapes, корректность step/reset

Sanity checks:

• Random policy должна терять деньги (транзакционные издержки)
• Buy & Hold reproducible through env
• Нет look-ahead в observation (нет future data)

Сроки: 2–4 недели

Базовая кастомная среда для single-asset — 3–5 дней. Multi-asset с реалистичным execution, order book, риск-ограничениями — 3–4 недели.