feat: okx support SubscribeBidAsk and SubscribeTrades

[Nouzan]

It is not a good idea to provide the support of SubscribeBidAsk and SubscribeTrades requests by just underlying using the SubscribeTickers request. Instead, we should use the API directly provided by OKX, since the tickers channel has a limited pushing rate (at most 1 update/100ms) but the trades channel doesn't.

[rise0chen]

How about impl SubscribeTrades by the trades channel, and impl SubscribeBidAsk by the tickers channel ?

[Nouzan]

I am not sure what would happen if the user subscribes to the tickers channel of the same instrument twice. If your goal is just to get a Service<SubscribeTrades> or a Service<SubscribeBidAsk>, maybe you can define wrappers for any Service<SubscribeTickers>. Just like how I implemented Service<SubscribeTickers> for Binance.

[rise0chen]

I think we should impl all service for every exchange.

I want to get price every minute. The price of QueryCandles is last minute, but I want to get realtime price. SubscribeTickers not impl for original Binance. So, I want to use SubscribeTrades which can impl for every exchange.

[rise0chen]

I have make a test, subscribe many times is same as subscribe once time.

The tickers channel will send a old msg immediately after you subscribe it. But, The trades channel won't send a old msg immediately, it maybe wait many time to get first msg if trade is few.

[rise0chen]

pub struct Exchange<C, Req, KC> {
    public_exc: Exc<C, Req>,
    private_exc: Exc<C, Req>,
    kline_exc: Exc<KC, QueryCandles>,
    pub inst: String,
}
impl<C, Req, KC> Exchange<C, Req, KC> {
    pub fn new(public_exc: Exc<C, Req>, private_exc: Exc<C, Req>, kline_exc: Exc<KC, QueryCandles>, inst: String) -> Self {
        Self {
            public_exc,
            private_exc,
            kline_exc,
            inst,
        }
    }
}
impl<C, Req, KC> Exchange<C, Req, KC>
where
    Exc<KC, QueryCandles>: FetchCandlesService,
{
    pub async fn get_history<R: RangeBounds<OffsetDateTime>>(&mut self, period: Period, range: R) -> anyhow::Result<CandleStream> {
        let sub = self.kline_exc.fetch_candles(&self.inst, period, range).await?;
        Ok(sub)
    }
}
impl<C, Req, KC> Exchange<C, Req, KC>
where
    Exc<C, Req>: SubscribeTradesService,
{
    pub async fn get_price(&mut self) -> anyhow::Result<Trade> {
        let mut sub = self.public_exc.subscribe_trades(&self.inst).await?;
        match sub.next().await {
            Some(Ok(data)) => Ok(data),
            e => Err(anyhow::anyhow!("get_price: {:?}", e)),
        }
    }
}

pub struct GridStrategy<C, Req, KC> {
    exchange: Exchange<C, Req, KC>,
}

impl<C, Req, KC> GridStrategy<C, Req, KC>
where
    Exc<C, Req>: ReconnectService,
    Exc<C, Req>: SubscribeOrdersService,
    Exc<C, Req>: CheckOrderService,
    Exc<C, Req>: TradingService,
    Exc<C, Req>: SubscribeTradesService,
    Exc<KC, QueryCandles>: FetchCandlesService,
{
}

I have to add kline_exc: Exc<KC, QueryCandles> if I use wrappers, It is unfriendly. I like use original exchange rather than wrappers.

[Nouzan]

Yes, you are right. It is a bit annoying. In fact, what I am going to solve in exc v0.6 is the same. I want to make a single Market type with all the required methods for every supported exchanges so that the users don't have to work with all kinds of wrappers. But I am still working on the design.

[Nouzan]

I think we should impl all service for every exchange.

I want to get price every minute. The price of QueryCandles is last minute, but I want to get realtime price. SubscribeTickers not impl for original Binance. So, I want to use SubscribeTrades which can impl for every exchange.

You can pick up every final services you need, and construct these services separately for different exchanges. For example,

struct Exchange {
    ticker: Box<dyn SubscribeTickersService>,
    trading: Box<dyn TradingService>,
}

impl Exchange {
    fn okx() -> Self {
        ...
    }

    fn binance() -> Self {
        ...
    }
}

Because these exchanges are different, they may be missing certain functionalities, so we have to handle these differences somewhere. I am going to handle them for users and release with exc v0.6, but there are many works to do.

[Nouzan]

But we have to make sure the underlying implementation is direct and efficient. So if you still want to add support for the trade and bid/ask subscription APIs of OKX, I may still insist on providing direct integration instead of relying on SubscribeTickers for implementation.

[rise0chen]

It is a good idea.

What's more, disconnect event should be solved. Maybe, every service need ReconnectService.

struct Exchange {
    ticker: Box<dyn SubscribeTickersService + ReconnectService>,
    trading: Box<dyn TradingService + ReconnectService>,
}

[Nouzan]

It is a good idea.

What's more, disconnect event should be solved. Maybe, every service need ReconnectService.

struct Exchange {
    ticker: Box<dyn SubscribeTickersService + ReconnectService>,
    trading: Box<dyn TradingService + ReconnectService>,
}

In fact a separate reconnection service may be easier to implement, we just need to ensure that these services are constructed from the same exchange object so that they have shared state.

struct Exchange {
    // Public services.
    ticker: Box<dyn SubscribeTickersService>,
    reconnect_public: Box<dyn ReconnectService>,
    // Private services.
    trading: Box<dyn TradingService>,
    reconnect_private: Box<dyn ReconnectService>,
}

[rise0chen]

Which method should be chosen:

1. delete SubscribeTickers
2. impl both SubscribeTickers and SubscribeBidAsk by the tickers channel

[rise0chen]

ensure that these services are constructed from the same exchange object is difficult, one service one connection is more easy to manage.

[Nouzan]

ensure that these services are constructed from the same exchange object is difficult, one service one connection is more easy to manage.

That would certainly be fine. However, just in case, we can actually clone Binance or Okx to share the same connection.

[rise0chen]

If you subscribe both trades and orders by a same connection. When the connection disconnect, this maybe happen:

`SubscribeTrades` recv disconnect event -> reconnect -> connected -> 
`SubscribeOrders` recv disconnect event -> reconnect -> connected -> 
`SubscribeTrades` recv disconnect event -> reconnect 
and loop

[Nouzan]

If you subscribe both trades and orders by a same connection. When the connection disconnect, this maybe happen:

`SubscribeTrades` recv disconnect event -> reconnect -> connected -> 
`SubscribeOrders` recv disconnect event -> reconnect -> connected -> 
`SubscribeTrades` recv disconnect event -> reconnect 
and loop

Oh, I think it wouldn't. Because they are sharing the same state, that means every services constructed by the same exchange object will agree on the connection state and only handle the disconnect event once.

[rise0chen]

When call reconnect function, it can query state, if state is connected, it won't reconnect?

[Nouzan]

When call reconnect function, it can query state, if state is connected, it won't reconnect?

In fact, Binance and Okx are services that will automatically reconnect on any transport failures (that is, they maintain a persistent connection). The implementations of ReconnectService for them are just to manually close the transports to trigger the automatic reconnection mechanism.

[rise0chen]

Box<dyn FetchCandlesService> can't work, I need use Box<dyn FetchCandlesService<Future = Box<dyn Future<Output = Result<<QueryCandles as Request>::Response, ExchangeError>>>>>， It's too difficult.

the value of the associated types `Future` (from trait `tower_service::Service`), `Future` (from trait `tower_service::Service`) must be specified associated type `Future` must be specified
specify the associated types: `SubscribeTradesService<Future = Type>`, `ReconnectService<Future = Type>`

[Nouzan]

It seems that we need to relax the bound of the traits like SubscribeTickersService. I am working on it and tracking by #59 .

[Nouzan]

It seems that we need to relax the bound of the traits like SubscribeTickersService. I am working on it and tracking by #59 .

I have create a draft pull request (#60) for it, and I am going to finish it as soon as possible.

[Nouzan]

60 has been merged, you can take a look at this example.

[rise0chen]

cool

wrappers are acceptable by this way. I close this PR.