Una plantilla de políticas que te permite usar el sector WebSocket sin problemas

Esta es una plantilla de mercado de WebSocket desarrollada oficialmente por FMZ, que puede ser copiada y guardada como una plantilla que se puede usar si se selecciona en la nueva política:https://www.fmz.com/strategy/470349

¿Por qué se necesita un WebSocket?

La estrategia actual de FMZ se basa principalmente en el envase tradicional de REST API, donde se establece una conexión a la red con cada paso de acceso a la API para obtener datos de mercado a través de consultas. Este método es simple y fácil de usar, y para la mayoría de las necesidades, hacerlo es completamente suficiente.

Sin embargo, el protocolo REST tiene un problema de demora inherente, que se amplifica cuando se requieren múltiples pares de transacciones o estrategias de múltiples intercambios. Aunque la función Go de la plataforma se puede ejecutar en paralelo, el problema de demora sigue existiendo, es difícil de satisfacer las necesidades de transacciones estratégicas de una frecuencia relativamente alta, y se tratan demasiadas transacciones, la frecuencia de consulta es demasiado rápida y se enfrenta a la limitación de la frecuencia de acceso a la plataforma de negociación.

La carga del servidor de los intercambios actuales es también muy pesada, ya que todos ofrecen protocolos WebSocket completos y se recomienda a los usuarios de las APIs utilizarlos. Comparado con el protocolo REST, el WebSocket ofrece una forma de conexión bidireccional duradera que permite a los intercambios enviar datos a los clientes en tiempo real, evitando solicitudes y respuestas frecuentes, lo que reduce enormemente la demora.

El modelo de mercado de WebSocket

La plataforma de comercio cuantificado de FMZ apoyó el protocolo WebSocket desde el principio y era relativamente fácil de llamar, pero era demasiado complejo para los usuarios novatos procesar múltiples suscripciones, suscribir múltiples mercados de intercambio y integrarse de manera eficiente y fácil en todo el proceso de estrategia. Esta plantilla de aceleración de datos de mercado en tiempo real de WebSocket, abierta al público, resuelve este problema, es muy fácil de usar, es completamente compatible con las llamadas de la API actualmente empaquetadas, y para la mayoría de las políticas de REST originales, es simple cambiarlas directamente para acelerar su política.

Las características principales:

• Apoyados por varias bolsasEsta política permite la conexión de WebSocket de varios intercambios, como Bitcoin, OKX, ByBit, Bitget, etc. Los usuarios pueden imitar el método de envasado de esta plantilla para soportar más intercambios.
• Suscripción personalizadaPermite suscribirse a canales de mercado específicos (como profundidad, transacciones, etc.) y procesar los datos recibidos de manera eficiente para su uso inmediato en la estrategia de transacción.
• El tratamiento de errores de alto nivelLos mecanismos de seguimiento de errores y de reconexión de WebSocket están integrados para garantizar la confiabilidad y continuidad del flujo de datos.

Una simple introducción a la implementación

Tenga en cuenta que esta estrategia utiliza TypeScript, que puede parecer un poco extraño si solo está familiarizado con JavaScript. TypeScript introduce un sistema de tipos y características de lenguaje más ricas basadas en JavaScript, para aplicaciones que requieren procesar lógica compleja como transacciones cuantitativas. El uso de TypeScript reduce los errores potenciales y mejora la legibilidad y la conservabilidad del código.

Además, la política utiliza el mecanismo de sincronización de la plataforma FMZ, donde los subdirectos pueden enviar mensajes al hilo principal a través de la función __threadPostMessage. Este método es sincronizado y se utiliza para notificar el hilo principal de la actualización de datos generados en los subdirectos. Se pueden compartir datos entre el hilo principal y los subdirectos a través de las funciones __threadGetData y __threadSetData. Este método permite que los hilos accedan y modifiquen el estado del intercambio.

El principal principio de esta estrategia es conectar a los principales intercambios de divisas digitales a través de WebSocket para recibir datos de mercado en tiempo real (como información profunda e información de transacción) para proporcionar soporte de datos para tomar decisiones de transacción cuantitativas.

1. Configuración de conexión de WebSocket

setupWebsocketFunción utilizada para iniciar una conexión WebSocket y recibir datos de mercado.main_exchangesEn el caso de las bolsas de divisas, las bolsas de divisas son las que necesitan conexión.

• MyDialFunción: crea una conexión WebSocket, registra el tiempo de conexión y produce el tiempo de cierre al cerrar la conexión.
• updateSymbolsFunción: regularmente comprobar si hay nuevas solicitudes de suscripción y actualizar la lista de transacciones actuales según sea necesario.

2. Procesamiento de datos

supportsLos objetos definen los intercambios que se apoyan y sus funciones de procesamiento (comoBinanceLa función de procesamiento de cada bolsa es responsable de analizar los mensajes recibidos y extraer los datos correspondientes.

• processMsgFunciónEl proceso de procesamiento de los mensajes de los intercambios identifica diferentes tipos de datos (como actualizaciones profundas, transacciones, etc.) y los formaliza en un objeto de evento unificado.

3. Datos de suscripción

En cada conexión, el sistema utiliza canales de datos de mercado relacionados con la suscripción en función de la transacción actual.

• getFunctionFunciónSe obtiene la función de procesamiento correspondiente según el nombre de la bolsa.
• this.wssPublicFunción: Inicializa la conexión WebSocket y inicia la recepción de datos.

4. Administración de los hilos

Se inicia un hilo para cada intercambio, recibe datos en tiempo real y los procesa mediante funciones de retorno.

• threadMarketFunción: Recibe datos, analiza y almacena la información de profundidad y transacción más reciente en sus subtemas.

5. Reescribir el método de obtención de datos

Para cada intercambio, reescriba el método de obtener información de profundidad y transacción, dando prioridad a los datos actualizados en tiempo real.

Cómo usar el modelo

1. IniciaciónUso:$.setupWebsocket()Iniciar la conexión WebSocket de la bolsa de destino.
2. SuscripciónEl sistema se suscribe automáticamente a los canales relacionados con la variedad que usted está negociando (por ejemplo, profundidad, transacción, etc.).
3. Obtención de datosPor teléfono:GetDepth()yGetTrades()Función que utiliza automáticamente datos WebSocket en tiempo real para la profundidad del mercado y la devolución de registros de transacciones.
4. El tratamiento incorrectoLa política incluye un mecanismo de seguimiento para registrar errores de conexión y datos y para intentar volver a conectar automáticamente cuando la conexión se interrumpe.

Si se incluye en la política la función EventLoop (), se convierte en un mecanismo de activación que se obtiene automáticamente inmediatamente cuando se actualizan los datos de wss, sin esperar a que no se actualicen los datos.

function main() {
    $.setupWebsocket()
    while (true) {
        exchanges.map(e=>{
            Log(e.GetName(), e.GetDepth())
            Log(e.GetName(), e.GetTrades())
        })
        EventLoop(100) // trigger by websocket
    }
}

Consulte mi anterior guía de estrategias para el comercio de monedas.https://www.fmz.com/digest-topic/10506En este caso, el sitio web de Google es el más fácil de usar para cambiar el soporte de WebSocket:

function MakeOrder() {
    for (let i in Info.trade_symbols) {
        let symbol = Info.trade_symbols[i];
        let buy_price = exchange.GetDepth(symbol + '_USDT').Asks[0].Price;
        let buy_amount = 50 / buy_price;
        if (Info.position[symbol].value < 2000){
            Trade(symbol, "buy", buy_price, buy_amount, symbol);
        }
    }
}

function OnTick() {
    try {
        UpdatePosition();
        MakeOrder();
        UpdateStatus();
    } catch (error) {
        Log("循环出错: " + error);
    }
}

function main() {
    $.setupWebsocket()
    InitInfo();
    while (true) {
        let loop_start_time = Date.now();
        if (Date.now() - Info.time.last_loop_time > Info.interval * 1000) {
            OnTick();
            Info.time.last_loop_time = Date.now();
            Info.time.loop_delay = Date.now() - loop_start_time;
        }
        Sleep(5);
    }
}

Cómo añadir nuevos intercambios por ti mismo

Si sigues la plantilla de la política, puedes hacer tu propia imitación del siguiente formato y consultar la documentación de la API de la bolsa:

    supports["Binance"] = function (ctx:ICtx) {
        let processMsg = function (obj) {
            let event = {
                ts: obj.E,
                instId: obj.s,
                depth: null,
                trades: [],
            }

            if (obj.e == "depthUpdate") {
                let depth = {
                    asks: [],
                    bids: []
                }
                obj.b.forEach(function (item) {
                    depth.bids.push({
                        price: Number(item[0]),
                        qty: Number(item[1])
                    })
                })
                obj.a.forEach(function (item) {
                    depth.asks.push({
                        price: Number(item[0]),
                        qty: Number(item[1])
                    })
                })
                event.depth = depth
            } else if (obj.e == 'bookTicker') {
                event.depth = {
                    asks: [{ price: Number(obj.a), qty: Number(obj.A) }],
                    bids: [{ price: Number(obj.b), qty: Number(obj.B) }]
                }
            } else if (obj.e == 'aggTrade') {
                event.ts = obj.E
                event.trades = [{
                    price: Number(obj.p),
                    qty: Number(obj.q),
                    ts: obj.T,
                    side: obj.m ? "sell" : "buy"
                }]
            } else if (typeof (obj.asks) !== 'undefined') {
                event.ts = obj.E || new Date().getTime()
                let depth = {
                    asks: [],
                    bids: []
                }
                obj.bids.forEach(function (item) {
                    depth.bids.push({
                        price: Number(item[0]),
                        qty: Number(item[1])
                    })
                })
                obj.asks.forEach(function (item) {
                    depth.asks.push({
                        price: Number(item[0]),
                        qty: Number(item[1])
                    })
                })
                event.depth = depth
            } else {
                return
            }
            return event
        }
        let channels = ["depth20@100ms", /*"bookTicker", */"aggTrade"]
 
        let ws = null
        let endPoint = "wss://stream.binance.com/stream"
        if (ctx.name == "Futures_Binance") {
            endPoint = "wss://fstream.binance.com/stream"
        }
        
        while (true) {
            if (!ws) {
                let subscribes = []
                ctx.symbols.forEach(function (symbol) {
                    channels.forEach(function (channel) {
                        subscribes.push(symbol.toLowerCase() + "@" + channel)
                    })
                })
                ws = MyDial(endPoint + (subscribes.length > 0 ? ("?streams=" + subscribes.join("/")) : ""))
            }
            if (!ws) {
                Sleep(1000)
                continue
            }
            updateSymbols(ctx, function(symbol:string, method:string) {
                ws.write(JSON.stringify({ 
                    "method": method.toUpperCase(), 
                    "params": channels.map(c=>symbol.toLowerCase()+'@'+c),
                    "id": 2
                }))
            })
            let msg = ws.read(1000)
            if (!msg) {
                if (msg == "") {
                    trace("websocket is closed")
                    ws.close()
                    ws = null
                }
                continue
            }
            if (msg == 'ping') {
                ws.write('pong')
            } else if (msg == 'pong') {

            } else {
                let obj = JSON.parse(msg)
                if (obj.error) {
                    trace(obj.error.msg, "#ff0000")
                    continue
                }
                if (!obj.stream) {
                    continue
                }
                if (obj.stream.indexOf("depth") != -1) {
                    if (typeof(obj.data.s) !== 'string') {
                        // patch
                        obj.data.s = obj.stream.split('@')[0].toUpperCase()
                    }
                }
                let event = processMsg(obj.data)
                if (event) {
                    ctx.callback(event)
                }
            }
        }
    }