brokapi_bcs/map/
market.rs

1//! Market data mapping: BCS wire models to brokapi-core domain types.
2
3use brokapi_core::Decimal;
4use std::time::Instant;
5
6use brokapi_bcs_protocol::http::models::{CandlesResponse, LastTradeRecord, OrderBookResponse};
7use brokapi_bcs_protocol::ws::models::{
8    LastCandleData, LastTradesData, MarketDataMessage, OrderBookData, QuotesData,
9};
10use brokapi_core::{
11    BookLevel, Candle, Event, MarketTrade, OrderBook, Price, Qty, Quote, Side, Timeframe,
12};
13
14use super::{instrument_id_ticker, parse_broker_ts, timestamps};
15
16/// Maps the candles of `POST /candles-chart` to domain candles of the timeframe `tf` that was asked.
17///
18/// The API gives one time per bar. It is used for both `open_ts` and `close_ts`. Whether that time is
19/// the open or the close of the bar is not verified against the live API, so do not rely on the
20/// difference between the two fields.
21pub fn map_rest_candles(resp: &CandlesResponse, tf: Timeframe, now: Instant) -> Vec<Candle> {
22    let instrument = instrument_id_ticker(&resp.ticker, &resp.class_code);
23    resp.bars
24        .iter()
25        .map(|bar| {
26            let close_ts = super::unix_nanos(bar.time);
27            Candle {
28                instrument: instrument.clone(),
29                timeframe: tf,
30                open: Price::new(bar.open),
31                high: Price::new(bar.high),
32                low: Price::new(bar.low),
33                close: Price::new(bar.close),
34                volume: Qty::new(bar.volume),
35                open_ts: close_ts,
36                close_ts,
37                ts: timestamps(now, Some(close_ts)),
38            }
39        })
40        .collect()
41}
42
43/// Maps a REST order book answer to an [`Event::OrderBook`] holding an [`OrderBook::Snapshot`].
44///
45/// BCS gives full books only, so the result is always a snapshot. Levels keep the broker's order.
46///
47/// # Examples
48///
49/// ```
50/// use std::time::Instant;
51/// use brokapi_bcs_protocol::http::models::OrderBookResponse;
52/// use brokapi_bcs::map::market::map_rest_order_book;
53/// use brokapi_core::{Event, OrderBook};
54///
55/// let resp: OrderBookResponse = serde_json::from_str(
56///     r#"{"ticker":"SBER","classCode":"TQBR","bids":[{"price":299.9,"quantity":100}],
57///         "asks":[{"price":300.1,"quantity":50}]}"#,
58/// )
59/// .unwrap();
60/// let Event::OrderBook(OrderBook::Snapshot { bids, asks, .. }) = map_rest_order_book(&resp, Instant::now())
61/// else {
62///     panic!("a snapshot")
63/// };
64/// assert_eq!((bids.len(), asks.len()), (1, 1));
65/// ```
66pub fn map_rest_order_book(book: &OrderBookResponse, now: Instant) -> Event {
67    let instrument = instrument_id_ticker(&book.ticker, &book.class_code);
68    let broker_ts = parse_broker_ts(book.date_time.as_deref());
69    Event::OrderBook(map_order_book_snapshot(
70        &instrument,
71        &book.bids,
72        &book.asks,
73        timestamps(now, broker_ts),
74    ))
75}
76
77/// Maps a market data WebSocket message to an event.
78///
79/// `None` for messages that carry no data: subscribe confirmations, error answers, undefined
80/// messages, a trade with an unknown side, and a candle message that lacks a ticker, a class code or a price. `sub` is passed
81/// into the result unchanged. The core fills it in.
82pub fn map_ws_market_data(
83    msg: &MarketDataMessage,
84    now: Instant,
85    sub: Option<brokapi_core::SubscriptionId>,
86) -> Option<MappedMarketEvent> {
87    let event = match msg {
88        MarketDataMessage::OrderBook(data) => Event::OrderBook(map_ws_order_book(data, now)),
89        MarketDataMessage::Quotes(data) => Event::Quote(map_quotes(data, now)),
90        MarketDataMessage::LastTrades(data) => Event::MarketTrade(map_last_trade(data, now)?),
91        MarketDataMessage::LastCandle(data) => map_last_candle(data, now)?,
92        MarketDataMessage::QuotesSuccess(_)
93        | MarketDataMessage::OrderBookSuccess(_)
94        | MarketDataMessage::LastTradesSuccess(_)
95        | MarketDataMessage::LastCandleSuccess(_)
96        | MarketDataMessage::ErrorResponse(_)
97        | MarketDataMessage::Undefined => return None,
98    };
99    Some(MappedMarketEvent { sub, event })
100}
101
102/// An event mapped from a market data message, with the subscription it belongs to if known.
103#[derive(Debug, Clone, PartialEq)]
104pub struct MappedMarketEvent {
105    /// The subscription the event belongs to, when the caller of the mapper knew it.
106    pub sub: Option<brokapi_core::SubscriptionId>,
107    /// The mapped event.
108    pub event: Event,
109}
110
111/// Maps a WebSocket order book message to an [`OrderBook::Snapshot`]. Every message of this stream
112/// is a full book, not a delta.
113pub fn map_ws_order_book(data: &OrderBookData, now: Instant) -> OrderBook {
114    let instrument = instrument_id_ticker(&data.ticker, &data.class_code);
115    let broker_ts = parse_broker_ts(data.date_time.as_deref());
116    map_order_book_snapshot(
117        &instrument,
118        &data.bids,
119        &data.asks,
120        timestamps(now, broker_ts),
121    )
122}
123
124fn map_order_book_snapshot(
125    instrument: &brokapi_core::InstrumentId,
126    bids: &[impl BookLevelSource],
127    asks: &[impl BookLevelSource],
128    ts: brokapi_core::Timestamps,
129) -> OrderBook {
130    OrderBook::Snapshot {
131        instrument: instrument.clone(),
132        bids: bids
133            .iter()
134            .map(|l| BookLevel {
135                price: Price::new(l.price()),
136                qty: Qty::new(l.quantity()),
137            })
138            .collect(),
139        asks: asks
140            .iter()
141            .map(|l| BookLevel {
142                price: Price::new(l.price()),
143                qty: Qty::new(l.quantity()),
144            })
145            .collect(),
146        ts,
147    }
148}
149
150/// A price level of an order book in either wire form (REST or WebSocket), so one function can map both.
151pub trait BookLevelSource {
152    /// Price of the level.
153    fn price(&self) -> Decimal;
154    /// Quantity at the level, as the broker sends it.
155    fn quantity(&self) -> Decimal;
156}
157
158impl BookLevelSource for brokapi_bcs_protocol::http::models::OrderBookLevel {
159    fn price(&self) -> Decimal {
160        self.price
161    }
162    fn quantity(&self) -> Decimal {
163        self.quantity
164    }
165}
166
167impl BookLevelSource for brokapi_bcs_protocol::ws::models::OrderBookLevel {
168    fn price(&self) -> Decimal {
169        self.price
170    }
171    fn quantity(&self) -> Decimal {
172        self.quantity
173    }
174}
175
176/// Maps a quotes message to a [`Quote`].
177///
178/// The message has one `volume`, and it is put on both sides (bid and ask). A side the message does
179/// not carry is `None`, and a missing volume is `0`. Whether that volume belongs to the bid, the ask
180/// or the last trade is not verified against the live API.
181pub fn map_quotes(data: &QuotesData, now: Instant) -> Quote {
182    let instrument = instrument_id_ticker(&data.ticker, &data.class_code);
183    let broker_ts = parse_broker_ts(data.date_time.as_deref());
184    Quote {
185        instrument,
186        bid: data.bid.map(|p| BookLevel {
187            price: Price::new(p),
188            qty: Qty::new(data.volume.unwrap_or(Decimal::ZERO)),
189        }),
190        ask: data.ask.map(|p| BookLevel {
191            price: Price::new(p),
192            qty: Qty::new(data.volume.unwrap_or(Decimal::ZERO)),
193        }),
194        ts: timestamps(now, broker_ts),
195    }
196}
197
198/// Maps a market trade message to a [`MarketTrade`].
199///
200/// The side is `BUY` / `1` or `SELL` / `2`. A trade whose side is none of these is dropped (`None`)
201/// rather than guessed, like the REST tape ([`map_rest_last_trade`]): a wrong side would corrupt
202/// anything built from the stream.
203pub fn map_last_trade(data: &LastTradesData, now: Instant) -> Option<MarketTrade> {
204    let instrument = instrument_id_ticker(&data.ticker, &data.class_code);
205    let side = match data.side.as_deref() {
206        Some("1") | Some("BUY") => Side::Buy,
207        Some("2") | Some("SELL") => Side::Sell,
208        _ => return None,
209    };
210    let broker_ts = parse_broker_ts(data.date_time.as_deref());
211    Some(MarketTrade {
212        instrument,
213        price: Price::new(data.price),
214        qty: Qty::new(data.quantity),
215        side,
216        trade_id: None,
217        ts: timestamps(now, broker_ts),
218    })
219}
220
221/// One trade of the tape (`POST /last-trades`) for `instrument`, the one asked for: the records of the
222/// docs carry no instrument. Side is `1` buy / `2` sell (older samples: `BUY` / `SELL`); a record whose
223/// side is neither is dropped rather than guessed.
224pub fn map_rest_last_trade(
225    trade: &LastTradeRecord,
226    instrument: &brokapi_core::InstrumentId,
227    now: Instant,
228) -> Option<MarketTrade> {
229    let side = match trade.side.as_deref() {
230        Some("1") | Some("BUY") => Side::Buy,
231        Some("2") | Some("SELL") => Side::Sell,
232        _ => return None,
233    };
234    let broker_ts = parse_broker_ts(trade.date_time.as_deref());
235    Some(MarketTrade {
236        instrument: instrument.clone(),
237        price: Price::new(trade.price),
238        qty: Qty::new(trade.quantity),
239        side,
240        trade_id: None,
241        ts: timestamps(now, broker_ts),
242    })
243}
244
245fn map_last_candle(data: &LastCandleData, now: Instant) -> Option<Event> {
246    let ticker = data.ticker.as_deref()?;
247    let class_code = data.class_code.as_deref()?;
248    let instrument = instrument_id_ticker(ticker, class_code);
249    let timeframe = data
250        .time_frame
251        .as_deref()
252        .and_then(parse_timeframe)
253        .unwrap_or(Timeframe::M1);
254    let broker_ts = parse_broker_ts(data.timestamp.as_deref());
255    let close_ts = broker_ts.unwrap_or_else(|| timestamps(now, None).rx_ts);
256    Some(Event::Candle(brokapi_core::Candle {
257        instrument,
258        timeframe,
259        open: Price::new(data.open?),
260        high: Price::new(data.high?),
261        low: Price::new(data.low?),
262        close: Price::new(data.close?),
263        volume: Qty::new(data.volume.unwrap_or(Decimal::ZERO)),
264        open_ts: close_ts,
265        close_ts,
266        ts: timestamps(now, broker_ts),
267    }))
268}
269
270fn parse_timeframe(raw: &str) -> Option<Timeframe> {
271    match raw {
272        "M1" => Some(Timeframe::M1),
273        "M5" => Some(Timeframe::M5),
274        "M10" => Some(Timeframe::M10),
275        "M15" => Some(Timeframe::M15),
276        "M30" => Some(Timeframe::M30),
277        "H1" => Some(Timeframe::H1),
278        "H4" => Some(Timeframe::H4),
279        // The docs call them `D` and `W`; the `1` forms are accepted too.
280        "D" | "D1" => Some(Timeframe::D1),
281        "W" | "W1" => Some(Timeframe::W1),
282        "MN" => Some(Timeframe::MN),
283        _ => None,
284    }
285}
286
287/// Detect BCS market-data subscribe error payloads (`errors` array).
288///
289/// Returns the message of the first error, or a default text when it has none. `None` when the text is
290/// not JSON or has no non-empty `errors` array.
291///
292/// # Examples
293///
294/// ```
295/// use brokapi_bcs::map::market::parse_subscribe_error;
296///
297/// let text = r#"{"errors":[{"message":"unknown instrument"}]}"#;
298/// assert_eq!(parse_subscribe_error(text).as_deref(), Some("unknown instrument"));
299/// assert_eq!(parse_subscribe_error(r#"{"data":[]}"#), None);
300/// ```
301pub fn parse_subscribe_error(text: &str) -> Option<String> {
302    let v: serde_json::Value = serde_json::from_str(text).ok()?;
303    let errors = v.get("errors")?.as_array()?;
304    let first = errors.first()?;
305    Some(
306        first
307            .get("message")
308            .and_then(|m| m.as_str())
309            .unwrap_or("subscription failed")
310            .to_string(),
311    )
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317
318    fn trade(side: Option<&str>) -> LastTradesData {
319        serde_json::from_value(serde_json::json!({
320            "ticker": "SBER", "classCode": "TQBR", "price": 100.5, "quantity": 3.0, "side": side
321        }))
322        .unwrap()
323    }
324
325    #[test]
326    fn stream_trade_side_is_read_and_never_guessed() {
327        let now = Instant::now();
328        assert_eq!(
329            map_last_trade(&trade(Some("BUY")), now).unwrap().side,
330            Side::Buy
331        );
332        assert_eq!(
333            map_last_trade(&trade(Some("2")), now).unwrap().side,
334            Side::Sell
335        );
336        assert!(map_last_trade(&trade(None), now).is_none());
337        assert!(map_last_trade(&trade(Some("???")), now).is_none());
338    }
339}
brokapi_bcs/map/market.rs

brokapi_bcs/map/
market.rs
