python版期货量化交易（AlgoPlus）案例（多进程处理子任务）

python因为简单、易上手，所以深受大家的喜爱，并且随着人工智能的不断发展与进步，python也一跃成为了最受欢迎的编程语言之一，俗话说：人生苦短，我用python。伴随着量化交易的崛起，上期所下面的子公司根据CTP接口封装出了python版本的api接口：Algoplus

文章目录

python版期货量化交易（AlgoPlus）案例（多进程处理子任务）
前言
一、AlgoPlus是什么？
二、使用步骤
- 1.引入库
- 2.账号配置
- 3.合成分钟线
- 4.Join函数
- 5.结果展示
总结

前言

为了策略的安全性，有必要自己搭建一套交易系统

提示：以下是本篇文章正文内容，下面案例可供参考

一、AlgoPlus是什么？

安装： pip install AlgoPlus

关于AlgoPlus的介绍请查看www.algoplus.com官网，这里不得不得不吐槽有一下，AlgoPlus官网有一段时间打不开。

二、使用步骤

1.引入库

代码如下（示例）：

# from CTP.MdApi import *
from AlgoPlus.CTP.FutureAccount import get_simnow_account, FutureAccount
from AlgoPlus.CTP.FutureAccount import SIMNOW_SERVER, MD_LOCATION, TD_LOCATIONfrom multiprocessing import Process, Queuefrom CTP.MdApi import run_bar_engine, run_tick_enginefrom CTP.TradeApi import run_trade_engine

2.账号配置

代码如下（示例）：

 # 账户配置future_account = FutureAccount(broker_id='9999',  # 期货公司BrokerID# server_dict={'TDServer': "180.168.146.187:10130", 'MDServer': '180.168.146.187:10131'},   # TESTserver_dict={'TDServer': "218.202.237.33:10102", 'MDServer': '218.202.237.33:10112'},       # 移动# TDServer为交易服务器，MDServer为行情服务器。服务器地址格式为"ip:port。"reserve_server_dict={},investor_id="****************",     # 账户password="****************",        # 密码app_id='simnow_client_test',        # 认证使用AppIDauth_code='0000000000000000',       # 认证使用授权码instrument_id_list=instrument_id_list,  # 订阅合约列表md_page_dir=MD_LOCATION,                # MdApi流文件存储地址，默认MD_LOCATIONtd_page_dir=TD_LOCATION                 # TraderApi流文件存储地址，默认TD_LOCATION)

SimNow提供了7x24小时的模拟服务器

3.合成分钟线

代码如下（algoplus提供官方示例）：

# ///深度行情通知def OnRtnDepthMarketData(self, pDepthMarketData):last_update_time = self.bar_dict[pDepthMarketData['InstrumentID']]["UpdateTime"]is_new_1minute = (pDepthMarketData['UpdateTime'][:-2] != last_update_time[:-2]) and pDepthMarketData['UpdateTime'] != b'21:00:00'  # 1分钟K线条件# is_new_5minute = is_new_1minute and int(pDepthMarketData['UpdateTime'][-4]) % 5 == 0  # 5分钟K线条件# is_new_10minute = is_new_1minute and pDepthMarketData['UpdateTime'][-4] == b"0"  # 10分钟K线条件# is_new_10minute = is_new_1minute and int(pDepthMarketData['UpdateTime'][-5:-3]) % 15 == 0  # 15分钟K线条件# is_new_30minute = is_new_1minute and int(pDepthMarketData['UpdateTime'][-5:-3]) % 30 == 0  # 30分钟K线条件# is_new_hour = is_new_1minute and int(pDepthMarketData['UpdateTime'][-5:-3]) % 60 == 0  # 60分钟K线条件# # 新K线开始if is_new_1minute and self.bar_dict[pDepthMarketData['InstrumentID']]["UpdateTime"] != b"99:99:99":for md_queue in self.md_queue_list:md_queue.put(self.bar_dict[pDepthMarketData['InstrumentID']])# 将Tick池化为Bartick_to_bar(self.bar_dict[pDepthMarketData['InstrumentID']], pDepthMarketData, is_new_1minute)

注意：我在向队列里添加数据时使用了深拷贝，官方给的示例有时无法得到正确的一分钟k线数据，因为当你在交易进程中还未拿到k线数据之前，已经被修改了。

4.Join函数

在Join函数中可以写策略逻辑：开仓、平仓等。

    def Join(self):lastPrice = 0  # 上根k线的收盘价while True:if self.status == 0:if not self.md_queue.empty():makeData = self.md_queue.get(True)# 撤单if self.local_position_dict:if self.local_position_dict[self.tickData['InstrumentID']]['Volume'] != 0:self.req_remove(self.tickData)print(f"====={makeData}")# 亏损超8个点止损，回撤6个点止损for instrument_id, position in self.local_position_dict.items():if self.symbol_close[instrument_id] == 1:if instrument_id not in self.md_dict.keys():breakif position['Volume'] != 0 and position['Direction'] == b'0':self.sell_close(b'', instrument_id, makeData['LastPrice'] - 6, 1)print(f"卖平仓：{lastPrice}，{makeData['LastPrice']}")else:self.buy_close(b'', instrument_id, makeData['LastPrice'] + 6, 1)print(f"买平仓：{lastPrice}，{makeData['LastPrice']}")if lastPrice != 0:sleep(1)   # 时间是59s的时候休眠1s，0s时开仓if makeData['LastPrice'] >= lastPrice:self.buy_open(b'', b'p2209', makeData['LastPrice'] + 6, 1)print(f"买开仓：{lastPrice}，{makeData['LastPrice']}")else:self.sell_open(b'', b'p2209', makeData['LastPrice'] - 6, 1)print(f"卖开仓：{lastPrice}，{makeData['LastPrice']}")lastPrice = makeData['LastPrice']# 初始化为最新价self.HighPrice = makeData['LastPrice']self.LowPrice = makeData['LastPrice']else:sleep(1)# 止盈止损self.check_position()