CTP量化投资技术研究 - AlgoPlus量化投资开源框架 - 将交易进行到底这是广告
本文将向大家展示使用AlgoPlus从账户层面，包括手动开仓或程序持仓，对持仓进行风控的例子。
很多商业软件也提供风控功能，例如云端止盈止损单，但是AlgoPlus可以帮助大家实现更个性化的风控方案。如果使用过程遇到任何问题请联系我们。
AlgoPlus是使用Cython、ctypes技术封装的Python版量化投资开源框架，不仅释放了GIL，而且充分利用CTP线程特性，既能满足低延时的交易需求，又提高了易用性。
止盈止损方法
常用的止盈止损方案有：固定止损、固定止盈、跟踪止损、阶梯止损、保本、时间止损。详细介绍可以参考：
在AlgoPlus量化投资开源项目的风控模块中，我们封装了上述止盈止损逻辑。AlgoPlus用户即可以此为账户风控引擎，也可从中抽离需要的逻辑放在自己的策略中。
配置账户参数
在account_info.py中配置账户信息，并订阅合约。如果没有订阅持仓合约行情，将无法进行止盈止损。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
'SimNow': FutureAccountInfo( broker_id='9999', # 期货公司BrokerID # TDServer为交易服务器，MDServer为行情服务器。服务器地址格式为"ip:port。" server_dict={'TDServer': "180.168.146.187:10130", 'MDServer': '180.168.146.187:10131'}, # 备用服务器地址 reserve_server_dict={}, investor_id='', # 账户 password='', # 密码 app_id='simnow_client_test', # 认证使用AppID auth_code='0000000000000000', # 认证使用授权码 # 订阅合约列表 instrument_id_list=[b'rb2001'], ), |
撤单次数数据结构
|
1 2 |
# {"InstrumentID": 0} self.order_action_num_dict = {} |
撤单次数是一个以合约名为键值的字典。收到撤单通知后，对撤单次数计算增加。
收到撤单通知时增加撤单次数计数
|
1 2 3 4 5 6 7 8 9 10 11 |
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 |
止盈止损参数数据结构
|
1 2 |
# {"InstrumentID": {"Type": []}} self.pl_parameter_dict = {} |
止损参数是一个以合约名为键值的字典，根据不同的止损止盈类型存储价差/时间差等参数。
其中，止损类型参数取值：
| b”0″ | 固定止盈 |
| b”1″ | 固定止损 |
价差/时间差等参数以列表形式存储，有些止损逻辑可能需要多个参数。
持仓数据结构
|
1 2 |
# {"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字段。
|
1 2 3 4 |
rtn_trade["IsLock"] = False # 平仓状态 rtn_trade["AnchorTime"] = timer() # 成交发生时间 rtn_trade["StopProfitDict"] = {} # 止盈触发价格，持仓期间实时更新 rtn_trade["StopLossDict"] = {} # 止损触发价格，持仓期间实时更新 |
成交通知
收到成交通知时放入一个列表中，等待后续处理，避免在此设计复杂的耗时操作。
|
1 2 3 4 5 6 7 |
def OnRtnTrade(self, pTrade): """ 当报单成交时，可以在本方法中获得通知。不适宜在回调函数里做比较耗时的操作。 :param pTrade: AlgoPlus.CTP.ApiStruct中的TradeField实例。 :return: """ self.local_rtn_trade_list.append(pTrade.to_dict_raw()) |
处理成交通知
根据开买卖开平字段将成交成交信息放入持仓数据结构中。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
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 |
实时监控当前行情价格是否触及止盈止损价
遍历持仓数据结构中的所有成交明细，判断最新行情是否触发某个止盈止损阈值，如果触发则录入平仓报单。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
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) |
止盈止损逻辑
根据止盈止损（固定止盈、固定止损、跟踪止损、阶梯止损、保本止损）逻辑计算出触发价格，存入成交明细字典中。这里给出了固定止盈和固定止损的例子供大家参考：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
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 |
使用方法
配置账户并设置止损参数之后，运行profit_loss_manager_example.py，即可对账户进行监控。从快期客户端下单，当持仓触发止损条件，程序会自动进行平仓。
完整代码下载地址：
码云：https://gitee.com/AlgoPlus/AlgoPlus
GitHub：https://github.com/CTPPlus/AlgoPlus
评论前必须登录！
注册