core/rustybot/src/managers.rs

use std::collections::HashMap;

use crate::connectors::{Client, ExchangeKind};
use crate::currency::SymbolPair;
use crate::events::{Event, SignalKind};
use crate::models::{Order, Position, PriceTicker};
use crate::strategy::PositionStrategy;
use crate::BoxError;

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,
        ))
    }

    // pub fn add_position(&mut self, position: Position) {
    //     let (new_position, events, signals) = {
    //         match &self.strategy {
    //             Some(strategy) => strategy.on_new_tick(&position, &self),
    //             None => (position, vec![], vec![]),
    //         }
    //     };
    //
    //     self.positions
    //         .entry(self.current_tick)
    //         .or_default()
    //         .push(new_position.clone());
    //
    //     // calling position state callbacks
    //     self.dispatcher
    //         .call_position_state_handlers(&new_position, &self);
    //
    //     // adding events and calling callbacks
    //     for e in events {
    //         self.add_event(e);
    //     }
    //
    //     // adding signals to current tick vector
    //     for s in signals {
    //         self.add_signal(s);
    //     }
    // }

    // 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>>, BoxError> {
        let opt_active_positions = self.client.active_positions(&self.pair).await?;
        let mut events = vec![];

        self.current_tick = tick;

        if opt_active_positions.is_none() {
            return Ok(None);
        }

        // we assume there is only ONE active position per pair
        match opt_active_positions
            .unwrap()
            .into_iter()
            .filter(|x| x.pair() == &self.pair)
            .next()
        {
            Some(position) => {
                // applying strategy to position
                let active_position = {
                    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(), active_position.clone());
                self.active_position = Some(active_position);
            }
            None => {
                self.active_position = None;
            }
        }

        if events.is_empty() {
            Ok(None)
        } else {
            Ok(Some(events))
        }
    }

    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.position_id() == id)
            .and_then(|x| Some(x))
    }
}

pub struct OrderManager {
    pair: SymbolPair,
    open_orders: Vec<Order>,
    client: Client,
}

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

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

pub struct ExchangeManager {
    kind: ExchangeKind,
    price_managers: Vec<PriceManager>,
    order_managers: Vec<OrderManager>,
    position_managers: Vec<PositionManager>,
    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()));
            price_managers.push(PriceManager::new(p.clone(), client.clone()));
        }

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

    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 {
            println!("Manager: {:?}", mgr);
            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)
    }
}