ccxt_client

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