core/rustybot/src/connectors.rs

use std::convert::{TryFrom, TryInto};
use std::error::Error;
use std::fmt::{Debug, Formatter};
use std::str::FromStr;
use std::sync::Arc;

use async_trait::async_trait;
use bitfinex::api::Bitfinex;
use bitfinex::orders::{CancelOrderForm, OrderMeta, OrderResponse};
use bitfinex::ticker::TradingPairTicker;
use futures_retry::{FutureRetry, RetryPolicy, StreamRetryExt};
use futures_util::task::FutureObj;
use log::trace;
use tokio::macros::support::Future;
use tokio::time::Duration;

use crate::currency::SymbolPair;
use crate::models::{
    ActiveOrder, OrderBook, OrderBookEntry, OrderForm, OrderKind, Position, PositionState,
    PriceTicker, TradingPlatform,
};
use crate::BoxError;

#[derive(PartialEq, Eq, Clone, Copy, Debug)]
pub enum Exchange {
    Bitfinex,
}

#[derive(Eq, PartialEq, Hash, Clone, Debug)]
pub enum ExchangeDetails {
    Bitfinex { api_key: String, api_secret: String },
}

/// You do **not** have to wrap the `Client` in an [`Rc`] or [`Arc`] to **reuse** it,
/// because it already uses an [`Arc`] internally.
#[derive(Clone, Debug)]
pub struct Client {
    exchange: Exchange,
    inner: Arc<Box<dyn Connector>>,
}

impl Client {
    pub fn new(exchange: &ExchangeDetails) -> Self {
        match exchange {
            ExchangeDetails::Bitfinex {
                api_key,
                api_secret,
            } => Self {
                exchange: Exchange::Bitfinex,
                inner: Arc::new(Box::new((BitfinexConnector::new(api_key, api_secret)))),
            },
        }
    }

    pub async fn active_positions(
        &self,
        pair: &SymbolPair,
    ) -> Result<Option<Vec<Position>>, BoxError> {
        // retrieving open positions and order book to calculate effective profit/loss
        let (positions, order_book) = tokio::join!(
            self.inner.active_positions(pair),
            self.inner.order_book(pair)
        );

        let (mut positions, order_book) = (positions?, order_book?);
        let (best_ask, best_bid) = (order_book.lowest_ask(), order_book.highest_bid());

        // updating positions with effective profit/loss
        // TODO: change fee with account's taker fee
        positions.iter_mut().flatten().for_each(|x| {
            if x.is_short() {
                x.update_profit_loss(best_ask, 0.2);
            } else {
                x.update_profit_loss(best_bid, 0.2);
            }
        });

        Ok(positions)
    }

    pub async fn current_prices(&self, pair: &SymbolPair) -> Result<TradingPairTicker, BoxError> {
        self.inner.current_prices(pair).await
    }

    pub async fn active_orders(&self, pair: &SymbolPair) -> Result<Vec<ActiveOrder>, BoxError> {
        Ok(self
            .inner
            .active_orders(pair)
            .await?
            .into_iter()
            .filter(|x| &x.symbol == pair)
            .collect())
    }

    pub async fn submit_order(&self, order: &OrderForm) -> Result<ActiveOrder, BoxError> {
        self.inner.submit_order(order).await
    }

    pub async fn order_book(&self, pair: &SymbolPair) -> Result<OrderBook, BoxError> {
        self.inner.order_book(pair).await
    }

    pub async fn cancel_order(&self, order: &ActiveOrder) -> Result<ActiveOrder, BoxError> {
        self.inner.cancel_order(order).await
    }
}

#[async_trait]
pub trait Connector: Send + Sync {
    fn name(&self) -> String;
    async fn active_positions(&self, pair: &SymbolPair) -> Result<Option<Vec<Position>>, BoxError>;
    async fn current_prices(&self, pair: &SymbolPair) -> Result<TradingPairTicker, BoxError>;
    async fn order_book(&self, pair: &SymbolPair) -> Result<OrderBook, BoxError>;
    async fn active_orders(&self, pair: &SymbolPair) -> Result<Vec<ActiveOrder>, BoxError>;
    async fn submit_order(&self, order: &OrderForm) -> Result<ActiveOrder, BoxError>;
    async fn cancel_order(&self, order: &ActiveOrder) -> Result<ActiveOrder, BoxError>;
}

impl Debug for dyn Connector {
    fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
        write!(f, "{}", self.name())
    }
}

/**************
* BITFINEX
**************/

pub struct BitfinexConnector {
    bfx: Bitfinex,
}

impl BitfinexConnector {
    const AFFILIATE_CODE: &'static str = "XPebOgHxA";

    fn handle_small_nonce_error(e: BoxError) -> RetryPolicy<BoxError> {
        if e.to_string().contains("nonce: small") {
            return RetryPolicy::WaitRetry(Duration::from_millis(1));
        }
        return RetryPolicy::ForwardError(e);
    }

    pub fn new(api_key: &str, api_secret: &str) -> Self {
        BitfinexConnector {
            bfx: Bitfinex::new(Some(api_key.into()), Some(api_secret.into())),
        }
    }

    fn format_trading_pair(pair: &SymbolPair) -> String {
        if pair.to_string().to_lowercase().contains("test") {
            format!("{}:{}", pair.base(), pair.quote())
        } else {
            format!("{}{}", pair.base(), pair.quote())
        }
    }
}

#[async_trait]
impl Connector for BitfinexConnector {
    fn name(&self) -> String {
        "Bitfinex".into()
    }

    async fn active_positions(&self, pair: &SymbolPair) -> Result<Option<Vec<Position>>, BoxError> {
        let (active_positions, _) = FutureRetry::new(
            move || self.bfx.positions.active_positions(),
            BitfinexConnector::handle_small_nonce_error,
        )
        .await
        .map_err(|(e, attempts)| e)?;

        let positions: Vec<_> = active_positions
            .into_iter()
            .filter_map(|x| x.try_into().ok())
            .filter(|x: &Position| x.pair() == pair)
            .collect();

        trace!("\t[PositionManager] Retrieved positions for {}", pair);
        Ok((!positions.is_empty()).then_some(positions))
    }

    async fn current_prices(&self, pair: &SymbolPair) -> Result<TradingPairTicker, BoxError> {
        let symbol_name = BitfinexConnector::format_trading_pair(pair);

        let ticker: TradingPairTicker = self.bfx.ticker.trading_pair(symbol_name).await?;

        Ok(ticker)
    }

    async fn active_orders(&self, _: &SymbolPair) -> Result<Vec<ActiveOrder>, BoxError> {
        let (response, _) = FutureRetry::new(
            move || self.bfx.orders.active_orders(),
            BitfinexConnector::handle_small_nonce_error,
        )
        .await
        .map_err(|(e, attempts)| e)?;

        Ok(response.iter().map(Into::into).collect())
    }

    async fn submit_order(&self, order: &OrderForm) -> Result<ActiveOrder, BoxError> {
        let symbol_name = format!("t{}", BitfinexConnector::format_trading_pair(order.pair()));

        let order_form = match order.kind() {
            OrderKind::Limit { price, amount } => {
                bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
            }
            OrderKind::Market { amount } => {
                bitfinex::orders::OrderForm::new(symbol_name, 0.0, amount, order.into())
            }
            OrderKind::Stop { price, amount } => {
                bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
            }
            OrderKind::StopLimit {
                price,
                amount,
                limit_price,
            } => bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
                .with_price_aux_limit(limit_price)?,
            OrderKind::TrailingStop { distance, amount } => {
                bitfinex::orders::OrderForm::new(symbol_name, 0.0, amount, order.into())
                    .with_price_trailing(distance)?
            }
            OrderKind::FillOrKill { price, amount } => {
                bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
            }
            OrderKind::ImmediateOrCancel { price, amount } => {
                bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
            }
        }
        .with_meta(OrderMeta::new(
            BitfinexConnector::AFFILIATE_CODE.to_string(),
        ));

        // retry to submit the order until it succeeds.
        // the function may fail due to concurrent signed requests
        // parsed in different times by the server
        let response = {
            loop {
                match self.bfx.orders.submit_order(&order_form).await {
                    Ok(response) => break response,
                    Err(e) => {
                        if !e.to_string().contains("nonce: small") {
                            return Err(e);
                        }
                        tokio::time::sleep(Duration::from_nanos(1)).await;
                    }
                }
            }
        };

        Ok((&response).try_into()?)
    }

    async fn order_book(&self, pair: &SymbolPair) -> Result<OrderBook, BoxError> {
        let symbol_name = BitfinexConnector::format_trading_pair(pair);

        let response = {
            loop {
                match self
                    .bfx
                    .book
                    .trading_pair(symbol_name.clone(), bitfinex::book::BookPrecision::P0)
                    .await
                {
                    Ok(response) => break response,
                    Err(e) => {
                        if !e.to_string().contains("nonce: small") {
                            return Err(e);
                        }
                        tokio::time::sleep(Duration::from_nanos(1)).await;
                    }
                }
            }
        };

        let entries = response
            .into_iter()
            .map(|x| OrderBookEntry::Trading {
                price: x.price,
                count: x.count as u64,
                amount: x.amount,
            })
            .collect();

        Ok(OrderBook::new(pair.clone()).with_entries(entries))
    }

    async fn cancel_order(&self, order: &ActiveOrder) -> Result<ActiveOrder, BoxError> {
        let cancel_form = order.into();

        let response = {
            loop {
                match self.bfx.orders.cancel_order(&cancel_form).await {
                    Ok(response) => break response,
                    Err(e) => {
                        if !e.to_string().contains("nonce: small") {
                            return Err(e);
                        }
                        tokio::time::sleep(Duration::from_nanos(1)).await;
                    }
                }
            }
        };

        Ok((&response).try_into()?)
    }
}

impl From<&ActiveOrder> for CancelOrderForm {
    fn from(o: &ActiveOrder) -> Self {
        Self::from_id(o.id)
    }
}

impl TryFrom<&bitfinex::orders::OrderResponse> for ActiveOrder {
    type Error = BoxError;

    fn try_from(response: &OrderResponse) -> Result<Self, Self::Error> {
        Ok(Self {
            exchange: Exchange::Bitfinex,
            id: response.id(),
            group_id: response.gid(),
            client_id: Some(response.cid()),
            symbol: SymbolPair::from_str(response.symbol())?,
            current_form: OrderForm::new(
                SymbolPair::from_str(response.symbol())?,
                response.into(),
                response.into(),
            ),
            creation_timestamp: 0,
            update_timestamp: 0,
        })
    }
}

impl TryInto<Position> for bitfinex::positions::Position {
    type Error = BoxError;

    fn try_into(self) -> Result<Position, Self::Error> {
        let state = {
            if self.status().to_lowercase().contains("active") {
                PositionState::Open
            } else {
                PositionState::Closed
            }
        };

        Ok(Position::new(
            SymbolPair::from_str(self.symbol())?,
            state,
            self.amount(),
            self.base_price(),
            self.pl(),
            self.pl_perc(),
            self.price_liq(),
            self.position_id(),
        )
        .with_creation_date(self.mts_create())
        .with_creation_update(self.mts_update()))
    }
}

impl From<&OrderForm> for bitfinex::orders::OrderKind {
    fn from(o: &OrderForm) -> Self {
        match o.platform() {
            TradingPlatform::Exchange => match o.kind() {
                OrderKind::Limit { .. } => bitfinex::orders::OrderKind::ExchangeLimit,
                OrderKind::Market { .. } => bitfinex::orders::OrderKind::ExchangeMarket,
                OrderKind::Stop { .. } => bitfinex::orders::OrderKind::ExchangeStop,
                OrderKind::StopLimit { .. } => bitfinex::orders::OrderKind::ExchangeStopLimit,
                OrderKind::TrailingStop { .. } => bitfinex::orders::OrderKind::ExchangeTrailingStop,
                OrderKind::FillOrKill { .. } => bitfinex::orders::OrderKind::ExchangeFok,
                OrderKind::ImmediateOrCancel { .. } => bitfinex::orders::OrderKind::ExchangeIoc,
            },
            TradingPlatform::Margin => match o.kind() {
                OrderKind::Limit { .. } => bitfinex::orders::OrderKind::Limit,
                OrderKind::Market { .. } => bitfinex::orders::OrderKind::Market,
                OrderKind::Stop { .. } => bitfinex::orders::OrderKind::Stop,
                OrderKind::StopLimit { .. } => bitfinex::orders::OrderKind::StopLimit,
                OrderKind::TrailingStop { .. } => bitfinex::orders::OrderKind::TrailingStop,
                OrderKind::FillOrKill { .. } => bitfinex::orders::OrderKind::Fok,
                OrderKind::ImmediateOrCancel { .. } => bitfinex::orders::OrderKind::Ioc,
            },
            _ => unimplemented!(),
        }
    }
}

impl From<&bitfinex::orders::OrderResponse> for TradingPlatform {
    fn from(response: &OrderResponse) -> Self {
        match response.order_type() {
            bitfinex::orders::OrderKind::Limit
            | bitfinex::orders::OrderKind::Market
            | bitfinex::orders::OrderKind::StopLimit
            | bitfinex::orders::OrderKind::Stop
            | bitfinex::orders::OrderKind::TrailingStop
            | bitfinex::orders::OrderKind::Fok
            | bitfinex::orders::OrderKind::Ioc => Self::Margin,
            _ => Self::Exchange,
        }
    }
}

impl From<&bitfinex::orders::ActiveOrder> for TradingPlatform {
    fn from(response: &bitfinex::orders::ActiveOrder) -> Self {
        match response.order_type() {
            bitfinex::orders::OrderKind::Limit
            | bitfinex::orders::OrderKind::Market
            | bitfinex::orders::OrderKind::StopLimit
            | bitfinex::orders::OrderKind::Stop
            | bitfinex::orders::OrderKind::TrailingStop
            | bitfinex::orders::OrderKind::Fok
            | bitfinex::orders::OrderKind::Ioc => Self::Margin,
            _ => Self::Exchange,
        }
    }
}

impl From<&bitfinex::orders::OrderResponse> for OrderKind {
    fn from(response: &OrderResponse) -> Self {
        match response.order_type() {
            bitfinex::orders::OrderKind::Limit | bitfinex::orders::OrderKind::ExchangeLimit => {
                Self::Limit {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::Market | bitfinex::orders::OrderKind::ExchangeMarket => {
                Self::Market {
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::Stop | bitfinex::orders::OrderKind::ExchangeStop => {
                Self::Stop {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::StopLimit
            | bitfinex::orders::OrderKind::ExchangeStopLimit => Self::StopLimit {
                price: response.price(),
                amount: response.amount(),
                limit_price: response.price_aux_limit().expect("Limit price not found!"),
            },
            bitfinex::orders::OrderKind::TrailingStop
            | bitfinex::orders::OrderKind::ExchangeTrailingStop => Self::TrailingStop {
                distance: response.price_trailing().expect("Distance not found!"),
                amount: response.amount(),
            },
            bitfinex::orders::OrderKind::Fok | bitfinex::orders::OrderKind::ExchangeFok => {
                Self::FillOrKill {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::Ioc | bitfinex::orders::OrderKind::ExchangeIoc => {
                Self::ImmediateOrCancel {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
        }
    }
}

impl From<&bitfinex::orders::ActiveOrder> for OrderKind {
    fn from(response: &bitfinex::orders::ActiveOrder) -> Self {
        match response.order_type() {
            bitfinex::orders::OrderKind::Limit | bitfinex::orders::OrderKind::ExchangeLimit => {
                Self::Limit {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::Market | bitfinex::orders::OrderKind::ExchangeMarket => {
                Self::Market {
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::Stop | bitfinex::orders::OrderKind::ExchangeStop => {
                Self::Stop {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::StopLimit
            | bitfinex::orders::OrderKind::ExchangeStopLimit => Self::StopLimit {
                price: response.price(),
                amount: response.amount(),
                limit_price: response.price_aux_limit().expect("Limit price not found!"),
            },
            bitfinex::orders::OrderKind::TrailingStop
            | bitfinex::orders::OrderKind::ExchangeTrailingStop => Self::TrailingStop {
                distance: response.price_trailing().expect("Distance not found!"),
                amount: response.amount(),
            },
            bitfinex::orders::OrderKind::Fok | bitfinex::orders::OrderKind::ExchangeFok => {
                Self::FillOrKill {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
            bitfinex::orders::OrderKind::Ioc | bitfinex::orders::OrderKind::ExchangeIoc => {
                Self::ImmediateOrCancel {
                    price: response.price(),
                    amount: response.amount(),
                }
            }
        }
    }
}

impl From<&bitfinex::orders::ActiveOrder> for ActiveOrder {
    fn from(order: &bitfinex::orders::ActiveOrder) -> Self {
        let pair = SymbolPair::from_str(&order.symbol()).expect("Invalid symbol!");

        Self {
            exchange: Exchange::Bitfinex,
            id: order.id(),
            group_id: order.group_id().map(|x| x as u64),
            client_id: Some(order.client_id()),
            symbol: pair.clone(),
            current_form: OrderForm::new(pair, order.into(), order.into()),
            creation_timestamp: order.creation_timestamp(),
            update_timestamp: order.update_timestamp(),
        }
    }
}

impl From<TradingPairTicker> for PriceTicker {
    fn from(t: TradingPairTicker) -> Self {
        Self {
            bid: t.bid,
            bid_size: t.bid_size,
            ask: t.ask,
            ask_size: t.ask_size,
            daily_change: t.daily_change,
            daily_change_perc: t.daily_change_perc,
            last_price: t.last_price,
            volume: t.volume,
            high: t.high,
            low: t.low,
        }
    }
}