10.1 强化学习交易

概念详解

强化学习（Reinforcement Learning, RL）在量化交易中代表了一种全新的策略范式。与传统的监督学习（预测未来收益率然后基于预测构建组合）不同，RL直接学习"在什么状态下采取什么行动来最大化累积回报"。这种端到端的学习方式天然适合交易问题的序列决策本质。

在量化交易中，RL的每个要素可以自然地映射到交易场景：

状态（State）：市场环境的表示，包括价格序列、技术指标、持仓信息、账户状态等
动作（Action）：交易决策，如买入/卖出/持有的离散动作，或持仓比例的连续动作
奖励（Reward）：交易结果的衡量，如PnL变化、夏普比、或风险调整后收益
策略（Policy）：从状态到动作的映射函数，即交易策略本身

RL在量化交易中的主要优势包括：

直接优化交易目标：不需要预测收益率的中间步骤，而是直接优化最终的PnL或夏普比
处理交易成本：RL可以将交易成本（佣金、滑点）自然地纳入奖励函数中，在学习过程中自动约束交易频率
自适应决策：RL策略可以根据市场状态（regime）动态调整行为
延迟奖励的处理：RL擅长处理短期行动影响长期回报的场景（如建仓时机影响后续退出机会）

主流的RL算法：

DQN（Deep Q-Network）：适用于离散动作空间（如买入/卖出/持有），使用经验回放和目标网络稳定训练
PPO（Proximal Policy Optimization）：适用于连续动作空间（如持仓比例），通过限制策略更新幅度保证训练稳定性，是当前量化交易中最常用的RL算法
SAC（Soft Actor-Critic）：最大熵RL算法，在探索和利用之间取得平衡，适合高度随机的金融环境
A2C/A3C：Actor-Critic架构的基础算法，同时学习策略（Actor）和价值函数（Critic）

名词解释：PPO

近端策略优化（Proximal Policy Optimization），一种强化学习算法，通过限制策略更新幅度来保证训练的稳定性，常用于连续动作空间。PPO是OpenAI的默认RL算法。

名词解释：马尔可夫决策过程（MDP）

RL的数学框架，由状态空间 $S$ 、动作空间 $A$ 、状态转移概率 $P$ 、奖励函数 $R$ 和折扣因子 $γ$ 组成。交易问题通常被建模为MDP或POMDP（部分可观测MDP）。

名词解释：经验回放（Experience Replay）

将Agent与环境的交互经验 $(s, a, r, s^{'})$ 存储在一个缓冲区中，训练时随机抽取批次数据。这打破了数据之间的时序相关性，提高了样本效率。

数学原理

马尔可夫决策过程

RL的数学基础是马尔可夫决策过程（MDP），定义为五元组 $(S, A, P, R, γ)$ ：

P (s_{t + 1} | s_{t}, a_{t}) = P [S_{t + 1} = s_{t + 1} | S_{t} = s_{t}, A_{t} = a_{t}]

R (s_{t}, a_{t}) = E [R_{t + 1} | S_{t} = s_{t}, A_{t} = a_{t}]

目标是最大化期望累积折扣奖励：

G_{t} = \sum_{k = 0}^{\infty} γ^{k} R_{t + k + 1}

其中 $γ \in [0, 1]$ 是折扣因子，表示未来奖励的重要性衰减。

价值函数与Q函数

状态价值函数 $V^{π} (s)$ 表示从状态 $s$ 出发，遵循策略 $π$ 的期望累积奖励：

V^{π} (s) = E_{π} [G_{t} | S_{t} = s]

动作价值函数（Q函数） $Q^{π} (s, a)$ 表示从状态 $s$ 出发，采取动作 $a$ ，之后遵循策略 $π$ 的期望累积奖励：

Q^{π} (s, a) = E_{π} [G_{t} | S_{t} = s, A_{t} = a]

最优Q函数满足贝尔曼最优方程：

Q^{*} (s, a) = E_{s^{'}} [R + γ max_{a^{'}} Q^{*} (s^{'}, a^{'})]

PPO的目标函数

PPO的核心创新是使用裁剪的替代目标函数：

L^{C L I P} (θ) = E_{t} [min (r_{t} (θ) {\hat{A}}_{t}, clip (r_{t} (θ), 1 - ϵ, 1 + ϵ) {\hat{A}}_{t})]

其中 $r_{t} (θ) = \frac{π_{θ} (a_{t} | s_{t})}{π_{θ_{o l d}} (a_{t} | s_{t})}$ 是新旧策略的概率比， ${\hat{A}}_{t}$ 是优势估计， $ϵ$ 是裁剪范围（通常为0.2）。

这个巧妙的裁剪机制确保：当优势为正时，不会过度增加该动作的概率；当优势为负时，不会过度降低该动作的概率。这正是PPO稳定性的来源。

优势函数估计

优势函数 $A (s, a) = Q (s, a) - V (s)$ 衡量某个动作比平均水平好多少。GAE（Generalized Advantage Estimation）是一种常用的优势估计方法：

{\hat{A}}_{t}^{G A E (γ, λ)} = \sum_{l = 0}^{\infty} (γ λ)^{l} δ_{t + l}

其中 $δ_{t} = r_{t} + γ V (s_{t + 1}) - V (s_{t})$ 是TD误差。

Python实战

📌 案例1：自定义交易环境

PYTHON33 行 · 1.1 KB

📄此处有展示代码33 行 · 1.1 KB展开 ▼

python

import gymnasium as gym

class TradingEnv(gym.Env):
    """单资产交易环境 — 自定义 gym.Env 子类的最小骨架"""
    def __init__(self):
        super().__init__()
        # 动作空间: 0=空仓, 1=半仓, 2=全仓
        self.action_space = gym.spaces.Discrete(3)
        # 观察空间: [position, price_return, volatility]
        self.observation_space = gym.spaces.Box(
            low=-np.inf, high=np.inf, shape=(3,), dtype=np.float32
        )

    def reset(self, *, seed=None, options=None):
        super().reset(seed=seed)
        self.position = 0.0
        self.cash = 100000.0
        return np.zeros(3, dtype=np.float32), {}

    def step(self, action):
        reward = 0.0
        terminated = False
        truncated = False
        return np.zeros(3, dtype=np.float32), reward, terminated, truncated, {}

# 演示 API：创建环境、查看空间
import numpy as np
env = TradingEnv()
obs, _ = env.reset()
print("自定义交易环境已创建")
print(f"  动作空间: {env.action_space}  (3 选 1)")
print(f"  观察空间: {env.observation_space.shape}  (3 维)")
print(f"  初始观察: {obs}")

▶ 点击展开可浏览运行结果

自定义交易环境已创建
  动作空间: Discrete(3)  (3 选 1)
  观察空间: (3,)  (3 维)
  初始观察: [0. 0. 0.]

📌 案例2：完整的RL交易环境与PPO训练

PYTHON236 行 · 9.9 KB

📄此处有展示代码236 行 · 9.9 KB展开 ▼

python

import numpy as np
import gymnasium as gym
from gymnasium import spaces
from collections import deque
from typing import Tuple, Optional

class TradingEnvironment(gym.Env):
    """
    单资产RL交易环境
    状态: [持仓, 价格变化率(N窗口), 波动率, 技术指标...]
    动作: 0=空仓, 1=持有(离散) 或 [-1, 1]连续仓位
    奖励: 每步的PnL变化 - 交易成本
    """
    def __init__(self, prices, window_size=20, 
                 transaction_cost_pct=0.001,
                 reward_scaling=100.0):
        super().__init__()
        self.prices = np.array(prices, dtype=np.float32)
        self.window_size = window_size
        self.transaction_cost = transaction_cost_pct
        self.reward_scaling = reward_scaling
        # 动作空间：离散(0=空仓, 1=半仓, 2=全仓)
        self.action_space = spaces.Discrete(3)
        # 观察空间：持仓 + 价格特征
        self.observation_space = spaces.Box(
            low=-np.inf, high=np.inf,
            shape=(1 + window_size + 2,),  # 持仓 + 价格窗口 + 波动率 + 趋势
            dtype=np.float32
        )
        self.reset()
    def _get_observation(self) -> np.ndarray:
        """构建当前状态表示"""
        # 价格窗口特征
        price_window = self.prices[self.current_step - self.window_size + 1:
                                   self.current_step + 1]
        returns = np.diff(price_window) / price_window[:-1]
        # 填充到fixed length
        if len(returns) < self.window_size:
            returns = np.pad(returns, (self.window_size - len(returns), 0), 
                           mode='constant')
        # 技术特征
        recent_prices = self.prices[max(0, self.current_step - self.window_size):
                                     self.current_step + 1]
        volatility = np.std(np.diff(recent_prices) / recent_prices[:-1]) if len(recent_prices) > 1 else 0
        # 趋势强度（当前价格相对于移动平均）
        if len(recent_prices) >= 5:
            sma = np.mean(recent_prices[-5:])
            trend = (recent_prices[-1] - sma) / sma
        else:
            trend = 0.0
        # 组合状态向量
        obs = np.array([
            self.position,        # 当前持仓 (0, 0.5, 1.0)
            *returns[-self.window_size:],  # 价格收益率序列
            volatility,           # 波动率
            trend,                # 趋势
        ], dtype=np.float32)
        return obs
    def reset(self, *, seed=None, options=None) -> np.ndarray:
        """重置环境到初始状态"""
        super().reset(seed=seed)
        self.current_step = self.window_size
        self.position = 0.0
        self.entry_price = 0.0
        self.total_pnl = 0.0
        self.trades = []
        return self._get_observation(), {}
    def step(self, action: int) -> Tuple[np.ndarray, float, bool, bool, dict]:
        """执行一步交易

        会计口径：
          · reward（奖励塑形）= 本步价格变动 × 持仓 - 本步交易成本
          · total_pnl（真实盈亏）= 建仓→平仓的持仓期收益 × 平均仓位 - 累计成本
        两者刻意分开：前者给 RL 密集反馈，后者给人类可读的真实业绩。
        """
        # 将离散动作映射为持仓
        target_position = np.array([0.0, 0.5, 1.0])[action]
        current_price = self.prices[self.current_step]
        prev_price = self.prices[self.current_step - 1] if self.current_step > 0 else current_price
        step_ret = (current_price - prev_price) / prev_price

        # 本步的奖励 = 价格变动带来的浮盈（按当前持仓计价）
        reward = step_ret * self.position
        cost = 0.0

        if self.position > 0 and target_position == 0:
            # 卖出（全部平仓）
            pnl_pct = (current_price - self.entry_price) / self.entry_price
            realized = pnl_pct * self.position - self.transaction_cost
            self.total_pnl += realized
            cost = self.transaction_cost
            self.trades.append(('SELL', current_price, realized))
            self.position = 0.0
            self.entry_price = 0.0
        elif self.position == 0 and target_position > 0:
            # 买入（建仓）
            self.position = target_position
            self.entry_price = current_price
            self.total_pnl -= self.transaction_cost
            cost = self.transaction_cost
            self.trades.append(('BUY', current_price, -self.transaction_cost))
        elif self.position > 0 and abs(target_position - self.position) > 1e-9:
            # 加减仓：先结算旧仓位，再按新仓位重新建仓
            pnl_pct = (current_price - self.entry_price) / self.entry_price
            realized = pnl_pct * self.position
            self.total_pnl += realized
            self.trades.append(('REBALANCE', current_price, realized))
            self.position = target_position
            self.entry_price = current_price
        # 其余情况：持仓不变，纯持有，不做任何结算

        reward = (reward - cost) * self.reward_scaling

        # 前进时间
        self.current_step += 1
        terminated = self.current_step >= len(self.prices) - 1
        truncated = False

        # 终止时强制平仓（注意：current_price 是推进前的价格，此处要用推进后的价格）
        if terminated and self.position > 0:
            exit_price = self.prices[min(self.current_step, len(self.prices) - 1)]
            final_pnl = (exit_price - self.entry_price) / self.entry_price * self.position
            self.total_pnl += final_pnl - self.transaction_cost
            reward += (final_pnl - self.transaction_cost) * self.reward_scaling
            self.trades.append(('SELL', exit_price, final_pnl - self.transaction_cost))
            self.position = 0.0
            self.entry_price = 0.0

        obs = self._get_observation() if not terminated else np.zeros_like(self._get_observation())
        info = {
            'total_pnl': self.total_pnl,
            'position': self.position,
            'current_price': current_price,
            'n_trades': len(self.trades)
        }
        return obs, reward, terminated, truncated, info

# ===== 使用PPO训练 =====

def train_rl_trader(prices, n_episodes=100):
    """
    使用Stable-Baselines3 PPO训练交易Agent
    需要安装: pip install stable-baselines3
    """
    try:
        from stable_baselines3 import PPO
        from stable_baselines3.common.vec_env import DummyVecEnv
        from stable_baselines3.common.callbacks import EvalCallback
    except ImportError:
        print("需要安装 stable-baselines3: pip install stable-baselines3")
        return None
    # 创建环境
    env = TradingEnvironment(prices, window_size=20)
    env = DummyVecEnv([lambda: env])
    # 创建PPO模型
    model = PPO(
        'MlpPolicy',
        env,
        learning_rate=3e-4,
        n_steps=2048,
        batch_size=64,
        n_epochs=10,
        gamma=0.99,
        gae_lambda=0.95,
        clip_range=0.2,
        ent_coef=0.01,  # 熵正则化系数（鼓励探索）
        verbose=1,
    )
    # 训练
    model.learn(total_timesteps=n_episodes * 1000)
    return model, env

# ===== 演示：环境自检 + 随机策略基线 =====
np.random.seed(42)
n_days = 500
# 模拟带趋势和波动聚类的价格序列
returns = np.random.randn(n_days) * 0.01
returns[100:200] += 0.003   # 上行趋势
returns[300:400] -= 0.002   # 下行趋势
prices = 100 * np.exp(np.cumsum(returns))

env = TradingEnvironment(prices, window_size=20)
print("=== RL 交易环境就绪 ===")
print(f"  价格序列长度    : {len(prices)}")
print(f"  价格范围        : [{prices.min():.1f}, {prices.max():.1f}]")
print(f"  动作空间        : {env.action_space}  (0=空仓, 1=半仓, 2=全仓)")
print(f"  观测空间维度    : {env.observation_space.shape[0]} "
      f"(持仓1 + 收益窗口{env.window_size} + 波动率1 + 趋势1)")

obs, _ = env.reset(seed=0)
print(f"\n  reset 后观测形状  : {obs.shape}, dtype={obs.dtype}")
step_ret = env.step(2)
print(f"  step(2) 返回元素 : {len(step_ret)} 个 -> (obs, reward, terminated, truncated, info)")
print(f"  info 字段        : {list(step_ret[4].keys())}")


def run_episode(policy, seed=0):
    """跑完整一轮 episode，返回 (总PnL, 交易次数, 步数)"""
    o, _ = env.reset(seed=seed)
    done = False
    steps = 0
    while not done:
        a = policy(o)
        o, r, term, trunc, info = env.step(a)
        done = term or trunc
        steps += 1
    return info['total_pnl'], info['n_trades'], steps


# 基线 1：随机策略
rng = np.random.default_rng(7)
random_pnl, random_trades, random_steps = run_episode(lambda o: int(rng.integers(0, 3)))

# 基线 2：始终全仓（近似买入持有）
always_pnl, always_trades, always_steps = run_episode(lambda o: 2)

# 基准：买入持有（无成本）
bh = prices[-1] / prices[env.window_size] - 1

print("\n=== 基线对比（单轮 episode，含 0.1% 单边成本）===")
print(f"  {'策略':<12}{'总收益':>12}{'交易次数':>10}{'步数':>8}")
print(f"  {'随机策略':<12}{random_pnl * 100:>11.2f}%{random_trades:>10}{random_steps:>8}")
print(f"  {'始终全仓':<12}{always_pnl * 100:>11.2f}%{always_trades:>10}{always_steps:>8}")
print(f"  {'买入持有(无成本)':<12}{bh * 100:>11.2f}%{1:>10}{len(prices) - env.window_size:>8}")

print("\n=== 观察 ===")
print("  · 随机策略频繁换手，交易成本侵蚀收益，通常显著跑输买入持有")
print("  · 始终全仓≈买入持有，差额即累计交易成本")
print("  · PPO 要超越上述基线才有训练价值——这正是 RL 交易的现实门槛")

try:
    import stable_baselines3  # noqa: F401
    print("\n  stable-baselines3 已安装，可调用 train_rl_trader(prices) 训练 PPO")
except ImportError:
    print("\n  提示: train_rl_trader(prices) 需先 pip install stable-baselines3")
    print("        （本页预计算结果由上述不依赖 SB3 的环境自检 + 基线对比生成）")

▶ 点击展开可浏览运行结果

=== RL 交易环境就绪 ===
  价格序列长度    : 500
  价格范围        : [88.5, 138.6]
  动作空间        : Discrete(3)  (0=空仓, 1=半仓, 2=全仓)
  观测空间维度    : 23 (持仓1 + 收益窗口20 + 波动率1 + 趋势1)

  reset 后观测形状  : (23,), dtype=float32
  step(2) 返回元素 : 5 个 -> (obs, reward, terminated, truncated, info)
  info 字段        : ['total_pnl', 'position', 'current_price', 'n_trades']

=== 基线对比（单轮 episode，含 0.1% 单边成本）===
  策略                   总收益      交易次数      步数
  随机策略              -7.35%       306     479
  始终全仓              16.42%         2     479
  买入持有(无成本)         16.62%         1     480

=== 观察 ===
  · 随机策略频繁换手，交易成本侵蚀收益，通常显著跑输买入持有
  · 始终全仓≈买入持有，差额即累计交易成本
  · PPO 要超越上述基线才有训练价值——这正是 RL 交易的现实门槛

  提示: train_rl_trader(prices) 需先 pip install stable-baselines3
        （本页预计算结果由上述不依赖 SB3 的环境自检 + 基线对比生成）

📌 案例3：RL策略评估与基准对比

PYTHON238 行 · 2.4 KB

📄此处有展示代码238 行 · 2.4 KB展开 ▼

python

import numpy as np

# ===== 前置数据(从案例2 复用,让案例3 可独立运行)=====
np.random.seed(42)
n_days = 500
_returns = np.random.randn(n_days) * 0.01
_returns[100:200] += 0.003
_returns[300:400] -= 0.002
prices = 100 * np.exp(np.cumsum(_returns))

def evaluate_rl_policy(model, env, n_episodes=10):



    """评估RL策略的表现"""



    episode_rewards = []



    episode_trades = []



    



    for ep in range(n_episodes):



        obs = env.reset()



        done = False



        total_reward = 0



        n_trades = 0



        



        while not done:



            action, _ = model.predict(obs, deterministic=True)



            obs, reward, done, info = env.step(action)



            total_reward += reward



            if info.get('position', 0) != 0:



                n_trades = info.get('n_trades', 0)



        



        episode_rewards.append(total_reward)



        episode_trades.append(n_trades)



    



    return {



        'avg_reward': np.mean(episode_rewards),



        'std_reward': np.std(episode_rewards),



        'avg_trades': np.mean(episode_trades),



        'sharpe': np.mean(episode_rewards) / np.std(episode_rewards) if np.std(episode_rewards) > 0 else 0



    }







def benchmark_comparison(prices, rl_model):



    """将RL策略与简单基准策略对比"""



    



    # 基准1：买入持有



    buy_hold_return = (prices[-1] - prices[0]) / prices[0]



    



    # 基准2：简单移动平均交叉



    sma_short = np.convolve(prices, np.ones(5)/5, mode='valid')



    sma_long = np.convolve(prices, np.ones(20)/20, mode='valid')



    



    # 对齐长度



    min_len = min(len(sma_short), len(sma_long))



    sma_short = sma_short[-min_len:]



    sma_long = sma_long[-min_len:]



    



    signals = np.where(sma_short > sma_long, 1, 0)



    sma_returns = np.diff(signals) * np.diff(np.log(prices[-min_len:]))



    sma_return = np.sum(sma_returns[~np.isnan(sma_returns)])



    



    print("策略对比:")



    print("-" * 40)



    print(f"  买入持有收益:    {buy_hold_return:.2%}")



    print(f"  均线交叉收益:    {sma_return:.2%}")



    if rl_model:



        # RL收益需要从训练结果中获取



        print(f"  RL策略收益:      待训练后评估")







# 执行基准对比（使用模拟价格）



benchmark_comparison(prices, None)

▶ 点击展开可浏览运行结果

策略对比:
----------------------------------------
  买入持有收益:    13.79%
  均线交叉收益:    13.54%

常见误区

误区1：RL是解决交易问题的"魔法"。 RL在交易中的应用面临诸多实际困难：金融环境是非平稳的（市场regime会变化）、样本效率低（需要大量训练数据）、奖励设计困难（稀疏奖励或不适当的奖励函数会导致学习失败）。RL不是"开箱即用"的方案。

误区2：在真实价格序列上训练RL一定能学到好的交易策略。 如果在整个历史价格序列上训练，RL会"记住"所有的高点和低点（过拟合），而不是学到泛化的交易规则。正确的做法是使用滚动窗口训练或留出真正的样本外测试期。

误区3：奖励函数就是PnL变化。 纯PnL驱动的奖励函数会导致RL学习到极高风险的策略（如在波动极大时全仓押注）。好的奖励函数通常需要包含风险惩罚项、交易频率惩罚等。

误区4：RL不需要特征工程。 RL的"端到端"学习能力虽然强大，但在金融数据（高噪声、低信噪比）中，提供良好的特征表示（技术指标、市场状态）仍然至关重要。直接喂入原始价格通常学不到有用的策略。

误区5：标准的RL算法参数适用于金融数据。 金融数据的非平稳性要求修改标准RL的超参数：更高的熵正则化（鼓励探索）、更激进的裁剪参数（防止在噪声中学到过拟合模式）、更多样的训练环境（多资产/多时段）。

实战练习

练习1： 设计并实现一个多资产RL交易环境，支持同时交易3-5只股票：

状态包括：每只股票的持仓、收益率、波动率、以及股票间的相关性
动作是离散的：对每只股票选择买入/卖出/持有
奖励函数包含组合层面的风险调整（如夏普比）
评估多资产RL是否比单资产RL有更好的风险调整后收益

练习2： 比较不同奖励函数对RL策略行为的影响：

奖励A：纯PnL变化
奖励B：PnL变化 - 0.01 * |动作| （惩罚频繁交易）
奖励C：夏普比代理（每次决策时的短期夏普比）
奖励D：Sortino比代理（只惩罚下行波动）
分析四种奖励函数下训练出的策略在风险、收益、换手率上的差异

练习3： 实现"Regime-Aware RL"——将市场Regime聚类（参考8.2章的方法）与RL结合：

使用K-Means将历史市场状态分为3个Regime
为每个Regime训练独立的RL策略
实时判断当前Regime并切换到对应的RL模型
对比Regime自适应RL与单一RL策略的表现

本章要点

强化学习（RL）直接学习从状态到动作的最优映射，天然适合交易的序列决策问题
PPO是量化交易中最常用的RL算法，其裁剪机制保证了训练的稳定性
RL交易环境的设计（状态空间、动作空间、奖励函数）是决定策略成败的关键
奖励函数需要考虑风险惩罚和交易成本，纯PnL驱动会导致高风险策略
RL在金融中的最大挑战是过拟合和非平稳性
RL策略应该被视为互补工具而非传统策略的替代，目前最成熟的应用是做市和最优执行

10.1 强化学习交易 ​

概念详解 ​

数学原理 ​

马尔可夫决策过程 ​

价值函数与Q函数 ​

PPO的目标函数 ​

优势函数估计 ​

Python实战 ​

📌 案例1：自定义交易环境 ​

📌 案例2：完整的RL交易环境与PPO训练 ​

📌 案例3：RL策略评估与基准对比 ​

常见误区 ​

实战练习 ​

延伸阅读 ​

本章要点 ​

10.1 强化学习交易

概念详解

数学原理

马尔可夫决策过程

价值函数与Q函数

PPO的目标函数

优势函数估计

Python实战

📌 案例1：自定义交易环境

📌 案例2：完整的RL交易环境与PPO训练

📌 案例3：RL策略评估与基准对比

常见误区

实战练习

延伸阅读

本章要点