brokapi_tbank/map/
market.rs

1//! Order books, market trades, candles.
2
3use std::time::Instant;
4
5use brokapi_core::{
6    BookLevel, Candle, InstrumentId, MarketTrade, OrderBook, Price, Qty, Quote, Side, Timeframe,
7    UnixNanos,
8};
9use brokapi_tbank_protocol::v1 as pb;
10
11use crate::ids::{price_of_quotation, proto_to_nanos, stamps};
12
13/// Order book depths T-Bank accepts (`SUBSCRIPTION_STATUS_DEPTH_IS_INVALID`: 1, 10, 20, 30, 40, 50).
14pub const VALID_DEPTHS: [u32; 6] = [1, 10, 20, 30, 40, 50];
15/// The deepest order book T-Bank serves.
16pub const MAX_DEPTH: u32 = 50;
17
18/// Smallest valid wire depth that covers `depth` levels.
19pub fn wire_depth(depth: u32) -> u32 {
20    let depth = depth.max(1);
21    VALID_DEPTHS
22        .iter()
23        .copied()
24        .find(|d| *d >= depth)
25        .unwrap_or(MAX_DEPTH)
26}
27
28fn level(o: &pb::Order) -> BookLevel {
29    BookLevel {
30        price: price(&o.price),
31        qty: Qty::from(o.quantity),
32    }
33}
34
35/// Levels best-first: bids by price descending, asks ascending, cut to `depth`.
36pub fn levels(orders: &[pb::Order], bids: bool, depth: usize) -> Vec<BookLevel> {
37    let mut v: Vec<BookLevel> = orders.iter().map(level).collect();
38    v.sort_by(|a, b| {
39        let o = a.price.cmp(&b.price);
40        if bids { o.reverse() } else { o }
41    });
42    v.truncate(depth);
43    v
44}
45
46/// An order book snapshot from the two sides of a broker message, cut to `depth` levels.
47///
48/// Quantities are in lots, as the broker sends them (the connector multiplies by the lot size). The time is the broker's `orderbook_ts` when given.
49pub fn book_snapshot(
50    instrument: &InstrumentId,
51    bids: &[pb::Order],
52    asks: &[pb::Order],
53    depth: usize,
54    broker_ts: Option<&prost_types::Timestamp>,
55    now: Instant,
56) -> OrderBook {
57    OrderBook::Snapshot {
58        instrument: instrument.clone(),
59        bids: levels(bids, true, depth),
60        asks: levels(asks, false, depth),
61        ts: stamps(now, broker_ts),
62    }
63}
64
65/// The best bid and the best ask of a broker order book message.
66///
67/// T-Bank has no quote stream: a quote is the first level of each side. A side with no orders is
68/// `None`.
69pub fn quote(
70    instrument: &InstrumentId,
71    bids: &[pb::Order],
72    asks: &[pb::Order],
73    broker_ts: Option<&prost_types::Timestamp>,
74    now: Instant,
75) -> Quote {
76    Quote {
77        instrument: instrument.clone(),
78        bid: levels(bids, true, 1).into_iter().next(),
79        ask: levels(asks, false, 1).into_iter().next(),
80        ts: stamps(now, broker_ts),
81    }
82}
83
84/// Trades without a direction cannot be given a `Side` and are skipped.
85pub fn trade(instrument: &InstrumentId, t: &pb::Trade, now: Instant) -> Option<MarketTrade> {
86    let side = match t.direction() {
87        pb::TradeDirection::Buy => Side::Buy,
88        pb::TradeDirection::Sell => Side::Sell,
89        pb::TradeDirection::Unspecified => return None,
90    };
91    Some(MarketTrade {
92        instrument: instrument.clone(),
93        price: price(&t.price),
94        qty: Qty::from(t.quantity),
95        side,
96        trade_id: None,
97        ts: stamps(now, t.time.as_ref()),
98    })
99}
100
101/// Length of one candle of timeframe `tf`, in nanoseconds.
102///
103/// A month is taken as 30 days, because the broker's candle carries only its open time and the close
104/// time has to be derived.
105///
106/// # Examples
107///
108/// ```
109/// use brokapi_core::Timeframe;
110/// use brokapi_tbank::map::market::timeframe_duration_nanos;
111///
112/// assert_eq!(timeframe_duration_nanos(Timeframe::M5), 300_000_000_000);
113/// assert_eq!(timeframe_duration_nanos(Timeframe::MN), 30 * 24 * 3600 * 1_000_000_000);
114/// ```
115pub fn timeframe_duration_nanos(tf: Timeframe) -> i64 {
116    const MIN: i64 = 60 * 1_000_000_000;
117    match tf {
118        Timeframe::M1 => MIN,
119        Timeframe::M5 => 5 * MIN,
120        Timeframe::M10 => 10 * MIN,
121        Timeframe::M15 => 15 * MIN,
122        Timeframe::M30 => 30 * MIN,
123        Timeframe::H1 => 60 * MIN,
124        Timeframe::H4 => 240 * MIN,
125        Timeframe::D1 => 1440 * MIN,
126        Timeframe::W1 => 7 * 1440 * MIN,
127        Timeframe::MN => 30 * 1440 * MIN,
128    }
129}
130
131/// Timeframe to the interval name used when subscribing to candles.
132pub fn subscription_interval(tf: Timeframe) -> pb::SubscriptionInterval {
133    match tf {
134        Timeframe::M1 => pb::SubscriptionInterval::OneMinute,
135        Timeframe::M5 => pb::SubscriptionInterval::FiveMinutes,
136        Timeframe::M10 => pb::SubscriptionInterval::SubscriptionInterval10Min,
137        Timeframe::M15 => pb::SubscriptionInterval::FifteenMinutes,
138        Timeframe::M30 => pb::SubscriptionInterval::SubscriptionInterval30Min,
139        Timeframe::H1 => pb::SubscriptionInterval::OneHour,
140        Timeframe::H4 => pb::SubscriptionInterval::SubscriptionInterval4Hour,
141        Timeframe::D1 => pb::SubscriptionInterval::OneDay,
142        Timeframe::W1 => pb::SubscriptionInterval::Week,
143        Timeframe::MN => pb::SubscriptionInterval::Month,
144    }
145}
146
147/// The interval of a candle stream message to timeframe, or `None` for an interval with no
148/// counterpart in the domain model.
149pub fn timeframe_from_interval(i: pb::SubscriptionInterval) -> Option<Timeframe> {
150    Some(match i {
151        pb::SubscriptionInterval::OneMinute => Timeframe::M1,
152        pb::SubscriptionInterval::FiveMinutes => Timeframe::M5,
153        pb::SubscriptionInterval::SubscriptionInterval10Min => Timeframe::M10,
154        pb::SubscriptionInterval::FifteenMinutes => Timeframe::M15,
155        pb::SubscriptionInterval::SubscriptionInterval30Min => Timeframe::M30,
156        pb::SubscriptionInterval::OneHour => Timeframe::H1,
157        pb::SubscriptionInterval::SubscriptionInterval4Hour => Timeframe::H4,
158        pb::SubscriptionInterval::OneDay => Timeframe::D1,
159        pb::SubscriptionInterval::Week => Timeframe::W1,
160        pb::SubscriptionInterval::Month => Timeframe::MN,
161        _ => return None,
162    })
163}
164
165/// Timeframe to the interval name used when requesting historic candles.
166pub fn candle_interval(tf: Timeframe) -> pb::CandleInterval {
167    match tf {
168        Timeframe::M1 => pb::CandleInterval::CandleInterval1Min,
169        Timeframe::M5 => pb::CandleInterval::CandleInterval5Min,
170        Timeframe::M10 => pb::CandleInterval::CandleInterval10Min,
171        Timeframe::M15 => pb::CandleInterval::CandleInterval15Min,
172        Timeframe::M30 => pb::CandleInterval::CandleInterval30Min,
173        Timeframe::H1 => pb::CandleInterval::Hour,
174        Timeframe::H4 => pb::CandleInterval::CandleInterval4Hour,
175        Timeframe::D1 => pb::CandleInterval::Day,
176        Timeframe::W1 => pb::CandleInterval::Week,
177        Timeframe::MN => pb::CandleInterval::Month,
178    }
179}
180
181fn price(q: &Option<pb::Quotation>) -> Price {
182    q.as_ref().map(price_of_quotation).unwrap_or(Price::ZERO)
183}
184
185fn candle_times(open: &Option<prost_types::Timestamp>, tf: Timeframe) -> (UnixNanos, UnixNanos) {
186    let open_ts = open.as_ref().map(proto_to_nanos).unwrap_or(UnixNanos::ZERO);
187    (
188        open_ts,
189        UnixNanos::from_nanos(open_ts.as_nanos() + timeframe_duration_nanos(tf)),
190    )
191}
192
193/// A candle of the candle stream to [`Candle`].
194///
195/// Volume is in lots, as the broker sends it (the connector multiplies by the lot size). `close_ts` is derived from the open time and the timeframe. The candle is the
196/// current state of a candle that may still be forming.
197pub fn stream_candle(
198    instrument: &InstrumentId,
199    tf: Timeframe,
200    c: &pb::Candle,
201    now: Instant,
202) -> Candle {
203    let (open_ts, close_ts) = candle_times(&c.time, tf);
204    Candle {
205        instrument: instrument.clone(),
206        timeframe: tf,
207        open: price(&c.open),
208        high: price(&c.high),
209        low: price(&c.low),
210        close: price(&c.close),
211        volume: Qty::from(c.volume),
212        open_ts,
213        close_ts,
214        ts: stamps(now, c.last_trade_ts.as_ref().or(c.time.as_ref())),
215    }
216}
217
218/// A candle of `GetCandles` to [`Candle`].
219///
220/// Volume is in lots, as the broker sends it (the connector multiplies by the lot size). `close_ts` is derived from the open time and the timeframe.
221pub fn historic_candle(
222    instrument: &InstrumentId,
223    tf: Timeframe,
224    c: &pb::HistoricCandle,
225    now: Instant,
226) -> Candle {
227    let (open_ts, close_ts) = candle_times(&c.time, tf);
228    Candle {
229        instrument: instrument.clone(),
230        timeframe: tf,
231        open: price(&c.open),
232        high: price(&c.high),
233        low: price(&c.low),
234        close: price(&c.close),
235        volume: Qty::from(c.volume),
236        open_ts,
237        close_ts,
238        ts: stamps(now, c.time.as_ref()),
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245    use brokapi_core::{Decimal, dec};
246    use brokapi_core::{InstrumentKey, NativeId};
247    use brokapi_tbank_protocol::money::quotation;
248
249    fn id() -> InstrumentId {
250        InstrumentId::new(
251            InstrumentKey::new("SBER", "TQBR"),
252            NativeId::Uid("u".into()),
253        )
254    }
255
256    fn o(p: &str, q: i64) -> pb::Order {
257        pb::Order {
258            price: Some(quotation(p)),
259            quantity: q,
260        }
261    }
262
263    #[test]
264    fn depth_rounds_up_to_valid() {
265        assert_eq!(wire_depth(0), 1);
266        assert_eq!(wire_depth(1), 1);
267        assert_eq!(wire_depth(2), 10);
268        assert_eq!(wire_depth(20), 20);
269        assert_eq!(wire_depth(21), 30);
270        assert_eq!(wire_depth(500), 50);
271    }
272
273    #[test]
274    fn book_sorted_best_first_and_cut() {
275        let bids = [o("99.0", 5), o("100.0", 3), o("98.5", 1)];
276        let asks = [o("101.5", 2), o("100.5", 7)];
277        let OrderBook::Snapshot {
278            bids: b, asks: a, ..
279        } = book_snapshot(&id(), &bids, &asks, 2, None, Instant::now())
280        else {
281            panic!("snapshot expected");
282        };
283        assert_eq!(
284            b.iter().map(|l| l.price.decimal()).collect::<Vec<_>>(),
285            vec![dec!(100.0), dec!(99.0)]
286        );
287        assert_eq!(
288            a.iter().map(|l| l.price.decimal()).collect::<Vec<_>>(),
289            vec![dec!(100.5), dec!(101.5)]
290        );
291        assert_eq!(b[0].qty.decimal(), dec!(3.0));
292    }
293
294    #[test]
295    fn quote_is_best_level_of_each_side() {
296        let q = quote(
297            &id(),
298            &[o("99.0", 5), o("100.0", 3)],
299            &[o("100.5", 7)],
300            None,
301            Instant::now(),
302        );
303        assert_eq!(q.bid.unwrap().price.decimal(), dec!(100.0));
304        assert_eq!(q.ask.unwrap().qty.decimal(), dec!(7.0));
305    }
306
307    #[test]
308    fn empty_side_gives_no_quote_level() {
309        let q = quote(&id(), &[], &[o("100.5", 7)], None, Instant::now());
310        assert!(q.bid.is_none());
311    }
312
313    #[test]
314    fn trade_direction_and_units() {
315        let t = pb::Trade {
316            direction: pb::TradeDirection::Sell as i32,
317            price: Some(pb::Quotation {
318                units: 273,
319                nano: 450_000_000,
320            }),
321            quantity: 12,
322            ..Default::default()
323        };
324        let m = trade(&id(), &t, Instant::now()).unwrap();
325        assert_eq!(
326            (m.side, m.price.decimal(), m.qty.decimal()),
327            (Side::Sell, dec!(273.45), dec!(12.0))
328        );
329        let none = pb::Trade {
330            quantity: 1,
331            ..Default::default()
332        };
333        assert!(trade(&id(), &none, Instant::now()).is_none());
334    }
335
336    #[test]
337    fn timeframe_roundtrip() {
338        for tf in [
339            Timeframe::M1,
340            Timeframe::M5,
341            Timeframe::M10,
342            Timeframe::M15,
343            Timeframe::M30,
344            Timeframe::H1,
345            Timeframe::H4,
346            Timeframe::D1,
347            Timeframe::W1,
348            Timeframe::MN,
349        ] {
350            assert_eq!(timeframe_from_interval(subscription_interval(tf)), Some(tf));
351        }
352    }
353
354    #[test]
355    fn candle_close_time_from_timeframe() {
356        let c = pb::HistoricCandle {
357            open: Some(quotation("10.0")),
358            high: Some(quotation("12.0")),
359            low: Some(quotation("9.5")),
360            close: Some(quotation("11.0")),
361            volume: 100,
362            time: Some(prost_types::Timestamp {
363                seconds: 1_000,
364                nanos: 0,
365            }),
366            ..Default::default()
367        };
368        let m = historic_candle(&id(), Timeframe::M5, &c, Instant::now());
369        assert_eq!(m.open_ts.as_nanos(), 1_000_000_000_000);
370        assert_eq!(m.close_ts.as_nanos(), 1_000_000_000_000 + 300_000_000_000);
371        assert_eq!(
372            (m.high.decimal(), m.volume.decimal()),
373            (dec!(12.0), dec!(100.0))
374        );
375    }
376}
brokapi_tbank/map/market.rs

brokapi_tbank/map/
market.rs
