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
136
137
138
139

use crate::{
    model::SubKind, ExchangeId, ExchangeTransformer, Identifiable, MarketEvent, Subscriber,
    Subscription, SubscriptionIds, SubscriptionMeta,
};
use barter_integration::{
    error::SocketError,
    model::{InstrumentKind, SubscriptionId},
    protocol::websocket::WsMessage,
    Transformer,
};
use model::{FtxMessage, FtxSubResponse};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::collections::HashMap;

/// [`Ftx`] specific data structures.
mod model;

/// `Ftx` [`Subscriber`] & [`ExchangeTransformer`] implementor for the collection
/// of `Spot` & `Futures` data.
#[derive(Clone, Eq, PartialEq, Debug, Deserialize, Serialize)]
pub struct Ftx {
    pub ids: SubscriptionIds,
}

impl Subscriber for Ftx {
    type SubResponse = FtxSubResponse;

    fn base_url() -> &'static str {
        "wss://ftx.com/ws/"
    }

    fn build_subscription_meta(
        subscriptions: &[Subscription],
    ) -> Result<SubscriptionMeta, SocketError> {
        // Allocate SubscriptionIds HashMap to track identifiers for each actioned Subscription
        let mut ids = SubscriptionIds(HashMap::with_capacity(subscriptions.len()));

        // Map Barter Subscriptions to Ftx channels
        let subscriptions = subscriptions
            .iter()
            .map(|subscription| {
                // Determine the Ftx specific channel & market for this Barter Subscription
                let (channel, market) = Self::get_channel_meta(subscription)?;

                // Construct Ftx specific subscription message
                let ftx_subscription = Self::subscription(channel, &market);

                // Use market as the SubscriptionId key in the SubscriptionIds
                ids.0.insert(SubscriptionId(market), subscription.clone());

                Ok(ftx_subscription)
            })
            .collect::<Result<Vec<_>, SocketError>>()?;

        Ok(SubscriptionMeta {
            ids,
            expected_responses: subscriptions.len(),
            subscriptions,
        })
    }
}

impl ExchangeTransformer for Ftx {
    const EXCHANGE: ExchangeId = ExchangeId::Ftx;
    fn new(ids: SubscriptionIds) -> Self {
        Self { ids }
    }
}

impl Transformer<MarketEvent> for Ftx {
    type Input = FtxMessage;
    type OutputIter = Vec<Result<MarketEvent, SocketError>>;

    fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
        let instrument = match self.ids.find_instrument(input.id()) {
            Ok(instrument) => instrument,
            Err(error) => return vec![Err(error)],
        };

        match input {
            FtxMessage::Trades { trades, .. } => trades
                .into_iter()
                .map(|trade| {
                    Ok(MarketEvent::from((
                        Ftx::EXCHANGE,
                        instrument.clone(),
                        trade,
                    )))
                })
                .collect(),
        }
    }
}

impl Ftx {
    /// Determine the `Ftx` channel metadata associated with an input Barter [`Subscription`].
    /// This includes the `Ftx` &str channel, and a `String` market identifier. Both are used to
    /// build an `Ftx` subscription payload.
    ///
    /// Example Ok Return: Ok("trades", "BTC/USDT")
    /// where channel == "trades" & market == "BTC/USDT".
    fn get_channel_meta(sub: &Subscription) -> Result<(&str, String), SocketError> {
        // Determine Ftx channel using the Subscription StreamKind
        let channel = match &sub.kind {
            SubKind::Trade => "trades",
            other => {
                return Err(SocketError::Unsupported {
                    entity: Self::EXCHANGE.as_str(),
                    item: other.to_string(),
                })
            }
        };

        // Determine Ftx market using the InstrumentKind
        let market = match &sub.instrument.kind {
            InstrumentKind::Spot => {
                format!("{}/{}", sub.instrument.base, sub.instrument.quote).to_uppercase()
            }
            InstrumentKind::FuturePerpetual => {
                format!("{}-PERP", sub.instrument.base).to_uppercase()
            }
        };

        Ok((channel, market))
    }

    /// Build a `Ftx` compatible subscription message using the channel & market provided.
    fn subscription(channel: &str, market: &str) -> WsMessage {
        WsMessage::Text(
            json!({
                "op": "subscribe",
                "channel": channel,
                "market": market,
            })
            .to_string(),
        )
    }
}