Liquidity Provider Integration

Liquidity Provider Integration

Liquidity provider (LP) integration is connecting external liquidity sources to your trading platform. Instead of waiting for organic market makers or building your own MM bot from scratch, the platform connects to professional LPs providing two-sided quotes via API.

Types of Liquidity Providers

Prime Brokers

Institutional LPs: Cumberland (DRW), Wintermute, Jump Crypto, B2C2, Galaxy Digital. Provide competitive quotes on large volumes via FIX protocol or REST/WebSocket API.

Requirements: KYB (Know Your Business), minimum monthly volume ($1M+), legal agreement, sometimes deposit.

Aggregated Liquidity APIs

B2Broker, FXCM Crypto, Leverate — aggregated liquidity providers available without institutional requirements. Suitable for small/medium exchanges.

DEX/AMM as Liquidity Source

For crypto platforms: aggregate liquidity from Uniswap, Curve, Balancer via 0x Protocol API or 1inch Fusion. Useful for long-tail tokens.

API Integration

REST API Integration (B2Broker Example)

import httpx
from decimal import Decimal
import asyncio

class B2BrokerLPClient:
    def __init__(self, api_key: str, api_secret: str, base_url: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = base_url
        self.client = httpx.AsyncClient()

    async def get_quote(
        self,
        symbol: str,
        side: str,
        quantity: Decimal
    ) -> LPQuote:
        payload = {
            'symbol': symbol,
            'side': side.upper(),
            'quantity': str(quantity)
        }
        headers = self._sign_request(payload)

        response = await self.client.post(
            f"{self.base_url}/v1/quote",
            json=payload,
            headers=headers
        )
        data = response.json()

        return LPQuote(
            provider='b2broker',
            symbol=symbol,
            side=side,
            price=Decimal(str(data['price'])),
            quantity=Decimal(str(data['quantity'])),
            quote_id=data['quoteId'],
            expires_in_ms=data['expiresIn']
        )

    async def execute_quote(self, quote_id: str) -> LPExecution:
        response = await self.client.post(
            f"{self.base_url}/v1/execute",
            json={'quoteId': quote_id},
            headers=self._sign_request({'quoteId': quote_id})
        )
        data = response.json()
        return LPExecution(
            execution_id=data['executionId'],
            fill_price=Decimal(str(data['fillPrice'])),
            fill_quantity=Decimal(str(data['fillQuantity'])),
            status=data['status']
        )

WebSocket Streaming Quotes

For real-time market data from LP:

class LPWebSocketFeed:
    async def subscribe(self, symbols: list[str]):
        async with websockets.connect(self.ws_url) as ws:
            # Authorization
            await ws.send(json.dumps({
                'type': 'auth',
                'apiKey': self.api_key,
                'signature': self.generate_signature()
            }))

            # Subscribe to quotes
            await ws.send(json.dumps({
                'type': 'subscribe',
                'channels': ['quotes'],
                'symbols': symbols
            }))

            async for message in ws:
                data = json.loads(message)
                if data['type'] == 'quote':
                    await self.on_quote(LPQuote(
                        provider=self.provider_name,
                        symbol=data['symbol'],
                        bid=Decimal(data['bid']),
                        ask=Decimal(data['ask']),
                        bid_size=Decimal(data['bidSize']),
                        ask_size=Decimal(data['askSize']),
                        timestamp=data['timestamp']
                    ))

Aggregating Multiple Providers

class LiquidityAggregator:
    def __init__(self, providers: list[BaseLPClient]):
        self.providers = providers
        self.quotes: dict[str, list[LPQuote]] = {}

    def on_quote_update(self, quote: LPQuote):
        """Update quote from specific provider"""
        symbol = quote.symbol
        if symbol not in self.quotes:
            self.quotes[symbol] = []

        # Remove old quote from this provider
        self.quotes[symbol] = [
            q for q in self.quotes[symbol]
            if q.provider != quote.provider
        ]
        self.quotes[symbol].append(quote)

    def get_best_bid_ask(self, symbol: str) -> BestBidAsk:
        quotes = self.quotes.get(symbol, [])
        valid = [q for q in quotes if not q.is_stale()]

        if not valid:
            return None

        best_bid = max(valid, key=lambda q: q.bid)
        best_ask = min(valid, key=lambda q: q.ask)

        return BestBidAsk(
            bid=best_bid.bid,
            bid_size=best_bid.bid_size,
            bid_provider=best_bid.provider,
            ask=best_ask.ask,
            ask_size=best_ask.ask_size,
            ask_provider=best_ask.provider,
            spread_bps=int((best_ask.ask - best_bid.bid) / best_bid.bid * 10000)
        )

Smart Order Routing (SOR)

SOR chooses optimal provider for each specific order:

class SmartOrderRouter:
    def route(self, order: ClientOrder, aggregator: LiquidityAggregator) -> RoutingPlan:
        available = aggregator.get_all_quotes(order.symbol)

        if order.type == 'market':
            return self.route_market(order, available)
        elif order.type == 'limit':
            return self.route_limit(order, available)

    def route_market(self, order: ClientOrder, quotes: list[LPQuote]) -> RoutingPlan:
        """TWAP-style: execute via best provider, split if insufficient volume"""
        remaining = order.quantity
        plan = []

        # Sort by price (best first)
        sorted_quotes = sorted(
            quotes,
            key=lambda q: q.ask if order.side == 'buy' else -q.bid
        )

        for quote in sorted_quotes:
            if remaining <= 0:
                break

            fill_qty = min(remaining, quote.ask_size if order.side == 'buy' else quote.bid_size)
            plan.append(RoutingLeg(
                provider=quote.provider,
                quantity=fill_qty,
                expected_price=quote.ask if order.side == 'buy' else quote.bid
            ))
            remaining -= fill_qty

        if remaining > 0:
            raise InsufficientLiquidity(f"Could not route full order, {remaining} remaining")

        return RoutingPlan(legs=plan, total_quantity=order.quantity)

Failover and Resilience

LPs can go offline, widen spreads, introduce restrictions. Resilient integration:

class ResilientLPManager:
    def __init__(self, providers: list, fallback_amm=None):
        self.providers = {p.name: p for p in providers}
        self.provider_health = {p.name: True for p in providers}
        self.fallback_amm = fallback_amm  # DEX as fallback

    async def get_quote_with_fallback(self, symbol, side, qty) -> LPQuote:
        # Try providers by priority
        for provider_name, provider in self.providers.items():
            if not self.provider_health[provider_name]:
                continue

            try:
                quote = await asyncio.wait_for(
                    provider.get_quote(symbol, side, qty),
                    timeout=2.0  # 2 seconds max
                )
                return quote
            except (asyncio.TimeoutError, LPError) as e:
                logger.warning(f"LP {provider_name} failed: {e}")
                await self.mark_unhealthy(provider_name)

        # All providers unavailable — fallback to AMM
        if self.fallback_amm:
            return await self.fallback_amm.get_quote(symbol, side, qty)

        raise NoLiquidityAvailable("All LP providers failed")

LP integration is not a one-time task, but ongoing engineering: providers update APIs, change terms, experience outages. Monitoring quality of fills (difference between quoted and executed price) shows real liquidity quality.