RQAlpha 策略示例实战指南：从买入持有到期货配对交易的 7 个完整回测代码

• 金融科技

RQAlpha 是一个可扩展、可替换的 Python 量化回测与交易框架，支持股票、股指期货、商品期货等多种证券。本文以 RQAlpha 官方文档「策略示例」（docs/source/intro/examples.rst）为骨架，完整讲解 7 个可直接运行的策略范例——从最简单的买入持有，到基于 TA-Lib 的 Golden Cross、MACD、RSI，再到海龟交易系统、股指期货 MACD 与商品期货跨品种配对交易。读完本文，你将掌握 RQAlpha 约定函数（init/before_trading/handle_bar/after_trading）的写法、股票与期货两套交易 API 的用法（order_percent、order_shares、order_target_value、buy_open、sell_close等）、历史数据查询（history_bars）与行情订阅（subscribe/update_universe）机制，并能够用一条rqalpha run命令把每个策略跑起来验证结果。

运行前提：示例代码与回测命令

文档中所有示例策略都遵循 RQAlpha 的约定函数框架：在init(context)中初始化全局变量、在handle_bar(context, bar_dict)中编写主逻辑，数据更新会自动触发。文档中的代码片段为保持简洁省略了 API 导入，而仓库中对应可运行版本（如 rqalpha/examples/buy_and_hold.py、rqalpha/examples/golden_cross.py）在文件开头统一添加了：

from rqalpha.apis import *

这一行将 RQAlpha 暴露的所有用户 API（logger、order_percent、history_bars、get_position、subscribe等）注入命名空间。API 的注册机制可以在 rqalpha/api.py 与 rqalpha/apis/api_abstract.py 中查看——每个 API 通过@export_as_api装饰器导出，并带有一系列参数校验与执行阶段约束装饰器。

以第一个策略为例，回测时需要明确以下要素，可通过 config.yml 传参或命令行传参：

假设策略存放在./rqalpha/examples/buy_and_hold.py，回测区间为2016-06-01至2016-12-01，起始资金 100000，基准设为沪深300，则运行命令为：

rqalpha run -f ./rqalpha/examples/buy_and_hold.py -s 2016-06-01 -e 2016-12-01 --account stock 100000 --benchmark 000300.XSHG

如果想要以图形方式查看回测结果，增加--plot参数：

rqalpha run -f ./rqalpha/examples/buy_and_hold.py -s 2016-06-01 -e 2016-12-01 --account stock 100000 --benchmark 000300.XSHG --plot

如果想把回测数据保存下来，用-o参数将结果保存成pkl文件，回测结束后可通过pandas.read_pickle读取进行后续分析：

rqalpha run -f ./rqalpha/examples/buy_and_hold.py -s 2016-06-01 -e 2016-12-01 --account stock 100000 --benchmark 000300.XSHG --plot -o result.pkl

import pandas as pd result_dict = pd.read_pickle('result.pkl') result_dict.keys() # [out]dict_keys(['total_portfolios', 'summary', 'benchmark_portfolios', 'benchmark_positions', 'stock_positions', 'trades', 'stock_portfolios'])

关于数据源同步与安装细节，可参考 docs/source/intro/install.rst 与 docs/source/intro/run_algorithm.rst。

第一个策略：买入并持有（Buy and Hold）

万事开头难，这是一个最简单的策略：在回测开始的第一天买入资金量 100% 的平安银行（000001.XSHE）并且一直持有。

# 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。 def init(context): logger.info("init") context.s1 = "000001.XSHE" update_universe(context.s1) # 是否已发送了order context.fired = False context.cnt = 1 def before_trading(context): logger.info("Before Trading", context.cnt) context.cnt += 1 # 你选择的证券的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新 def handle_bar(context, bar_dict): context.cnt += 1 logger.info("handle_bar", context.cnt) # 开始编写你的主要的算法逻辑 # bar_dict[order_book_id] 可以拿到某个证券的bar信息 # context.portfolio 可以拿到现在的投资组合状态信息 # 使用order_shares(id_or_ins, amount)方法进行落单 # TODO: 开始编写你的算法吧！ if not context.fired: # order_percent并且传入1代表买入该股票并且使其占有投资组合的100% order_percent(context.s1, 1) context.fired = True

这个策略展示了 RQAlpha 三个核心约定函数的协作：

• init(context)：策略启动时执行一次。这里用update_universe(context.s1)将平安银行加入关注证券池，并在context上保存全局状态（fired标记是否已发过单）。从源码看（rqalpha/apis/api_base.py），update_universe是覆盖式操作而非增量添加，会在下一个 bar 事件触发时生效。
• before_trading(context)：每天交易开始前调用一次，适合做每日初始化。
• handle_bar(context, bar_dict)：每个 bar 数据更新时触发。bar_dict[order_book_id]拿到某个证券的 bar 信息，context.portfolio拿到投资组合状态。

下单使用order_percent(context.s1, 1)：根据 rqalpha/apis/api_abstract.py 的签名，order_percent发送一个花费价值等于目前投资组合（市场价值和现金总和）一定百分比现金的买卖单，percent需在[-1, 1]区间内，1即满仓买入；股数会自动向下调整为一手（A 股 100 股）的倍数，资金不足时会使用最大可用资金发单。

回测结果如下所示（对应文档中的买入持有策略回测图）：

Golden Cross 算法示例

Golden Cross（金叉）是最经典的趋势跟随策略：短期均线上穿长期均线时买入，下穿时卖出。以下使用 TA-Lib 的SMA（simple moving average）方法实现：

import talib # 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。 def init(context): context.s1 = "000001.XSHE" # 设置这个策略当中会用到的参数，在策略中可以随时调用，这个策略使用长短均线，我们在这里设定长线和短线的区间，在调试寻找最佳区间的时候只需要在这里进行数值改动 context.SHORTPERIOD = 20 context.LONGPERIOD = 120 # 你选择的证券的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新 def handle_bar(context, bar_dict): # 开始编写你的主要的算法逻辑 # bar_dict[order_book_id] 可以拿到某个证券的bar信息 # context.portfolio 可以拿到现在的投资组合状态信息 # 使用order_shares(id_or_ins, amount)方法进行落单 # TODO: 开始编写你的算法吧！ # 因为策略需要用到均线，所以需要读取历史数据 prices = history_bars(context.s1, context.LONGPERIOD+1, '1d', 'close') # 使用talib计算长短两根均线，均线以array的格式表达 short_avg = talib.SMA(prices, context.SHORTPERIOD) long_avg = talib.SMA(prices, context.LONGPERIOD) plot("short avg", short_avg[-1]) plot("long avg", long_avg[-1]) # 获取当前投资组合中股票的仓位 cur_position = get_position(context.s1).quantity # 计算现在portfolio中的现金可以购买多少股票 shares = context.portfolio.cash/bar_dict[context.s1].close # 如果短均线从上往下跌破长均线，也就是在目前的bar短线平均值低于长线平均值，而上一个bar的短线平均值高于长线平均值 if short_avg[-1] - long_avg[-1] < 0 and short_avg[-2] - long_avg[-2] > 0 and cur_position > 0: # 进行清仓 order_target_value(context.s1, 0) # 如果短均线从下往上突破长均线，为入场信号 if short_avg[-1] - long_avg[-1] > 0 and short_avg[-2] - long_avg[-2] < 0: # 满仓入股 order_shares(context.s1, shares)

本策略引入了三个重要 API：

• history_bars(order_book_id, bar_count, frequency, fields)：获取指定合约的历史 k 线行情，返回numpy.ndarray。依据 rqalpha/apis/api_base.py 的说明，frequency支持'1d'、'1m'、'1w'；fields支持datetime、open、high、low、close、volume、total_turnover，期货另有open_interest、settlement、prev_settlement等字段；还可传skip_suspended（是否跳过停牌数据）、include_now（是否包含当前数据）、adjust_type（复权类型，默认pre前复权，可选pre/none/post）。注意：日回测中无法获取分钟历史数据，history_bars也不能在init中调用。
• plot(name, value)：绘制指标曲线，出现在回测结果图与输出报告中。
• get_position(order_book_id, direction)：获取某标的的持仓对象，direction默认为POSITION_DIRECTION.LONG；cur_position = get_position(context.s1).quantity取得当前持股数量。

交易逻辑上，short_avg[-1] < long_avg[-1]且short_avg[-2] > long_avg[-2]表示短均线刚刚下穿长均线（死叉），此时若有持仓则调用order_target_value(context.s1, 0)清仓；反之短均线上穿长均线（金叉）则用全部现金买入。order_target_value会买入/卖出并自动调整该证券仓位到目标价值（见 rqalpha/apis/api_abstract.py），传入 0 即为清仓；order_shares则是指定股数下单，负数代表卖出（rqalpha/apis/api_abstract.py）。

Golden Cross 策略运行命令：

rqalpha run -s 2014-01-01 -e 2016-01-01 -f rqalpha/examples/golden_cross.py --account stock 100000 -p -bm 000001.XSHE

其中-p等价于--plot，-bm等价于--benchmark。使用plot函数绘制的内容（short avg / long avg）会出现在输出的结果中：

单股票 MACD 算法示例

MACD 策略使用 TA-Lib 的MACD方法，通过 MACD 线突破 signal 线产生交易信号。MACD 由长短均线之差（DIF）及其均线（DEA/signal）构成：

import talib # 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。 def init(context): context.s1 = "000001.XSHE" # 使用MACD需要设置长短均线和macd平均线的参数 context.SHORTPERIOD = 12 context.LONGPERIOD = 26 context.SMOOTHPERIOD = 9 context.OBSERVATION = 100 # 你选择的证券的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新 def handle_bar(context, bar_dict): # 开始编写你的主要的算法逻辑 # bar_dict[order_book_id] 可以拿到某个证券的bar信息 # context.portfolio 可以拿到现在的投资组合状态信息 # 使用order_shares(id_or_ins, amount)方法进行落单 # TODO: 开始编写你的算法吧！ # 读取历史数据，使用sma方式计算均线准确度和数据长度无关，但是在使用ema方式计算均线时建议将历史数据窗口适当放大，结果会更加准确 prices = history_bars(context.s1, context.OBSERVATION,'1d','close') # 用Talib计算MACD取值，得到三个时间序列数组，分别为macd, signal 和 hist macd, signal, hist = talib.MACD(prices, context.SHORTPERIOD, context.LONGPERIOD, context.SMOOTHPERIOD) plot("macd", macd[-1]) plot("macd signal", signal[-1]) # macd 是长短均线的差值，signal是macd的均线，使用macd策略有几种不同的方法，我们这里采用macd线突破signal线的判断方法 # 如果macd从上往下跌破macd_signal if macd[-1] - signal[-1] < 0 and macd[-2] - signal[-2] > 0: # 获取当前投资组合中股票的仓位 curPosition = get_position(context.s1).quantity #进行清仓 if curPosition > 0: order_target_value(context.s1, 0) # 如果短均线从下往上突破长均线，为入场信号 if macd[-1] - signal[-1] > 0 and macd[-2] - signal[-2] < 0: # 满仓入股 order_target_percent(context.s1, 1)

参数含义：SHORTPERIOD = 12、LONGPERIOD = 26是 MACD 的标准快慢均线周期，SMOOTHPERIOD = 9是 signal 线的平滑周期，OBSERVATION = 100是历史数据窗口。talib.MACD返回三个时间序列数组：macd（长短均线差值）、signal（macd 的均线）、hist（柱状图，即两者之差）。文档特别提示：使用 SMA 方式计算均线时准确度与数据长度无关，但使用 EMA 方式时建议将历史数据窗口适当放大，结果会更准确。

入场使用order_target_percent(context.s1, 1)：按 rqalpha/apis/api_abstract.py，它买入/卖出证券以自动调整该证券仓位到占投资组合总价值的目标百分比（percent需在[0, 1]内），其内部计算position_to_adjust = target_position - current_position，正值买入、负值卖出。文档还提示：若需大量调整股票仓位，推荐使用order_target_portfolio（见 rqalpha/mod/rqalpha_mod_sys_accounts/api/order_target_portfolio.py），性能更好。

多股票 RSI 算法示例

RSI（相对强弱指标）策略对多只股票组成的股票池循环计算 RSI，超买（RSI 高于阈值）清仓、超卖（RSI 低于阈值）按剩余现金比例买入：

import talib # 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。 def init(context): # 选择我们感兴趣的股票 context.s1 = "000001.XSHE" context.s2 = "601988.XSHG" context.s3 = "000068.XSHE" context.stocks = [context.s1, context.s2, context.s3] context.TIME_PERIOD = 14 context.HIGH_RSI = 85 context.LOW_RSI = 30 context.ORDER_PERCENT = 0.3 # 你选择的证券的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新 def handle_bar(context, bar_dict): # 开始编写你的主要的算法逻辑 # bar_dict[order_book_id] 可以拿到某个证券的bar信息 # context.portfolio 可以拿到现在的投资组合状态信息 # 使用order_shares(id_or_ins, amount)方法进行落单 # TODO: 开始编写你的算法吧！ # 对我们选中的股票集合进行loop，运算每一只股票的RSI数值 for stock in context.stocks: # 读取历史数据 prices = history_bars(stock, context.TIME_PERIOD+1, '1d', 'close') # 用Talib计算RSI值 rsi_data = talib.RSI(prices, timeperiod=context.TIME_PERIOD)[-1] cur_position = get_position(stock).quantity # 用剩余现金的30%来购买新的股票 target_available_cash = context.portfolio.cash * context.ORDER_PERCENT # 当RSI大于设置的上限阀值，清仓该股票 if rsi_data > context.HIGH_RSI and cur_position > 0: order_target_value(stock, 0) # 当RSI小于设置的下限阀值，用剩余cash的一定比例补仓该股 if rsi_data < context.LOW_RSI: logger.info("target available cash caled: " + str(target_available_cash)) # 如果剩余的现金不够一手 - 100shares，那么会被ricequant 的order management system reject掉 order_value(stock, target_available_cash)

这个示例展示了多标的策略的典型写法：

• 在init中用列表context.stocks维护股票池（平安银行000001.XSHE、中国银行601988.XSHG、赛格三星000068.XSHE），handle_bar中for stock in context.stocks循环处理。
• talib.RSI(prices, timeperiod=14)[-1]取最新一期的 RSI 值，HIGH_RSI = 85、LOW_RSI = 30为超买超卖阈值。
• 买入使用order_value(stock, target_available_cash)：按 rqalpha/apis/api_abstract.py 的说明，order_value用想花费的金额买入股票，股数会自动调整成 100 的倍数；当提交买单时，cash_amount表示希望买入股票消耗的金额（包含税费），最终买入股数与税费参数设置相关。文档特别提醒：如果剩余现金不够一手（100 股），订单会被订单管理系统 reject。
• 清仓使用order_target_value(stock, 0)。

海龟交易系统

海龟交易系统（Turtle Trading）是经典的趋势跟踪策略，以 ATR（平均真实波幅）衡量波动并决定仓位大小，采用「突破入场、分批加仓、ATR 止损」的规则：

import numpy as np import talib import math def get_extreme(array_high_price_result, array_low_price_result): np_array_high_price_result = np.array(array_high_price_result[:-1]) np_array_low_price_result = np.array(array_low_price_result[:-1]) max_result = np_array_high_price_result.max() min_result = np_array_low_price_result.min() return [max_result, min_result] def get_atr_and_unit( atr_array_result, atr_length_result, portfolio_value_result): atr = atr_array_result[ atr_length_result-1] unit = math.floor(portfolio_value_result * .01 / atr) return [atr, unit] def get_stop_price(first_open_price_result, units_hold_result, atr_result): stop_price = first_open_price_result - 2 * atr_result \ + (units_hold_result - 1) * 0.5 * atr_result return stop_price def init(context): context.trade_day_num = 0 context.unit = 0 context.atr = 0 context.trading_signal = 'start' context.pre_trading_signal = '' context.units_hold_max = 4 context.units_hold = 0 context.quantity = 0 context.max_add = 0 context.first_open_price = 0 context.s = '000300.XSHG' context.open_observe_time = 55 context.close_observe_time = 20 context.atr_time = 20 def handle_bar(context, bar_dict): portfolio_value = context.portfolio.portfolio_value high_price = history_bars(context.s, context.open_observe_time+1, '1d', 'high') low_price_for_atr = history_bars(context.s, context.open_observe_time+1, '1d', 'low') low_price_for_extreme = history_bars(context.s, context.close_observe_time+1, '1d', 'low') close_price = history_bars(context.s, context.open_observe_time+2, '1d', 'close') close_price_for_atr = close_price[:-1] atr_array = talib.ATR(high_price, low_price_for_atr, close_price_for_atr, timeperiod=context.atr_time) maxx = get_extreme(high_price, low_price_for_extreme)[0] minn = get_extreme(high_price, low_price_for_extreme)[1] atr = atr_array[-2] if context.trading_signal != 'start': if context.units_hold != 0: context.max_add += 0.5 * get_atr_and_unit(atr_array, atr_array.size, portfolio_value)[0] else: context.max_add = bar_dict[context.s].last cur_position = get_position(context.s).quantity available_cash = context.portfolio.cash market_value = context.portfolio.market_value if (cur_position > 0 and bar_dict[context.s].last < get_stop_price(context.first_open_price, context.units_hold, atr)): context.trading_signal = 'stop' else: if cur_position > 0 and bar_dict[context.s].last < minn: context.trading_signal = 'exit' else: if (bar_dict[context.s].last > context.max_add and context.units_hold != 0 and context.units_hold < context.units_hold_max and available_cash > bar_dict[context.s].last*context.unit): context.trading_signal = 'entry_add' else: if bar_dict[context.s].last > maxx and context.units_hold == 0: context.max_add = bar_dict[context.s].last context.trading_signal = 'entry' atr = get_atr_and_unit(atr_array, atr_array.size, portfolio_value)[0] if context.trade_day_num % 5 == 0: context.unit = get_atr_and_unit(atr_array, atr_array.size, portfolio_value)[1] context.trade_day_num += 1 context.quantity = context.unit if (context.trading_signal != context.pre_trading_signal or (context.units_hold < context.units_hold_max and context.units_hold > 1) or context.trading_signal == 'stop'): if context.trading_signal == 'entry': context.quantity = context.unit if available_cash > bar_dict[context.s].last*context.quantity: order_shares(context.s, context.quantity) context.first_open_price = bar_dict[context.s].last context.units_hold = 1 if context.trading_signal == 'entry_add': context.quantity = context.unit order_shares(context.s, context.quantity) context.units_hold += 1 if context.trading_signal == 'stop': if context.units_hold > 0: order_shares(context.s, -context.quantity) context.units_hold -= 1 if context.trading_signal == 'exit': if cur_position > 0: order_shares(context.s, -cur_position) context.units_hold = 0 context.pre_trading_signal = context.trading_signal

海龟策略的机制要点：

• ATR 与仓位：talib.ATR(high, low, close, timeperiod=20)计算平均真实波幅；unit = floor(portfolio_value * 1% / atr)定义「1 单位」仓位——每单位风险控制在组合价值的 1% 内，每 5 个交易日（trade_day_num % 5 == 0）更新一次。
• 突破入场：bar_dict[context.s].last > maxx（55 日最高价突破）且无持仓时触发entry；已持仓时价格突破max_add（逐级抬高的加仓线，每次加仓后上移 0.5 倍 ATR）则触发entry_add，最多加到units_hold_max = 4单位。
• 止损与退出：stop_price = first_open_price - 2*ATR + (units_hold-1)*0.5*ATR为逐单位递进止损线，跌破触发stop（减一单位）；价格跌破 20 日最低价minn触发exit（全部清仓）。
• 下单统一使用order_shares，卖出的amount传负数（如order_shares(context.s, -context.quantity)）。

完整的可运行版本（含from rqalpha.apis import *）位于 rqalpha/examples/turtle.py，集成测试参考 tests/integration_tests/test_backtest_results/test_s_turtle.py。

股指期货 MACD 日回测

前面的示例都基于股票账户，这个示例转向期货：使用 TA-Lib 对股指期货主力合约做日级别 MACD 回测，并展示期货专用的开平仓 API 与持仓方向概念：

# 可以自己import我们平台支持的第三方python模块，比如pandas、numpy等 import talib # 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递 def init(context): # context内引入全局变量s1，存储目标合约信息 context.s1 = 'IF1606' # 使用MACD需要设置长短均线和macd平均线的参数 context.SHORTPERIOD = 12 context.LONGPERIOD = 26 context.SMOOTHPERIOD = 9 context.OBSERVATION = 50 #初始化时订阅合约行情。订阅之后的合约行情会在handle_bar中进行更新 subscribe(context.s1) # 你选择的期货数据更新将会触发此段逻辑，例如日线或分钟线更新 def handle_bar(context, bar_dict): # 开始编写你的主要的算法逻辑 # 获取历史收盘价序列，history_bars函数直接返回ndarray，方便之后的有关指标计算 prices = history_bars(context.s1, context.OBSERVATION, '1d', 'close') # 用Talib计算MACD取值，得到三个时间序列数组，分别为macd,signal 和 hist macd, signal, hist = talib.MACD(prices, context.SHORTPERIOD, context.LONGPERIOD, context.SMOOTHPERIOD) # macd 是长短均线的差值，signal是macd的均线，如果短均线从下往上突破长均线，为入场信号，进行买入开仓操作 if macd[-1] - signal[-1] > 0 and macd[-2] - signal[-2] < 0: sell_qty = get_position(context.s1, POSITION_DIRECTION.SHORT).quantity # 先判断当前卖方仓位，如果有，则进行平仓操作 if sell_qty > 0: buy_close(context.s1, 1) # 买入开仓 buy_open(context.s1, 1) if macd[-1] - signal[-1] < 0 and macd[-2] - signal[-2] > 0: buy_qty = get_position(context.s1, POSITION_DIRECTION.LONG).quantity # 先判断当前买方仓位，如果有，则进行平仓操作 if buy_qty > 0: sell_close(context.s1, 1) # 卖出开仓 sell_open(context.s1, 1)

本示例的关键知识点：

• subscribe(context.s1)：初始化时订阅合约行情。依据 rqalpha/apis/api_base.py 的说明，日级别回测中不需要订阅合约；分钟回测中若设置了期货账户，则需要订阅策略关注的期货合约，框架会根据订阅的期货合约品种触发对应交易时间的handle_bar；tick 回测中则需订阅每一个关注的标的。subscribe也可直接传入合约列表。
• 持仓方向：POSITION_DIRECTION.LONG（多）与POSITION_DIRECTION.SHORT（空），get_position(context.s1, POSITION_DIRECTION.SHORT).quantity获取空头持仓量。POSITION_DIRECTION等常量由 rqalpha/apis/api_base.py 通过export_as_api导出。
• 期货四元操作（见 rqalpha/apis/api_abstract.py）：
  • buy_open(id_or_ins, amount)：买入开仓；
  • sell_close(id_or_ins, amount)：平买仓（卖出平多）；
  • sell_open(id_or_ins, amount)：卖出开仓（开空）；
  • buy_close(id_or_ins, amount)：平卖仓（买入平空）；
  • 四个函数均支持close_today参数（是否发平今仓单，默认False发平仓单）。

策略逻辑：MACD 上穿 signal 时，若存在空头仓位先buy_close平空，再buy_open开多；MACD 下穿 signal 时，若存在多头仓位先sell_close平多，再sell_open开空。文档使用的合约IF1606是 2016 年 6 月到期的沪深300股指期货合约。

商品期货跨品种配对交易

最后一个示例是分钟级别的配对交易（Pairs Trading）策略，运用简单移动平均与布林带作为交易信号源，以黄金（AU1612）与白银（AG1612）两种商品期货为例，做跨品种价差套利。策略中的移动窗口选择为 60 分钟，即每天开盘 60 分钟内不做任何交易，积累数据计算移动平均值；对冲比率（HedgeRatio）可通过研究平台使用 statsmodels 的 OLS 方法对历史价格序列做线性回归估计得到，具体估计窗口可根据策略需要自行选择。简单起见，示例中期货价格未做对数差处理：

import numpy as np # 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。 def init(context): context.s1 = 'AG1612' context.s2 = 'AU1612' # 设置全局计数器 context.counter = 0 # 设置滚动窗口 context.window = 60 # 设置对冲手数,通过研究历史数据进行价格序列回归得到该值 context.ratio = 15 context.up_cross_up_limit = False context.down_cross_down_limit = False # 设置入场临界值 context.entry_score = 2 # 初始化时订阅合约行情。订阅之后的合约行情会在handle_bar中进行更新 subscribe([context.s1, context.s2]) # before_trading此函数会在每天交易开始前被调用，当天只会被调用一次 def before_trading(context): # 样例商品期货在回测区间内有夜盘交易,所以在每日开盘前将计数器清零 context.counter = 0 # 你选择的期货数据更新将会触发此段逻辑，例如日线或分钟线更新 def handle_bar(context, bar_dict): # 获取当前一对合约的仓位情况。如尚未有仓位,则对应持仓量都为0 long_pos_a = get_position(context.s1, POSITION_DIRECTION.LONG) short_pos_a = get_position(context.s1, POSITION_DIRECTION.SHORT) long_pos_b = get_position(context.s2, POSITION_DIRECTION.LONG) short_pos_b = get_position(context.s2, POSITION_DIRECTION.SHORT) context.counter += 1 # 当累积满一定数量的bar数据时候,进行交易逻辑的判断 if context.counter > context.window: # 获取当天历史分钟线价格队列 price_array_a = history_bars(context.s1, context.window, '1m', 'close') price_array_b = history_bars(context.s2, context.window, '1m', 'close') # 计算价差序列、其标准差、均值、上限、下限 spread_array = price_array_a - context.ratio * price_array_b std = np.std(spread_array) mean = np.mean(spread_array) up_limit = mean + context.entry_score * std down_limit = mean - context.entry_score * std # 获取当前bar对应合约的收盘价格并计算价差 price_a = bar_dict[context.s1].close price_b = bar_dict[context.s2].close spread = price_a - context.ratio * price_b # 如果价差低于预先计算得到的下限,则为建仓信号,'买入'价差合约 if spread <= down_limit and not context.down_cross_down_limit: # 可以通过logger打印日志 logger.info('spread: {}, mean: {}, down_limit: {}'.format(spread, mean, down_limit)) logger.info('创建买入价差中...') # 获取当前剩余的应建仓的数量 qty_a = 1 - long_pos_a.quantity qty_b = context.ratio - short_pos_b.sell_quantity # 由于存在成交不超过下一bar成交量25%的限制,所以可能要通过多次发单成交才能够成功建仓 if qty_a > 0: buy_open(context.s1, qty_a) if qty_b > 0: sell_open(context.s2, qty_b) if qty_a == 0 and qty_b == 0: # 已成功建立价差的'多仓' context.down_cross_down_limit = True logger.info('买入价差仓位创建成功!') # 如果价差向上回归移动平均线,则为平仓信号 if spread >= mean and context.down_cross_down_limit: logger.info('spread: {}, mean: {}, down_limit: {}'.format(spread, mean, down_limit)) logger.info('对买入价差仓位进行平仓操作中...') # 由于存在成交不超过下一bar成交量25%的限制,所以可能要通过多次发单成交才能够成功建仓 qty_a = long_pos_a.quantity qty_b = short_pos_b.quantity if qty_a > 0: sell_close(context.s1, qty_a) if qty_b > 0: buy_close(context.s2, qty_b) if qty_a == 0 and qty_b == 0: context.down_cross_down_limit = False logger.info('买入价差仓位平仓成功!') # 如果价差高于预先计算得到的上限,则为建仓信号,'卖出'价差合约 if spread >= up_limit and not context.up_cross_up_limit: logger.info('spread: {}, mean: {}, up_limit: {}'.format(spread, mean, up_limit)) logger.info('创建卖出价差中...') qty_a = 1 - short_pos_a.quantity qty_b = context.ratio - long_pos_b.quantity if qty_a > 0: sell_open(context.s1, qty_a) if qty_b > 0: buy_open(context.s2, qty_b) if qty_a == 0 and qty_b == 0: context.up_cross_up_limit = True logger.info('卖出价差仓位创建成功') # 如果价差向下回归移动平均线,则为平仓信号 if spread < mean and context.up_cross_up_limit: logger.info('spread: {}, mean: {}, up_limit: {}'.format(spread, mean, up_limit)) logger.info('对卖出价差仓位进行平仓操作中...') qty_a = short_pos_a.quantity qty_b = long_pos_b.quantity if qty_a > 0: buy_close(context.s1, qty_a) if qty_b > 0: sell_close(context.s2, qty_b) if qty_a == 0 and qty_b == 0: context.up_cross_up_limit = False logger.info('卖出价差仓位平仓成功!')

该策略的完整可运行版本位于 rqalpha/examples/pair_trading.py，核心设计：

• 价差构造：spread = price_a - ratio * price_b，其中ratio = 15是对冲手数（手数配比），通过历史价格序列回归得到。
• 布林带信号：对 60 根 1 分钟价差序列计算标准差std与均值mean，构造上下轨mean ± 2*std。价差突破下轨（spread <= down_limit）视为价差被低估，「买入价差」（做多 A、做空 B）；价差突破上轨（spread >= up_limit）视为价差被高估，「卖出价差」（做空 A、做多 B）；价差回归均值时平仓。
• 状态锁存：up_cross_up_limit/down_cross_down_limit两个布尔变量记录当前是否已建仓，避免重复发单。
• 夜盘处理：由于样例商品期货在回测区间内有夜盘交易，before_trading中每天开盘前将计数器context.counter清零。
• 成交限制：代码注释强调，由于存在成交不超过下一 bar 成交量 25% 的限制，建仓可能需要通过多次发单成交才能成功，因此每次先计算剩余应建仓数量（如qty_a = 1 - long_pos_a.quantity），只对剩余部分下单，直到目标仓位全部建立（qty_a == 0 and qty_b == 0）才置位状态标记。

示例策略的测试与进一步阅读

文档中的策略并非孤立示例，仓库的集成测试对其中多个策略做了回测断言，可作为理解框架行为的参考：

• tests/integration_tests/test_backtest_results/test_s_buy_and_hold.py：买入持有策略测试，其config展示了 RQAlpha 回测配置的标准结构——base下的start_date、end_date、frequency（1d）、matching_type（current_bar）、benchmark（000300.XSHG）、accounts（stock: 1000000），以及extra.log_level；
• 另有 test_s_turtle.py、test_s_turtle_signal.py、test_s_dual_thrust.py、test_s_split.py 等海龟及其他策略的回测验证。

如果想进一步深入，建议继续阅读：

• docs/source/intro/tutorial.rst：10 分钟教程，讲解策略运行流程、约定函数、数据查询 API 与交易 API 的分类；
• docs/source/api/base_api.rst：基础 API 手册，覆盖all_instruments、instruments、history_bars、current_snapshot、get_trading_dates、is_suspended、is_st_stock等全部数据查询接口；
• docs/source/api/extend_api.rst：米筐金融、财务、合约历史数据等扩展数据接口；
• rqalpha/examples/：本文全部示例的可运行源码，以及 run_file_demo.py、run_code_demo.py、run_func_demo.py 等不同的策略加载方式演示；
• tests/integration_tests/test_backtest_results/：各策略的回测测试与输出样例（outs/目录下的.txt结果文件）。

从「买入并持有」到「期货跨品种配对交易」，这 7 个示例覆盖了 RQAlpha 最核心的能力面：约定函数驱动的事件循环、股票与期货的双账户交易 API、历史行情查询与行情订阅，以及 TA-Lib / numpy 等第三方库与策略的无缝集成。你可以直接复制本文代码保存为.py文件，配置好数据源后用rqalpha run跑通每一个策略，再逐步替换标的、调整参数，搭建出自己的策略框架。

• 金融科技