diff --git a/lib/sanbase/external_services/coinmarketcap/historical_fetcher.ex b/lib/sanbase/external_services/coinmarketcap/historical_fetcher.ex new file mode 100644 index 0000000000..2c2821025e --- /dev/null +++ b/lib/sanbase/external_services/coinmarketcap/historical_fetcher.ex @@ -0,0 +1,97 @@ +defmodule Sanbase.ExternalServices.Coinmarketcap.HistoricalFetcher do + @moduledoc ~s""" + Fetches historical CMC ticker data for a range of timestamps. + + Given from/to datetimes and an interval (in seconds), fires one + TickerFetcher.work/1 request per timestamp in the range. + + ## Usage + + # Backfill the last 15 hours with 5-minute intervals: + from = DateTime.utc_now() |> DateTime.add(-15 * 3600) + to = DateTime.utc_now() + Sanbase.ExternalServices.Coinmarketcap.HistoricalFetcher.run(from, to, 300) + + # With custom options (projects_number, sleep between requests): + Sanbase.ExternalServices.Coinmarketcap.HistoricalFetcher.run( + from, to, 300, + projects_number: 5000, + sleep_between_requests_ms: 3000 + ) + """ + + require Logger + + alias Sanbase.ExternalServices.Coinmarketcap.TickerFetcher + + @default_sleep_ms 2_000 + + @doc """ + Fetches historical data for each timestamp in the range [from, to] + spaced `interval_seconds` apart. + + Options: + - `:projects_number` - number of top projects to fetch (default from config) + - `:sleep_between_requests_ms` - milliseconds to sleep between requests (default #{@default_sleep_ms}) + """ + @spec run(DateTime.t(), DateTime.t(), pos_integer(), keyword()) :: + {:ok, non_neg_integer()} | {:error, String.t()} + def run(%DateTime{} = from, %DateTime{} = to, interval_seconds, opts \\ []) + when is_integer(interval_seconds) and interval_seconds > 0 do + if DateTime.compare(from, to) == :gt do + {:error, "`from` datetime must be before `to` datetime"} + else + timestamps = generate_timestamps(from, to, interval_seconds) + sleep_ms = Keyword.get(opts, :sleep_between_requests_ms, @default_sleep_ms) + work_opts = Keyword.drop(opts, [:sleep_between_requests_ms]) + + Logger.info( + "[CMC Historical] Starting backfill from #{DateTime.to_iso8601(from)} " <> + "to #{DateTime.to_iso8601(to)} with #{interval_seconds}s interval " <> + "(#{length(timestamps)} requests)" + ) + + total = length(timestamps) + + {successes, failures} = + timestamps + |> Enum.with_index(1) + |> Enum.reduce({0, 0}, fn {datetime, index}, {ok_count, err_count} -> + Logger.info( + "[CMC Historical] Request #{index}/#{total}: #{DateTime.to_iso8601(datetime)}" + ) + + result = + try do + TickerFetcher.work(Keyword.put(work_opts, :datetime, datetime)) + rescue + e -> + Logger.error( + "[CMC Historical] Exception for #{DateTime.to_iso8601(datetime)}: #{Exception.message(e)}" + ) + + {:error, :exception} + end + + if total > 1, do: Process.sleep(sleep_ms) + + case result do + :ok -> {ok_count + 1, err_count} + {:error, _} -> {ok_count, err_count + 1} + end + end) + + Logger.info( + "[CMC Historical] Backfill complete. " <> + "Successful: #{successes}, Failed: #{failures}" + ) + + {:ok, successes} + end + end + + defp generate_timestamps(from, to, interval_seconds) do + Stream.iterate(from, fn dt -> DateTime.add(dt, interval_seconds, :second) end) + |> Enum.take_while(fn dt -> DateTime.compare(dt, to) != :gt end) + end +end diff --git a/lib/sanbase/external_services/coinmarketcap/ticker.ex b/lib/sanbase/external_services/coinmarketcap/ticker.ex index 129a8ef2b1..2b01b0db92 100644 --- a/lib/sanbase/external_services/coinmarketcap/ticker.ex +++ b/lib/sanbase/external_services/coinmarketcap/ticker.ex @@ -76,20 +76,41 @@ defmodule Sanbase.ExternalServices.Coinmarketcap.Ticker do |> String.to_integer() end - Logger.info("[CMC] Fetching the realtime data for top #{projects_number} projects") + datetime = Keyword.get(opts, :datetime) - "v1/cryptocurrency/listings/latest?start=1&sort=market_cap&limit=#{projects_number}&cryptocurrency_type=all&convert=USD,BTC" + base_params = + "start=1&sort=market_cap&limit=#{projects_number}&cryptocurrency_type=all&convert=USD,BTC" + + {url, label} = + case datetime do + %DateTime{} -> + date_str = + datetime + |> DateTime.truncate(:second) + |> DateTime.to_iso8601() + + {"v1/cryptocurrency/listings/historical?date=#{date_str}&#{base_params}", + "historical (#{date_str})"} + + _ -> + {"v1/cryptocurrency/listings/latest?#{base_params}", "realtime"} + end + + Logger.info("[CMC] Fetching #{label} data for top #{projects_number} projects") + + url |> get() |> case do {:ok, %Tesla.Env{status: 200, body: body}} -> Logger.info( - "[CMC] Successfully fetched the realtime data for top #{projects_number} projects." + "[CMC] Successfully fetched #{label} data for top #{projects_number} projects." ) {:ok, parse_json(body)} - {:ok, %Tesla.Env{status: status}} -> - error = "Failed fetching top #{projects_number} projects' information. Status: #{status}" + {:ok, %Tesla.Env{status: status, body: body}} -> + error = + "Failed fetching top #{projects_number} projects' information. Status: #{status}. Body: #{inspect(body)}" Logger.warning(error) {:error, error} diff --git a/lib/sanbase/external_services/coinmarketcap/ticker_fetcher.ex b/lib/sanbase/external_services/coinmarketcap/ticker_fetcher.ex index 6c9ff149e5..5e49de5ca5 100644 --- a/lib/sanbase/external_services/coinmarketcap/ticker_fetcher.ex +++ b/lib/sanbase/external_services/coinmarketcap/ticker_fetcher.ex @@ -91,59 +91,99 @@ defmodule Sanbase.ExternalServices.Coinmarketcap.TickerFetcher do savings-crvusd ] def work(opts \\ []) do - Logger.info("[CMC] Fetching realtime data from coinmarketcap") - # Fetch current coinmarketcap data for many tickers - # It fetches data for the first N projects, where N is specified in - # the COINMARKETCAP_API_PROJECTS_NUMBER env var + datetime = Keyword.get(opts, :datetime) + historical? = match?(%DateTime{}, datetime) + + if historical? do + Logger.info( + "[CMC] Fetching historical data from coinmarketcap for #{DateTime.to_iso8601(datetime)}" + ) + else + Logger.info("[CMC] Fetching realtime data from coinmarketcap") + end + + # Fetch coinmarketcap data for many tickers. + # For historical mode, fail early if the API returns an error. + # For realtime mode, continue with empty list (original behavior). tickers = case Ticker.fetch_data(opts) do - {:ok, tickers} -> tickers - _ -> [] - end - - fetched_slugs = MapSet.new(tickers, & &1.slug) + {:ok, tickers} -> + tickers - # Handle separately tokens that might be out of top N. - custom_cmc_slugs = @custom_cmc_slugs |> Enum.reject(&(&1 in fetched_slugs)) + {:error, error} when historical? -> + {:error, error} - # Do not break when some of the handpicked assets is no longer supported. - # On 03.09.2025 we had an issue where wrapped-fantom started causing HTTP 400 - # which in turn broke everything below {:ok, _} = fetch_data_by_slug - # and made the exporter fail and did not export the already fetched tickers - custom_tickers = - case Ticker.fetch_data_by_slug(custom_cmc_slugs) do - {:ok, custom_tickers} -> custom_tickers - _ -> [] + _ -> + [] end - tickers = tickers ++ custom_tickers + # Early return on historical fetch failure + if match?({:error, _}, tickers) do + tickers + else + # Custom slugs are only fetched for realtime data. + # The listings/historical endpoint covers the top N projects; + # quotes/historical has a different response format and is not used here. + tickers = + if historical? do + tickers + else + fetched_slugs = MapSet.new(tickers, & &1.slug) + + # Handle separately tokens that might be out of top N. + custom_cmc_slugs = @custom_cmc_slugs |> Enum.reject(&(&1 in fetched_slugs)) + + # Do not break when some of the handpicked assets is no longer supported. + # On 03.09.2025 we had an issue where wrapped-fantom started causing HTTP 400 + # which in turn broke everything below {:ok, _} = fetch_data_by_slug + # and made the exporter fail and did not export the already fetched tickers + custom_tickers = + case Ticker.fetch_data_by_slug(custom_cmc_slugs) do + {:ok, custom_tickers} -> custom_tickers + _ -> [] + end - # Create a map where the coinmarketcap_id is key and the values is the list of - # santiment slugs that have that coinmarketcap_id - cmc_id_to_slugs_mapping = coinmarketcap_to_santiment_slug_map() + tickers ++ custom_tickers + end - tickers = - remove_not_valid_prices(tickers, cmc_id_to_slugs_mapping) + # Create a map where the coinmarketcap_id is key and the values is the list of + # santiment slugs that have that coinmarketcap_id + cmc_id_to_slugs_mapping = coinmarketcap_to_santiment_slug_map() - # Create a project if it's a new one in the top projects and we don't have it - if System.get_env("INSERT_CMC_TOP_N_PROJECTS_INTO_DB") == "1" do - tickers - |> Enum.sort_by(& &1.rank, :asc) - |> Enum.take(top_projects_to_follow()) - |> Enum.each(&insert_or_update_project/1) - end + tickers = + remove_not_valid_prices(tickers, cmc_id_to_slugs_mapping) - # Store the data in LatestCoinmarketcapData in postgres + # Create a project if it's a new one in the top projects and we don't have it. + # Skip for historical data. + if not historical? and System.get_env("INSERT_CMC_TOP_N_PROJECTS_INTO_DB") == "1" do + tickers + |> Enum.sort_by(& &1.rank, :asc) + |> Enum.take(top_projects_to_follow()) + |> Enum.each(&insert_or_update_project/1) + end - tickers - |> Enum.each(&store_latest_coinmarketcap_data!/1) + # Store the data in LatestCoinmarketcapData in postgres. + # Skip for historical data to avoid overwriting latest values. + if not historical? do + tickers + |> Enum.each(&store_latest_coinmarketcap_data!/1) + end - tickers - |> export_to_kafka(cmc_id_to_slugs_mapping) + tickers + |> export_to_kafka(cmc_id_to_slugs_mapping) + + if historical? do + Logger.info( + "[CMC] Fetching historical data from coinmarketcap for #{DateTime.to_iso8601(datetime)} done. #{length(tickers)} tickers exported." + ) + else + Logger.info( + "[CMC] Fetching realtime data from coinmarketcap done. The data is imported in the database." + ) + end - Logger.info( - "[CMC] Fetching realtime data from coinmarketcap done. The data is imported in the database." - ) + :ok + end end defp coinmarketcap_to_santiment_slug_map() do