Aplicação da função __Thread no design de estratégia JavaScript

No design da estratégia FMZ original, se você precisar usar operações simultâneas assíncronas, você só poderá usarexchange.Go()A função é usada para implementar a simultaneidade da interface de encapsulamento FMZ, e algumas operações personalizadas (funções) não podem ser executadas simultaneamente. Embora esse design melhore muito a eficiência da execução do programa de políticas, ele ainda é muito desconhecido para estudantes que têm experiência em design simultâneo em linguagens de programação nativas.

Mesmo alguns novos alunos que são novos na negociação quantitativa usando FMZ não entendemexchange.Go()Uso de funções, usoexchange.Go()Ainda parece que as instruções são executadas uma por uma no código executado sequencialmente. Neste artigo, vamos explorar os novos recursos de threading simultâneo da plataforma FMZ:__Thread()O uso de uma série de funções e estratégia de design assíncrono do programa.

1. Design concorrente simples

Se quisermos executar um thread filho simultaneamente para executar uma função personalizada que escrevemos enquanto o thread principal da estratégia estiver em execução, podemos usar um design semelhante ao código a seguir. Personalize uma função no código de estratégiaGetTickerAsync(), escreva a funcionalidade específica desta função. Esta função executa um loop infinito.whileA interface da API FMZ é chamada continuamente em um loop:GetTicker()Para obter dados de mercado.

Então use__threadSetData(0, "ticker", t)Esta frase escreve um dado no thread principal. O nome do dado éticker, o valor dos dados étAgora mesmoGetTicker()O valor de retorno de .

__threadSetData(0, "ticker", t)

Depois de projetar a função personalizada para execução simultânea de threads, podemos escrevermain()O código na função émain()No início da função, usamos:

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

Crie um thread simultâneo, que começa a executarGetTickerAsync()função. entãomain()A função começa a executar seuwhileLoop, receber em loopGetTickerAsync()A função atualiza os dados e então imprime:

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

Exemplo de código completo:

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)
    }
}

Teste de operação de disco real:

Este é o design de aplicação mais simples. Em seguida, vamos dar uma olhada em alguns outros designs de demanda.

2. Design de ordem simultânea

Uma função pode ser projetada para criar 10 threads ao mesmo tempo, e cada thread executa uma função de operação de ordem. existirmain()Projetar uma funçãowhileLoop, instruções de interação de estratégia de detecção. Receba instruções interativas:placeMultipleOrdersBasta chamar esta função de ordem simultâneatestPlaceMultipleOrders()。

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

Adicione o design de interação de estratégia na página de edição de estratégia e defina um botão com o comando: placeMultipleOrders

Exemplo de código completo:

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)
    }
}

• O teste usa o método de ordem pendente, aumentando de 80% para 90% do preço atual, usando um teste de ambiente de disco simulado e clicando no botão interativo para acionar uma ordem de teste:

• Após clicar no botão “placeMultipleOrders”, a mensagem de prompt é:

• O log de estratégia mostra operações de ordens simultâneas:

3. Crie uma conexão WebSocket em uma função de execução de thread simultânea

Essa exigência foi levantada por um usuário do FMZ que queria um exemplo simples para demonstrar como usá-lo em threads simultâneos.WebSocketConecte e projete como passar dados para o thread principalmain()função.

Na verdade, é muito simples e é semelhante à criação de threads simultâneos no exemplo anterior. Apenas a comunicação entre threads é usada__threadPeekMessage()Funções e__threadPostMessage()função. Tomando a chamada de interface da API WebSocket da Binance Exchange como exemplo, também precisamos prestar atenção à operação de fechamento da conexão WebSocket no design. O exemplo a seguir também mostra como notificar um thread concorrente para pará-lo.

Exemplo de código completo:

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("扫尾函数执行完毕")
}

Teste de operação de disco real:

Você pode vermain()A função recebe continuamente dados de mercado da conexão WebSocket criada pelo thread simultâneo.

Quando a estratégia for interrompida, a função de varredura começará a funcionar: