Add Kraken daily OHLC poller with database storage

Polls Kraken API for 1-day BTC/USD candles aligned to 00:01 UTC,
mirroring the existing hourly poller pattern.

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

resources/migrations/20260218100000-create-kraken-day.down.sql

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

resources/migrations/20260218100000-create-kraken-day.up.sql

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

@@ -25,3 +25,20 @@ SET open = EXCLUDED.open,
 -- :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
+
+-- :name upsert-kraken-day! :! :n
+-- :doc Upsert a Kraken daily OHLC candle
+INSERT INTO kraken_day (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-day :? :1
+-- :doc Get the most recent Kraken daily candle by timestamp
+SELECT ts FROM kraken_day ORDER BY ts DESC LIMIT 1

resources/system.edn

@@ -59,4 +59,7 @@
   :uri      "wss://stream.binance.com:9443/ws/btcusdt@trade"}
 
  :kraken/ohlc
+ {:query-fn #ig/ref :db.sql/query-fn}
+
+ :kraken/ohlc-day
  {:query-fn #ig/ref :db.sql/query-fn}}

src/clj/pmagnus/btcprice/core.clj

@@ -17,7 +17,8 @@
    ;; WebSocket clients
    [pmagnus.btcprice.ws.binance]
    ;; Pollers
-   [pmagnus.btcprice.kraken.ohlc])
+   [pmagnus.btcprice.kraken.ohlc]
+   [pmagnus.btcprice.kraken.ohlc-daily])
   (:gen-class))
 
 (defonce system (atom nil))

src/clj/pmagnus/btcprice/kraken/ohlc_daily.clj

@@ -0,0 +1,123 @@
+(ns pmagnus.btcprice.kraken.ohlc-daily
+  (: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=1440")
+
+(defn- fetch-ohlc
+  "HTTP GET to Kraken OHLC endpoint (daily). 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_day table."
+  [query-fn candles]
+  (doseq [candle candles]
+    (query-fn :upsert-kraken-day! candle)))
+
+(defn- seed-since-from-db
+  "Query DB for the latest daily candle timestamp. Returns it or nil."
+  [query-fn]
+  (some-> (query-fn :get-latest-kraken-day {})
+          :ts))
+
+(defn- poll!
+  "Fetch daily 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)
+        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 daily candles"))
+    (count candles)))
+
+(defn- ms-until-next-daily-poll
+  "Milliseconds from now until next 00:01:00 UTC."
+  []
+  (let [now    (java.time.ZonedDateTime/now java.time.ZoneOffset/UTC)
+        next   (-> now
+                   (.truncatedTo java.time.temporal.ChronoUnit/DAYS)
+                   (.plusMinutes 1))
+        target (if (.isAfter now next)
+                 (.plusDays next 1)
+                 next)]
+    (.toMillis (java.time.Duration/between now target))))
+
+(defn- start-poll-loop!
+  "Start a background future that polls Kraken daily OHLC aligned to 00:01:00 UTC.
+   When `has-data?` is false, fetches immediately to backfill."
+  [client query-fn since-atom running? has-data?]
+  (future
+    (when-not has-data?
+      (try
+        (poll! client query-fn since-atom)
+        (catch Exception e
+          (log/error e "Kraken daily OHLC initial poll failed"))))
+    (while @running?
+      (let [wait (ms-until-next-daily-poll)]
+        (log/info "Next Kraken daily OHLC poll in" (quot wait 60000) "minutes")
+        (Thread/sleep wait)
+        (when @running?
+          (try
+            (poll! client query-fn since-atom)
+            (catch Exception e
+              (log/error e "Kraken daily OHLC poll failed"))))))))
+
+(defmethod ig/init-key :kraken/ohlc-day
+  [_ {:keys [query-fn]}]
+  (let [client    (HttpClient/newHttpClient)
+        db-since  (seed-since-from-db query-fn)
+        since     (atom db-since)
+        running?  (atom true)
+        has-data? (some? db-since)
+        fut       (start-poll-loop! client query-fn since running? has-data?)]
+    (if has-data?
+      (log/info "Starting Kraken daily OHLC poller, next fetch in" (quot (ms-until-next-daily-poll) 60000) "minutes")
+      (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]}]
+  (log/info "Stopping Kraken daily OHLC poller")
+  (reset! running? false)
+  (future-cancel future))