Rust

Anthid provides Protocol Buffer definitions that can be used to generate Rust clients with Tonic. This guide demonstrates connecting to the Streaming API and receiving real-time broker order and position updates.

Prerequisites

Add the required dependencies.

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"
tonic = { version = "0.14", features = ["tls-webpki-roots"] }
tonic-prost = "0.14"
prost = "0.14"
prost-types = "0.14"

Generate Rust Types

Install the Protocol Buffer compiler, then clone the Anthid definitions.

git clone https://github.com/anthid-labs/proto.git anthid-proto

Example build.rs:

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_prost_build::configure()
        .build_client(true)
        .build_server(false)
        .compile_protos(
            &[
                "anthid-proto/trading/trading_service.proto",
            ],
            &["anthid-proto"],
        )?;

    Ok(())
}

Example Cargo.toml build dependency:

[build-dependencies]
tonic-prost-build = "0.14"

Authentication

Anthid authenticates streaming requests with either a Bearer access token or the x-api-key metadata header. This guide uses an API key.

use tonic::metadata::MetadataValue;

request.metadata_mut().insert(
    "x-api-key",
    MetadataValue::try_from("YOUR_API_KEY")?,
);

The SubscribeEvents endpoint provides a server-streaming interface suitable for most applications.

use tonic::{
    metadata::MetadataValue,
    Request,
};

pub mod trading {
    tonic::include_proto!("trading");
}

use trading::{
    trading_service_client::TradingServiceClient,
    EventType,
    Subscribe,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>>
{
    let mut client =
        TradingServiceClient::connect(
            "https://stream.anthid.com",
        )
        .await?;

    let mut request = Request::new(
        Subscribe {
            trading_account_id:
                "YOUR_TRADING_ACCOUNT_ID"
                    .to_string(),
            environment:
                trading::BrokerEnvironment::Paper
                    as i32,
            subscriptions: vec![
                EventType::BrokerOrder as i32,
                EventType::BrokerPosition
                    as i32,
            ],
            ..Default::default()
        },
    );

    request.metadata_mut().insert(
        "x-api-key",
        MetadataValue::try_from(
            "YOUR_API_KEY",
        )?,
    );

    let mut stream = client
        .subscribe_events(request)
        .await?
        .into_inner();

    while let Some(event) =
        stream.message().await?
    {
        println!("{event:?}");
    }

    Ok(())
}

environment is required: set it to trading::BrokerEnvironment::Paper as i32 or trading::BrokerEnvironment::Live as i32 to match the account, because Default::default() leaves it unspecified and the server refuses that. Set order_delivery: trading::OrderDelivery::Primed as i32 on Subscribe to receive retained order snapshots first. Default::default() leaves instrument filters empty and order delivery live-only. Positions always prime retained state. Wait for the corresponding Primed event before treating initial delivery as complete. Terminal orders and zero-quantity positions expire from retained stream state after one hour. Rebuild a fresh snapshot set on reconnect so an old cache does not retain entries that no longer replay. See Rebuilding a cached view.

Handling Events

Each TradingEvent contains a single event variant.

while let Some(event) =
    stream.message().await?
{
    match event.event {
        Some(
            trading::trading_event::Event::Heartbeat(_),
        ) => {
            println!("heartbeat");
        }

        Some(
            trading::trading_event::Event::Status(
                status,
            ),
        ) => {
            println!(
                "status: {}",
                status.message
            );
        }

        Some(
            trading::trading_event::Event::Error(
                error,
            ),
        ) => {
            println!(
                "error: {}",
                error.message
            );
        }

        Some(
            trading::trading_event::Event::BrokerOrder(
                order,
            ),
        ) => {
            println!(
                "order: {:?}",
                order
            );
        }

        Some(
            trading::trading_event::Event::BrokerPosition(
                position,
            ),
        ) => {
            println!(
                "position: {:?}",
                position
            );
        }

        Some(trading::trading_event::Event::Primed(primed)) => {
            println!("initial state complete: {}", primed.event_type);
        }

        Some(trading::trading_event::Event::BrokerTrade(trade)) => {
            println!("trade: {trade:?}");
        }

        Some(trading::trading_event::Event::IntentAction(action)) => {
            println!("intent action: {action:?}");
        }

        None => {}
    }
}

Bidirectional Streaming

Backend and native applications may use StreamEvents when subscriptions need to be modified over an existing connection.

use tokio_stream::iter;
use tonic::Request;

use trading::{
    EventType,
    StreamEventsRequest,
    Subscribe,
};

let request_stream = iter(vec![
    StreamEventsRequest {
        msg: Some(
            trading::stream_events_request::Msg::Subscribe(
                Subscribe {
                    trading_account_id:
                        "YOUR_TRADING_ACCOUNT_ID"
                            .to_string(),
                    environment:
                        trading::BrokerEnvironment::Paper
                            as i32,
                    subscriptions: vec![
                        EventType::BrokerOrder
                            as i32,
                        EventType::BrokerPosition
                            as i32,
                    ],
                    ..Default::default()
                },
            ),
        ),
    },
]);

let mut request =
    Request::new(request_stream);

request.metadata_mut().insert(
    "x-api-key",
    MetadataValue::try_from(
        "YOUR_API_KEY",
    )?,
);

let response = client
    .stream_events(request)
    .await?;

let mut stream =
    response.into_inner();

Handling Disconnects

Applications should automatically reconnect when a stream closes unexpectedly. After reconnecting:

Create a new stream.
Authenticate using a Bearer access token or API key.
Re-subscribe to required trading accounts.
Resume event processing.

Streaming Overview
Python Guide
Trading Accounts
Orders
Positions
Ledger
Intents

Connections

Organizations

Controls

Accounts

Intents

Ledger

Instruments

Stream

MCP

Billing

Prerequisites

Generate Rust Types

Authentication

Handling Events

Bidirectional Streaming

Handling Disconnects

​Prerequisites

​Generate Rust Types

​Authentication

​Subscribe to Events

​Handling Events

​Bidirectional Streaming

​Handling Disconnects

​Related Resources

Prerequisites

Generate Rust Types

Authentication

Subscribe to Events

Handling Events

Bidirectional Streaming

Handling Disconnects

Related Resources