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

金融科技【免费下载链接】rqalphaA extendable, replaceable Python algorithmic backtest trading framework supporting multiple securities项目地址https://gitcode.com/gh_mirrors/rq/rqalpha点击查看免费下载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 暴露的所有用户 APIlogger、order_percent、history_bars、get_position、subscribe等注入命名空间。API 的注册机制可以在 rqalpha/api.py 与 rqalpha/apis/api_abstract.py 中查看——每个 API 通过export_as_api装饰器导出并带有一系列参数校验与执行阶段约束装饰器。以第一个策略为例回测时需要明确以下要素可通过 config.yml 传参或命令行传参回测要素说明命令行参数数据源路径本地 bundle 数据路径-d/--data-bundle-path策略文件路径.py策略文件-f/--file回测起始时间如2016-06-01-s/--start-date回测结束时间如2016-12-01-e/--end-date起始资金按账户类型分配如股票账户 100000--account stock 100000Benchmark基准如沪深300指数000300.XSHG--benchmark 000300.XSHG假设策略存放在./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.pklimport 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.pyupdate_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 的SMAsimple 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.LONGPERIOD1, 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)本策略引入了三个重要 APIhistory_bars(order_book_id, bar_count, frequency, fields)获取指定合约的历史 k 线行情返回numpy.ndarray。依据 rqalpha/apis/api_base.py 的说明frequency支持1d、1m、1wfields支持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.LONGcur_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长短均线差值、signalmacd 的均线、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_PERIOD1, 1d, close) # 用Talib计算RSI值 rsi_data talib.RSI(prices, timeperiodcontext.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.XSHEhandle_bar中for stock in context.stocks循环处理。talib.RSI(prices, timeperiod14)[-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_time1, 1d, high) low_price_for_atr history_bars(context.s, context.open_observe_time1, 1d, low) low_price_for_extreme history_bars(context.s, context.close_observe_time1, 1d, low) close_price history_bars(context.s, context.open_observe_time2, 1d, close) close_price_for_atr close_price[:-1] atr_array talib.ATR(high_price, low_price_for_atr, close_price_for_atr, timeperiodcontext.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, timeperiod20)计算平均真实波幅unit floor(portfolio_value * 1% / atr)定义「1 单位」仓位——每单位风险控制在组合价值的 1% 内每 5 个交易日trade_day_num % 5 0更新一次。突破入场bar_dict[context.s].last maxx55 日最高价突破且无持仓时触发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_bartick 回测中则需订阅每一个关注的标的。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.pybuy_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、frequency1d、matching_typecurrent_bar、benchmark000300.XSHG、accountsstock: 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.rst10 分钟教程讲解策略运行流程、约定函数、数据查询 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跑通每一个策略再逐步替换标的、调整参数搭建出自己的策略框架。赞分享金融科技【免费下载链接】rqalphaA extendable, replaceable Python algorithmic backtest trading framework supporting multiple securities项目地址https://gitcode.com/gh_mirrors/rq/rqalpha点击查看免费下载相关推荐7个必学的pyalgotrade交易策略案例从入门到实战Python算法交易库pyalgotrade是量化交易领域的强大工具它提供了丰富的技术指标和策略框架让交易者能够快速构建和测试各种交易策略。无论你是量化交易金融科技5个RQAlpha实战策略示例快速提升你的量化交易技能5个RQAlpha实战策略示例快速提升你的量化交易技能 RQAlpha是一个可扩展、可替换的Python算法回测和交易框架支持多种证券类型。作为量化交易初学金融科技RQAlpha量化交易框架入门教程从策略编写到回测分析RQAlpha量化交易框架入门教程从策略编写到回测分析 前言 RQAlpha是一个功能强大的Python量化交易回测框架它提供了完整的策略开发、回测和风险分金融科技上一篇2025终极packer.nvim安全配置指南从插件管理到代码防护全流程下一篇Spot Micro高级定制指南自定义步态与性能优化创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考