Aplicación de la función __Thread en el diseño de estrategias de JavaScript

En el diseño de estrategia FMZ original, si necesita utilizar operaciones concurrentes asincrónicas, solo puede utilizarexchange.Go()La función se utiliza para implementar la concurrencia de la interfaz de encapsulación FMZ, y algunas operaciones personalizadas (funciones) no se pueden ejecutar simultáneamente. Aunque este diseño mejora enormemente la eficiencia de la ejecución del programa de políticas, todavía resulta muy desconocido para los estudiantes que tienen experiencia en diseño concurrente en lenguajes de programación nativos.

Incluso algunos estudiantes nuevos que son nuevos en el comercio cuantitativo utilizando FMZ no comprendenexchange.Go()Uso de funciones, usoexchange.Go()Todavía parece que las declaraciones se ejecutan una por una en el código ejecutado secuencialmente. En este artículo, exploraremos las nuevas funciones de subprocesamiento simultáneo de la plataforma FMZ:__Thread()La utilización de una serie de funciones y estrategias de diseño asincrónico del programa.

1. Diseño concurrente simple

Si queremos ejecutar un hilo secundario simultáneamente para ejecutar una función personalizada que escribimos mientras se ejecuta el hilo principal de la estrategia, podemos usar un diseño similar al siguiente código. Personalizar una función en el código de estrategiaGetTickerAsync(), escribe la funcionalidad específica de esta función. Esta función ejecuta un bucle infinito.whileLa interfaz API FMZ se llama continuamente en un bucle:GetTicker()Para obtener datos del mercado.

Luego usa__threadSetData(0, "ticker", t)Esta oración escribe datos en el hilo principal. El nombre de los datos esticker, el valor de los datos estAhora mismoGetTicker()El valor de retorno de .

__threadSetData(0, "ticker", t)

Después de diseñar la función personalizada para la ejecución simultánea de subprocesos, podemos escribirmain()El código en la función esmain()Al comienzo de la función, utilizamos:

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

Crea un hilo simultáneo, que comienza a ejecutarseGetTickerAsync()función. entoncesmain()La función comienza a ejecutar suwhileBucle, recibir en bucleGetTickerAsync()La función actualiza los datos y luego imprime:

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

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

Prueba de funcionamiento del disco real:

Este es el diseño de aplicación más simple. A continuación, veamos otros diseños de demanda.

2. Diseño de órdenes concurrentes

Se puede diseñar una función para crear 10 subprocesos al mismo tiempo, y cada subproceso ejecuta una función de operación de orden. existirmain()Diseñar una funciónwhileInstrucciones de interacción de estrategia de detección de bucle. Recibir instrucciones interactivas:placeMultipleOrdersSimplemente llame a esta función de orden concurrentetestPlaceMultipleOrders()。

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

Agregue un diseño de interacción de estrategia en la página de edición de estrategia y configure un botón con el comando: placeMultipleOrders

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

• La prueba utiliza el método de orden pendiente, aumentando del 80% al 90% del precio actual, utilizando una prueba de entorno de disco simulado y haciendo clic en el botón interactivo para activar una orden de prueba:

• Después de hacer clic en el botón “placeMultipleOrders”, el mensaje de aviso es:

• El registro de estrategia muestra operaciones de órdenes simultáneas:

3. Cree una conexión WebSocket en una función de ejecución de subproceso concurrente

Este requisito fue planteado por un usuario de FMZ que quería un ejemplo simple para demostrar cómo usarlo en subprocesos simultáneos.WebSocketConectar y diseñar cómo pasar datos al hilo principalmain()función.

En realidad es muy sencillo y es similar a la creación de subprocesos simultáneos en el ejemplo anterior. Solo se utiliza la comunicación entre hilos.__threadPeekMessage()Funciones y__threadPostMessage()función. Si tomamos como ejemplo la llamada a la interfaz API de WebSocket de Binance Exchange, también debemos prestar atención a la operación de cierre de la conexión WebSocket en el diseño. El siguiente ejemplo también muestra cómo notificar a un hilo concurrente para que lo detenga.

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

Prueba de funcionamiento del disco real:

Puedes vermain()La función recibe continuamente datos de mercado desde la conexión WebSocket creada por el hilo simultáneo.

Cuando se detiene la estrategia, la función de barrido comenzará a funcionar: