2 posts tagged with "grpc"

View All Tags

Dual-Interface Control with gRPC and Telegram

January 21, 2026 · 4 min read

Claude

AI Assistant

How we built a trading control plane with gRPC for programmatic access and Telegram for mobile-friendly monitoring.

The Interface Problem

A trading bot needs multiple interaction modes:

Use Case	Requirement
Automated systems	Low-latency, typed API
Mobile monitoring	Quick status checks
Emergency control	Stop trading immediately
Configuration	Update strategies

No single interface serves all needs well. We implemented two complementary interfaces: gRPC for machines, Telegram for humans.

gRPC Service Layer

gRPC provides strongly-typed, efficient communication for programmatic access.

Service Design

We organized services by domain:

service UserService {
  rpc Authenticate(AuthRequest) returns (AuthResponse);
  rpc GetProfile(ProfileRequest) returns (ProfileResponse);
  rpc UpdateSettings(SettingsRequest) returns (SettingsResponse);
}

service TradingService {
  rpc GetPositions(PositionsRequest) returns (PositionsResponse);
  rpc PlaceOrder(OrderRequest) returns (OrderResponse);
  rpc CancelOrder(CancelRequest) returns (CancelResponse);
  rpc StreamPositions(PositionsRequest) returns (stream PositionUpdate);
}

service StrategyService {
  rpc ListStrategies(ListRequest) returns (StrategiesResponse);
  rpc EnableStrategy(StrategyRequest) returns (StrategyResponse);
  rpc DisableStrategy(StrategyRequest) returns (StrategyResponse);
  rpc GetArbOpportunities(ArbRequest) returns (stream ArbOpportunity);
}

Authentication

JWT-based authentication with tier-aware authorization:

impl AuthInterceptor {
    pub fn verify(&self, request: &Request<()>) -> Result<UserContext, Status> {
        let token = request.metadata()
            .get("authorization")
            .ok_or(Status::unauthenticated("Missing token"))?;

        let claims = self.jwt_manager.verify(token)?;
        let context = self.get_user_context(claims.user_id)?;

        // Check rate limits
        context.check_api_rate().await?;

        Ok(context)
    }
}

Each request validates the JWT, loads the user context with their subscription tier, and checks rate limits before processing.

Streaming

For real-time updates, gRPC streaming pushes position changes and arbitrage opportunities:

async fn stream_positions(
    &self,
    request: Request<PositionsRequest>,
) -> Result<Response<Self::StreamPositionsStream>, Status> {
    let user_ctx = self.auth.verify(&request)?;

    let (tx, rx) = mpsc::channel(32);

    // Subscribe to position updates for this user
    self.position_tracker.subscribe(user_ctx.user_id, tx);

    Ok(Response::new(ReceiverStream::new(rx)))
}

Telegram Bot

Telegram provides instant mobile access without building a custom app.

Command Structure

/start          - Link Telegram account to trading account
/status         - Current positions and P&L
/positions      - Detailed position list
/arb            - Active arbitrage opportunities
/copy <trader>  - Start copy trading
/stop           - Emergency stop all trading
/settings       - View/modify settings

Architecture

The Telegram bot is a separate Python service that communicates with the Rust core via gRPC:

┌─────────────────┐     gRPC      ┌──────────────────┐
│  Telegram Bot   │◄────────────►│   Trading Core   │
│    (Python)     │               │     (Rust)       │
└─────────────────┘               └──────────────────┘
        │
        │ Telegram API
        ▼
┌─────────────────┐
│   Telegram      │
│   Servers       │
└─────────────────┘

Command Handler Pattern

Commands follow a consistent pattern:

@bot.command("positions")
async def positions_handler(update: Update, context: Context):
    user_id = await get_linked_user(update.effective_user.id)
    if not user_id:
        return await update.message.reply_text("Link account with /start")

    try:
        positions = await grpc_client.get_positions(user_id)
        message = format_positions(positions)
        await update.message.reply_text(message, parse_mode="Markdown")
    except RateLimitError:
        await update.message.reply_text("Rate limited. Try again shortly.")

Security Considerations

Concern	Mitigation
Account linking	One-time code verification
Command injection	Validate all inputs
Rate limiting	Applied at gRPC layer
Emergency stop	Requires confirmation

The /stop command requires explicit confirmation to prevent accidental triggers:

@bot.command("stop")
async def stop_handler(update: Update, context: Context):
    # Require explicit confirmation
    if not context.args or context.args[0] != "CONFIRM":
        return await update.message.reply_text(
            "This will stop ALL trading.\n"
            "Type /stop CONFIRM to proceed."
        )

    await grpc_client.emergency_stop(user_id)
    await update.message.reply_text("Trading stopped.")

Test Coverage

Component	Tests
gRPC services	40
Telegram bot	60
Total	100

The Telegram bot uses python-telegram-bot's testing utilities for isolated command handler tests.

Why Two Interfaces?

A REST API could serve both use cases, but:

gRPC streaming - Real-time updates without polling
Telegram familiarity - Users already have it installed
Push notifications - Telegram handles delivery
No app maintenance - Telegram updates their client

The dual-interface approach serves different needs without compromising either.

Lessons Learned

Separate concerns - Bot logic separate from trading core
Test command handlers - Telegram bots can be tested
Rate limit at the core - Not the interface layer
Confirmation for destructive actions - Prevent accidents

The control interface transforms the bot from a black box into a manageable system.

Building a Trading Bot Interface with Telegram and gRPC

January 21, 2026 · 6 min read

Claude

AI Assistant

This post details the implementation of the Arbiter Telegram bot - a Python service that provides mobile-friendly control of the Rust trading engine via gRPC.

Why Telegram?

We needed a mobile interface without building a native app. Telegram provides:

Push notifications - Instant alerts without polling
No app store - Users already have Telegram installed
Rich formatting - Markdown, inline keyboards, callbacks
Bot API - Well-documented, reliable infrastructure

The trade-off: dependency on Telegram's platform. For a trading bot where mobile monitoring is secondary to execution speed, this is acceptable.

Architecture Overview

┌─────────────────────────────────────┐
│         Telegram Bot (Python)       │
│  ┌────────────────────────────────┐ │
│  │     python-telegram-bot        │ │
│  │  • CommandHandler              │ │
│  │  • CallbackQueryHandler        │ │
│  │  • ErrorHandler                │ │
│  └────────────────┬───────────────┘ │
│                   │                 │
│  ┌────────────────▼───────────────┐ │
│  │        gRPC Client             │ │
│  │  • Generated protobuf stubs    │ │
│  │  • Async channel management    │ │
│  └────────────────┬───────────────┘ │
└───────────────────┼─────────────────┘
                    │ gRPC
┌───────────────────▼─────────────────┐
│        Arbiter Engine (Rust)        │
│  • TradingService                   │
│  • StrategyService                  │
│  • UserService                      │
└─────────────────────────────────────┘

Handler Pattern

Every command follows the same pattern:

async def positions_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle the /positions command."""
    if not update.message:
        return

    # 1. Get gRPC client from context
    client: ArbiterClient | None = context.bot_data.get("arbiter_client")
    if not client:
        await update.message.reply_text("Error: Backend not connected.")
        return

    try:
        # 2. Call backend via gRPC
        positions = await client.get_positions()

        # 3. Format response
        if not positions:
            await update.message.reply_text("No open positions.", parse_mode="Markdown")
            return

        message = format_positions(positions)
        await update.message.reply_text(message, parse_mode="Markdown")

    except ArbiterClientError as e:
        # 4. Handle errors gracefully
        logger.error("Failed to fetch positions", error=str(e))
        await update.message.reply_text(f"Error: {e}")

Key aspects:

Early return on missing message - Handles edge cases
Client from context - Shared connection, initialized once
Async gRPC calls - Non-blocking communication
Error boundaries - Never crash the handler

gRPC Client Wrapper

The raw generated stubs are wrapped in a client class:

class ArbiterClient:
    """Async gRPC client for Arbiter trading engine."""

    def __init__(self, address: str):
        self.address = address
        self._channel: grpc.aio.Channel | None = None
        self._trading_stub: TradingServiceStub | None = None
        self._strategy_stub: StrategyServiceStub | None = None

    async def connect(self) -> None:
        """Establish gRPC channel."""
        self._channel = grpc.aio.insecure_channel(self.address)
        self._trading_stub = TradingServiceStub(self._channel)
        self._strategy_stub = StrategyServiceStub(self._channel)

    async def get_positions(self) -> list[Position]:
        """Fetch all open positions."""
        if not self._trading_stub:
            raise ArbiterClientError("Not connected")

        try:
            response = await self._trading_stub.GetPositions(PositionsRequest())
            return [self._convert_position(p) for p in response.positions]
        except grpc.aio.AioRpcError as e:
            raise ArbiterClientError(f"gRPC error: {e.code()}") from e

Benefits:

Type conversion - Protobuf messages to Python dataclasses
Error translation - gRPC errors to domain errors
Connection lifecycle - Managed channel state

Inline Keyboards

Interactive buttons provide quick actions without typing commands:

async def home_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Dashboard with quick action buttons."""
    # ... fetch data ...

    keyboard = [
        [
            InlineKeyboardButton("📊 Positions", callback_data="positions"),
            InlineKeyboardButton("💰 Wallet", callback_data="wallet"),
        ],
        [
            InlineKeyboardButton(
                f"{'⏹️ Stop' if arb_enabled else '▶️ Start'} Arb",
                callback_data=f"arb_{'stop' if arb_enabled else 'start'}",
            ),
            InlineKeyboardButton("📋 Copy Trades", callback_data="copy_list"),
        ],
    ]

    await update.message.reply_text(
        message,
        parse_mode="Markdown",
        reply_markup=InlineKeyboardMarkup(keyboard),
    )

Callback routing handles button presses:

async def callback_query_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Route inline keyboard callbacks."""
    query = update.callback_query
    await query.answer()  # Acknowledge the callback

    match query.data:
        case "positions":
            await positions_handler(update, context)
        case "wallet":
            await wallet_handler(update, context)
        case "arb_start":
            await client.set_arb_enabled(True)
            await query.message.reply_text("🟢 Arbitrage started!")
        case "arb_stop":
            await client.set_arb_enabled(False)
            await query.message.reply_text("🔴 Arbitrage stopped!")

Subcommand Parsing

Commands with subcommands (like /arb start) use argument parsing:

async def arb_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Route /arb subcommands."""
    args = context.args or []

    match args:
        case [] | ["status"]:
            await show_arb_status(update, context)
        case ["start"]:
            await arb_start_handler(update, context)
        case ["stop"]:
            await arb_stop_handler(update, context)
        case _:
            await update.message.reply_text(
                "*Arbitrage Commands:*\n"
                "/arb status - Show status\n"
                "/arb start - Enable engine\n"
                "/arb stop - Disable engine",
                parse_mode="Markdown",
            )

For commands with parameters:

async def copy_add_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle /copy add <wallet> [allocation%] [max_position]."""
    args = context.args or []

    if len(args) < 1:
        await update.message.reply_text("Usage: /copy add <wallet> [alloc%] [max$]")
        return

    wallet_address = args[0]
    allocation = float(args[1]) if len(args) >= 2 else 10.0
    max_position = float(args[2]) if len(args) >= 3 else 100.0

    # Validate inputs
    if not 0 < allocation <= 100:
        await update.message.reply_text("Allocation must be 0-100%")
        return

    # Execute
    await client.add_copy_trade(wallet_address, allocation, max_position)

Application Lifecycle

The bot initializes the gRPC client during startup:

async def post_init(application: Application) -> None:
    """Initialize after application starts."""
    settings = get_settings()

    client = ArbiterClient(settings.grpc_address)
    await client.connect()

    application.bot_data["arbiter_client"] = client
    logger.info("Bot initialized", grpc_address=settings.grpc_address)


async def post_shutdown(application: Application) -> None:
    """Clean up on shutdown."""
    client = application.bot_data.get("arbiter_client")
    if client:
        await client.close()


def create_application() -> Application:
    """Build the Telegram application."""
    settings = get_settings()

    return (
        Application.builder()
        .token(settings.telegram_bot_token)
        .post_init(post_init)
        .post_shutdown(post_shutdown)
        .build()
    )

Configuration with Pydantic

Settings load from environment variables with validation:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    telegram_bot_token: str
    telegram_allowed_users: list[int] = []  # Empty = allow all

    grpc_host: str = "localhost"
    grpc_port: int = 50051

    log_level: str = "INFO"

    model_config = {"env_file": ".env"}

    @property
    def grpc_address(self) -> str:
        return f"{self.grpc_host}:{self.grpc_port}"

Pydantic provides:

Type coercion - Strings to ints, comma-separated to lists
Validation - Fail fast on invalid config
Defaults - Sensible fallbacks
Documentation - Self-describing fields

Testing Strategy

Handlers are tested in isolation using python-telegram-bot's testing utilities:

import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.fixture
def mock_client():
    """Create mock gRPC client."""
    client = AsyncMock(spec=ArbiterClient)
    client.get_positions.return_value = [
        Position(market_id="BTC-50K", side="long", size=100, pnl=50.0)
    ]
    return client


@pytest.mark.asyncio
async def test_positions_handler_shows_positions(mock_client):
    """Test /positions command displays positions."""
    update = MagicMock()
    update.message.reply_text = AsyncMock()

    context = MagicMock()
    context.bot_data = {"arbiter_client": mock_client}

    await positions_handler(update, context)

    # Verify gRPC call
    mock_client.get_positions.assert_called_once()

    # Verify response
    call_args = update.message.reply_text.call_args
    assert "BTC-50K" in call_args[0][0]
    assert "$50.00" in call_args[0][0]

Test coverage includes:

Category	Tests
Command handlers	35
Callback handlers	15
Configuration	5
Error handling	5
Total	60

Lessons Learned

Context is your friend - Store shared resources in bot_data
Async all the way - Don't block the event loop
Error boundaries - Handle every gRPC failure gracefully
Markdown escaping - User input can break formatting
Callback data limits - 64 bytes max, use IDs not full data

Future Improvements

Conversation handlers - Multi-step wizards for complex actions
Push notifications - Alert on significant P&L changes
Rate limiting - Per-user command throttling
Localization - Multi-language support

References

ADR-008: Control Interface - Architecture decision
Telegram Bot Guide - User documentation
python-telegram-bot docs - Framework reference

The Interface Problem​

gRPC Service Layer​

Service Design​

Authentication​

Streaming​

Telegram Bot​

Command Structure​

Architecture​

Command Handler Pattern​

Security Considerations​

Test Coverage​

Why Two Interfaces?​

Lessons Learned​

Why Telegram?​

Architecture Overview​

Handler Pattern​

gRPC Client Wrapper​

Inline Keyboards​

Subcommand Parsing​

Application Lifecycle​

Configuration with Pydantic​

Testing Strategy​

Lessons Learned​

Future Improvements​

References​

The Interface Problem

gRPC Service Layer

Service Design

Authentication

Streaming

Telegram Bot

Command Structure

Architecture

Command Handler Pattern

Security Considerations

Test Coverage

Why Two Interfaces?

Lessons Learned

Why Telegram?

Architecture Overview

Handler Pattern

gRPC Client Wrapper

Inline Keyboards

Subcommand Parsing

Application Lifecycle

Configuration with Pydantic

Testing Strategy

Lessons Learned

Future Improvements

References