Application de la fonction __Thread dans la conception de stratégies JavaScript

Dans la conception de la stratégie FMZ d’origine, si vous devez utiliser des opérations simultanées asynchrones, vous ne pouvez utiliser queexchange.Go()La fonction est utilisée pour implémenter la concurrence de l’interface d’encapsulation FMZ, et certaines opérations personnalisées (fonctions) ne peuvent pas être exécutées simultanément. Bien que cette conception améliore considérablement l’efficacité de l’exécution des programmes de politique, elle est encore très peu familière aux étudiants qui ont de l’expérience en conception simultanée dans les langages de programmation natifs.

Même certains nouveaux étudiants qui débutent dans le trading quantitatif en utilisant FMZ ne comprennent pasexchange.Go()Utilisation des fonctions, utilisationexchange.Go()Il semble toujours que les instructions soient exécutées une par une dans le code exécuté séquentiellement. Dans cet article, explorons les nouvelles fonctionnalités de threading simultané de la plateforme FMZ :__Thread()L’utilisation d’une série de fonctions et de stratégies de conception asynchrone du programme.

1. Conception simultanée simple

Si nous voulons exécuter un thread enfant simultanément pour exécuter une fonction personnalisée que nous avons écrite pendant que le thread principal de la stratégie est en cours d’exécution, nous pouvons utiliser une conception similaire au code suivant. Personnaliser une fonction dans le code de stratégieGetTickerAsync(), écrivez la fonctionnalité spécifique de cette fonction. Cette fonction exécute une boucle infinie.whileL’interface API FMZ est appelée en continu dans une boucle :GetTicker()Pour obtenir des données de marché.

Ensuite, utilisez__threadSetData(0, "ticker", t)Cette phrase écrit une donnée dans le thread principal. Le nom de la donnée estticker, la valeur des données esttTout de suiteGetTicker()La valeur de retour de .

__threadSetData(0, "ticker", t)

Après avoir conçu la fonction personnalisée pour l’exécution simultanée de threads, nous pouvons écriremain()Le code dans la fonction estmain()Au début de la fonction, nous utilisons :

__Thread(GetTickerAsync, 0)   // GetTickerAsync为需要并发执行的自定义函数，0为这个传入GetTickerAsync函数的参数

Créer un thread simultané, qui commence à s’exécuterGetTickerAsync()fonction. alorsmain()La fonction commence à exécuter sonwhileBoucle, recevoir en boucleGetTickerAsync()La fonction met à jour les données puis imprime :

var t = __threadGetData(0, "ticker")
Log(t)

Exemple de code complet :

function GetTickerAsync(index) {
    while (true) {
        var t = exchanges[index].GetTicker()
        __threadSetData(0, "ticker", t)
        Sleep(500)
    }
}

function main() {
    __Thread(GetTickerAsync, 0)

    while(true) {
        var t = __threadGetData(0, "ticker")
        Log(t)
        Sleep(1000)
    }
}

Test de fonctionnement du disque réel :

Il s’agit de la conception d’application la plus simple. Voyons maintenant d’autres conceptions de demande.

2. Conception de commandes simultanées

Une fonction peut être conçue pour créer 10 threads en même temps, et chaque thread exécute une fonction d’opération de commande. existermain()Concevoir une fonctionwhileBoucle, instructions d’interaction de stratégie de détection. Recevez des instructions interactives :placeMultipleOrdersAppelez simplement cette fonction de commande simultanéetestPlaceMultipleOrders()。

if (cmd == "placeMultipleOrders") {
    // ...
}

Ajoutez une conception d’interaction de stratégie sur la page d’édition de stratégie et définissez un bouton avec la commande : placeMultipleOrders

Exemple de code complet :

function placeOrder(exIndex, type, price, amount) {
    var id = null 
    if (type == "Buy") {
        id = exchanges[exIndex].Buy(price, amount)
    } else if (type == "Sell") {
        id = exchanges[exIndex].Sell(price, amount)
    } else {
        throw "type error! type:" + type
    }
}

function testPlaceMultipleOrders(index, beginPrice, endPrice, step, type, amount) {
    Log("beginPrice:", beginPrice, ", endPrice:", endPrice, ", step:", step, ", type:", type, ", amount:", amount)
    var tids = []
    for (var p = beginPrice; p <= endPrice; p += step) {
        var tid = __Thread(placeOrder, index, type, p, amount)
        tids.push(tid)
        Sleep(10)
    }
    Sleep(1000)
    for (var i = 0; i < tids.length; i++) {
        __threadTerminate(tids[i])
    }
}

function main() {
    while(true) {
        LogStatus(_D())
        var cmd = GetCommand()
        if (cmd) {
            if (cmd == "placeMultipleOrders") {
                var t = _C(exchange.GetTicker)
                var beginPrice = t.Last * 0.8
                var endPrice = t.Last * 0.9
                var step = t.Last * 0.01
                testPlaceMultipleOrders(0, beginPrice, endPrice, step, "Buy", 0.01)
                var orders = exchange.GetOrders()
                for (var i = 0; i < orders.length; i++) {
                    Log(orders[i])
                }
            }
        }
        Sleep(1000)
    }
}

• Le test utilise la méthode de commande en attente, augmentant de 80 % à 90 % du prix actuel, en utilisant un test d’environnement de disque simulé et en cliquant sur le bouton interactif pour déclencher une commande de test :

• Après avoir cliqué sur le bouton « placeMultipleOrders », le message d’invite est :

• Le journal de stratégie affiche les opérations de commande simultanées :

3. Créer une connexion WebSocket dans une fonction d’exécution de thread simultanée

Cette exigence a été soulevée par un utilisateur de FMZ qui souhaitait un exemple simple pour démontrer comment l’utiliser dans des threads simultanés.WebSocketConnectez-vous et concevez comment transmettre des données au thread principalmain()fonction.

C’est en fait très simple et similaire à la création de threads simultanés dans l’exemple précédent. Seule la communication entre les threads est utilisée__threadPeekMessage()Fonctions et__threadPostMessage()fonction. En prenant comme exemple l’appel d’interface API WebSocket de Binance Exchange, nous devons également prêter attention à l’opération de fermeture de la connexion WebSocket dans la conception. L’exemple suivant montre également comment notifier un thread concurrent de l’arrêter.

Exemple de code complet :

var tid = null 

function createWS() {
    // wss://stream.binance.com:9443/ws/<streamName> , <symbol>@ticker
    
    var stream = "wss://stream.binance.com:9443/ws/btcusdt@ticker"    
    var ws = Dial(stream)
    Log("创建WS连接:", stream)
    
    while (true) {
        var data = ws.read()
        if (data) {            
            __threadPostMessage(0, data)
        }
        Log("接收到WS链接推送的数据，data:", data)
        
        // __threadPeekMessage 超时参数设置-1，不阻塞
        var msg = __threadPeekMessage(-1)
        if (msg) {
            if (msg == "stop") {
                Log("并发线程Id:", __threadId(), "接收到stop指令")
                break
            }
        }
    }

    Log("并发线程执行完毕，关闭ws连接")
    ws.close()
}

function main() {
    tid = __Thread(createWS)
    Log("创建并发线程，线程Id:", tid)

    while(true) {
        // __threadPeekMessage 的超时参数设置为0，阻塞等待数据
        var data = __threadPeekMessage(0)
        Log("接收到并发线程", ", Id:", tid, ", 发送的数据，data:", data, "#FF0000")
        
        var tbl = {
            type : "table", 
            title : "<symbol>@ticker频道推送消息",
            cols : ["事件类型", "事件时间", "交易对", "24小时价格变化", "24小时价格变化百分比", "平均价格", "最新成交价格", "24小时内成交量", "24小时内成交额"],
            rows : []
        }

        try {
            data = JSON.parse(data)
            tbl.rows.push([data.e, _D(data.E), data.s, data.p, data.P, data.w, data.c, data.v, data.q])
        } catch (e) {
            Log("e.name:", e.name, "e.stack:", e.stack, "e.message:", e.message)
        }
        LogStatus(_D(), "\n`" + JSON.stringify(tbl) + "`")
    }
}

function onexit() {
    Log("扫尾函数，向Id为", tid, "的并发线程发送stop指令")
    __threadPostMessage(tid, "stop")
    Log("等待Id为", tid, "的并发线程停止")
    __threadJoin(tid)
    Log("扫尾函数执行完毕")
}

Test de fonctionnement du disque réel :

Tu peux voirmain()La fonction reçoit en continu des données de marché à partir de la connexion WebSocket créée par le thread simultané.

Lorsque la stratégie est arrêtée, la fonction de balayage commence à fonctionner :