Compare commits

...
4 Commits
Author SHA1 Message Date
magnusandClaude Opus 4.6 8b5c1faeaa Add deposits rebuild endpoint and fix date serialization
Add POST /api/deposits/rebuild to re-project all bank_to_exchange events.
Fix date-off-by-one by converting java.sql.Date to LocalDate via
ReadableColumn and serializing with Jackson JavaTimeModule.
Sort all event/deposit queries by occurred_at, id.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 17:30:43 +01:00
magnusandClaude Opus 4.6 96d12f170b Change events.occurred_at from TIMESTAMPTZ to DATE
Make date required, remove default. Simplifies date handling
throughout the controller.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 16:04:24 +01:00
magnusandClaude Opus 4.6 130815e32f Add BINANCE_ENABLED and STRIKE_ENABLED env flags
Skip live WebSocket/API connections when set to false.
Defaults to true so Docker keeps fetching.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 15:54:50 +01:00
magnusandClaude Opus 4.6 793c63927c Add wallet_type column to distinguish exchanges from wallets
Exchanges (can receive bank deposits) vs wallets (cold storage only).
Supports ?type= filter on GET /api/wallets.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-08 15:48:41 +01:00
13 changed files with 152 additions and 45 deletions
+2
View File
@@ -26,6 +26,8 @@
;; Serialization ;; Serialization
metosin/muuntaja {:mvn/version "0.6.11"} 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"} luminus-transit/luminus-transit {:mvn/version "0.1.6"}
;; Database ;; Database
@@ -0,0 +1 @@
ALTER TABLE wallets DROP COLUMN wallet_type;
@@ -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');
@@ -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;
@@ -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;
+24 -7
View File
@@ -70,15 +70,19 @@ SELECT rate_date, eur, dkk, eur_dkk, updated_at FROM currencies WHERE rate_date
-- :name insert-wallet! :! :n -- :name insert-wallet! :! :n
-- :doc Insert a new wallet -- :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 :? :* -- :name get-all-wallets :? :*
-- :doc 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 -- :name get-wallet-by-id :? :1
-- :doc Get a wallet by ID -- :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 -------------------------------------------------------------------- -- Events --------------------------------------------------------------------
@@ -98,7 +102,7 @@ SELECT e.id, e.event_type, e.occurred_at, e.recorded_at,
FROM events e FROM events e
LEFT JOIN wallets fw ON fw.id = e.from_wallet_id LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
LEFT JOIN wallets tw ON tw.id = e.to_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 :? :* -- :name get-events-by-type :? :*
-- :doc Get events filtered 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 fw ON fw.id = e.from_wallet_id
LEFT JOIN wallets tw ON tw.id = e.to_wallet_id LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
WHERE e.event_type = :event-type 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 :? :* -- :name get-events-by-wallet :? :*
-- :doc Get events involving a specific 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 fw ON fw.id = e.from_wallet_id
LEFT JOIN wallets tw ON tw.id = e.to_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 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 -- :name insert-deposit! :! :n
-- :doc Insert a projected deposit row -- :doc Insert a projected deposit row
@@ -136,7 +153,7 @@ VALUES (:event-id, :occurred-at, :exchange, :fiat-amount, :fiat-currency, :amoun
SELECT id, event_id, occurred_at, exchange, fiat_amount, fiat_currency, SELECT id, event_id, occurred_at, exchange, fiat_amount, fiat_currency,
amount_eur, amount_dkk, amount_usd, note amount_eur, amount_dkk, amount_usd, note
FROM deposits FROM deposits
ORDER BY occurred_at DESC ORDER BY occurred_at DESC, id DESC
-- :name get-wallet-balances :? :* -- :name get-wallet-balances :? :*
-- :doc Compute sats balance per wallet from events -- :doc Compute sats balance per wallet from events
+4 -2
View File
@@ -52,7 +52,8 @@
:ws/binance :ws/binance
{:query-fn #ig/ref :db.sql/query-fn {: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 :kraken/ohlc
{:query-fn #ig/ref :db.sql/query-fn} {:query-fn #ig/ref :db.sql/query-fn}
@@ -67,4 +68,5 @@
:strike/ticker :strike/ticker
{:query-fn #ig/ref :db.sql/query-fn {:query-fn #ig/ref :db.sql/query-fn
:api-key #env STRIKE_API_KEY :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"]}}
+17 -1
View File
@@ -2,6 +2,7 @@
(:require (:require
[clojure.tools.logging :as log] [clojure.tools.logging :as log]
[integrant.core :as ig] [integrant.core :as ig]
[next.jdbc.result-set]
[pmagnus.btcdata.config :as config] [pmagnus.btcdata.config :as config]
[pmagnus.btcdata.env :refer [defaults]] [pmagnus.btcdata.env :refer [defaults]]
@@ -20,7 +21,22 @@
[pmagnus.btcdata.kraken.ohlc-daily] [pmagnus.btcdata.kraken.ohlc-daily]
[pmagnus.btcdata.frankfurter.rates] [pmagnus.btcdata.frankfurter.rates]
[pmagnus.btcdata.strike.ticker]) [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)) (defonce system (atom nil))
+16 -11
View File
@@ -136,19 +136,24 @@
(log/error e "Strike poll failed"))))))) (log/error e "Strike poll failed")))))))
(defmethod ig/init-key :strike/ticker (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 (defmethod ig/halt-key! :strike/ticker
[_ {:keys [running? future]}] [_ {:keys [running? future]}]
(log/info "Stopping Strike ticker poller") (log/info "Stopping Strike ticker poller")
(reset! running? false) (reset! running? false)
(future-cancel future)) (when future (future-cancel future)))
@@ -5,17 +5,20 @@
[pmagnus.btcdata.frankfurter.rates :as rates] [pmagnus.btcdata.frankfurter.rates :as rates]
[ring.util.http-response :as response]) [ring.util.http-response :as response])
(:import (:import
[java.time Instant LocalDate ZoneOffset] [java.time LocalDate]
[java.util UUID])) [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] (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) (if (str/blank? name)
(response/bad-request {:error "name is required"}) (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}))))) (response/created "/api/wallets" {:name name})))))
(defn get-wallet-balances [{:keys [query-fn]} _req] (defn get-wallet-balances [{:keys [query-fn]} _req]
@@ -48,11 +51,10 @@
(try (try
(let [wallet (query-fn :get-wallet-by-id {:id to-wallet-id}) (let [wallet (query-fn :get-wallet-by-id {:id to-wallet-id})
exchange (or (:name wallet) "Unknown") 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! (rates/fetch-and-persist-for-date!
(:client frankfurter) (:url frankfurter) (:client frankfurter) (:url frankfurter)
query-fn date)) query-fn occurred-at))
amounts (if rates amounts (if rates
(convert-amount fiat-amount fiat-currency rates) (convert-amount fiat-amount fiat-currency rates)
{:amount-eur nil :amount-dkk nil :amount-usd nil})] {:amount-eur nil :amount-dkk nil :amount-usd nil})]
@@ -72,9 +74,9 @@
event-type (:event_type params)] event-type (:event_type params)]
(if-not (#{"bank_to_exchange" "exchange_to_wallet" "wallet_to_wallet"} event-type) (if-not (#{"bank_to_exchange" "exchange_to_wallet" "wallet_to_wallet"} event-type)
(response/bad-request {:error "Invalid 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 row {:event-type event-type
:occurred-at occurred-at :occurred-at occurred-at
:sats (:sats params) :sats (:sats params)
@@ -96,6 +98,36 @@
(defn list-deposits [{:keys [query-fn]} _req] (defn list-deposits [{:keys [query-fn]} _req]
(response/ok (query-fn :get-all-deposits {}))) (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-events [{:keys [query-fn]} req] (defn list-events [{:keys [query-fn]} req]
(let [event-type (get-in req [:query-params "type"])] (let [event-type (get-in req [:query-params "type"])]
(if event-type (if event-type
@@ -1,11 +1,23 @@
(ns pmagnus.btcdata.web.middleware.formats (ns pmagnus.btcdata.web.middleware.formats
(:require (:require
[jsonista.core :as j]
[luminus-transit.time :as time] [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 (def instance
(m/create (m/create
(-> m/default-options (-> 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] (update-in [:formats "application/transit+json" :decoder-opts]
(partial merge time/time-deserialization-handlers)) (partial merge time/time-deserialization-handlers))
(update-in [:formats "application/transit+json" :encoder-opts] (update-in [:formats "application/transit+json" :encoder-opts]
+3 -1
View File
@@ -174,7 +174,9 @@
{:get (fn [req] (tx/list-events opts req)) {:get (fn [req] (tx/list-events opts req))
:post (fn [req] (tx/create-event! opts req))}] :post (fn [req] (tx/create-event! opts req))}]
["/deposits" ["/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))}]])
(defn route-data [opts] (defn route-data [opts]
(merge (merge
+15 -11
View File
@@ -71,17 +71,21 @@
(.join)))) (.join))))
(defmethod ig/init-key :ws/binance (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 (defmethod ig/halt-key! :ws/binance
[_ {:keys [state]}] [_ {:keys [state]}]