core/rustybot/src/managers.rs

use std::collections::HashMap;
use std::ops::Neg;

use bitfinex::ticker::TradingPairTicker;
use log::error;

use crate::connectors::{Client, ExchangeKind};
use crate::currency::SymbolPair;
use crate::events::{Dispatcher, Event, SignalKind};
use crate::models::{ExecutedOrder, OrderForm, OrderKind, Position, PriceTicker};
use crate::strategy::{FastOrderStrategy, OrderStrategy, PositionStrategy};
use crate::BoxError;
use tokio::sync::mpsc::Receiver;

pub struct EventManager {
    events: Vec<Event>,
}

#[derive(Clone, Debug)]
pub struct PriceManager {
    pair: SymbolPair,
    prices: Vec<PriceEntry>,
    client: Client,
}

impl PriceManager {
    pub fn new(pair: SymbolPair, client: Client) -> Self {
        PriceManager {
            pair,
            prices: Vec::new(),
            client,
        }
    }

    pub fn add_entry(&mut self, entry: PriceEntry) {
        self.prices.push(entry);
    }

    pub async fn update(&mut self, tick: u64) -> Result<PriceEntry, BoxError> {
        let current_prices = self.client.current_prices(&self.pair).await?.into();

        Ok(PriceEntry::new(
            tick,
            current_prices,
            self.pair.clone(),
            None,
            None,
        ))
    }

    // fn add_event(&mut self, event: Event) {
    //     self.events.push(event);
    //
    //     self.dispatcher.call_event_handlers(&event, &self);
    // }
    //
    // fn add_signal(&mut self, signal: SignalKind) {
    //     self.signals.insert(self.current_tick(), signal);
    // }

    pub fn pair(&self) -> &SymbolPair {
        &self.pair
    }
}

#[derive(Clone, Debug)]
pub struct PriceEntry {
    tick: u64,
    pair: SymbolPair,
    price: PriceTicker,
    events: Option<Vec<Event>>,
    signals: Option<Vec<SignalKind>>,
}

impl PriceEntry {
    pub fn new(
        tick: u64,
        price: PriceTicker,
        pair: SymbolPair,
        events: Option<Vec<Event>>,
        signals: Option<Vec<SignalKind>>,
    ) -> Self {
        PriceEntry {
            tick,
            pair,
            price,
            events,
            signals,
        }
    }

    pub fn tick(&self) -> u64 {
        self.tick
    }
    pub fn pair(&self) -> &SymbolPair {
        &self.pair
    }
    pub fn price(&self) -> PriceTicker {
        self.price
    }
    pub fn events(&self) -> &Option<Vec<Event>> {
        &self.events
    }
    pub fn signals(&self) -> &Option<Vec<SignalKind>> {
        &self.signals
    }
}

#[derive(Debug)]
pub struct PositionManager {
    current_tick: u64,
    pair: SymbolPair,
    positions_history: HashMap<u64, Position>,
    active_position: Option<Position>,
    client: Client,
    strategy: Option<Box<dyn PositionStrategy>>,
}

impl PositionManager {
    pub fn new(pair: SymbolPair, client: Client) -> Self {
        PositionManager {
            current_tick: 0,
            pair,
            positions_history: HashMap::new(),
            active_position: None,
            client,
            strategy: None,
        }
    }

    pub fn with_strategy(mut self, strategy: Box<dyn PositionStrategy>) -> Self {
        self.strategy = Some(strategy);
        self
    }

    pub fn current_tick(&self) -> u64 {
        self.current_tick
    }

    pub async fn update(
        &mut self,
        tick: u64,
    ) -> Result<(Option<Vec<Event>>, Option<OrderForm>), BoxError> {
        let opt_active_positions = self.client.active_positions(&self.pair).await?;
        let mut events = vec![];

        self.current_tick = tick;

        // we assume there is only ONE active position per pair
        match opt_active_positions {
            // no open positions, no events and no order forms returned
            None => return Ok((None, None)),

            Some(positions) => {
                // checking if there are positions open for our pair
                match positions
                    .into_iter()
                    .filter(|x| x.pair() == &self.pair)
                    .next()
                {
                    // no open positions for our pair, setting active position to none
                    None => self.active_position = None,

                    // applying strategy to open position and saving into struct
                    Some(position) => {
                        let position_after_strategy = {
                            match &self.strategy {
                                Some(strategy) => {
                                    let (pos, strategy_events, _) =
                                        strategy.on_new_tick(position, &self);

                                    events.extend(strategy_events);

                                    pos
                                }
                                None => position,
                            }
                        };

                        self.positions_history
                            .insert(self.current_tick(), position_after_strategy.clone());
                        self.active_position = Some(position_after_strategy);
                    }
                }
            }
        };

        Ok(((events.is_empty().then_some(events)), None))
    }

    pub fn position_previous_tick(&self, id: u64, tick: Option<u64>) -> Option<&Position> {
        let tick = match tick {
            Some(tick) => {
                if tick < 1 {
                    1
                } else {
                    tick
                }
            }
            None => self.current_tick() - 1,
        };

        self.positions_history
            .get(&tick)
            .filter(|x| x.id() == id)
            .and_then(|x| Some(x))
    }
}

pub type TrackedPositionsMap = HashMap<u64, ExecutedOrder>;

pub struct OrderManager {
    // receiver: Receiver<SignalKind>,
    tracked_positions: TrackedPositionsMap,
    pair: SymbolPair,
    open_orders: Vec<ExecutedOrder>,
    client: Client,
    strategy: Box<dyn OrderStrategy>,
}

impl OrderManager {
    const UNDERCUT_PERC: f64 = 0.005;

    pub fn new(
        // receiver: Receiver<SignalKind>,
        pair: SymbolPair,
        client: Client,
        strategy: Box<dyn OrderStrategy>,
    ) -> Self {
        OrderManager {
            // receiver,
            pair,
            open_orders: Vec::new(),
            client,
            strategy,
            tracked_positions: HashMap::new(),
        }
    }

    pub async fn close_position(&mut self, position: &Position) -> Result<(), BoxError> {
        let open_order = self.tracked_positions.get(&position.id());

        // checking if the position has an open order.
        // If so, the strategy method is called, otherwise we open
        // an undercut limit order at the best current price.
        match open_order {
            Some(open_order) => {
                self.tracked_positions = self
                    .strategy
                    .on_position_close(open_order, &self.tracked_positions);
            }
            None => {
                let current_prices = self.client.current_prices(&self.pair).await?;
                let closing_price = self.best_closing_price(&position, &current_prices)?;

                // submitting order
                let order_form = OrderForm::new(
                    &self.pair,
                    closing_price,
                    position.amount().neg(),
                    OrderKind::Limit,
                );

                match self.client.submit_order(order_form).await {
                    Err(e) => error!("Could not submit order: {}", e),
                    Ok(o) => {
                        self.tracked_positions.insert(position.id(), o);
                    }
                };
            }
        }

        Ok(())
    }

    pub fn update(&self) -> Option<Vec<Event>> {
        unimplemented!()
    }

    pub fn best_closing_price(
        &self,
        position: &Position,
        price_ticker: &TradingPairTicker,
    ) -> Result<f64, BoxError> {
        let price = {
            if position.is_short() {
                price_ticker.ask
            } else {
                price_ticker.bid
            }
        };

        Ok(price * (1.0 - OrderManager::UNDERCUT_PERC))
    }
}

pub struct ExchangeManager {
    kind: ExchangeKind,
    price_managers: Vec<PriceManager>,
    order_managers: Vec<OrderManager>,
    position_managers: Vec<PositionManager>,
    dispatcher: Dispatcher,
    client: Client,
}

impl ExchangeManager {
    pub fn new(kind: &ExchangeKind, pairs: &Vec<SymbolPair>) -> Self {
        let client = Client::new(kind);

        let mut position_managers = Vec::new();
        let mut order_managers = Vec::new();
        let mut price_managers = Vec::new();

        for p in pairs {
            position_managers.push(PositionManager::new(p.clone(), client.clone()));
            order_managers.push(OrderManager::new(
                p.clone(),
                client.clone(),
                Box::new(FastOrderStrategy {}),
            ));
            price_managers.push(PriceManager::new(p.clone(), client.clone()));
        }

        ExchangeManager {
            kind: kind.clone(),
            position_managers,
            order_managers,
            price_managers,
            client,
            dispatcher: Dispatcher::new(),
        }
    }

    pub fn with_position_strategy(mut self, strategy: Box<dyn PositionStrategy>) -> Self {
        self.position_managers = self
            .position_managers
            .into_iter()
            .map(|x| x.with_strategy(dyn_clone::clone_box(&*strategy)))
            .collect();

        self
    }

    pub async fn update_managers(&mut self, tick: u64) -> Result<(), BoxError> {
        self.update_price_managers(tick).await?;
        self.update_position_managers(tick).await?;

        Ok(())
    }

    async fn update_position_managers(
        &mut self,
        tick: u64,
    ) -> Result<Option<Vec<Event>>, BoxError> {
        for mgr in &mut self.position_managers {
            mgr.update(tick).await?;
        }

        Ok(None)
    }

    async fn update_price_managers(&mut self, tick: u64) -> Result<Option<Vec<Event>>, BoxError> {
        let futures: Vec<_> = self
            .price_managers
            .clone()
            .into_iter()
            // the only reason you need the async block is that the future
            // returned by x.update(tick) borrows from x
            // so we create a future that first takes ownership of x, then uses it to call x.update
            .map(|mut x| async move { x.update(tick).await })
            .map(tokio::spawn)
            .collect();

        let mut price_entries = vec![];

        for f in futures {
            price_entries.push(f.await??);
        }

        for manager in &mut self.price_managers {
            let prices: Vec<_> = price_entries
                .drain_filter(|x| x.pair() == manager.pair())
                .collect();

            for p in prices {
                manager.add_entry(p);
            }
        }

        Ok(None)
    }
}