From b49e8afd59d00e76ad08a89d770215505ff18b86 Mon Sep 17 00:00:00 2001 From: Per Magnus Petersen Date: Fri, 13 Mar 2026 13:06:36 +0100 Subject: [PATCH] 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 --- ...260313000000-create-kraken-minute.down.sql | 1 + ...20260313000000-create-kraken-minute.up.sql | 11 ++ resources/queries.sql | 17 +++ resources/system.edn | 4 + src/clj/pmagnus/btcdata/core.clj | 1 + .../pmagnus/btcdata/kraken/ohlc_minute.clj | 107 ++++++++++++++++++ 6 files changed, 141 insertions(+) create mode 100644 resources/migrations/20260313000000-create-kraken-minute.down.sql create mode 100644 resources/migrations/20260313000000-create-kraken-minute.up.sql create mode 100644 src/clj/pmagnus/btcdata/kraken/ohlc_minute.clj diff --git a/resources/migrations/20260313000000-create-kraken-minute.down.sql b/resources/migrations/20260313000000-create-kraken-minute.down.sql new file mode 100644 index 0000000..dc93d5e --- /dev/null +++ b/resources/migrations/20260313000000-create-kraken-minute.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS kraken_minute; diff --git a/resources/migrations/20260313000000-create-kraken-minute.up.sql b/resources/migrations/20260313000000-create-kraken-minute.up.sql new file mode 100644 index 0000000..63adab2 --- /dev/null +++ b/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() +); diff --git a/resources/queries.sql b/resources/queries.sql index da916d3..1797e76 100644 --- a/resources/queries.sql +++ b/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, diff --git a/resources/system.edn b/resources/system.edn index 6f2070d..524269d 100644 --- a/resources/system.edn +++ b/resources/system.edn @@ -63,6 +63,10 @@ {: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"] :query-fn #ig/ref :db.sql/query-fn} diff --git a/src/clj/pmagnus/btcdata/core.clj b/src/clj/pmagnus/btcdata/core.clj index 8d42836..d4bdb8b 100644 --- a/src/clj/pmagnus/btcdata/core.clj +++ b/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) diff --git a/src/clj/pmagnus/btcdata/kraken/ohlc_minute.clj b/src/clj/pmagnus/btcdata/kraken/ohlc_minute.clj new file mode 100644 index 0000000..6004a75 --- /dev/null +++ b/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))