cxmr-ws-client-binance 0.0.1

Binance WebSocket feed client.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

use std::iter::FromIterator;

use chrono::offset::Utc;
use hashbrown::HashMap;
use serde_json::Value;

use cxmr_currency::CurrencyPair;
use cxmr_exchanges::Market;
use cxmr_exchanges::{Exchange, Order, OrderSide, Trade};
use cxmr_feeds::{Event, Events, OrderBook, OrderRate};
use cxmr_ws_client::{Command, Error, Parser, Protocol, Subscription};

/// Binance Public feed WebSocket protocol.
pub struct BinancePublicProtocol;

// Implementation of Binance Public feed WebSocket protocol.
impl Protocol<Events> for BinancePublicProtocol {
    /// Protocol parser.
    type Parser = BinancePublicParser;

    /// Creates protocol parser and stream address.
    fn subscription(sub: &Subscription) -> Option<(Command, Self::Parser)> {
        // let parser = BinancePrivateParser
        match sub {
            Subscription::Pairs(pairs) => {
                let address = subscription_address(pairs);
                Some((Command::Connect(address), BinancePublicParser::new(pairs)))
            }
            Subscription::User(_) => None,
        }
    }
}

/// Binance Public feed WebSocket protocol parser.
pub struct BinancePublicParser {
    subscribed: HashMap<String, Market>,
}

impl BinancePublicParser {
    fn new(pairs: &Vec<CurrencyPair>) -> Self {
        let subscribed = HashMap::from_iter(pairs.iter().map(|pair| {
            let id = format!("{}{}", pair.quote(), pair.base()).to_lowercase();
            let market = Market::new(Exchange::Binance, pair.clone());
            (id, market)
        }));
        BinancePublicParser { subscribed }
    }
}

// Implementation of Binance WebSocket protocol parser.
impl Parser<Events> for BinancePublicParser {
    /// Parses received WebSocket message.
    fn parse(&mut self, text: &str) -> Result<Option<Events>, Error> {
        let msg = ::serde_json::from_str::<Value>(text).map_err(|e| Error::Client(e.into()))?;
        let mut split = msg.get("stream")?.as_str()?.split('@');
        let chan_id = split.next()?;
        let channel = split.next()?;
        let data = msg.get("data")?;
        match channel {
            "depth20" => {
                // let _seq = data.get("lastUpdateId")?.as_u64()?;
                let asks = data
                    .get("asks")?
                    .as_array()?
                    .iter()
                    .map(|order| {
                        Ok(OrderRate {
                            rate: order.get(0)?.as_str()?.parse()?,
                            amount: order.get(1)?.as_str()?.parse()?,
                        })
                    })
                    .collect::<Result<Vec<OrderRate>, Error>>()?;
                let bids = data
                    .get("bids")?
                    .as_array()?
                    .iter()
                    .map(|order| {
                        Ok(OrderRate {
                            rate: order.get(0)?.as_str()?.parse()?,
                            amount: order.get(1)?.as_str()?.parse()?,
                        })
                    })
                    .collect::<Result<Vec<OrderRate>, Error>>()?;
                Ok(Some(Events {
                    market: self.subscribed.get(chan_id)?.clone(),
                    events: vec![Event::OrderBook(OrderBook {
                        pair: None,
                        asks: asks,
                        bids: bids,
                    })],
                    timestamp: Some(Utc::now().timestamp_millis() as u64),
                }))
            }
            "trade" => {
                let timestamp = data.get("E")?.as_u64()?;
                let rate = data.get("p")?.as_str()?.parse::<f64>()?;
                let amount = data.get("q")?.as_str()?.parse::<f64>()?;
                let trade_id = data.get("t")?.as_i64()?;
                let side = if data.get("m")?.as_bool()? {
                    OrderSide::Bid
                } else {
                    OrderSide::Ask
                };
                Ok(Some(Events {
                    market: self.subscribed.get(chan_id)?.clone(),
                    events: vec![Event::Trade(Trade {
                        id: Some(trade_id),
                        order: Order {
                            side: side,
                            rate: rate,
                            amount: amount,
                        },
                        timestamp: timestamp,
                    })],
                    timestamp: Some(Utc::now().timestamp_millis() as u64),
                }))
            }
            _ => Ok(None),
        }
    }
}

fn subscription_address(pairs: &Vec<CurrencyPair>) -> String {
    format!(
        "wss://stream.binance.com:9443/stream?streams={}",
        pairs
            .into_iter()
            .map(|pair| {
                let pair = format!("{}{}", pair.quote(), pair.base()).to_lowercase();
                format!("{}@trade/{}@depth20", pair, pair)
            })
            .collect::<Vec<String>>()
            .join("/")
    )
}