前几日受到《利用 tradingview 指标对接发明者实盘机器人》文章启发，自己也尝试复现了一下。但作者写的比较粗略，也略晦涩难懂。
所以我使用了Top 10 Python libraries of 2019 的3个库重构了一份server。
注意 tradingview webhook功能为会员功能，需要购买会员！
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建立一个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)
接下去我们把机器人运行起来
这里要非常注意 机器人ID要填写到python代码中
第一步
新建一个python文件。起个名字 fz1.py
第二步
编写导入模块
import time
import hashlib
import httpx
from fastapi import FastAPI
from pydantic import BaseModel
第三步
在fmz主页注册api
并找到托管主机ID
并定义到 python代码中。
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。
以上就是全部核心代码的实现。
我略微处理了一下异常处理模块以及重试机制。将所有代码整合。
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)
记住要讲文件保存成fz1的文件名
接着我们打开命令行工具。启动python服务端
命令如下：
uvicorn fz1:webhook --host 0.0.0.0 --port 8001 --reload
其中代表使用fz1的py文件并调用webhook任务。
当你看到如图所示，Application startup complete就代表启动成功了
这时候你可以打开浏览器，访问 http://127.0.0.1:8001/redoc
fast api 已经为你实现了api接口文档，和自动调试器。
我们使用的是chrome 插件 TALEND API TESTER
你也可以用其他的http调试工具完成。
模拟本地tradingview发送webhook消息。
将模式调整为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
完成以后点击发送
如图所示。本地测试已经完成了。
我们从本地模拟发送到webhook的数据通过python 传递到托管设备，并顺利获取到了相关数据。
本段落部分属于个人网络环境，每个人不一样。
这些步骤主要是允许 tradingview 访问我的外网ip，
并指定80端口重定向到本地设备的8001端口。
将设备规则执行以后。tradingview就能顺利推送消息给我们了。
我们打开tradingview 并在一个高级指标上新建一个报警，
这个方法可以用于tradingview各种高级指标自动下单。
勾选webhook，并填写服务端地址http://xxx.xxx.xxx.xxx/tv/
下方消息框填入字段消息。
{"apikey": "FMZ", "exchange": "bybit", "symbol": "BTC-USD", "side": "buy", "amount": "100"}
如图所示 在指标上已经显示了我们的预警信息
当价格穿越指标以后，tradingview产生报警
