TensorTrade：拥有自定义 Gym 环境的强化学习交易框架 — 2026 指南

python -m venv tensortrade-env
source tensortrade-env/bin/activate  # Linux/Mac
# tensortrade-env\Scripts\activate  # Windows

# Upgrade pip
pip install --upgrade pip

# Core framework
pip install tensortrade==1.2.0

# RL algorithms
pip install stable-baselines3==2.5.0

# Data fetching
pip install ccxt==4.4.0 yfinance==0.2.54

# Utilities
pip install pandas==2.2.3 numpy==1.26.4

import tensortrade
import gymnasium as gym
import stable_baselines3

print(f"TensorTrade version: {tensortrade.__version__}")
print(f"Gymnasium version: {gym.__version__}")
print(f"Stable Baselines3 version: {stable_baselines3.__version__}")

import pandas as pd
import yfinance as yf
from tensortrade.env.default import create
from tensortrade.feed.core import Stream, DataFeed
from tensortrade.oms.exchanges import Exchange
from tensortrade.oms.services.execution.simulated import execute_order
from tensortrade.oms.instruments import USD, BTC
from tensortrade.oms.wallets import Wallet, Portfolio

# Download BTC-USD data
df = yf.download("BTC-USD", start="2025-01-01", end="2026-03-01")
df.columns = [c[0] if isinstance(c, tuple) else c for c in df.columns]

# Create simulated exchange
exchange = Exchange("yfinance", service=execute_order)(
    Stream.source(list(df["Close"]), dtype="float").rename("USD-BTC")
)

# Set up portfolio
cash_wallet = Wallet(exchange, 10000 * USD)  # $10,000 starting
coin_wallet = Wallet(exchange, 0 * BTC)
portfolio = Portfolio(USD, [
    cash_wallet,
    coin_wallet
])

# Build data feed
feed = DataFeed([
    Stream.source(list(df["Open"]), dtype="float").rename("open"),
    Stream.source(list(df["High"]), dtype="float").rename("high"),
    Stream.source(list(df["Low"]), dtype="float").rename("low"),
    Stream.source(list(df["Close"]), dtype="float").rename("close"),
    Stream.source(list(df["Volume"]), dtype="float").rename("volume"),
])

# Create trading environment
env = create(
    portfolio=portfolio,
    action_scheme="managed-risk",  # actions: hold, buy, sell with sizing
    reward_scheme="risk-adjusted", # reward based on returns / volatility
    feed=feed,
    window_size=20,  # 20-period observation window
    max_allowed_loss=0.10  # stop if portfolio drops 10%
)

print(f"Observation space: {env.observation_space}")
print(f"Action space: {env.action_space}")

from stable_baselines3 import PPO
from stable_baselines3.common.callbacks import EvalCallback

# Initialize PPO agent
agent = PPO(
    policy="MlpPolicy",
    env=env,
    verbose=1,
    learning_rate=3e-4,
    n_steps=2048,
    batch_size=64,
    n_epochs=10,
    gamma=0.99,
    gae_lambda=0.95,
    clip_range=0.2,
    tensorboard_log="./tensorboard_logs/"
)

# Train for 100k timesteps
agent.learn(total_timesteps=100_000)

# Save the trained model
agent.save("ppo_btc_trader_v1")

import ta  # technical analysis library

# Compute RSI
rsi = ta.momentum.RSIIndicator(df["Close"], window=14).rsi().fillna(50)

# Compute MACD
macd = ta.trend.MACD(df["Close"])
macd_line = macd.macd().fillna(0)
macd_signal = macd.macd_signal().fillna(0)

# Add to feed
feed = DataFeed([
    Stream.source(list(df["Close"]), dtype="float").rename("close"),
    Stream.source(list(rsi), dtype="float").rename("rsi"),
    Stream.source(list(macd_line), dtype="float").rename("macd"),
    Stream.source(list(macd_signal), dtype="float").rename("macd_signal"),
    Stream.source(list(df["Volume"]), dtype="float").rename("volume"),
])

import ray
from ray import tune
from ray.rllib.algorithms.ppo import PPOConfig

ray.init()

config = (
    PPOConfig()
    .environment("TradingEnv", env_config={"portfolio": portfolio, "feed": feed})
    .framework("torch")
    .resources(num_gpus=1)
    .rollouts(num_rollout_workers=4)
)

tune.run(
    "PPO",
    config=config.to_dict(),
    stop={"timesteps_total": 500_000},
    checkpoint_at_end=True,
    storage_path="~/ray_results"
)

import ccxt

# Connect to Binance via CCXT
binance = ccxt.binance({
    "apiKey": "YOUR_API_KEY",
    "secret": "YOUR_SECRET",
    "enableRateLimit": True,
})

# Fetch recent OHLCV data
ohlcv = binance.fetch_ohlcv("BTC/USDT", timeframe="1h", limit=500)
ohlcv_df = pd.DataFrame(
    ohlcv,
    columns=["timestamp", "open", "high", "low", "close", "volume"]
)

# Use in TensorTrade environment
# Note: live trading requires additional risk management

import numpy as np

class SortinoRewardScheme:
    def __init__(self, risk_free_rate=0.02, window=30):
        self.risk_free_rate = risk_free_rate
        self.window = window
        self.returns = []

    def get_reward(self, portfolio: "Portfolio") -> float:
        profit_loss = portfolio.profit_loss
        self.returns.append(profit_loss)
        if len(self.returns) < self.window:
            return 0.0
        recent_returns = np.array(self.returns[-self.window:])
        excess = recent_returns - self.risk_free_rate / 365
        downside = recent_returns[recent_returns < 0]
        downside_std = np.std(downside) if len(downside) > 0 else 1e-6
        sortino = np.mean(excess) / downside_std
        return float(sortino)

# Use in environment
env = create(
    portfolio=portfolio,
    action_scheme="managed-risk",
    reward_scheme=SortinoRewardScheme(),
    feed=feed,
    window_size=20,
)

from tensortrade.oms.exchanges import Exchange
from tensortrade.oms.instruments import USD, BTC

# Simulated price divergences between two exchanges
binance_exchange = Exchange("binance", service=execute_order)(
    Stream.source(list(binance_prices), dtype="float").rename("USD-BTC")
)

coinbase_exchange = Exchange("coinbase", service=execute_order)(
    Stream.source(list(coinbase_prices), dtype="float").rename("USD-BTC")
)

# Portfolio spans both exchanges
binance_wallet = Wallet(binance_exchange, 5000 * USD)
coinbase_wallet = Wallet(coinbase_exchange, 5000 * USD)
btc_binance = Wallet(binance_exchange, 0 * BTC)
btc_coinbase = Wallet(coinbase_exchange, 0 * BTC)

multi_portfolio = Portfolio(USD, [
    binance_wallet, coinbase_wallet, btc_binance, btc_coinbase
])

class KellyCriterionActionScheme:
    """Sizes bets using fractional Kelly criterion."""
    def __init__(self, kelly_fraction=0.3):
        self.kelly_fraction = kelly_fraction
        self.win_rate = 0.5
        self.avg_win = 0.02
        self.avg_loss = 0.01

    def compute_size(self, action, portfolio):
        # Update statistics from trade history
        kelly = (self.win_rate / self.avg_loss -
                 (1 - self.win_rate) / self.avg_win) if self.avg_win > 0 else 0
        kelly = max(0, min(kelly, 0.5))  # Cap at 50%
        return kelly * self.kelly_fraction * portfolio.base_balance

# 1. Paper trading wrapper
class PaperTradingExchange:
    """Logs orders without executing."""
    def execute(self, order):
        print(f"[PAPER] {order.side} {order.quantity} @ {order.price}")
        return {"status": "filled", "price": order.price}

# 2. Circuit breaker
class CircuitBreaker:
    def __init__(self, max_drawdown=0.05, daily_loss_limit=0.03):
        self.max_drawdown = max_drawdown
        self.daily_loss_limit = daily_loss_limit
        self.daily_pnl = 0
        self.peak = 0

    def check(self, portfolio):
        if portfolio.net_worth > self.peak:
            self.peak = portfolio.net_worth
        drawdown = (self.peak - portfolio.net_worth) / self.peak
        if drawdown > self.max_drawdown:
            raise RuntimeError(f"Circuit breaker: drawdown {drawdown:.2%}")

# 3. Model versioning
import datetime
model_version = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
agent.save(f"models/ppo_prod_{model_version}.zip")