Loading ...

关于FMZ对接Tradingview下单的方法

Author: homily, Created: 2020-04-04 00:28:01, Updated: 2020-04-22 13:28:55

灵感由来

前几日受到《利用 tradingview 指标对接发明者实盘机器人》文章启发,自己也尝试复现了一下。但作者写的比较粗略,也略晦涩难懂。 所以我使用了Top 10 Python libraries of 2019 的3个库重构了一份server。

注意 tradingview webhook功能为会员功能,需要购买会员!

本系统webhook特性如下:

  • 1、标准化json消息传递
  • 2、加入try…except…异常处理
  • 3、加入with…as…内存回收
  • 4、采用async异步
  • 5、采用ASGI协议
  • 6、采用AUTO HTTP1.X-2.0协议
  • 7、代码经过 100% test coverage
  • 8、装饰器的运用使程序简洁、执行效率高、稳定可靠

一、准备工作

  1. 您需要安装 python 3.8.2 或以上版本 https://www.python.org/downloads/
  2. 您需要安装pip包管理器 https://pip.pypa.io/en/stable/
  3. 安装必要的包
   pip install httpx
   pip install starlette
   pip install pydantic
   pip install fastapi
   pip install uvicorn

HTTPX

是一个功能齐全的http访问库。获2019 top 10 第一名。 HTTPX is a fully featured HTTP client for Python 3, which provides sync and async APIs, and support for both HTTP/1.1 and HTTP/2. https://www.python-httpx.org/

Starlette

Starlette 是一个轻量级ASGI标准库/工具包。获 top 10 第二名。 Starlette is a lightweight ASGI framework/toolkit, which is ideal for building high performance asyncio services. https://www.uvicorn.org/

FastAPI

FastAPI 是一个基于Starlette的API快速实现库。获 top 10 第三名。 FastAPI framework, high performance, easy to learn, fast to code, ready for production https://fastapi.tiangolo.com/

Uvicorn

Uvicorn 是一个轻量级 ASGI 服务端。 Uvicorn is a lightning-fast ASGI server, built on uvloop and httptools. https://www.uvicorn.org/

pydantic

pydantic 是一个基于python的数据格式管理库,已用于Microsoft、Amazon Web Services、Uber Data validation and settings management using python type annotations. https://pydantic-docs.helpmanual.io/

题外话: 其实这些库已经代表了当前行业的最高标准!!。. 你还能用他们写很多其他的东西。 我已经用其对接了5大交易所,监控资金流向。okex的检测频率甚至稳定在0.1秒。。 或者说…可以搭建一套千万级并发的交易所…

人生苦短,我用python。 一行代码能解决的问题,绝不用一堆。 请往后看!

原理及实现

我们要做的事情很简单,就是接收tradingview post过来的数据再编码发送http get传递给FMZ的托管者。 有点类似中间件服务。

FMZ终端接收器

我们要去fmz建立一个python策略 具体代码如下:

import json
import time
def main():
    while True:
        cmd = GetCommand() #接收webhook推送的数据
        if cmd:
            Log(cmd) #log全部json 数据
            text=eval(cmd) #拆分json数据
            Log('apikey=',text['apikey']) #分别输出apikey
            Log('exchange=',text['exchange']) #分别输出exhange
            Log('symbol=',text['symbol']) #分别输出symbol
            Log('side=',text['side']) #分别输出side
            Log('amount=',text['amount']) #分别输出amount
        Sleep(500)

img 接下去我们把机器人运行起来

img

这里要非常注意 机器人ID要填写到python代码中

webhook server 服务端搭建

第一步 新建一个python文件。起个名字 fz1.py

第二步 编写导入模块

import time
import hashlib
import httpx
from fastapi import FastAPI
from pydantic import BaseModel

第三步 在fmz主页注册api 并找到托管主机ID 并定义到 python代码中。 img

accessKey = '1XXXXXXXXXXX34XXXXXXXXXXXXXXXXbb'
secretKey = 'a1XXXXXXXa88fXXXXXXXXXXXXXXX5870'
RobotId   = 18XXX0

这里的RobotID需要在前面的托管策略中获得ID

第四步 定义一个api函数来提交CMD指令给fmz托管。

async def api(CMD):
    
    params = {
        'version': '1.0',
        'access_key': accessKey,
        'method': 'CommandRobot',
        'args': f'[{RobotId},"{CMD}"]',
        'nonce': int(time.time() * 1000),
        }
    #先按照fmz api手册规定的参数,定义http访问参数params
    
    params['sign'] = hashlib.md5(f"{params['version']}|{params['method']}|{params['args']}|{params['nonce']}|{secretKey}".encode('utf-8')).hexdigest()
    #对所有参数进行md5加密处理生成sign字段。
    
    async with httpx.AsyncClient() as client:  
    #使用with..as..模式建立内存回收机制的httpx 异步 client
        response = await client.get('https://www.fmz.com/api/v1',params=params) 
        #使用get方法请求fmz api
    return(response.json())  
    #httpx直接调用json格式返回

第四步 自定义数据类型(json格式) 使用pydantic模块的BaseModel方法来定义一个类型 当然你也可以自己定义更多的字段,这完全取决于你自己想要达到的效果。

class Item(BaseModel):  
    apikey: str         #接受apikey 用于验证发送数据方
    exchange: str       #交易所字段
    symbol:str          #交易品种字段
    side:str            #交易方向字段
    amount: int         #交易量

以上全部是非空字段,传参如果缺少一个会报错, 如果需要传递可空参数,请参阅BaseModel手册。

第五步 编写Tradingview接收模块(webhook server) webhook 的协议使用的是http post方法。 所以我们需要定义一个http server端并接受post传递过来的数据。

webhook = FastAPI()  #语句来定义服务端

@webhook.post("/tv/") #语法糖创建post方法接口
async def create_item(item: Item): #创建消息机制,调用自定义数据类型
    print(item)#输出接收的消息
    item_dict = item.dict() #将数据类型转换字典
    await api(item_dict) #将字典数据交给传递给api函数,并推送到fmz。

python服务端全部代码

以上就是全部核心代码的实现。 我略微处理了一下异常处理模块以及重试机制。将所有代码整合。

import time
import hashlib
import httpx
from fastapi import FastAPI
from pydantic import BaseModel

accessKey = '1XXXXXXXXXXX34XXXXXXXXXXXXXXXXbb'
secretKey = 'a1XXXXXXXa88fXXXXXXXXXXXXXXX5870'
RobotId   = 18XXX0

async def api(CMD):
    #http params
    params = {
        'version': '1.0',
        'access_key': accessKey,
        'method': 'CommandRobot',
        'args': f'[{RobotId},"{CMD}"]',
        'nonce': int(time.time() * 1000),
        }

    params['sign'] = hashlib.md5(f"{params['version']}|{params['method']}|{params['args']}|{params['nonce']}|{secretKey}".encode('utf-8')).hexdigest()

    async with httpx.AsyncClient() as client:
        for i in range(5):
            try:
                response = await client.get('https://www.fmz.com/api/v1',params=params)
                assert response.status_code == 200
                break
            except:
                print(f'ERR!  connect "www.fmz.com" filed , again times {i+1}!')
                await asyncio.sleep(2)
                continue
        return(response.json())


class Item(BaseModel):
    apikey: str
    exchange: str 
    symbol:str
    side:str
    amount: int 

webhook = FastAPI() #new webhook server

@webhook.post("/tv/") #set post url root
async def create_item(item: Item): #get webhook item
    print(item)
    if (item.apikey=='FMZ'):
        item_dict = item.dict()
        ret=await api(item_dict)
        if(ret['code']==0 and ret['data']['result']==True ):
            print ({'return':'ok'})
        else:
            print({'return':'error'})
            print(ret)

img

服务端的启动

记住要讲文件保存成fz1的文件名 接着我们打开命令行工具。启动python服务端 命令如下:

uvicorn fz1:webhook --host 0.0.0.0 --port 8001 --reload

其中代表使用fz1的py文件并调用webhook任务。

  • host 表示接收0.0.0.0的ip访问,也就是全部。
  • port 表示将端口设定为8001 (这里需要注意的是,tradingview只通过80端口访问,你需要在防火墙做好端口映射,或者直接使用80)
  • reload 表示如果py主文件有迭代,服务端自动重载最新版本。

img

当你看到如图所示,Application startup complete就代表启动成功了

这时候你可以打开浏览器,访问 http://127.0.0.1:8001/redoc

img

fast api 已经为你实现了api接口文档,和自动调试器。

本地调试

我们使用的是chrome 插件 TALEND API TESTER 你也可以用其他的http调试工具完成。 模拟本地tradingview发送webhook消息。 img 将模式调整为post url地址是 127.0.0.1:8001/tv/ post的body字段写上json

  "apikey": "FMZ",
  "exchange":"okex",
  "symbol": "BTC-USD-200626",
  "side": "buy", 
  "amount": 1
}

同时在HEADER中会自动生成application/json

完成以后点击发送

img img img img

如图所示。本地测试已经完成了。 我们从本地模拟发送到webhook的数据通过python 传递到托管设备,并顺利获取到了相关数据。

将webhook部署到外网

本段落部分属于个人网络环境,每个人不一样。

  1. 登陆Juniper 终端 img
  2. 增加rule-set outside-to-inside
  3. 指定链路走向 from untruset to trust
  4. 增加tradingview 源地址IP断 source-address [ 58.35.162.147/32 52.89.214.238/32 34.212.75.30/32 54.218.53.128/32 52.32.178.7/32 ];
  5. 指定目的地址(本机外网ip) destination-address xxx.xxx.xxx.xxx/32;
  6. 指定目的端口(本机外网port) destination-port 80;
  7. 做端口重定向 destination-nat pool inside-8001; img

这些步骤主要是允许 tradingview 访问我的外网ip, 并指定80端口重定向到本地设备的8001端口。

将设备规则执行以后。tradingview就能顺利推送消息给我们了。

tradingvew 消息测试

我们打开tradingview 并在一个高级指标上新建一个报警, 这个方法可以用于tradingview各种高级指标自动下单。

img

勾选webhook,并填写服务端地址http://xxx.xxx.xxx.xxx/tv/ 下方消息框填入字段消息。 {“apikey”: “FMZ”, “exchange”: “bybit”, “symbol”: “BTC-USD”, “side”: “buy”, “amount”: “100”} img img 如图所示 在指标上已经显示了我们的预警信息

当价格穿越指标以后,tradingview产生报警 img img img img

特别鸣谢《利用 tradingview 指标对接发明者实盘机器人》 扁豆子

特别鸣谢 币旋风社区 燃在顺势 技术支持

特别鸣谢 武汉易物云 大吴老师 技术支持


Related

More

aliwithtaozi import asyncio ?

Peppa 官方已经支持直接生成url对接tradingview的webhook来给机器人发指令 https://www.fmz.com/api#%E7%9B%B4%E6%8E%A5%E9%AA%8C%E8%AF%81

小草 写的不错,实现起来也很简洁。要不官方直接支持了吧,目前账户里添加的webhook好像只能用于推送消息

homily 啊?

轻轻的云 不知道为啥,官方这个延迟差不多得1分钟左右 脚本才能出日志下单。。。。。

轻轻的云 不知道为啥,官方这个延迟差不多得1分钟左右 脚本才能出日志下单。。。。。

congcong009 同上,复议

shawnQ 同上