The Bjarkan SOR SDK provides access to real-time market data across multiple exchanges. This includes both orderbook data and trade data. You can choose between aggregated orderbook data across all exchanges or separate data for each exchange and symbol. Live market data is crucial for making informed trading decisions and is used by the smart order router for order execution.

​
Setting Up Market Data Streams

Before you can receive market data, you need to configure and start the appropriate data stream:

import asyncio
from bjarkan import BjarkanSDK

async def main():
    async with BjarkanSDK() as sdk:
        # Configure orderbook
        await sdk.set_config(
            type="orderbook",
            aggregated=True,  # Set to False for separate exchange orderbooks
            exchanges=["binance", "bybit", "okx"],
            symbols=["BTC/USDT", "ETH/USDT"],
            depth=20
        )
        
        # Define callback function
        async def process_orderbook(orderbook):
            print(f"Received orderbook for {orderbook['symbol']}")
            print(f"Best bid: {orderbook['bids'][0][0]}, Best ask: {orderbook['asks'][0][0]}")
        
        # Start the orderbook stream
        await sdk.start_stream("orderbook", callback=process_orderbook)
        
        # Run for 60 seconds
        await asyncio.sleep(60)

if __name__ == "__main__":
    asyncio.run(main())

​
Configuration Examples and Returned Data

Let’s look at specific examples of configurations and the corresponding data structure you’ll receive in the callbacks.

​
Example 1: Aggregated Orderbook

When you set up an aggregated orderbook configuration with multiple exchanges:

await sdk.set_config(
    type="orderbook",
    aggregated=True,
    exchanges=["binance", "bybit", "okx", "htx", "coinbase"],
    symbols=["BTC/USDT", "ETH/USDT"],
    depth=200
)

The callback will receive individual orderbook updates like this:

{
    "timestamp": 1740504896368,       # Local timestamp (milliseconds)
    "exchange_timestamp": 1740504896291,  # Latest exchange timestamp
    "symbol": "BTC/USDT",             # Just one symbol per update
    "bids": [
        # [price, amount, exchange]
        [87612.38, 0.12546, "binance"],
        [87612.37, 0.04211, "okx"],
        [87612.35, 0.53982, "bybit"],
        [87612.32, 0.10000, "htx"],
        [87612.29, 0.02514, "coinbase"],
        # ... more bid entries from all exchanges, sorted by price
    ],
    "asks": [
        # [price, amount, exchange]
        [87612.39, 0.05231, "binance"],
        [87612.40, 0.11875, "okx"],
        [87612.42, 0.22461, "bybit"],
        [87612.45, 0.15000, "coinbase"],
        # ... more ask entries from all exchanges, sorted by price
    ]
}

For this configuration:

  • Each update contains data for a single symbol
  • Bids and asks from all exchanges are combined and sorted by price
  • The callback is triggered whenever there’s a change to the orderbook on any exchange
  • Updates are sent one at a time for each symbol

​
Example 2: Non-Aggregated Orderbook

When you set up a non-aggregated orderbook configuration:

await sdk.set_config(
    type="orderbook",
    aggregated=false,
    exchanges=["htx", "coinbase"],
    symbols=["ETH/USDT", "BTC/USDT"],
    depth=200
)

The callback will receive individual orderbook updates for each exchange-symbol combination:

{
    "timestamp": 1740504899057,       # Local timestamp (milliseconds)
    "exchange_timestamp": 1740504899014,  # Exchange timestamp
    "symbol": "BTC/USDT",             # Symbol for this update
    "bids": [
        # All bids from a single exchange
        [87609.2, 0.01276021, "coinbase"],
        [87609.19, 0.02126701, "coinbase"],
        [87609.15, 0.00812456, "coinbase"],
        # ... more bid entries from coinbase only
    ],
    "asks": [
        # All asks from the same exchange
        [87609.21, 0.03542187, "coinbase"],
        [87609.25, 0.12589354, "coinbase"],
        [87609.28, 0.00256913, "coinbase"],
        # ... more ask entries from coinbase only
    ]
}

With this configuration:

  • Each update contains data for a single exchange and a single symbol
  • A separate update will arrive for each exchange-symbol combination
  • Updates are triggered independently whenever an exchange’s orderbook changes
  • One update might be for “BTC/USDT” on Coinbase, the next might be for “ETH/USDT” on HTX

​
Example 3: Trade Stream with Size Filter

When you set up a trade stream with size filtering:

await sdk.set_config(
    type="trades",
    exchanges=["binance", "bybit", "okx", "coinbase"],
    symbols=["BTC/USDT", "ETH/USDT"],
    size={
        "BTC/USDT": {"USDT": 10000},  # Only trades ≥ $10,000
        "ETH/USDT": {"USDT": 10000}   # Only trades ≥ $10,000
    }
)

The callback will receive individual trade updates one at a time:

{
    "id": "4596689614",               # Exchange's trade ID
    "symbol": "BTC/USDT",             # Symbol for this trade
    "exchange": "binance",            # Exchange where trade occurred
    "timestamp": 1740504905426,       # Local timestamp (milliseconds)
    "exchange_timestamp": 1740504905304,  # Exchange timestamp
    "price": 87612.38,                # Trade price
    "amount": 0.12546,                # Trade amount (≥ $10,000 due to size filter)
    "side": "sell"                    # Trade direction
}

With this configuration:

  • Each callback delivers a single trade
  • Only trades that meet the size filter are included
  • Trades come in real-time as they occur on any of the configured exchanges
  • Each trade includes details from one exchange for one symbol

​
Fetching the Latest Orderbook Data

You can programmatically get the latest orderbook data at any time:

# Get the latest orderbook data
latest_orderbooks = await sdk.get_latest_data("orderbook")

The structure of the returned data depends on your configuration:

​
For Aggregated Orderbooks (aggregated=True):

{
    "BTC/USDT": {
        # Latest aggregated orderbook for BTC/USDT
        "timestamp": 1740504896368,
        "exchange_timestamp": 1740504896291,
        "symbol": "BTC/USDT",
        "bids": [...],
        "asks": [...]
    },
    "ETH/USDT": {
        # Latest aggregated orderbook for ETH/USDT
        ...
    }
}

​
For Non-Aggregated Orderbooks (aggregated=False):

{
    "binance": {
        "BTC/USDT": {
            # Latest orderbook for BTC/USDT on Binance
            "timestamp": 1740504899057,
            "exchange_timestamp": 1740504899014,
            "symbol": "BTC/USDT",
            "bids": [...],
            "asks": [...]
        },
        "ETH/USDT": {
            # Latest orderbook for ETH/USDT on Binance
            ...
        }
    },
    "bybit": {
        # Latest orderbooks for Bybit
        ...
    }
}

​
Fetching the Latest Trade Data

Similarly, you can get the latest trades:

# Get the latest collected trades
latest_trades = await sdk.get_latest_data("trades")

The structure will be:

{
    "binance": {
        "BTC/USDT": [
            # List of recent trades for BTC/USDT on Binance
            {
                "id": "4596689614",
                "symbol": "BTC/USDT",
                "exchange": "binance",
                "timestamp": 1740504905426,
                "exchange_timestamp": 1740504905304,
                "price": 87612.38,
                "amount": 0.12546,
                "side": "sell"
            },
            # More trades...
        ],
        "ETH/USDT": [
            # List of recent trades for ETH/USDT on Binance
            ...
        ]
    },
    "bybit": {
        # Recent trades for Bybit
        ...
    }
}

​
Fee-Aware Data

When you configure fees_bps in your configuration, the SDK will adjust the orderbook prices to account for exchange fees. This provides several advantages:

  1. True Arbitrage Detection: Identify genuine arbitrage opportunities by seeing prices with fees included.
  2. Accurate Price Comparison: Compare prices across exchanges fairly by accounting for fee differences.
  3. Realistic Execution: Better understand what your actual execution price will be after fees.

For example, with a fee of 0.1% (10 bps):

  • A bid price of 50,000 would be adjusted to 49,950 (50,000 / 1.001)
  • An ask price of 50,000 would be adjusted to 50,050 (50,000 * 1.001)

When using callbacks, the data you receive will already have these fee adjustments applied if you’ve configured fees_bps.

​
Advanced Usage: Custom Analysis

You can combine orderbook and trade data for advanced market analysis:

class MarketAnalyzer:
    def __init__(self):
        self.orderbooks = {}
        self.recent_trades = []
        self.bid_ask_spreads = {}
        
    async def orderbook_callback(self, orderbook):
        symbol = orderbook['symbol']
        
        # Store the latest orderbook for each symbol
        self.orderbooks[symbol] = orderbook
        
        # Calculate and store bid-ask spread
        if orderbook['bids'] and orderbook['asks']:
            best_bid = orderbook['bids'][0][0]
            best_ask = orderbook['asks'][0][0]
            spread = best_ask - best_bid
            spread_pct = (spread / best_bid) * 100
            self.bid_ask_spreads[symbol] = {
                'spread': spread,
                'spread_pct': spread_pct,
                'timestamp': orderbook['timestamp']
            }
            
            # Print current market conditions
            print(f"\n{symbol} Market Update:")
            print(f"Bid-Ask Spread: {spread:.2f} ({spread_pct:.3f}%)")
        
    async def trade_callback(self, trade):
        # Store recent trade
        self.recent_trades.append(trade)
        
        # Optional: limit stored trades
        if len(self.recent_trades) > 100:
            self.recent_trades.pop(0)
            
        # Analyze trade in context of orderbook
        symbol = trade['symbol']
        orderbook = self.orderbooks.get(symbol)
        
        if orderbook and orderbook['bids'] and orderbook['asks']:
            best_bid = orderbook['bids'][0][0]
            best_ask = orderbook['asks'][0][0]
            
            # Detect market orders (trades at or beyond the best bid/ask)
            if trade['side'] == 'buy' and trade['price'] >= best_ask:
                print(f"Market buy detected: {trade['amount']} @ {trade['price']}")
            elif trade['side'] == 'sell' and trade['price'] <= best_bid:
                print(f"Market sell detected: {trade['amount']} @ {trade['price']}")

​
Best Practices

When working with market data in the Bjarkan SOR SDK:

  1. Efficient Processing: Keep callback functions lightweight and non-blocking
  2. State Management: Maintain clean data structures for storing and analyzing market data
  3. Data Volume: Be prepared to handle high-frequency updates in active markets
  4. Error Handling: Implement robust error handling in all callbacks
  5. Reconnection Logic: Handle potential disconnections gracefully
  6. Memory Management: For long-running applications, implement data retention policies
  7. Timestamp Awareness: Pay attention to both local and exchange timestamps for accurate sequencing

Using these techniques, you can build sophisticated trading systems that utilize real-time market data for analysis, signal generation, and order execution.