brokapi_finam/map/
market.rs

1//! Order book, quotes, trades, bars.
2
3use std::collections::BTreeMap;
4use std::time::Instant;
5
6use brokapi_core::{
7    BookLevel, Candle, Error, InstrumentId, MarketTrade, OrderBook, Price, Qty, Quote, Side,
8    Timeframe, UnixNanos,
9};
10use brokapi_finam_protocol::pb::google::r#type::Decimal;
11use brokapi_finam_protocol::pb::grpc::tradeapi::v1::Side as PbSide;
12use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata as pb;
13use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata::{
14    TimeFrame, order_book, stream_order_book,
15};
16use chrono::{DateTime, Datelike, Months, TimeZone, Utc};
17
18use crate::error::unsupported;
19use crate::util::{price, price0, qty0, timestamps, ts_nanos};
20
21/// Row action of a stream update.
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub enum RowAction {
24    /// The level is removed from the book.
25    Remove,
26    /// ADD, UPDATE and an unspecified action all mean "set this level to this size".
27    Set,
28}
29
30/// Side-neutral view of a book row (unary and stream rows are different generated types).
31#[derive(Debug, Clone, Copy, PartialEq)]
32pub struct RowView {
33    /// Price of the level.
34    pub price: Price,
35    /// `sell_size` rows are asks, `buy_size` rows are bids.
36    pub is_ask: bool,
37    /// Size of the level in pieces. A zero size on a [`RowAction::Set`] row removes the level.
38    pub size: Qty,
39    /// What to do with the level.
40    pub action: RowAction,
41    /// Time of the row at the broker, when it sends one.
42    pub ts: Option<UnixNanos>,
43}
44
45/// Rows of a unary `OrderBook` answer in the neutral form. Rows with no side or no price are skipped.
46pub fn rest_rows(rows: &[order_book::Row]) -> Vec<RowView> {
47    rows.iter()
48        .filter_map(|r| {
49            let (is_ask, size) = match &r.side {
50                Some(order_book::row::Side::SellSize(d)) => (true, qty0(&Some(d.clone()))),
51                Some(order_book::row::Side::BuySize(d)) => (false, qty0(&Some(d.clone()))),
52                None => return None,
53            };
54            Some(RowView {
55                price: price(&r.price)?,
56                is_ask,
57                size,
58                action: if r.action == order_book::row::Action::Remove as i32 {
59                    RowAction::Remove
60                } else {
61                    RowAction::Set
62                },
63                ts: r.timestamp.as_ref().map(ts_nanos),
64            })
65        })
66        .collect()
67}
68
69/// Rows of an order book stream message in the neutral form. Rows with no side or no price are skipped.
70pub fn stream_rows(rows: &[stream_order_book::Row]) -> Vec<RowView> {
71    rows.iter()
72        .filter_map(|r| {
73            let (is_ask, size) = match &r.side {
74                Some(stream_order_book::row::Side::SellSize(d)) => (true, qty0(&Some(d.clone()))),
75                Some(stream_order_book::row::Side::BuySize(d)) => (false, qty0(&Some(d.clone()))),
76                None => return None,
77            };
78            Some(RowView {
79                price: price(&r.price)?,
80                is_ask,
81                size,
82                action: if r.action == stream_order_book::row::Action::Remove as i32 {
83                    RowAction::Remove
84                } else {
85                    RowAction::Set
86                },
87                ts: r.timestamp.as_ref().map(ts_nanos),
88            })
89        })
90        .collect()
91}
92
93/// Result of applying rows to a [`BookState`].
94#[derive(Debug, Clone, Copy, PartialEq, Eq)]
95pub enum ApplyOutcome {
96    /// The rows were applied to the book.
97    Applied,
98    /// An increment arrived while no snapshot is held: the book is unusable until resubscribed.
99    NeedSnapshot,
100}
101
102/// Local order book rebuilt from Finam's snapshot + action rows.
103///
104/// # Examples
105///
106/// ```
107/// use brokapi_finam::map::market::{ApplyOutcome, BookState, RowAction, RowView};
108///
109/// let row = |price: &str, is_ask, size: &str, action| RowView {
110///     price: price.parse().unwrap(),
111///     is_ask,
112///     size: size.parse().unwrap(),
113///     action,
114///     ts: None,
115/// };
116/// let mut book = BookState::default();
117///
118/// // An increment with no snapshot under it is refused: the stream must be reopened.
119/// let late = [row("100", false, "5", RowAction::Set)];
120/// assert_eq!(book.apply(&late, false), ApplyOutcome::NeedSnapshot);
121///
122/// // A snapshot replaces the book, then rows change single levels.
123/// let snapshot = [row("100", false, "5", RowAction::Set), row("101", true, "3", RowAction::Set)];
124/// assert_eq!(book.apply(&snapshot, true), ApplyOutcome::Applied);
125/// assert!(book.has_snapshot());
126/// book.apply(&[row("100", false, "0", RowAction::Remove), row("99", false, "7", RowAction::Set)], false);
127///
128/// let (bids, asks) = book.levels(10);
129/// assert_eq!(bids.len(), 1);
130/// assert_eq!(bids[0].price, "99".parse()?);
131/// assert_eq!(asks[0].qty, "3".parse()?);
132/// # Ok::<(), brokapi_core::NumberError>(())
133/// ```
134#[derive(Debug, Clone, Default)]
135pub struct BookState {
136    bids: BTreeMap<Price, Qty>,
137    asks: BTreeMap<Price, Qty>,
138    have_snapshot: bool,
139}
140
141impl BookState {
142    /// Forgets the whole book, including the snapshot. The next increment is refused until a new snapshot.
143    pub fn reset(&mut self) {
144        *self = Self::default();
145    }
146
147    /// `true` once a snapshot was applied since creation or the last [`reset`](Self::reset).
148    pub fn has_snapshot(&self) -> bool {
149        self.have_snapshot
150    }
151
152    /// `is_snapshot` replaces the whole book (a snapshot message is assumed to carry all levels).
153    ///
154    /// Without a snapshot an increment is not applied and the result is [`ApplyOutcome::NeedSnapshot`].
155    pub fn apply(&mut self, rows: &[RowView], is_snapshot: bool) -> ApplyOutcome {
156        if is_snapshot {
157            self.bids.clear();
158            self.asks.clear();
159            self.have_snapshot = true;
160        } else if !self.have_snapshot {
161            return ApplyOutcome::NeedSnapshot;
162        }
163        for r in rows {
164            let side = if r.is_ask {
165                &mut self.asks
166            } else {
167                &mut self.bids
168            };
169            let key = r.price;
170            match r.action {
171                RowAction::Remove => {
172                    side.remove(&key);
173                }
174                RowAction::Set if r.size.is_positive() => {
175                    side.insert(key, r.size);
176                }
177                // A zero size is a removal in disguise.
178                RowAction::Set => {
179                    side.remove(&key);
180                }
181            }
182        }
183        ApplyOutcome::Applied
184    }
185
186    /// Best-first levels, at most `depth` per side.
187    pub fn levels(&self, depth: usize) -> (Vec<BookLevel>, Vec<BookLevel>) {
188        let bids = self
189            .bids
190            .iter()
191            .rev()
192            .take(depth)
193            .map(|(p, q)| level(*p, *q))
194            .collect();
195        let asks = self
196            .asks
197            .iter()
198            .take(depth)
199            .map(|(p, q)| level(*p, *q))
200            .collect();
201        (bids, asks)
202    }
203}
204
205fn level(price: Price, qty: Qty) -> BookLevel {
206    BookLevel { price, qty }
207}
208
209/// Full snapshot event from the local book.
210pub fn book_snapshot(
211    instrument: &InstrumentId,
212    state: &BookState,
213    depth: usize,
214    broker_ts: Option<UnixNanos>,
215    now: Instant,
216) -> OrderBook {
217    let (bids, asks) = state.levels(depth);
218    let mut ts = timestamps(None, now);
219    ts.broker_ts = broker_ts;
220    OrderBook::Snapshot {
221        instrument: instrument.clone(),
222        bids,
223        asks,
224        ts,
225    }
226}
227
228/// Unary `OrderBook` response -> snapshot.
229pub fn map_rest_order_book(
230    instrument: &InstrumentId,
231    resp: &pb::OrderBookResponse,
232    depth: u32,
233    now: Instant,
234) -> OrderBook {
235    let rows = resp
236        .orderbook
237        .as_ref()
238        .map(|b| rest_rows(&b.rows))
239        .unwrap_or_default();
240    let mut state = BookState::default();
241    state.apply(&rows, true);
242    let broker_ts = rows.iter().filter_map(|r| r.ts).max();
243    book_snapshot(instrument, &state, depth as usize, broker_ts, now)
244}
245
246/// `Quote` -> core quote. Finam sends 0 for a missing side.
247pub fn map_quote(instrument: &InstrumentId, q: &pb::Quote, now: Instant) -> Quote {
248    let side = |price: &Option<Decimal>, size: &Option<Decimal>| {
249        let p = price0(price);
250        p.is_positive().then(|| level(p, qty0(size)))
251    };
252    Quote {
253        instrument: instrument.clone(),
254        bid: side(&q.bid, &q.bid_size),
255        ask: side(&q.ask, &q.ask_size),
256        ts: timestamps(q.timestamp.as_ref(), now),
257    }
258}
259
260/// `Trade` -> market trade. A trade without a side cannot be expressed (core `Side` is
261/// mandatory), so it is dropped by returning `None`.
262pub fn map_trade(instrument: &InstrumentId, t: &pb::Trade, now: Instant) -> Option<MarketTrade> {
263    let side = match PbSide::try_from(t.side).unwrap_or(PbSide::Unspecified) {
264        PbSide::Buy => Side::Buy,
265        PbSide::Sell => Side::Sell,
266        PbSide::Unspecified => return None,
267    };
268    Some(MarketTrade {
269        instrument: instrument.clone(),
270        price: price0(&t.price),
271        qty: qty0(&t.size),
272        side,
273        trade_id: (!t.trade_id.is_empty()).then(|| t.trade_id.clone()),
274        ts: timestamps(t.timestamp.as_ref(), now),
275    })
276}
277
278/// Core timeframe -> Finam. M10 has no Finam equivalent (see `wire_timeframe`).
279pub fn timeframe_to_pb(tf: Timeframe) -> Result<TimeFrame, Error> {
280    Ok(match tf {
281        Timeframe::M1 => TimeFrame::M1,
282        Timeframe::M5 => TimeFrame::M5,
283        Timeframe::M15 => TimeFrame::M15,
284        Timeframe::M30 => TimeFrame::M30,
285        Timeframe::H1 => TimeFrame::H1,
286        Timeframe::H4 => TimeFrame::H4,
287        Timeframe::D1 => TimeFrame::D,
288        Timeframe::W1 => TimeFrame::W,
289        Timeframe::MN => TimeFrame::Mn,
290        Timeframe::M10 => return Err(unsupported("timeframe_m10")),
291    })
292}
293
294/// Timeframe asked from Finam for a core timeframe: the same one, except M10, which is built from M5
295/// candles (`map::m10`).
296pub fn wire_timeframe(tf: Timeframe) -> Result<TimeFrame, Error> {
297    match tf {
298        Timeframe::M10 => Ok(TimeFrame::M5),
299        other => timeframe_to_pb(other),
300    }
301}
302
303/// Finam gives the bar open time only; close is open + frame length (UTC for week/month).
304pub fn bar_close(tf: Timeframe, open: UnixNanos) -> UnixNanos {
305    const SEC: i64 = 1_000_000_000;
306    let fixed = |secs: i64| UnixNanos::from_nanos(open.as_nanos() + secs * SEC);
307    match tf {
308        Timeframe::M1 => fixed(60),
309        Timeframe::M5 => fixed(300),
310        Timeframe::M10 => fixed(600),
311        Timeframe::M15 => fixed(900),
312        Timeframe::M30 => fixed(1800),
313        Timeframe::H1 => fixed(3600),
314        Timeframe::H4 => fixed(4 * 3600),
315        Timeframe::D1 => fixed(86_400),
316        Timeframe::W1 => fixed(7 * 86_400),
317        Timeframe::MN => {
318            let secs = open.as_nanos().div_euclid(SEC);
319            let dt: DateTime<Utc> = Utc.timestamp_opt(secs, 0).single().unwrap_or_default();
320            let next = dt
321                .checked_add_months(Months::new(1))
322                .map(|d| d.with_day(dt.day()).unwrap_or(d))
323                .unwrap_or(dt);
324            UnixNanos::from_nanos(next.timestamp() * SEC)
325        }
326    }
327}
328
329/// `Bar` -> core candle. Absent prices and volume become 0. `close_ts` is computed from the open time
330/// and the timeframe (month and longer frames follow the calendar), because the bar carries only
331/// the open time.
332pub fn map_bar(instrument: &InstrumentId, tf: Timeframe, b: &pb::Bar, now: Instant) -> Candle {
333    let open_ts = b
334        .timestamp
335        .as_ref()
336        .map(ts_nanos)
337        .unwrap_or(UnixNanos::ZERO);
338    Candle {
339        instrument: instrument.clone(),
340        timeframe: tf,
341        open: price0(&b.open),
342        high: price0(&b.high),
343        low: price0(&b.low),
344        close: price0(&b.close),
345        volume: qty0(&b.volume),
346        open_ts,
347        close_ts: bar_close(tf, open_ts),
348        ts: timestamps(b.timestamp.as_ref(), now),
349    }
350}
brokapi_finam/map/market.rs

brokapi_finam/map/
market.rs
