defmodule CCXT.Multi do @moduledoc """ 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 for efficient parallel fetching - **Configurable timeouts**: Per-exchange timeout with sensible defaults - **Per-exchange symbols**: Map inputs allow different symbols per exchange - **Result helpers**: Easy extraction of successes and failures ## Usage # Uniform symbol across all exchanges result = Multi.fetch_tickers([CCXT.Bybit, CCXT.Binance], "BTC/USDT") # => %{CCXT.Bybit => {:ok, %{...}}, CCXT.Binance => {:ok, %{...}}} # Per-exchange symbols (e.g., Bitfinex uses "tBTCUSD" format) result = Multi.fetch_tickers(%{CCXT.Bybit => "BTC/USDT", CCXT.Bitfinex => "tBTCUSD"}) # => %{CCXT.Bybit => {:ok, %{...}}, CCXT.Bitfinex => {:ok, %{...}}} # Get only successful results tickers = Multi.successes(result) # => %{CCXT.Bybit => %{...}, CCXT.Binance => %{...}} # from_multi/1 is an alias for successes/1 tickers = Multi.from_multi(result) # => %{CCXT.Bybit => %{...}, CCXT.Bitfinex => %{...}} # Check which exchanges failed failures = Multi.failures(result) # => %{} (empty if all succeeded) # Generic parallel call for any function result = Multi.parallel_call(exchanges, :fetch_balance, [credentials], timeout: 15_000) ## Common Pitfalls - **Always filter before consuming**: Multi results contain `{:ok, _}` and `{:error, _}` tuples. Call `successes/1` (or `from_multi/1`) before passing to downstream consumers that expect unwrapped values. - **Use map form for cross-exchange symbols**: Exchanges use different symbol formats for the same instrument. Deribit uses `"BTC-PERPETUAL"` where Bybit uses `"BTC/USDT:USDT"`. The map form `%{CCXT.Deribit => "BTC-PERPETUAL", CCXT.Bybit => "BTC/USDT:USDT"}` handles this. ## Notes - All functions are public endpoint calls (no authentication required) - For authenticated calls, use `parallel_call/4` with credentials as args - Timeout is per-exchange, not total (total time ≈ max(individual timeouts)) """ use Descripex, namespace: "/multi" @default_timeout_ms 10_000 @typedoc "Result map with exchange module as key and {:ok, value} | {:error, reason} as value" @type result(t) :: %{module() => {:ok, t} | {:error, term()}} @typedoc "Map of exchange modules to per-exchange first argument (e.g., symbol)" @type exchange_map :: %{module() => term()} api(:fetch_tickers, "Fetch tickers from multiple exchanges in parallel.", params: [ exchange_or_map: [kind: :value, description: "List of exchange modules or %{module => symbol} map"], symbol: [kind: :value, description: "Unified symbol (list form only)", default: nil], opts: [kind: :value, description: "Options: :timeout (ms, default 10_000)", default: []] ], returns: %{type: :map, description: "%{module => {:ok, ticker} | {:error, reason}}"} ) @doc """ Fetches tickers from multiple exchanges in parallel. Returns partial results - one exchange failing doesn't kill the whole request. Essential for dashboards and price comparison. Accepts either a list of exchanges with a shared symbol, or a map of `%{exchange_module => symbol}` for per-exchange symbol formats. ## Parameters - `exchange_map` - Map of exchange module to symbol (e.g., `%{CCXT.Bybit => "BTC/USDT", CCXT.Bitfinex => "tBTCUSD"}`) - `exchange_modules` - List of exchange modules (e.g., `[CCXT.Bybit, CCXT.Binance]`) - `symbol` - Unified symbol (e.g., `"BTC/USDT"`) — used with list form - `opts` - Options: - `:timeout` - Timeout per exchange in ms (default: #{@default_timeout_ms}) ## Examples iex> result = CCXT.Multi.fetch_tickers(%{CCXT.Bybit => "BTC/USDT"}) iex> is_map(result) true iex> result = CCXT.Multi.fetch_tickers([CCXT.Bybit, CCXT.Binance], "BTC/USDT") iex> is_map(result) true """ @spec fetch_tickers(exchange_map()) :: result(map()) @spec fetch_tickers(exchange_map(), keyword()) :: result(map()) @spec fetch_tickers([module()], String.t()) :: result(map()) @spec fetch_tickers([module()], String.t(), keyword()) :: result(map()) def fetch_tickers(%{} = map), do: fetch_tickers(map, []) @doc "Map form with options, or list+symbol form (defaults opts to `[]`). See `fetch_tickers/1`." def fetch_tickers(%{} = map, opts) when is_list(opts), do: parallel_call(map, :fetch_ticker, [], opts) def fetch_tickers(list, symbol) when is_list(list) and is_binary(symbol), do: fetch_tickers(list, symbol, []) @doc "List form with a shared symbol and options. Accepts `:timeout` in opts. See `fetch_tickers/1`." def fetch_tickers(list, symbol, opts) when is_list(list) and is_binary(symbol) and is_list(opts) do parallel_call(list, :fetch_ticker, [symbol], opts) end api(:fetch_order_books, "Fetch order books from multiple exchanges in parallel.", params: [ exchange_or_map: [kind: :value, description: "List of exchange modules or %{module => symbol} map"], symbol: [kind: :value, description: "Unified symbol (list form only)", default: nil], opts: [kind: :value, description: "Options: :timeout (ms), :limit (depth)", default: []] ], returns: %{type: :map, description: "%{module => {:ok, order_book} | {:error, reason}}"} ) @doc """ Fetches order books from multiple exchanges in parallel. Accepts either a list of exchanges with a shared symbol, or a map of `%{exchange_module => symbol}` for per-exchange symbol formats. ## Parameters - `exchange_map` - Map of exchange module to symbol - `exchange_modules` - List of exchange modules - `symbol` - Unified symbol — used with list form - `opts` - Options: - `:timeout` - Timeout per exchange in ms (default: #{@default_timeout_ms}) - `:limit` - Order book depth limit (passed to exchange) ## Examples iex> result = CCXT.Multi.fetch_order_books(%{CCXT.Bybit => "BTC/USDT"}) iex> is_map(result) true iex> result = CCXT.Multi.fetch_order_books([CCXT.Bybit, CCXT.Binance], "BTC/USDT") iex> is_map(result) true """ @spec fetch_order_books(exchange_map()) :: result(map()) @spec fetch_order_books(exchange_map(), keyword()) :: result(map()) @spec fetch_order_books([module()], String.t()) :: result(map()) @spec fetch_order_books([module()], String.t(), keyword()) :: result(map()) def fetch_order_books(%{} = map), do: fetch_order_books(map, []) @doc "Map form with options, or list+symbol form (defaults opts to `[]`). See `fetch_order_books/1`." def fetch_order_books(%{} = map, opts) when is_list(opts) do {call_opts, fetch_opts} = Keyword.split(opts, [:timeout]) {limit, fetch_opts} = Keyword.pop(fetch_opts, :limit) parallel_call(map, :fetch_order_book, [limit, fetch_opts], call_opts) end def fetch_order_books(list, symbol) when is_list(list) and is_binary(symbol), do: fetch_order_books(list, symbol, []) @doc "List form with a shared symbol and options. Accepts `:timeout` and `:limit` in opts. See `fetch_order_books/1`." def fetch_order_books(list, symbol, opts) when is_list(list) and is_binary(symbol) and is_list(opts) do {call_opts, fetch_opts} = Keyword.split(opts, [:timeout]) {limit, fetch_opts} = Keyword.pop(fetch_opts, :limit) parallel_call(list, :fetch_order_book, [symbol, limit, fetch_opts], call_opts) end api(:fetch_ohlcv, "Fetch OHLCV candles from multiple exchanges in parallel.", params: [ exchange_or_map: [kind: :value, description: "List of exchange modules or %{module => symbol} map"], timeframe: [kind: :value, description: ~s{Candle timeframe (e.g. "1h", "1d")}], opts: [kind: :value, description: "Options: :timeout (ms), :since (ms), :limit, :category", default: []] ], returns: %{type: :map, description: "%{module => {:ok, [candle]} | {:error, reason}}"} ) @doc """ Fetches OHLCV candles from multiple exchanges in parallel. Accepts either a list of exchanges with a shared symbol, or a map of `%{exchange_module => symbol}` for per-exchange symbol formats. When using a map, `timeframe` is a required second argument. ## Parameters - `exchange_map` - Map of exchange module to symbol - `exchange_modules` - List of exchange modules - `symbol` - Unified symbol — used with list form - `timeframe` - Candle timeframe (e.g., `"1h"`, `"1d"`) - `opts` - Options: - `:timeout` - Timeout per exchange in ms (default: #{@default_timeout_ms}) - `:since` - Start time in milliseconds - `:limit` - Number of candles - `:category` - Override inferred market category (e.g., `"linear"`, `"inverse"`, `"spot"`). Inferred automatically from symbol format: `"BTC/USDT:USDT"` → `"linear"`, `"BTC/USD:BTC"` → `"inverse"`, `"BTC/USDT"` → exchange default. ## Examples iex> result = CCXT.Multi.fetch_ohlcv(%{CCXT.Bybit => "BTC/USDT"}, "1h") iex> is_map(result) true iex> result = CCXT.Multi.fetch_ohlcv([CCXT.Bybit, CCXT.Binance], "BTC/USDT", "1h") iex> is_map(result) true """ @spec fetch_ohlcv(exchange_map(), String.t()) :: result(list()) @spec fetch_ohlcv(exchange_map(), String.t(), keyword()) :: result(list()) @spec fetch_ohlcv([module()], String.t(), String.t()) :: result(list()) @spec fetch_ohlcv([module()], String.t(), String.t(), keyword()) :: result(list()) def fetch_ohlcv(%{} = map, timeframe) when is_binary(timeframe), do: fetch_ohlcv(map, timeframe, []) @doc "Map form with timeframe and options, or list+symbol+timeframe form (defaults opts to `[]`). See `fetch_ohlcv/2`." def fetch_ohlcv(%{} = map, timeframe, opts) when is_binary(timeframe) and is_list(opts) do {call_opts, fetch_opts} = Keyword.split(opts, [:timeout]) {explicit_category, fetch_opts} = Keyword.pop(fetch_opts, :category) {since, fetch_opts} = Keyword.pop(fetch_opts, :since) {limit, fetch_opts} = Keyword.pop(fetch_opts, :limit) entries = Map.to_list(map) timeout = Keyword.get(call_opts, :timeout, @default_timeout_ms) # Cannot delegate to parallel_call/4 here — category injection depends on each # exchange's own symbol, so we must compute per-entry opts before the call. entries |> Task.async_stream( fn {module, symbol} -> per_opts = inject_category(symbol, explicit_category, fetch_opts) call_exchange(module, :fetch_ohlcv, [symbol, timeframe, since, limit, per_opts]) end, timeout: timeout, on_timeout: :kill_task, ordered: true ) |> Enum.zip(entries) |> Map.new(fn {result, {module, _}} -> {module, normalize_result(result)} end) end def fetch_ohlcv(list, symbol, timeframe) when is_list(list) and is_binary(symbol) and is_binary(timeframe), do: fetch_ohlcv(list, symbol, timeframe, []) @doc "List form with a shared symbol, timeframe, and options. Accepts `:timeout`, `:since`, `:limit`, `:category` in opts. See `fetch_ohlcv/2`." def fetch_ohlcv(list, symbol, timeframe, opts) when is_list(list) and is_binary(symbol) and is_binary(timeframe) and is_list(opts) do {call_opts, fetch_opts} = Keyword.split(opts, [:timeout]) {explicit_category, fetch_opts} = Keyword.pop(fetch_opts, :category) {since, fetch_opts} = Keyword.pop(fetch_opts, :since) {limit, fetch_opts} = Keyword.pop(fetch_opts, :limit) fetch_opts = inject_category(symbol, explicit_category, fetch_opts) parallel_call(list, :fetch_ohlcv, [symbol, timeframe, since, limit, fetch_opts], call_opts) end api(:fetch_trades, "Fetch recent trades from multiple exchanges in parallel.", params: [ exchange_or_map: [kind: :value, description: "List of exchange modules or %{module => symbol} map"], symbol: [kind: :value, description: "Unified symbol (list form only)", default: nil], opts: [kind: :value, description: "Options: :timeout (ms), :since (ms), :limit", default: []] ], returns: %{type: :map, description: "%{module => {:ok, [trade]} | {:error, reason}}"} ) @doc """ Fetches recent trades from multiple exchanges in parallel. Accepts either a list of exchanges with a shared symbol, or a map of `%{exchange_module => symbol}` for per-exchange symbol formats. ## Parameters - `exchange_map` - Map of exchange module to symbol - `exchange_modules` - List of exchange modules - `symbol` - Unified symbol — used with list form - `opts` - Options: - `:timeout` - Timeout per exchange in ms (default: #{@default_timeout_ms}) - `:since` - Start time in milliseconds - `:limit` - Number of trades to fetch ## Examples iex> result = CCXT.Multi.fetch_trades(%{CCXT.Bybit => "BTC/USDT"}) iex> is_map(result) true iex> result = CCXT.Multi.fetch_trades([CCXT.Bybit, CCXT.Binance], "BTC/USDT") iex> is_map(result) true """ @spec fetch_trades(exchange_map()) :: result(list()) @spec fetch_trades(exchange_map(), keyword()) :: result(list()) @spec fetch_trades([module()], String.t()) :: result(list()) @spec fetch_trades([module()], String.t(), keyword()) :: result(list()) def fetch_trades(%{} = map), do: fetch_trades(map, []) @doc "Map form with options, or list+symbol form (defaults opts to `[]`). See `fetch_trades/1`." def fetch_trades(%{} = map, opts) when is_list(opts) do {call_opts, fetch_opts} = Keyword.split(opts, [:timeout]) {since, fetch_opts} = Keyword.pop(fetch_opts, :since) {limit, fetch_opts} = Keyword.pop(fetch_opts, :limit) parallel_call(map, :fetch_trades, [since, limit, fetch_opts], call_opts) end def fetch_trades(list, symbol) when is_list(list) and is_binary(symbol), do: fetch_trades(list, symbol, []) @doc "List form with a shared symbol and options. Accepts `:timeout`, `:since`, `:limit` in opts. See `fetch_trades/1`." def fetch_trades(list, symbol, opts) when is_list(list) and is_binary(symbol) and is_list(opts) do {call_opts, fetch_opts} = Keyword.split(opts, [:timeout]) {since, fetch_opts} = Keyword.pop(fetch_opts, :since) {limit, fetch_opts} = Keyword.pop(fetch_opts, :limit) parallel_call(list, :fetch_trades, [symbol, since, limit, fetch_opts], call_opts) end api(:parallel_call, "Call any function on multiple exchanges in parallel.", params: [ exchange_or_map: [kind: :value, description: "List of exchange modules or %{module => first_arg} map"], function_name: [kind: :value, description: "Function atom (e.g. :fetch_ticker, :fetch_balance)"], args: [ kind: :value, description: "Arguments list (list form: full args; map form: shared args after per-exchange value)" ], opts: [kind: :value, description: "Options: :timeout (ms, default 10_000)", default: []] ], returns: %{type: :map, description: "%{module => {:ok, value} | {:error, reason}}"} ) @doc """ Generic parallel call - call any function on multiple exchanges. Accepts either a list of exchange modules with shared args, or a map of `%{exchange_module => value}` where `value` is prepended as the first argument to each exchange call (enabling per-exchange symbols or credentials). This is the core function that other specialized functions use. Can be used for any exchange method. ## Parameters - `exchange_map` - Map of `%{module => first_arg}`. The map value is prepended to `shared_args` for each exchange call. - `exchange_modules` - List of exchange modules - `function_name` - Function atom (e.g., `:fetch_ticker`, `:fetch_balance`) - `args` / `shared_args` - Arguments to pass (list form: full args; map form: shared args appended after the per-exchange value) - `opts` - Options: - `:timeout` - Timeout per exchange in ms (default: #{@default_timeout_ms}) ## Examples # List form: same args for all exchanges iex> result = CCXT.Multi.parallel_call([CCXT.Bybit], :fetch_ticker, ["BTC/USDT"]) iex> is_map(result) true # Map form: per-exchange first arg (e.g., different symbol formats) iex> result = CCXT.Multi.parallel_call(%{CCXT.Bybit => "BTC/USDT"}, :fetch_ticker, []) iex> is_map(result) true """ @spec parallel_call(exchange_map(), atom(), [term()], keyword()) :: result(term()) @spec parallel_call([module()], atom(), [term()], keyword()) :: result(term()) def parallel_call(exchange_or_map, function_name, args, opts \\ []) def parallel_call(%{} = exchange_args_map, function_name, shared_args, opts) when is_list(shared_args) and is_list(opts) do entries = Map.to_list(exchange_args_map) timeout = Keyword.get(opts, :timeout, @default_timeout_ms) entries |> Task.async_stream( fn {module, value} -> call_exchange(module, function_name, [value | shared_args]) end, timeout: timeout, on_timeout: :kill_task, ordered: true ) |> Enum.zip(entries) |> Map.new(fn {result, {module, _}} -> {module, normalize_result(result)} end) end def parallel_call([], _function_name, _args, _opts), do: %{} def parallel_call(exchange_modules, function_name, args, opts) when is_list(exchange_modules) and is_list(args) and is_list(opts) do timeout = Keyword.get(opts, :timeout, @default_timeout_ms) exchange_modules |> Task.async_stream( fn module -> call_exchange(module, function_name, args) end, timeout: timeout, on_timeout: :kill_task, ordered: true ) |> Enum.zip(exchange_modules) |> Map.new(fn {result, module} -> {module, normalize_result(result)} end) end api(:successes, "Extract only successful results, discarding errors.", params: [ results: [kind: :value, description: "Multi result map from fetch_*/parallel_call"] ], returns: %{type: :map, description: "%{module => unwrapped_value} with errors removed"} ) @doc """ Returns only successful results, discarding errors. Unwraps `{:ok, value}` tuples to just values. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{price: 100}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.successes(results) %{CCXT.Bybit => %{price: 100}} """ @spec successes(result(t)) :: %{module() => t} when t: var def successes(results) when is_map(results) do results |> Enum.filter(fn {_module, result} -> match?({:ok, _}, result) end) |> Map.new(fn {module, {:ok, value}} -> {module, value} end) end api(:from_multi, "Convenience alias for successes/1.", params: [ results: [kind: :value, description: "Multi result map from fetch_*/parallel_call"] ], returns: %{type: :map, description: "%{module => unwrapped_value} with errors removed"}, composes_with: [:successes] ) @doc """ Convenience alias for `successes/1`. Returns only successful results, discarding errors. Unwraps `{:ok, value}` tuples to just values. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{price: 100}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.from_multi(results) %{CCXT.Bybit => %{price: 100}} """ @spec from_multi(result(t)) :: %{module() => t} when t: var def from_multi(results) when is_map(results), do: successes(results) api(:failures, "Extract only failed results, discarding successes.", params: [ results: [kind: :value, description: "Multi result map from fetch_*/parallel_call"] ], returns: %{type: :map, description: "%{module => error_reason} with successes removed"} ) @doc """ Returns only failed results. Unwraps `{:error, reason}` tuples to just reasons. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{price: 100}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.failures(results) %{CCXT.Binance => :timeout} """ @spec failures(result(term())) :: %{module() => term()} def failures(results) when is_map(results) do results |> Enum.filter(fn {_module, result} -> match?({:error, _}, result) end) |> Map.new(fn {module, {:error, reason}} -> {module, reason} end) end api(:success_count, "Count successful results.", params: [results: [kind: :value, description: "Multi result map"]], returns: %{type: :integer, description: "Number of {:ok, _} entries"} ) @doc """ Returns the count of successful results. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.success_count(results) 1 """ @spec success_count(result(term())) :: non_neg_integer() def success_count(results) when is_map(results) do Enum.count(results, fn {_module, result} -> match?({:ok, _}, result) end) end api(:failure_count, "Count failed results.", params: [results: [kind: :value, description: "Multi result map"]], returns: %{type: :integer, description: "Number of {:error, _} entries"} ) @doc """ Returns the count of failed results. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.failure_count(results) 1 """ @spec failure_count(result(term())) :: non_neg_integer() def failure_count(results) when is_map(results) do Enum.count(results, fn {_module, result} -> match?({:error, _}, result) end) end api(:all_succeeded?, "Check if all calls succeeded.", params: [results: [kind: :value, description: "Multi result map"]], returns: %{type: :boolean, description: "true if every entry is {:ok, _}"} ) @doc """ Checks if all calls succeeded. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{}}, CCXT.Binance => {:ok, %{}}} iex> CCXT.Multi.all_succeeded?(results) true iex> results = %{CCXT.Bybit => {:ok, %{}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.all_succeeded?(results) false """ @spec all_succeeded?(result(term())) :: boolean() def all_succeeded?(results) when is_map(results) do Enum.all?(results, fn {_module, result} -> match?({:ok, _}, result) end) end api(:any_succeeded?, "Check if any call succeeded.", params: [results: [kind: :value, description: "Multi result map"]], returns: %{type: :boolean, description: "true if at least one entry is {:ok, _}"} ) @doc """ Checks if any call succeeded. ## Examples iex> results = %{CCXT.Bybit => {:ok, %{}}, CCXT.Binance => {:error, :timeout}} iex> CCXT.Multi.any_succeeded?(results) true iex> results = %{CCXT.Bybit => {:error, :a}, CCXT.Binance => {:error, :b}} iex> CCXT.Multi.any_succeeded?(results) false """ @spec any_succeeded?(result(term())) :: boolean() def any_succeeded?(results) when is_map(results) do Enum.any?(results, fn {_module, result} -> match?({:ok, _}, result) end) end # =========================================================================== # Private Helpers # =========================================================================== @doc false # Calls the exchange function and wraps result in {:ok, _} | {:error, _} @spec call_exchange(module(), atom(), [term()]) :: {:ok, term()} | {:error, term()} defp call_exchange(module, function_name, args) do Code.ensure_loaded!(module) if function_exported?(module, function_name, length(args)) do apply(module, function_name, args) else {:error, {:function_not_exported, {module, function_name, length(args)}}} end rescue e -> {:error, {:exception, Exception.message(e)}} catch kind, reason -> {:error, {kind, reason}} end @doc false # Infers OHLCV category from symbol format and injects into opts[:params]. # explicit_category overrides inference when provided. defp inject_category(symbol, nil, fetch_opts) do case infer_category_from_symbol(symbol) do nil -> fetch_opts cat -> put_category_param(fetch_opts, cat) end end defp inject_category(_symbol, category, fetch_opts) do put_category_param(fetch_opts, category) end @doc false # Returns "linear", "inverse", or nil from CCXT unified symbol format. # Only infers for simple perpetual swap format (no dash in settle). # Skips options/futures (e.g. "BTC/USDT:USDT-240329-C") — settle has a dash suffix. # linear: settle == quote, no dash (e.g. "BTC/USDT:USDT") # inverse: settle != quote, no dash (e.g. "BTC/USD:BTC") # spot/unknown: nil — use exchange default defp infer_category_from_symbol(symbol) do case CCXT.Symbol.parse(symbol) do {:ok, %{settle: nil}} -> nil {:ok, %{settle: settle, quote: quote}} when settle == quote -> "linear" {:ok, %{settle: settle}} -> # Skip if settle contains "-" — indicates option/dated future (e.g. "BTC-240329") # Those need explicit :category override via escape hatch if String.contains?(settle, "-"), do: nil, else: "inverse" {:error, _} -> nil end end @doc false # Merges category into opts[:params], normalizing all keys to strings first. # opts[:params] can be a keyword list, list of 2-tuples, or map — all handled. # Key normalization prevents atom :category and string "category" coexisting after # apply_endpoint_mappings converts atoms to strings (helpers.ex:56). Without it, # "category" override could be nondeterministic depending on map iteration order. defp put_category_param(fetch_opts, value) do params = fetch_opts |> Keyword.get(:params, %{}) |> Map.new(fn {k, v} -> {to_string(k), v} end) |> Map.put("category", value) Keyword.put(fetch_opts, :params, params) end @doc false # Normalizes Task.async_stream result to {:ok, _} | {:error, _}. # Functions returning {:ok, value} or {:error, reason} are passed through. # Functions returning raw values (not tuples) are auto-wrapped in {:ok, value}. @spec normalize_result({:ok, term()} | {:exit, term()}) :: {:ok, term()} | {:error, term()} defp normalize_result({:ok, {:ok, value}}), do: {:ok, value} defp normalize_result({:ok, {:error, reason}}), do: {:error, reason} defp normalize_result({:ok, other}), do: {:ok, other} defp normalize_result({:exit, :timeout}), do: {:error, :timeout} defp normalize_result({:exit, reason}), do: {:error, {:exit, reason}} end