记录学习CTP的OrderRef/OrderActionRef字段规则，原文参考https://zhuanlan.zhihu.com/p/89602892
OrderRef用来标识报单，OrderActionRef用来标识标撤单。
CTP量化投资API要求报单的OrderRef/OrderActionRef字段在同一线程内必须是递增的，长度不超过13的数字字符串。
如果包含非数字字符，或者非递增关系，都会触发以下的错误：
{'ErrorID': 22, 'ErrorMsg': 'CTP:报单错误：不允许重复报单'}
如果为每个策略开启一个线程，即一个TraderApi管理一个策略，则指定策略ID，为每个策略分配一个报单区间，据此在回报/通知中分辨出对应策略。
如果一个TraderApi作为主引擎管理多个策略，则需要在引擎层面管理OrderRef/OrderActionRef，然后维护策略与OrderRef/OrderActionRef的关系表，或者制定一个可标识策略的递增规则，据此在回报/通知中分辨出对应策略。
记录学习CTP出现4097、8193报错，原文参考https://zhuanlan.zhihu.com/p/89263750
在客户端程序与期货公司行情、交易前置建立连接之后，服务器会定时发送心跳包确认连接是否正常。当网络连接出现异常，客户端可以在回调函数OnFrontDisconnected中收到通知：
///当客户端与交易后台通信连接断开时，该方法被调用。当发生这个情况后，API会自动重新连接，客户端可不做处理。
///@param nReason 错误原因
/// 0x1001 网络读失败 4097
/// 0x1002 网络写失败 4098
/// 0x2001 接收心跳超时 8193
/// 0x2002 发送心跳失败 8194
/// 0x2003 收到错误报文 8195
void OnFrontDisconnected(int nReason) {};
交易者最常遇到的就是4097（十六进制：0x1001）和8193（十六进制：0x2001）报错。
如果网络异常，很容易发现问题。
但是初次使用CTP的交易者会发现，在网络正常情况下，也会报这个错误。究竟是哪里出了问题呢？
其实，这是不熟悉CTP异步执行特性导致的。
CTP所有的方法都是异步执行的，也就是说，调用一个方法返回时，该方法并没有执行完成，而是刚开始执行。如果主线程没有等待子线程执行完成就结束了，会触发OnFrontDisconnected错误。
from AlgoPlus.CTP.MdApi import MdApi
class TickEngine(MdApi):
#5-8行
# def __init__(self, md_server, broker_id, investor_id, password, app_id, auth_code
# , instrument_id_list, md_queue_list=None
# , page_dir='', using_udp=False, multicast=False):
# self.Join()
# ///深度行情通知
def OnRtnDepthMarketData(self, pDepthMarketData):
print(pDepthMarketData)
if __name__ == '__main__':
import sys
sys.path.append("..")
from account_info import my_future_account_info_dict
future_account = my_future_account_info_dict['SimNow']
tick_engine = TickEngine(future_account.server_dict['MDServer']
, future_account.broker_id
, future_account.investor_id
, future_account.password
, future_account.app_id
, future_account.auth_code
, future_account.instrument_id_list
, None
, future_account.md_page_dir)
# 30行
# tick_engine.Join()
解决办法很简单，就是在主线程结束之前调用Join方法，等待子线程执行。
将以上5-8行代码的注释取消，或者在30行之后调用tick_engine的Join方法，就正常了。
网络异常情况下的解决方案
无需做其他处理，等待CTP自动重连，重连成功后自动登录账户，然后就可以正常使用了。
MdApi同样会自动重连，且重连成功后自动登录账户，但是需要重新订阅行情，否则不会收到断开前订阅的行情数据。
AlgoPlus已封装了MdApi重连成功时订阅合约的功能。
记录学习用AlgoPlus构建自己的交易盈亏风控系统，原文参考https://zhuanlan.zhihu.com/p/88655638
常用的止盈止损方案有：固定止损、固定止盈、跟踪止损、阶梯止损、保本、时间止损。
# {"InstrumentID": 0}
self.order_action_num_dict = {}
撤单次数是一个以合约名为键值的字典。收到撤单通知后，对撤单次数计算增加。
收到撤单通知时增加撤单次数计数
def OnRtnOrder(self, pOrder):
"""
当收到订单状态变化时，可以在本方法中获得通知。不适宜在回调函数里做比较耗时的操作。
:param pOrder: AlgoPlus.CTP.ApiStruct中OrderField的实例。
:return:
"""
if pOrder.OrderStatus == b"5":
if pOrder.InstrumentID in self.action_num_dict.keys():
self.action_num_dict[pOrder.InstrumentID] += 1
else:
self.action_num_dict[pOrder.InstrumentID] = 1
`pl_parameter = {
'StrategyID': 9,
# 盈损参数，'0'代表止盈, '1'代表止损，绝对价差
'ProfitLossParameter': {
b'rb2010': {'0': [2], '1': [2]},
b'ni2007': {'0': [20], '1': [20]},
},
}`
# {"InstrumentID": {"Type": []}}
self.pl_parameter_dict = {}
止损参数是一个以合约名为键值的字典，根据不同的止损止盈类型存储价差/时间差等参数。其中，止损类型参数取值：
Type止损逻辑b”0″固定止盈b”1″固定止损
价差/时间差等参数以列表形式存储，有些止损逻辑可能需要多个参数。
# {"InstrumentID": {"LongVolume": 0, "LongPositionList": [], "ShortVolume": 0, "ShortPositionList": []}}
self.local_position_dict = {}
账户持仓以合约名为键值存入字典中，LongVolume统计合约总多头持仓，ShortVolume统计合约总空头持仓，LongPositionList、ShortPositionList以OrderRef为单位存储成交明细。
成交明细是在AlgoPlus.CTP.ApiStruct中TradeField基础上附加IsLock、AnchorTime、StopProfitDict、StopLossDict、MaxProfitPrice字段。
rtn_trade["IsLock"] = False # 平仓状态
rtn_trade["AnchorTime"] = timer() # 成交发生时间
rtn_trade["StopProfitDict"] = {} # 止盈触发价格，持仓期间实时更新
rtn_trade["StopLossDict"] = {} # 止损触发价格，持仓期间实时更新
成交通知
收到成交通知时放入一个列表中，等待后续处理，避免在此设计复杂的耗时操作。
def OnRtnTrade(self, pTrade):
"""
当报单成交时，可以在本方法中获得通知。不适宜在回调函数里做比较耗时的操作。
:param pTrade: AlgoPlus.CTP.ApiStruct中的TradeField实例。
:return:
"""
self.local_rtn_trade_list.append(pTrade.to_dict_raw())
根据开买卖开平字段将成交成交信息放入持仓数据结构中。
def process_rtn_trade(self):
"""
从上次订单ID位置开始处理订单数据。
:return:
"""
last_rtn_trade_id = len(self.local_rtn_trade_list)
for rtn_trade in self.local_rtn_trade_list[self.last_rtn_trade_id:last_rtn_trade_id]:
if rtn_trade["InstrumentID"] not in self.instrument_id_registered:
self.instrument_id_registered.append(rtn_trade["InstrumentID"])
rtn_trade["IsLock"] = False
rtn_trade["AnchorTime"] = timer()
rtn_trade["StopProfitDict"] = {}
rtn_trade["StopLossDict"] = {}
if rtn_trade["InstrumentID"] not in self.local_position_dict.keys():
self.local_position_dict[rtn_trade["InstrumentID"]] = {"LongVolume": 0, "LongPositionList": [], "ShortVolume": 0, "ShortPositionList": []}
local_position_info = self.local_position_dict[rtn_trade["InstrumentID"]]
# 开仓
if rtn_trade["OffsetFlag"] == b'0':
self.update_stop_price(rtn_trade)
if rtn_trade["Direction"] == b'0':
local_position_info["LongVolume"] += rtn_trade["Volume"]
local_position_info["LongPositionList"].append(rtn_trade)
elif rtn_trade["Direction"] == b'1':
local_position_info["ShortVolume"] += rtn_trade["Volume"]
local_position_info["ShortPositionList"].append(rtn_trade)
elif rtn_trade["Direction"] == b'0':
local_position_info["ShortVolume"] = max(local_position_info["ShortVolume"] - rtn_trade["Volume"], 0)
elif rtn_trade["Direction"] == b'1':
local_position_info["LongVolume"] = max(local_position_info["LongVolume"] - rtn_trade["Volume"], 0)
self.last_rtn_trade_id = last_rtn_trade_id
遍历持仓数据结构中的所有成交明细，判断最新行情是否触发某个止盈止损阈值，如果触发则录入平仓报单。
def check_position(self):
"""
检查所有持仓是否触发持仓阈值。
"""
try:
for instrument_id, position_info in self.local_position_dict.items():
for long_position in position_info["LongPositionList"]:
if not long_position["IsLock"]:
trigger = False
order_price = None
for stop_profit in long_position["StopProfitDict"].values():
if self.md_dict[instrument_id]["LastPrice"] > stop_profit:
trigger = True
order_price = self.get_stop_profit_price(instrument_id, long_position["Direction"])
break
if not trigger:
for stop_loss in long_position["StopLossDict"].values():
if self.md_dict[instrument_id]["LastPrice"] < stop_loss:
trigger = True
order_price = self.get_stop_loss_price(instrument_id, long_position["Direction"])
break
if trigger and order_price:
self.order_ref += 1
self.sell_close(long_position["ExchangeID"], instrument_id, order_price, long_position["Volume"], self.order_ref)
long_position["IsLock"] = True
for short_position in position_info["ShortPositionList"]:
if not short_position["IsLock"]:
trigger = False
order_price = None
for stop_profit in short_position["StopProfitDict"].values():
if self.md_dict[instrument_id]["LastPrice"] < stop_profit:
trigger = True
order_price = self.get_stop_profit_price(instrument_id, short_position["Direction"])
break
if not trigger:
for stop_loss in short_position["StopLossDict"].values():
if self.md_dict[instrument_id]["LastPrice"] > stop_loss:
trigger = True
order_price = self.get_stop_loss_price(instrument_id, short_position["Direction"])
break
if trigger and order_price:
self.order_ref += 1
self.buy_close(short_position["ExchangeID"], instrument_id, order_price, short_position["Volume"], self.order_ref)
short_position["IsLock"] = True
except Exception as err:
self._write_log(err)
根据止盈止损（固定止盈、固定止损、跟踪止损、阶梯止损、保本止损）逻辑计算出触发价格，存入成交明细字典中。这里给出了固定止盈和固定止损的例子
def update_stop_price(self, position_info):
"""
获取止盈止损阈值。止损类型参考https://7jia.com/1002.html
:param position_info: 持仓信息
:return:
"""
for instrument_id, pl_dict in self.pl_parameter_dict.items():
if isinstance(pl_dict, dict):
for pl_type, delta in pl_dict.items():
# 固定止盈
sgn = 1 if position_info["Direction"] == b'0' else -1
if pl_type == b"0":
position_info["StopProfitDict"][b"0"] = position_info["Price"] + delta[0] * sgn
# 固定止损
elif pl_type == b"1":
position_info["StopLossDict"][b"1"] = position_info["Price"] - delta[0] * sgn
在多线程multiprocessing模块中，有两个类，Queue（队列）和Process（进程）
队列Queue：
Queue是python中的标准库，可以直接import引用在队列中;
Queue.Queue(maxsize)创建队列对象，如果不提供maxsize,则队列数无限制。
# _*_ encoding:utf-8 _*_
import Queue
q = Queue.Queue(10)
q.put('LOVE')
q.put('You')
print (q.get())
print (q.get())
当一个队列为空的时候,用get取回堵塞,所以一般取队列的时候会用,get_nowait()方法,这个方法在向一个空队列取值的时候会抛一个Empty异常,所以一般会先判断队列是否为空,如果不为空则取值;
不阻塞的方式取队列
判断队列是否为空，为空返回True，不为空返回False
返回队列的长度
Queue.get([block[, timeout]]) 获取队列，timeout等待时间
Queue.get_nowait() 相当Queue.get(False)
非阻塞 Queue.put(item) 写入队列，timeout等待时间
Queue.put_nowait(item) 相当Queue.put(item, False)
Multiprocessing中使用子进程的概念Process：
from multiprocessing import Process
可以通过Process来构造一个子进程
p=Process(target=fun,args=(args))
再通过p.start()来启动子进程
再通过p.join()方法来使得子进程运行结束后再执行父进程
在multiprocessing中使用pool：
如果需要多个子进程时可以考虑使用进程池（pool）来管理
Pool创建子进程的方法与Process不同,是通过p.apply_async(func,args=(args))实现,一个池子里能同时运行的任务是取决你电脑CPU的数量,如果是4个CPU,那么会有task0,task1,task2,task3同时启动,task4需要在某个进程结束后才开始。
多个子进程间的通信:
多个子进程间的通信就要采用第一步中的队列Queue,比如,有以下需求,一个子进程向队列中写数据,另一个进程从队列中取数据;
# _*_ encoding:utf-8 _*_
from multiprocessing import Process,Queue,Pool,Pipe
import os,time,random
#写数据进程执行的代码：
def write(p):
for value in ['A','B','C']:
print ('Write---Before Put value---Put %s to queue...' % value)
p.put(value)
print ('Write---After Put value')
time.sleep(random.random())
print ('Write---After sleep')
#读数据进程执行的代码：
def read(p):
while True:
print ('Read---Before get value')
value = p.get(True)
print ('Read---After get value---Get %s from queue.' % value)
if __name__ == '__main__':
#父进程创建Queue，并传给各个子进程：
p = Queue()
pw = Process(target=write,args=(p,))
pr = Process(target=read,args=(p,))
#启动子进程pw，写入：
pw.start()
#启动子进程pr，读取:
pr.start()
#等待pw结束：
pw.join()
#pr进程里是死循环，无法等待其结束，只能强行终止：
pr.terminate()
买卖报单、撤单、查询概括了所有的交易业务。这些业务都是由交易者主动发起的，并且提供必要的信息。交易者只需要传递相应参数，AlgoPlus就可以按照CTP标准组织信息并发起请求。
b"ExchangeID": b"SHFE", b"Direction": 1, b"Volume": 1
exchange_id：b"SHFE"
instrument_id：b"rb2001"
order_vol：1
买开仓
buy_open(exchange_id, instrument_id, order_price, order_vol)
卖平仓
sell_close(exchange_id, instrument_id, order_price, order_vol, is_today)
卖开仓
sell_open(exchange_id, instrument_id, order_price, order_vol)
买平仓
buy_close(exchange_id, instrument_id, order_price, order_vol, is_today)
撤单
req_order_action(exchange_id, instrument_id, order_ref, order_sysid)
查成交
req_qry_trade()
查持仓
req_qry_investor_position()
查资金账户
req_qry_trading_account()
查合约
req_qry_instrument()
交易者发起的买卖报单、撤单请求后，期货公司柜台会推送一条响应信息，AlgoPlus的回调函数以此作为参数被调用。
OnRspOrderInsert(pInputOrder, pRspInfo, nRequestID, bIsLast)
OnRspOrderAction(pInputOrderAction, pRspInfo, nRequestID, bIsLast)
pRspInfo是一个python字典，内容如下
{
'ErrorID': 0, # 错误代码
'ErrorMsg': "", # 错误信息
}
买卖报单、撤单请求到达交易所被执行的过程中，交易所实时推送订单状态变化信息，AlgoPlus的回调函数OnRtnOrder(pOrder)以此作为参数被调用。pOrder是一个python字典，内容如下
{
'BrokerID': "", # 经纪公司代码
'InvestorID': "", # 投资者代码
'InstrumentID': "", # 合约代码
'OrderRef': "", # 报单引用
'UserID': "", # 用户代码
'OrderPriceType': "", # 报单价格条件
'Direction': "", # 买卖方向
'CombOffsetFlag': "", # 组合开平标志
'CombHedgeFlag': "", # 组合投机套保标志
'LimitPrice': 0.0, # 价格
'VolumeTotalOriginal': 0, # 数量
'TimeCondition': "", # 有效期类型
'GTDDate': "", # GTD日期
'VolumeCondition': "", # 成交量类型
'MinVolume': 0, # 最小成交量
'ContingentCondition': "", # 触发条件
'StopPrice': 0.0, # 止损价
'ForceCloseReason': "", # 强平原因
'IsAutoSuspend': 0, # 自动挂起标志
'BusinessUnit': "", # 业务单元
'RequestID': 0, # 请求编号
'OrderLocalID': "", # 本地报单编号
'ExchangeID': "", # 交易所代码
'ParticipantID': "", # 会员代码
'ClientID': "", # 客户代码
'ExchangeInstID': "", # 合约在交易所的代码
'TraderID': "", # 交易所交易员代码
'InstallID': 0, # 安装编号
'OrderSubmitStatus': "", # 报单提交状态
'NotifySequence': 0, # 报单提示序号
'TradingDay': "", # 交易日
'SettlementID': 0, # 结算编号
'OrderSysID': "", # 报单编号
'OrderSource': "", # 报单来源
'OrderStatus': "", # 报单状态
'OrderType': "", # 报单类型
'VolumeTraded': 0, # 今成交数量
'VolumeTotal': 0, # 剩余数量
'InsertDate': "", # 报单日期
'InsertTime': "", # 委托时间
'ActiveTime': "", # 激活时间
'SuspendTime': "", # 挂起时间
'UpdateTime': "", # 最后修改时间
'CancelTime': "", # 撤销时间
'ActiveTraderID': "", # 最后修改交易所交易员代码
'ClearingPartID': "", # 结算会员编号
'SequenceNo': 0, # 序号
'FrontID': 0, # 前置编号
'SessionID': 0, # 会话编号
'UserProductInfo': "", # 用户端产品信息
'StatusMsg': "", # 状态信息
'UserForceClose': 0, # 用户强平标志
'ActiveUserID': "", # 操作用户代码
'BrokerOrderSeq': 0, # 经纪公司报单编号
'RelativeOrderSysID': "", # 相关报单
'ZCETotalTradedVolume': 0, # 郑商所成交数量
'IsSwapOrder': 0, # 互换单标志
'BranchID': "", # 营业部编号
'InvestUnitID': "", # 投资单元代码
'AccountID': "", # 资金账号
'CurrencyID': "", # 币种代码
'IPAddress': "", # IP地址
'MacAddress': "", # Mac地址
}
OrderStatus取值及含义
全部成交
OrderStatus_AllTraded = b'0'
#部分成交还在队列中
OrderStatus_PartTradedQueueing = b'1'
#部分成交不在队列中
OrderStatus_PartTradedNotQueueing = b'2'
#未成交还在队列中
OrderStatus_NoTradeQueueing = b'3'
#未成交不在队列中
OrderStatus_NoTradeNotQueueing = b'4'
#撤单
OrderStatus_Canceled = b'5'
#未知
OrderStatus_Unknown = b'a'
#尚未触发
OrderStatus_NotTouched = b'b'
#已触发
OrderStatus_Touched = b'c'
OrderSubmitStatus取值及含义
#已经提交
OrderSubmitStatus_InsertSubmitted = b'0'
#撤单已经提交
OrderSubmitStatus_CancelSubmitted = b'1'
#修改已经提交
OrderSubmitStatus_ModifySubmitted = b'2'
#已经接受
OrderSubmitStatus_Accepted = b'3'
#报单已经被拒绝
OrderSubmitStatus_InsertRejected = b'4'
#撤单已经被拒绝
OrderSubmitStatus_CancelRejected = b'5'
#改单已经被拒绝
OrderSubmitStatus_ModifyRejected = b'6'
当订单有成交发生时，交易所还会推送一条成交信息，AlgoPlus的回调函数OnRtnTrade(pTrade)以此作为参数被调用。除了成交价格之外，pTrade中的其他信息在pOrder中都有。
行情数据通知
创建行情接口实例时，AlgoPlus会根据交易者传递的合约列表参数自动订阅合约。在盘中，AlgoPlus的回调函数OnRtnDepthMarketData(pDepthMarketData)以实时行情数据为参数被调用。pDepthMarketData是一个python字典，内容如下：
{
'TradingDay': b'20200113', # 交易日
'InstrumentID': b'rb2005', # 合约代码
'ExchangeID': b'', # 交易所代码
'ExchangeInstID': b'', # 合约在交易所的代码
'LastPrice': 3559.0, # 最新价
'PreSettlementPrice': 3568.0, # 上次结算价
'PreClosePrice': 3571.0, # 昨收盘
'PreOpenInterest': 1357418.0, # 昨持仓量
'OpenPrice': 3565.0, # 今开盘
'HighestPrice': 3567.0, # 最高价
'LowestPrice': 3544.0, # 最低价
'Volume': 347796, # 数量
'Turnover': 12361542250.0, # 成交金额
'OpenInterest': 1345077.0, # 持仓量
'ClosePrice': 1.7976931348623157e+308, # 今收盘
'SettlementPrice': 1.7976931348623157e+308, # 本次结算价
'UpperLimitPrice': 3782.0, # 涨停板价
'LowerLimitPrice': 3353.0, # 跌停板价
'PreDelta': 0.0, # 昨虚实度
'CurrDelta': 1.7976931348623157e+308, # 今虚实度
'UpdateTime': b'23:00:01', # 最后修改时间
'UpdateMillisec': 0, # 最后修改毫秒
'BidPrice1': 3559.0, # 申买价一
'BidVolume1': 158, # 申买量一
'AskPrice1': 3560.0, # 申卖价一
'AskVolume1': 18, # 申卖量一
'BidPrice2': 1.7976931348623157e+308, # 申买价二
'BidVolume2': 0, # 申买量二
'AskPrice2': 1.7976931348623157e+308, # 申卖价二
'AskVolume2': 0, # 申卖量二
'BidPrice3': 1.7976931348623157e+308, # 申买价三
'BidVolume3': 0, # 申买量三
'AskPrice3': 1.7976931348623157e+308, # 申卖价三
'AskVolume3': 0, # 申卖量三
'BidPrice4': 1.7976931348623157e+308, # 申买价四
'BidVolume4': 0, # 申买量四
'AskPrice4': 1.7976931348623157e+308, # 申卖价四
'AskVolume4': 0, # 申卖量四
'BidPrice5': 1.7976931348623157e+308, # 申买价五
'BidVolume5': 0, # 申买量五
'AskPrice5': 1.7976931348623157e+308, # 申卖价五
'AskVolume5': 0, # 申卖量五
'AveragePrice': 35542.50839572623, # 当日均价
'ActionDay': b'20200110' # 业务日期
}
记录学习使用AlgoPlus接收期货实时行情，原文参考https://zhuanlan.zhihu.com/p/86082225
关于CTP
CTP是Comprehensive Transaction Platform的简称。CTP有MdApi和TraderApi两个独立的开放接口。
MdApi负责行情相关操作（订阅、接收）。
TraderApi负责交易相关的操作（买、卖、撤、查）。
MdApi与TraderApi方法的执行过程都是异步的，每一个请求都对应一个或多个负责接收执行结果的回调函数。例如，通过ReqOrderInsert方法向交易所发出买开仓指令，对应的回调方法OnRtnOrder可以实时接收交易所服务器发回来的执行通知。
MdApi是行情接口，使用时只需要传递账户参数创建一个实例就可以了。示例：
from AlgoPlus.CTP.MdApi import MdApi
class TickEngine(MdApi):
# 深度行情通知
def OnRtnDepthMarketData(self, pDepthMarketData):
print(pDepthMarketData)
# print(f"{pDepthMarketData.InstrumentID}当前最新价：{pDepthMarketData.LastPrice}")
if __name__ == '__main__':
from account_info import my_future_account_info_dict
future_account = my_future_account_info_dict['SimNow']
tick_engine = TickEngine(future_account.server_dict['MDServer']
, future_account.broker_id
, future_account.investor_id
, future_account.password
, future_account.app_id
, future_account.auth_code
, future_account.instrument_id_list
, None
, future_account.md_page_dir)
tick_engine.Join()
1、从AlgoPlus.CTP.MdApi文件中导入MdApi类。MdApi已对工作流程的前六步进行了封装。
2、TickEngine是MdApi的子类。TickEngine类主要实现收到行情的数据处理算法，示例只将收到的行情打印出来。
3、创建行情接口实例前，需要导入账户信息。示例的账户信息存放在同一个目录下的account_info.py文件中。
4、交易时间运行以上代码就可以将接收到的实时期货行情打印出来。
5、回调函数OnRtnDepthMarketData接收到的pDepthMarketData行情是DepthMarketDataField结构体的实例，在AlgoPlus.CTP.ApiStruct中被定义。以调用属性的方式可以获取行情任意字段的数值，例如pDepthMarketData.LastPrice表示最新价。DepthMarketDataField包括以下字段：
class DepthMarketDataField(BaseField):
"""深度行情"""
_fields_ = [
('TradingDay', c_char * 9) # ///交易日
, ('InstrumentID', c_char * 31) # 合约代码
, ('ExchangeID', c_char * 9) # 交易所代码
, ('ExchangeInstID', c_char * 31) # 合约在交易所的代码
, ('LastPrice', c_double) # 最新价
, ('PreSettlementPrice', c_double) # 上次结算价
, ('PreClosePrice', c_double) # 昨收盘
, ('PreOpenInterest', c_double) # 昨持仓量
, ('OpenPrice', c_double) # 今开盘
, ('HighestPrice', c_double) # 最高价
, ('LowestPrice', c_double) # 最低价
, ('Volume', c_int) # 数量
, ('Turnover', c_double) # 成交金额
, ('OpenInterest', c_double) # 持仓量
, ('ClosePrice', c_double) # 今收盘
, ('SettlementPrice', c_double) # 本次结算价
, ('UpperLimitPrice', c_double) # 涨停板价
, ('LowerLimitPrice', c_double) # 跌停板价
, ('PreDelta', c_double) # 昨虚实度
, ('CurrDelta', c_double) # 今虚实度
, ('UpdateTime', c_char * 9) # 最后修改时间
, ('UpdateMillisec', c_int) # 最后修改毫秒
, ('BidPrice1', c_double) # 申买价一
, ('BidVolume1', c_int) # 申买量一
, ('AskPrice1', c_double) # 申卖价一
, ('AskVolume1', c_int) # 申卖量一
, ('BidPrice2', c_double) # 申买价二
, ('BidVolume2', c_int) # 申买量二
, ('AskPrice2', c_double) # 申卖价二
, ('AskVolume2', c_int) # 申卖量二
, ('BidPrice3', c_double) # 申买价三
, ('BidVolume3', c_int) # 申买量三
, ('AskPrice3', c_double) # 申卖价三
, ('AskVolume3', c_int) # 申卖量三
, ('BidPrice4', c_double) # 申买价四
, ('BidVolume4', c_int) # 申买量四
, ('AskPrice4', c_double) # 申卖价四
, ('AskVolume4', c_int) # 申卖量四
, ('BidPrice5', c_double) # 申买价五
, ('BidVolume5', c_int) # 申买量五
, ('AskPrice5', c_double) # 申卖价五
, ('AskVolume5', c_int) # 申卖量五
, ('AveragePrice', c_double) # 当日均价
, ('ActionDay', c_char * 9) # 业务日期
]
说明：
1、队列是实现行情进程与策略进程之间共享数据的最简单有效的方案，也是AlgoPlus默认使用的方案。
2、每个策略对应一个队列，将这些队列的列表赋值给参数md_queue_list。
3、在OnRtnDepthMarketData中，将收到的行情放入所有队列。
import time
from datetime import datetime, timedelta
from multiprocessing import Process, Queue
from AlgoPlus.CTP.TraderApi import TraderApi
from AlgoPlus.CTP.ApiStruct import *
from tick_engine import TickEngine
class TraderEngine(TraderApi):
def __init__(self, td_server, broker_id, investor_id, password, app_id, auth_code, md_queue=None
, page_dir='', private_resume_type=2, public_resume_type=2):
self.order_ref = 0 # 报单引用
self.order_time = None # 报单时间
self.order_status = b"" # 订单状态
self.Join()
# 撤单
def req_order_action(self, exchange_id, instrument_id, order_ref, order_sysid=''):
input_order_action_field = InputOrderActionField(
BrokerID=self.broker_id,
InvestorID=self.investor_id,
UserID=self.investor_id,
ExchangeID=exchange_id,
ActionFlag="0",
InstrumentID=instrument_id,
FrontID=self.front_id,
SessionID=self.session_id,
OrderSysID=order_sysid,
OrderRef=str(order_ref),
)
l_retVal = self.ReqOrderAction(input_order_action_field)
# 报单
def req_order_insert(self, exchange_id, instrument_id, order_price, order_vol, order_ref, direction, offset_flag):
input_order_field = InputOrderField(
BrokerID=self.broker_id,
InvestorID=self.investor_id,
ExchangeID=exchange_id,
InstrumentID=instrument_id,
UserID=self.investor_id,
OrderPriceType="2",
Direction=direction,
CombOffsetFlag=offset_flag,
CombHedgeFlag="1",
LimitPrice=order_price,
VolumeTotalOriginal=order_vol,
TimeCondition="3",
VolumeCondition="1",
MinVolume=1,
ContingentCondition="1",
StopPrice=0,
ForceCloseReason="0",
IsAutoSuspend=0,
OrderRef=str(order_ref),
)
l_retVal = self.ReqOrderInsert(input_order_field)
# 买开仓
def buy_open(self, exchange_ID, instrument_id, order_price, order_vol, order_ref):
self.req_order_insert(exchange_ID, instrument_id, order_price, order_vol, order_ref, '0', '0')
# 卖开仓
def sell_open(self, exchange_ID, instrument_id, order_price, order_vol, order_ref):
self.req_order_insert(exchange_ID, instrument_id, order_price, order_vol, order_ref, '1', '0')
# 买平仓
def buy_close(self, exchange_ID, instrument_id, order_price, order_vol, order_ref):
if exchange_ID == "SHFE" or exchange_ID == "INE":
self.req_order_insert(exchange_ID, instrument_id, order_price, order_vol, order_ref, '0', '3')
else:
self.req_order_insert(exchange_ID, instrument_id, order_price, order_vol, order_ref, '0', '1')
# 卖平仓
def sell_close(self, exchange_ID, instrument_id, order_price, order_vol, order_ref):
if exchange_ID == "SHFE" or exchange_ID == "INE":
self.req_order_insert(exchange_ID, instrument_id, order_price, order_vol, order_ref, '1', '3')
else:
self.req_order_insert(exchange_ID, instrument_id, order_price, order_vol, order_ref, '1', '1')
# 报单通知
def OnRtnOrder(self, pOrder):
self.order_status = pOrder.OrderStatus
if pOrder.OrderStatus == b"a":
status_msg = "未知状态！"
elif pOrder.OrderStatus == b"0":
if pOrder.Direction == b"0":
if pOrder.CombOffsetFlag == b"0":
status_msg = "买开仓已全部成交！"
else:
status_msg = "买平仓已全部成交！"
else:
if pOrder.CombOffsetFlag == b"0":
status_msg = "卖开仓已全部成交！"
else:
status_msg = "卖平仓已全部成交！"
elif pOrder.OrderStatus == b"1":
status_msg = "部分成交！"
elif pOrder.OrderStatus == b"3":
status_msg = "未成交！"
elif pOrder.OrderStatus == b"5":
status_msg = "已撤！"
else:
status_msg = "其他！"
self._write_log(f"{status_msg}=>{pOrder}")
def Join(self):
while True:
if self.status == 0:
last_md = None
# 如果队列非空，从队列中取数据
while not self.md_queue.empty():
last_md = self.md_queue.get(block=False)
if last_md:
# ############################################################################# #
if self.order_ref == 0:
# 涨停买开仓
self.order_ref += 1
self.buy_open(test_exchange_id, test_instrument_id, last_md.BidPrice1, test_vol, self.order_ref)
self.order_time = datetime.now()
self._write_log(f"=>买开仓请求！")
if self.order_ref == 1 and self.order_status == b"3" and datetime.now() - self.order_time > timedelta(seconds=3):
self.order_status = b""
self.req_order_action(test_exchange_id, test_instrument_id, self.order_ref)
self._write_log(f"=>发出撤单请求！")
if self.order_ref == 1 and self.order_status == b"0":
self.order_ref += 1
self.order_status = b""
self.sell_close(test_exchange_id, test_instrument_id, last_md.BidPrice1, test_vol, self.order_ref)
self.order_time = datetime.now()
self._write_log(f"=>买开仓已全部成交，发出卖平仓请求！")
# ############################################################################# #
if self.order_ref == 1:
if self.order_status == b"5":
print("老爷，买开仓单超过3秒未成交，已撤销，这里的测试工作已经按照您的吩咐全部完成！")
break
elif datetime.now() - self.order_time > timedelta(seconds=3):
print("买开仓执行等待中！")
elif self.order_ref == 2:
if self.order_status == b"0":
print("老爷，卖平仓单已成交，这里的测试工作已经按照您的吩咐全部完成！")
break
elif datetime.now() - self.order_time > timedelta(seconds=3):
print("卖平仓执行等待中！")
else:
time.sleep(1)
说明：
1、直接在TraderApi的子类中编写策略是最简单的方案。但是，该方案不适合单账户策略比较多的情况，因为CTP支持同时在线的终端个数有限。如果策略比较多，则创建有限个TraderApi，在独立的Strategy类与MdApi和TraderApi之间实现共享数据。
2、在Join方法中实现了策略逻辑：登录成功之后，先以排队价发开仓委托，如果挂单超过3秒未成交，则撤单并退出策略。如果开仓全部成交，则以对手价发平仓委托，等待全部成交后退出策略。
3、Join方法中的策略每次执行时，从队列中取出所有数据，以最后一笔行情的盘口价格作为委托价。
4、OnRtnOrder收到订单状态通知时更新本地订单状态、持仓手数，在策略中根据状态的变化进行后续操作。
## 多进程
# 请在这里填写需要测试的合约数据
# 警告：该例子只支持上期所品种平今仓测试
test_exchange_id = 'SHFE' # 交易所
test_instrument_id = 'ag1912' # 合约代码
test_vol = 1 # 报单手数
share_queue = Queue(maxsize=100) # 队列
if __name__ == "__main__":
import sys
sys.path.append("..")
from account_info import my_future_account_info_dict
future_account = my_future_account_info_dict['SimNow']
# 行情进程
md_process = Process(target=TickEngine, args=(future_account.server_dict['MDServer']
, future_account.broker_id
, future_account.investor_id
, future_account.password
, future_account.app_id
, future_account.auth_code
, future_account.instrument_id_list
, [share_queue]
, future_account.md_page_dir)
)
# 交易进程
trader_process = Process(target=TraderEngine, args=(future_account.server_dict['TDServer']
, future_account.broker_id
, future_account.investor_id
, future_account.password
, future_account.app_id
, future_account.auth_code
, share_queue
, future_account.td_page_dir)
)
md_process.start()
trader_process.start()
md_process.join()
trader_process.join()
说明：
1、前几节的例子中需要手动设置涨跌停价作为报单价，这里我们以实时行情的盘口价作为报单价，所以不再需要设置涨跌停价参数。
2、share_queue是一个队列，在多进程中，队列数据可以实现共享。
3、md_process和trader_process分别是行情进程和交易进程。这两个进程通过队列share_queue共享数据。
4、所有进程的join方法须在start方法之后最后调用。
5、这段代码放置在策略代码最后，执行即可看到执行结果。
6、参考这个例子可以很方便的扩展一对多、多对多的进程间数据共享模式。
AlgoPlus的设计在登录时通过查询获取初始持仓、可用资金，成交发生时自动增减持仓数量，当平仓报单时自动增加冻结数量。当报撤单、出入金时自动增减可用资金。
## 智能交易指令
买卖智能开平指令
只关注买卖，不关注平仓还是开仓，优先平仓，无持仓的情况下再开仓。
除了buyOpen、sellClose、sellOpen、buyClose、closeLong、closeShort这些指定了开平方向的指令，其他都是智能开平指令。
记录学习基于AlgoPlus的量化交易开发准备工作，原文参考https://www.zhihu.com/column/AlgoPlus
yum install -y git
yum install -y vim
yum -y install wget
yum install -y bzip2
wget --no-check-certificate https://mirrors.tuna.tsinghua.edu.cn/anaconda/archive/Anaconda3-2021.11-Linux-x86_64.sh
bash Anaconda3-2021.11-Linux-x86_64.sh -u
vim /etc/profile
export PATH=/root/anaconda3/bin:$PATH
source /etc/profile
conda create -n foralgo python=3.7
conda env list
source activate foralgo
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/r
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/pro
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/msys2
#显示检索路径
conda config --set show_channel_urls yes
#显示镜像通道
conda config --show channels
#更新pip源
pip install -i https://pypi.doubanio.com/simple pip -U --user
vim /etc/yum.repos.d/mongodb-org-4.2.repo
编辑以下内容
[mongodb-enterprise]
name=MongoDB Enterprise Repository
baseurl=https://repo.mongodb.com/yum/redhat/$releasever/mongodb-enterprise/4.2/$basearch/
gpgcheck=1
enabled=1
gpgkey=https://www.mongodb.org/static/pgp/server-4.2.asc
使用yum进行安装
yum install -y mongodb-org
vim /etc/mongod.conf
systemLog:
destination: file #日志输出方式。file/syslog,如果是file，需指定path，默认是输出到标准输出流中
path: /var/log/mongodb/mongod/log #日志路径
logAppend: false #启动时，日志追加在已有日志文件内还是备份旧日志后，创建新文件记录日志, 默认false
net:
port: 27017 #监听端口，默认27017
bindIp: 127.0.0.1 #绑定监听的ip，设置为127.0.0.1时，只会监听本机
maxIncomingConnections: 65536 #最大连接数，可接受的连接数还受限于操作系统配置的最大连接数
wireObjectCheck: true #校验客户端的请求，防止错误的或无效BSON插入,多层文档嵌套的对象会有轻微性能影响,默认true
processManagement:
fork: true # 后台运行
security:
authorization: enabled # enabled/disabled #开启客户端认证
storage:
dbPath: /var/lib/mongodb # 数据库地址
journal:
enabled: true #启动journal,64位系统默认开启，32位默认关闭
启动mongo
mongod -f /etc/mongod.conf
pip install AlgoPlus -i http://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
docker pull centos:7
yum install -y git
yum install -y vim
yum -y install wget
yum install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gcc make
wget https://www.python.org/ftp/python/3.7.0/Python-3.7.0.tar.xz
tar xvf Python-3.7.0.tar.xz
mv Python-3.7.0 python3
cd python3
./configure --prefix=/usr/local/python3
make && make install
检查python3.7的编译器：
/usr/local/python3/bin/python3.7
建立Python3和pip3的软链
ln -s /usr/local/python3/bin/python3 /usr/bin/python3
并将/usr/local/python3/bin加入PATH
（1）vim /etc/profile
（2）按“I”，然后贴上下面内容：
# vim ~/.bash_profile
# .bash_profile
# Get the aliases and functions
if [ -f ~/.bashrc ]; then
. ~/.bashrc
fi
# User specific environment and startup programs
PATH=$PATH:$HOME/bin:/usr/local/python3/bin
export PATH
vim /etc/yum.repos.d/mongodb-org-4.2.repo
编辑以下内容
[mongodb-enterprise]
name=MongoDB Enterprise Repository
baseurl=https://repo.mongodb.com/yum/redhat/$releasever/mongodb-enterprise/4.2/$basearch/
gpgcheck=1
enabled=1
gpgkey=https://www.mongodb.org/static/pgp/server-4.2.asc
使用yum进行安装
yum install -y mongodb-org
vim /etc/mongod.conf
systemLog:
destination: file #日志输出方式。file/syslog,如果是file，需指定path，默认是输出到标准输出流中
path: /var/log/mongodb/mongod/log #日志路径
logAppend: false #启动时，日志追加在已有日志文件内还是备份旧日志后，创建新文件记录日志, 默认false
net:
port: 27017 #监听端口，默认27017
bindIp: 127.0.0.1 #绑定监听的ip，设置为127.0.0.1时，只会监听本机
maxIncomingConnections: 65536 #最大连接数，可接受的连接数还受限于操作系统配置的最大连接数
wireObjectCheck: true #校验客户端的请求，防止错误的或无效BSON插入,多层文档嵌套的对象会有轻微性能影响,默认true
processManagement:
fork: true # 后台运行
security:
authorization: enabled # enabled/disabled #开启客户端认证
storage:
dbPath: /var/lib/mongodb # 数据库地址
journal:
enabled: true #启动journal,64位系统默认开启，32位默认关闭
启动mongo
mongod -f /etc/mongod.conf
systemctl start
yum install vixie-cron
yum install crontabs
service crond status
启动rsyslog&crond服务
systemctl start rsyslog
systemctl start crond
systemctl restart crond
tail -f /var/log/cron
crontab -l 查看所有定时任务
crontab -e 编辑修改定时任务
更改时区
tzselect
vim /etc/crontab
添加变量 CRON_TZ=Asia/Shanghai
cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
service crond restart
示例
30 20 * * 1-5 python3 /home/stock/data_main.py
0 5 * * 2-6 python3 /home/stock/frm_main.py
30 7 * * 1-5 python3 /home/stock/stock_match.py
30 8 * * 1-5 python3 /home/stock/timer.py
docker commit -a "ikangbow" -m "描述信心" 730661ccf053 ikangbow/myquant:1.0.1
docker push ikangbow/myquant:1.0.1
docker run -e TZ="Asia/Shanghai" --privileged -it -d --name myquant ikangbow/myquant:1.0.3
wget http://prdownloads.sourceforge.net/ta-lib/ta-lib-0.4.0-src.tar.gz
cd ta-lib/
./configure --prefix=/usr
make && make install
cd /usr
find -name libta_lib.so.0
vim /etc/profile
export LD_LIBRARY_PATH=/usr/lib64
pip install ta-lib -U
cp /usr/lib/libta_lib.* /usr/lib64/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
corePoolSize：核心线程最大数量，即线程池中常驻线程的最大数量。线程池新建线程的时候，如果当前线程总数小于corePoolSize，则新建的是核心线程；如果超过了corePoolSize则新建的是非核心线程。
maximumPoolSize：线程池中运行的最大线程数，包括核心线程和非核心线程。
keepAliveTime：线程池中空闲线程所存活的最长时间（仅适用于非核心线程）。
unit：表示超出核心线程数之外的线程的空闲存活时间，也就是核心线程不会消除，但是超出核心线程数的部分线程如果空闲一定的时间则会被消除，可以通过keepAliveTime来设置空闲时间。
workQueue：用来存放任务的阻塞队列，假设我们现在核心线程都被使用，还有任务进来则全部放入队列，直到整个队列被放满但再持续进入则会创建新的线程。
ThreadFactory：实际上是一个线程工厂，用来生产线程执行任务，我们可以选择使用默认的创建工厂，产生的线程都在同一个组内，拥有相同的优先级，且都不是守护线程。当然我们也可以选择自定义线程工厂，一般会根据业务来制定不同的线程工厂。
Handler：任务拒绝策略，有两种情况，第一种是当我们调用shutdown等方法关闭线程池后，这时候即使线程池内部还有没执行完的任务正在执行，但是由于线程池已经关闭，我们再继续想线程池提交任务就会遭到拒绝，另一种情况就是当达到最大线程数，线程池已经没有能力继续处理新提交的任务时，这时也就会拒绝。
JDK自带有三个类加载器：BootStrapClassLoader，ExtClassLoader，AppClassLoaser
BootStrapClassLoader是ExtClassLoader的父类加载器，默认加载%JAVA_HOME%/lib下的jar包和class文件。
ExtClassLoader是AppClassLoader的父类加载器，负责加载%JAVA_HOME%/jre/lib/ext目录下的一些扩展jar。
AppClassLoader是自定义类加载器的父类，负责加载classpath下的类文件，是系统类加载器，线程上下文加载器继承ClassLoader实现自定义类加载器。
当某个类加载器需要加载.class文件时，它首先会把这个任务委托给他的上级类加载器，递归这个操作，如果上级的类加载器没有加载，自己才会去加载这个类。即向上委派，实际上就是查找缓存，看是否加载了该类，有则返回，没有继续向上；向下查找，查找加载的路径，有则加载，没有继续向下查找。
作用：主要为了安全性，避免用户自己编写的类动态替换java的一些核心类。同时也避免了类的重复加载，因为JVM中区分不同的类，不仅仅是根据类名，相同的class文件被不同的classLoader加载就是不同的两个类。
Java虚拟机对class文件采用的是按需加载的方式，也就是说当需要使用该类时才会将它的class文件加载到内存生成class对象，加载某个类class文件时，java虚拟机采用的是双亲委派机制，即把请求交由父类处理。
如果一个类加载器收到了类加载请求，它并不会自己先加载，而是把这个请求委托给父类的加载器去执行
如果父类加载器还存在其他父类加载器，则进一步向上委托，依次递归，请求最终将到达顶层的引导类加载器
如果父类加载器可以完成类加载的任务，就成功返回，倘若父类加载器无法完成加载任务，子类加载器材会尝试自己去加载，这就是双亲委派机制
父类加载器一层一层往下分配任务，如果子类加载器能加载，则加载此类，如果一直向下查找也无法加载此类，则抛出异常。
避免类的重复加载。
保护程序安全，防止核心API被随意篡改。