[Python] Python强化学习实战：基于TensorTrade框架的量化交易策略构建教程

你是否对强化学习（Reinforcement Learning）感兴趣，却被复杂的概念和实现所困扰？TensorTrade 这一Python库的出现，让构建基于强化学习的交易策略变得触手可及。作为一个开源框架，TensorTrade 设计灵活，能够与 NumPy、Pandas、Gym、TensorFlow 及 Keras 等主流库无缝集成。本文将带你一步步从数据获取到模型训练，完成一个AI驱动的交易算法入门项目。

第一步：获取并处理历史价格数据

在启动强化学习训练前，我们需要准备历史交易数据作为训练环境的基础。这里使用 yfinance 库来获取数据，它无需复杂注册，简单易用。

import yfinance
import pandas_ta as ta

# 定义股票代码和日期范围
TICKER = ‘TTRD’  # 替换为你要交易的股票代码
TRAIN_START_DATE = ‘2021-02-09’  # 训练集开始日期
TRAIN_END_DATE = ‘2021-09-30’    # 训练集结束日期
EVAL_START_DATE = ‘2021-10-01’   # 评估集开始日期
EVAL_END_DATE = ‘2021-11-12’     # 评估集结束日期

# 创建 yfinance 对象
yf_ticker = yfinance.Ticker(ticker=TICKER)

# 获取训练数据集
df_training = yf_ticker.history(start=TRAIN_START_DATE, end=TRAIN_END_DATE)
df_training.drop([‘Dividends’, ‘Stock Splits’], axis=1, inplace=True)  # 删除不需要的列
df_training[“Volume”] = df_training[“Volume”].astype(int)  # 将成交量转换为整数

# 添加技术指标
df_training.ta.log_return(append=True, length=16)  # 对数收益率
df_training.ta.rsi(append=True, length=14)         # RSI 指标
df_training.ta.macd(append=True, fast=12, slow=26)  # MACD 指标
df_training.to_csv(‘training.csv’, index=False)    # 保存为 CSV 文件

# 获取评估数据集（代码结构与训练集相同）
df_evaluation = yf_ticker.history(start=EVAL_START_DATE, end=EVAL_END_DATE)
df_evaluation.drop([‘Dividends’, ‘Stock Splits’], axis=1, inplace=True)
df_evaluation[“Volume”] = df_evaluation[“Volume”].astype(int)
df_evaluation.ta.log_return(append=True, length=16)
df_evaluation.ta.rsi(append=True, length=14)
df_evaluation.ta.macd(append=True, fast=12, slow=26)
df_evaluation.to_csv(‘evaluation.csv’, index=False)

这段代码不仅完成了基础的数据清洗，还为数据添加了对数收益率、RSI和MACD等常用技术指标，确保后续的智能体（Agent）能够基于多维特征进行决策，而非盲目操作。

第二步：构建TensorTrade交易环境

TensorTrade 的工作方式借鉴了 OpenAI Gym。你需要提供价格数据，它会为智能体搭建一个用于学习和试错的模拟交易环境。

import pandas as pd
from tensortrade.feed.core import DataFeed, Stream
from tensortrade.oms.instruments import Instrument
from tensortrade.oms.exchanges import Exchange, ExchangeOptions
from tensortrade.oms.services.execution.simulated import execute_order
from tensortrade.oms.wallets import Wallet, Portfolio
import tensortrade.env.default as default

def create_env(config):
    “”“创建交易环境的函数”“”
    # 读取数据集
    dataset = pd.read_csv(config[“csv_filename”], parse_dates=[‘Datetime’])

    # 设置交易手续费
    commission = 0.0035

    # 创建价格数据流
    price = Stream.source(list(dataset[“Close”]), dtype=“float”).rename(“price”)

    # 配置交易所
    exchange_options = ExchangeOptions(commission=commission)
    ttse_exchange = Exchange(“TTSE”, service=execute_order, options=exchange_options)

    # 定义交易工具（货币和股票）
    USD = Instrument(“USD”, 2, “US Dollar”)     # 美元，精度为 2 位小数
    TTRD = Instrument(“TTRD”, 2, “TensorTrade Corp”)  # 股票

    # 创建钱包和投资组合
    cash = Wallet(ttse_exchange, 1000 * USD)  # 初始资金 1000 美元
    asset = Wallet(ttse_exchange, TTRD)        # 股票钱包
    portfolio = Portfolio(USD, [cash, asset])  # 投资组合

    # 创建渲染数据流（用于可视化）
    renderer_feed = DataFeed([
        Stream.source(list(dataset[“Datetime”])).rename(“date”),
        Stream.source(list(dataset[“Open”]), dtype=“float”).rename(“open”),
        Stream.source(list(dataset[“High”]), dtype=“float”).rename(“high”),
        Stream.source(list(dataset[“Low”]), dtype=“float”).rename(“low”),
        Stream.source(list(dataset[“Close”]), dtype=“float”).rename(“close”),
        Stream.source(list(dataset[“Volume”]), dtype=“float”).rename(“volume”),
    ])

    # 创建特征数据流
    features = []
    for c in dataset.columns[1:]:
        features.append(
            Stream.source(list(dataset[c]), dtype=“float”).rename(c)
        )
    feed = DataFeed(features)
    feed.compile()

    # 定义奖励方案和动作方案
    reward_scheme = default.rewards.SimpleProfit(window_size=config[“reward_window_size”])
    action_scheme = default.actions.BSH(cash=cash, asset=asset)  # 买入/卖出/持有

    # 创建并返回环境
    env = default.create(
        feed=feed,
        portfolio=portfolio,
        action_scheme=action_scheme,
        reward_scheme=reward_scheme,
        renderer_feed=renderer_feed,
        renderer=[],
        window_size=config[“window_size”],
        max_allowed_loss=config[“max_allowed_loss”]  # 最大允许亏损
    )
    return env

至此，一个微型的交易沙盒环境就搭建完成了。智能体可以在这个环境中执行买入、卖出或持有操作，并通过与环境交互来学习如何最大化收益。

第三步：集成Ray进行自动化训练与超参数调优

手动调整强化学习模型的超参数效率低下。利用 Ray Tune 框架，我们可以自动化这一搜索过程，大幅提升开发效率。

import ray
import os
from ray import tune
from ray.tune.registry import register_env

# 定义超参数搜索空间
FC_SIZE = tune.grid_search([[256, 256], [1024], [128, 64, 32]])  # 全连接层大小
LEARNING_RATE = tune.grid_search([0.001, 0.0005, 0.00001])       # 学习率
MINIBATCH_SIZE = tune.grid_search([5, 10, 20])                   # 小批量大小

cwd = os.getcwd()
ray.init()  # 初始化 Ray

# 注册自定义环境
register_env(“MyTrainingEnv”, create_env)

# 训练环境配置
env_config_training = {
    “window_size”: 14,          # 观察窗口大小
    “reward_window_size”: 7,    # 奖励计算窗口
    “max_allowed_loss”: 0.10,   # 最大允许亏损 10%
    “csv_filename”: os.path.join(cwd, ‘training.csv’),
}

# 评估环境配置
env_config_evaluation = {
    “max_allowed_loss”: 1.00,   # 评估时允许更大亏损
    “csv_filename”: os.path.join(cwd, ‘evaluation.csv’),
}

# 开始训练
analysis = tune.run(
    run_or_experiment=“PPO”,           # 使用 PPO 算法
    name=“MyExperiment1”,              # 实验名称
    metric=“episode_reward_mean”,      # 优化指标
    mode=“max”,                        # 最大化奖励
    stop={“training_iteration”: 5},    # 训练迭代次数
    config={
        “env”: “MyTrainingEnv”,
        “env_config”: env_config_training,
        “framework”: “torch”,          # 使用 PyTorch 框架
        “num_workers”: 1,              # 工作进程数
        “lr”: LEARNING_RATE,
        “model”: {“fcnet_hiddens”: FC_SIZE},
        “sgd_minibatch_size”: MINIBATCH_SIZE,
        “evaluation_interval”: 1,
        “evaluation_config”: {
            “env_config”: env_config_evaluation,
            “explore”: False          # 评估时不探索
        }
    },
    num_samples=1,
    checkpoint_freq=1                 # 检查点保存频率
)

Ray 会自动尝试我们定义的各种超参数组合，并行运行多个训练任务，并最终找出表现最佳的配置，从而避免了繁琐的手动调参过程。

第四步：设计自定义奖励函数

奖励函数是引导智能体学习方向的关键。TensorTrade 在此处的设计非常友好，使得创建贴合交易逻辑的自定义奖励函数变得简单直观。

from tensortrade.env.default.rewards import TensorTradeRewardScheme
from tensortrade.feed.core import DataFeed, Stream

class PBR(TensorTradeRewardScheme):
    “”“基于持仓的奖励方案（Position-Based Reward）”“”
    registered_name = “pbr”

    def __init__(self, price: Stream):
        super().__init__()
        self.position = -1  # 初始持仓状态

        # 计算价格变化
        r = Stream.sensor(price, lambda p: p.value, dtype=“float”).diff()
        # 获取当前持仓
        position = Stream.sensor(self, lambda rs: rs.position, dtype=“float”)
        # 奖励 = 价格变化 * 持仓方向
        reward = (r * position).fillna(0).rename(“reward”)

        self.feed = DataFeed([reward])
        self.feed.compile()

    def on_action(self, action: int):
        “”“根据动作更新持仓方向”“”
        self.position = 1 if action == 0 else -1  # 0 表示做多，其他表示做空

    def get_reward(self, portfolio):
        “”“获取当前奖励值”“”
        return self.feed.next()[“reward”]

    def reset(self):
        “”“重置奖励方案”“”
        self.position = -1
        self.feed.reset()

这个自定义的 PBR 奖励方案将奖励与持仓方向紧密关联：当做多（持仓为1）时，价格上涨产生正奖励；当做空（持仓为-1）时，价格下跌产生正奖励。这更符合实际交易中的盈亏逻辑。

总结与展望

TensorTrade 作为一个实践人工智能与量化交易结合的工具，并非没有挑战。如果奖励函数设计不合理，智能体的学习效果可能会不尽如人意。然而，它确是将强化学习应用于具体交易场景的理想入门框架。

其核心优势包括：

1. 开源与高灵活性：允许开发者深度定制交易环境、奖励机制和智能体。
2. 生态集成良好：与 Python 数据科学生态（Pandas, NumPy）和主流深度学习框架（TensorFlow, PyTorch）无缝衔接。
3. 抽象层次合理：简化了强化学习环境的构建过程，让开发者更专注于策略逻辑本身。
4. 支持分布式训练：可结合 Ray 等工具进行高效的超参数搜索与模型训练。

对于同时对强化学习和量化交易感兴趣的开发者而言，TensorTrade 提供了一个低门槛、高自由度的实践平台，是探索AI驱动交易策略的优质起点。