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_LOCATION

from multiprocessing import Process, Queue

from CTP.MdApi import run_bar_engine, run_tick_engine

from 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'},   # TEST
        server_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',        # 认证使用AppID
        auth_code='0000000000000000',       # 认证使用授权码
        instrument_id_list=instrument_id_list,  # 订阅合约列表
        md_page_dir=MD_LOCATION,                # MdApi流文件存储地址，默认MD_LOCATION
        td_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池化为Bar
        tick_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():
                                break
                            if 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()

           

5.结果展示

源代码链接: https://pan.baidu.com/s/10HnZf89cEAbXtlbrJRlVEA .

提取码：yooq

如果您熟悉c++11，请看CTP开发案例：

CTP接口开发链接: https://blog.csdn.net/syealfalfa/article/details/124994132 .

总结

请关注www.algoplus.com官网的最新消息