diff --git a/resources/migrations/20260218000000-create-kraken-hour.down.sql b/resources/migrations/20260218000000-create-kraken-hour.down.sql new file mode 100644 index 0000000..073213e --- /dev/null +++ b/resources/migrations/20260218000000-create-kraken-hour.down.sql @@ -0,0 +1 @@ +DROP TABLE kraken_hour; diff --git a/resources/migrations/20260218000000-create-kraken-hour.up.sql b/resources/migrations/20260218000000-create-kraken-hour.up.sql new file mode 100644 index 0000000..eb1ad69 --- /dev/null +++ b/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() +); diff --git a/resources/queries.sql b/resources/queries.sql index 98297fd..4c7a108 100644 --- a/resources/queries.sql +++ b/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 diff --git a/resources/system.edn b/resources/system.edn index 7475742..11673ab 100644 --- a/resources/system.edn +++ b/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}} diff --git a/src/clj/pmagnus/btcprice/core.clj b/src/clj/pmagnus/btcprice/core.clj index e312ec4..b3b5c7b 100644 --- a/src/clj/pmagnus/btcprice/core.clj +++ b/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)) diff --git a/src/clj/pmagnus/btcprice/kraken/ohlc.clj b/src/clj/pmagnus/btcprice/kraken/ohlc.clj new file mode 100644 index 0000000..b38620b --- /dev/null +++ b/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))