Chapter 19 (#21)

* working through chapter 19

* finished chapter 19

* self code review against franton code

Author: Bodhert

apps/indicator/.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

apps/indicator/.gitignore

@@ -0,0 +1,26 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+indicator-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

apps/indicator/README.md

@@ -0,0 +1,21 @@
+# Indicator
+
+**TODO: Add description**
+
+## Installation
+
+If [available in Hex](https://hex.pm/docs/publish), the package can be installed
+by adding `indicator` to your list of dependencies in `mix.exs`:
+
+```elixir
+def deps do
+  [
+    {:indicator, "~> 0.1.0"}
+  ]
+end
+```
+
+Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
+and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
+be found at <https://hexdocs.pm/indicator>.
+

apps/indicator/lib/indicator.ex

@@ -0,0 +1,11 @@
+defmodule Indicator do
+  @moduledoc """
+  Documentation for `Indicator`.
+  """
+  def aggregate_ohlcs(symbol) do
+    DynamicSupervisor.start_child(
+      Indicator.DynamicSupervisor,
+      {Indicator.Ohlc.Worker, symbol}
+    )
+  end
+end

apps/indicator/lib/indicator/application.ex

@@ -0,0 +1,19 @@
+defmodule Indicator.Application do
+  # See https://hexdocs.pm/elixir/Application.html
+  # for more information on OTP Applications
+  @moduledoc false
+
+  use Application
+
+  @impl true
+  def start(_type, _args) do
+    children = [
+      {DynamicSupervisor, strategy: :one_for_one, name: Indicator.DynamicSupervisor}
+    ]
+
+    # See https://hexdocs.pm/elixir/Supervisor.html
+    # for other strategies and supported options
+    opts = [strategy: :one_for_one, name: Indicator.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
+end

apps/indicator/lib/indicator/ohlc.ex

@@ -0,0 +1,80 @@
+defmodule Indicator.Ohlc do
+  require Logger
+  alias Core.Struct.TradeEvent
+
+  @pubsub_client Application.compile_env(:core, :pubsub_client)
+  @enforce_keys [:symbol, :start_time, :duration]
+  defstruct [:symbol, :start_time, :duration, :open, :high, :low, :close]
+
+  def process([_ | _] = ohlcs, %TradeEvent{} = trade_event) do
+    {old_ohlcs, new_ohlcs} = merge_prices(ohlcs, trade_event.price, trade_event.trade_time)
+    Enum.each(old_ohlcs, &maybe_broadcast/1)
+    new_ohlcs
+  end
+
+  def process(symbol, %TradeEvent{} = trade_event) do
+    generate_ohlcs(symbol, trade_event.price, trade_event.trade_time)
+  end
+
+  def merge_prices(ohlcs, price, trade_time) do
+    results = ohlcs |> Enum.map(&merge_price(&1, price, trade_time))
+
+    {
+      results |> Enum.map(&elem(&1, 0)) |> Enum.filter(& &1),
+      results |> Enum.map(&elem(&1, 1))
+    }
+  end
+
+  def merge_price(%__MODULE__{} = ohlc, price, trade_time) do
+    if within_current_timeframe(ohlc.start_time, ohlc.duration, trade_time) do
+      {nil, %{ohlc | low: min(ohlc.low, price), high: max(ohlc.high, price), close: price}}
+    else
+      {ohlc, generate_ohlc(ohlc.symbol, ohlc.duration, price, trade_time)}
+    end
+  end
+
+  def within_current_timeframe(start_time, duration, trade_time) do
+    end_time = start_time + duration * 60
+    trade_time = div(trade_time, 1000)
+    start_time <= trade_time && trade_time < end_time
+  end
+
+  def generate_ohlcs(symbol, price, trade_time) do
+    [1, 5, 15, 60, 4 * 60, 24 * 60]
+    |> Enum.map(
+      &generate_ohlc(
+        symbol,
+        &1,
+        price,
+        trade_time
+      )
+    )
+  end
+
+  def generate_ohlc(symbol, duration, price, trade_time) do
+    # start_time = div(div(div(trade_time, 1000), 60), duration)
+    start_time = trade_time |> div(1000) |> div(60) |> div(duration)
+
+    %__MODULE__{
+      symbol: symbol,
+      start_time: start_time,
+      duration: duration,
+      open: price,
+      high: price,
+      low: price,
+      close: price
+    }
+  end
+
+  defp maybe_broadcast(nil), do: :ok
+
+  defp maybe_broadcast(%__MODULE__{} = ohlc) do
+    Logger.debug("Broadcasting OHLC: #{inspect(ohlc)}")
+
+    @pubsub_client.broadcast(
+      Core.PubSub,
+      "OHLC:#{ohlc.symbol}",
+      ohlc
+    )
+  end
+end

apps/indicator/lib/indicator/ohlc/worker.ex

@@ -0,0 +1,28 @@
+defmodule Indicator.Ohlc.Worker do
+  use GenServer
+  require Logger
+  alias Core.Struct.TradeEvent
+
+  @logger Application.compile_env(:core, :logger)
+  @pubsub_client Application.compile_env(:core, :pubsub_client)
+
+  def start_link(symbol) do
+    GenServer.start_link(__MODULE__, symbol)
+  end
+
+  def init(symbol) do
+    symbol = String.upcase(symbol)
+    @logger.debug("Initializing a new OHLC worker for #{symbol}")
+
+    @pubsub_client.subscribe(
+      Core.PubSub,
+      "TRADE_EVENTS:#{symbol}"
+    )
+
+    {:ok, symbol}
+  end
+
+  def handle_info(%TradeEvent{} = trade_event, ohlc) do
+    {:noreply, Indicator.Ohlc.process(ohlc, trade_event)}
+  end
+end

apps/indicator/mix.exs

@@ -0,0 +1,33 @@
+defmodule Indicator.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :indicator,
+      version: "0.1.0",
+      build_path: "../../_build",
+      config_path: "../../config/config.exs",
+      deps_path: "../../deps",
+      lockfile: "../../mix.lock",
+      elixir: "~> 1.17",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger],
+      mod: {Indicator.Application, []}
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:core, in_umbrella: true},
+      {:phoenix_pubsub, "~> 2.0"}
+    ]
+  end
+end

apps/indicator/test/indicator_test.exs

@@ -0,0 +1,8 @@
+defmodule IndicatorTest do
+  use ExUnit.Case
+  doctest Indicator
+
+  test "greets the world" do
+    assert Indicator.hello() == :world
+  end
+end

apps/indicator/test/test_helper.exs

@@ -0,0 +1 @@
+ExUnit.start()