diff --git a/src/clj/pmagnus/btcdata/kraken/ohlc.clj b/src/clj/pmagnus/btcdata/kraken/ohlc.clj index c698508..3003561 100644 --- a/src/clj/pmagnus/btcdata/kraken/ohlc.clj +++ b/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. - When `has-data?` is false, fetches immediately to backfill." - [client query-fn since-atom running? has-data?] + "Start a background future that polls Kraken every 60 seconds." + [client query-fn since-atom running?] (future - (when-not has-data? - (try - (poll! client query-fn since-atom) - (catch Exception e - (log/error e "Kraken OHLC initial poll failed")))) + (try + (poll! client query-fn since-atom true) + (catch Exception e + (log/error e "Kraken OHLC initial poll failed"))) (while @running? - (let [wait (ms-until-next-poll)] - (log/info "Next Kraken 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 OHLC poll failed")))))))) + (Thread/sleep 60000) + (when @running? + (try + (poll! client query-fn since-atom false) + (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? has-data?)] - (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)")) + fut (start-poll-loop! client query-fn since running?)] + (log/info "Starting Kraken OHLC poller (60s interval)") {:running? running? :future fut :since since})))