pineforge-feed: OKX, Bybit and Binance USD-M venues (slice 2)

native/feed/CMakeLists.txt

@@ -45,7 +45,8 @@ if(PINEFORGE_FEED_SANITIZER)
   add_link_options(-fsanitize=${PINEFORGE_FEED_SANITIZER})
 endif()
 
-add_library(pineforge_feed_core STATIC src/core.cpp src/state.cpp src/transport.cpp src/binance.cpp)
+add_library(pineforge_feed_core STATIC src/core.cpp src/state.cpp src/transport.cpp src/binance.cpp src/okx.cpp
+  src/bybit.cpp src/session.cpp)
 target_include_directories(pineforge_feed_core PUBLIC src vendor)
 target_link_libraries(pineforge_feed_core PUBLIC CURL::libcurl OpenSSL::Crypto Threads::Threads)
 target_compile_options(pineforge_feed_core PRIVATE -Wall -Wextra -Wpedantic -Werror)
@@ -62,5 +63,5 @@ if(BUILD_TESTING)
   add_test(NAME feed-unit COMMAND feed-unit)
   find_package(Python3 REQUIRED COMPONENTS Interpreter)
   add_test(NAME feed-mock COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tests/mock_venue.py $<TARGET_FILE:pineforge-feed>)
-  set_tests_properties(feed-mock PROPERTIES TIMEOUT 180)
+  set_tests_properties(feed-mock PROPERTIES TIMEOUT 420)
 endif()

native/feed/src/binance.hpp

@@ -1,55 +1,48 @@
 // SPDX-License-Identifier: Apache-2.0
 #pragma once
-#include "state.hpp"
 #include "transport.hpp"
 #include "venue.hpp"
-#include <map>
 
 namespace pineforge::feed {
 Trade binance_trade(const Json& trade, bool websocket);
 Kline binance_kline(const Json& row, bool websocket);
+// One USD-M aggregate print: `a` is the sequence, `q` the quantity. Its raw `f..l` range is validated,
+// never expanded into imaginary raw trades.
+Trade usdm_aggregate(const Json& row);
+Frame binance_frame(const std::string& message);
+Connection usdm_connection(const Config& config);
+unsigned int usdm_kline_weight(std::size_t limit);
 
 class BinanceSpot final : public Venue {
     Config config_;
     HttpClient http_;
 public:
-    explicit BinanceSpot(const Config& config) : config_(config), http_(config) {}
-    std::string streams() const override;
-    VenueEvent decode(const std::string& message) const override;
+    explicit BinanceSpot(const Config& config) : config_(config), http_(config, binance_spot_rest(config.symbol)) {}
+    Connection connection() const override;
+    std::vector<VenueEvent> decode(const std::string& message) const override;
     std::vector<Trade> history(std::uint64_t from, std::size_t limit = 1000) override;
     std::uint64_t first_trade_id(std::int64_t minute) override;
     std::vector<Kline> klines(std::int64_t start, std::int64_t end) override;
+    std::string kline_source() const override { return "/api/v3/klines"; }
 };
 
-class FeedSession {
-    State& state_;
-    Venue& venue_;
-    std::function<void(const std::string&)> output_;
-    Aggregate aggregate_;
-    std::map<std::uint64_t, Trade> pending_;
-    std::size_t pending_bytes_ = 0;
-    std::int64_t watermark_ = -1;
-    std::uint64_t new_messages_ = 0;
-    bool replayed_ = false;
-    void emit(const std::string& line, const std::optional<Bar>& proof = std::nullopt);
-    void anchor(std::uint64_t first);
-    std::uint64_t next_id() const;
-    void put(const Trade& trade);
-    void emit_trade(const Trade& trade);
-    void heal_to(std::uint64_t target);
-    void drain();
-    void historical_minute();
-    void closed(const Kline& kline);
+// Binance USD-M perpetuals: confirmed bars, or the opt-in `agg-ticks` aggregate-print mode. Both use the
+// routed /market stream; the unrouted origin no longer serves aggTrade or kline streams.
+class BinanceUsdm final : public Venue {
+    Config config_;
+    HttpClient http_;
 public:
-    FeedSession(State& state, Venue& venue, std::function<void(const std::string&)> output);
-    void connected();
-    // stage() verifies an event and stages its messages; publish() group-commits them, then writes them
-    // out; salvage() publishes what was already verified before a stop. ingest() is all three for one event.
-    void stage(const VenueEvent& event);
-    void publish();
-    void salvage() noexcept;
-    void ingest(const VenueEvent& event);
+    explicit BinanceUsdm(const Config& config);
+    Connection connection() const override { return usdm_connection(config_); }
+    std::vector<VenueEvent> decode(const std::string& message) const override;
+    std::vector<Trade> history(std::uint64_t from, std::size_t limit = 1000) override;
+    std::uint64_t first_trade_id(std::int64_t minute) override;
+    std::optional<Trade> predecessor_if_ready(std::int64_t minute) override;
+    std::vector<Kline> klines(std::int64_t start, std::int64_t end) override;
+    TickProof tick_proof() const override { return TickProof::NextPrintFence; }
+    std::int64_t retention_ms() const override { return 48LL * 3600000; }
+    // Observed live: an aggregate dated in one minute held a fill the next minute's kline counts.
+    bool candle_is_print_sum() const override { return false; }
+    std::string kline_source() const override { return "/fapi/v1/klines"; }
 };
-void run_feed(const Config& config);
-void warmup(const Config& config, const std::string& output);
 }

native/feed/src/bybit.cpp

@@ -0,0 +1,135 @@
+// SPDX-License-Identifier: Apache-2.0
+#include "bybit.hpp"
+#include <algorithm>
+#include <charconv>
+
+namespace pineforge::feed {
+namespace {
+std::int64_t milliseconds(const Json& value) {
+    // WebSocket rows carry integers, REST rows decimal strings.
+    const auto text = value.kind == Json::Kind::String ? value.text() : value.value;
+    if (value.kind != Json::Kind::String && value.kind != Json::Kind::Number) throw Error(23, "invalid Bybit time");
+    std::int64_t result = 0;
+    const auto parsed = std::from_chars(text.data(), text.data() + text.size(), result);
+    if (text.empty() || text.find_first_not_of("0123456789") != std::string::npos || parsed.ec != std::errc{} ||
+        parsed.ptr != text.data() + text.size()) throw Error(23, "invalid Bybit time");
+    return result;
+}
+RestPolicy bybit_rest() {
+    RestPolicy policy;
+    policy.endpoints = {"/v5/market/kline", "/v5/market/instruments-info"};
+    // 600 requests per 5 seconds per IP; this spends well under a fifth of it.
+    policy.milliseconds_per_request = 100;
+    policy.throttled = [](const std::string& body) { return body.find("\"retCode\":10006") != std::string::npos; };
+    // Bybit answers HTTP 403 for its roughly ten-minute IP rate ban: a restart after the ban recovers.
+    policy.rejected = [](long status, const std::string&) {
+        if (status == 403) throw Error(20, "Bybit IP rate ban (HTTP 403); restart after the venue's ban window");
+    };
+    // Server timeout and server error: the venue asks to retry.
+    policy.transient = [](long, const std::string& body) {
+        for (const char* code : {"10000", "10016"})
+            for (const char* end : {",", "}"})
+                if (body.find(std::string("\"retCode\":") + code + end) != std::string::npos) return true;
+        return false;
+    };
+    return policy;
+}
+std::string category(const Config& config) { return config.market == "linear" ? "linear" : "spot"; }
+}
+
+Kline bybit_kline(const Json& row) {
+    if (row.at("interval").text() != "1") throw Error(23, "only one-minute Bybit klines are supported");
+    Kline kline;
+    kline.bar = {milliseconds(row.at("start")), row.at("open").text(), row.at("high").text(), row.at("low").text(),
+                 row.at("close").text(), row.at("volume").text()};
+    kline.bar.validate();
+    if (milliseconds(row.at("end")) != kline.bar.ts + 59999) throw Error(23, "invalid inclusive one-minute Bybit kline end");
+    const auto& confirm = row.at("confirm");
+    if (confirm.kind != Json::Kind::Bool) throw Error(23, "confirmation must be a JSON boolean");
+    kline.confirmed = confirm.value == "true";
+    return kline;
+}
+std::vector<Kline> bybit_rows(const Json& list, std::int64_t start, std::size_t count) {
+    if (list.kind != Json::Kind::Array || list.items.size() > 1000) throw Error(23, "invalid Bybit kline list");
+    std::vector<Kline> result;
+    auto expected = start;
+    for (auto item = list.items.rbegin(); item != list.items.rend(); ++item) {
+        if (item->kind != Json::Kind::Array || item->items.size() != 7) throw Error(23, "Bybit REST kline must have seven fields");
+        const auto& row = item->items;
+        Kline kline;
+        kline.bar = {milliseconds(row[0]), row[1].text(), row[2].text(), row[3].text(), row[4].text(), row[5].text()};
+        kline.bar.validate();
+        if (kline.bar.ts != expected) throw Error(20, "historical klines have a missing, unordered or unexpected minute");
+        expected += 60000;
+        result.push_back(kline);
+    }
+    if (result.size() != count) throw Error(20, "historical klines have a missing minute");
+    return result;
+}
+Frame bybit_frame(const std::string& message) {
+    try {
+        const auto envelope = parse_json(message);
+        const auto* operation = envelope.find("op");
+        if (!operation) return Frame::Data;
+        if (operation->text() == "subscribe") {
+            const auto* success = envelope.find("success");
+            if (!success || success->kind != Json::Kind::Bool || success->value != "true") throw Error(23, "Bybit rejected the subscription");
+        }
+        return Frame::Control;
+    } catch (const Error&) { throw; }
+    catch (const std::exception&) { throw Error(23, "invalid Bybit WebSocket JSON"); }
+}
+
+Bybit::Bybit(const Config& config) : config_(config), http_(config, bybit_rest()) {
+    const auto instruments = result("/v5/market/instruments-info?category=" + category(config_) + "&symbol=" + config_.symbol);
+    try {
+        const auto& list = instruments.at("list");
+        if (list.kind != Json::Kind::Array || list.items.size() != 1) throw Error(23, "unknown Bybit symbol");
+        const auto& row = list.items.front();
+        if (row.at("symbol").text() != config_.symbol) throw Error(23, "Bybit instrument metadata does not match the symbol");
+        if (row.at("status").text() != "Trading") throw Error(23, "Bybit instrument is not trading");
+        if (config_.market == "linear") {
+            const auto type = row.at("contractType").text();
+            if (type != "LinearPerpetual" && type != "LinearFutures") throw Error(23, "Bybit instrument is not a linear contract");
+        }
+    } catch (const Error&) { throw; }
+    catch (const std::exception&) { throw Error(23, "invalid Bybit instrument metadata"); }
+}
+Json Bybit::result(const std::string& path) {
+    const auto response = http_.get(path, 1);
+    try {
+        const auto& code = response.at("retCode");
+        if (code.kind != Json::Kind::Number || code.value != "0") throw Error(23, "Bybit rejected the public request (retCode " + code.value + ")");
+        return response.at("result");
+    } catch (const Error&) { throw; }
+    catch (const std::exception&) { throw Error(23, "invalid Bybit REST envelope"); }
+}
+Connection Bybit::connection() const {
+    return {"/v5/public/" + category(config_), {"{\"op\":\"subscribe\",\"args\":[\"kline.1." + config_.symbol + "\"]}"},
+            "{\"op\":\"ping\"}", &bybit_frame};
+}
+std::vector<VenueEvent> Bybit::decode(const std::string& message) const {
+    try {
+        const auto envelope = parse_json(message);
+        if (envelope.at("topic").text() != "kline.1." + config_.symbol) throw Error(23, "unsupported Bybit topic");
+        const auto& data = envelope.at("data");
+        if (data.kind != Json::Kind::Array || data.items.size() > 1024) throw Error(23, "invalid Bybit data array");
+        // A push can carry the minute that just closed (confirm=true) and the next forming one.
+        std::vector<VenueEvent> events;
+        for (const auto& row : data.items) events.push_back({VenueEvent::Kind::Kline, {}, bybit_kline(row)});
+        return events;
+    } catch (const Error&) { throw; }
+    catch (const std::exception&) { throw Error(23, "invalid Bybit message shape"); }
+}
+std::vector<Kline> Bybit::klines(std::int64_t start, std::int64_t end) {
+    if (start < 0 || start % 60000 || end <= start || end % 60000) throw Error(23, "invalid exclusive kline range");
+    // `start` and `end` are inclusive candle starts and rows arrive newest first: a wider range than the
+    // limit would silently drop its oldest rows, so each request asks for exactly the rows it can hold.
+    const auto count = static_cast<std::size_t>(std::min<std::int64_t>(1000, (end - start) / 60000));
+    const auto list = result("/v5/market/kline?category=" + category(config_) + "&symbol=" + config_.symbol + "&interval=1&start=" +
+        std::to_string(start) + "&end=" + std::to_string(start + static_cast<std::int64_t>(count - 1) * 60000) + "&limit=" + std::to_string(count));
+    try { return bybit_rows(list.at("list"), start, count); }
+    catch (const Error&) { throw; }
+    catch (const std::exception&) { throw Error(23, "invalid Bybit kline result"); }
+}
+}

native/feed/src/bybit.hpp

@@ -0,0 +1,28 @@
+// SPDX-License-Identifier: Apache-2.0
+#pragma once
+#include "transport.hpp"
+#include "venue.hpp"
+
+namespace pineforge::feed {
+// {start, end, interval, open, close, high, low, volume, turnover, confirm, timestamp}; only confirm=true closes.
+Kline bybit_kline(const Json& row);
+// REST rows arrive newest first: they are reversed and must cover [start, start + count minutes) exactly.
+std::vector<Kline> bybit_rows(const Json& list, std::int64_t start, std::size_t count);
+Frame bybit_frame(const std::string& message);
+
+// Bybit v5 spot and linear: confirmed one-minute bars only. Ticks are refused at startup: trade IDs are
+// not a contiguous cursor and the public REST cannot page back through trades. Bars never come from
+// trades, because trade-built bars disagree with Bybit candles.
+class Bybit final : public Venue {
+    Config config_;
+    HttpClient http_;
+    Json result(const std::string& path);
+public:
+    explicit Bybit(const Config& config);
+    Connection connection() const override;
+    std::vector<VenueEvent> decode(const std::string& message) const override;
+    std::vector<Trade> history(std::uint64_t, std::size_t) override { throw Error(23, "Bybit has no pageable trade history"); }
+    std::vector<Kline> klines(std::int64_t start, std::int64_t end) override;
+    std::string kline_source() const override { return "/v5/market/kline"; }
+};
+}

native/feed/src/core.cpp

@@ -3,6 +3,7 @@
 #include <openssl/evp.h>
 #include <openssl/rand.h>
 #include <algorithm>
+#include <cctype>
 #include <cerrno>
 #include <charconv>
 #include <chrono>
@@ -238,6 +239,85 @@ Decimal Decimal::add(const Decimal& other) const {
     return Decimal(sum, scale);
 }
 
+Decimal Decimal::multiply(const Decimal& other) const {
+    if (zero() || other.zero()) return Decimal("0", 0);
+    // The constructor bounds the product: at most 128 coefficient digits and 32 fractional digits.
+    std::string product(coefficient_.size() + other.coefficient_.size(), '0');
+    for (std::size_t left = coefficient_.size(); left > 0; --left) {
+        int carry = 0;
+        for (std::size_t right = other.coefficient_.size(); right > 0; --right) {
+            auto& digit = product[left + right - 1];
+            const int value = (digit - '0') + (coefficient_[left - 1] - '0') * (other.coefficient_[right - 1] - '0') + carry;
+            digit = static_cast<char>('0' + value % 10);
+            carry = value / 10;
+        }
+        product[left - 1] = static_cast<char>(product[left - 1] + carry);
+    }
+    return Decimal(product, scale_ + other.scale_);
+}
+
+std::string Decimal::str() const {
+    if (!scale_) return coefficient_;
+    auto digits = coefficient_;
+    if (digits.size() <= scale_) digits.insert(0, scale_ - digits.size() + 1, '0');
+    digits.insert(digits.size() - scale_, 1, '.');
+    return digits;
+}
+
+std::string scaled_quantity(const std::string& token, const std::string& multiplier) {
+    const Decimal amount(token);
+    if (multiplier == "1") return token;
+    return amount.multiply(Decimal(multiplier)).str();
+}
+
+std::string json_member(const std::string& document, const std::string& name) {
+    std::size_t index = 0;
+    const auto space = [&] { while (index < document.size() && std::isspace(static_cast<unsigned char>(document[index]))) ++index; };
+    const auto string_end = [&](std::size_t start) {
+        for (auto cursor = start + 1; cursor < document.size(); ++cursor) {
+            if (document[cursor] == '\\') ++cursor;
+            else if (document[cursor] == '"') return cursor + 1;
+        }
+        throw Error(23, "unterminated JSON string");
+    };
+    const auto value_end = [&](std::size_t start) {
+        if (start >= document.size()) throw Error(23, "truncated JSON member");
+        if (document[start] == '"') return string_end(start);
+        std::size_t depth = 0, cursor = start;
+        while (cursor < document.size()) {
+            const char letter = document[cursor];
+            if (letter == '"') { cursor = string_end(cursor); continue; }
+            if (letter == '{' || letter == '[') ++depth;
+            else if (letter == '}' || letter == ']') {
+                if (!depth) return cursor;
+                if (!--depth) return cursor + 1;
+            } else if (letter == ',' && !depth) return cursor;
+            ++cursor;
+        }
+        throw Error(23, "truncated JSON member");
+    };
+    space();
+    if (index >= document.size() || document[index] != '{') throw Error(23, "expected a JSON object document");
+    ++index;
+    for (;;) {
+        space();
+        if (index >= document.size() || document[index] != '"') throw Error(23, "JSON member not found: " + name);
+        const auto key_end = string_end(index);
+        const auto key = document.substr(index + 1, key_end - index - 2);
+        index = key_end;
+        space();
+        if (index >= document.size() || document[index] != ':') throw Error(23, "invalid JSON object member");
+        ++index;
+        space();
+        const auto start = index;
+        index = value_end(start);
+        if (key == name) return document.substr(start, index - start);
+        space();
+        if (index >= document.size() || document[index] != ',') throw Error(23, "JSON member not found: " + name);
+        ++index;
+    }
+}
+
 void Trade::validate() const {
     if (!id || ts < 0 || ts > std::numeric_limits<std::int64_t>::max() - 60000 || Decimal(price).zero() || Decimal(qty).zero())
         throw Error(23, "invalid positive venue trade");
@@ -247,8 +327,10 @@ std::string Trade::wire() const {
     return "{\"type\":\"tick\",\"ts\":" + std::to_string(ts) + ",\"seq\":" + std::to_string(id) +
         ",\"price\":" + price + ",\"qty\":" + qty + "}";
 }
+// Exact decimal values, not lexemes: a venue may render one number as "100.0" on its WebSocket and as
+// "100" over REST. Emitted tokens stay verbatim; only the comparison is by value.
 bool Trade::operator==(const Trade& other) const {
-    return id == other.id && ts == other.ts && price == other.price && qty == other.qty;
+    return id == other.id && ts == other.ts && Decimal(price) == Decimal(other.price) && Decimal(qty) == Decimal(other.qty);
 }
 void Bar::validate() const {
     if (ts < 0 || ts % 60000 || ts > std::numeric_limits<std::int64_t>::max() - 60000)
@@ -265,7 +347,8 @@ std::string Bar::wire() const {
         ",\"h\":" + high + ",\"l\":" + low + ",\"c\":" + close + ",\"v\":" + volume + "}}";
 }
 bool Bar::operator==(const Bar& other) const {
-    return ts == other.ts && open == other.open && high == other.high && low == other.low && close == other.close && volume == other.volume;
+    return ts == other.ts && Decimal(open) == Decimal(other.open) && Decimal(high) == Decimal(other.high) &&
+        Decimal(low) == Decimal(other.low) && Decimal(close) == Decimal(other.close) && Decimal(volume) == Decimal(other.volume);
 }
 void Aggregate::add(const Trade& trade) {
     trade.validate();