欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。 我们提供了精选的国内外量化投研的 180+ 篇高质量文章，并每日更新最新研究成果，涵盖策略开发、因子分析、风险管理等核心领域。 无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！
引言
想象一下，有一个系统能够自动分析市场数据、做出交易决策，并且能从自己的错误中不断学习——这就是 QuantScope，一个基于强化学习的高频算法交易系统。
本文将带你深入了解这个使用 Q-learning 算法的外汇交易系统，它能够在 CAD/USD 货币对上自动执行交易策略。无论你是对量化交易感兴趣，还是想了解强化学习在金融领域的应用，这篇文章都会给你带来启发。
系统核心概念：Scope（时间尺度）
QuantScope 最独特的设计是引入了 "Scope"（时间尺度）的概念。简单来说，就是让系统同时从不同的时间维度观察市场：
- 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:
return1
return0
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: # 低于阈值，超卖
return1
elif RSI > (100 - threshold): # 高于阈值，超买
return-1
return0
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 isNone:
# 首次遇到该状态-动作对，直接设置为奖励值
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 isnotNone:
return self.get_action(self.states)
returnNone
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 的运行可以概括为以下循环：
- 决策：Q-learning 算法选择最优动作（买入/卖出/观望）
- 进化：表现差的代理被淘汰，表现好的代理获得更大交易量
关键设计亮点
共享学习
所有代理共用一个 Q 值表，这意味着一个代理发现的盈利模式，其他代理可以立即利用。这种集体智慧大大加速了学习过程。
适者生存
当代理的绩效低于阈值时会被移除，同时系统会在需要时生成新的代理。这种机制确保只有有效的策略能够存活。
多尺度分析
不同的时间尺度捕捉不同的市场特征，短期尺度适合捕捉快速波动，长期尺度适合把握大趋势。
总结
QuantScope 展示了如何将强化学习应用于量化交易领域。通过结合技术分析指标和 Q-learning 算法，系统能够：
虽然这是一个教学性质的项目，但它清晰地展示了算法交易系统的核心架构。如果你对量化交易或强化学习感兴趣，这个项目是一个很好的学习起点。
需要注意的是，真实的交易环境远比模拟复杂，存在滑点、手续费、流动性等诸多因素。这个系统更适合作为学习工具，而非直接用于实盘交易。
参考文章
加入专注于财经数据与量化投研的知识星球【数据科学实战】，获取本文完整研究解析、代码实现细节。财经数据与量化投研知识社区
核心权益如下：
- 赠送《财经数据宝典》完整文档，汇集多年财经数据维护经验
- 赠送《量化投研宝典》完整文档，汇集多年量化投研领域经验
- 赠送《PyBroker-入门及实战》视频课程，手把手学习量化策略开发
- 每日分享高质量量化投研文章（已更新 180+篇）、代码和相关资料
星球已有丰富内容积累，包括量化投研论文、财经高频数据、 PyBroker 视频教程、定期直播、数据分享和答疑解难。适合对量化投研和财经数据分析有兴趣的学习者及从业者。欢迎加入我们！
