Python Q-learning强化学习实战：构建多尺度CAD/USD自动交易系统

想象一个系统，能够持续分析市场数据、自主做出交易决策，并能从成败经验中不断学习优化——这就是QuantScope，一个基于强化学习原理构建的高频算法交易系统。本文将深入解析如何使用Python实现Q-learning算法，并将其应用于CAD/USD货币对的自动化交易策略中。无论你是对量化交易感兴趣，还是希望探索人工智能在金融领域的实际应用，本文都将为你提供清晰的实践路径。

系统核心概念：Scope（时间尺度）

QuantScope 最独特的设计在于引入了 “Scope” （时间尺度）的概念。简而言之，系统被设计为能够从不同的时间维度同步观察市场：

• Scope 1：分析每一个市场报价，旨在捕捉瞬时的价格波动。
• Scope 50：每50个报价分析一次，关注中短期的趋势变化。
• Scope 1000：每1000个报价分析一次，用以把握长期的行情走势。

这种多时间尺度的融合分析，使系统既能敏捷响应短线的交易机会，又不至于因噪音干扰而错失重大的趋势行情。

核心模块解析

1. 资金管理模块（Bankroll）

作为系统的“金库”，该模块负责精准记录每一笔资金的流入与流出。

class Bankroll(object):
    def __init__(self, vault, funds):
        self.init_logging(vault)  # 初始化日志系统
        self.bankroll = funds     # 设置初始资金
        self.transactions = 0     # 交易计数器
        self.logger.info('Bankroll initialized with $ {}'.format(funds))

    def transaction(self, val):
        """
        处理每一笔交易
        val 为正数表示盈利入账，负数表示开仓支出
        """
        self.bankroll += val
        self.transactions += 1
        self.logger.info('Transaction {id}: $ {val} added to bankroll: $ {br}'.format(
            id=self.transactions, val=val, br=self.bankroll))

        # 资金安全检查：防止透支
        if self.bankroll < 0:
            raise Exception('We ran out of money')

    def get_bankroll(self):
        """获取当前可用资金"""
        return self.bankroll

2. 技术指标模块（Indicators）

此模块充当系统的“感知器官”，通过计算9个维度的技术指标来量化市场状态。

class Indicators(object):
    def __init__(self, log=None):
        self.logger = log
        # 初始化 9 维状态向量，每个维度代表一个技术指标
        self.state = (0, 0, 0, 0, 0, 0, 0, 0, 0)

    def get_states(self, quotes):
        """
        根据历史报价计算所有技术指标
        返回一个 9 元组，每个元素为 1（看涨）、-1（看跌）或 0（中性）
        """
        self.quotes = quotes
        self.state = (
            self.crossover_indicator(self.quotes, 5, 7),    # 超短期均线交叉
            self.crossover_indicator(self.quotes, 5, 20),   # 短期均线交叉
            self.crossover_indicator(self.quotes, 7, 30),   # 短中期均线交叉
            self.crossover_indicator(self.quotes, 12, 26),  # MACD 标准周期
            self.crossover_indicator(self.quotes, 50, 100), # 中期均线交叉
            self.crossover_indicator(self.quotes, 50, 200), # 长期均线交叉（黄金交叉）
            self.MACD_sig_line(self.quotes, 12, 26, 9),     # MACD 信号线交叉
            self.MACD_zero_cross(self.quotes, 12, 26),      # MACD 零轴交叉
            self.RSI(self.quotes, 14, 25)                   # 相对强弱指标
        )
        return self.state

    def moving_average(self, size, sliced):
        """
        计算指数移动平均线（EMA）
        size: 周期长度
        sliced: 价格数据切片
        """
        multiplier = float((2 / (float(size) + 1)))  # 计算平滑系数
        ema = sum(sliced) / float(size)              # 初始值为简单平均

        # 迭代计算 EMA
        for value in sliced:
            ema = (multiplier * value) + ((1 - multiplier) * ema)

        return ema

    def crossover_indicator(self, q, x, y):
        """
        检测均线交叉信号
        x: 快速均线周期
        y: 慢速均线周期
        返回: 1（金叉）、-1（死叉）、0（无交叉）
        """
        # 当前时刻的均线值
        fast_now = self.moving_average(x, q[-x:])
        slow_now = self.moving_average(y, q[-y:])
        # 前一时刻的均线值
        fast_prev = self.moving_average(x, q[-x-1:-1])
        slow_prev = self.moving_average(y, q[-y-1:-1])

        # 检测死叉：快线从上方穿越到下方
        if fast_now < slow_now and fast_prev > slow_prev:
            return -1
        # 检测金叉：快线从下方穿越到上方
        elif fast_now > slow_now and fast_prev < slow_prev:
            return 1

        return 0

    def RSI(self, q, period, threshold):
        """
        计算相对强弱指标（RSI）
        period: 计算周期
        threshold: 超买超卖阈值
        返回: 1（超卖，买入信号）、-1（超买，卖出信号）、0（中性）
        """
        updays = []   # 上涨日
        downdays = [] # 下跌日

        # 统计涨跌情况
        for i in range(min(period, len(q) - 1)):
            if q[1 + i] < q[i]:
                updays.append(q[1 + i])
            elif q[1 + i] > q[i]:
                downdays.append(q[1 + i])

        # 计算相对强度
        try:
            RS = self.moving_average(period, updays) / self.moving_average(period, downdays)
        except:
            RS = 0

        # 计算 RSI 值（0-100 区间）
        RSI = 100 - (100 / (1 + RS))

        # 判断超买超卖
        if RSI < threshold:       # 低于阈值，超卖
            return 1
        elif RSI > (100 - threshold):  # 高于阈值，超买
            return -1

        return 0

3. Q-learning 学习模块（QLearn）

这是系统的“决策大脑”，通过强化学习框架持续优化交易策略。

class QLearn(object):
    def __init__(self, all_actions, state_size, alpha):
        self.alpha = alpha         # 学习率
        self.all_actions = all_actions  # 可选动作：买入(1)、卖出(-1)、观望(0)
        self.state_sz = state_size  # 状态空间维度
        self.q = {}                 # Q 值表

    def Q(self, s, a):
        """
        查询 Q 值表
        s: 当前状态
        a: 执行的动作
        返回: 该状态-动作对的期望收益
        """
        return self.q.get((tuple(s), a), 0.0)  # 未见过的状态返回 0

    def get_action(self, s):
        """
        根据当前状态选择最优动作
        采用 ε-贪婪策略：90% 选最优，10% 随机探索
        """
        # 状态维度检查
        if len(s) != self.state_sz:
            raise Exception('invalid state dim')

        # 10% 概率随机探索，避免陷入局部最优
        if random.random() < 0.10:
            return random.choice(self.all_actions)

        # 90% 概率选择 Q 值最大的动作
        best_action = self.all_actions[0]
        best_q_val = float('-inf')

        for action in self.all_actions:
            q_val = self.Q(s, action)
            if q_val > best_q_val:
                best_q_val = q_val
                best_action = action

        return best_action

    def updateQ(self, state, action, reward, value):
        """
        更新 Q 值表
        使用 Q-learning 更新公式：Q(s,a) ← Q(s,a) + α[target - Q(s,a)]
        """
        oldv = self.q.get((tuple(state), action), None)

        if oldv is None:
            # 首次遇到该状态-动作对，直接设置为奖励值
            self.q[(tuple(state), action)] = reward
        else:
            # 增量更新：新值 = 旧值 + 学习率 × (目标值 - 旧值)
            self.q[(tuple(state), action)] = oldv + self.alpha * (value - oldv)

    def learnQ(self, state1, action, state2, reward):
        """
        完整的 Q-learning 学习过程
        state1: 动作前的状态
        action: 执行的动作
        state2: 动作后的状态
        reward: 获得的奖励（交易盈亏）
        """
        # 计算新状态下的最大 Q 值
        best_q_new = max([self.Q(state2, a) for a in self.all_actions])

        # 更新 Q 值表
        target = reward + self.alpha * best_q_new
        self.updateQ(state1, action, reward, target)

4. 智能交易代理（Agent）

Agent 是一个集感知、决策与执行于一体的完整交易实体。

class Agent(Learning, Indicators, Order):
    """
    交易代理：继承自 Learning（学习能力）、Indicators（市场感知）、Order（交易执行）
    每个代理同一时间最多持有一个仓位
    """

    def __init__(self, scope, q, alpha, reward, discount, quotes, bankroll, log=None):
        self.logger = log
        self.scope = scope
        self.actions = [1, -1, 0]  # 买入、卖出、观望

        # 初始化三个父类
        Indicators.__init__(self, log)
        Order.__init__(self, scope, bankroll, log)
        Learning.__init__(self, q, alpha, reward, discount, self.state, self.actions)

        self.num_trades = 0      # 交易次数
        self.performance = 1     # 绩效得分
        self.volume = max(self.performance, 1)  # 交易量与绩效挂钩
        self.status = {'status': 'idle', 'action': ''}  # 当前状态
        self.quotes = quotes

    def learn(self):
        """
        感知市场并做出决策
        返回推荐的交易动作
        """
        # 保存上一个状态
        self.prev_states = self.states
        # 根据最新行情计算当前状态
        self.states = self.get_states(self.quotes)

        # 确保有足够的历史数据再做决策
        if self.prev_states is not None:
            return self.get_action(self.states)
        return None

    def trade(self):
        """
        执行交易逻辑
        根据学习模块的建议开仓或平仓
        """
        response = self.learn()

        if response == 1:  # 买入信号
            if self.status['status'] != 'open':
                self.open_position(order='buy')  # 开多仓
            elif self.status['action'] == 'sell':
                self.close_position()  # 平空仓

        elif response == -1:  # 卖出信号
            if self.status['status'] != 'open':
                self.open_position(order='sell')  # 开空仓
            elif self.status['action'] == 'buy':
                self.close_position()  # 平多仓

    def open_position(self, order):
        """开仓"""
        self.open_order(order, self.quotes[-1], self.volume)
        self.status['status'] = 'open'
        self.status['action'] = order
        self.num_trades += 1

    def close_position(self):
        """平仓并学习"""
        self.close_order(self.status['action'], self.quotes[-1])
        profit = self.get_profit()

        # 关键：用交易结果更新 Q 值表
        self.learnQ(self.states, self.status['action'], self.prev_states, profit)

        # 更新绩效
        self.update_performance(profit)
        self.status['status'] = 'idle'

    def update_performance(self, profit):
        """
        更新代理绩效
        绩效影响未来的交易量：表现好的代理获得更大的交易权限
        """
        self.performance += profit * self.volume * self.num_trades

5. 总控模块（Executive）

Executive 作为系统的“指挥官”，负责协调所有模块的协同运作。

class Executive(object):
    def __init__(self):
        self.init_logging()
        self.bankroll = Bankroll('logs/bankroll.log', 1000)  # 初始资金 1000 美元
        self.all_quotes = []  # 全部历史数据
        self.quotes = []      # 已处理的数据
        self.scopes = []      # 时间尺度列表

        self.load_csv()       # 加载历史行情
        self.load_scopes()    # 创建多个时间尺度

    def supervise(self):
        """
        主循环：驱动整个交易模拟
        """
        hop = 0 # 当前时间步

        while hop < len(self.all_quotes):
            self.logger.info('Hop {hop} Bankroll: {bankroll}'.format(
                hop=hop, bankroll=self.bankroll.get_bankroll()))

            # 获取新报价
            new_quote = self.get_new_quote(hop)

            # 更新所有活跃的时间尺度
            for scope in self.active_scopes(hop):
                scope.refresh(new_quote)  # 更新代理状态
                scope.trade()             # 执行交易

            hop += 1

    def active_scopes(self, hop):
        """
        生成器：返回当前时间步应该激活的时间尺度
        例如：hop=100 时，Scope 1 和 Scope 50 都会激活，但 Scope 1000 不会
        """
        for scope in self.scopes:
            if hop % scope.scope == 0:
                yield scope

    def load_scopes(self):
        """创建多个时间尺度，每个尺度独立运行一组交易代理"""
        for scope_val in {1, 50, 1000}:
            self.scopes.append(Scope(
                scope_val,           # 时间分辨率
                {},                  # 共享的 Q 值表
                0.7,                 # 学习率
                tuple(),             # 奖励结构
                0.314,               # 折扣因子
                11,                  # 最大代理数量
                self.quotes,
                self.bankroll,
                self.logger
            ))

# 程序入口
if __name__ == "__main__":
    trader = Executive()
    trader.supervise()

系统工作流程

整个 QuantScope 系统的运行可以概括为一个持续优化的循环：

1. 观察：各Agent通过技术指标模块感知并量化当前市场状态。
2. 决策：Q-learning 算法基于当前状态和历史经验，选择最优动作（买入、卖出或观望）。
3. 执行：通过订单模块将决策转化为实际的交易操作。
4. 学习：根据交易产生的盈亏（奖励），更新Q值表，修正对状态-动作价值的评估。
5. 进化：绩效持续低下的Agent被淘汰，表现优异的Agent则获得更大的资金权重，实现策略的优胜劣汰。

关键设计亮点

• 共享学习：所有交易代理共享同一个中央Q值表。这意味着任何一个代理发现的盈利模式，都能即时被其他代理学习与利用，形成了高效的“集体智慧”，极大加速了系统整体的学习进程。
• 适者生存：系统内嵌了代理绩效评估与淘汰机制。当代理的绩效低于设定阈值时会被移除，同时系统会在适当时机生成新的代理进行探索。这确保了活跃的策略池始终由相对有效的策略主导。
• 多尺度分析：通过并行的多个时间尺度（Scope），系统能够同步捕捉不同时间维度的市场特征。短期尺度擅长捕捉瞬时波动带来的套利机会，而长期尺度则有助于把握和跟随主要的趋势行情。

总结

QuantScope 项目清晰地展示了如何将强化学习技术落地于量化交易领域。通过整合经典的技术分析指标与现代的Q-learning算法，该系统实现了：

• 从历史数据中自动学习并优化交易策略。
• 在多个时间尺度上并发进行市场分析。
• 通过绩效机制动态优化代理种群。
• 达成全自动的交易决策与执行闭环。

尽管本项目更侧重于教学与原理演示，但它完整勾勒了一个自动化算法交易系统的核心架构。对于希望入门量化交易或强化学习应用的开发者而言，这是一个极具价值的实践范本。需要注意的是，实盘交易环境远比模拟复杂，需考虑滑点、手续费、流动性限制、市场冲击成本等诸多因素。因此，该系统更适合作为学习和研究工具，在投入实盘前必须经过更加严格和全面的测试与验证。