4、自定义强化学习环境：继承gym.Env、定义状态空间与动作空间、实现step与reset方法

好，咱们进入正题。上一章我们聊了强化学习的基本框架，这一章我带你亲手搭一个交易环境。说白了，就是让我们的策略能在一个模拟的“市场”里跑起来。

我个人习惯，做量化回测的第一步不是写策略，而是先定义环境。为什么？因为环境决定了你的智能体能学到什么、不能学到什么。你想想看，如果状态空间里漏掉了关键信息，比如成交量变化，那模型再怎么训练也白搭。

4.1 为什么需要自定义环境？

市面上有很多现成的强化学习库，比如OpenAI Gym自带了一些经典环境。但做交易回测，你得面对几个现实问题：

  • 数据是私有的：你的历史行情数据、因子数据，别人没有
  • 规则是自定义的：手续费怎么算？滑点模型用哪种？这些都得自己定
  • 状态是动态的：持仓、账户余额、市场状态，这些需要实时更新

所以，继承gym.Env自己写一个环境，几乎是必经之路。我在项目中遇到过好几次，直接用别人的环境改，结果改到最后发现还不如从头写——因为牵一发而动全身。

4.2 继承gym.Env：骨架搭起来

先看一个最简版本的环境骨架。记住，核心就两个方法：reset和step。

import gym
from gym import spaces
import numpy as np

class TradingEnv(gym.Env):
    """自定义交易环境"""
    
    def __init__(self, data, initial_balance=10000):
        super(TradingEnv, self).__init__()
        
        # 数据
        self.data = data
        self.initial_balance = initial_balance
        
        # 状态空间：连续空间
        # 假设状态包含：价格、持仓量、账户余额、已实现收益
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf, shape=(4,), dtype=np.float32
        )
        
        # 动作空间：离散动作
        # 0: 卖出, 1: 持有, 2: 买入
        self.action_space = spaces.Discrete(3)
        
        # 内部状态
        self.current_step = 0
        self.balance = initial_balance
        self.holdings = 0
        self.total_reward = 0
        
    def reset(self):
        """重置环境到初始状态"""
        self.current_step = 0
        self.balance = self.initial_balance
        self.holdings = 0
        self.total_reward = 0
        return self._get_obs()
    
    def step(self, action):
        """执行一步交易"""
        # 获取当前价格
        price = self.data[self.current_step]
        
        # 执行动作
        if action == 0:  # 卖出
            reward = self._sell(price)
        elif action == 2:  # 买入
            reward = self._buy(price)
        else:  # 持有
            reward = 0
        
        # 更新步数
        self.current_step += 1
        done = self.current_step >= len(self.data) - 1
        
        # 计算奖励
        self.total_reward += reward
        
        return self._get_obs(), reward, done, {}
    
    def _get_obs(self):
        """获取当前状态"""
        price = self.data[self.current_step]
        return np.array([
            price,
            self.holdings,
            self.balance,
            self.total_reward
        ], dtype=np.float32)
    
    def _sell(self, price):
        """卖出逻辑"""
        if self.holdings > 0:
            revenue = self.holdings * price
            self.balance += revenue
            self.holdings = 0
            return revenue * 0.001  # 简单奖励
        return 0
    
    def _buy(self, price):
        """买入逻辑"""
        if self.balance > price:
            shares = self.balance // price
            cost = shares * price
            self.balance -= cost
            self.holdings += shares
            return -cost * 0.001  # 简单惩罚
        return 0

嗯，这里要注意。上面的代码只是一个演示，实际项目中你还需要考虑滑点、手续费、交易限制等。我曾经犯过一个错误，就是忽略了滑点，结果回测曲线漂亮得不行，实盘一跑就崩。

4.3 定义状态空间与动作空间

状态空间和动作空间的定义，直接决定了你的模型能学到什么。我一般遵循一个原则：够用就好，别贪多。

状态空间

常见的交易状态包括：

状态维度 说明 示例值
当前价格 最新成交价 100.5
持仓量 当前持有的股票数量 100
账户余额 可用现金 50000
已实现收益 累计盈亏 1500
技术指标 如RSI、MACD等 0.65

我个人习惯，会把价格做归一化处理。比如用当前价格除以过去N天的平均价格，这样模型更容易收敛。你想想看，如果价格从100涨到200，数值范围变了，模型还得重新适应，多麻烦。

动作空间

动作空间有两种常见形式：

  • 离散动作：比如{卖出, 持有, 买入}，适合做方向判断
  • 连续动作：比如输出一个[-1, 1]之间的值，表示仓位比例

我建议初学者先从离散动作开始。为什么？因为连续动作的探索空间太大，训练起来很慢。等你把离散动作玩明白了，再升级到连续动作也不迟。

4.4 实现step与reset方法

这两个方法是环境的核心。咱们拆开来看。

reset方法

reset的作用很简单：把环境恢复到初始状态。但这里有个坑——每次reset应该从数据的不同位置开始。否则模型会记住数据顺序，导致过拟合。

def reset(self):
    # 随机选择一个起始点
    start_idx = np.random.randint(0, len(self.data) - self.window_size)
    self.current_step = start_idx
    self.balance = self.initial_balance
    self.holdings = 0
    self.total_reward = 0
    return self._get_obs()

我曾经犯过一个错误，就是固定从数据开头开始。结果模型学会了“前100天不要交易，第101天买入”这种毫无意义的模式。后来改成随机起始点，效果好了很多。

step方法

step方法执行一个动作，返回四个值：

  • obs：新的状态
  • reward：即时奖励
  • done：是否结束
  • info：额外信息（可选）

这里最关键的其实是奖励函数的设计。我见过很多人直接用“盈亏金额”作为奖励，结果模型学会了“死扛亏损单”——因为只要不平仓，亏损就不算数。这显然不对。

我的建议：奖励函数要同时考虑收益和风险。比如用夏普比率、最大回撤等指标。或者简单点，用“每笔交易的盈亏”加上“持仓时间的惩罚”。

4.5 核心逻辑流程图

下面这张图展示了自定义交易环境的完整流程。你可以看到，从初始化到每一步的交互，再到最终的结束判断，形成了一个闭环。

自定义交易环境核心流程 __init__() 初始化 reset() 重置环境 _get_obs() 获取状态 step(action) 执行动作 done? 是 结束回合 否 继续下一步 图：交易环境核心流程——初始化 → 重置 → 获取状态 → 执行动作 → 判断结束

4.6 避坑指南

我曾经踩过的坑：
  • 数据泄露：在计算状态时用了未来数据。比如用当天的收盘价作为状态，但当天还没收盘。解决办法是只用历史数据。
  • 奖励滞后：奖励只给在最后一步，导致中间步骤没有学习信号。建议每步都给一个小奖励，比如持仓盈亏的变化。
  • 动作空间过大：一开始就搞连续动作空间，模型根本学不会。建议从3个离散动作开始。

4.7 完整示例：一个可运行的交易环境

最后，给你一个可以直接运行的完整示例。这个环境包含了手续费、滑点、以及简单的奖励函数。

import gym
from gym import spaces
import numpy as np
import pandas as pd

class RealisticTradingEnv(gym.Env):
    """更真实的交易环境"""
    
    def __init__(self, df, initial_balance=100000, commission=0.001, slippage=0.0005):
        super(RealisticTradingEnv, self).__init__()
        
        self.df = df
        self.initial_balance = initial_balance
        self.commission = commission  # 手续费率
        self.slippage = slippage      # 滑点率
        
        # 状态：价格、持仓、余额、收益、技术指标
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf, shape=(8,), dtype=np.float32
        )
        
        # 动作：0=卖出, 1=持有, 2=买入
        self.action_space = spaces.Discrete(3)
        
        self.reset()
    
    def reset(self):
        self.current_step = np.random.randint(100, len(self.df) - 100)
        self.balance = self.initial_balance
        self.holdings = 0
        self.total_pnl = 0
        self.trades = 0
        
        return self._get_obs()
    
    def step(self, action):
        price = self.df.iloc[self.current_step]['close']
        
        # 考虑滑点
        if action == 0:  # 卖出
            exec_price = price * (1 - self.slippage)
            reward = self._sell(exec_price)
        elif action == 2:  # 买入
            exec_price = price * (1 + self.slippage)
            reward = self._buy(exec_price)
        else:
            reward = 0
        
        # 更新
        self.current_step += 1
        done = self.current_step >= len(self.df) - 1
        
        # 计算持仓市值变化作为额外奖励
        if self.holdings > 0:
            price_change = self.df.iloc[self.current_step]['close'] - price
            reward += self.holdings * price_change * 0.01
        
        self.total_pnl += reward
        
        return self._get_obs(), reward, done, {'pnl': self.total_pnl, 'trades': self.trades}
    
    def _get_obs(self):
        row = self.df.iloc[self.current_step]
        return np.array([
            row['close'] / 100,  # 归一化
            self.holdings / 1000,
            self.balance / self.initial_balance,
            self.total_pnl / self.initial_balance,
            row.get('rsi', 50) / 100,
            row.get('macd', 0) / 10,
            row.get('volume', 0) / 1000000,
            self.trades / 100
        ], dtype=np.float32)
    
    def _sell(self, price):
        if self.holdings > 0:
            revenue = self.holdings * price
            fee = revenue * self.commission
            self.balance += revenue - fee
            self.holdings = 0
            self.trades += 1
            return revenue - fee
        return 0
    
    def _buy(self, price):
        if self.balance > price:
            shares = int(self.balance // price)
            cost = shares * price
            fee = cost * self.commission
            self.balance -= cost + fee
            self.holdings += shares
            self.trades += 1
            return -(cost + fee)
        return 0
关键点总结：
  • 继承gym.Env，实现reset和step
  • 状态空间用spaces.Box，动作空间用spaces.Discrete
  • 奖励函数要兼顾收益和风险
  • 别忘了考虑手续费和滑点

好了，这一章的内容就到这里。你把这个环境跑起来，然后试着训练一个简单的DQN智能体，看看效果如何。下一章我们会聊如何把策略和这个环境对接起来。

专注资料整理

← 上一章 📖 返回目录 下一章 →