Python

Anthid provides generated Protocol Buffer definitions that can be used to build Python clients for the Streaming API. This guide demonstrates how to generate the Python gRPC client and subscribe to real-time trading events.

Prerequisites

Install the required Python packages.

python -m pip install grpcio grpcio-tools protobuf

Download the Protobuf Definitions

Clone the Anthid protobuf repository.

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

Generate Python Client Code

Generate Python classes from the protobuf definitions.

mkdir -p generated/python

python -m grpc_tools.protoc \
  -I anthid-proto \
  --python_out=generated/python \
  --grpc_python_out=generated/python \
  anthid-proto/trading/*.proto

export PYTHONPATH="$PWD/generated/python:$PYTHONPATH"

The generated trading namespace contains the message definitions and gRPC service stubs required to interact with Anthid services.

Authentication

Streaming requests authenticate with either a Bearer access token or an API key passed as gRPC metadata. This guide uses an API key.

metadata = (
    ("x-api-key", "YOUR_API_KEY"),
)

The example below demonstrates connecting to the server-streaming endpoint and subscribing to broker order and position events.

import grpc

from trading import trading_service_pb2 as trading_pb2
from trading import trading_service_pb2_grpc as trading_pb2_grpc
from trading import trading_types_pb2


API_KEY = "YOUR_API_KEY"

request = trading_pb2.Subscribe(
    trading_account_id="YOUR_TRADING_ACCOUNT_ID",
    environment=trading_types_pb2.BROKER_ENVIRONMENT_PAPER,
    subscriptions=[
        trading_pb2.EVENT_TYPE_BROKER_ORDER,
        trading_pb2.EVENT_TYPE_BROKER_POSITION,
    ],
)

credentials = grpc.ssl_channel_credentials()

with grpc.secure_channel(
    "stream.anthid.com:443",
    credentials,
) as channel:
    stub = trading_pb2_grpc.TradingServiceStub(
        channel,
    )

    stream = stub.SubscribeEvents(
        request,
        metadata=(
            ("x-api-key", API_KEY),
        ),
    )

    for event in stream:
        if event.HasField("heartbeat"):
            print("heartbeat")

        elif event.HasField("status"):
            print(
                "status:",
                event.status.message,
            )

        elif event.HasField("error"):
            print(
                "error:",
                event.error.message,
            )

        elif event.HasField("primed"):
            print("initial state complete:", event.primed.event_type)

        elif event.HasField("broker_order"):
            print(
                "order:",
                event.broker_order,
            )

        elif event.HasField("broker_position"):
            print(
                "position:",
                event.broker_position,
            )

environment is required: set it to trading_types_pb2.BROKER_ENVIRONMENT_PAPER or trading_types_pb2.BROKER_ENVIRONMENT_LIVE to match the account. The enum is defined in trading_types.proto, so it is generated into trading_types_pb2 rather than trading_service_pb2. To receive retained order snapshots first, set order_delivery=trading_pb2.ORDER_DELIVERY_PRIMED in Subscribe. 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.

Bidirectional Streaming

Backend and native applications may prefer the bidirectional StreamEvents endpoint. This endpoint allows subscriptions and unsubscriptions to be sent over an existing stream without creating a new connection.

import grpc
from queue import Queue

from trading import trading_service_pb2 as trading_pb2
from trading import trading_service_pb2_grpc as trading_pb2_grpc
from trading import trading_types_pb2


outbound = Queue()


def requests():
    while True:
        yield outbound.get()


outbound.put(
    trading_pb2.StreamEventsRequest(
        subscribe=trading_pb2.Subscribe(
            trading_account_id=
                "YOUR_TRADING_ACCOUNT_ID",
            environment=trading_types_pb2.BROKER_ENVIRONMENT_PAPER,
            subscriptions=[
                trading_pb2.EVENT_TYPE_BROKER_ORDER,
                trading_pb2.EVENT_TYPE_BROKER_POSITION,
            ],
        )
    )
)


credentials = grpc.ssl_channel_credentials()

with grpc.secure_channel(
    "stream.anthid.com:443",
    credentials,
) as channel:
    stub = trading_pb2_grpc.TradingServiceStub(
        channel,
    )

    stream = stub.StreamEvents(
        requests(),
        metadata=(
            ("x-api-key", "YOUR_API_KEY"),
        ),
    )

    for event in stream:
        print(event)

Handling Disconnects

Applications should automatically reconnect when a stream is interrupted. After reconnecting:

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

Streaming Overview
Trading Accounts
Orders
Positions
Ledger
Intents

Connections

Organizations

Controls

Accounts

Intents

Ledger

Instruments

Stream

MCP

Billing

Prerequisites

Download the Protobuf Definitions

Generate Python Client Code

Authentication

Bidirectional Streaming

Handling Disconnects

​Prerequisites

​Download the Protobuf Definitions

​Generate Python Client Code

​Authentication

​Subscribe to Events

​Bidirectional Streaming

​Handling Disconnects

​Related Resources

Prerequisites

Download the Protobuf Definitions

Generate Python Client Code

Authentication

Subscribe to Events

Bidirectional Streaming

Handling Disconnects

Related Resources