[原创] TradingViewWebHook报警直连FMZ机器人

2020-12-11
1582
0
0
0
发明者量化

数字货币期货

最近越来越多的TradingView使用者，把TradingView图表信号连通到发明者量化平台（FMZ.COM）上根据图表信号，让FMZ上的机器人策略执行交易，这样对于编程技术小白来说省去了大量代码编写、设计工作。直接可以让指标用于程序化、自动化交易，降低了不少程序化、量化交易开发门槛。对于TradingViewWebHook实现实盘自动交易，有好几种设计方案。

前一篇方案：https://www.fmz.com/digest-topic/5533。

之前的方案，走的是发明者量化交易平台扩展API接口，给机器人发送指令。今天我们一起来看另一种方案，让TradingView的报警WebHook请求直接发送给FMZ量化交易平台机器人，实现直接发送指令，命令机器人交易。

机器人策略源码

策略使用Python编写，在使用该策略创建机器人并启动后，机器人会创建一个线程，该线程会启动一个服务监听设置的端口。等待外部请求并处理。我测试的时候是用在服务器上的托管者测试的，托管者所在设备必须能被外部访问。机器人执行交易时，使用的是市价单接口，当然也可以改造这个策略，实现限价单下单逻辑。为了简单易懂、程序精简，这里使用了市价单，所以必须交易所支持市价单才行。

'''
请求格式：http://x.x.x.x:xxxx/data?access_key=xxx&secret_key=yyy&type=buy&amount=0.001
策略机器人参数：
- 类型：加密字符串，AccessKey , SecretKey ，可以用FMZ平台的低权限的API KEY，或者自己生成KEY也可以。
- 类型：字符串，合约ID，ContractType
- 类型：数值，端口号，Port
'''

import _thread
import json
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse

def url2Dict(url):
    query = urlparse(url).query  
    params = parse_qs(query)  
    result = {key: params[key][0] for key in params}  
    return result

class Executor(BaseHTTPRequestHandler):
    def do_GET(self):
        try:
            dictParam = url2Dict(self.path)
            Log("测试", dictParam)
        except Exception as e:
            Log("Provider do_GET error, e:", e)
    def do_POST(self):
        try:
            self.send_response(200)
            self.send_header("Content-type", "application/json")
            self.end_headers()
            dictParam = url2Dict(self.path)
            
            # 校验
            if len(dictParam) == 4 and dictParam["access_key"] == AccessKey and dictParam["secret_key"] == SecretKey:
                del dictParam["access_key"]
                del dictParam["secret_key"]
                Log("接收到请求", "参数：", dictParam, "#FF0000")
                '''
                map[access_key:xxx amount:0.001 secret_key:yyy type:buy]
                '''
                isSpot = True
                if exchange.GetName().find("Futures") != -1:
                    if ContractType != "":
                        exchange.SetContractType(ContractType)
                        isSpot = False 
                    else :
                        raise "未设置期货合约"
                
                if isSpot and dictParam["type"] == "buy":
                    exchange.Buy(-1, float(dictParam["amount"]))
                    Log(exchange.GetAccount())
                elif isSpot and dictParam["type"] == "sell":
                    exchange.Sell(-1, float(dictParam["amount"]))
                    Log(exchange.GetAccount())
                elif not isSpot and dictParam["type"] == "long":
                    exchange.SetDirection("buy")
                    exchange.Buy(-1, float(dictParam["amount"]))
                    Log("持仓：", exchange.GetPosition())
                elif not isSpot and dictParam["type"] == "short":
                    exchange.SetDirection("sell")
                    exchange.Sell(-1, float(dictParam["amount"]))
                    Log("持仓：", exchange.GetPosition())
                elif not isSpot and dictParam["type"] == "cover_long":
                    exchange.SetDirection("closebuy")
                    exchange.Sell(-1, float(dictParam["amount"]))
                    Log("持仓：", exchange.GetPosition())
                elif not isSpot and dictParam["type"] == "cover_short":
                    exchange.SetDirection("closesell")
                    exchange.Buy(-1, float(dictParam["amount"]))
                    Log("持仓：", exchange.GetPosition())
            
            # 写入数据应答
            self.wfile.write(json.dumps({"state": "ok"}).encode())
        except Exception as e:
            Log("Provider do_POST error, e:", e)


def createServer(host):
    try:
        server = ThreadingHTTPServer(host, Executor)
        Log("Starting server, listen at: %s:%s" % host)
        server.serve_forever()
    except Exception as e:
        Log("createServer error, e:", e)
        raise Exception("stop")

def main():
    # 开启一个线程
    try:
        _thread.start_new_thread(createServer, (("0.0.0.0", Port), ))         # VPS服务器上测试        
    except Exception as e:        
        Log("错误信息：", e)
        raise Exception("stop")    
    Log("账户资产信息：", _C(exchange.GetAccount))
    while True:
        if exchange.GetName() == "Futures_CTP":
            if exchange.IO("status"):
                LogStatus(_D(), "CTP连接")
            else:
                LogStatus(_D(), "CTP未连接")
        else:
            LogStatus(_D())
        Sleep(2000)

策略参数：

TradingView的WebHook报警请求

报警请求设置为：

http://xxx.xxx.xxx.xxx:80/data?access_key=e3809e173e23004821a9bfb6a468e308&secret_key=45a811e0009d91ad21154e79d4074bc6&type=sell&amount=0.1

由于Trading View发送的是POST请求，所以监听服务中要监听POST请求，并且Trading View对于http协议只允许用80端口。

xxx.xxx.xxx.xxx，为机器人所在托管者的设备IP地址。填写自己的设备具体IP地址，需要注意必须能被外网访问才行。
access_key，secret_key可以自己生成，只要WebHook报警请求中的access_key，secret_key填写与机器人参数上配置的一致即可。
type，交易方向，买入或者卖出、开仓或者平仓，注意现货期货是区分的。如果是期货，注意机器人参数上要设置期货合约代码，并且配置的交易所对象需要是期货交易所。
amount，交易数量。

运行测试

使用wexApp模拟盘测试。

END

完整策略地址：https://www.fmz.com/strategy/221850

方案中的access_key，secret_key仅仅为识别，对于使用http并无安全性。该方案仅仅作为思路、抛砖引玉，实际应用应当增加安全方面的考虑，使用https通信。

更新

由于HTTPServer本身有些坑，考虑使用ThreadingHTTPServer代替。
参考：https://docs.python.org/3.7/library/http.server.html
需要Python3.7版本。
增加了GET请求测试
可以使用浏览器，地址栏输入请求链接。
```
http://xxx.xx.xx.xx/data?access_key=123&secret_key=123&type=buy&amount=1
```
测试：
增加了商品期货连接状态的判断