Loading ...

TradingViewWebHook报警直连FMZ机器人

Author: 小小梦, Created: 2020-08-03 10:50:04, Updated: 2020-11-25 17:39:27

TradingViewWebHook报警直连FMZ机器人

最近越来越多的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 HTTPServer, 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_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 = HTTPServer(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:
        LogStatus(_D())
        Sleep(2000)

策略参数: img

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_keysecret_key可以自己生成,只要WebHook报警请求中的access_keysecret_key填写与机器人参数上配置的一致即可。
  • type,交易方向,买入或者卖出、开仓或者平仓,注意现货期货是区分的。如果是期货,注意机器人参数上要设置期货合约代码,并且配置的交易所对象需要是期货交易所。
  • amount,交易数量。

运行测试

使用wexApp模拟盘测试。

img

img

END

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

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

更新

由于HTTPServer本身有些坑,考虑使用ThreadingHTTPServer代替。 参考:https://docs.python.org/3.7/library/http.server.html 需要Python3.7版本。

HTTPServer问题的资料: https://www.zybuluo.com/JunQiu/note/1350528


Related

More

melo23 有JavaScript 代码吗?

melo23 比如币安 永续合约 开单之后 ,怎么设置 止损呢?想 开单价+2刀设置止损,这个怎么弄?

小小梦 这个只能Python

小小梦 那是TV图表上 交易信号的问题了,这个机器人仅仅是执行交易信号,不考虑交易策略的。你可以理解成你的TV图表上的WebHook信号是交易命令发布者,FMZ的机器人仅仅是命令执行者。至于你什么时候开仓,什么时候止损、止盈,那是要在TV图表上设置的。