Add kraken_minute table with 15-second polling

1-minute OHLC candles from Kraken, polled every 15 seconds to keep the
in-progress candle fresh. Unlike hourly/daily pollers, this one keeps
the last candle and upserts it as it updates.

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

docker-compose.yml

@@ -19,6 +19,7 @@ services:
       CORS_ORIGIN: "${DOCKER_CORS_ORIGIN}"
       FRANKFURTER_URL: "${DOCKER_FRANKFURTER_URL}"
       STRIKE_API_KEY: "${STRIKE_API_KEY}"
+      KRAKEN_ENABLED: "true"
     extra_hosts:
       - "postgres:host-gateway"
 

resources/migrations/20260313000000-create-kraken-minute.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS kraken_minute;

resources/migrations/20260313000000-create-kraken-minute.up.sql

@@ -0,0 +1,11 @@
+CREATE TABLE kraken_minute (
+  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

@@ -155,6 +155,23 @@ SELECT id, event_id, occurred_at, exchange, fiat_amount, fiat_currency,
 FROM deposits
 ORDER BY occurred_at DESC, id DESC
 
+-- :name upsert-kraken-minute! :! :n
+-- :doc Upsert a Kraken minute OHLC candle
+INSERT INTO kraken_minute (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-minute :? :1
+-- :doc Get the most recent Kraken minute candle timestamp
+SELECT ts FROM kraken_minute ORDER BY ts DESC LIMIT 1
+
 -- :name get-wallet-balances :? :*
 -- :doc Compute sats balance per wallet from events
 SELECT w.id, w.name,

resources/system.edn

@@ -56,10 +56,16 @@
   :enabled? #or [#env BINANCE_ENABLED "true"]}
 
  :kraken/ohlc
- {:query-fn #ig/ref :db.sql/query-fn}
+ {:query-fn #ig/ref :db.sql/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
 
  :kraken/ohlc-day
- {:query-fn #ig/ref :db.sql/query-fn}
+ {:query-fn #ig/ref :db.sql/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
+
+ :kraken/ohlc-minute
+ {:query-fn #ig/ref :db.sql/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
 
  :frankfurter/rates
  {:url      #or [#env FRANKFURTER_URL "http://localhost:8080"]

src/clj/pmagnus/btcdata/core.clj

@@ -19,6 +19,7 @@
    ;; Pollers
    [pmagnus.btcdata.kraken.ohlc]
    [pmagnus.btcdata.kraken.ohlc-daily]
+   [pmagnus.btcdata.kraken.ohlc-minute]
    [pmagnus.btcdata.frankfurter.rates]
    [pmagnus.btcdata.strike.ticker])
   (:gen-class)

src/clj/pmagnus/btcdata/kraken/ohlc.clj

@@ -103,19 +103,22 @@
               (log/error e "Kraken OHLC poll failed"))))))))
 
 (defmethod ig/init-key :kraken/ohlc
-  [_ {:keys [query-fn]}]
+  [_ {:keys [query-fn enabled?]}]
-  (let [client    (HttpClient/newHttpClient)
+  (if (= enabled? "false")
-        db-since  (seed-since-from-db query-fn)
+    (do (log/info "Kraken OHLC poller disabled")
-        since     (atom db-since)
+        {:running? (atom false)})
-        running?  (atom true)
+    (let [client    (HttpClient/newHttpClient)
-        has-data? (some? db-since)
+          db-since  (seed-since-from-db query-fn)
-        fut       (start-poll-loop! client query-fn since running? has-data?)]
+          since     (atom db-since)
-    (if has-data?
+          running?  (atom true)
-      (log/info "Starting Kraken OHLC poller, next fetch in" (quot (ms-until-next-poll) 60000) "minutes")
+          has-data? (some? db-since)
-      (log/info "Starting Kraken OHLC poller, fetching immediately (no data)"))
+          fut       (start-poll-loop! client query-fn since running? has-data?)]
-    {:running? running?
+      (if has-data?
-     :future   fut
+        (log/info "Starting Kraken OHLC poller, next fetch in" (quot (ms-until-next-poll) 60000) "minutes")
-     :since    since}))
+        (log/info "Starting Kraken OHLC poller, fetching immediately (no data)"))
+      {:running? running?
+       :future   fut
+       :since    since})))
 
 (defmethod ig/halt-key! :kraken/ohlc
   [_ {:keys [running? future]}]

src/clj/pmagnus/btcdata/kraken/ohlc_daily.clj

@@ -102,19 +102,22 @@
               (log/error e "Kraken daily OHLC poll failed"))))))))
 
 (defmethod ig/init-key :kraken/ohlc-day
-  [_ {:keys [query-fn]}]
+  [_ {:keys [query-fn enabled?]}]
-  (let [client    (HttpClient/newHttpClient)
+  (if (= enabled? "false")
-        db-since  (seed-since-from-db query-fn)
+    (do (log/info "Kraken daily OHLC poller disabled")
-        since     (atom db-since)
+        {:running? (atom false)})
-        running?  (atom true)
+    (let [client    (HttpClient/newHttpClient)
-        has-data? (some? db-since)
+          db-since  (seed-since-from-db query-fn)
-        fut       (start-poll-loop! client query-fn since running? has-data?)]
+          since     (atom db-since)
-    (if has-data?
+          running?  (atom true)
-      (log/info "Starting Kraken daily OHLC poller, next fetch in" (quot (ms-until-next-daily-poll) 60000) "minutes")
+          has-data? (some? db-since)
-      (log/info "Starting Kraken daily OHLC poller, fetching immediately (no data)"))
+          fut       (start-poll-loop! client query-fn since running? has-data?)]
-    {:running? running?
+      (if has-data?
-     :future   fut
+        (log/info "Starting Kraken daily OHLC poller, next fetch in" (quot (ms-until-next-daily-poll) 60000) "minutes")
-     :since    since}))
+        (log/info "Starting Kraken daily OHLC poller, fetching immediately (no data)"))
+      {:running? running?
+       :future   fut
+       :since    since})))
 
 (defmethod ig/halt-key! :kraken/ohlc-day
   [_ {:keys [running? future]}]

src/clj/pmagnus/btcdata/kraken/ohlc_minute.clj

@@ -0,0 +1,107 @@
+(ns pmagnus.btcdata.kraken.ohlc-minute
+  (: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=1")
+
+(defn- fetch-ohlc
+  "HTTP GET to Kraken OHLC endpoint for 1-minute candles.
+   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_minute table."
+  [query-fn candles]
+  (doseq [candle candles]
+    (query-fn :upsert-kraken-minute! 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-minute {})
+          :ts))
+
+(defn- poll!
+  "Fetch OHLC data including the last in-progress candle, upsert all.
+   Returns the count of saved candles."
+  [client query-fn since-atom]
+  (let [result   (fetch-ohlc client @since-atom)
+        last-ts  (:last result)
+        pair-key (first (remove #{:last} (keys result)))
+        raw      (get result pair-key)
+        candles  (map parse-candle raw)]
+    (when (seq candles)
+      (save-candles! query-fn candles)
+      (when last-ts
+        (reset! since-atom last-ts))
+      (log/info "Fetched" (count candles) "Kraken minute candles"))
+    (count candles)))
+
+(defn- start-poll-loop!
+  "Start a background future that polls Kraken every 15 seconds."
+  [client query-fn since-atom running?]
+  (future
+    (try
+      (poll! client query-fn since-atom)
+      (catch Exception e
+        (log/error e "Kraken minute initial poll failed")))
+    (while @running?
+      (Thread/sleep 15000)
+      (when @running?
+        (try
+          (poll! client query-fn since-atom)
+          (catch Exception e
+            (log/error e "Kraken minute poll failed")))))))
+
+(defmethod ig/init-key :kraken/ohlc-minute
+  [_ {:keys [query-fn enabled?]}]
+  (if (= enabled? "false")
+    (do (log/info "Kraken minute poller disabled")
+        {:running? (atom false)})
+    (let [client    (HttpClient/newHttpClient)
+          db-since  (seed-since-from-db query-fn)
+          since     (atom db-since)
+          running?  (atom true)
+          fut       (start-poll-loop! client query-fn since running?)]
+      (log/info "Starting Kraken minute poller (15s interval)")
+      {:running? running?
+       :future   fut
+       :since    since})))
+
+(defmethod ig/halt-key! :kraken/ohlc-minute
+  [_ {:keys [running? future]}]
+  (log/info "Stopping Kraken minute poller")
+  (reset! running? false)
+  (future-cancel future))