最近では,TradingViewのチャートシグナルを FMZプラットフォームに接続しているユーザーも増えています (FMZ.COM直接,インディケーターは,多くのプログラムおよび定量的な取引開発の障壁を軽減する,プログラムおよび自動取引に使用することができます. 自動取引を実現するためのいくつかの設計スキームがあります.TradingViewWebHook.

前回の解決法:https://www.fmz.com/digest-topic/5533.

前回の計画では,FMZプラットフォームのAPIインターフェースを拡張してロボットに指示を送信することでした.今日,別の解決策を見てみましょう. TradingViewのアラーム WebHook リクエストを直接FMZプラットフォームロボットに送信し,直接指示を送信し,ロボット取引を注文することができます.

ロボット戦略 ソースコード

この戦略は Python で書かれています. ロボットが作成され,この戦略を使用し始めた後,ロボットはスレッドを作成し,設定ポートを監視するサービスを起動します. 外部の要求と処理を待っています. 私がテストしたとき,それはサーバー上のホストによってテストされ,ホストが位置するデバイスは外部からアクセス可能でなければなりません. ロボットが取引を実行するとき,それはマーケットオーダーインターフェイスを使用します. さらに,この戦略は,市場オーダーインターフェースを実装するために変更することもできます.limit 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)

戦略パラメータ:

トレーディングビューのWebHookアラーム要求

アラーム要求の設定は:

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

取引視野が送信するのでPOST監視サービスが監視しなければならない.POSTポート80を許可する.http protocol.

• xxx.xxx.xxx.xxxロボットが位置するホストのデバイスIPアドレスです. 特定のIPアドレスを入力してください. デバイスは外部ネットワークからアクセス可能である必要があります.

• についてaccess_keyそしてsecret_key単体で生成できる限り,access_keyそしてsecret_keyについてWebHookロボットのパラメータに設定されているものと同じです

• Type,取引方向,購入または販売,開封または閉鎖,スポットと先物との区別が注意してください. 将来の場合は,先物契約コードがロボットパラメータに設定され,設定された交換オブジェクトが先物取引である必要があることに注意してください.

• amount, 取引数

実行テスト

使用wexApp実際の市場テストをシミュレートするために

エンド

戦略の完全な住所:https://www.fmz.com/strategy/221850

についてaccess_keyそしてsecret_key制度は識別のみで,使用の保証はありませんhttpこの解決策は,単なるアイデアと導入に過ぎません.実用的な応用では,セキュリティの考慮が加えられ,httpsコミュニケーションを活用すべきです