Add Frankfurter EUR/DKK rate polling and include rates in WebSocket stream

Polls self-hosted Frankfurter API hourly for USD-based EUR and DKK exchange
rates. Both the WebSocket HTML fragments and the /api/price/latest JSON
endpoint now include the currency rates alongside BTC price.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

CLAUDE.md

@@ -44,8 +44,8 @@ Docker runs alongside local dev on different ports and databases:
 ## API Endpoints
 
 - `GET /api/health` — Health check
-- `GET /api/price/ws` — WebSocket endpoint for live HTML price fragments (HTMX-compatible)
+- `GET /api/price/ws` — WebSocket endpoint for live HTML price fragments with EUR/DKK rates (HTMX-compatible)
-- `GET /api/price/latest` — Latest price as JSON
+- `GET /api/price/latest` — Latest price + EUR/DKK rates as JSON
 
 ## Source Layout
 
@@ -55,6 +55,8 @@ src/clj/pmagnus/btcdata/
 ├── config.clj            # System config loader
 ├── ws/
 │   └── binance.clj       # Binance WebSocket client
+├── frankfurter/
+│   └── rates.clj         # EUR/DKK exchange rate poller
 ├── kraken/
 │   ├── ohlc.clj          # Hourly OHLC poller
 │   └── ohlc_daily.clj    # Daily OHLC poller

docker-compose.yml

@@ -4,10 +4,13 @@ services:
     restart: unless-stopped
     ports:
       - "4101:4101"
+    depends_on:
+      - frankfurter
     environment:
       BTCDATA_PORT: "4101"
       JDBC_URL: "jdbc:postgresql://postgres:5432/btcprod?user=postgres&password=ratata,123"
       CORS_ORIGIN: "http://localhost:4041"
+      FRANKFURTER_URL: "http://frankfurter:8080"
     extra_hosts:
       - "postgres:host-gateway"
 

resources/system.edn

@@ -46,7 +46,8 @@
  :reitit.routes/api
  {:base-path    "/api"
   :query-fn     #ig/ref :db.sql/query-fn
-  :binance   #ig/ref :ws/binance}
+  :binance      #ig/ref :ws/binance
+  :frankfurter  #ig/ref :frankfurter/rates}
 
  :ws/binance
  {:query-fn #ig/ref :db.sql/query-fn
@@ -56,4 +57,7 @@
  {:query-fn #ig/ref :db.sql/query-fn}
 
  :kraken/ohlc-day
- {:query-fn #ig/ref :db.sql/query-fn}}
+ {:query-fn #ig/ref :db.sql/query-fn}
+
+ :frankfurter/rates
+ {:url #or [#env FRANKFURTER_URL "http://localhost:8080"]}}

src/clj/pmagnus/btcdata/frankfurter/rates.clj

@@ -0,0 +1,66 @@
+(ns pmagnus.btcdata.frankfurter.rates
+  (:require
+   [clojure.data.json :as json]
+   [clojure.tools.logging :as log]
+   [integrant.core :as ig])
+  (:import
+   [java.net URI]
+   [java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers]
+   [java.time Instant]))
+
+(defn- fetch-rates
+  "GET /latest?from=USD&to=EUR,DKK from Frankfurter. Returns {:EUR x :DKK y}."
+  [^HttpClient client base-url]
+  (let [url     (str base-url "/latest?from=USD&to=EUR,DKK")
+        request (-> (HttpRequest/newBuilder)
+                    (.uri (URI. url))
+                    (.header "Accept" "application/json")
+                    (.GET)
+                    (.build))
+        resp    (.send client request (HttpResponse$BodyHandlers/ofString))
+        body    (json/read-str (.body resp) :key-fn keyword)]
+    (:rates body)))
+
+(defn- poll!
+  "Fetch rates and update the atom."
+  [client base-url rates-atom]
+  (let [rates (fetch-rates client base-url)]
+    (when (and (:EUR rates) (:DKK rates))
+      (reset! rates-atom
+              {:eur        (bigdec (str (:EUR rates)))
+               :dkk        (bigdec (str (:DKK rates)))
+               :updated-at (Instant/now)})
+      (log/info "Frankfurter rates — EUR:" (:EUR rates) "DKK:" (:DKK rates)))))
+
+(defn- start-poll-loop!
+  "Fetch immediately, then poll every 60 minutes."
+  [client base-url rates-atom running?]
+  (future
+    (try
+      (poll! client base-url rates-atom)
+      (catch Exception e
+        (log/error e "Frankfurter initial fetch failed")))
+    (while @running?
+      (Thread/sleep 3600000)
+      (when @running?
+        (try
+          (poll! client base-url rates-atom)
+          (catch Exception e
+            (log/error e "Frankfurter poll failed")))))))
+
+(defmethod ig/init-key :frankfurter/rates
+  [_ {:keys [url]}]
+  (log/info "Starting Frankfurter rates poller:" url)
+  (let [client   (HttpClient/newHttpClient)
+        rates    (atom nil)
+        running? (atom true)
+        fut      (start-poll-loop! client url rates running?)]
+    {:rates    rates
+     :running? running?
+     :future   fut}))
+
+(defmethod ig/halt-key! :frankfurter/rates
+  [_ {:keys [running? future]}]
+  (log/info "Stopping Frankfurter rates poller")
+  (reset! running? false)
+  (future-cancel future))

src/clj/pmagnus/btcdata/web/routes/api.clj

@@ -18,15 +18,17 @@
    [java.time.format DateTimeFormatter]
    [java.util.concurrent LinkedBlockingQueue TimeUnit]))
 
-(defn- price->json [{:keys [price prev-price recorded-at]}]
+(defn- price->json [{:keys [price prev-price recorded-at]} rates]
   (json/write-str
-   {:price       (str price)
+   (cond-> {:price       (str price)
             :prev_price  (when prev-price (str prev-price))
-    :recorded_at (str recorded-at)}))
+            :recorded_at (str recorded-at)}
+     rates (assoc :eur (str (:eur rates))
+                  :dkk (str (:dkk rates))))))
 
 (def ^:private ts-fmt (DateTimeFormatter/ofPattern "dd-MM HH:mm:ss"))
 
-(defn- price->html [{:keys [price prev-price recorded-at]}]
+(defn- price->html [{:keys [price prev-price recorded-at]} rates]
   (let [p  (double (bigdec price))
         pp (when prev-price (double (bigdec prev-price)))
         color (cond
@@ -39,50 +41,64 @@
          "<div class=\"bg-white rounded-2xl shadow-sm border border-gray-200 p-6\">"
          "<div class=\"text-center\">"
          "<p class=\"text-4xl font-bold " color "\">$" (format "%,.2f" p) "</p>"
+         (when rates
+           (let [eur (double (:eur rates))
+                 dkk (double (:dkk rates))]
+             (str "<div class=\"flex justify-center gap-4 mt-3\">"
+                  "<span class=\"text-sm text-gray-600\">EUR " (format "%.4f" eur) "</span>"
+                  "<span class=\"text-sm text-gray-600\">DKK " (format "%.4f" dkk) "</span>"
+                  "</div>")))
          "<p class=\"text-xs text-gray-400 mt-2\">Updated " ts "</p>"
          "</div></div></div>")))
 
-(defn- start-ws-send-loop! [^WebSocketChannel channel price-atom]
+(defn- start-ws-send-loop! [^WebSocketChannel channel price-atom rates-atom]
   (let [queue (LinkedBlockingQueue.)
-        wkey  (keyword (gensym "ws-"))]
+        wkey  (keyword (gensym "ws-"))
+        rkey  (keyword (gensym "ws-r-"))]
     (.set (.getReceiveSetter channel)
           (proxy [AbstractReceiveListener] []))
     (.resumeReceives channel)
     (add-watch price-atom wkey
-      (fn [_ _ _ v] (.offer queue v)))
+      (fn [_ _ _ _] (.offer queue :update)))
-    (when-let [v @price-atom]
+    (add-watch rates-atom rkey
-      (.offer queue v))
+      (fn [_ _ _ _] (.offer queue :update)))
+    (when @price-atom
+      (.offer queue :update))
     (future
       (try
         (loop []
           (when (.isOpen channel)
-            (if-let [v (.poll queue 15 TimeUnit/SECONDS)]
+            (if (.poll queue 15 TimeUnit/SECONDS)
-              (WebSockets/sendTextBlocking (price->html v) channel)
+              (when-let [v @price-atom]
+                (WebSockets/sendTextBlocking (price->html v @rates-atom) channel))
               (WebSockets/sendTextBlocking "" channel))
             (recur)))
         (catch Exception _)
         (finally
           (remove-watch price-atom wkey)
+          (remove-watch rates-atom rkey)
           (when (.isOpen channel)
             (try (.close channel) (catch Exception _))))))))
 
-(defn- ws-price-handler [{:keys [binance]} req]
+(defn- ws-price-handler [{:keys [binance frankfurter]} req]
   (let [^HttpServerExchange exchange (:server-exchange req)
         price-atom (:latest-price binance)
+        rates-atom (:rates frankfurter)
         callback (proxy [WebSocketConnectionCallback] []
                    (onConnect [_ws-exchange channel]
-                     (start-ws-send-loop! channel price-atom)))
+                     (start-ws-send-loop! channel price-atom rates-atom)))
         handler (WebSocketProtocolHandshakeHandler. callback)]
     (.handleRequest handler exchange)
     nil))
 
-(defn- latest-price-handler [{:keys [binance]} _req]
+(defn- latest-price-handler [{:keys [binance frankfurter]} _req]
   (let [price-atom (:latest-price binance)
+        rates-atom (:rates frankfurter)
         v          @price-atom]
     (if v
       {:status  200
        :headers {"Content-Type" "application/json"}
-       :body    (price->json v)}
+       :body    (price->json v @rates-atom)}
       {:status  503
        :headers {"Content-Type" "application/json"}
        :body    (json/write-str {:error "No price data yet"})})))