本文就示例代码讲解优矿中常用的API接口

优矿简介

• 一站式量化研究平台
  海量金融大数据+多资产策略研究+高性能回测与模拟交易+专业风险模型+高效优化器+丰富归因分析。
• 海量金融大数据
  提供各类资产的财务、因子、主题、宏观行业特色大数据，以及量化场景下的PIT数据，保障量化过程不引入未来数据。
• 多资产回测框架
  股票、期货、指数、场内外基金等多资产多策略回测。丰富的衍生工具，保证多因子策略、事件驱动等快速实现。
• 本地SDK
  支持云端数据与本地的交互。云端使用的大数据，及本地自定义数据，可自动双向对接，实现与本地现有环境的无缝集成。
• 量化因子库
  提供400+量化因子库，除了传统的投资因子等，还提供了特色Alpha因子，例如分析师评级、分析师盈利预测等。
• 风险模型
  接轨国际化风险模型算法，采用优质原始数据，提供10种风格因子、28种行业因子，全面揭示市场行业风险。

1.开始第一个策略回测

1.1 注册账号

登录优矿注册账号
注册完成之后，开始研究。

新建策略，自动生成策略回测模板

示例代码：

start = '2017-01-01' # 回测起始时间 end = '2018-01-01' # 回测结束时间 universe = DynamicUniverse('HS300') # 证券池，支持股票、基金、期货、指数四种资产 benchmark = 'HS300' # 策略参考标准 freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测 refresh_rate = 1 # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟 # 配置账户信息，支持多资产多账户 accounts = { 
    'fantasy_account': AccountConfig(account_type='security', capital_base=) } def initialize(context): pass # 每个单位时间(如果按天回测,则每天调用一次,如果按分钟,则每分钟调用一次)调用一次 def handle_data(context): previous_date = context.previous_date.strftime('%Y-%m-%d') # 获取因子PE的的历史数据集，截止到前一个交易日 hist = context.history(symbol=context.get_universe(exclude_halt=True), attribute='PE', time_range=1, style='tas')[previous_date] # 将因子值从小到大排序，并取前100支股票作为目标持仓 signal = hist['PE'].order(ascending=True) target_position = signal[:100].index # 获取当前账户信息 account = context.get_account('fantasy_account') current_position = account.get_positions(exclude_halt=True) # 卖出当前持有，但目标持仓没有的部分 for stock in set(current_position).difference(target_position): account.order_to(stock, 0) # 根据目标持仓权重，逐一委托下单 for stock in target_position: account.order(stock, 10000) 

1.2 回测参数

具体参数的api文档请关注官方，你可以永远相信它。

• start、end
  回测开始和结束时间，日线数据只能回测到2007年1月1日，分钟级别数据只能回测到2009年1月1日。值得注意的是，参数格式必须是“YYYY-MM-DD”和”YYYYMMDD“，并且在设置start和end的时候需要考虑到max_history_window引入的提前量，例如max_history_window=30时，start最早只能是2007-2-15。
• universe证券池
  策略回测的证券池，下单与历史数据获取都是限制于universe中的证券。支持全部的A股、全部可以在二级市场交易的ETF和LOF。
  有两种获取证券的方式，固定的资产列表和动态的证券池。

universe = ['000001,XSHF','.XHSG'] # 静态证券池 universe = DynamicUniverse('HS300') 

• apply_fitter
  将筛选条件作用于动态证券池。
• benchmark参考标准
  支持5个常用指数板块"SHCI,SH50,SH180,HS300,ZZ500"，支持个股"000001.XSHE"
• freq和refresh_rate策略运行频率
  • refresh_rate（日线）有三种取值：
    • 使用整数1表示每个交易日调仓策略
    • 使用Weekly(1)表示每周第一个交易日调仓
    • 使用Monthly(1,-1)表示每个月第一个和最后一个交易日调仓
  • refresh_rate（分钟线）有三种取值：
    • 使用(1,2)表示每个交易日每隔2分钟进行调仓
    • 使用(2,[‘10:30’,“14:30”])表示每2个交易日制定特定时间调仓
    • 使用(Weekly(1,-1),[‘10:30’,“14:30”])表示周第一个和最后一个交易日指定时间调仓
    • 使用(Monthly(1),120)表示每个月第一个交易日，间隔120分钟调仓

1.3 账户配置

具体参数的api文档请关注官方，你可以永远相信它。

优矿回测框架的交易账户配置函数，优矿最新框架支持多种交易品种，多个交易账户同时进行回测。

单个账户配置：

accounts = { 
    'security_account': AccountConfig(account_type='security', capital_base=, position_base = { 
   '.XSHG':1000}, cost_base = { 
   '.XSHG':10.05}, commission = Commission(buycost=0.001, sellcost=0.002, unit='perValue'), slippage = Slippage(value=0.0, unit='perValue')) } 

多个账户配置，Commission、Slippage 可以对多账户进行全局配置：

# 对两个账户进行全局的交易费用和滑点设置 commission = Commission(buycost=0.001, sellcost=0.002, unit='perValue') slippage = Slippage(value=0.0, unit='perValue') accounts = { 
    'security_account1': AccountConfig(account_type='security', capital_base=, position_base = { 
   '.XSHG':1000}, cost_base = { 
   '.XSHG':10.05}, commission = commission, slippage = slippage), 'security_account2': AccountConfig(account_type='security', capital_base=, position_base = { 
   '.XSHG':2000}, cost_base = { 
   '.XSHG':10.05}, commission = commission, slippage = slippage) } 

具体字段含义，请查看官方文档。

1.4 策略基本方法

1.4.1 initialize策略初始化函数

initialize是您实现策略初始化环境的函数，用于设置一些全局变量信息，该函数只在策略初始化时运行一次。

1.4.2 handle_data策略运行逻辑

您可以在handle_data中实现策略运行逻辑，策略框架会根据初始化配置的策略算法运行频率调用该函数进行策略调仓，handle_data是策略的核心部分，该函数每个调仓周期都会调用一次。

1.4.3 post_trading_day 盘后处理函数

您可以在post_trading_day中实现策略盘后处理逻辑（可选）。每天运行策略结束后，可以使用这个函数，做一些盘后操作，比如当日交易总结、预计算因子值等。

1.5 下单函数

• order下单指定股数

stock_account.order(symbol, amount) # 股票账户下单指定股数 futures_account.order(symbol, amount, 'open') # 期货账户委托开仓指定份数 otc_fund_account.order(symbol, amount) # 场外账户申购指定份数基金 index_account.order(symbol, amount, 'open') # 指数账户委托开仓指定份数 

可以对股票，期货，指数策略进行下单操作。
根据指定的参数，进行策略订单委托下单指定股数。
订单类型支持市价单或限价单，限价单需设置order_type为’limit’，并设置下单价格。

• order_to下单到指定股数

stock_account.order_to(symbol, amount) # 股票策略下单到指定股数 

• order_pct下单指定百分比

stock_account.order_pct(symbol, pct) # 股票策略下单指定百分比 

• order_pct_to:

• close_all_positions全部平仓

close_all_positions(symbol) 

卖出当前所有持仓，期货平仓时包含多头持仓和空头持仓。

• cancel_order撤单

start = '2017-01-01' # 回测起始时间 end = '2017-01-07' # 回测结束时间 benchmark = 'HS300' # 策略参考标准 universe = ['000001.XSHE'] # 证券池，支持股票和基金 capital_base =  # 起始资金 freq = 'm' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测 refresh_rate = (2, 1) # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟 def initialize(account): account.record_orders = [] def handle_data(account): if account.universe: if '13:14' == account.current_minute: del account.record_orders[:] for stk in account.universe: price = account.referencePrice[stk] * 0.99 order_id = order(stk, 500, price=price, otype='limit') account.record_orders.append(order_id) elif '13:15' == account.current_minute: for order_id in account.record_orders: _order = get_order(order_id) assert _order.state == OrderState.OPEN # assert _order.state=="ToFill" success = cancel_order(order_id) elif '13:16' == account.current_minute: # print account.blotter[0].state assert account.blotter[0].state == OrderState.CANCELED assert account.cash ==  

2. 多因子选股策略

量化选股策略主要分为基本面选股和市场行为选股两种，多因子模型是在基本面选股中常见的一种选股模型。

在不同市场环境下总有一些因子会发挥作用，多因子选股模型有两种方法，一种是打分法，一种是回归法。

• 打分法是指根据各个因子的大小对股票进行打分，按照一定的权重对个股加权得到一个总分，再根据总分筛选出分数较高的股票。
• 回归法是指按照股票历史数据的收益率对多因子进行回归，从而得到一个回归方程，然后将最新的因子值代入回归方程中得到一个对未来股票收益的预测，并以此为依据为未来收益选股进行预测。

建立多因子选股模型分为：选取候选因子、检验候选因子的有效性、剔除有效性但冗余的因子、构建综合评分模型、评价和改进综合评分模型等5个步骤。

2.1 单因子分析流程

这边给出了单因子：PE选股策略
回测区间：2014-01-01到2022-05-01，基准是沪深300，中证800成分股，策略每20天换仓一次。
因子选取：市盈率PE
因子处理：去极值，标准化处理
组合构建：整体分位数+风格中性权重

import pandas as pd import numpy as np start = '2014-01-01' # 回测起始时间 end = '2022-05-01' # 回测结束时间 universe = DynamicUniverse('HS300') # 证券池，支持股票、基金、期货、指数四种资产 benchmark = 'HS300' # 策略参考标准 freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测 refresh_rate = Monthly(1) # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟 # 配置账户信息，支持多资产多账户 accounts = { 
    'fantasy_account': AccountConfig(account_type='security', capital_base=) } def initialize(context): pass # 每个单位时间(如果按天回测,则每天调用一次,如果按分钟,则每分钟调用一次)调用一次 def handle_data(context): universe = context.get_universe() yesterday = context.previous_date.strftime("%Y-%m-%d") data = context.history(universe,["PE"],time_range=1,style='tas') data = data[yesterday] factor = data["PE"].dropna() # 因子分析 factor = pd.Series(winsorize(factor.to_dict())) factor = 1.0/factor factor = factor.replace([np.inf,-np.inf],0.0) signal = standardize(dict(factor))# 标准化 # 组合构建 wts = simple_long_only(signal,yesterday) # 交易部分 account = context.get_account('fantasy_account') current_pos = account.get_positions(exclude_halt=True) target_pos = wts.keys() # 卖出当前持有，单目标持仓没有的部分 for stock in set(current_pos).difference(target_pos): account.order_to(stock,0) # 根据目标权重，逐一下单 for stock in target_pos: account.order_pct_to(stock,wts[stock]) 

2.2 附加值分析

首先要找到回测效果满意的因子，然后根据已有模型选择合适的单因子进行多因子合成，查看回测整体效果是否提高。将市盈率、流通市值对数、等权合成后测试整体效果，代码如下：

import pandas as pd import numpy as np start = '2014-01-01' # 回测起始时间 end = '2017-05-01' # 回测结束时间 universe = DynamicUniverse('HS300')+DynamicUniverse('ZZ500') # 证券池，支持股票、基金、期货、指数四种资产 benchmark = 'HS300' # 策略参考标准 freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测 refresh_rate = Monthly(1) # 调仓频率，表示执行handle_data的时间间隔，若freq = 'd'时间间隔的单位为交易日，若freq = 'm'时间间隔为分钟 # 配置账户信息，支持多资产多账户 accounts = { 
    'fantasy_account': AccountConfig(account_type='security', capital_base=) } def initialize(context): # 初始化虚拟账户状态 context.signal_generator = SignalGenerator(Signal("PE"),Signal("LCAP")) # 每个单位时间(如果按天回测,则每天调用一次,如果按分钟,则每分钟调用一次)调用一次 def handle_data(context): universe = context.get_universe() yesterday = context.previous_date.strftime("%Y-%m-%d") data = context.history(universe,["PE","LCAP"],time_range=1,style='tas') data = data[yesterday] factor = data["PE"].dropna() # 因子分析 factor = pd.Series(winsorize(factor.to_dict())) factor = 1.0/factor # factor = factor.replace([np.inf,-np.inf],0.0) signal_pe = standardize(dict(factor))# 标准化 factor = data["LCAP"].dropna() # 因子分析 factor = pd.Series(winsorize(factor.to_dict())) factor = 1.0/factor # factor = factor.replace([np.inf,-np.inf],0.0) signal_lcap = standardize(dict(factor))# 标准化 # 信号合成 signal = (0.5*pd.Series(signal_pe).add(0.5*pd.Series(signal_lcap))) # 组合构建 wts = simple_long_only(signal,yesterday) # 交易部分 account = context.get_account('fantasy_account') current_pos = account.get_positions(exclude_halt=True) target_pos = wts.keys() # 卖出当前持有，单目标持仓没有的部分 for stock in set(current_pos).difference(target_pos): account.order_to(stock,0) # 根据目标权重，逐一下单 for stock in target_pos: account.order_pct_to(stock,wts[stock]) 

2.3 多因子（对冲）策略

多因子对冲策略具有如下特点：

• 影响价格因子且筛选广度大
• 各因子配分比选择可依策略需求进行调整，操作弹性大
• 搜集大量分析数据以提高策略可靠性

1.策略配置
回测区间：2017-1-1到2019-1-1
股票池：沪深300成分股
基准：沪深300
每周最后一个交易日换仓

2.因子数据处理
因子选取：净利润增长率、权益收益率、RSI
因子处理：去极值、标准化、中性化处理
组合构建：等权配置

3.空头使用期货IF进行对冲
对冲条件如下：
(1):用当月合约做空
(2)距离合约到期日3天，则移仓换月

策略如下：

from CAL.PyCAL import * import numpy as np from pandas import DataFrame start = '2017-01-01' # 回测起始时间 end = '2019-01-01'# 回测结束时间 universe = StockUniverse('HS300') + ['IFL0', 'IFL1'] # 证券池，支持股票和基金 benchmark = 'HS300' # 策略参考基准 freq = 'd' # 策略类型，'d'表示日间策略使用日线回测，'m'表示日内策略使用分钟线回测 refresh_rate = 1 # 读取每月末交易日信息 trade_dates = DataAPI.TradeCalGet(exchangeCD=u"XSHG", beginDate=start, endDate=end,field=['calendarDate', 'isOpen','isMonthEnd'], pandas="1") trade_dates = (trade_dates[(trade_dates.isOpen==1) & (trade_dates.isMonthEnd==1)]['calendarDate']).tolist() # 账户初始化配置 stock_commission = Commission(buycost=0.0005, sellcost=0.0005, unit='perValue') futures_commission = Commission(buycost=0.00005, sellcost=0.00005, unit='perValue') slippage = Slippage(value=0, unit='perValue') accounts = { 
    'stock_account': AccountConfig(account_type='security', capital_base=, commission=stock_commission, slippage=slippage), 'futures_account': AccountConfig(account_type='futures', capital_base=, commission=futures_commission, slippage=slippage) } # 策略算法 def initialize(context): context.need_to_switch_position = False context.contract_holding = '' def handle_data(context): universe = context.get_universe(exclude_halt=True) current_date = context.current_date.strftime('%Y-%m-%d') if current_date in trade_dates: yesterday = context.previous_date.strftime('%Y-%m-%d') data = context.history(universe, ['NetProfitGrowRate', 'ROE', 'RSI'], time_range=1, style='tas') data = data[yesterday] signal_composite = DataFrame() # 净利润增长率 NetProfitGrowRate = data['NetProfitGrowRate'].dropna() signal_NetProfitGrowRate = standardize(neutralize(winsorize(NetProfitGrowRate), yesterday)) signal_composite['NetProfitGrowRate'] = signal_NetProfitGrowRate # 权益收益率 ROE = data['ROE'].dropna() signal_ROE = standardize(neutralize(winsorize(ROE), yesterday)) signal_composite['ROE'] = signal_ROE # RSI RSI = data['RSI'].dropna() signal_RSI = standardize(winsorize(RSI)) signal_composite['RSI'] = signal_RSI #　信号合成，各因子权重 weight = np.array([0.3, 0.4, 0.3]) signal_composite['total_score'] = np.dot(signal_composite, weight) # 组合构建 total_score = signal_composite['total_score'].to_dict() wts = simple_long_only(total_score, yesterday) post_portfolio_value = handle_stock_orders(context, wts) handle_futures_orders(context, post_portfolio_value) # 移仓换月 switch_positions(context) # 订单委托 # 买入股票，返回股票多头总市值，方便空头开对冲 def handle_stock_orders(context, target_weights): account = context.get_account('stock_account') current_position = account.get_positions(exclude_halt=True) target_position = target_weights.keys() # 卖出当前持有，但目标持仓没有的部分 for stock in set(current_position).difference(target_position): account.order_to(stock, 0) post_portfolio_value = 0 for stock in target_position: weight = target_weights.get(stock) order_id = account.order_pct_to(stock, weight) order = account.get_order(order_id) # 将买入的订单进行汇总，估计多头总市值 if order and order.direction == 1: post_portfolio_value += context.current_price(stock) * order.order_amount return post_portfolio_value # 开空头对冲 def handle_futures_orders(context, stock_positions_value): stock_account = context.get_account('stock_account') future_account = context.get_account("futures_account") stock_position = stock_account.get_positions() # 获取股票仓位 contract_holding = context.contract_holding # 记录期货合约名称 # 有多头股票仓位，使用期货进行空头对冲 if stock_positions_value: futures_position = future_account.get_position(contract_holding) # 没有空头持仓，则建仓进行对冲 if not futures_position: contract_current_month = context.get_symbol('IFL0') multiplier = get_asset(contract_current_month).multiplier # 获取合约层数 futures_price = context.current_price(contract_current_month) total_hedging_amount = int(stock_positions_value / futures_price / multiplier) # 计算开仓数量 log.info(u'%s没有持仓，准备建仓。空头开仓%s手' % (contract_current_month, total_hedging_amount)) orders = future_account.order(contract_current_month, -1 * total_hedging_amount, "open") log.info(future_account.get_order(orders).order_time) context.contract_holding = contract_current_month # 已经有空头持仓，则判断是否需要调仓 else: contract_holding = context.contract_holding contract_current_month = context.get_symbol('IFL0') futures_price = context.current_price(contract_current_month) multiplier = get_asset(contract_holding).multiplier # 计算当前对冲需要的期货手数 total_hedging_amount = int(stock_positions_value / futures_price / multiplier) hedging_amount_diff = total_hedging_amount - futures_position.short_amount # 调仓阈值，可以适当放大，防止反复调仓 threshold = 2 if hedging_amount_diff >= threshold: log.info(u'空头调仓。[合约名:%s,当前空头手数:%s,目标空头手数:%s]' % (contract_holding, int(futures_position.short_amount), total_hedging_amount)) # 多开空仓 future_account.order(contract_holding, -1 * int(hedging_amount_diff), "open") elif hedging_amount_diff <= -threshold: log.info(u'空头调仓。[合约名:%s,当前空头手数:%s,目标空头手数:%s]' % (contract_holding, int(futures_position.short_amount), total_hedging_amount)) # 平掉部分空仓 future_account.order(contract_holding, int(abs(hedging_amount_diff)), "close") def switch_positions(context): stock_account = context.get_account('stock_account') future_account = context.get_account("futures_account") # 将主力连续合约映射为实际合约 contract_current_month = context.get_symbol('IFL0') # 判断是否需要移仓换月 contract_holding = context.contract_holding if not contract_holding: contract_holding = contract_current_month if contract_holding: last_trade_date = get_asset(contract_holding).last_trade_date # 当月合约离交割日只有3天 days_to_expire = (last_trade_date- context.current_date).days if days_to_expire < 3: log.info(u'距离%s到期，还有%s天' % (contract_holding, days_to_expire)) contract_next_month = context.get_symbol('IFL1') futures_position = future_account.get_position(contract_holding) if futures_position: current_holding = futures_position.short_amount log.info(u'移仓换月。[平仓旧合约:%s,开仓新合约:%s,手数:%s]' % (contract_holding, contract_next_month, int(current_holding))) if current_holding == 0: return future_account.order(contract_holding, current_holding, "close") future_account.order(contract_next_month, -1 * current_holding, "open") context.contract_holding = contract_next_month