基于Q-Learning的股票市场自动化交易Agent：从原理到实战

1次阅读

没有评论

共计 3243 个字符，预计需要花费 9 分钟才能阅读完成。

传统股票交易策略通常依赖于技术指标、基本面分析或人工经验判断。这些方法存在几个明显缺陷：

市场环境变化快，人工策略难以及时调整
技术指标组合复杂，参数优化耗时费力
情绪因素影响大，难以保持交易纪律性

强化学习通过模拟 ” 环境 - 行为 - 反馈 ” 的循环，让 Agent 在交易过程中自主学习最优策略。Q-Learning 作为经典的 value-based 方法，特别适合中小型状态空间的交易决策问题。

在各类 RL 算法中，我们需要根据交易场景特点做出选择：

Q-Learning：
优点：原理简单，适合离散动作空间
局限：难以处理高维状态（如原始 K 线数据）
DQN：
优点：通过神经网络逼近 Q 函数，能处理复杂状态
局限：训练稳定性较差，需要经验回放
PPO：
优点：策略梯度方法，适合连续动作空间
局限：超参数敏感，训练周期长

对于股票交易这种中等复杂度的离散决策问题，基础 Q -Learning 往往能达到不错的起步效果。

有效的状态表示应该包含市场关键信息：

价格相关：当日涨跌幅、5 日均线偏离度
成交量：量比、换手率
技术指标：RSI(14)、MACD(12,26,9)
持仓状态：当前仓位比例

# 状态编码示例
def get_state(data, position):
    features = [data['close'].pct_change(),
        (data['close'] - data['close'].rolling(5).mean()) / data['close'].rolling(5).std(),
        data['volume'] / data['volume'].rolling(20).mean(),
        ta.rsi(data['close'], length=14),
        position  # 当前持仓比例 (0-1)
    ]
    return np.concatenate([f.values[-1:] for f in features])

采用离散三动作设计：

买入（仓位增加 30%）
持有（保持当前仓位）
卖出（仓位减少 30%）

好的奖励应该平衡收益与风险：

def calculate_reward(prev_portfolio, current_portfolio, max_drawdown):
    returns = (current_portfolio - prev_portfolio) / prev_portfolio
    risk_penalty = 0.5 * max_drawdown  # 惩罚最大回撤
    return returns - risk_penalty

import pandas as pd
import numpy as np
import talib as ta

# 数据加载与特征工程
def preprocess_data(filepath):
    df = pd.read_csv(filepath)
    df['date'] = pd.to_datetime(df['date'])
    df = df.set_index('date')

    # 添加技术指标
    df['rsi'] = ta.RSI(df['close'], timeperiod=14)
    df['macd'], _, _ = ta.MACD(df['close'])
    df['ma5'] = df['close'].rolling(5).mean()

    # 标准化处理
    for col in ['close', 'volume', 'rsi', 'macd']:
        df[col] = (df[col] - df[col].mean()) / df[col].std()

    return df

import tensorflow as tf

class QNetwork(tf.keras.Model):
    def __init__(self, state_size, action_size):
        super().__init__()
        self.dense1 = tf.keras.layers.Dense(64, activation='relu')
        self.dense2 = tf.keras.layers.Dense(64, activation='relu')
        self.output_layer = tf.keras.layers.Dense(action_size)

    def call(self, state):
        x = self.dense1(state)
        x = self.dense2(x)
        return self.output_layer(x)

class TradingAgent:
    def __init__(self, state_size, action_size):
        self.q_network = QNetwork(state_size, action_size)
        self.optimizer = tf.keras.optimizers.Adam(learning_rate=0.001)
        self.gamma = 0.95  # 折扣因子
        self.epsilon = 1.0  # 探索率

    def train_step(self, states, actions, rewards, next_states, dones):
        # 计算目标 Q 值
        next_q = self.q_network(next_states)
        max_next_q = tf.reduce_max(next_q, axis=1)
        targets = rewards + (1 - dones) * self.gamma * max_next_q

        with tf.GradientTape() as tape:
            q_values = self.q_network(states)
            action_masks = tf.one_hot(actions, depth=3)
            selected_q = tf.reduce_sum(q_values * action_masks, axis=1)
            loss = tf.reduce_mean(tf.square(targets - selected_q))

        grads = tape.gradient(loss, self.q_network.trainable_variables)
        self.optimizer.apply_gradients(zip(grads, self.q_network.trainable_variables))

        return loss

使用 walk-forward 验证而非简单 train-test split
在 Q -Network 中添加 Dropout 层
定期在验证集上评估性能

# 在环境类中添加延迟模拟
class TradingEnv:
    def __init__(self, data, latency=3):
        self.data = data
        self.latency = latency
        self.current_step = latency  # 跳过前 N 条数据

    def step(self, action):
        # 执行动作后，前进 latency 步
        self.current_step += self.latency
        # ... 返回延迟后的状态和奖励