Merge pull request #12 from sgobotta/feature/upgrade-to-daily-graphs-…

…for-dollar-quotes

feature/upgrade to daily graphs for dollar quotes

assets/js/charts/line_chart.js

@@ -34,8 +34,8 @@ export default class {
             suggestedMin: 50
           },
           y: {
-            suggestedMax: 200,
-            suggestedMin: 50
+            suggestedMax: 1200,
+            suggestedMin: 800
           }
         }
       },

lib/ex_finance/currencies.ex

@@ -374,9 +374,17 @@ defmodule ExFinance.Currencies do
     stream_name =
       get_stream_name("currency-history_" <> supplier_name <> "_" <> type)
 
-    with {:ok, entries} <- Redis.Client.fetch_history(stream_name, 40),
-         sorted_entries <- Enum.reverse(entries),
-         history <- map_currency_history(sorted_entries) do
+    days_before_now = -20
+
+    since =
+      DateTime.utc_now()
+      |> DateTime.add(days_before_now, :day)
+      |> DateTime.to_unix(:millisecond)
+
+    with {:ok, entries} <-
+           Redis.Client.fetch_reverse_stream_since(stream_name, since),
+         filtered_entries <- filter_history_entries(entries),
+         history <- map_currency_history(filtered_entries) do
       {:ok, history}
     else
       error ->
@@ -395,6 +403,18 @@ defmodule ExFinance.Currencies do
       :error
   end
 
+  @spec filter_history_entries([Redis.Stream.Entry.t()]) :: [
+          Redis.Stream.Entry.t()
+        ]
+  defp filter_history_entries(entries) do
+    entries
+    |> Enum.group_by(fn entry ->
+      "#{entry.datetime.year}-#{entry.datetime.month}-#{entry.datetime.day}"
+    end)
+    |> Enum.map(fn {_datetime, entries} -> hd(entries) end)
+    |> Enum.sort_by(& &1.datetime, :asc)
+  end
+
   @spec map_currency_history([Redis.Stream.Entry.t()]) :: [
           {NaiveDateTime.t(), Currency.t()}
         ]

lib/redis/client.ex

@@ -72,6 +72,11 @@ defmodule Redis.Client do
     Redix.command(:redix, [command, stream_name, "+", "-", "COUNT", count])
   end
 
+  @spec fetch_reverse_stream_since(binary(), binary() | non_neg_integer()) ::
+          Stream.response()
+  def fetch_reverse_stream_since(stream_name, since),
+    do: Stream.xrevrange(stream_name, "+", since)
+
   @spec fetch_last_stream_entry(String.t()) ::
           {:ok, Stream.Entry.t()}
           | {:error, :no_result | :stream_parse_error}
@@ -177,47 +182,3 @@ defmodule Redis.Client do
         acc ++ [Atom.to_string(key), value]
       end)
 end
-
-defmodule Redis.Stream.Entry do
-  @moduledoc false
-  require Integer
-
-  @enforce_keys [:id, :values, :datetime]
-  defstruct id: nil, values: nil, datetime: nil
-
-  @type t :: %__MODULE__{}
-
-  @doc """
-  Given a redis stream entry returns a readable map representation of the
-  entry values.
-  """
-  @spec from_raw_entry(any()) :: map()
-  def from_raw_entry([entry_id, entry]) do
-    datetime = parse_stream_entry_id(entry_id)
-
-    Enum.reduce(Enum.with_index(entry), {[], []}, fn {value, index},
-                                                     {keys, values} ->
-      case Integer.is_even(index) do
-        true ->
-          {keys ++ [value], values}
-
-        false ->
-          {keys, values ++ [value]}
-      end
-    end)
-    |> then(fn {keys, values} -> Enum.zip(keys, values) end)
-    |> Enum.into(%{})
-    |> then(fn values ->
-      %__MODULE__{id: entry_id, values: values, datetime: datetime}
-    end)
-  end
-
-  @spec parse_stream_entry_id(String.t()) :: DateTime.t()
-  defp parse_stream_entry_id(entry_id) do
-    entry_id
-    |> String.split("-")
-    |> hd
-    |> String.to_integer()
-    |> DateTime.from_unix!(:millisecond)
-  end
-end

lib/redis/stream.ex

@@ -0,0 +1,93 @@
+defmodule Redis.Stream do
+  @moduledoc """
+  Module definition for redis streams
+  """
+
+  alias Redis.Client
+  alias Redis.Stream
+
+  require Logger
+
+  @type since :: binary() | non_neg_integer()
+  @type until :: binary() | non_neg_integer()
+  @type response :: {:ok, [Stream.Entry.t()]} | {:error, :stream_parse_error}
+
+  @spec xrevrange(binary(), since(), until()) :: response()
+  def xrevrange(stream_name, until \\ "+", since \\ "-") do
+    Redix.command(:redix, ["XREVRANGE", stream_name, until, since])
+    |> parse_response()
+  end
+
+  @spec parse_response(Client.redix_response()) ::
+          {:ok, [map()]} | {:error, :stream_parse_error}
+  defp parse_response(reply) do
+    with {:ok, entries} <- parse_reply(reply),
+         parsed_entries <- parse_stream_entries(entries) do
+      {:ok, parsed_entries}
+    end
+  rescue
+    error ->
+      Logger.error(
+        "There was an error while processing the stream result error=#{inspect(error)}"
+      )
+
+      {:error, :stream_parse_error}
+  end
+
+  @spec parse_reply({atom, list | binary} | any) ::
+          {:error, :no_result} | {:ok, any} | any
+  defp parse_reply({:ok, []}), do: {:error, :no_result}
+  defp parse_reply({:ok, _result} = result), do: result
+
+  @spec parse_stream_entries([any()]) :: [Stream.Entry.t()]
+  defp parse_stream_entries(entries),
+    do: entries |> Enum.map(&parse_stream_entry/1)
+
+  @spec parse_stream_entry(list()) :: Stream.Entry.t()
+  defp parse_stream_entry([_entry_id, _entry_values] = entry),
+    do: Stream.Entry.from_raw_entry(entry)
+end
+
+defmodule Redis.Stream.Entry do
+  @moduledoc false
+  require Integer
+
+  @enforce_keys [:id, :values, :datetime]
+  defstruct id: nil, values: nil, datetime: nil
+
+  @type t :: %__MODULE__{}
+
+  @doc """
+  Given a redis stream entry returns a readable map representation of the
+  entry values.
+  """
+  @spec from_raw_entry(any()) :: t()
+  def from_raw_entry([entry_id, entry]) do
+    datetime = parse_stream_entry_id(entry_id)
+
+    Enum.reduce(Enum.with_index(entry), {[], []}, fn {value, index},
+                                                     {keys, values} ->
+      case Integer.is_even(index) do
+        true ->
+          {keys ++ [value], values}
+
+        false ->
+          {keys, values ++ [value]}
+      end
+    end)
+    |> then(fn {keys, values} -> Enum.zip(keys, values) end)
+    |> Enum.into(%{})
+    |> then(fn values ->
+      %__MODULE__{id: entry_id, values: values, datetime: datetime}
+    end)
+  end
+
+  @spec parse_stream_entry_id(String.t()) :: DateTime.t()
+  defp parse_stream_entry_id(entry_id) do
+    entry_id
+    |> String.split("-")
+    |> hd
+    |> String.to_integer()
+    |> DateTime.from_unix!(:millisecond)
+  end
+end

version.txt

@@ -1 +1 @@
-0.1.2
+0.1.3