inv2004 / coinbase-pro-rs Goto Github PK

It seems that Coinbase has removed base_min_size from their product api, which causes get_product to crash to an error:

thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value:
Serde {
  error: Error("missing field `base_min_size`", line: 1, column: 388),
  data: "{\"id\":\"CRV-EUR\",\"base_currency\":\"CRV\",\"quote_currency\":\"EUR\",\"quote_increment\":\"0.0001\",\"base_increment\":\"0.01\",\"display_name\":\"CRV/EUR\",\"min_market_funds\":\"0.84\",\"margin_enabled\":false,\"fx_stablecoin\":false,\"max_slippage_percentage\":\"0.03000000\",\"post_only\":false,\"limit_only\":false,\"cancel_only\":false,\"trading_disabled\":false,\"status\":\"online\",\"status_message\":\"\",\"auction_mode\":false}"
}

All, where do you find the API docs for the api.pro.coinbase.com? When I go into their new cloud docs all I see is something that they call exchange; api.exchange.coinbase.com. Even the link at https://docs.pro.coinbase.com sends me to the same 'exchange' docs. I don't see the api.pro.coinbase.com anywhere in there docs.

I tried using the same secret, passphrase and key with exchange but it doesn't work. Am I missing something or are trying to move the api url from api.pro.coinbase.com to api.exchange.coinbase.com? If not can someone send me a link to the pro docs?

Thanks.

The official API (and the docs copied from there) allow specifying "all" orders, but the code only allows passing Option<OrderStatus>:

https://docs.rs/coinbase-pro-rs/0.6.9/src/coinbase_pro_rs/private.rs.html#357-372

Not sure if the parameter would need it's own enum, or allow a list of order statuses, or if "all" should be the default.
As is it is a bit clumsy to get "all", as that would require looping over the enum.

Hi, I want to implement dynamic subscribe functionality.
coinbase-pro API support subscription to new items in the middle of data reception.

I think it can be done without breaking backward compatibility by simply changes WSFeed's return type

from

impl Stream<Item = Result<Message, CBError>>

to

impl Stream<Item = Result<Message, CBError>> + Sink<TMessage, Error = TError> + Unpin

below is more detailed PoC code for this changes

use async_trait::async_trait;

use futures::{future, Sink, Stream};
use futures_util::{future::TryFutureExt, sink::SinkExt, stream::TryStreamExt};
use serde_json;
use tokio_tungstenite::{
    connect_async, tungstenite::Error as TError, tungstenite::Message as TMessage,
};

fn convert_msg(msg: TMessage) -> Message {
    match msg {
        TMessage::Text(str) => serde_json::from_str(&str).unwrap_or_else(|e| {
            Message::InternalError(CBError::Serde {
                error: e,
                data: str,
            })
        }),
        _ => unreachable!(), // filtered in stream
    }
}
use crate::{structs::wsfeed::Subscribe, CBError};
use crate::{structs::wsfeed::*, WSError};

#[async_trait]
pub trait CBWSExt {
    async fn subscribe(&mut self, subscribe: Subscribe) -> Result<(), CBError>;
}

#[async_trait]
impl<T> CBWSExt for T
where
    T: Stream<Item = Result<Message, CBError>> + Sink<TMessage, Error = TError> + Unpin + Send,
{
    async fn subscribe(&mut self, subscribe: Subscribe) -> Result<(), CBError> {
        let subscribe = serde_json::to_string(&subscribe).unwrap();
        self.send(TMessage::Text(subscribe))
            .map_err(|e| CBError::Websocket(WSError::Send(e)))
            .await
    }
}

pub async fn connect_cbws(
    uri: &str,
) -> Result<
    impl Stream<Item = Result<Message, CBError>> + Sink<TMessage, Error = TError> + Unpin,
    CBError,
> {
    let (stream, _) = connect_async(uri)
        .map_err(|e| CBError::Websocket(WSError::Connect(e)))
        .await?;

    log::debug!("subsription sent");

    Ok(stream
        .try_filter(|msg| future::ready(msg.is_text()))
        .map_ok(convert_msg)
        .map_err(|e| CBError::Websocket(WSError::Read(e))))
}

async fn test(uri: &str, product_ids: &[&str], channels: &[ChannelType]) {
    let subscribe = Subscribe {
        _type: SubscribeCmd::Subscribe,
        product_ids: product_ids.into_iter().map(|x| x.to_string()).collect(),
        channels: channels
            .to_vec()
            .into_iter()
            .map(|x| Channel::Name(x))
            .collect::<Vec<_>>(),
        auth: None,
    };

    let mut cbws = connect_cbws(uri).await.expect("failed to connect.");
    cbws.subscribe(subscribe).await.unwrap();
}

is it okay to modify the code in this way?

I want to know if you are thinking about a better way or if there was a reason for not implementing the subscription functionality.

thanks!

The Public::get_trades() method does not support pagination which is required to retrieve historical trade data. To avoid a breaking change this could be addressed by adding a new method resembling pub fn get_trades_paginated(&self, product_id: &str, before: Option<usize>, after: Option<usize>) ....

new_size field is missing in Full::Change enum variant.

I think that the corresponding value is an optional value, because one of the official coinbase pro API reference's example has no new_size field too.

we need to checks if there is a document specifying the fields of each variant, and that there are no problems with other variants.

with the most recent commits.

thank you! :)

I'm getting this error running the ws_url url of coinbase pro. Any ideas?

`
extern crate coinbase_pro_rs;
extern crate hyper;
extern crate tokio;

use coinbase_pro_rs::{ ASync, Public, WS_URL, structs };
use hyper::rt::Future;

fn main() {
let client: Public = Public::new_with_keep_alive(WS_URL, false);

let btc_usd = client
    .get_book::<structs::public::BookRecordL1>("BTC-USD")
    .map_err(|err| {
        println!("{:?}", err);
    })
    .and_then(|book| {
        println!("BTC-USD: {:?}", book);
        Ok(())
    });
tokio::run(btc_usd);

}
`

• choose decimal lib (rust_decimal ?)
• compare performance
• implement

Hello!

Have you planned to update this project with new async std and/or tokio updates?

Thanks!