core/src/connectors.rs

743 lines
25 KiB
Rust

use std::convert::{TryFrom, TryInto};
use std::fmt::{Debug, Formatter};
use std::str::FromStr;
use std::sync::Arc;
use async_trait::async_trait;
use bitfinex::api::Bitfinex;
use bitfinex::book::BookPrecision;
use bitfinex::orders::{CancelOrderForm, OrderMeta};
use bitfinex::responses::{OrderResponse, TradeResponse};
use bitfinex::ticker::TradingPairTicker;
use futures_retry::RetryPolicy;
use log::trace;
use tokio::macros::support::Future;
use tokio::time::Duration;
use crate::BoxError;
use crate::currency::{Symbol, SymbolPair};
use crate::models::{
ActiveOrder, OrderBook, OrderBookEntry, OrderDetails, OrderFee, OrderForm, OrderKind, Position,
PositionState, PriceTicker, Trade, TradingFees, TradingPlatform, WalletKind,
};
#[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, fees) = tokio::join!(
self.inner.active_positions(pair),
self.inner.order_book(pair),
self.inner.trading_fees()
);
let (mut positions, order_book, fees) = (positions?, order_book?, fees?);
let (best_ask, best_bid) = (order_book.lowest_ask(), order_book.highest_bid());
if positions.is_none() {
return Ok(None);
}
let derivative_taker = fees
.iter()
.filter_map(|x| match x {
TradingFees::Taker {
platform,
percentage,
} if platform == &TradingPlatform::Derivative => Some(percentage),
_ => None,
})
.next()
.ok_or("Could not retrieve derivative taker fee!")?;
let margin_taker = fees
.iter()
.filter_map(|x| match x {
TradingFees::Taker {
platform,
percentage,
} if platform == &TradingPlatform::Margin => Some(percentage),
_ => None,
})
.next()
.ok_or("Could not retrieve margin taker fee!")?;
// updating positions with effective profit/loss
positions.iter_mut().flatten().for_each(|x| {
let fee = match x.platform() {
TradingPlatform::Funding | TradingPlatform::Exchange => {
unimplemented!()
}
TradingPlatform::Margin => margin_taker,
TradingPlatform::Derivative => derivative_taker,
};
if x.is_short() {
x.update_profit_loss(best_ask, *fee);
} else {
x.update_profit_loss(best_bid, *fee);
}
});
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.pair() == &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
}
pub async fn transfer_between_wallets(
&self,
from: &WalletKind,
to: &WalletKind,
symbol: Symbol,
amount: f64,
) -> Result<(), BoxError> {
self.inner
.transfer_between_wallets(from, to, symbol, amount)
.await
}
pub async fn trades_from_order(
&self,
order: &OrderDetails,
) -> Result<Option<Vec<Trade>>, BoxError> {
self.inner.trades_from_order(order).await
}
pub async fn orders_history(
&self,
pair: &SymbolPair,
) -> Result<Option<Vec<OrderDetails>>, BoxError> {
self.inner.orders_history(pair).await
}
pub async fn trading_fees(&self) -> Result<Vec<TradingFees>, BoxError> { self.inner.trading_fees().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>;
async fn transfer_between_wallets(
&self,
from: &WalletKind,
to: &WalletKind,
symbol: Symbol,
amount: f64,
) -> Result<(), BoxError>;
async fn trades_from_order(&self, order: &OrderDetails)
-> Result<Option<Vec<Trade>>, BoxError>;
async fn orders_history(
&self,
pair: &SymbolPair,
) -> Result<Option<Vec<OrderDetails>>, BoxError>;
async fn trading_fees(&self) -> Result<Vec<TradingFees>, 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));
}
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")
|| pair.to_string().to_lowercase().contains("f0")
{
format!("{}:{}", pair.base(), pair.quote())
} else {
format!("{}{}", pair.base(), pair.quote())
}
}
// retry to submit the request until it succeeds.
// the function may fail due to concurrent signed requests
// parsed in different times by the server
async fn retry_nonce<F, Fut, O>(mut func: F) -> Result<O, BoxError>
where
F: FnMut() -> Fut,
Fut: Future<Output=Result<O, BoxError>>,
{
let response = {
loop {
match func().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)
}
}
#[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 =
BitfinexConnector::retry_nonce(|| self.bfx.positions.active_positions()).await?;
let positions: Vec<_> = active_positions
.into_iter()
.filter_map(|x| x.try_into().ok())
.filter(|x: &Position| x.pair() == pair)
.collect();
trace!("\tRetrieved 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 order_book(&self, pair: &SymbolPair) -> Result<OrderBook, BoxError> {
let symbol_name = BitfinexConnector::format_trading_pair(pair);
let response = BitfinexConnector::retry_nonce(|| {
self.bfx.book.trading_pair(&symbol_name, BookPrecision::P0)
})
.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 active_orders(&self, _: &SymbolPair) -> Result<Vec<ActiveOrder>, BoxError> {
let response = BitfinexConnector::retry_nonce(|| self.bfx.orders.active_orders()).await?;
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 amount = order.amount();
let order_form = {
match order.kind() {
OrderKind::Limit { price } => {
bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
}
OrderKind::Market => {
bitfinex::orders::OrderForm::new(symbol_name, 0.0, amount, order.into())
}
OrderKind::Stop { price } => {
bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
}
OrderKind::StopLimit { stop_price: price, limit_price } => {
bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
.with_price_aux_limit(Some(limit_price))?
}
OrderKind::TrailingStop { distance } => {
bitfinex::orders::OrderForm::new(symbol_name, 0.0, amount, order.into())
.with_price_trailing(Some(distance))?
}
OrderKind::FillOrKill { price } => {
bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
}
OrderKind::ImmediateOrCancel { price } => {
bitfinex::orders::OrderForm::new(symbol_name, price, amount, order.into())
}
}
.with_meta(Some(OrderMeta::new(
BitfinexConnector::AFFILIATE_CODE.to_string(),
)))
// TODO: CHANGEME!
.with_leverage(Some(15))
};
let response =
BitfinexConnector::retry_nonce(|| self.bfx.orders.submit_order(&order_form)).await?;
// parsing response into ActiveOrder and adding leverage from order form
let order_response: ActiveOrder = (&response).try_into()?;
// TODO: CHANGEME!!!!
Ok(order_response.with_leverage(Some(15.0)))
}
async fn cancel_order(&self, order: &ActiveOrder) -> Result<ActiveOrder, BoxError> {
let cancel_form = order.into();
let response =
BitfinexConnector::retry_nonce(|| self.bfx.orders.cancel_order(&cancel_form)).await?;
Ok((&response).try_into()?)
}
async fn transfer_between_wallets(
&self,
from: &WalletKind,
to: &WalletKind,
symbol: Symbol,
amount: f64,
) -> Result<(), BoxError> {
BitfinexConnector::retry_nonce(|| {
self.bfx.account.transfer_between_wallets(
from.into(),
to.into(),
symbol.to_string(),
amount,
)
})
.await?;
Ok(())
}
async fn trades_from_order(
&self,
order: &OrderDetails,
) -> Result<Option<Vec<Trade>>, BoxError> {
let response = BitfinexConnector::retry_nonce(|| {
self.bfx
.trades
.generated_by_order(order.pair().trading_repr(), order.id())
})
.await?;
if response.is_empty() {
Ok(None)
} else {
Ok(Some(response.iter().map(Into::into).collect()))
}
}
async fn orders_history(
&self,
pair: &SymbolPair,
) -> Result<Option<Vec<OrderDetails>>, BoxError> {
let response =
BitfinexConnector::retry_nonce(|| self.bfx.orders.history(Some(pair.trading_repr())))
.await?;
let mapped_vec: Vec<_> = response.iter().map(Into::into).collect();
Ok((!mapped_vec.is_empty()).then_some(mapped_vec))
}
async fn trading_fees(&self) -> Result<Vec<TradingFees>, BoxError> {
let mut fees = vec![];
let accountfees =
BitfinexConnector::retry_nonce(|| self.bfx.account.account_summary()).await?;
// Derivatives
let derivative_taker = TradingFees::Taker {
platform: TradingPlatform::Derivative,
percentage: accountfees.derivative_taker() * 100.0,
};
let derivative_maker = TradingFees::Maker {
platform: TradingPlatform::Derivative,
percentage: accountfees.derivative_rebate() * 100.0,
};
fees.push(derivative_taker);
fees.push(derivative_maker);
// Exchange
let exchange_taker = TradingFees::Taker {
platform: TradingPlatform::Exchange,
percentage: accountfees.taker_to_fiat() * 100.0,
};
let exchange_maker = TradingFees::Maker {
platform: TradingPlatform::Exchange,
percentage: accountfees.maker_fee() * 100.0,
};
fees.push(exchange_taker);
fees.push(exchange_maker);
// Margin
let margin_taker = TradingFees::Taker {
platform: TradingPlatform::Margin,
percentage: accountfees.taker_to_fiat() * 100.0,
};
let margin_maker = TradingFees::Maker {
platform: TradingPlatform::Margin,
percentage: accountfees.maker_fee() * 100.0,
};
fees.push(margin_taker);
fees.push(margin_maker);
Ok(fees)
}
}
impl From<&ActiveOrder> for CancelOrderForm {
fn from(o: &ActiveOrder) -> Self {
Self::from_id(o.id())
}
}
impl TryFrom<&bitfinex::responses::OrderResponse> for ActiveOrder {
type Error = BoxError;
fn try_from(response: &OrderResponse) -> Result<Self, Self::Error> {
let pair = SymbolPair::from_str(response.symbol())?;
Ok(ActiveOrder::new(
Exchange::Bitfinex,
response.id(),
pair.clone(),
OrderForm::new(pair, response.into(), response.into(), response.amount()),
response.mts_create(),
response.mts_update(),
)
.with_group_id(response.gid())
.with_client_id(Some(response.cid())))
}
}
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
}
};
let platform = {
if self.symbol().to_ascii_lowercase().contains("f0") {
TradingPlatform::Derivative
} else {
TradingPlatform::Margin
}
};
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(),
platform,
self.leverage(),
)
.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 | TradingPlatform::Derivative => 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::responses::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::responses::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(),
}
}
bitfinex::orders::OrderKind::Market | bitfinex::orders::OrderKind::ExchangeMarket => {
Self::Market
}
bitfinex::orders::OrderKind::Stop | bitfinex::orders::OrderKind::ExchangeStop => {
Self::Stop {
price: response.price(),
}
}
bitfinex::orders::OrderKind::StopLimit
| bitfinex::orders::OrderKind::ExchangeStopLimit => Self::StopLimit {
stop_price: response.price(),
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!"),
},
bitfinex::orders::OrderKind::Fok | bitfinex::orders::OrderKind::ExchangeFok => {
Self::FillOrKill {
price: response.price(),
}
}
bitfinex::orders::OrderKind::Ioc | bitfinex::orders::OrderKind::ExchangeIoc => {
Self::ImmediateOrCancel {
price: response.price(),
}
}
}
}
}
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(),
}
}
bitfinex::orders::OrderKind::Market | bitfinex::orders::OrderKind::ExchangeMarket => {
Self::Market {}
}
bitfinex::orders::OrderKind::Stop | bitfinex::orders::OrderKind::ExchangeStop => {
Self::Stop {
price: response.price(),
}
}
bitfinex::orders::OrderKind::StopLimit
| bitfinex::orders::OrderKind::ExchangeStopLimit => Self::StopLimit {
stop_price: response.price(),
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!"),
},
bitfinex::orders::OrderKind::Fok | bitfinex::orders::OrderKind::ExchangeFok => {
Self::FillOrKill {
price: response.price(),
}
}
bitfinex::orders::OrderKind::Ioc | bitfinex::orders::OrderKind::ExchangeIoc => {
Self::ImmediateOrCancel {
price: response.price(),
}
}
}
}
}
impl From<&bitfinex::orders::ActiveOrder> for ActiveOrder {
fn from(order: &bitfinex::orders::ActiveOrder) -> Self {
let pair = SymbolPair::from_str(&order.symbol()).expect("Invalid symbol!");
ActiveOrder::new(
Exchange::Bitfinex,
order.id(),
pair.clone(),
OrderForm::new(pair, order.into(), order.into(), order.amount()),
order.creation_timestamp(),
order.update_timestamp(),
)
.with_client_id(Some(order.client_id()))
.with_group_id(order.group_id())
}
}
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,
}
}
}
impl From<&WalletKind> for &bitfinex::account::WalletKind {
fn from(k: &WalletKind) -> Self {
match k {
WalletKind::Exchange => &bitfinex::account::WalletKind::Exchange,
WalletKind::Margin => &bitfinex::account::WalletKind::Margin,
WalletKind::Funding => &bitfinex::account::WalletKind::Funding,
}
}
}
impl From<&bitfinex::orders::ActiveOrder> for OrderDetails {
fn from(order: &bitfinex::orders::ActiveOrder) -> Self {
Self::new(
Exchange::Bitfinex,
order.id(),
SymbolPair::from_str(order.symbol()).unwrap(),
order.into(),
order.into(),
order.update_timestamp(),
)
}
}
// TODO: fields are hardcoded, to fix
impl From<&bitfinex::responses::TradeResponse> for Trade {
fn from(response: &TradeResponse) -> Self {
let pair = SymbolPair::from_str(&response.symbol()).unwrap();
let fee = {
if response.is_maker() {
OrderFee::Maker(response.fee())
} else {
OrderFee::Taker(response.fee())
}
};
Self {
trade_id: response.trade_id(),
pair,
execution_timestamp: response.execution_timestamp(),
price: response.execution_price(),
amount: response.execution_amount(),
fee,
fee_currency: Symbol::new(response.symbol().to_owned()),
}
}
}