Parallel fetch operations across multiple exchanges.
Enables fetching data from multiple exchanges concurrently with graceful handling of partial failures. Essential for dashboards, price comparison, and arbitrage detection.
Key Features
- Partial failure handling: one exchange failing doesn't kill the whole request
- Concurrent execution: uses
Task.async_stream/3for efficient parallel fetching - Configurable timeouts: per-exchange timeout with a sensible default
- Per-exchange symbols: map inputs allow different symbols per exchange
- Result helpers: easy extraction of successes and failures
Exchange struct, not module atoms
Unlike the original ccxt_client_bak port (which dispatched on generated
exchange modules like Bourse.Bybit), this version operates on %Bourse.Exchange{}
structs built via Bourse.exchange/2. Each parallel call routes through the
unified API (Bourse.fetch_ticker/3, etc.) with the exchange struct as the first
argument, and result maps are keyed by the %Bourse.Exchange{} struct.
Usage
{:ok, bybit} = Bourse.exchange("bybit")
{:ok, binance} = Bourse.exchange("binance")
# Uniform symbol across all exchanges
result = Bourse.Multi.fetch_tickers([bybit, binance], "BTC/USDT")
# => %{bybit => {:ok, %{...}}, binance => {:ok, %{...}}}
# Per-exchange symbols (exchanges use different symbol formats)
result = Bourse.Multi.fetch_tickers(%{bybit => "BTC/USDT:USDT", deribit => "BTC-PERPETUAL"})
# Get only successful results (unwrapped)
tickers = Bourse.Multi.successes(result)
# => %{bybit => %{...}, binance => %{...}}
# Check which exchanges failed (unwrapped reasons)
failures = Bourse.Multi.failures(result)
# => %{} (empty if all succeeded)
# Generic parallel call for any unified function
result = Bourse.Multi.parallel_call([bybit, binance], :fetch_balance, [], timeout: 15_000)Common Pitfalls
- Always filter before consuming: Multi results mix
{:ok, _}and{:error, _}tuples. Callsuccesses/1before passing to downstream consumers that expect unwrapped values. - Use map form for cross-exchange symbols: the same instrument has different
symbol formats per exchange. The map form
%{deribit => "BTC-PERPETUAL", bybit => "BTC/USDT:USDT"}handles this.
Notes
fetch_tickers/2,3andfetch_order_books/2,3are public endpoint calls (no authentication required).- For authenticated calls, use
parallel_call/4against exchanges built with credentials. - Timeout is per-exchange, not total (total time ≈
max(individual timeouts)).
Summary
Types
Map of exchange structs to a per-exchange first argument (e.g., symbol)
Result map keyed by exchange struct, with {:ok, value} | {:error, reason} values
Functions
Returns only failed results, discarding successes.
Fetches order books from multiple exchanges in parallel.
Map form with options, or list+symbol form (defaults opts to []). See fetch_order_books/1.
List form with a shared symbol and options. Accepts :timeout and :limit in opts. See fetch_order_books/1.
Fetches tickers from multiple exchanges in parallel.
Map form with options, or list+symbol form (defaults opts to []). See fetch_tickers/1.
List form with a shared symbol and options. Accepts :timeout in opts. See fetch_tickers/1.
Generic parallel call — invokes any unified Bourse function on multiple exchanges.
Returns only successful results, discarding errors.
Types
@type exchange_map() :: %{required(Bourse.Exchange.t()) => term()}
Map of exchange structs to a per-exchange first argument (e.g., symbol)
@type result(t) :: %{required(Bourse.Exchange.t()) => {:ok, t} | {:error, term()}}
Result map keyed by exchange struct, with {:ok, value} | {:error, reason} values
Functions
@spec failures(result(term())) :: %{required(Bourse.Exchange.t()) => term()}
Returns only failed results, discarding successes.
Unwraps {:error, reason} tuples to just reasons.
Examples
iex> results = %{a: {:ok, %{price: 100}}, b: {:error, :timeout}}
iex> Bourse.Multi.failures(results)
%{b: :timeout}@spec fetch_order_books(exchange_map()) :: result(map())
Fetches order books from multiple exchanges in parallel.
Accepts either a list of exchanges with a shared symbol, or a map of
%{exchange => symbol} for per-exchange symbol formats.
Parameters
exchange_map—%{%Exchange{} => symbol}exchanges— list of%Exchange{}symbol— unified symbol, used with list formopts::timeout— per-exchange timeout in ms (default:10000):limit— order book depth limit (forwarded toBourse.fetch_order_book/3)- any other option is forwarded to
Bourse.fetch_order_book/3
Examples
iex> Bourse.Multi.fetch_order_books([], "BTC/USDT")
%{}
iex> Bourse.Multi.fetch_order_books(%{})
%{}@spec fetch_order_books( exchange_map(), keyword() ) :: result(map())
@spec fetch_order_books([Bourse.Exchange.t()], String.t()) :: result(map())
Map form with options, or list+symbol form (defaults opts to []). See fetch_order_books/1.
@spec fetch_order_books([Bourse.Exchange.t()], String.t(), keyword()) :: result(map())
List form with a shared symbol and options. Accepts :timeout and :limit in opts. See fetch_order_books/1.
@spec fetch_tickers(exchange_map()) :: result(map())
Fetches tickers from multiple exchanges in parallel.
Returns partial results — one exchange failing doesn't kill the whole request.
Accepts either a list of exchanges with a shared symbol, or a map of
%{exchange => symbol} for per-exchange symbol formats.
Parameters
exchange_map—%{%Exchange{} => symbol}(per-exchange symbols)exchanges— list of%Exchange{}(shared symbol)symbol— unified symbol (e.g.,"BTC/USDT"), used with list formopts::timeout— per-exchange timeout in ms (default:10000)- any other option is forwarded to
Bourse.fetch_ticker/3
Examples
iex> Bourse.Multi.fetch_tickers([], "BTC/USDT")
%{}
iex> Bourse.Multi.fetch_tickers(%{})
%{}@spec fetch_tickers( exchange_map(), keyword() ) :: result(map())
@spec fetch_tickers([Bourse.Exchange.t()], String.t()) :: result(map())
Map form with options, or list+symbol form (defaults opts to []). See fetch_tickers/1.
@spec fetch_tickers([Bourse.Exchange.t()], String.t(), keyword()) :: result(map())
List form with a shared symbol and options. Accepts :timeout in opts. See fetch_tickers/1.
@spec parallel_call( exchange_map() | [Bourse.Exchange.t()], atom(), [term()], keyword() ) :: result(term())
Generic parallel call — invokes any unified Bourse function on multiple exchanges.
Accepts either a list of %Exchange{} structs with shared args, or a map of
%{exchange => value} where value is prepended as the first argument (after
the exchange) to each call — enabling per-exchange symbols or other values.
Each call routes through Bourse.<function_name>(exchange, args...). This is the
core function the specialized helpers build on.
Parameters
exchange_or_map— list of%Exchange{}, or%{%Exchange{} => first_arg}function_name— unified function atom (e.g.,:fetch_ticker,:fetch_balance)args— argument list. List form: full args after the exchange. Map form: shared args appended after the per-exchange value.opts::timeout— per-exchange timeout in ms (default:10000)
Examples
iex> Bourse.Multi.parallel_call([], :fetch_ticker, ["BTC/USDT"])
%{}
iex> Bourse.Multi.parallel_call(%{}, :fetch_ticker, [])
%{}@spec successes(result(t)) :: %{required(Bourse.Exchange.t()) => t} when t: var
Returns only successful results, discarding errors.
Unwraps {:ok, value} tuples to just values. The result map is keyed the same
way as the input (by %Exchange{} for fetch_*/parallel_call output).
Examples
iex> results = %{a: {:ok, %{price: 100}}, b: {:error, :timeout}}
iex> Bourse.Multi.successes(results)
%{a: %{price: 100}}