wickra_exchange_core/
feeds.rs

1//! Typed derivatives feeds and zero-glue conversions into the wickra-core
2//! indicator input types.
3//!
4//! An exchange publishes microstructure over several channels — trades, depth,
5//! funding, open interest, liquidations, positioning. This module gives each a
6//! typed representation and a **direct conversion into the exact value type the
7//! wickra-core indicator families consume**, so a recorded or live frame feeds an
8//! indicator with no hand-written glue:
9//!
10//! | Source (this crate)                     | Target (`wickra_core`) |
11//! |-----------------------------------------|------------------------|
12//! | [`TradePrint`]                          | [`Trade`]              |
13//! | [`OrderBookSnapshot`]                   | [`OrderBook`]          |
14//! | [`DerivativesFeed`] (via [`DerivativesTickBuilder`]) | [`DerivativesTick`] |
15//! | [`BreadthMember`] slice                 | [`CrossSection`]       |
16//!
17//! Every venue client emits [`TradePrint`] and [`OrderBookSnapshot`] on its
18//! stream, and the futures clients emit [`DerivativesFeed`] frames as
19//! [`Event::Derivatives`](crate::events::Event::Derivatives) once the channel is
20//! subscribed with
21//! [`DerivativesStream`](crate::traits::DerivativesStream). Open interest and
22//! long/short positioning are polled rather than pushed -- no venue streams them
23//! -- so they are read through that same trait and returned directly.
24//!
25//! A [`BreadthMember`] universe is still assembled by the caller from tickers.
26//!
27//! The derivatives channels arrive independently, so [`DerivativesTickBuilder`]
28//! folds each into a running tick and emits a validated [`DerivativesTick`] on
29//! demand.
30
31use rust_decimal::prelude::ToPrimitive;
32use rust_decimal::Decimal;
33use serde::{Deserialize, Serialize};
34use wickra_core::{CrossSection, DerivativesTick, Level, Member, OrderBook, Side, Trade};
35
36use crate::error::{Error, Result};
37use crate::events::{OrderBookSnapshot, TradePrint};
38use crate::symbol::Symbol;
39use crate::types::OrderSide;
40
41/// Convert an exact [`Decimal`] to the `f64` the indicator core consumes. A
42/// value outside `f64`'s range becomes `NaN`, which the core's own validators
43/// reject — so an out-of-range figure surfaces as a conversion error rather than
44/// a silent zero.
45fn to_f64(value: Decimal) -> f64 {
46    value.to_f64().unwrap_or(f64::NAN)
47}
48
49/// Map an exchange aggressor side to the core's trade side.
50fn core_side(side: OrderSide) -> Side {
51    match side {
52        OrderSide::Buy => Side::Buy,
53        OrderSide::Sell => Side::Sell,
54    }
55}
56
57/// Convert a public [`TradePrint`] into the core [`Trade`] value consumed by the
58/// microstructure indicator family.
59///
60/// # Errors
61///
62/// Returns [`Error::Deserialization`] if the price or size is out of the core's
63/// valid range (non-finite, non-positive price, or negative size).
64pub fn trade_from_print(print: &TradePrint) -> Result<Trade> {
65    Trade::new(
66        to_f64(print.price),
67        to_f64(print.quantity),
68        core_side(print.aggressor),
69        print.timestamp,
70    )
71    .map_err(|e| Error::Deserialization(e.to_string()))
72}
73
74/// Convert a depth [`OrderBookSnapshot`] into the core [`OrderBook`] consumed by
75/// the order-book imbalance / order-flow indicators.
76///
77/// # Errors
78///
79/// Returns [`Error::Deserialization`] if either side is empty, a level is
80/// invalid, the sides are not strictly best-first, or the book is crossed.
81pub fn order_book_from_snapshot(snapshot: &OrderBookSnapshot) -> Result<OrderBook> {
82    let to_levels = |levels: &[crate::events::BookLevel]| -> Result<Vec<Level>> {
83        levels
84            .iter()
85            .map(|level| {
86                Level::new(to_f64(level.price), to_f64(level.quantity))
87                    .map_err(|e| Error::Deserialization(e.to_string()))
88            })
89            .collect()
90    };
91    OrderBook::new(to_levels(&snapshot.bids)?, to_levels(&snapshot.asks)?)
92        .map_err(|e| Error::Deserialization(e.to_string()))
93}
94
95/// One symbol's contribution to a market-breadth cross-section.
96#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
97pub struct BreadthMember {
98    /// Price change versus the previous close (sign classifies advance/decline).
99    pub change: Decimal,
100    /// Period volume for the symbol (non-negative).
101    pub volume: Decimal,
102    /// Whether the symbol printed a new period high.
103    pub new_high: bool,
104    /// Whether the symbol printed a new period low.
105    pub new_low: bool,
106}
107
108/// Convert a universe of [`BreadthMember`]s into the core [`CrossSection`]
109/// consumed by the breadth indicator family.
110///
111/// # Errors
112///
113/// Returns [`Error::Deserialization`] if the universe is empty or a member's
114/// change/volume is out of the core's valid range.
115pub fn cross_section(members: &[BreadthMember], timestamp: i64) -> Result<CrossSection> {
116    let members = members
117        .iter()
118        .map(|member| {
119            Member::new(
120                to_f64(member.change),
121                to_f64(member.volume),
122                member.new_high,
123                member.new_low,
124            )
125        })
126        .collect();
127    CrossSection::new(members, timestamp).map_err(|e| Error::Deserialization(e.to_string()))
128}
129
130/// A funding-rate update for a perpetual market.
131#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
132pub struct FundingRate {
133    /// The market.
134    pub symbol: Symbol,
135    /// Funding rate for the interval (may be negative).
136    pub rate: Decimal,
137    /// The perpetual mark price at the funding print.
138    pub mark_price: Decimal,
139    /// Venue timestamp (milliseconds since the Unix epoch).
140    pub timestamp: i64,
141}
142
143/// An open-interest update: outstanding contracts / notional.
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
145pub struct OpenInterest {
146    /// The market.
147    pub symbol: Symbol,
148    /// Open interest (non-negative).
149    pub open_interest: Decimal,
150    /// Venue timestamp (milliseconds since the Unix epoch).
151    pub timestamp: i64,
152}
153
154/// A forced liquidation print.
155#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
156pub struct Liquidation {
157    /// The market.
158    pub symbol: Symbol,
159    /// The side of the forced order hitting the market: a liquidated **long**
160    /// sells ([`OrderSide::Sell`]), a liquidated **short** buys
161    /// ([`OrderSide::Buy`]).
162    pub side: OrderSide,
163    /// Liquidation price.
164    pub price: Decimal,
165    /// Liquidated quantity.
166    pub quantity: Decimal,
167    /// Venue timestamp (milliseconds since the Unix epoch).
168    pub timestamp: i64,
169}
170
171/// A long/short positioning ratio update.
172#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
173pub struct LongShortRatio {
174    /// The market.
175    pub symbol: Symbol,
176    /// Aggregate long size / long account count (non-negative).
177    pub long_size: Decimal,
178    /// Aggregate short size / short account count (non-negative).
179    pub short_size: Decimal,
180    /// Venue timestamp (milliseconds since the Unix epoch).
181    pub timestamp: i64,
182}
183
184/// A mark/index price update.
185#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
186pub struct MarkIndex {
187    /// The market.
188    pub symbol: Symbol,
189    /// Perpetual mark price (strictly positive).
190    pub mark_price: Decimal,
191    /// Spot / index price the perpetual tracks (strictly positive).
192    pub index_price: Decimal,
193    /// Venue timestamp (milliseconds since the Unix epoch).
194    pub timestamp: i64,
195}
196
197/// A derivatives channel a venue can **push** over its WebSocket.
198///
199/// Not every channel is a stream. Open interest and long/short positioning are
200/// polled figures on every venue in this crate -- they are published on a fixed
201/// cadence over REST and no venue pushes them -- so they are read with
202/// [`DerivativesStream::open_interest`](crate::traits::DerivativesStream::open_interest)
203/// and
204/// [`long_short_ratio`](crate::traits::DerivativesStream::long_short_ratio)
205/// rather than subscribed to. Modelling them as subscriptions would have made
206/// the API look symmetric and the data arrive never.
207#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
208pub enum DerivativesChannel {
209    /// The funding rate and the mark price it is charged against.
210    Funding,
211    /// Mark price and the index price the perpetual tracks.
212    MarkIndex,
213    /// Forced liquidations hitting the book.
214    Liquidations,
215}
216
217/// One channel of the derivatives microstructure feed.
218#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
219pub enum DerivativesFeed {
220    /// A funding-rate print.
221    Funding(FundingRate),
222    /// An open-interest print.
223    OpenInterest(OpenInterest),
224    /// A forced-liquidation print.
225    Liquidation(Liquidation),
226    /// A long/short positioning print.
227    LongShortRatio(LongShortRatio),
228    /// A mark/index price print.
229    MarkIndex(MarkIndex),
230}
231
232/// Folds independent [`DerivativesFeed`] channels into a running
233/// [`DerivativesTick`].
234///
235/// The derivatives channels (funding, open interest, liquidations, positioning,
236/// mark/index) arrive on their own cadences; apply each as it lands, then
237/// [`build`](Self::build) a validated tick for the perpetual-futures indicator
238/// family. Liquidation notionals accumulate until [`reset_flows`](Self::reset_flows)
239/// clears them for the next interval.
240#[derive(Debug, Clone, Default)]
241pub struct DerivativesTickBuilder {
242    funding_rate: f64,
243    mark_price: f64,
244    index_price: f64,
245    futures_price: f64,
246    open_interest: f64,
247    long_size: f64,
248    short_size: f64,
249    long_liquidation: f64,
250    short_liquidation: f64,
251    timestamp: i64,
252}
253
254impl DerivativesTickBuilder {
255    /// A fresh builder with every field zeroed.
256    #[must_use]
257    pub fn new() -> Self {
258        Self::default()
259    }
260
261    /// Fold one feed channel into the running tick.
262    pub fn apply(&mut self, feed: &DerivativesFeed) {
263        match feed {
264            DerivativesFeed::Funding(funding) => {
265                self.funding_rate = to_f64(funding.rate);
266                self.mark_price = to_f64(funding.mark_price);
267                self.timestamp = funding.timestamp;
268            }
269            DerivativesFeed::OpenInterest(oi) => {
270                self.open_interest = to_f64(oi.open_interest);
271                self.timestamp = oi.timestamp;
272            }
273            DerivativesFeed::Liquidation(liq) => {
274                let notional = to_f64(liq.price * liq.quantity);
275                match liq.side {
276                    OrderSide::Sell => self.long_liquidation += notional,
277                    OrderSide::Buy => self.short_liquidation += notional,
278                }
279                self.timestamp = liq.timestamp;
280            }
281            DerivativesFeed::LongShortRatio(ratio) => {
282                self.long_size = to_f64(ratio.long_size);
283                self.short_size = to_f64(ratio.short_size);
284                self.timestamp = ratio.timestamp;
285            }
286            DerivativesFeed::MarkIndex(mark) => {
287                self.mark_price = to_f64(mark.mark_price);
288                self.index_price = to_f64(mark.index_price);
289                // Absent a dated-futures channel, the perpetual tracks its own mark.
290                self.futures_price = to_f64(mark.mark_price);
291                self.timestamp = mark.timestamp;
292            }
293        }
294    }
295
296    /// Clear the interval-accumulated liquidation notionals.
297    pub fn reset_flows(&mut self) {
298        self.long_liquidation = 0.0;
299        self.short_liquidation = 0.0;
300    }
301
302    /// Build a validated [`DerivativesTick`] from the folded state.
303    ///
304    /// # Errors
305    ///
306    /// Returns [`Error::Deserialization`] if a required price channel
307    /// (mark/index/futures) has not yet been applied, so the tick would violate
308    /// the core's positivity invariants.
309    pub fn build(&self) -> Result<DerivativesTick> {
310        DerivativesTick::new(
311            self.funding_rate,
312            self.mark_price,
313            self.index_price,
314            self.futures_price,
315            self.open_interest,
316            self.long_size,
317            self.short_size,
318            // Taker buy/sell volume is carried by the trade channel, not these
319            // derivatives feeds; it stays zero until a trade-derived source sets it.
320            0.0,
321            0.0,
322            self.long_liquidation,
323            self.short_liquidation,
324            self.timestamp,
325        )
326        .map_err(|e| Error::Deserialization(e.to_string()))
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333    use crate::events::BookLevel;
334    use rust_decimal_macros::dec;
335
336    fn sym() -> Symbol {
337        Symbol::new("BTC", "USDT")
338    }
339
340    #[test]
341    fn trade_print_converts_to_core_trade() {
342        let print = TradePrint {
343            symbol: sym(),
344            price: dec!(20000.5),
345            quantity: dec!(0.25),
346            aggressor: OrderSide::Sell,
347            timestamp: 1,
348        };
349        let trade = trade_from_print(&print).unwrap();
350        assert!((trade.price - 20000.5).abs() < 1e-9);
351        assert!((trade.size - 0.25).abs() < 1e-12);
352        assert_eq!(trade.side, Side::Sell);
353        assert_eq!(trade.timestamp, 1);
354    }
355
356    #[test]
357    fn snapshot_converts_to_core_order_book() {
358        let snapshot = OrderBookSnapshot {
359            symbol: sym(),
360            last_update_id: 7,
361            bids: vec![
362                BookLevel::new(dec!(100), dec!(2)),
363                BookLevel::new(dec!(99), dec!(3)),
364            ],
365            asks: vec![
366                BookLevel::new(dec!(101), dec!(1)),
367                BookLevel::new(dec!(102), dec!(4)),
368            ],
369            timestamp: 0,
370        };
371        let book = order_book_from_snapshot(&snapshot).unwrap();
372        assert!((book.best_bid().unwrap().price - 100.0).abs() < 1e-9);
373        assert!((book.best_ask().unwrap().price - 101.0).abs() < 1e-9);
374        assert_eq!(book.mid(), Some(100.5));
375    }
376
377    #[test]
378    fn crossed_snapshot_is_rejected() {
379        let snapshot = OrderBookSnapshot {
380            symbol: sym(),
381            last_update_id: 0,
382            bids: vec![BookLevel::new(dec!(102), dec!(1))],
383            asks: vec![BookLevel::new(dec!(101), dec!(1))],
384            timestamp: 0,
385        };
386        assert!(matches!(
387            order_book_from_snapshot(&snapshot).unwrap_err(),
388            Error::Deserialization(_)
389        ));
390    }
391
392    #[test]
393    fn breadth_universe_converts_to_cross_section() {
394        let members = [
395            BreadthMember {
396                change: dec!(1.5),
397                volume: dec!(1000),
398                new_high: true,
399                new_low: false,
400            },
401            BreadthMember {
402                change: dec!(-0.5),
403                volume: dec!(500),
404                new_high: false,
405                new_low: true,
406            },
407        ];
408        let section = cross_section(&members, 42).unwrap();
409        assert_eq!(section.advancers(), 1);
410        assert_eq!(section.decliners(), 1);
411        assert_eq!(section.timestamp, 42);
412    }
413
414    #[test]
415    fn empty_universe_is_rejected() {
416        assert!(matches!(
417            cross_section(&[], 0).unwrap_err(),
418            Error::Deserialization(_)
419        ));
420    }
421
422    #[test]
423    fn derivatives_channels_fold_into_a_tick() {
424        let mut builder = DerivativesTickBuilder::new();
425        builder.apply(&DerivativesFeed::MarkIndex(MarkIndex {
426            symbol: sym(),
427            mark_price: dec!(20000),
428            index_price: dec!(19995),
429            timestamp: 10,
430        }));
431        builder.apply(&DerivativesFeed::Funding(FundingRate {
432            symbol: sym(),
433            rate: dec!(-0.0001),
434            mark_price: dec!(20010),
435            timestamp: 11,
436        }));
437        builder.apply(&DerivativesFeed::OpenInterest(OpenInterest {
438            symbol: sym(),
439            open_interest: dec!(1234.5),
440            timestamp: 12,
441        }));
442        builder.apply(&DerivativesFeed::LongShortRatio(LongShortRatio {
443            symbol: sym(),
444            long_size: dec!(60),
445            short_size: dec!(40),
446            timestamp: 13,
447        }));
448        // A liquidated long is a forced sell.
449        builder.apply(&DerivativesFeed::Liquidation(Liquidation {
450            symbol: sym(),
451            side: OrderSide::Sell,
452            price: dec!(20000),
453            quantity: dec!(0.5),
454            timestamp: 14,
455        }));
456
457        let tick = builder.build().unwrap();
458        assert!((tick.funding_rate + 0.0001).abs() < 1e-12);
459        assert!((tick.mark_price - 20010.0).abs() < 1e-9);
460        assert!((tick.index_price - 19995.0).abs() < 1e-9);
461        assert!((tick.open_interest - 1234.5).abs() < 1e-9);
462        assert!((tick.long_size - 60.0).abs() < 1e-9);
463        assert!((tick.long_liquidation - 10000.0).abs() < 1e-9);
464        assert!(tick.short_liquidation.abs() < 1e-9);
465        assert_eq!(tick.timestamp, 14);
466    }
467
468    #[test]
469    fn short_liquidation_accumulates_and_resets() {
470        let mut builder = DerivativesTickBuilder::new();
471        builder.apply(&DerivativesFeed::MarkIndex(MarkIndex {
472            symbol: sym(),
473            mark_price: dec!(100),
474            index_price: dec!(100),
475            timestamp: 1,
476        }));
477        // A liquidated short is a forced buy; accumulate two of them.
478        for _ in 0..2 {
479            builder.apply(&DerivativesFeed::Liquidation(Liquidation {
480                symbol: sym(),
481                side: OrderSide::Buy,
482                price: dec!(100),
483                quantity: dec!(1),
484                timestamp: 2,
485            }));
486        }
487        assert!((builder.build().unwrap().short_liquidation - 200.0).abs() < 1e-9);
488        builder.reset_flows();
489        assert!(builder.build().unwrap().short_liquidation.abs() < 1e-9);
490    }
491
492    #[test]
493    fn build_before_a_price_channel_is_rejected() {
494        // No mark/index applied: prices are zero, violating the core's positivity.
495        assert!(matches!(
496            DerivativesTickBuilder::new().build().unwrap_err(),
497            Error::Deserialization(_)
498        ));
499    }
500}
wickra_exchange_core/feeds.rs

wickra_exchange_core/
feeds.rs
