Fall back to DB for price when Binance websocket not connected

The /api/price/latest endpoint now reads the latest row from
binance_price table when the in-memory atom is empty on startup.

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

deps.edn

@@ -26,6 +26,8 @@
 
         ;; Serialization
         metosin/muuntaja                     {:mvn/version "0.6.11"}
+        metosin/jsonista                     {:mvn/version "0.3.12"}
+        com.fasterxml.jackson.datatype/jackson-datatype-jsr310 {:mvn/version "2.18.3"}
         luminus-transit/luminus-transit       {:mvn/version "0.1.6"}
 
         ;; Database

docker-compose.yml

@@ -19,6 +19,7 @@ services:
       CORS_ORIGIN: "${DOCKER_CORS_ORIGIN}"
       FRANKFURTER_URL: "${DOCKER_FRANKFURTER_URL}"
       STRIKE_API_KEY: "${STRIKE_API_KEY}"
+      KRAKEN_ENABLED: "true"
     extra_hosts:
       - "postgres:host-gateway"
 

resources/migrations/20260308100000-add-wallet-type.down.sql

@@ -0,0 +1 @@
+ALTER TABLE wallets DROP COLUMN wallet_type;

resources/migrations/20260308100000-add-wallet-type.up.sql

@@ -0,0 +1,4 @@
+ALTER TABLE wallets ADD COLUMN wallet_type TEXT NOT NULL DEFAULT 'exchange';
+--;;
+UPDATE wallets SET wallet_type = 'wallet'
+WHERE name IN ('Cold', 'Coldcard', 'Nunchuk Multi Sig');

resources/migrations/20260308200000-events-occurred-at-date.down.sql

@@ -0,0 +1,5 @@
+ALTER TABLE deposits ALTER COLUMN occurred_at TYPE TIMESTAMPTZ USING occurred_at::timestamptz;
+--;;
+ALTER TABLE events ALTER COLUMN occurred_at SET DEFAULT NOW();
+--;;
+ALTER TABLE events ALTER COLUMN occurred_at TYPE TIMESTAMPTZ USING occurred_at::timestamptz;

resources/migrations/20260308200000-events-occurred-at-date.up.sql

@@ -0,0 +1,5 @@
+ALTER TABLE events ALTER COLUMN occurred_at TYPE DATE USING occurred_at::date;
+--;;
+ALTER TABLE events ALTER COLUMN occurred_at DROP DEFAULT;
+--;;
+ALTER TABLE deposits ALTER COLUMN occurred_at TYPE DATE USING occurred_at::date;

resources/migrations/20260313000000-create-kraken-minute.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS kraken_minute;

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()
+);

resources/migrations/20260316000000-create-buys.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS buys;

resources/migrations/20260316000000-create-buys.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE buys (
+  id          BIGSERIAL PRIMARY KEY,
+  event_id    BIGINT NOT NULL REFERENCES events(id) UNIQUE,
+  occurred_at DATE NOT NULL,
+  wallet      TEXT NOT NULL,
+  amount_eur  NUMERIC(18,2) NOT NULL
+);

resources/migrations/20260316100000-buys-add-sats.down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE buys DROP COLUMN fee_sats;
+--;;
+ALTER TABLE buys DROP COLUMN sats;

resources/migrations/20260316100000-buys-add-sats.up.sql

@@ -0,0 +1,3 @@
+ALTER TABLE buys ADD COLUMN sats BIGINT NOT NULL DEFAULT 0;
+--;;
+ALTER TABLE buys ADD COLUMN fee_sats BIGINT;

resources/queries.sql

@@ -70,15 +70,19 @@ SELECT rate_date, eur, dkk, eur_dkk, updated_at FROM currencies WHERE rate_date
 
 -- :name insert-wallet! :! :n
 -- :doc Insert a new wallet
-INSERT INTO wallets (name) VALUES (:name)
+INSERT INTO wallets (name, wallet_type) VALUES (:name, :wallet-type)
 
 -- :name get-all-wallets :? :*
 -- :doc Get all wallets
-SELECT id, name, created_at FROM wallets ORDER BY name
+SELECT id, name, wallet_type, created_at FROM wallets ORDER BY name
+
+-- :name get-wallets-by-type :? :*
+-- :doc Get wallets filtered by type
+SELECT id, name, wallet_type, created_at FROM wallets WHERE wallet_type = :wallet-type ORDER BY name
 
 -- :name get-wallet-by-id :? :1
 -- :doc Get a wallet by ID
-SELECT id, name, created_at FROM wallets WHERE id = :id
+SELECT id, name, wallet_type, created_at FROM wallets WHERE id = :id
 
 -- Events --------------------------------------------------------------------
 
@@ -98,7 +102,7 @@ SELECT e.id, e.event_type, e.occurred_at, e.recorded_at,
 FROM events e
 LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
 LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
-ORDER BY e.occurred_at DESC
+ORDER BY e.occurred_at DESC, e.id DESC
 
 -- :name get-events-by-type :? :*
 -- :doc Get events filtered by type
@@ -111,7 +115,7 @@ FROM events e
 LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
 LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
 WHERE e.event_type = :event-type
-ORDER BY e.occurred_at DESC
+ORDER BY e.occurred_at DESC, e.id DESC
 
 -- :name get-events-by-wallet :? :*
 -- :doc Get events involving a specific wallet
@@ -124,7 +128,20 @@ FROM events e
 LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
 LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
 WHERE e.from_wallet_id = :wallet-id OR e.to_wallet_id = :wallet-id
-ORDER BY e.occurred_at DESC
+ORDER BY e.occurred_at DESC, e.id DESC
+
+-- :name truncate-deposits! :! :n
+-- :doc Delete all rows from the deposits read table
+TRUNCATE deposits
+
+-- :name get-deposit-events :? :*
+-- :doc Get bank_to_exchange events with exchange name, oldest first
+SELECT e.id, e.occurred_at, e.fiat_amount, e.fiat_currency,
+       e.to_wallet_id, tw.name AS exchange, e.note
+FROM events e
+LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
+WHERE e.event_type = 'bank_to_exchange'
+ORDER BY e.occurred_at ASC, e.id ASC
 
 -- :name insert-deposit! :! :n
 -- :doc Insert a projected deposit row
@@ -136,7 +153,58 @@ VALUES (:event-id, :occurred-at, :exchange, :fiat-amount, :fiat-currency, :amoun
 SELECT id, event_id, occurred_at, exchange, fiat_amount, fiat_currency,
        amount_eur, amount_dkk, amount_usd, note
 FROM deposits
-ORDER BY occurred_at DESC
+ORDER BY occurred_at DESC, id DESC
+
+-- :name truncate-buys! :! :n
+-- :doc Delete all rows from the buys read table
+TRUNCATE buys
+
+-- :name get-buy-events :? :*
+-- :doc Get exchange_to_wallet events with wallet name, oldest first
+SELECT e.id, e.occurred_at, e.sats, e.fee_sats, e.fiat_amount, e.fiat_currency,
+       e.to_wallet_id, tw.name AS wallet
+FROM events e
+LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
+WHERE e.event_type = 'exchange_to_wallet'
+ORDER BY e.occurred_at ASC, e.id ASC
+
+-- :name insert-buy! :! :n
+-- :doc Insert a projected buy row
+INSERT INTO buys (event_id, occurred_at, wallet, sats, fee_sats, amount_eur)
+VALUES (:event-id, :occurred-at, :wallet, :sats, :fee-sats, :amount-eur)
+
+-- :name get-all-buys :? :*
+-- :doc Get all projected buys ordered by date descending
+SELECT id, event_id, occurred_at, wallet, sats, fee_sats, amount_eur
+FROM buys
+ORDER BY occurred_at DESC, id DESC
+
+-- :name get-kraken-hour-latest-24 :? :*
+-- :doc Get the 24 most recent Kraken hourly candles
+SELECT ts, open, high, low, close, vwap, volume, trade_count, created_at
+FROM kraken_hour ORDER BY ts DESC LIMIT 24
+
+-- :name get-kraken-minute-latest-60 :? :*
+-- :doc Get the 60 most recent Kraken minute candles
+SELECT ts, open, high, low, close, vwap, volume, trade_count, created_at
+FROM kraken_minute ORDER BY ts DESC LIMIT 60
+
+-- :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

resources/system.edn

@@ -52,13 +52,20 @@
 
  :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"
+  :enabled? #or [#env BINANCE_ENABLED "true"]}
 
  :kraken/ohlc
- {:query-fn #ig/ref :db.sql/query-fn}
+ {:query-fn #ig/ref :db.sql/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
 
  :kraken/ohlc-day
- {:query-fn #ig/ref :db.sql/query-fn}
+ {: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"]
@@ -67,4 +74,5 @@
  :strike/ticker
  {:query-fn        #ig/ref :db.sql/query-fn
   :api-key         #env STRIKE_API_KEY
-  :sats-eur-amount #or [#env SATS_EUR_AMOUNT "55"]}}
+  :sats-eur-amount #or [#env SATS_EUR_AMOUNT "55"]
+  :enabled?        #or [#env STRIKE_ENABLED "true"]}}

src/clj/pmagnus/btcdata/core.clj

@@ -2,6 +2,7 @@
   (:require
    [clojure.tools.logging :as log]
    [integrant.core :as ig]
+   [next.jdbc.result-set]
    [pmagnus.btcdata.config :as config]
    [pmagnus.btcdata.env :refer [defaults]]
 
@@ -18,9 +19,25 @@
    ;; 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))
+  (:gen-class)
+  (:import
+   [java.sql Date Timestamp]))
+
+;; Read SQL DATE as LocalDate and TIMESTAMP as Instant (timezone-safe)
+(extend-protocol next.jdbc.result-set/ReadableColumn
+  Date
+  (read-column-by-label [v _]
+    (.toLocalDate v))
+  (read-column-by-index [v _ _]
+    (.toLocalDate v))
+  Timestamp
+  (read-column-by-label [v _]
+    (.toInstant v))
+  (read-column-by-index [v _ _]
+    (.toInstant v)))
 
 (defonce system (atom nil))
 

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

@@ -54,68 +54,53 @@
           :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]}]
+  [_ {:keys [query-fn enabled?]}]
-  (let [client    (HttpClient/newHttpClient)
+  (if (= enabled? "false")
-        db-since  (seed-since-from-db query-fn)
+    (do (log/info "Kraken OHLC poller disabled")
-        since     (atom db-since)
+        {:running? (atom false)})
-        running?  (atom true)
+    (let [client    (HttpClient/newHttpClient)
-        has-data? (some? db-since)
+          db-since  (seed-since-from-db query-fn)
-        fut       (start-poll-loop! client query-fn since running? has-data?)]
+          since     (atom db-since)
-    (if has-data?
+          running?  (atom true)
-      (log/info "Starting Kraken OHLC poller, next fetch in" (quot (ms-until-next-poll) 60000) "minutes")
+          fut       (start-poll-loop! client query-fn since running?)]
-      (log/info "Starting Kraken OHLC poller, fetching immediately (no data)"))
+      (log/info "Starting Kraken OHLC poller (60s interval)")
-    {:running? running?
+      {:running? running?
-     :future   fut
+       :future   fut
-     :since    since}))
+       :since    since})))
 
 (defmethod ig/halt-key! :kraken/ohlc
   [_ {:keys [running? future]}]

src/clj/pmagnus/btcdata/kraken/ohlc_daily.clj

@@ -102,19 +102,22 @@
               (log/error e "Kraken daily OHLC poll failed"))))))))
 
 (defmethod ig/init-key :kraken/ohlc-day
-  [_ {:keys [query-fn]}]
+  [_ {:keys [query-fn enabled?]}]
-  (let [client    (HttpClient/newHttpClient)
+  (if (= enabled? "false")
-        db-since  (seed-since-from-db query-fn)
+    (do (log/info "Kraken daily OHLC poller disabled")
-        since     (atom db-since)
+        {:running? (atom false)})
-        running?  (atom true)
+    (let [client    (HttpClient/newHttpClient)
-        has-data? (some? db-since)
+          db-since  (seed-since-from-db query-fn)
-        fut       (start-poll-loop! client query-fn since running? has-data?)]
+          since     (atom db-since)
-    (if has-data?
+          running?  (atom true)
-      (log/info "Starting Kraken daily OHLC poller, next fetch in" (quot (ms-until-next-daily-poll) 60000) "minutes")
+          has-data? (some? db-since)
-      (log/info "Starting Kraken daily OHLC poller, fetching immediately (no data)"))
+          fut       (start-poll-loop! client query-fn since running? has-data?)]
-    {:running? running?
+      (if has-data?
-     :future   fut
+        (log/info "Starting Kraken daily OHLC poller, next fetch in" (quot (ms-until-next-daily-poll) 60000) "minutes")
-     :since    since}))
+        (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]}]

src/clj/pmagnus/btcdata/kraken/ohlc_minute.clj

@@ -0,0 +1,114 @@
+(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]
+   [java.time Instant ZoneOffset]
+   [java.time.format DateTimeFormatter]))
+
+(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.
+   On initial fetch saves everything; subsequent fetches only take last 5.
+   Returns the count of saved candles."
+  [client query-fn since-atom initial?]
+  (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 (if initial? raw (take-last 5 raw)))]
+    (when (seq candles)
+      (save-candles! query-fn candles)
+      (when last-ts
+        (reset! since-atom last-ts))
+      (let [fmt    (DateTimeFormatter/ofPattern "HH:mm")
+            c      (last candles)
+            ts-str (.format (.atOffset (Instant/ofEpochSecond (:ts c)) ZoneOffset/UTC) fmt)]
+        (log/info "Kraken minute:" (count candles) "candles, latest" ts-str "UTC"
+                  "O" (str (:open c)) "H" (str (:high c)) "L" (str (:low c)) "C" (str (:close c)))))
+    (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 true)
+      (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 false)
+          (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))

src/clj/pmagnus/btcdata/strike/ticker.clj

@@ -136,19 +136,24 @@
             (log/error e "Strike poll failed")))))))
 
 (defmethod ig/init-key :strike/ticker
-  [_ {:keys [query-fn api-key sats-eur-amount]}]
+  [_ {:keys [query-fn api-key sats-eur-amount enabled?] :or {enabled? "true"}}]
-  (log/info "Starting Strike ticker poller")
+  (if-not (= enabled? "true")
-  (let [client       (HttpClient/newHttpClient)
+    (do (log/info "Strike ticker disabled")
-        running?     (atom true)
+        {:running?     (atom false)
-        latest-rates (atom nil)
+         :future       nil
-        fee-cache    (atom nil)
+         :latest-rates (atom nil)})
-        fut          (start-poll-loop! client api-key query-fn running? latest-rates sats-eur-amount fee-cache)]
+    (do (log/info "Starting Strike ticker poller")
-    {:running?     running?
+        (let [client       (HttpClient/newHttpClient)
-     :future       fut
+              running?     (atom true)
-     :latest-rates latest-rates}))
+              latest-rates (atom nil)
+              fee-cache    (atom nil)
+              fut          (start-poll-loop! client api-key query-fn running? latest-rates sats-eur-amount fee-cache)]
+          {:running?     running?
+           :future       fut
+           :latest-rates latest-rates}))))
 
 (defmethod ig/halt-key! :strike/ticker
   [_ {:keys [running? future]}]
   (log/info "Stopping Strike ticker poller")
   (reset! running? false)
-  (future-cancel future))
+  (when future (future-cancel future)))

src/clj/pmagnus/btcdata/web/controllers/transactions.clj

@@ -5,17 +5,20 @@
    [pmagnus.btcdata.frankfurter.rates :as rates]
    [ring.util.http-response :as response])
   (:import
-   [java.time Instant LocalDate ZoneOffset]
+   [java.time LocalDate]
    [java.util UUID]))
 
-(defn list-wallets [{:keys [query-fn]} _req]
+(defn list-wallets [{:keys [query-fn]} req]
-  (response/ok (query-fn :get-all-wallets {})))
+  (let [wallet-type (get-in req [:query-params "type"])]
+    (if wallet-type
+      (response/ok (query-fn :get-wallets-by-type {:wallet-type wallet-type}))
+      (response/ok (query-fn :get-all-wallets {})))))
 
 (defn create-wallet! [{:keys [query-fn]} req]
-  (let [{:keys [name]} (:body-params req)]
+  (let [{:keys [name wallet_type]} (:body-params req)]
     (if (str/blank? name)
       (response/bad-request {:error "name is required"})
-      (do (query-fn :insert-wallet! {:name name})
+      (do (query-fn :insert-wallet! {:name name :wallet-type (or wallet_type "exchange")})
           (response/created "/api/wallets" {:name name})))))
 
 (defn get-wallet-balances [{:keys [query-fn]} _req]
@@ -48,11 +51,10 @@
   (try
     (let [wallet   (query-fn :get-wallet-by-id {:id to-wallet-id})
           exchange (or (:name wallet) "Unknown")
-          date     (.toLocalDate (.atOffset occurred-at ZoneOffset/UTC))
+          rates    (or (query-fn :get-currency-rates-by-date {:rate-date occurred-at})
-          rates    (or (query-fn :get-currency-rates-by-date {:rate-date date})
                        (rates/fetch-and-persist-for-date!
                          (:client frankfurter) (:url frankfurter)
-                         query-fn date))
+                         query-fn occurred-at))
           amounts  (if rates
                      (convert-amount fiat-amount fiat-currency rates)
                      {:amount-eur nil :amount-dkk nil :amount-usd nil})]
@@ -67,14 +69,37 @@
     (catch Exception e
       (log/error e "Failed to project deposit for event" event-id))))
 
+(defn- project-buy!
+  "Project an exchange_to_wallet event into the buys read table."
+  [{:keys [query-fn frankfurter]} event-id occurred-at to-wallet-id sats fee-sats fiat-amount fiat-currency]
+  (try
+    (let [wallet-name (or (:name (query-fn :get-wallet-by-id {:id to-wallet-id})) "Unknown")
+          amount-eur  (if (= fiat-currency "EUR")
+                        (bigdec fiat-amount)
+                        (let [rates (or (query-fn :get-currency-rates-by-date {:rate-date occurred-at})
+                                        (rates/fetch-and-persist-for-date!
+                                          (:client frankfurter) (:url frankfurter)
+                                          query-fn occurred-at))]
+                          (:amount-eur (convert-amount fiat-amount fiat-currency rates))))]
+      (when amount-eur
+        (query-fn :insert-buy!
+                  {:event-id    event-id
+                   :occurred-at occurred-at
+                   :wallet      wallet-name
+                   :sats        (or sats 0)
+                   :fee-sats    fee-sats
+                   :amount-eur  amount-eur})))
+    (catch Exception e
+      (log/error e "Failed to project buy for event" event-id))))
+
 (defn create-event! [{:keys [query-fn] :as opts} req]
   (let [params (:body-params req)
         event-type (:event_type params)]
     (if-not (#{"bank_to_exchange" "exchange_to_wallet" "wallet_to_wallet"} event-type)
       (response/bad-request {:error "Invalid event_type"})
-      (let [occurred-at (if-let [ts (:occurred_at params)]
+      (let [occurred-at (if-let [d (:occurred_at params)]
-                          (Instant/parse ts)
+                          (LocalDate/parse d)
-                          (Instant/now))
+                          (LocalDate/now))
             row {:event-type     event-type
                  :occurred-at    occurred-at
                  :sats           (:sats params)
@@ -91,11 +116,82 @@
                             (:fiat_amount params)
                             (:fiat_currency params)
                             (:note params)))
+        (when (and (= event-type "exchange_to_wallet") (:fiat_amount params))
+          (project-buy! opts (:id result) occurred-at
+                         (some-> (:to_wallet_id params) UUID/fromString)
+                         (:sats params)
+                         (:fee_sats params)
+                         (:fiat_amount params)
+                         (:fiat_currency params)))
         (response/created "/api/events" {:status "ok"})))))
 
 (defn list-deposits [{:keys [query-fn]} _req]
   (response/ok (query-fn :get-all-deposits {})))
 
+(defn rebuild-deposits! [{:keys [query-fn frankfurter]} _req]
+  (query-fn :truncate-deposits! {})
+  (let [events (query-fn :get-deposit-events {})
+        n      (reduce
+                 (fn [cnt {:keys [id occurred_at fiat_amount fiat_currency exchange note]}]
+                   (if-not fiat_amount
+                     cnt
+                     (try
+                       (let [rates   (or (query-fn :get-currency-rates-by-date {:rate-date occurred_at})
+                                         (rates/fetch-and-persist-for-date!
+                                           (:client frankfurter) (:url frankfurter)
+                                           query-fn occurred_at))
+                             amounts (if rates
+                                       (convert-amount fiat_amount fiat_currency rates)
+                                       {:amount-eur nil :amount-dkk nil :amount-usd nil})]
+                         (query-fn :insert-deposit!
+                                   (merge {:event-id      id
+                                           :occurred-at   occurred_at
+                                           :exchange      (or exchange "Unknown")
+                                           :fiat-amount   fiat_amount
+                                           :fiat-currency fiat_currency
+                                           :note          note}
+                                          amounts))
+                         (inc cnt))
+                       (catch Exception e
+                         (log/error e "Failed to rebuild deposit for event" id)
+                         cnt))))
+                 0 events)]
+    (response/ok {:rebuilt n})))
+
+(defn list-buys [{:keys [query-fn]} _req]
+  (response/ok (query-fn :get-all-buys {})))
+
+(defn rebuild-buys! [{:keys [query-fn frankfurter]} _req]
+  (query-fn :truncate-buys! {})
+  (let [events (query-fn :get-buy-events {})
+        n      (reduce
+                 (fn [cnt {:keys [id occurred_at sats fee_sats fiat_amount fiat_currency wallet]}]
+                   (if-not fiat_amount
+                     cnt
+                     (try
+                       (let [amount-eur
+                             (if (= fiat_currency "EUR")
+                               (bigdec fiat_amount)
+                               (let [rates (or (query-fn :get-currency-rates-by-date {:rate-date occurred_at})
+                                               (rates/fetch-and-persist-for-date!
+                                                 (:client frankfurter) (:url frankfurter)
+                                                 query-fn occurred_at))]
+                                 (:amount-eur (convert-amount fiat_amount fiat_currency rates))))]
+                         (when amount-eur
+                           (query-fn :insert-buy!
+                                     {:event-id    id
+                                      :occurred-at occurred_at
+                                      :wallet      (or wallet "Unknown")
+                                      :sats        (or sats 0)
+                                      :fee-sats    fee_sats
+                                      :amount-eur  amount-eur}))
+                         (inc cnt))
+                       (catch Exception e
+                         (log/error e "Failed to rebuild buy for event" id)
+                         cnt))))
+                 0 events)]
+    (response/ok {:rebuilt n})))
+
 (defn list-events [{:keys [query-fn]} req]
   (let [event-type (get-in req [:query-params "type"])]
     (if event-type

src/clj/pmagnus/btcdata/web/middleware/formats.clj

@@ -1,11 +1,23 @@
 (ns pmagnus.btcdata.web.middleware.formats
   (:require
+   [jsonista.core :as j]
    [luminus-transit.time :as time]
-   [muuntaja.core :as m]))
+   [muuntaja.core :as m])
+  (:import
+   [com.fasterxml.jackson.databind SerializationFeature]
+   [com.fasterxml.jackson.datatype.jsr310 JavaTimeModule]))
+
+(def ^:private mapper
+  (j/object-mapper
+   {:modules [(JavaTimeModule.)]
+    :decode-key-fn true
+    :configure {SerializationFeature/WRITE_DATES_AS_TIMESTAMPS false}}))
 
 (def instance
   (m/create
    (-> m/default-options
+       (assoc-in [:formats "application/json" :decoder-opts] {:mapper mapper})
+       (assoc-in [:formats "application/json" :encoder-opts] {:mapper mapper})
        (update-in [:formats "application/transit+json" :decoder-opts]
                   (partial merge time/time-deserialization-handlers))
        (update-in [:formats "application/transit+json" :encoder-opts]

src/clj/pmagnus/btcdata/web/routes/api.clj

@@ -87,10 +87,14 @@
      {:on-open (fn [{:keys [^WebSocketChannel channel]}]
                  (start-strike-ws-loop! channel strike-atom))}}))
 
-(defn- latest-price-handler [{:keys [binance frankfurter]} _req]
+(defn- latest-price-handler [{:keys [binance frankfurter query-fn]} _req]
   (let [price-atom (:latest-price binance)
         rates-atom (:rates frankfurter)
-        v          @price-atom]
+        v          (or @price-atom
+                       (when-let [row (query-fn :get-latest-binance-price {})]
+                         {:price       (:price row)
+                          :prev-price  (:price row)
+                          :recorded-at (:recorded_at row)}))]
     (if v
       {:status  200
        :headers {"Content-Type" "application/json"}
@@ -144,6 +148,31 @@
          :headers {"Content-Type" "application/json"}
          :body    (json/write-str {:error "Invalid date format, use YYYY-MM-DD"})}))))
 
+(def ^:private ohlc-fmt
+  (java.time.format.DateTimeFormatter/ofPattern "yyyy-MM-dd HH:mm:ss"))
+
+(defn- ohlc-row->map [r]
+  {:ts          (:ts r)
+   :datetime    (.format (.atOffset (java.time.Instant/ofEpochSecond (:ts r))
+                                    java.time.ZoneOffset/UTC) ohlc-fmt)
+   :open        (str (:open r))
+   :high        (str (:high r))
+   :low         (str (:low r))
+   :close       (str (:close r))
+   :vwap        (str (:vwap r))
+   :volume      (str (:volume r))
+   :trade_count (:trade_count r)})
+
+(defn- kraken-hour-handler [{:keys [query-fn]} _req]
+  {:status  200
+   :headers {"Content-Type" "application/json"}
+   :body    (json/write-str (mapv ohlc-row->map (query-fn :get-kraken-hour-latest-24 {})))})
+
+(defn- kraken-minute-handler [{:keys [query-fn]} _req]
+  {:status  200
+   :headers {"Content-Type" "application/json"}
+   :body    (json/write-str (mapv ohlc-row->map (query-fn :get-kraken-minute-latest-60 {})))})
+
 (defn- api-routes [opts]
   [["/swagger.json"
     {:get {:no-doc  true
@@ -174,7 +203,17 @@
     {:get  (fn [req] (tx/list-events opts req))
      :post (fn [req] (tx/create-event! opts req))}]
    ["/deposits"
-    {:get (fn [req] (tx/list-deposits opts req))}]])
+    {:get (fn [req] (tx/list-deposits opts req))}]
+   ["/deposits/rebuild"
+    {:post (fn [req] (tx/rebuild-deposits! opts req))}]
+   ["/buys"
+    {:get (fn [req] (tx/list-buys opts req))}]
+   ["/buys/rebuild"
+    {:post (fn [req] (tx/rebuild-buys! opts req))}]
+   ["/kraken-hour"
+    {:get (fn [req] (kraken-hour-handler opts req))}]
+   ["/kraken-minute"
+    {:get (fn [req] (kraken-minute-handler opts req))}]])
 
 (defn route-data [opts]
   (merge

src/clj/pmagnus/btcdata/ws/binance.clj

@@ -71,17 +71,21 @@
         (.join))))
 
 (defmethod ig/init-key :ws/binance
-  [_ {:keys [query-fn uri]}]
+  [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}]
-  (log/info "Starting Binance WebSocket listener:" uri)
+  (if-not (= enabled? "true")
-  (let [latest-price (atom nil)
+    (do (log/info "Binance WebSocket disabled")
-        state        (atom {:running?   true
+        {:state        (atom {:running? false})
-                            :last-write 0
+         :latest-price (atom nil)})
-                            :buffer     (StringBuilder.)
+    (do (log/info "Starting Binance WebSocket listener:" uri)
-                            :ws         nil})
+        (let [latest-price (atom nil)
-        ws           (connect! uri query-fn state latest-price)]
+              state        (atom {:running?   true
-    (swap! state assoc :ws ws)
+                                  :last-write 0
-    {:state        state
+                                  :buffer     (StringBuilder.)
-     :latest-price latest-price}))
+                                  :ws         nil})
+              ws           (connect! uri query-fn state latest-price)]
+          (swap! state assoc :ws ws)
+          {:state        state
+           :latest-price latest-price}))))
 
 (defmethod ig/halt-key! :ws/binance
   [_ {:keys [state]}]