火币量化API的简单使用-白红宇

火币网()是国内比较有名的数据货币交易所，它提供了API接口，可以用来作量化交易。

申请火币API接口

如下图所示，登陆火币官网后用户头像下面有一个API管理。

API管理

点开后创建API key

创建API key

这里需要输入三个安全验证码，分别是安全邮箱的验证码、手机验证码和google验证码。

生成API key成功

上图显示的两个参数的含义是，Access Key API 访问密钥， Secret Key 签名认证加密所使用的密钥(仅申请时可见)，是用户接入火币网API的凭证，因为涉及到你的资产安全，必须牢记。

如何使用火币网API 接口

我把官网的API介绍搬过来了，说的很明白，详细的看官网文档，见文末参考资料。

接口类型

火币为用户提供两种接口，您可根据自己的使用场景和偏好来选择适合的方式进行查询行情、交易或提现。

REST API

REST，即Representational State Transfer的缩写，是目前较为流行的基于HTTP的一种通信机制，每一个URL代表一种资源。
交易或资产提现等一次性操作，建议开发者使用REST API进行操作。

WebSocket API

WebSocket是HTML5一种新的协议(Protocol)。它实现了客户端与服务器全双工通信，通过一次简单的握手就可以建立客户端和服务器连接，服务器可以根据业务规则主动推送信息给客户端。

市场行情和买卖深度等信息，建议开发者使用WebSocket API进行获取。

接口鉴权

以上两种接口均包含公共接口和私有接口两种类型。

公共接口可用于获取基础信息和行情数据。公共接口无需认证即可调用。

私有接口可用于交易管理和账户管理。每个私有请求必须使用您的API Key进行签名验证。

接入URLs

您可以自行比较使用api.huobi.pro和api-aws.huobi.pro两个域名的延迟情况，选择延迟低的进行使用。

其中，api-aws.huobi.pro域名对使用aws云服务的用户做了一定的链路延迟优化。

SDK与代码示例

SDK(推荐)

其它代码示例

简单应用

我想通过python来访问火币API接口，获得近期的k线数据，然后根据历史信息计算出的相应的指标(主要是rsi指标)，来找到买卖的时间点，并进行自动化交易。如何实现这样的功能？

在上有一个

的例子，基于这个修改是最快速方便的。

使用git clone到本地

git clone https://github.com/huobiapi/REST-Python3-demo.git

文件目录

我们需要修改Utils.py和HuobiServices.py这两个文件

修改其中的全局变量

将刚才审请的acess_key和secret_key填入20和21行。

image.png

第一次执行get_account()函数后，会返回一个account_id，将其赋值给全局变量ACCOUNT_ID,这个值与帐户有关，在交易请求时作为参数，在代码中硬编码写入，就不用重复向后台请求了。如果只读取行情数据，不进行实盘交易的情况下，就不需要它，也用不到这个函数。我审请的api接口权限为读取，因此不需要改这儿。

demo代码实现

下面我们写一个使用火币api进行量化交易的demo,代码的功能很简单，就是获取k线数据，计算rsi指标，根据rsi指标来判断交易时机，这是一个非常粗糙的策略，几乎没有实用价值。

开发运行环境

ubuntu19.04 server

python3.7 (Anaconda 3)

python需要的库

talib

这是一个python中一个非常有名的金融库，里面提供了大量的方法，可以帮助我们快速计算各个指标，详细用法参考官网，这个库基于numpy，建议在anaconda环境下用pip安装。

#-*- coding:utf-8 -*-from HuobiServices import *import timeimport numpy as npimport os,jsonfrom talib import RSIimport requestsfrom utils import send_mail#保存配置信息config = {}#日志文件log_file = 'log/log.txt'#配置文件config_file = 'conf.json'#日志函数def log(msg):    global log_file    cur_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time())))    if not os.path.isfile(log_file):        open(log_file,'w')    open(log_file,'a').write('%s: %s\n'%(cur_time,msg))#获取我的帐户def get_my_balance(currency):    data = list(filter(lambda x: x['currency'] == currency and x['type'] == 'trade',get_balance()['data']['list']))[0]['balance']    return float(data[:data.find('.')+7])#获取当前的usdt/btc价格def get_current_price():    url = 'https://www.huobi.co/-/x/pro/market/overview5?r=ny2seo'    response = requests.get(url)    if response.status_code == 200:        data = response.json()        price_usdt= data['data'][2]['close']        return price_usdt    else:        return 0            #根据k线计算rsi，周期为14def get_rsi(data):    #调用talib库计算rsi指标    close = np.array([x['close'] for x in data])    real = RSI(close,14)    return real#买操作def buy(cur_price):    global config    #获取当前我帐户中的usdt    my_usdt = get_my_balance('usdt')    #取小数点后6位    my_usdt = float('%0.6f' % my_usdt)     #若my_usdt不为0的话    if  int(my_usdt) > 0:        #下单，以cur_price价钱全部买btc        ret = send_order(amount=my_usdt,source='api',symbol='btcusdt',_type='buy-market')        #config['BUY_PRICE']记下当前买的价格        config['BUY_PRICE'] = cur_price        #log(str(ret))        #在日志文件中记下这个买操作        log('buy at %f usdt/btc,current balance: %f btc,%f usdt' % (            cur_price,            get_my_balance('btc')            ,get_my_balance('usdt')            ))#卖操作def sell(cur_price):    global config    #获取我的帐户中的btc数量    my_btc = get_my_balance('btc')    #取小数点后6位    my_btc = float('%0.6f' % my_btc)    #若大于0.0001，因为btc是最小交易单位为万分之一比特币    if  my_btc > 0.0001:        #下单卖        ret = send_order(amount=my_btc,source='api',symbol='btcusdt',_type='sell-market')        #config['SELL_PRICE']记下当前的卖的价格        config['SELL_PRICE'] = cur_price        #log(str(ret))        #将这次卖操作写入日志文件        log('sell at %f usdt/btc,current balance: %f btc,%f usdt' % (            cur_price,            get_my_balance('btc'),            get_my_balance('usdt')            ))#买时机监控开关buy_monitor_switch = False#卖时机监控开关sell_monitor_switch = False#量化主函数def trade():    #声明全局变量    global config    global buy_monitor_switch,sell_monitor_switch    #获取k线数据，获取从当前向前200个k线数据，每个k线时长为config['period']    data = get_kline('btcusdt',"%dmin" % config['period'],200)['data']    #数据反转    data.reverse()    #计算rsi,取最后三个rsi值    rsi1,rsi2,rsi3  = get_rsi(data)[-3:]    #根据时间戳获取最后一个k线的时间    cur_time_ = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(data[-1]['id']))        #获取当前的btc/usdt的价格    cur_price = get_current_price()    log('%s rsi:%f, price:%f usdt/btc' % (cur_time_,rsi3,cur_price))        #如果rsi3小于30，开启买监控    if rsi3 < 30:        buy_monitor_switch =  True    #如果rsi大于70，开启卖监控    if rsi3 > 70:        sell_monitor_switch =  True    if buy_monitor_switch:              #若rsi曲线出线 触底反弹 或者 直接向上突破30 时，发出买的信息         if (rsi2 < rsi1 and rsi2 < rsi3) or (rsi3 > 30):            log('It is time to buy')                        try:                #向用户发出买的信息                send_mail(title='Buy Signal rsi=%f price= %f' % (rsi3,cur_price),msg='%s rsi:%f, price:%f usdt/btc' % (cur_time_,rsi3,cur_price))            except:                pass        #若rsi向上已经突破30，关闭买操作监控        if rsi3 > 30 :            buy_monitor_switch = False            if sell_monitor_switch:        #若rsi出现 触顶下降 时 或者 向下突破70，发出卖的信息        if (rsi2 > rsi1 and rsi2 > rsi3) or (rsi3 < 70):            log('It is time to sell')            try:                #向用户发出卖的信号                send_mail(title='Sell Signal rsi=%f price= %f' % (rsi3,cur_price),msg='%s rsi:%f, price:%f usdt/btc' % (cur_time_,rsi3,cur_price))            except:                pass        #若rsi向下突破70，关闭卖操作监控        if rsi3 < 70:            sell_monitor_switch = False   #主函数  def main():    #声明全局变量    global config    #k线的时间区间为15min    config['period'] = 15    while True:        #获取当前的分和秒        cur_miniute,cur_second = time.strftime("%M:%S", time.localtime(int(time.time()))).split(':')        cur_miniute = int(cur_miniute)        cur_second = int(cur_second)        #如果当前时间为整刻钟        if cur_miniute % config['period'] == 0 and cur_second == 0:            try:                #加载配置文件                config = json.load(open(config_file,'r'))                #监控操作                trade()            except:                #出错的话继续                continue            finally:                #配置信息写入文本文件                json.dump(config,open(config_file,'w'))            time.sleep(1)main()

配置文件是一个json文件

{"period": 15, "BUY_PRICE": 11403.96, "SELL_PRICE": 11430.09}

上面代码中使用了一个可以发邮件的功能

#-*- coding:utf-8 -*-import smtplibfrom email.mime.multipart import MIMEMultipartfrom email.mime.text import MIMETextdef send_mail(title,msg):    mail_content = msg    #发送方的邮箱和密码    sender_address = 'xxxx@gmail.com'    sender_pass = 'xxxxx'    #接收方的邮箱，多个邮箱的可以用分号隔开    receiver_address = 'xxxx@qq.com'    #Setup the MIME    message = MIMEMultipart()    message['From'] = sender_address    message['To'] = receiver_address    message['Subject'] = title   #邮件的title    #The body and the attachments for the mail    message.attach(MIMEText(mail_content, 'plain'))    #Create SMTP session for sending the mail    session = smtplib.SMTP('smtp.gmail.com', 587) #use gmail with port    session.starttls() #enable security    session.login(sender_address, sender_pass) #login with mail_id and password    text = message.as_string()    session.sendmail(sender_address, receiver_address, text)    session.quit()    #print('Mail Sent')

使用微信开启qq邮件提醒的话，就可以在第1时间得到交易行情信息。

这个demo没有进行实际交易，因为这个交易策略实在是太简单粗暴，没有应用价值，在实际的量化模型中往往要考虑多个指标，还要考虑交易手续费的问题。一个实用的量化模型可以给你带来丰厚的收益，当然一个失败的模型也能让你赔的很惨，千万不要用实盘去测试你的代码，用历史数据进行回测是正确的方法，如果你的模型在历史数据上运行的很好，收益率不错，再考虑在实盘上操作。

参考资料

火币API官方文档

火币API示例代码

talib文档(英文)

talib文档(中文)