Pr 727 create oracle services data agregator

config/config.exs

@@ -134,7 +134,11 @@ config :archethic, Archethic.OracleChain,
   ]
 
 config :archethic, Archethic.OracleChain.Services.UCOPrice,
-  provider: Archethic.OracleChain.Services.UCOPrice.Providers.Coingecko
+  providers: [
+    Archethic.OracleChain.Services.UCOPrice.Providers.Coingecko,
+    Archethic.OracleChain.Services.UCOPrice.Providers.CoinMarketCap,
+    Archethic.OracleChain.Services.UCOPrice.Providers.CoinPaprika
+  ]
 
 config :archethic, ArchethicWeb.FaucetController,
   seed:

config/test.exs

@@ -88,7 +88,8 @@ config :archethic, Archethic.OracleChain.Scheduler,
   polling_interval: "0 0 * * * *",
   summary_interval: "0 0 0 * * *"
 
-config :archethic, Archethic.OracleChain.Services.UCOPrice, provider: MockUCOPriceProvider
+config :archethic, Archethic.OracleChain.Services.UCOPrice,
+  providers: [MockUCOPriceProvider1, MockUCOPriceProvider2, MockUCOPriceProvider3]
 
 # -----Start-of-Networking-tests-configs-----
 

lib/archethic/oracle_chain/services/uco_price.ex

@@ -6,27 +6,88 @@ defmodule Archethic.OracleChain.Services.UCOPrice do
   require Logger
 
   alias Archethic.OracleChain.Services.Impl
+  alias Archethic.Utils
 
   @behaviour Impl
 
   @pairs ["usd", "eur"]
 
   @impl Impl
   @spec fetch() :: {:ok, %{required(String.t()) => any()}} | {:error, any()}
-  def fetch, do: provider().fetch(@pairs)
+  def fetch do
+    ## Start a supervisor for the feching tasks
+    {:ok, fetching_tasks_supervisor} = Task.Supervisor.start_link()
+    ## retrieve prices from configured providers and filter results marked as errors
+    prices =
+      Task.Supervisor.async_stream_nolink(
+        fetching_tasks_supervisor,
+        providers(),
+        fn provider ->
+          case provider.fetch(@pairs) do
+            {:ok, _prices} = result ->
+              result
+
+            {:error, reason} ->
+              provider_name = provider |> to_string() |> String.split(".") |> List.last()
+
+              Logger.warning(
+                "Service UCOPrice cannot fetch values from " <>
+                  "provider: #{inspect(provider_name)} with reason : #{inspect(reason)}."
+              )
+
+              {:error, provider}
+          end
+        end,
+        on_timeout: :kill_task
+      )
+      |> Stream.filter(&match?({:ok, {:ok, _}}, &1))
+      |> Stream.map(fn
+        {_, {_, result = %{}}} ->
+          result
+      end)
+      ## Here stream looks like : [%{"eur"=>[0.44], "usd"=[0.32]}, ..., %{"eur"=>[0.42, 0.43], "usd"=[0.35]}]
+      |> Enum.reduce(%{}, &agregate_providers_data/2)
+      |> Enum.reduce(%{}, fn {currency, values}, acc ->
+        Map.put(acc, currency, Utils.median(values))
+      end)
+
+    ## split prices in a list per currency. If a service returned a list of prices of a currency,
+    ## they will be medianed first before being added to list
+    #      |> split_prices()
+    ## compute median per currency list
+    #      |> median_prices()
+
+    Supervisor.stop(fetching_tasks_supervisor, :normal, 3_000)
+    {:ok, prices}
+  end
+
+  @spec agregate_providers_data(map(), map()) :: map()
+  defp agregate_providers_data(provider_results, acc) do
+    provider_results
+    |> Enum.reduce(acc, fn
+      {currency, values}, acc when values != [] ->
+        Map.update(acc, String.downcase(currency), values, fn
+          previous_values ->
+            previous_values ++ values
+        end)
+
+      {_currency, _values}, acc ->
+        acc
+    end)
+  end
 
   @impl Impl
   @spec verify?(%{required(String.t()) => any()}) :: boolean
   def verify?(prices_prior = %{}) do
-    case provider().fetch(@pairs) do
+    case fetch() do
+      {:ok, prices_now} when prices_now == %{} ->
+        Logger.error("Cannot fetch UCO price - reason: no data from any service.")
+        false
+
       {:ok, prices_now} ->
         Enum.all?(@pairs, fn pair ->
           compare_price(Map.fetch!(prices_prior, pair), Map.fetch!(prices_now, pair))
         end)
-
-      {:error, reason} ->
-        Logger.warning("Cannot fetch UCO price - reason: #{inspect(reason)}")
-        false
     end
   end
 
@@ -79,7 +140,7 @@ defmodule Archethic.OracleChain.Services.UCOPrice do
 
   def parse_data(_), do: {:error, :invalid_data}
 
-  defp provider do
-    Application.get_env(:archethic, __MODULE__) |> Keyword.fetch!(:provider)
+  defp providers do
+    Application.get_env(:archethic, __MODULE__) |> Keyword.fetch!(:providers)
   end
 end

lib/archethic/oracle_chain/services/uco_price/providers/coingecko.ex → lib/archethic/oracle_chain/services/uco_price/providers/coin_gecko.ex

@@ -34,7 +34,12 @@ defmodule Archethic.OracleChain.Services.UCOPrice.Providers.Coingecko do
            :httpc.request(:get, {query, []}, httpc_options, []),
          {:ok, payload} <- Jason.decode(body),
          {:ok, prices} <- Map.fetch(payload, "archethic") do
-      {:ok, prices}
+      formatted_prices =
+        prices
+        |> Enum.map(fn {pair, price} -> {pair, [price]} end)
+        |> Map.new()
+
+      {:ok, formatted_prices}
     else
       {:ok, {{_, _, status}, _, _}} ->
         {:error, status}

lib/archethic/oracle_chain/services/uco_price/providers/coin_marketcap.ex

@@ -0,0 +1,66 @@
+defmodule Archethic.OracleChain.Services.UCOPrice.Providers.CoinMarketCap do
+  @moduledoc false
+
+  alias Archethic.OracleChain.Services.UCOPrice.Providers.Impl
+
+  @behaviour Impl
+
+  require Logger
+
+  @impl Impl
+  @spec fetch(list(binary())) :: {:ok, %{required(String.t()) => any()}} | {:error, any()}
+  def fetch(pairs) when is_list(pairs) do
+    query = 'https://coinmarketcap.com/currencies/uniris/markets/'
+
+    httpc_options = [
+      ssl: [
+        verify: :verify_peer,
+        cacertfile: CAStore.file_path(),
+        depth: 3,
+        customize_hostname_check: [
+          match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
+        ]
+      ],
+      connect_timeout: 1000,
+      timeout: 2000
+    ]
+
+    returned_prices =
+      Task.async_stream(pairs, fn pair ->
+        headers = [
+          {'user-agent',
+           'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36'},
+          {'accept', 'text/html'},
+          {'accept-language', 'en-US,en;q=0.9,es;q=0.8'},
+          {'upgrade-insecure-requests', '1'},
+          {'referer', 'https://archethic.net/'},
+          {'Cookie', 'currency=#{pair}'}
+        ]
+
+        with {:ok, {{_, 200, 'OK'}, _headers, body}} <-
+               :httpc.request(:get, {query, headers}, httpc_options, []),
+             {:ok, document} <- Floki.parse_document(body) do
+          price =
+            Floki.find(document, "div.priceTitle > div.priceValue > span")
+            |> Floki.text()
+            |> String.graphemes()
+            |> Enum.filter(&(&1 in [".", "0", "1", "2", "3", "4", "5", "6", "7", "8", "9"]))
+            |> Enum.into("")
+            |> String.to_float()
+
+          {:ok, {pair, [price]}}
+        else
+          :error ->
+            {:error, "invalid content"}
+
+          {:error, _} = e ->
+            e
+        end
+      end)
+      |> Stream.filter(&match?({:ok, {:ok, _}}, &1))
+      |> Stream.map(fn {:ok, {:ok, val}} -> val end)
+      |> Enum.into(%{})
+
+    {:ok, returned_prices}
+  end
+end

lib/archethic/oracle_chain/services/uco_price/providers/coin_paprika.ex

@@ -0,0 +1,68 @@
+defmodule Archethic.OracleChain.Services.UCOPrice.Providers.CoinPaprika do
+  @moduledoc false
+
+  alias Archethic.OracleChain.Services.UCOPrice.Providers.Impl
+
+  @behaviour Impl
+
+  require Logger
+
+  @impl Impl
+  @spec fetch(list(binary())) :: {:ok, %{required(String.t()) => any()}} | {:error, any()}
+  def fetch(pairs) when is_list(pairs) do
+    pairs_str = Enum.join(pairs, ",")
+
+    httpc_options = [
+      ssl: [
+        verify: :verify_peer,
+        cacertfile: CAStore.file_path(),
+        depth: 2,
+        customize_hostname_check: [
+          match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
+        ]
+      ],
+      connect_timeout: 1000,
+      timeout: 2000
+    ]
+
+    query =
+      String.to_charlist(
+        "https://api.coinpaprika.com/v1/coins/uco-uniris/markets?quotes=#{pairs_str}"
+      )
+
+    with {:ok, {{_, 200, 'OK'}, _headers, body}} <-
+           :httpc.request(:get, {query, []}, httpc_options, []),
+         {:ok, payload} <- Jason.decode(body) do
+      quotes =
+        payload
+        |> Enum.map(fn %{"quotes" => quotes} -> quotes end)
+
+      prices =
+        pairs
+        |> Enum.map(fn pair ->
+          values =
+            quotes
+            |> Enum.map(&get_in(&1, [String.upcase(pair), "price"]))
+            |> Enum.reject(fn price -> is_nil(price) end)
+
+          {pair, values}
+        end)
+        |> Enum.reject(fn {_pair, prices} -> prices == [] end)
+        |> Enum.into(%{})
+
+      {:ok, prices}
+    else
+      {:ok, {{_, _, status}, _, _}} ->
+        {:error, status}
+
+      {:error, %Jason.DecodeError{}} ->
+        {:error, "invalid content"}
+
+      :error ->
+        {:error, "invalid content"}
+
+      {:error, _} = e ->
+        e
+    end
+  end
+end

lib/archethic/utils.ex

@@ -708,26 +708,19 @@ defmodule Archethic.Utils do
   """
   @spec median([number]) :: number | nil
   def median([]), do: nil
+  ## To avoid all calculation from general clause to follow
 
-  def median(list) when is_list(list) do
-    midpoint =
-      (length(list) / 2)
-      |> Float.floor()
-      |> round
+  def median([number]), do: number
+  ## To avoid all calculation from general clause to follow
 
-    {l1, l2} =
-      Enum.sort(list)
-      |> Enum.split(midpoint)
+  def median(numbers) do
+    sorted = Enum.sort(numbers)
+    length_list = length(sorted)
 
-    case length(l2) > length(l1) do
-      true ->
-        [med | _] = l2
-        med
-
-      false ->
-        [m1 | _] = l2
-        [m2 | _] = Enum.reverse(l1)
-        (m1 + m2) / 2
+    case rem(length_list, 2) do
+      1 -> Enum.at(sorted, div(length_list, 2))
+      ## If we have an even number, media is the average of the two medium numbers
+      0 -> Enum.slice(sorted, div(length_list, 2) - 1, 2) |> Enum.sum() |> Kernel./(2)
     end
   end
 

mix.exs

@@ -76,7 +76,6 @@ defmodule Archethic.MixProject do
       # Test
       {:mox, "~> 1.0", only: [:test]},
       {:stream_data, "~> 0.5", only: [:test], runtime: false},
-      {:floki, "~> 0.33", only: :test},
 
       # P2P
       {:ranch, "~> 2.1", override: true},
@@ -108,7 +107,8 @@ defmodule Archethic.MixProject do
       {:ex_json_schema, "~> 0.9", override: true},
       {:pathex, "~> 2.4"},
       {:easy_ssl, "~> 1.3"},
-      {:castore, "~> 0.1"}
+      {:castore, "~> 0.1"},
+      {:floki, "~> 0.33"}
     ]
   end
 

mix.lock

@@ -32,7 +32,7 @@
   "exjsonpath": {:hex, :exjsonpath, "0.9.0", "87e593eb0deb53aa0688ca9f9edc9fb3456aca83c82245f83201ea04d696feba", [:mix], [], "hexpm", "8d7a8e9ba784e1f7a67c6f1074a3ac91a3a79a45969514ee5d95cea5bf749627"},
   "file_system": {:hex, :file_system, "0.2.10", "fb082005a9cd1711c05b5248710f8826b02d7d1784e7c3451f9c1231d4fc162d", [:mix], [], "hexpm", "41195edbfb562a593726eda3b3e8b103a309b733ad25f3d642ba49696bf715dc"},
   "floki": {:hex, :floki, "0.34.0", "002d0cc194b48794d74711731db004fafeb328fe676976f160685262d43706a8", [:mix], [], "hexpm", "9c3a9f43f40dde00332a589bd9d389b90c1f518aef500364d00636acc5ebc99c"},
-  "flow": {:hex, :flow, "1.2.1", "cfe984b2078ced0bc92807737909abe1b158288256244cc77d03ad96cbef1571", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}], "hexpm", "3f9fe6a4b28b8e82822d7a851c8d8fe3ac3c0e597aa2cf3cccd81f9af561abee"},
+  "flow": {:hex, :flow, "1.2.3", "1d9239be6a6ec8ef33f22200c8d8b3a756a28145476c7e096fbb3fda04e12f5c", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}], "hexpm", "77eeb976cb5152a1168290ce9015b8b2d41b70eb5265a36d7538a817237891d9"},
   "gen_stage": {:hex, :gen_stage, "1.1.2", "b1656cd4ba431ed02c5656fe10cb5423820847113a07218da68eae5d6a260c23", [:mix], [], "hexpm", "9e39af23140f704e2b07a3e29d8f05fd21c2aaf4088ff43cb82be4b9e3148d02"},
   "gen_state_machine": {:hex, :gen_state_machine, "3.0.0", "1e57f86a494e5c6b14137ebef26a7eb342b3b0070c7135f2d6768ed3f6b6cdff", [:mix], [], "hexpm", "0a59652574bebceb7309f6b749d2a41b45fdeda8dbb4da0791e355dd19f0ed15"},
   "git_hooks": {:hex, :git_hooks, "0.7.3", "09489e94d88dfc767662e22aff2b6208bd7cf555a19dd0e1477cca4683ce0701", [:mix], [{:blankable, "~> 1.0.0", [hex: :blankable, repo: "hexpm", optional: false]}, {:recase, "~> 0.7.0", [hex: :recase, repo: "hexpm", optional: false]}], "hexpm", "d6ddedeb4d3a8602bc3f84e087a38f6150a86d9e790628ed8bc70e6d90681659"},
@@ -56,7 +56,7 @@
   "phoenix": {:hex, :phoenix, "1.6.15", "0a1d96bbc10747fd83525370d691953cdb6f3ccbac61aa01b4acb012474b047d", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 1.0 or ~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.2", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "d70ab9fbf6b394755ea88b644d34d79d8b146e490973151f248cacd122d20672"},
   "phoenix_html": {:hex, :phoenix_html, "3.2.0", "1c1219d4b6cb22ac72f12f73dc5fad6c7563104d083f711c3fcd8551a1f4ae11", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "36ec97ba56d25c0136ef1992c37957e4246b649d620958a1f9fa86165f8bc54f"},
   "phoenix_live_dashboard": {:hex, :phoenix_live_dashboard, "0.7.2", "97cc4ff2dba1ebe504db72cb45098cb8e91f11160528b980bd282cc45c73b29c", [:mix], [{:ecto, "~> 3.6.2 or ~> 3.7", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_mysql_extras, "~> 0.5", [hex: :ecto_mysql_extras, repo: "hexpm", optional: true]}, {:ecto_psql_extras, "~> 0.7", [hex: :ecto_psql_extras, repo: "hexpm", optional: true]}, {:mime, "~> 1.6 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 0.18.3", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "0e5fdf063c7a3b620c566a30fcf68b7ee02e5e46fe48ee46a6ec3ba382dc05b7"},
-  "phoenix_live_view": {:hex, :phoenix_live_view, "0.18.3", "2e3d009422addf8b15c3dccc65ce53baccbe26f7cfd21d264680b5867789a9c1", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.6.15 or ~> 1.7.0", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.1", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c8845177a866e017dcb7083365393c8f00ab061b8b6b2bda575891079dce81b2"},
+  "phoenix_live_view": {:hex, :phoenix_live_view, "0.18.6", "460c36977643d76fc8e0b6b3c4bba703c0ef21abc74233cf7dc15d1c1696832f", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.6.15 or ~> 1.7.0", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 3.1", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.2 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "ce2768fb44c3c370df13fc4f0dc70623b662a93a201d8d7d87c4ba6542bc6b73"},
   "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.1.1", "ba04e489ef03763bf28a17eb2eaddc2c20c6d217e2150a61e3298b0f4c2012b5", [:mix], [], "hexpm", "81367c6d1eea5878ad726be80808eb5a787a23dee699f96e72b1109c57cdd8d9"},
   "phoenix_template": {:hex, :phoenix_template, "1.0.0", "c57bc5044f25f007dc86ab21895688c098a9f846a8dda6bc40e2d0ddc146e38f", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "1b066f99a26fd22064c12b2600a9a6e56700f591bf7b20b418054ea38b4d4357"},
   "phoenix_view": {:hex, :phoenix_view, "2.0.2", "6bd4d2fd595ef80d33b439ede6a19326b78f0f1d8d62b9a318e3d9c1af351098", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}], "hexpm", "a929e7230ea5c7ee0e149ffcf44ce7cf7f4b6d2bfe1752dd7c084cdff152d36f"},