diff --git a/resources/migrations/20260218100000-create-kraken-day.down.sql b/resources/migrations/20260218100000-create-kraken-day.down.sql new file mode 100644 index 0000000..a8f05eb --- /dev/null +++ b/resources/migrations/20260218100000-create-kraken-day.down.sql @@ -0,0 +1 @@ +DROP TABLE kraken_day; diff --git a/resources/migrations/20260218100000-create-kraken-day.up.sql b/resources/migrations/20260218100000-create-kraken-day.up.sql new file mode 100644 index 0000000..b694557 --- /dev/null +++ b/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() +); diff --git a/resources/queries.sql b/resources/queries.sql index 4c7a108..7c7a9ba 100644 --- a/resources/queries.sql +++ b/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 diff --git a/resources/system.edn b/resources/system.edn index 1c17688..a662a2c 100644 --- a/resources/system.edn +++ b/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}} diff --git a/src/clj/pmagnus/btcprice/core.clj b/src/clj/pmagnus/btcprice/core.clj index b3b5c7b..6c3a6bb 100644 --- a/src/clj/pmagnus/btcprice/core.clj +++ b/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)) diff --git a/src/clj/pmagnus/btcprice/kraken/ohlc_daily.clj b/src/clj/pmagnus/btcprice/kraken/ohlc_daily.clj new file mode 100644 index 0000000..632d05a --- /dev/null +++ b/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))