Change Kraken hourly poller to 60s interval, fetch last 5 candles

Keeps the in-progress hour candle fresh instead of only updating at
the top of each hour.

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

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

@@ -54,53 +54,38 @@
           :ts))
 
 (defn- poll!
-  "Fetch OHLC data, drop the last (in-progress) candle, save completed ones.
+  "Fetch OHLC data, upsert candles. On initial fetch saves all completed candles
+   (drops last in-progress). On subsequent fetches saves last 5 including in-progress.
    Returns the count of saved candles."
-  [client query-fn since-atom]
+  [client query-fn since-atom initial?]
   (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))]
+        candles  (map parse-candle (if initial? (butlast raw) (take-last 5 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"))
+      (log/info "Fetched" (count candles) "Kraken hourly candles"))
     (count candles)))
 
-(defn- ms-until-next-poll
-  "Milliseconds from now until next HH:01:00 UTC."
-  []
-  (let [now    (java.time.ZonedDateTime/now java.time.ZoneOffset/UTC)
-        next   (-> now
-                   (.truncatedTo java.time.temporal.ChronoUnit/HOURS)
-                   (.plusMinutes 1))
-        target (if (.isAfter now next)
-                 (.plusHours next 1)
-                 next)]
-    (.toMillis (java.time.Duration/between now target))))
-
 (defn- start-poll-loop!
-  "Start a background future that polls Kraken aligned to HH:01:00 UTC.
+  "Start a background future that polls Kraken every 60 seconds."
-   When `has-data?` is false, fetches immediately to backfill."
+  [client query-fn since-atom running?]
-  [client query-fn since-atom running? has-data?]
   (future
-    (when-not has-data?
+    (try
-      (try
+      (poll! client query-fn since-atom true)
-        (poll! client query-fn since-atom)
+      (catch Exception e
-        (catch Exception e
+        (log/error e "Kraken OHLC initial poll failed")))
-          (log/error e "Kraken OHLC initial poll failed"))))
     (while @running?
-      (let [wait (ms-until-next-poll)]
+      (Thread/sleep 60000)
-        (log/info "Next Kraken OHLC poll in" (quot wait 60000) "minutes")
+      (when @running?
-        (Thread/sleep wait)
+        (try
-        (when @running?
+          (poll! client query-fn since-atom false)
-          (try
+          (catch Exception e
-            (poll! client query-fn since-atom)
+            (log/error e "Kraken OHLC poll failed")))))))
-            (catch Exception e
-              (log/error e "Kraken OHLC poll failed"))))))))
 
 (defmethod ig/init-key :kraken/ohlc
   [_ {:keys [query-fn enabled?]}]
@@ -111,11 +96,8 @@
           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?)]
-          fut       (start-poll-loop! client query-fn since running? has-data?)]
+      (log/info "Starting Kraken OHLC poller (60s interval)")
-      (if has-data?
-        (log/info "Starting Kraken OHLC poller, next fetch in" (quot (ms-until-next-poll) 60000) "minutes")
-        (log/info "Starting Kraken OHLC poller, fetching immediately (no data)"))
       {:running? running?
        :future   fut
        :since    since})))