【Python量化交易】——1、封装交易所API

在刚刚过去的一个星期里，博主一直在捣鼓 Python量化交易 的内容。在写这篇文章的时候已经用python实现网格法自动交易的功能，其次也成功将脚本部署到服务器自动运行，另外又用 flask模块 完成 WebAPI 的封装，实现了交易状况的 实时监控 。接下来的几篇文章里我将逐一介绍我是如何不断掉坑以及爬坑的艰辛历程， 感兴趣的小伙伴们不要错过~
个人博客地址：ht/tps://www.asyu17.cn/

传送门
【Python量化交易】——1、封装交易所API
【Python量化交易】——2、利用python实现网格法交易策略以及回测
【Python量化交易】——3、存储模块以及WebAPI的封装
【Python量化交易】——4、主函数
【Python量化交易】——5、部署至linux服务器

目录

  • 获取accesskey以及secretkey
  • 封装Market模块
  • 封装Trade模块
  • 结语
  • 微信公众号

获取accesskey以及secretkey

我是在中币网注册的交易账号，所以我就以中币网为例。为避免广告嫌疑我就不放传送门了。。。
登陆完成后，在头像下面有一个API设置的选项，点击进入。
API设置
输入必要信息，添加一个api，随后记录下获取的accesskey和secretkey（这两个号码要记住，后面的交易模块全部需要这两个参数）
在这里插入图片描述

封装Market模块

点击 行情 ，这几个接口我们全部需要封装。
在这里插入图片描述
先建立一个 Market类 为保证一定能获取请求结果，我们封装 get模块 ,完成后，我们封装其余几个请求（所有请求参数放在data里面）
调用示例在下面：

Market.py

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
52
53
54
import requests
import time 

# 市场部分
class market(object):
	# 官方请求函数
    def get(self, url):
        # 返回字典类型
        while True:
            try:
                r = requests.get(url)
            except Exception:
                time.sleep(0.5)
                continue
            if r.status_code != 200:
                time.sleep(0.5)
                continue
            r_info = r.json()
            r.close()
            return r_info
 # 获取所有币的当前数据
    def GetAllCoinCurrentData(self):
        url = 'http://api.zb.live/data/v1/allTicker'
        data = self.get(url=url)
        return data

    # 获取一个币的当前数据
    def GetOneCoinCurrentData(self, market):
        url = 'http://api.zb.live/data/v1/ticker?market=' + market
        data = self.get(url=url)
        return data

    #  获取市场深度
    def GetMarketDepth(self, market, size):
        # 市场深度是指在不影响当前价格的前提下可以完成的最大交易量
        # 应该是获取市场上有多少买单和多少卖单
        url = 'http://api.zb.live/data/v1/depth?market=' + market + '&size=' + size
        data = self.get(url=url)
        return data

    # 获取指定市场最新五十条成交数据
    def GetHistoryTradeData(self, market):
        url = 'http://api.zb.live/data/v1/trades?market=' + market
        data = self.get(url=url)
        return data

    # 获取K线数据
    def GetKlineData(self, market, type='1day', size='100'):
        # 最多获取最新的1000条
        # type	String	1min/3min/5min/15min/30min/1day/3day/1week/1hour/2hour/4hour/6hour/12hour
        # since 为获取这个时间戳之后的数据
        url = 'http://api.zb.live/data/v1/kline?market=' + market + '&type=' + type + '&size=' + size
        data = self.get(url=url)
        return data

封装Trade模块

进入 API文档 ，点击 交易
在这里插入图片描述
画圈的这五个是 必须 要封装的，我们就以封装这几个为例，其他都是一模一样的。
在这里插入图片描述
在交易模块中，参数比较杂，我们在封装的时候就 全部放到data字典里 ，然后用urllib.parse函数来解析并放到url里。

特别提醒：
在 Encryption 函数加密时，data字典先不添加sign和reqTime参数，加密完成获取到sign后，再往data里面添加这两个参数！！！
调用示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
data = {
    'accesskey': 你的accesskey,
    'method': 'getAccountInfo',
}
sign = zbTrade.Encryption(data=data, secretKey=secretKey)
data = {
    'accesskey': accessKey,
    'method': 'getAccountInfo',
    'sign': sign,
    'reqTime': int(time.time() * 1000)
}
res = zbTrade.GetAccountInfo(data=data)
print(res)

Trade.py

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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
import requests
import time
import hmac
import hashlib
from urllib import parse


# 交易
class trade(object):
    # 官方请求函数
    def get(self, url):
        # 返回字典类型
        while True:
            try:
                r = requests.get(url)
            except Exception:
                time.sleep(0.5)
                continue
            if r.status_code != 200:
                time.sleep(0.5)
                continue
            r_info = r.json()
            r.close()
            return r_info

    # 委托下单
    def Order(self, data):
        # accesskey	String	accesskey
        # market	String	交易币种计价币种
        # tradeParams	String	交易参数，数组格式[[价格，数量]，[价格，数量]]，最多20个
        # tradeType	int	交易类型1/0[buy/sell]
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        # acctType	int	杠杆 1/0[杠杆/现货]（可选参数,默认为: 0 现货）
        url = 'https://trade.zb.live/api/order?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    # 取消委托下单
    def CancelOrder(self, data):
        # accesskey	String	accesskey
        # currency	String	交易币种计价币种
        # id	long	委托挂单号
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        # customerOrderId	String	自定义ID（4-36位字符，数字或字母组成，customerOrderId和id只能选一个填写）
        url = 'https://trade.zb.live/api/cancelOrder?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    # 获取当前市场上的委托单
    def GetOrder(self, data):
        # accesskey	String	accesskey
        # tradeType	int	交易类型1/0[buy/sell]
        # currency	String	交易币种计价币种
        # pageIndex	int	当前页数
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        url = 'https://trade.zb.live/api/getOrder?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    # 获取未成交或部份成交挂单
    def GetUnfinishedOrders(self,data):
        # method	String	直接赋值getUnfinishedOrdersIgnoreTradeType
        # accesskey	String	accesskey
        # currency	String	交易币种计价币种
        # pageIndex	int	当前页数
        # pageSize	int	每页数量
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        url = 'https://trade.zb.live/api/getUnfinishedOrdersIgnoreTradeType?' + parse.urlencode(data)
        data = self.get(url=url)
        return data
    # 获取充值地址
    def GetRechargeAddress(self, data):
        # accesskey	String	accesskey
        # currency	String	币种
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        url = 'https://trade.zb.live/api/getUserAddress?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    # 获取提现地址
    def GetWithdrawalAddress(self, data):
        # method	String	直接赋值getWithdrawAddress
        # accesskey	String	accesskey
        # currency	String	币种
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        url = 'https://trade.zb.live/api/getWithdrawAddress?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    # 提现
    def Withdrawal(self, data):
        # accesskey	String	accesskey
        # amount	float	提现金额
        # currency	String	币种
        # fees	float	提现矿工费
        # itransfer	int	是否同意bitbank系内部转账(0不同意，1同意，默认不同意)
        # method	String	直接赋值withdraw
        # receiveAddr	String	接收地址（必须是认证了的地址，bts的话，以"账户_备注"这样的格式）
        # safePwd	String	资金安全密码
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        url = 'https://trade.zb.live/api/getChargeRecord?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    def GetAccountInfo(self, data):
        # method	String	直接赋值getAccountInfo
        # accesskey	String	accesskey
        # sign	String	请求加密签名串
        # reqTime	long	当前时间毫秒数
        url = 'https://trade.zb.live/api/getAccountInfo?' + parse.urlencode(data)
        data = self.get(url=url)
        return data

    # 完成sign加密
    def Encryption(self, data,
                   secretKey):
        # data最后需要转换的格式如：accesskey=359f9eaa-53ec-491d-b04b-dc8462698123&method=getAccountInfo
        accessKey = parse.urlencode(data)
        secretKey = hashlib.sha1(secretKey.encode('utf-8')).hexdigest()
        sign = hmac.new(key=bytes(secretKey.encode('utf-8')), msg=accessKey.encode('utf-8'),
                        digestmod=hashlib.md5).hexdigest()
        return sign

结语

以上就完成了交易市场API的最简单的封装。在下篇文章里我们将用python实现网格法交易策略，并利用在这篇文章中封装的API进行交易策略的回测。

微信公众号：

小术快跑