Persist closed candles in PostgreSQL and read them back

.gitattributes

@@ -0,0 +1,3 @@
+# The CI job regenerates this fixture and compares it to the committed copy,
+# so its line endings must not depend on the checkout platform.
+ingest/fixtures/public-candles.json text eol=lf

.github/workflows/check.yml

@@ -26,3 +26,80 @@ jobs:
           grep -q 'Pattern Forge' /tmp/pattern-forge-page.html
           test "$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:3000/api/markets/INVALID)" = 400
           test "$(docker exec pattern-forge-test id -u)" != 0
+          # Without a database the stored reader must say so, not appear healthy.
+          test "$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:3000/api/stored/BTC)" = 503
+
+  persistence:
+    runs-on: ubuntu-latest
+    # GUESS: UNCALIBRATED GUESS — CI timeout, not application performance.
+    timeout-minutes: 20
+    services:
+      postgres:
+        image: postgres:18-alpine
+        env:
+          POSTGRES_DB: patternforge
+          POSTGRES_USER: patternforge
+          # CI-only credential for an ephemeral service container.
+          POSTGRES_PASSWORD: patternforge
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U patternforge -d patternforge"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 12
+    env:
+      DATABASE_URL: postgresql://patternforge:patternforge@127.0.0.1:5432/patternforge
+      # SOURCE: the fixed instant the recorded payload is validated against.
+      # See ingest/fixtures/build_fixture.py.
+      FIXTURE_NOW_MS: "1788220800000"
+      # SOURCE: 48 of the 50 recorded BTC hourly rows are closed and unique.
+      EXPECTED_HOURLY: "48"
+    steps:
+      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+      # SOURCE: actions/setup-python v5 and actions/setup-node v4 pinned by commit.
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
+        with:
+          python-version: "3.13"
+      - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
+        with:
+          node-version: "24"
+      - name: Install dependencies
+        run: |
+          python -m pip install --disable-pip-version-check -r ingest/requirements.txt
+          npm ci
+      - name: The recorded fixture is the committed one
+        run: |
+          python -m ingest.fixtures.build_fixture
+          git diff --exit-code -- ingest/fixtures/public-candles.json
+
+      - name: Ingest, repeat and confirm the stored rows
+        run: python -m unittest ingest.tests.test_persistence -v
+
+      - name: Load the fixture in one process, then read it back in another
+        shell: bash
+        run: |
+          set -euo pipefail
+          python -m ingest.ingest \
+            --fixture ingest/fixtures/public-candles.json \
+            --now "$FIXTURE_NOW_MS" --symbols BTC --intervals 1h --show-stored
+          # A second pass must not grow the table: the primary key is the guard.
+          python -m ingest.ingest \
+            --fixture ingest/fixtures/public-candles.json \
+            --now "$FIXTURE_NOW_MS" --symbols BTC --intervals 1h
+          # This process never saw the ingester, so the rows can only come from PostgreSQL.
+          node scripts/check_persistence.mjs "$EXPECTED_HOURLY"
+
+      - name: Restart the database and replay what survived
+        shell: bash
+        run: |
+          set -euo pipefail
+          container="$(docker ps --filter ancestor=postgres:18-alpine --format '{{.ID}}' | head -n 1)"
+          test -n "$container"
+          docker restart "$container"
+          # GUESS: UNCALIBRATED GUESS — bounded wait for the restarted server.
+          for attempt in $(seq 1 30); do
+            if docker exec "$container" pg_isready -U patternforge -d patternforge; then break; fi
+            sleep 2
+          done
+          node scripts/check_persistence.mjs "$EXPECTED_HOURLY"

.gitignore

@@ -3,4 +3,5 @@ node_modules/
 .next/
 .env*
 *.tsbuildinfo
+__pycache__/
 .github/

README.md

@@ -73,8 +73,10 @@ The existing Lightweight Charts dependency and attribution are retained.
 Use the existing Node environment and lockfile. `npm run dev` starts Next;
 For a fresh clone, use Node 22.13 or newer and run `npm ci` first.
 `npm test` runs mechanics and archive-validation tests; `npm run test:build`
-also produces the production build. Deployment uses the existing linked Vercel
-project. Keep the raw downloads outside this checkout; public JSON is generated
+also produces the production build. The ingester's offline tests run with
+`python -m unittest discover -s ingest/tests -t .`; the PostgreSQL checks in
+`ingest/tests/test_persistence.py` are skipped unless `DATABASE_URL` is set.
+Deployment uses the existing linked Vercel project. Keep the raw downloads outside this checkout; public JSON is generated
 data, not a hand-edited fixture. Do not run any trading service to test this UI.
 
 ## API and deployment
@@ -104,8 +106,11 @@ docker run --rm -p 127.0.0.1:3000:3000 pattern-forge
 The multi-stage image runs as a non-root user and excludes local environment
 files. [CI](https://github.com/coder058/pattern-forge/actions/workflows/check.yml)
 builds that image, runs tests during the build, starts it, and checks readiness,
-the page and an invalid-market request. Vercel remains the public deployment;
-it does not use this Docker image. No new paid service is required.
+the page and an invalid-market request. A second job starts a real PostgreSQL
+service, ingests the recorded payload twice, reads it back from a separate
+process, restarts the database and reads it again. Vercel remains the public
+deployment; it does not use this Docker image and does not serve stored candles.
+No new paid service is required.
 
 The quote feed uses the documented
 [allMids subscription](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/websocket/subscriptions).
@@ -118,6 +123,56 @@ questions, and mixing a current quote into an archived candle would make replay
 misleading. I used an API to centralize validation and share duplicate requests,
 not to hide an exchange URL behind an unnecessary microservice.
 
+## Stored candles
+
+Snapshots were previously held in one process's memory, so a restart lost them
+and the workspace could show nothing when the endpoint was unreachable. A small
+Python ingester now writes closed candles to PostgreSQL, and the app reads them
+back through `GET /api/stored/BTC` (also ETH and SOL). The chart, the replay
+cursor and the timeframe comparison are unchanged: a stored market is just
+another source in the selector, listed under **Stored candles**.
+
+```sh
+# Database, ingester and app together. The named volume keeps the candles.
+docker compose up --build
+
+# One pass against the live endpoint, into an existing database.
+DATABASE_URL=postgresql://... python -m ingest.ingest --symbols BTC --intervals 1h,4h
+
+# Offline: load the recorded payload instead of calling the endpoint.
+DATABASE_URL=postgresql://... python -m ingest.ingest \
+  --fixture ingest/fixtures/public-candles.json --now 1788220800000 --symbols BTC --intervals 1h
+```
+
+The ingester reads the same documented
+[candleSnapshot endpoint](https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/info-endpoint)
+the browser uses, and applies the same boundary rules in `ingest/candles.py`:
+a still-forming candle, a candle that does not span its interval, a non-numeric
+price and inconsistent OHLC bounds are all refused before any write. Rows
+sharing an opening timestamp with different values are treated as a conflict,
+not a duplicate, and reject the whole response.
+
+`(symbol, interval, open_time)` is the primary key, so re-reading the same
+window cannot create a second row. The upsert suppresses no-op updates, which
+means each pass reports how many candles were inserted, updated and left
+unchanged; a corrected candle updates in place instead of appearing twice.
+Every attempt, including a failure, is recorded in `ingest_runs` with its own
+`fetch_ms` and `write_ms`, measured around the fetch and write calls. Reads
+report `Server-Timing: query;dur=` for the handler.
+
+Those numbers are wall-clock measurements between explicit start and end points
+in one process. They are not throughput, database server time, network time or
+exchange-to-screen latency, and no synchronized clocks are involved.
+
+Limits: closed candles only, so this is a persisted snapshot history and not a
+tick feed. One ingester process, no retention, partitioning or backfill policy,
+and reads are bounded to 300 bars per interval. An empty store answers 404 with
+an instruction rather than an empty chart, and a database that is unreachable
+answers 503 rather than silently falling back to the live endpoint. The
+duplicate and restart behaviour is checked against a real PostgreSQL server in
+CI, using the recorded payload; the counts describe one statement's view, not
+concurrent ingesters.
+
 ## A reproducible investigation
 
 Choose recorded Gold, move the replay cursor back, then enable timeframe

app/api/stored/[symbol]/route.ts

@@ -0,0 +1,5 @@
+import { candleQuery } from "../../../db";
+import { createStoredApi } from "../../../storedApi";
+
+export const dynamic = "force-dynamic";
+export const GET = createStoredApi(candleQuery);

app/db.ts

@@ -0,0 +1,33 @@
+import { Pool } from "pg";
+import type { CandleQuery } from "./storedApi.ts";
+
+// GUESS: UNCALIBRATED GUESS — small pool and bounded waits for a single
+// container. Not measured capacity.
+const POOL_SIZE = 4;
+const CONNECT_TIMEOUT_MS = 5_000;
+const IDLE_TIMEOUT_MS = 30_000;
+
+let pool: Pool | undefined;
+
+/** Created on first use, so a build or a page render without DATABASE_URL still works. */
+function candlePool() {
+  const connectionString = process.env.DATABASE_URL;
+  if (!connectionString) throw new Error("DATABASE_URL is not set.");
+  pool ??= new Pool({
+    connectionString,
+    max: POOL_SIZE,
+    connectionTimeoutMillis: CONNECT_TIMEOUT_MS,
+    idleTimeoutMillis: IDLE_TIMEOUT_MS,
+  });
+  return pool;
+}
+
+export const candleQuery: CandleQuery = async (text, values) =>
+  (await candlePool().query(text, values)).rows;
+
+/** Release the pool so a one-shot script can exit. Not used by the server. */
+export async function closeCandlePool() {
+  const open = pool;
+  pool = undefined;
+  await open?.end();
+}

app/marketSources.ts

@@ -10,7 +10,7 @@ export type MarketSource = {
   symbol: string;
   label: string;
   group: string;
-  kind: "public" | "recording" | "case";
+  kind: "public" | "stored" | "recording" | "case";
   venue: string;
   path?: string;
   count?: number;
@@ -41,6 +41,19 @@ export const SOURCES: MarketSource[] = [
     kind: "public" as const,
     venue: "Hyperliquid",
   })),
+  // Served from PostgreSQL by the ingester, so these survive a restart of the
+  // application and remain readable when the upstream endpoint is unreachable.
+  ...["BTC", "ETH", "SOL"].map((symbol) => ({
+    id: `stored-${symbol}`,
+    symbol,
+    label: (
+      { BTC: "Bitcoin", ETH: "Ethereum", SOL: "Solana" } as Record<string, string>
+    )[symbol],
+    group: "Stored candles",
+    kind: "stored" as const,
+    venue: "Hyperliquid via PostgreSQL",
+    note: "Read from the local database. Empty until the ingester has run.",
+  })),
   ...catalog.map((item) => ({ ...item, kind: "recording" as const })),
   {
     id: "saved-btc",
@@ -54,7 +67,7 @@ export const SOURCES: MarketSource[] = [
   },
 ];
 export const SOURCE_INTERVALS = (source: MarketSource): Interval[] =>
-  source.kind === "recording"
+  source.kind === "recording" || source.kind === "stored"
     ? ["1h", "4h", "1d"]
     : source.kind === "case"
       ? ["5m", "30m", "1h", "4h"]
@@ -64,9 +77,16 @@ export async function loadMarket(
   source: MarketSource,
   signal: AbortSignal,
 ): Promise<LoadedMarket> {
-  if (source.kind === "public") {
-    const response = await fetch(`/api/markets/${source.symbol}`, { signal });
-    if (!response.ok) throw new Error("Public candles are unavailable. Try Refresh or choose a recording.");
+  if (source.kind === "public" || source.kind === "stored") {
+    const stored = source.kind === "stored";
+    const response = await fetch(`/api/${stored ? "stored" : "markets"}/${source.symbol}`, { signal });
+    if (!response.ok) {
+      const reason = await response.json().catch(() => undefined);
+      if (typeof reason?.error === "string") throw new Error(reason.error);
+      throw new Error(stored
+        ? "Stored candles are unavailable. Check the database, or choose a recording."
+        : "Public candles are unavailable. Try Refresh or choose a recording.");
+    }
     const snapshot = await response.json();
     if (snapshot.symbol !== source.symbol || !Number.isSafeInteger(snapshot.asOf))
       throw new Error("The snapshot does not match the selected market.");

app/storedApi.ts

@@ -0,0 +1,38 @@
+import { PUBLIC_SYMBOLS } from "./marketApi.ts";
+import { STORED_BARS, STORED_INTERVALS, STORED_QUERY, toFrames, type StoredRow } from "./storedMarket.ts";
+
+export type CandleQuery = (text: string, values: unknown[]) => Promise<StoredRow[]>;
+
+/** Read persisted candles for one allowlisted market.
+ * An empty store is a normal state before the ingester has run, so it answers
+ * 404 with an instruction rather than pretending the database is broken.
+ */
+export function createStoredApi(query: CandleQuery, now = Date.now) {
+  return async (input: Request): Promise<Response> => {
+    const symbol = new URL(input.url).pathname.split("/").at(-1) ?? "";
+    if (!PUBLIC_SYMBOLS.includes(symbol as typeof PUBLIC_SYMBOLS[number])) {
+      return Response.json({ error: "Choose BTC, ETH or SOL." }, { status: 400 });
+    }
+    const started = now();
+    try {
+      const rows = await query(STORED_QUERY, [symbol, STORED_INTERVALS, STORED_BARS]);
+      const { frames, counts, asOf } = toFrames(rows);
+      if (!Object.keys(frames).length) {
+        return Response.json(
+          { error: "No candles are stored for this market yet. Run the ingester, then reload." },
+          { status: 404, headers: { "Cache-Control": "no-store" } });
+      }
+      return Response.json({ ...{ frames, asOf }, symbol, venue: "Hyperliquid", stored: counts, source: "database" }, {
+        headers: {
+          "Cache-Control": "no-store",
+          // SOURCE: measured query and mapping wall time inside this handler.
+          // Not database server time, network time or end-to-end latency.
+          "Server-Timing": `query;dur=${Math.max(0, now() - started)}`,
+        },
+      });
+    } catch {
+      return Response.json({ error: "Stored candles are unavailable. Check the database connection." },
+        { status: 503, headers: { "Cache-Control": "no-store" } });
+    }
+  };
+}

app/storedMarket.ts

@@ -0,0 +1,74 @@
+import { INTERVAL_MS, type Bar, type Interval } from "./marketAnalysis.ts";
+
+/** Intervals the ingester persists by default. */
+export const STORED_INTERVALS: Interval[] = ["1h", "4h", "1d"];
+// GUESS: UNCALIBRATED GUESS — read bound per interval, matching the existing
+// display history in app/publicSnapshot.ts. Not a trading lookback.
+export const STORED_BARS = 300;
+
+/** One round trip for every interval: rank each interval's rows, keep the newest.
+ * The primary key already prevents duplicate candles, so no grouping is needed here.
+ */
+export const STORED_QUERY = `
+SELECT interval, open_time, close_time, "open", "high", "low", "close", volume
+  FROM (
+    SELECT interval, open_time, close_time, "open", "high", "low", "close", volume,
+           row_number() OVER (PARTITION BY interval ORDER BY open_time DESC) AS position
+      FROM candles
+     WHERE symbol = $1 AND interval = ANY($2)
+  ) ranked
+ WHERE position <= $3
+ ORDER BY interval, open_time
+`;
+
+export type StoredRow = Record<string, unknown>;
+
+/** node-postgres returns bigint columns as strings, so widen then check the range. */
+function epochMilliseconds(value: unknown) {
+  const result = typeof value === "string" ? Number(value) : value;
+  if (typeof result !== "number" || !Number.isSafeInteger(result) || result < 0)
+    throw new Error("Stored candle has an unusable timestamp.");
+  return result;
+}
+
+function price(value: unknown) {
+  const result = typeof value === "string" ? Number(value) : value;
+  if (typeof result !== "number" || !Number.isFinite(result))
+    throw new Error("Stored candle has an unusable price or volume.");
+  return result;
+}
+
+/** Group stored rows into per-interval frames without trusting the row contents.
+ * A stored row still has to describe a whole candle of its interval; anything
+ * else is a storage fault and is reported rather than charted.
+ */
+export function toFrames(rows: Iterable<StoredRow>) {
+  const frames: Partial<Record<Interval, Bar[]>> = {};
+  const byInterval = new Map<Interval, Map<number, Bar>>();
+  let asOf = 0;
+  for (const row of rows) {
+    const interval = row.interval as Interval;
+    if (!STORED_INTERVALS.includes(interval)) continue;
+    const t = epochMilliseconds(row.open_time);
+    const closeTime = epochMilliseconds(row.close_time);
+    if (closeTime !== t + INTERVAL_MS[interval])
+      throw new Error("Stored candle does not span its interval.");
+    const o = price(row.open), h = price(row.high), l = price(row.low);
+    const c = price(row.close), v = price(row.volume);
+    if (l <= 0 || h < Math.max(o, c, l) || l > Math.min(o, c) || v < 0)
+      throw new Error("Stored candle has inconsistent bounds.");
+    // Only closed candles are ever persisted, so a stored row is closed by construction.
+    const bar: Bar = { t, closeTime, o, h, l, c, v, closed: true };
+    if (!byInterval.has(interval)) byInterval.set(interval, new Map());
+    byInterval.get(interval)!.set(t, bar);
+    if (closeTime > asOf) asOf = closeTime;
+  }
+  const counts: Partial<Record<Interval, number>> = {};
+  for (const [interval, bars] of byInterval) {
+    frames[interval] = [...bars.values()].sort((a, b) => a.t - b.t);
+    counts[interval] = frames[interval]!.length;
+  }
+  // The cutoff is the newest stored close, so a restart replays what is on disk
+  // rather than assuming the ingester is currently running.
+  return { frames, counts, asOf };
+}