В процессе загрузки ресурсов... загрузка...

Схема политики, которая позволяет беспрепятственно использовать веб-сокет

Автор:Трава, Создано: 2024-10-30 09:49:20, Обновлено: 2024-11-05 17:45:31

Это официальный шаблон FMZ для WebSocket, который можно скопировать и сохранить в качестве шаблона, который можно использовать при нажатии на этот шаблон в новой политике:https://www.fmz.com/strategy/470349

Зачем нужен WebSocket?

В настоящее время стратегия FMZ основана на традиционном REST API-оболочке, при котором на каждом этапе доступа к API устанавливается сетевое соединение для получения рыночных данных путем опроса.

Однако REST-протокол имеет проблемы с задержкой, которые усиливаются при необходимости использования нескольких торговых пар и стратегий для нескольких бирж. Хотя Go-функции платформы могут выполняться одновременно, проблемы с задержкой остаются, что затрудняет удовлетворение потребностей в стратегических сделках с относительно высокой частотой, а также слишком большое количество сделок, частота обращений слишком быстро и ограничение частоты доступа к торговым платформам.

В настоящее время серверные нагрузки на биржи также очень высоки, и все они предлагают исправные протоколы WebSocket и рекомендуют пользователям API использовать. По сравнению с протоколом REST, WebSocket предоставляет устойчивый двусторонний способ подключения, который позволяет бирже отправлять данные клиентам в режиме реального времени, избегая частых запросов и ответов, что значительно снижает задержку. В целом, если задержка доступа к REST API составляет около 20 мс, то задержка доступа к WebSocket составляет около 2 мс.

Введение шаблона WebSocket

Квалификационная торговая платформа FMZ поддерживала протокол WebSocket с самого начала и была относительно удобной для вызова, но для новичков обрабатывать несколько подписок, подписки на несколько биржевых рынков и эффективно и легко встраиваться во все стратегические процессы было слишком сложно. Этот открытый шаблон WebSocket Real-time Market Data Acceleration решает эту проблему, очень простой в использовании, полностью совместим с текущими упакованными API-призывами, для большинства из первоначальных REST-политик, простой переход прямо в использование, чтобы ускорить вашу политику.

Основные особенности:

Поддержка от многих биржЭта политика поддерживает подключение WebSocket к нескольким биржам, таким как Binance, OKX, ByBit и Bitget, и пользователи могут имитировать метод упаковки шаблона, чтобы поддержать больше бирж.
Подписка на заказ: позволяет подписываться на определенные рынковые каналы (например, глубина, сделки и т.д.) и эффективно обрабатывать полученные данные для мгновенного использования в стратегии торговли;
Высокая погрешность обработкиВстроенные механизмы отслеживания ошибок и WebSocket для обеспечения надежности и непрерывности потока данных.

Очень простое представление о принципах реализации

Обратите внимание, что эта стратегия использует TypeScript, что может показаться немного незнакомым, если вы просто знакомы с JavaScript. TypeScript внедряет типную систему и более богатые языковые функции на основе JavaScript. Использование TypeScript может уменьшить потенциальные ошибки, улучшить читаемость и устойчивость кода.

В дополнение к этому политика использует синхронный механизм платформы FMZ, который позволяет потоку передавать сообщения через функцию __threadPostMessage. Этот способ является синхронным и используется для уведомления потока об обновлении данных, созданных в потоке. Данные могут быть обменены между потоком и потоком с помощью функций __threadGetData и __threadSetData.

Основной принцип этой стратегии заключается в том, чтобы подключить основные цифровые валютные биржи через WebSocket и получить рыночные данные в режиме реального времени (например, глубинную информацию и информацию о сделках), чтобы обеспечить поддержку данных для количественных торговых решений.

1. Настройки подключения WebSocket

setupWebsocketФункция используется для инициализации соединения WebSocket, для получения рыночных данных.main_exchangesВ этом случае, если вы хотите, чтобы ваш сайт был доступен для всех пользователей, вы можете использовать его.

MyDialФункции: создает соединение WebSocket, записывает время соединения и выводит время закрытия при закрытии соединения.
updateSymbolsФункции: регулярно проверяет, есть ли новые запросы на подписку, и обновляет список текущих сделок по мере необходимости.

2. Обработка данных

supportsОбъект определяет поддерживаемые биржи и их функции обработки (например,BinanceПроцессорная функция каждой биржи отвечает за анализ полученных сообщений и извлечение соответствующих данных.

processMsgФункции: обрабатывает сообщения с бирж, идентифицирует различные типы данных (например, глубокие обновления, сделки и т.д.) и форматирует их в единые объекты событий;

3. Данные о подписке

При каждом подключении система будет переводить данные рынка по каналам, связанным с подпиской, в зависимости от текущей сделки.

getFunctionФункции: получение соответствующей обработки в соответствии с именем биржи.
this.wssPublicФункции: инициализировать соединение WebSocket и начать прием данных.

4. Управление потоками

Запускается одна нить для каждой биржи, получается информация в режиме реального времени и обрабатывается с помощью функций обратного вызова.

threadMarketФункции: получать данные в подпроводках, анализировать и хранить новейшую глубину и информацию о сделках.

5. Перезапись методов получения данных

Для каждой биржи переписывается способ получения глубины и информации о сделках, приоритетом которого является возвращение обновленных данных в режиме реального времени.

Использование шаблонов

ИнициализацияИспользование:$.setupWebsocket()Инициализировать соединение WebSocket с целевой биржей.
ПодписатьсяСистема автоматически подписывается на соответствующие каналы (например, глубина, торговля и т.д.) для того сорта, в котором вы торгуете.
Получение данныхНазвание:GetDepth()иGetTrades()Функция, автоматически использующая данные WebSocket в режиме реального времени для глубины рынка и возвращения записей сделок.
Неправильное обращение: Политика включает в себя механизм отслеживания, который используется для записи ошибок соединения и данных и автоматического попытки восстановления соединения при прерывании соединения.

Если в политику включить функцию EventLoop (), то она превратится в триггерный механизм, который автоматически сразу же будет доступен, когда обновляются данные wss, а не ждать, пока не будут обновлены данные. Это эквивалентно умной функции Sleep, но, конечно же, можно использовать ее напрямую.

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

Ссылаясь на мой предыдущий материал о стратегии торговли в разных валютахhttps://www.fmz.com/digest-topic/10506Здесь вы можете очень легко изменить его для поддержки 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);
    }
}

Как самостоятельно добавить новые биржи

Если вы будете следовать шаблону политики и копировать формат, указанный ниже, то сможете ознакомиться с API-документацией биржи:

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

Больше