Recently, more and more TradingView users have connected the TradingView chart signal to FMZ platform (FMZ.COM) and let the robot strategy on FMZ execute the transaction according to the chart signal, which saves a lot of code writing and design work. Directly, indicators can be used for programmatic and automated trading, which reduces the barriers for many programmatic and quantitative trading development. There are several design schemes for realizing automatic trading on TradingViewWebHook.
The previous solution: https://www.fmz.com/bbs-topic/9817.
The previous plan was to extend the API interface of FMZ Platform to send instructions to the robot. Today, let’s take a look at another solution. Let TradingView’s alarm WebHook request be sent directly to the FMZ platform robot, so that it can directly send instructions and order robot transactions.
Robot strategy source code
The strategy is written in Python. After the robot is created and started using this strategy, the robot will create a thread, which will start a service to monitor the set port. Waiting for external requests and processing. When I tested it, it was tested by the host on the server, and the device where the host is located must be accessible from the outside. When the robot executes the transaction, it uses the market order interface. in addition, this strategy can also be modified to implement the limit order order logic. In order to be easy to understand and streamlined, the market order is used here, so the exchange must support the market order.
'''
Request format: http://x.x.x.x:xxxx/data?access_key=xxx&secret_key=yyy&type=buy&amount=0.001
Strategy robot parameters:
- Type: Encrypted string, AccessKey, SecretKey, you can use the low-privileged API KEY of the FMZ platform, or you can generate the KEY yourself.
- Type: string, contract ID, ContractType
- Type: numeric value, port number, 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)
# check
if len(dictParam) == 4 and dictParam["access_key"] == AccessKey and dictParam["secret_key"] == SecretKey:
del dictParam["access_key"]
del dictParam["secret_key"]
Log("Request received", "parameter:", 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 "No futures contract set"
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("Holding Position:", exchange.GetPosition())
elif not isSpot and dictParam["type"] == "short":
exchange.SetDirection("sell")
exchange.Sell(-1, float(dictParam["amount"]))
Log("Holding Position:", exchange.GetPosition())
elif not isSpot and dictParam["type"] == "cover_long":
exchange.SetDirection("closebuy")
exchange.Sell(-1, float(dictParam["amount"]))
Log("Holding Position:", exchange.GetPosition())
elif not isSpot and dictParam["type"] == "cover_short":
exchange.SetDirection("closesell")
exchange.Buy(-1, float(dictParam["amount"]))
Log("Holding Position:", exchange.GetPosition())
# Write data response
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():
# Start a thread
try:
_thread.start_new_thread(createServer, (("0.0.0.0", Port), )) # Test on VPS server
except Exception as e:
Log("Error message:", e)
raise Exception("stop")
Log("Account asset information:", _C(exchange.GetAccount))
while True:
LogStatus(_D())
Sleep(2000)
Strategy parameters:
TradingView's WebHook alarm request
The alarm request setting is:
http://xxx.xxx.xxx.xxx:80/data?access_key=e3809e173e23004821a9bfb6a468e308&secret_key=45a811e0009d91ad21154e79d4074bc6&type=sell&amount=0.1
Since TradingView sends POST requests, the monitoring service must monitor POST requests, and TradingView only allows port 80 for the http protocol.
xxx.xxx.xxx.xxxis the device IP address of the host where the robot is located. Fill in the specific IP address of your own device, you need to be aware that it must be accessible from the external network.- The
access_keyandsecret_keycan be generated by themselves, as long as theaccess_keyandsecret_keyin theWebHookalarm request are the same as those configured on the robot parameters. Type, trading direction, buying or selling, opening or closing, note that spots and futures are distinguished. If it is a futures, note that the futures contract code must be set on the robot parameters, and the configured exchange object needs to be a futures exchange.amount, the number of transactions.
Running Test
Use wexApp to simulate the real market test.
END
Full strategy address: https://www.fmz.com/strategy/221850
The access_key and secret_key in the scheme are only for identification, and there is no security for using http. This solution is just an idea and an introduction. In practical applications, security considerations should be added and https communication should be used.