inv2004 / coinbase-pro-rs

Coinbase pro client for Rust

Coinbase pro client for Rust

Supports SYNC/ASYNC/Websocket-feed data support

Features

• private and public API
• sync and async support
• websocket-feed support

Examples

Cargo.toml:

[dependencies]
coinbase-pro-rs = "0.7.1"

Async

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

fn main() {
    let client: Public<ASync> = Public::new_with_keep_alive(SANDBOX_URL, false);
    // if keep_alive is not disables - tokio::run will hold the connection without exiting the example
    let f = client.get_time()
        .map_err(|_| ())
        .and_then(|time| {
            println!("Coinbase.time: {}", time.iso);
            Ok(())
        });

    tokio::run(f); // waiting for tokio
}

Sync

use coinbase_pro_rs::{Public, Sync, SANDBOX_URL};

fn main() {
   let client: Public<Sync> = Public::new(SANDBOX_URL);
   let time = client.get_time().unwrap();
   println!("Coinbase.time: {}", time.iso);
}

Websocket

use futures::{Future, Stream};
use coinbase_pro_rs::{WSFeed, WS_SANDBOX_URL};
use coinbase_pro_rs::structs::wsfeed::*;

fn main() {
    let stream = WSFeed::connect(WS_SANDBOX_URL,
        &["BTC-USD"], &[ChannelType::Heartbeat])
        .await
        .unwrap();

    let f = stream
        .take(10)
        .for_each(|msg| {
        match msg {
            Message::Heartbeat {sequence, last_trade_id, time, ..} => println!("{}: seq:{} id{}",
                                                                               time, sequence, last_trade_id),
            Message::Error {message} => println!("Error: {}", message),
            Message::InternalError(_) => panic!("internal_error"),
            other => println!("{:?}", other)
        }
        Ok(())
    });

    tokio::run(f.map_err(|_| panic!("stream fail")));
}

Api supported:

• SYNC
• ASYNC
• Websocket-Feed

API

• Requests
• Pagination
• Types
• Private
  • Authentication
  • Accounts
  • Orders
  • Fills
  • Deposits
    • List
  • Withdrawals
    • List
  • Payment Methods
  • Coinbase Accounts
  • Fees
  • Reports
  • User Account
• Market Data
  • Products
  • Currencies
  • Time
• Websocket Feed
  • heartbeat
  • ticker
  • level2
  • user
  • matches
  • full

FIX API

by request

OrderBook

https://github.com/inv2004/orderbook-rs

Tests

cargo test

coinbase-pro-rs's Issues

Question on API Docs

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.

Add pagination to the Public::get_trades API

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>) ....

Could you publish a new release?

with the most recent commits.

thank you! :)

base_min_size not present in get_product, causing a crash

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}"
}

decimal (optional?) support

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

get_orders() API doesn't allow getting "all" orders

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.

dynamic subscription

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!

Async updates

Hello!

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

Thanks!

Avoid creation of runtime on every Sync request

Serde { error: Error("EOF while parsing a value", line: 1, column: 0), data: "" }

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);

}
`

new_size value missing in Full::Change json message

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.