core/rustybot/src/events.rs

use std::collections::{HashMap, HashSet};
use std::future::Future;

use bitfinex::positions::Position;
use tokio::stream::StreamExt;
use tokio::task::JoinHandle;

use crate::pairs::PairStatus;
use crate::positions::{PositionState, PositionWrapper};
use crate::BoxError;

#[derive(Copy, Clone)]
pub enum SignalKind {
    ClosePosition,
    OpenPosition,
}

#[derive(Copy, Clone)]
pub struct EventMetadata {
    position_id: Option<u64>,
    order_id: Option<u64>,
}

#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub enum EventKind {
    NewMinimum,
    NewMaximum,
    ReachedLoss,
    ReachedBreakEven,
    ReachedMinProfit,
    ReachedGoodProfit,
    ReachedMaxLoss,
    TrailingStopSet,
    TrailingStopMoved,
    OrderSubmitted,
    NewTick,
    Any,
}

#[derive(Copy, Clone)]
pub struct Event {
    kind: EventKind,
    tick: u64,
    metadata: Option<EventMetadata>,
}

impl Event {
    pub fn new(kind: EventKind, tick: u64, metadata: Option<EventMetadata>) -> Self {
        Event {
            kind,
            tick,
            metadata,
        }
    }

    fn has_metadata(&self) -> bool {
        self.metadata.is_some()
    }

    pub fn kind(&self) -> EventKind {
        self.kind
    }
    pub fn tick(&self) -> u64 {
        self.tick
    }
    pub fn metadata(&self) -> Option<EventMetadata> {
        self.metadata
    }
}

pub struct EventDispatcher {
    event_handlers: HashMap<EventKind, Vec<Box<dyn Fn(&Event, &PairStatus) -> JoinHandle<()>>>>,
    position_state_handlers:
        HashMap<PositionState, Vec<Box<dyn Fn(&PositionWrapper, &PairStatus) -> JoinHandle<()>>>>,

    on_any_event_handlers: Vec<Box<dyn Fn(&Event, &PairStatus) -> JoinHandle<()>>>,
    on_any_position_state_handlers:
        Vec<Box<dyn Fn(&PositionWrapper, &PairStatus) -> JoinHandle<()>>>,
}

impl EventDispatcher {
    pub fn new() -> Self {
        EventDispatcher {
            event_handlers: HashMap::new(),
            position_state_handlers: HashMap::new(),
            on_any_event_handlers: Vec::new(),
            on_any_position_state_handlers: Vec::new(),
        }
    }

    pub fn call_event_handlers(&self, event: &Event, status: &PairStatus) {
        if let Some(functions) = self.event_handlers.get(&event.kind()) {
            for f in functions {
                f(event, status);
            }
        }

        for f in &self.on_any_event_handlers {
            f(event, status);
        }
    }

    pub fn call_position_state_handlers(&self, pw: &PositionWrapper, status: &PairStatus) {
        if let Some(state) = pw.state() {
            if let Some(functions) = self.position_state_handlers.get(&state) {
                for f in functions {
                    f(pw, status);
                }
            }
        }

        for f in &self.on_any_position_state_handlers {
            f(pw, status);
        }
    }

    pub fn register_event_handler<F: 'static, Fut: 'static>(&mut self, event: EventKind, f: F)
    where
        F: Fn(&Event, &PairStatus) -> Fut,
        Fut: Future<Output = ()> + Send,
    {
        let f = Box::new(move |e, s| tokio::spawn(f(&e, s)));

        match event {
            EventKind::Any => self.on_any_event_handlers.push(f),
            _ => self.event_handlers.entry(event).or_default().push(f),
        }
    }

    pub fn register_positionstate_handler<F: 'static, Fut: 'static>(
        &mut self,
        state: PositionState,
        f: F,
    ) where
        F: Fn(&PositionWrapper, &PairStatus) -> Fut,
        Fut: Future<Output = ()> + Send,
    {
        let f = Box::new(move |pw, s| tokio::spawn(f(&pw, s)));

        match state {
            PositionState::Any => self.on_any_position_state_handlers.push(f),
            _ => self
                .position_state_handlers
                .entry(state)
                .or_default()
                .push(f),
        }
    }
}