rustrade_data/exchange/binance/

kline.rs

//! Live Binance kline (candle) WebSocket payloads and their normalisation to
//! [`Candle`](crate::subscription::candle::Candle).
//!
//! Covers both the [`BinanceSpot`](crate::exchange::binance::spot::BinanceSpot) `@kline_<interval>`
//! stream and the
//! [`BinanceFuturesUsdMarket`](crate::exchange::binance::futures::BinanceFuturesUsdMarket)
//! `@continuousKline_<interval>` stream (perpetual-only). Endpoints are public/unauthenticated.
//!
//! # Closed candles only (no repaint)
//!
//! rustrade emits **closed candles only** — an in-progress kline (`k.x == false`) yields an empty
//! [`MarketIter`](crate::event::MarketIter), so consumers never see a repainting/lookahead value.
//! The exclusive `close_time` boundary is recomputed library-side as `open + interval` (see
//! [`close_time_from_open`](crate::subscription::candle::close_time_from_open)), **not** taken from
//! Binance's wire `T` (its `period-end − 1ms` convention) — consumers comparing against the raw `T`
//! will see a 1ms difference by design.
//!
//! # Reconnection: no replay, no dedup (consumer policy)
//!
//! Across a reconnect the underlying [`Connector`](crate::exchange::Connector) /
//! `ReconnectingStream` re-subscribes but does **not** replay or de-duplicate: a closed candle
//! straddling the disconnect may be **re-delivered or skipped**. De-duplication and gap-back-fill
//! are **consumer policy, not library policy** (consistent with rustrade's "no consumer-specific
//! policy in the library" rule). A consumer wanting a gapless series should reconcile the live
//! candle stream against a
//! [`fetch_candles`](crate::exchange::binance::historical::BinanceHistoricalClient::fetch_candles)
//! backfill keyed on `close_time` (the field both paths agree on, since `open ≡ close − interval`).

use super::BinanceChannel;
use crate::{
    Identifier,
    error::DataError,
    event::{MarketEvent, MarketIter},
    subscription::candle::{Candle, CandleInterval, close_time_from_open},
};
use chrono::{DateTime, Utc};
use rust_decimal::Decimal;
use rustrade_instrument::exchange::ExchangeId;
use rustrade_integration::subscription::SubscriptionId;
use serde::{Deserialize, Serialize};
use smol_str::{SmolStr, format_smolstr};

/// The inner `k` payload shared by the [`BinanceSpot`](super::spot::BinanceSpot) `@kline_`
/// and [`BinanceFuturesUsdMarket`](super::futures::BinanceFuturesUsdMarket) `@continuousKline_` streams.
///
/// Both surfaces carry an identical `k` object for the fields rustrade consumes; they differ
/// only in where the market symbol lives (spot top-level `s` vs futures top-level `ps`) and
/// in the continuous payload omitting `k.s`. This struct deliberately omits the symbol so it
/// can be reused by both — the symbol is captured by the outer model and used to build the
/// [`SubscriptionId`].
///
/// ### Correctness notes
/// - OHLCV are JSON **strings** on the wire (e.g. `"0.01634790"`) → parsed via `de_str` to
///   [`Decimal`], never through an `f64` (which silently truncates precision).
/// - `open_time` (`t`) is the candle's open instant; the exclusive `close_time` boundary is
///   recomputed library-side via [`close_time_from_open`] (`open + interval`), **not** taken
///   from the wire `T` (Binance's `period-end − 1ms` convention).
#[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize, Serialize)]
pub struct BinanceKlineData {
    #[serde(
        alias = "t",
        deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
    )]
    pub open_time: DateTime<Utc>,
    #[serde(alias = "i")]
    pub interval: CandleInterval,
    #[serde(
        alias = "o",
        deserialize_with = "rustrade_integration::serde::de::de_str"
    )]
    pub open: Decimal,
    #[serde(
        alias = "h",
        deserialize_with = "rustrade_integration::serde::de::de_str"
    )]
    pub high: Decimal,
    #[serde(
        alias = "l",
        deserialize_with = "rustrade_integration::serde::de::de_str"
    )]
    pub low: Decimal,
    #[serde(
        alias = "c",
        deserialize_with = "rustrade_integration::serde::de::de_str"
    )]
    pub close: Decimal,
    #[serde(
        alias = "v",
        deserialize_with = "rustrade_integration::serde::de::de_str"
    )]
    pub volume: Decimal,
    #[serde(alias = "n")]
    pub trade_count: u64,
    /// `true` once the kline interval has closed. rustrade emits **closed candles only**
    /// (no repaint/lookahead) — an in-progress kline yields an empty [`MarketIter`].
    #[serde(alias = "x")]
    pub closed: bool,
}

impl BinanceKlineData {
    /// Map a kline payload to a normalised [`Candle`] [`MarketEvent`], honouring the
    /// closed-only policy and the `close_time = open + interval` boundary contract.
    ///
    /// - In-progress klines (`x == false`) yield an empty [`MarketIter`] (no event).
    /// - A closed kline whose computed `close_time` overflows the representable
    ///   [`DateTime<Utc>`] range yields an observable [`DataError`] rather than a silent
    ///   drop or a plausible-but-wrong timestamp (unreachable for real intervals, but the
    ///   boundary contract forbids a silent fallback).
    fn into_market_iter<InstrumentKey>(
        self,
        exchange_id: ExchangeId,
        instrument: InstrumentKey,
    ) -> MarketIter<InstrumentKey, Candle> {
        if !self.closed {
            return MarketIter(vec![]);
        }

        match close_time_from_open(self.open_time, self.interval.to_step()) {
            Some(close_time) => MarketIter(vec![Ok(MarketEvent {
                time_exchange: close_time,
                time_received: Utc::now(),
                exchange: exchange_id,
                instrument,
                kind: Candle {
                    close_time,
                    open: self.open,
                    high: self.high,
                    low: self.low,
                    close: self.close,
                    volume: self.volume,
                    trade_count: self.trade_count,
                },
            })]),
            None => MarketIter(vec![Err(DataError::Socket(format!(
                "Binance candle close_time overflow: open_time {} + interval {} exceeds the representable DateTime<Utc> range",
                self.open_time, self.interval
            )))]),
        }
    }
}

/// [`BinanceSpot`](super::spot::BinanceSpot) real-time kline (candle) message.
///
/// ### Raw Payload Example
/// See docs: <https://binance-docs.github.io/apidocs/spot/en/#kline-candlestick-streams>
/// ```json
/// {
///     "e": "kline",
///     "E": 1638747660000,
///     "s": "BTCUSDT",
///     "k": {
///         "t": 1638747660000, "T": 1638747719999, "s": "BTCUSDT", "i": "1m",
///         "f": 100, "L": 200, "o": "0.0010", "c": "0.0020", "h": "0.0025",
///         "l": "0.0015", "v": "1000", "n": 100, "x": false, "q": "1.0000",
///         "V": "500", "Q": "0.500", "B": "123456"
///     }
/// }
/// ```
#[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize, Serialize)]
pub struct BinanceKline {
    /// Top-level market symbol, UPPERCASE (e.g. `BTCUSDT`).
    #[serde(alias = "s")]
    pub symbol: SmolStr,
    #[serde(alias = "k")]
    pub kline: BinanceKlineData,
}

impl Identifier<Option<SubscriptionId>> for BinanceKline {
    /// Builds the instrument-map key `{channel}|{MARKET}` (e.g. `@kline_1m|BTCUSDT`) from the
    /// payload's interval (`k.i`) and UPPERCASE symbol (`s`) — matching
    /// [`ExchangeSub::id`](crate::exchange::ExchangeSub), **not** the lowercase stream name.
    fn id(&self) -> Option<SubscriptionId> {
        Some(SubscriptionId(format_smolstr!(
            "{}|{}",
            BinanceChannel::spot_candle(self.kline.interval).0,
            self.symbol
        )))
    }
}

impl<InstrumentKey> From<(ExchangeId, InstrumentKey, BinanceKline)>
    for MarketIter<InstrumentKey, Candle>
{
    fn from((exchange_id, instrument, kline): (ExchangeId, InstrumentKey, BinanceKline)) -> Self {
        kline.kline.into_market_iter(exchange_id, instrument)
    }
}

/// [`BinanceFuturesUsdMarket`](super::futures::BinanceFuturesUsdMarket) real-time continuous-contract
/// (perpetual) kline message.
///
/// Differs from the spot [`BinanceKline`] shape: the symbol is the top-level `ps` (pair) and
/// `ct` (contract type), and the inner `k` object has **no** `s` field.
///
/// ### Raw Payload Example
/// See docs: <https://binance-docs.github.io/apidocs/futures/en/#continuous-contract-kline-candlestick-streams>
/// ```json
/// {
///     "e": "continuous_kline",
///     "E": 1607443058651,
///     "ps": "BTCUSDT",
///     "ct": "PERPETUAL",
///     "k": {
///         "t": 1607443020000, "T": 1607443079999, "i": "1m", "f": 116467658886,
///         "L": 116468012423, "o": "18787.00", "c": "18804.04", "h": "18804.04",
///         "l": "18786.54", "v": "197.664", "n": 543, "x": false, "q": "3715253.19494",
///         "V": "184.769", "Q": "3472925.84746", "B": "0"
///     }
/// }
/// ```
#[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize, Serialize)]
pub struct BinanceContinuousKline {
    /// Top-level pair symbol, UPPERCASE (e.g. `BTCUSDT`). For a perpetual-only venue the pair
    /// equals the symbol.
    #[serde(alias = "ps")]
    pub pair: SmolStr,
    // The wire `ct` (contract type, e.g. `"PERPETUAL"`) is intentionally not deserialized: the
    // `_perpetual@continuousKline_` subscription this model decodes only ever receives PERPETUAL
    // frames, and `id()` hardcodes the perpetual channel prefix to match. A non-perpetual frame
    // (impossible given the subscription) would build a non-matching `SubscriptionId` and be
    // dropped at the instrument-map lookup rather than mis-attributed.
    #[serde(alias = "k")]
    pub kline: BinanceKlineData,
}

impl Identifier<Option<SubscriptionId>> for BinanceContinuousKline {
    /// Builds the instrument-map key `{channel}|{MARKET}` (e.g.
    /// `_perpetual@continuousKline_1m|BTCUSDT`) from the payload's interval (`k.i`) and
    /// UPPERCASE pair (`ps`) — matching [`ExchangeSub::id`](crate::exchange::ExchangeSub). The
    /// continuous payload has no `k.s`, so the pair is the only symbol source.
    fn id(&self) -> Option<SubscriptionId> {
        Some(SubscriptionId(format_smolstr!(
            "{}|{}",
            BinanceChannel::futures_candle(self.kline.interval).0,
            self.pair
        )))
    }
}

impl<InstrumentKey> From<(ExchangeId, InstrumentKey, BinanceContinuousKline)>
    for MarketIter<InstrumentKey, Candle>
{
    fn from(
        (exchange_id, instrument, kline): (ExchangeId, InstrumentKey, BinanceContinuousKline),
    ) -> Self {
        kline.kline.into_market_iter(exchange_id, instrument)
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
mod tests {
    use super::*;
    use chrono::TimeZone;
    use rust_decimal_macros::dec;

    /// Spot `@kline_1m` frame for a closed candle.
    const SPOT_CLOSED: &str = r#"
    {
        "e": "kline", "E": 1638747660000, "s": "BTCUSDT",
        "k": {
            "t": 1638747660000, "T": 1638747719999, "s": "BTCUSDT", "i": "1m",
            "f": 100, "L": 200, "o": "0.0010", "c": "0.0020", "h": "0.0025",
            "l": "0.0015", "v": "1000", "n": 100, "x": true, "q": "1.0000",
            "V": "500", "Q": "0.500", "B": "123456"
        }
    }"#;

    /// Futures `continuousKline_1s` frame for an in-progress candle (note: no `k.s`).
    const FUTURES_OPEN: &str = r#"
    {
        "e": "continuous_kline", "E": 1607443058651, "ps": "BTCUSDT", "ct": "PERPETUAL",
        "k": {
            "t": 1607443020000, "T": 1607443079999, "i": "1s",
            "f": 116467658886, "L": 116468012423, "o": "18787.00", "c": "18804.04",
            "h": "18804.04", "l": "18786.54", "v": "197.664", "n": 543, "x": false,
            "q": "3715253.19494", "V": "184.769", "Q": "3472925.84746", "B": "0"
        }
    }"#;

    /// Futures `continuousKline_1m` frame for a **closed** candle (note: no `k.s`; the symbol
    /// source is the top-level `ps` pair).
    const FUTURES_CLOSED: &str = r#"
    {
        "e": "continuous_kline", "E": 1607443079999, "ps": "BTCUSDT", "ct": "PERPETUAL",
        "k": {
            "t": 1607443020000, "T": 1607443079999, "i": "1m",
            "f": 116467658886, "L": 116468012423, "o": "18787.00", "c": "18804.04",
            "h": "18810.00", "l": "18786.54", "v": "197.664", "n": 543, "x": true,
            "q": "3715253.19494", "V": "184.769", "Q": "3472925.84746", "B": "0"
        }
    }"#;

    #[test]
    fn spot_kline_deserialises_and_builds_map_key() {
        let kline = serde_json::from_str::<BinanceKline>(SPOT_CLOSED).unwrap();
        assert_eq!(kline.symbol, "BTCUSDT");
        assert_eq!(kline.kline.interval, CandleInterval::Min1);
        assert!(kline.kline.closed);
        assert_eq!(kline.kline.open, dec!(0.0010));
        assert_eq!(
            kline.id(),
            Some(SubscriptionId::from("@kline_1m|BTCUSDT")),
            "map key must be {{channel}}|{{MARKET}}, not the lowercase stream name"
        );
    }

    #[test]
    fn futures_continuous_kline_deserialises_without_k_s() {
        let kline = serde_json::from_str::<BinanceContinuousKline>(FUTURES_OPEN).unwrap();
        assert_eq!(kline.pair, "BTCUSDT");
        assert_eq!(kline.kline.interval, CandleInterval::Sec1);
        assert!(!kline.kline.closed);
        assert_eq!(
            kline.id(),
            Some(SubscriptionId::from(
                "_perpetual@continuousKline_1s|BTCUSDT"
            ))
        );
    }

    #[test]
    fn high_precision_ohlcv_round_trips_via_decimal_not_f64() {
        // A value an f64 intermediate would truncate; str→Decimal must preserve it exactly.
        let input = r#"
        {
            "e": "kline", "E": 1, "s": "ETHUSDT",
            "k": { "t": 0, "T": 1, "s": "ETHUSDT", "i": "1m", "o": "0.000000010000000",
                   "c": "0", "h": "0", "l": "0", "v": "0", "n": 0, "x": true }
        }"#;
        let kline = serde_json::from_str::<BinanceKline>(input).unwrap();
        assert_eq!(kline.kline.open, dec!(0.000000010000000));
    }

    #[test]
    fn closed_candle_maps_to_candle_with_boundary_close_time() {
        let kline = serde_json::from_str::<BinanceKline>(SPOT_CLOSED).unwrap();
        let MarketIter(events) =
            MarketIter::<u64, Candle>::from((ExchangeId::BinanceSpot, 1u64, kline));
        assert_eq!(events.len(), 1);
        let event = events.into_iter().next().unwrap().unwrap();
        // close_time = open (1638747660000ms) + 1m = 1638747720000ms — NOT the wire `T`
        // (1638747719999ms, the `period-end − 1ms` convention).
        let expected = Utc.timestamp_millis_opt(1638747720000).unwrap();
        assert_eq!(event.kind.close_time, expected);
        assert_eq!(event.time_exchange, expected);
        assert_eq!(event.kind.trade_count, 100);
    }

    #[test]
    fn in_progress_candle_emits_nothing() {
        let kline = serde_json::from_str::<BinanceContinuousKline>(FUTURES_OPEN).unwrap();
        let MarketIter(events) =
            MarketIter::<u64, Candle>::from((ExchangeId::BinanceFuturesUsd, 1u64, kline));
        assert!(events.is_empty(), "in-progress klines must not emit events");
    }

    #[test]
    fn closed_continuous_kline_maps_with_boundary_close_time_via_ps_pair() {
        // Symmetric to `closed_candle_maps_to_candle_with_boundary_close_time`, but for the
        // futures continuous frame: the symbol comes from `ps` (no `k.s`) and the OHLCV must
        // map through unchanged.
        let kline = serde_json::from_str::<BinanceContinuousKline>(FUTURES_CLOSED).unwrap();
        let MarketIter(events) =
            MarketIter::<u64, Candle>::from((ExchangeId::BinanceFuturesUsd, 7u64, kline));
        assert_eq!(
            events.len(),
            1,
            "a closed continuous kline must emit exactly one candle"
        );
        let event = events.into_iter().next().unwrap().unwrap();
        // close_time = open (1607443020000ms) + 1m = 1607443080000ms — the exclusive boundary,
        // NOT the wire `T` (1607443079999ms).
        let expected = Utc.timestamp_millis_opt(1607443080000).unwrap();
        assert_eq!(event.kind.close_time, expected);
        assert_eq!(event.time_exchange, expected);
        assert_eq!(event.kind.open, dec!(18787.00));
        assert_eq!(event.kind.high, dec!(18810.00));
        assert_eq!(event.kind.low, dec!(18786.54));
        assert_eq!(event.kind.close, dec!(18804.04));
        assert_eq!(event.kind.volume, dec!(197.664));
        assert_eq!(event.kind.trade_count, 543);
    }

    #[test]
    fn zero_volume_candle_is_delivered_not_filtered() {
        // Binance REST gap-fills zero-trade periods (V=0, OHLC=prev_close); the library must
        // not drop them. (WS omits them, but if one arrives it is still delivered.)
        let input = r#"
        {
            "e": "kline", "E": 1, "s": "BTCUSDT",
            "k": { "t": 0, "T": 59999, "s": "BTCUSDT", "i": "1m", "o": "100", "c": "100",
                   "h": "100", "l": "100", "v": "0", "n": 0, "x": true }
        }"#;
        let kline = serde_json::from_str::<BinanceKline>(input).unwrap();
        let MarketIter(events) =
            MarketIter::<u64, Candle>::from((ExchangeId::BinanceSpot, 1u64, kline));
        assert_eq!(events.len(), 1);
        let event = events.into_iter().next().unwrap().unwrap();
        assert_eq!(event.kind.volume, dec!(0));
        assert_eq!(event.kind.trade_count, 0);
    }
}