Merge pull request 'Add daily BTC/USD endpoint with historical caching and Swagger UI' (#2) from day-history into master

Reviewed-on: #2

README.md

@@ -9,6 +9,33 @@ Bitcoin data backend service. Fetches BTC prices from Binance (WebSocket) and Kr
 | `GET /api/price/ws` | WebSocket — sends HTML fragments with live price updates (HTMX-compatible) |
 | `GET /api/price/latest` | Latest price as JSON |
 | `GET /api/health` | Health check |
+| `GET /api/price/day/:date` | Completed UTC BTC/USD daily OHLC, e.g. `/api/price/day/2020-01-01` |
+| `GET /api/price/day?date=YYYY-MM-DD` | Same daily OHLC endpoint using a query parameter |
+| `GET /api/swagger.json` | Swagger specification |
+| `GET /swagger` | Interactive Swagger UI |
+
+### Daily OHLC
+
+Dates must be valid `YYYY-MM-DD` values strictly before today in UTC. Both
+`kraken_day` and `kraken_day_history` are checked before contacting Kraken.
+Missing recent candles are fetched from the OHLC API and saved in `kraken_day`.
+Older days outside Kraken's 720-candle window are calculated from paginated
+historical trades and saved in `kraken_day_history` for subsequent requests.
+Only trades in `[00:00 UTC, next midnight UTC)` contribute; VWAP is volume-weighted.
+
+Example response (prices and volume are JSON numbers):
+
+```json
+{"date":"2020-01-01","pair":"BTC/USD","open":7200,"high":7300,"low":7100,"close":7250,"vwap":7230.5,"volume":123.45,"trade_count":1000}
+```
+
+Errors are JSON: 400 for invalid, current or future dates, 404 for days without
+trades, and 502 for Kraken failures. Failed calculations are not cached.
+Historical requests wait for completion and can take several minutes; clients
+and reverse proxies need sufficiently long request timeouts. Kraken requests
+have a 60-second per-page timeout and historical pages are spaced one second apart.
+
+The `kraken_day_history` migration runs automatically on application startup.
 
 ## Quick Start
 

resources/migrations/20260317000000-create-kraken-day-history.down.sql

@@ -0,0 +1 @@
+DROP TABLE kraken_day_history;

resources/migrations/20260317000000-create-kraken-day-history.up.sql

@@ -0,0 +1,11 @@
+CREATE TABLE kraken_day_history (
+  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/queries.sql

@@ -47,6 +47,26 @@ SET open = EXCLUDED.open,
 -- :doc Get the most recent Kraken daily candle by timestamp
 SELECT ts FROM kraken_day ORDER BY ts DESC LIMIT 1
 
+-- :name get-kraken-day-by-ts :? :1
+SELECT ts, open, high, low, close, vwap, volume, trade_count
+FROM kraken_day WHERE ts = :ts
+
+-- :name get-kraken-day-history-by-ts :? :1
+SELECT ts, open, high, low, close, vwap, volume, trade_count
+FROM kraken_day_history WHERE ts = :ts
+
+-- :name upsert-kraken-day-history! :! :n
+INSERT INTO kraken_day_history (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 insert-strike-price! :! :n
 -- :doc Insert a Strike ticker snapshot
 INSERT INTO strike_price (rates, quote_sats) VALUES (:rates, :quote-sats)

src/clj/pmagnus/btcdata/kraken/day.clj

@@ -0,0 +1,111 @@
+(ns pmagnus.btcdata.kraken.day
+  "On-demand completed UTC candles. Database failures are not upstream failures."
+  (:require [clojure.data.json :as json])
+  (:import [java.net URI]
+           [java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers]
+           [java.time Duration LocalDate ZoneOffset]
+           [java.math RoundingMode]))
+
+(defonce ^:private client
+  (delay (-> (HttpClient/newBuilder)
+             (.connectTimeout (Duration/ofSeconds 15))
+             (.build))))
+
+(defn- upstream-error [message]
+  (ex-info message {:type ::upstream}))
+
+(defn request!
+  "Fetch a Kraken public result, with a per-page timeout (not an overall deadline)."
+  [endpoint]
+  (try
+    (let [request (-> (HttpRequest/newBuilder)
+                      (.uri (URI/create (str "https://api.kraken.com/0/public/" endpoint)))
+                      (.timeout (Duration/ofSeconds 60))
+                      (.header "Accept" "application/json")
+                      (.GET) (.build))
+          response (.send ^HttpClient @client request (HttpResponse$BodyHandlers/ofString))]
+      (when-not (= 200 (.statusCode response))
+        (throw (upstream-error "Kraken returned an HTTP error")))
+      (let [body (json/read-str (.body response) :key-fn keyword :bigdec true)]
+        (when (or (seq (:error body)) (not (map? (:result body))))
+          (throw (upstream-error "Invalid or unsuccessful Kraken response")))
+        (:result body)))
+    (catch InterruptedException e
+      (.interrupt (Thread/currentThread))
+      (throw (ex-info "Kraken request interrupted" {:type ::upstream} e)))
+    (catch Exception e
+      (throw (ex-info "Failed to fetch Kraken data" {:type ::upstream} e)))))
+
+(defn- pair-rows [result]
+  (let [rows (:XXBTZUSD result)]
+    (when-not (vector? rows)
+      (throw (upstream-error "Missing BTC/USD data in Kraken response")))
+    rows))
+
+(defn- candle [[ts open high low close vwap volume trade-count]]
+  {:ts (long ts) :open (bigdec open) :high (bigdec high)
+   :low (bigdec low) :close (bigdec close) :vwap (bigdec vwap)
+   :volume (bigdec volume) :trade-count (long trade-count)})
+
+(defn- add-trade [acc [price volume]]
+  (let [p (bigdec price) v (bigdec volume)]
+    (-> (or acc {:open p :high p :low p :volume 0M :value 0M :trade-count 0})
+        (assoc :close p)
+        (update :high max p)
+        (update :low min p)
+        (update :volume + v)
+        (update :value + (* p v))
+        (update :trade-count inc))))
+
+(defn pause! [] (Thread/sleep 1000))
+
+(defn- historical-candle! [start]
+  (let [end (+ start 86400)]
+    (loop [cursor (str (*' (max 0 start) 1000000000)) acc nil]
+      (let [result (request! (str "Trades?pair=XBTUSD&since=" cursor))
+            rows (pair-rows result)
+            acc (reduce add-trade acc
+                        (filter (fn [row]
+                                  (let [ts (bigdec (nth row 2))]
+                                    (and (<= start ts) (< ts end))))
+                                rows))
+            done? (or (empty? rows) (some #(>= (bigdec (nth % 2)) end) rows))]
+        (if done?
+          (when acc
+            (when-not (pos? (:volume acc))
+              (throw (upstream-error "Kraken returned zero daily volume")))
+            (-> acc
+                (assoc :ts start :vwap (.divide ^java.math.BigDecimal (:value acc)
+                                               ^java.math.BigDecimal (:volume acc)
+                                               8 RoundingMode/HALF_UP))
+                (dissoc :value)))
+          (let [next-cursor (str (:last result))]
+            (when-not (> (bigint next-cursor) (bigint cursor))
+              (throw (upstream-error "Kraken trade pagination did not advance")))
+            (pause!)
+            (recur next-cursor acc)))))))
+
+(defn- fetch-candle! [ts]
+  (try
+    (let [rows (pair-rows (request! (str "OHLC?pair=XBTUSD&interval=1440&since=" (max 0 ts))))
+          match (some #(when (= ts (long (first %))) %) rows)]
+      (cond
+        match [:upsert-kraken-day! (candle match)]
+        (or (empty? rows) (< ts (long (ffirst rows))))
+        (when-let [c (historical-candle! ts)] [:upsert-kraken-day-history! c])
+        :else nil))
+    (catch Exception e
+      (throw (ex-info "Failed to obtain Kraken daily candle" {:type ::upstream} e)))))
+
+(defn find-day!
+  "Look up both caches before fetching. Returns a candle or nil. Date must be completed."
+  [query-fn ^LocalDate date]
+  (let [ts (.toEpochSecond (.atStartOfDay date) ZoneOffset/UTC)
+        params {:ts ts}]
+    (or (query-fn :get-kraken-day-by-ts params)
+        (query-fn :get-kraken-day-history-by-ts params)
+        (when-let [[save-query c] (fetch-candle! ts)]
+          (query-fn save-query c)
+          ;; Read stored numeric precision so first and cached responses match.
+          (query-fn (if (= save-query :upsert-kraken-day!)
+                      :get-kraken-day-by-ts :get-kraken-day-history-by-ts) params)))))

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

@@ -0,0 +1,45 @@
+(ns pmagnus.btcdata.web.controllers.day
+  (:require [pmagnus.btcdata.kraken.day :as day])
+  (:import [java.time LocalDate ZoneOffset]
+           [java.time.format DateTimeParseException]))
+
+(defn- response [status body]
+  {:status status :body body})
+
+(defn- parse-date [s]
+  (when (and (string? s) (re-matches #"[0-9]{4}-[0-9]{2}-[0-9]{2}" s))
+    (try (LocalDate/parse s) (catch DateTimeParseException _ nil))))
+
+(defn handle [{:keys [query-fn]} req]
+  (let [s (or (get-in req [:path-params :date])
+              (get-in req [:query-params "date"]))
+        date (parse-date s)]
+    (cond
+      (nil? date) (response 400 {:error "Invalid date, use YYYY-MM-DD"})
+      (not (.isBefore date (LocalDate/now ZoneOffset/UTC)))
+      (response 400 {:error "Only completed UTC days are allowed"})
+      :else
+      (try
+        (if-let [row (day/find-day! query-fn date)]
+          (response 200 (assoc (select-keys row [:open :high :low :close :vwap :volume :trade_count])
+                               :date (str date) :pair "BTC/USD"))
+          (response 404 {:error "No BTC/USD trading data for this day"}))
+        (catch clojure.lang.ExceptionInfo e
+          (if (= ::day/upstream (:type (ex-data e)))
+            (response 502 {:error "Failed to fetch or calculate Kraken daily data"})
+            (throw e)))))))
+
+(def response-schema
+  [:map [:date :string] [:pair :string]
+   [:open number?] [:high number?] [:low number?] [:close number?]
+   [:vwap number?] [:volume number?] [:trade_count int?]])
+
+(defn route [opts parameter-location]
+  {:get {:summary "BTC/USD daily OHLC for a completed UTC date"
+         :description "Checks stored candles first. Missing recent candles are fetched from Kraken; older days are calculated from historical trades and cached. Historical requests wait for completion and may take several minutes."
+         :parameters {parameter-location [:map [:date :string]]}
+         :responses {200 {:body response-schema}
+                     400 {:body [:map [:error :string]]}
+                     404 {:body [:map [:error :string]]}
+                     502 {:body [:map [:error :string]]}}
+         :handler (partial handle opts)}})

src/clj/pmagnus/btcdata/web/handler.clj

@@ -2,6 +2,7 @@
   (:require
    [integrant.core :as ig]
    [reitit.ring :as ring]
+   [reitit.swagger-ui :as swagger-ui]
    [pmagnus.btcdata.web.middleware.core :as middleware]))
 
 (defmethod ig/init-key :handler/ring
@@ -10,7 +11,8 @@
    (router)
    (ring/routes
     (when (some? api-path)
-      (reitit.ring/create-default-handler))
+      (swagger-ui/create-swagger-ui-handler
+       {:path "/swagger" :url (str api-path "/swagger.json")}))
     (ring/create-default-handler
      {:not-found          (constantly {:status 404 :body "Page not found"})
       :method-not-allowed (constantly {:status 405 :body "Not allowed"})

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

@@ -4,6 +4,7 @@
    [integrant.core :as ig]
    [pmagnus.btcdata.frankfurter.rates :as rates]
    [pmagnus.btcdata.web.controllers.health :as health]
+   [pmagnus.btcdata.web.controllers.day :as day]
    [pmagnus.btcdata.web.controllers.transactions :as tx]
    [pmagnus.btcdata.web.middleware.exception :as exception]
    [pmagnus.btcdata.web.middleware.formats :as formats]
@@ -188,6 +189,8 @@
      :middleware []}]
    ["/price/latest"
     {:get (fn [req] (latest-price-handler opts req))}]
+   ["/price/day" (day/route opts :query)]
+   ["/price/day/:date" (day/route opts :path)]
    ["/currencies/:date"
     {:get (fn [req] (currencies-handler opts req))}]
    ["/strike/ws"

test/clj/pmagnus/btcdata/kraken/day_test.clj

@@ -0,0 +1,77 @@
+(ns pmagnus.btcdata.kraken.day-test
+  (:require [clojure.test :refer [deftest is]]
+            [pmagnus.btcdata.kraken.day :as day])
+  (:import [java.time LocalDate ZoneOffset]))
+
+(def date (LocalDate/parse "2020-01-01"))
+(def ts (.toEpochSecond (.atStartOfDay date) ZoneOffset/UTC))
+
+(defn database [store writes]
+  (fn [q p]
+    (case q
+      :get-kraken-day-by-ts (get @store [:recent (:ts p)])
+      :get-kraken-day-history-by-ts (get @store [:history (:ts p)])
+      (:upsert-kraken-day! :upsert-kraken-day-history!)
+      (let [table (if (= q :upsert-kraken-day!) :recent :history)]
+        (swap! writes conj q)
+        (swap! store assoc [table (:ts p)]
+               (-> p (assoc :trade_count (:trade-count p)) (dissoc :trade-count)))))))
+
+(deftest cached-data-never-fetches
+  (doseq [table [:recent :history]]
+    (let [row {:ts ts} store (atom {[table ts] row})]
+      (with-redefs [day/request! (fn [_] (throw (Exception. "Unexpected fetch")))]
+        (is (= row (day/find-day! (database store (atom [])) date)))))))
+
+(deftest recent-candle-is-persisted
+  (let [store (atom {}) writes (atom []) q (database store writes)]
+    (with-redefs [day/request! (fn [_] {:XXBTZUSD [[ts "10" "12" "9" "11" "10.5" "2" 3]]})]
+      (is (= 11M (:close (day/find-day! q date))))
+      (is (= [:upsert-kraken-day!] @writes)))))
+
+(deftest historical-pagination-and-cache
+  (let [store (atom {}) writes (atom []) q (database store writes)
+        calls (atom []) cursor (str (*' (+ ts 5) 1000000000))]
+    (with-redefs [day/pause! (fn [])
+                  day/request!
+                  (fn [url]
+                    (swap! calls conj url)
+                    (case (count @calls)
+                      1 {:XXBTZUSD [[(+ ts 86400) "1" "1" "1" "1" "1" "1" 1]]}
+                      2 {:XXBTZUSD [["10" "2" ts] ["20" "1" (+ ts 5)]] :last cursor}
+                      3 {:XXBTZUSD [["5" "1" (+ ts 86399.9M)]
+                                   ["999" "1" (+ ts 86400)]] :last "unused"}))]
+      (let [result (day/find-day! q date)]
+        (is (= {:open 10M :high 20M :low 5M :close 5M :volume 4M
+                :vwap 11.25000000M :trade_count 3 :ts ts} result))
+        (is (= result (day/find-day! q date)))
+        (is (= 3 (count @calls)))
+        (is (.endsWith (last @calls) cursor))
+        (is (= [:upsert-kraken-day-history!] @writes))))))
+
+(deftest failed-or-empty-history-is-not-persisted
+  (doseq [result [{:XXBTZUSD [] :last "0"}
+                  {:XXBTZUSD [["1" "1" ts]] :last "0"}
+                  {:wrong []}]]
+    (let [writes (atom []) calls (atom 0)]
+      (with-redefs [day/request! (fn [_]
+                                  (if (= 1 (swap! calls inc))
+                                    {:XXBTZUSD [[(+ ts 86400)]]} result))]
+        (if (= [] (:XXBTZUSD result))
+          (is (nil? (day/find-day! (database (atom {}) writes) date)))
+          (is (thrown? clojure.lang.ExceptionInfo
+                       (day/find-day! (database (atom {}) writes) date))))
+        (is (empty? @writes))))))
+
+(deftest upstream-failure-does-not-save-a-partial-day
+  (let [writes (atom []) calls (atom 0)]
+    (with-redefs [day/pause! (fn [])
+                  day/request! (fn [_]
+                                 (case (swap! calls inc)
+                                   1 {:XXBTZUSD [[(+ ts 86400)]]}
+                                   2 {:XXBTZUSD [["10" "1" ts]]
+                                      :last (str (*' (inc ts) 1000000000))}
+                                   (throw (Exception. "Connection failed"))))]
+      (is (thrown? clojure.lang.ExceptionInfo
+                   (day/find-day! (database (atom {}) writes) date)))
+      (is (empty? @writes)))))

test/clj/pmagnus/btcdata/web/controllers/day_test.clj

@@ -0,0 +1,54 @@
+(ns pmagnus.btcdata.web.controllers.day-test
+  (:require [clojure.test :refer [deftest is]]
+            [clojure.data.json :as json]
+            [integrant.core :as ig]
+            [pmagnus.btcdata.kraken.day :as day]
+            [pmagnus.btcdata.web.controllers.day :as controller]
+            [pmagnus.btcdata.web.routes.api :as api]
+            [pmagnus.btcdata.web.handler]
+            [reitit.ring :as ring]))
+
+(def app
+  (ring/ring-handler
+   (ring/router [(ig/init-key :reitit.routes/api {:base-path "/api" :query-fn nil})])))
+
+(defn request [uri query]
+  (app {:request-method :get :uri uri :query-string query
+        :headers {"accept" "application/json"}}))
+
+(deftest routes-and-numeric-json
+  (with-redefs [day/find-day! (fn [_ _]
+                              {:open 10M :high 12M :low 9M :close 11M
+                               :vwap 10.5M :volume 2M :trade_count 3})]
+    (doseq [[uri query] [["/api/price/day/2020-01-01" nil]
+                        ["/api/price/day" "date=2020-01-01"]]]
+      (let [r (request uri query) body (json/read-str (slurp (:body r)) :key-fn keyword)]
+        (is (= 200 (:status r)))
+        (is (= "BTC/USD" (:pair body)))
+        (is (= "2020-01-01" (:date body)))
+        (is (number? (:open body)))))))
+
+(deftest invalid-dates
+  (doseq [s [nil "bad" "2020-02-30" "2020-1-01" "9999-01-01"
+             (str (java.time.LocalDate/now java.time.ZoneOffset/UTC))]]
+    (with-redefs [day/find-day! (fn [& _] (throw (Exception. "Must not fetch")))]
+      (is (= 400 (:status (controller/handle {} {:query-params {"date" s}})))))))
+
+(deftest error-responses
+  (with-redefs [day/find-day! (constantly nil)]
+    (is (= 404 (:status (request "/api/price/day/2020-01-01" nil)))))
+  (with-redefs [day/find-day! (fn [& _] (throw (ex-info "upstream" {:type ::day/upstream})))]
+    (is (= 502 (:status (request "/api/price/day/2020-01-01" nil))))))
+
+(deftest swagger-documents-both-routes
+  (let [r (request "/api/swagger.json" nil)
+        body (json/read-str (slurp (:body r)))]
+    (is (= 200 (:status r)))
+    (is (get-in body ["paths" "/api/price/day" "get"]))
+    (is (get-in body ["paths" "/api/price/day/{date}" "get"]))))
+
+(deftest swagger-ui-is-served
+  (let [handler (ig/init-key :handler/ring
+                            {:router #(ring/router []) :api-path "/api"})]
+    (is (= 302 (:status (handler {:request-method :get :uri "/swagger" :headers {}}))))
+    (is (= 200 (:status (handler {:request-method :get :uri "/swagger/" :headers {}}))))))