Add Kraken hourly OHLC poller with database storage

Fetches Bitcoin hourly candles from the Kraken public REST API
and upserts them into a kraken_hour table on a 5-minute poll interval,
following the existing Integrant component pattern.

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

resources/migrations/20260218000000-create-kraken-hour.down.sql

@@ -0,0 +1 @@
+DROP TABLE kraken_hour;

resources/migrations/20260218000000-create-kraken-hour.up.sql

@@ -0,0 +1,11 @@
+CREATE TABLE kraken_hour (
+  ts          BIGINT         NOT NULL PRIMARY KEY,
+  open        NUMERIC(18,8)  NOT NULL,
+  high        NUMERIC(18,8)  NOT NULL,
+  low         NUMERIC(18,8)  NOT NULL,
+  close       NUMERIC(18,8)  NOT NULL,
+  vwap        NUMERIC(18,8)  NOT NULL,
+  volume      NUMERIC(24,8)  NOT NULL,
+  trade_count INTEGER        NOT NULL,
+  created_at  TIMESTAMPTZ    NOT NULL DEFAULT NOW()
+);

resources/queries.sql

@@ -7,3 +7,21 @@ INSERT INTO binance_price (price) VALUES (:price)
 -- :name get-latest-binance-price :? :1
 -- :doc Get the most recent Binance BTC price
 SELECT price, recorded_at FROM binance_price ORDER BY id DESC LIMIT 1
+
+-- :name upsert-kraken-hour! :! :n
+-- :doc Upsert a Kraken hourly OHLC candle
+INSERT INTO kraken_hour (ts, open, high, low, close, vwap, volume, trade_count)
+VALUES (:ts, :open, :high, :low, :close, :vwap, :volume, :trade-count)
+ON CONFLICT (ts) DO UPDATE
+SET open = EXCLUDED.open,
+    high = EXCLUDED.high,
+    low  = EXCLUDED.low,
+    close = EXCLUDED.close,
+    vwap  = EXCLUDED.vwap,
+    volume = EXCLUDED.volume,
+    trade_count = EXCLUDED.trade_count
+
+-- :name get-latest-kraken-hour :? :1
+-- :doc Get the most recent Kraken hourly candle by timestamp
+SELECT ts, open, high, low, close, vwap, volume, trade_count, created_at
+FROM kraken_hour ORDER BY ts DESC LIMIT 1

resources/system.edn

@@ -56,4 +56,8 @@
 
  :ws/binance
  {:query-fn #ig/ref :db.sql/query-fn
-  :uri      "wss://stream.binance.com:9443/ws/btcusdt@trade"}}
+  :uri      "wss://stream.binance.com:9443/ws/btcusdt@trade"}
+
+ :kraken/ohlc
+ {:query-fn    #ig/ref :db.sql/query-fn
+  :interval-ms 300000}}

src/clj/pmagnus/btcprice/core.clj

@@ -15,7 +15,9 @@
    [pmagnus.btcprice.web.routes.api]
    [pmagnus.btcprice.web.routes.ui]
    ;; WebSocket clients
-   [pmagnus.btcprice.ws.binance])
+   [pmagnus.btcprice.ws.binance]
+   ;; Pollers
+   [pmagnus.btcprice.kraken.ohlc])
   (:gen-class))
 
 (defonce system (atom nil))

src/clj/pmagnus/btcprice/kraken/ohlc.clj

@@ -0,0 +1,101 @@
+(ns pmagnus.btcprice.kraken.ohlc
+  (: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]))
+
+(def ^:private kraken-ohlc-url
+  "https://api.kraken.com/0/public/OHLC?pair=XBTUSD&interval=60")
+
+(defn- fetch-ohlc
+  "HTTP GET to Kraken OHLC endpoint. Returns parsed JSON result map.
+   When `since` is provided, appends &since= to fetch only newer candles."
+  [^HttpClient client since]
+  (let [url     (if since
+                  (str kraken-ohlc-url "&since=" since)
+                  kraken-ohlc-url)
+        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)]
+    (when-let [errors (seq (:error body))]
+      (throw (ex-info "Kraken API error" {:errors errors})))
+    (:result body)))
+
+(defn- parse-candle
+  "Convert a Kraken OHLC array [ts, open, high, low, close, vwap, volume, count]
+   to a map with bigdec values."
+  [[ts open high low close vwap volume count]]
+  {:ts          (long ts)
+   :open        (bigdec open)
+   :high        (bigdec high)
+   :low         (bigdec low)
+   :close       (bigdec close)
+   :vwap        (bigdec vwap)
+   :volume      (bigdec volume)
+   :trade-count (int count)})
+
+(defn- save-candles!
+  "Upsert each candle into the kraken_hour table."
+  [query-fn candles]
+  (doseq [candle candles]
+    (query-fn :upsert-kraken-hour! candle)))
+
+(defn- seed-since-from-db
+  "Query DB for the latest candle timestamp. Returns it or nil."
+  [query-fn]
+  (some-> (query-fn :get-latest-kraken-hour {})
+          :ts))
+
+(defn- poll!
+  "Fetch OHLC data, drop the last (in-progress) candle, save completed ones.
+   Returns the count of saved candles."
+  [client query-fn since-atom]
+  (let [result   (fetch-ohlc client @since-atom)
+        ;; Kraken returns a map with the pair key and a "last" key
+        last-ts  (:last result)
+        pair-key (first (remove #{:last} (keys result)))
+        raw      (get result pair-key)
+        candles  (map parse-candle (butlast raw))]
+    (when (seq candles)
+      (save-candles! query-fn candles)
+      (when last-ts
+        (reset! since-atom last-ts))
+      (log/info "Fetched" (count candles) "completed Kraken hourly candles"))
+    (count candles)))
+
+(defn- start-poll-loop!
+  "Start a background future that polls Kraken every `interval-ms`.
+   Returns the future."
+  [client query-fn since-atom running? interval-ms]
+  (future
+    (while @running?
+      (try
+        (poll! client query-fn since-atom)
+        (catch Exception e
+          (log/error e "Kraken OHLC poll failed")))
+      (Thread/sleep interval-ms))))
+
+(defmethod ig/init-key :kraken/ohlc
+  [_ {:keys [query-fn interval-ms]}]
+  (log/info "Starting Kraken OHLC poller, interval:" interval-ms "ms")
+  (let [client    (HttpClient/newHttpClient)
+        since     (atom (seed-since-from-db query-fn))
+        running?  (atom true)
+        fut       (start-poll-loop! client query-fn since running?
+                                    (or interval-ms 300000))]
+    {:running? running?
+     :future   fut
+     :since    since}))
+
+(defmethod ig/halt-key! :kraken/ohlc
+  [_ {:keys [running? future]}]
+  (log/info "Stopping Kraken OHLC poller")
+  (reset! running? false)
+  (future-cancel future))