cmc historical job

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

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}

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