From 8b457304d980acd354e27033ab75d29d3518869e Mon Sep 17 00:00:00 2001 From: Jordi Lluis Date: Sun, 6 Sep 2026 13:39:12 +0200 Subject: [PATCH] Persist closed candles in PostgreSQL and read them back Snapshots only lived in one process's memory, so a restart lost them and the workspace had nothing to show when the endpoint was unreachable. A Python ingester reads the same documented candleSnapshot endpoint, applies the same boundary rules as the browser, and writes closed candles to PostgreSQL. (symbol, interval, open_time) is the primary key and the upsert suppresses no-op updates, so a repeated window inserts nothing, a corrected candle updates in place, and each pass reports inserted, updated and unchanged counts. Every attempt is recorded with the wall time measured around its fetch and write. GET /api/stored/:symbol reads those rows back for the existing chart, replay cursor and timeframe comparison. An empty store is a 404 with an instruction; an unreachable database is a 503, never a silent fall back to the live endpoint. Compose runs the database, ingester and app against a named volume. A second CI job ingests the recorded payload twice against a real PostgreSQL server, reads it back from a separate process, restarts the database and reads it again. Co-authored-by: Cursor --- .gitattributes | 3 + .github/workflows/check.yml | 77 ++ .gitignore | 1 + README.md | 63 +- app/api/stored/[symbol]/route.ts | 5 + app/db.ts | 33 + app/marketSources.ts | 30 +- app/storedApi.ts | 38 + app/storedMarket.ts | 74 ++ docker-compose.yml | 51 ++ ingest/Dockerfile | 12 + ingest/__init__.py | 1 + ingest/candles.py | 90 +++ ingest/fixtures/__init__.py | 0 ingest/fixtures/build_fixture.py | 60 ++ ingest/fixtures/public-candles.json | 1044 +++++++++++++++++++++++++++ ingest/ingest.py | 188 +++++ ingest/requirements.txt | 2 + ingest/schema.sql | 38 + ingest/store.py | 93 +++ ingest/tests/__init__.py | 0 ingest/tests/test_candles.py | 83 +++ ingest/tests/test_ingest.py | 158 ++++ ingest/tests/test_persistence.py | 130 ++++ package-lock.json | 160 ++++ package.json | 2 + scripts/check_persistence.mjs | 61 ++ tests/stored-api.test.mjs | 72 ++ tests/stored-market.test.mjs | 71 ++ 29 files changed, 2631 insertions(+), 9 deletions(-) create mode 100644 .gitattributes create mode 100644 app/api/stored/[symbol]/route.ts create mode 100644 app/db.ts create mode 100644 app/storedApi.ts create mode 100644 app/storedMarket.ts create mode 100644 docker-compose.yml create mode 100644 ingest/Dockerfile create mode 100644 ingest/__init__.py create mode 100644 ingest/candles.py create mode 100644 ingest/fixtures/__init__.py create mode 100644 ingest/fixtures/build_fixture.py create mode 100644 ingest/fixtures/public-candles.json create mode 100644 ingest/ingest.py create mode 100644 ingest/requirements.txt create mode 100644 ingest/schema.sql create mode 100644 ingest/store.py create mode 100644 ingest/tests/__init__.py create mode 100644 ingest/tests/test_candles.py create mode 100644 ingest/tests/test_ingest.py create mode 100644 ingest/tests/test_persistence.py create mode 100644 scripts/check_persistence.mjs create mode 100644 tests/stored-api.test.mjs create mode 100644 tests/stored-market.test.mjs diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..64dc73a --- /dev/null +++ b/.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 diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 2c99250..7ae5d54 100644 --- a/.github/workflows/check.yml +++ b/.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" diff --git a/.gitignore b/.gitignore index 56cc672..b78f92a 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,5 @@ node_modules/ .next/ .env* *.tsbuildinfo +__pycache__/ .github/ diff --git a/README.md b/README.md index 94c8fe9..ce93d8c 100644 --- a/README.md +++ b/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 diff --git a/app/api/stored/[symbol]/route.ts b/app/api/stored/[symbol]/route.ts new file mode 100644 index 0000000..2f2d071 --- /dev/null +++ b/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); diff --git a/app/db.ts b/app/db.ts new file mode 100644 index 0000000..14b5702 --- /dev/null +++ b/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(); +} diff --git a/app/marketSources.ts b/app/marketSources.ts index b97fed2..b853db4 100644 --- a/app/marketSources.ts +++ b/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 + )[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 { - 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."); diff --git a/app/storedApi.ts b/app/storedApi.ts new file mode 100644 index 0000000..8dc636c --- /dev/null +++ b/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; + +/** 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 => { + 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" } }); + } + }; +} diff --git a/app/storedMarket.ts b/app/storedMarket.ts new file mode 100644 index 0000000..8378ac9 --- /dev/null +++ b/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; + +/** 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) { + const frames: Partial> = {}; + const byInterval = new Map>(); + 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> = {}; + 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 }; +} diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..ac2ee36 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,51 @@ +# One documented public endpoint, persisted to PostgreSQL, read back by the app. +# The named volume is the point: stop everything, start it again, and the stored +# candles are still there to replay. +services: + db: + image: postgres:18-alpine + environment: + POSTGRES_DB: patternforge + POSTGRES_USER: patternforge + # SOURCE: local development credential only. Supply a real secret through + # the environment before running this anywhere reachable. + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-patternforge} + healthcheck: + test: ["CMD-SHELL", "pg_isready -U patternforge -d patternforge"] + # GUESS: UNCALIBRATED GUESS — startup allowance, not a measured boot time. + interval: 5s + timeout: 5s + retries: 12 + volumes: + - candles:/var/lib/postgresql/data + # Bound to loopback so a development database is not exposed on the network. + ports: + - "127.0.0.1:5432:5432" + + ingest: + build: + context: . + dockerfile: ingest/Dockerfile + environment: + DATABASE_URL: postgresql://patternforge:${POSTGRES_PASSWORD:-patternforge}@db:5432/patternforge + # GUESS: UNCALIBRATED GUESS — a five-minute pass is well inside the shortest + # stored interval. It is not a freshness guarantee. + command: ["--every", "300", "--symbols", "BTC,ETH,SOL", "--intervals", "1h,4h,1d"] + depends_on: + db: + condition: service_healthy + restart: unless-stopped + + web: + build: + context: . + environment: + DATABASE_URL: postgresql://patternforge:${POSTGRES_PASSWORD:-patternforge}@db:5432/patternforge + ports: + - "127.0.0.1:3000:3000" + depends_on: + db: + condition: service_healthy + +volumes: + candles: diff --git a/ingest/Dockerfile b/ingest/Dockerfile new file mode 100644 index 0000000..3614697 --- /dev/null +++ b/ingest/Dockerfile @@ -0,0 +1,12 @@ +# SOURCE: python:3.13-slim; psycopg[binary] ships libpq, so no compiler is needed. +FROM python:3.13-slim +WORKDIR /app +ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 +COPY ingest/requirements.txt ./ingest/requirements.txt +RUN pip install --no-cache-dir -r ingest/requirements.txt +COPY ingest ./ingest +RUN python -m unittest discover -s ingest/tests -t . \ + && useradd --create-home --uid 10001 ingester +USER ingester +# Arguments are supplied by the compose command, for example --every 300. +ENTRYPOINT ["python", "-m", "ingest.ingest"] diff --git a/ingest/__init__.py b/ingest/__init__.py new file mode 100644 index 0000000..e654995 --- /dev/null +++ b/ingest/__init__.py @@ -0,0 +1 @@ +"""Candle ingestion for Pattern Forge: read a public endpoint, persist closed bars.""" diff --git a/ingest/candles.py b/ingest/candles.py new file mode 100644 index 0000000..968f4a4 --- /dev/null +++ b/ingest/candles.py @@ -0,0 +1,90 @@ +"""Closed-candle validation for the ingestion path. + +SOURCE: these rules mirror closedPublicCandles in app/publicMarket.ts. The +browser and the ingester read the same upstream shape, so the same boundary +rules apply in both languages. This is a deliberate duplicate, not a shared +library: a candle that the browser would refuse must not reach the database. +""" + +# SOURCE: Hyperliquid candleSnapshot interval identifiers and their durations. +INTERVAL_MS = { + "1m": 60_000, + "5m": 300_000, + "15m": 900_000, + "30m": 1_800_000, + "1h": 3_600_000, + "4h": 14_400_000, + "1d": 86_400_000, +} + + +class CandleError(ValueError): + """A candle response cannot be trusted and must not be persisted.""" + + +def _number(value): + """Accept a JSON number or a non-empty numeric string, as the upstream sends both.""" + if isinstance(value, bool): + raise CandleError("Candle contains non-numeric price or volume.") + if isinstance(value, (int, float)): + result = float(value) + elif isinstance(value, str) and value.strip(): + try: + result = float(value) + except ValueError: + raise CandleError("Candle contains non-numeric price or volume.") from None + else: + raise CandleError("Candle contains non-numeric price or volume.") + if result != result or result in (float("inf"), float("-inf")): + raise CandleError("Candle contains non-numeric price or volume.") + return result + + +def _integer(value): + return isinstance(value, int) and not isinstance(value, bool) + + +def closed_candles(payload, now_ms, interval): + """Return sorted, provably closed candles, or raise CandleError. + + Candles whose inclusive close is at or after ``now_ms`` are still forming and + are dropped rather than stored. Rows sharing an opening timestamp with + different values are a conflict, not a duplicate, and reject the response. + """ + if interval not in INTERVAL_MS: + raise CandleError(f"Unsupported interval: {interval}") + interval_ms = INTERVAL_MS[interval] + if not isinstance(payload, list): + raise CandleError("The exchange returned an invalid candle response.") + by_time = {} + for item in payload: + if not isinstance(item, dict): + raise CandleError("Malformed candle row.") + open_time, close_inclusive = item.get("t"), item.get("T") + if not _integer(open_time) or not _integer(close_inclusive): + raise CandleError("Candle timestamps do not match the requested interval.") + if open_time < 0 or close_inclusive < open_time: + raise CandleError("Candle timestamps do not match the requested interval.") + if close_inclusive + 1 != open_time + interval_ms: + raise CandleError("Candle timestamps do not match the requested interval.") + if close_inclusive >= now_ms: + continue + opening, high, low, closing, volume = (_number(item.get(key)) for key in ("o", "h", "l", "c", "v")) + if low <= 0 or high < max(opening, closing, low) or low > min(opening, closing) or volume < 0: + raise CandleError("Candle OHLC bounds are inconsistent.") + row = { + "open_time": open_time, + "close_time": close_inclusive + 1, + "open": opening, + "high": high, + "low": low, + "close": closing, + "volume": volume, + } + previous = by_time.get(open_time) + if previous is not None and previous != row: + raise CandleError("Conflicting candles share an opening timestamp.") + by_time[open_time] = row + if not by_time: + raise CandleError("No provably closed candles were returned.") + return [by_time[key] for key in sorted(by_time)] diff --git a/ingest/fixtures/__init__.py b/ingest/fixtures/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingest/fixtures/build_fixture.py b/ingest/fixtures/build_fixture.py new file mode 100644 index 0000000..59d5379 --- /dev/null +++ b/ingest/fixtures/build_fixture.py @@ -0,0 +1,60 @@ +"""Regenerate the recorded candle payloads used by tests and continuous integration. + +The values are synthetic and deterministic. Tests must not depend on the live +endpoint, so the fixture stands in for it, including the two cases that matter +for persistence: a candle that is still forming, and an exact duplicate. + + python -m ingest.fixtures.build_fixture +""" +import datetime +import json +from pathlib import Path + +HOUR_MS = 3_600_000 +# The fixture is validated against this fixed instant, so "closed" never drifts. +NOW_MS = int(datetime.datetime(2026, 9, 1, tzinfo=datetime.timezone.utc).timestamp() * 1000) +OUTPUT = Path(__file__).with_name("public-candles.json") + + +def bar(open_time, interval_ms, seed, symbol, interval): + opening = 60_000 + seed * 12.5 + closing = opening + (7.5 if seed % 3 else -9.25) + return { + "t": open_time, + "T": open_time + interval_ms - 1, + "o": opening, + "h": max(opening, closing) + 4.5, + "l": min(opening, closing) - 3.25, + "c": closing, + "v": 10.0 + seed, + "s": symbol, + "i": interval, + "n": 5, + } + + +def series(interval_ms, count, symbol, interval): + start = NOW_MS - interval_ms * count + return [bar(start + index * interval_ms, interval_ms, index, symbol, interval) for index in range(count)] + + +def build(): + btc_hourly = series(HOUR_MS, 48, "BTC", "1h") + closed_unique = len(btc_hourly) + # Still forming: its inclusive close is not before NOW_MS, so it must be dropped. + btc_hourly.append(bar(NOW_MS, HOUR_MS, 99, "BTC", "1h")) + # An exact repeat of a closed candle: one stored row, not two. + btc_hourly.append(dict(btc_hourly[5])) + fixture = { + "BTC": {"1h": btc_hourly, "4h": series(HOUR_MS * 4, 12, "BTC", "4h")}, + "ETH": {"1h": series(HOUR_MS, 24, "ETH", "1h")}, + } + # Written with newline="\n" so the committed file is byte-identical whether + # it is regenerated on Windows or on the Linux CI runner. + with OUTPUT.open("w", encoding="utf-8", newline="\n") as handle: + handle.write(json.dumps(fixture, indent=1) + "\n") + return {"now_ms": NOW_MS, "btc_1h_payload": len(btc_hourly), "btc_1h_closed_unique": closed_unique} + + +if __name__ == "__main__": + print(json.dumps(build(), indent=1)) diff --git a/ingest/fixtures/public-candles.json b/ingest/fixtures/public-candles.json new file mode 100644 index 0000000..65c2334 --- /dev/null +++ b/ingest/fixtures/public-candles.json @@ -0,0 +1,1044 @@ +{ + "BTC": { + "1h": [ + { + "t": 1788048000000, + "T": 1788051599999, + "o": 60000.0, + "h": 60004.5, + "l": 59987.5, + "c": 59990.75, + "v": 10.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788051600000, + "T": 1788055199999, + "o": 60012.5, + "h": 60024.5, + "l": 60009.25, + "c": 60020.0, + "v": 11.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788055200000, + "T": 1788058799999, + "o": 60025.0, + "h": 60037.0, + "l": 60021.75, + "c": 60032.5, + "v": 12.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788058800000, + "T": 1788062399999, + "o": 60037.5, + "h": 60042.0, + "l": 60025.0, + "c": 60028.25, + "v": 13.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788062400000, + "T": 1788065999999, + "o": 60050.0, + "h": 60062.0, + "l": 60046.75, + "c": 60057.5, + "v": 14.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788066000000, + "T": 1788069599999, + "o": 60062.5, + "h": 60074.5, + "l": 60059.25, + "c": 60070.0, + "v": 15.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788069600000, + "T": 1788073199999, + "o": 60075.0, + "h": 60079.5, + "l": 60062.5, + "c": 60065.75, + "v": 16.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788073200000, + "T": 1788076799999, + "o": 60087.5, + "h": 60099.5, + "l": 60084.25, + "c": 60095.0, + "v": 17.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788076800000, + "T": 1788080399999, + "o": 60100.0, + "h": 60112.0, + "l": 60096.75, + "c": 60107.5, + "v": 18.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788080400000, + "T": 1788083999999, + "o": 60112.5, + "h": 60117.0, + "l": 60100.0, + "c": 60103.25, + "v": 19.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788084000000, + "T": 1788087599999, + "o": 60125.0, + "h": 60137.0, + "l": 60121.75, + "c": 60132.5, + "v": 20.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788087600000, + "T": 1788091199999, + "o": 60137.5, + "h": 60149.5, + "l": 60134.25, + "c": 60145.0, + "v": 21.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788091200000, + "T": 1788094799999, + "o": 60150.0, + "h": 60154.5, + "l": 60137.5, + "c": 60140.75, + "v": 22.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788094800000, + "T": 1788098399999, + "o": 60162.5, + "h": 60174.5, + "l": 60159.25, + "c": 60170.0, + "v": 23.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788098400000, + "T": 1788101999999, + "o": 60175.0, + "h": 60187.0, + "l": 60171.75, + "c": 60182.5, + "v": 24.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788102000000, + "T": 1788105599999, + "o": 60187.5, + "h": 60192.0, + "l": 60175.0, + "c": 60178.25, + "v": 25.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788105600000, + "T": 1788109199999, + "o": 60200.0, + "h": 60212.0, + "l": 60196.75, + "c": 60207.5, + "v": 26.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788109200000, + "T": 1788112799999, + "o": 60212.5, + "h": 60224.5, + "l": 60209.25, + "c": 60220.0, + "v": 27.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788112800000, + "T": 1788116399999, + "o": 60225.0, + "h": 60229.5, + "l": 60212.5, + "c": 60215.75, + "v": 28.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788116400000, + "T": 1788119999999, + "o": 60237.5, + "h": 60249.5, + "l": 60234.25, + "c": 60245.0, + "v": 29.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788120000000, + "T": 1788123599999, + "o": 60250.0, + "h": 60262.0, + "l": 60246.75, + "c": 60257.5, + "v": 30.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788123600000, + "T": 1788127199999, + "o": 60262.5, + "h": 60267.0, + "l": 60250.0, + "c": 60253.25, + "v": 31.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788127200000, + "T": 1788130799999, + "o": 60275.0, + "h": 60287.0, + "l": 60271.75, + "c": 60282.5, + "v": 32.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788130800000, + "T": 1788134399999, + "o": 60287.5, + "h": 60299.5, + "l": 60284.25, + "c": 60295.0, + "v": 33.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788134400000, + "T": 1788137999999, + "o": 60300.0, + "h": 60304.5, + "l": 60287.5, + "c": 60290.75, + "v": 34.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788138000000, + "T": 1788141599999, + "o": 60312.5, + "h": 60324.5, + "l": 60309.25, + "c": 60320.0, + "v": 35.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788141600000, + "T": 1788145199999, + "o": 60325.0, + "h": 60337.0, + "l": 60321.75, + "c": 60332.5, + "v": 36.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788145200000, + "T": 1788148799999, + "o": 60337.5, + "h": 60342.0, + "l": 60325.0, + "c": 60328.25, + "v": 37.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788148800000, + "T": 1788152399999, + "o": 60350.0, + "h": 60362.0, + "l": 60346.75, + "c": 60357.5, + "v": 38.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788152400000, + "T": 1788155999999, + "o": 60362.5, + "h": 60374.5, + "l": 60359.25, + "c": 60370.0, + "v": 39.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788156000000, + "T": 1788159599999, + "o": 60375.0, + "h": 60379.5, + "l": 60362.5, + "c": 60365.75, + "v": 40.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788159600000, + "T": 1788163199999, + "o": 60387.5, + "h": 60399.5, + "l": 60384.25, + "c": 60395.0, + "v": 41.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788163200000, + "T": 1788166799999, + "o": 60400.0, + "h": 60412.0, + "l": 60396.75, + "c": 60407.5, + "v": 42.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788166800000, + "T": 1788170399999, + "o": 60412.5, + "h": 60417.0, + "l": 60400.0, + "c": 60403.25, + "v": 43.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788170400000, + "T": 1788173999999, + "o": 60425.0, + "h": 60437.0, + "l": 60421.75, + "c": 60432.5, + "v": 44.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788174000000, + "T": 1788177599999, + "o": 60437.5, + "h": 60449.5, + "l": 60434.25, + "c": 60445.0, + "v": 45.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788177600000, + "T": 1788181199999, + "o": 60450.0, + "h": 60454.5, + "l": 60437.5, + "c": 60440.75, + "v": 46.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788181200000, + "T": 1788184799999, + "o": 60462.5, + "h": 60474.5, + "l": 60459.25, + "c": 60470.0, + "v": 47.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788184800000, + "T": 1788188399999, + "o": 60475.0, + "h": 60487.0, + "l": 60471.75, + "c": 60482.5, + "v": 48.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788188400000, + "T": 1788191999999, + "o": 60487.5, + "h": 60492.0, + "l": 60475.0, + "c": 60478.25, + "v": 49.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788192000000, + "T": 1788195599999, + "o": 60500.0, + "h": 60512.0, + "l": 60496.75, + "c": 60507.5, + "v": 50.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788195600000, + "T": 1788199199999, + "o": 60512.5, + "h": 60524.5, + "l": 60509.25, + "c": 60520.0, + "v": 51.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788199200000, + "T": 1788202799999, + "o": 60525.0, + "h": 60529.5, + "l": 60512.5, + "c": 60515.75, + "v": 52.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788202800000, + "T": 1788206399999, + "o": 60537.5, + "h": 60549.5, + "l": 60534.25, + "c": 60545.0, + "v": 53.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788206400000, + "T": 1788209999999, + "o": 60550.0, + "h": 60562.0, + "l": 60546.75, + "c": 60557.5, + "v": 54.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788210000000, + "T": 1788213599999, + "o": 60562.5, + "h": 60567.0, + "l": 60550.0, + "c": 60553.25, + "v": 55.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788213600000, + "T": 1788217199999, + "o": 60575.0, + "h": 60587.0, + "l": 60571.75, + "c": 60582.5, + "v": 56.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788217200000, + "T": 1788220799999, + "o": 60587.5, + "h": 60599.5, + "l": 60584.25, + "c": 60595.0, + "v": 57.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788220800000, + "T": 1788224399999, + "o": 61237.5, + "h": 61242.0, + "l": 61225.0, + "c": 61228.25, + "v": 109.0, + "s": "BTC", + "i": "1h", + "n": 5 + }, + { + "t": 1788066000000, + "T": 1788069599999, + "o": 60062.5, + "h": 60074.5, + "l": 60059.25, + "c": 60070.0, + "v": 15.0, + "s": "BTC", + "i": "1h", + "n": 5 + } + ], + "4h": [ + { + "t": 1788048000000, + "T": 1788062399999, + "o": 60000.0, + "h": 60004.5, + "l": 59987.5, + "c": 59990.75, + "v": 10.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788062400000, + "T": 1788076799999, + "o": 60012.5, + "h": 60024.5, + "l": 60009.25, + "c": 60020.0, + "v": 11.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788076800000, + "T": 1788091199999, + "o": 60025.0, + "h": 60037.0, + "l": 60021.75, + "c": 60032.5, + "v": 12.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788091200000, + "T": 1788105599999, + "o": 60037.5, + "h": 60042.0, + "l": 60025.0, + "c": 60028.25, + "v": 13.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788105600000, + "T": 1788119999999, + "o": 60050.0, + "h": 60062.0, + "l": 60046.75, + "c": 60057.5, + "v": 14.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788120000000, + "T": 1788134399999, + "o": 60062.5, + "h": 60074.5, + "l": 60059.25, + "c": 60070.0, + "v": 15.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788134400000, + "T": 1788148799999, + "o": 60075.0, + "h": 60079.5, + "l": 60062.5, + "c": 60065.75, + "v": 16.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788148800000, + "T": 1788163199999, + "o": 60087.5, + "h": 60099.5, + "l": 60084.25, + "c": 60095.0, + "v": 17.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788163200000, + "T": 1788177599999, + "o": 60100.0, + "h": 60112.0, + "l": 60096.75, + "c": 60107.5, + "v": 18.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788177600000, + "T": 1788191999999, + "o": 60112.5, + "h": 60117.0, + "l": 60100.0, + "c": 60103.25, + "v": 19.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788192000000, + "T": 1788206399999, + "o": 60125.0, + "h": 60137.0, + "l": 60121.75, + "c": 60132.5, + "v": 20.0, + "s": "BTC", + "i": "4h", + "n": 5 + }, + { + "t": 1788206400000, + "T": 1788220799999, + "o": 60137.5, + "h": 60149.5, + "l": 60134.25, + "c": 60145.0, + "v": 21.0, + "s": "BTC", + "i": "4h", + "n": 5 + } + ] + }, + "ETH": { + "1h": [ + { + "t": 1788134400000, + "T": 1788137999999, + "o": 60000.0, + "h": 60004.5, + "l": 59987.5, + "c": 59990.75, + "v": 10.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788138000000, + "T": 1788141599999, + "o": 60012.5, + "h": 60024.5, + "l": 60009.25, + "c": 60020.0, + "v": 11.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788141600000, + "T": 1788145199999, + "o": 60025.0, + "h": 60037.0, + "l": 60021.75, + "c": 60032.5, + "v": 12.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788145200000, + "T": 1788148799999, + "o": 60037.5, + "h": 60042.0, + "l": 60025.0, + "c": 60028.25, + "v": 13.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788148800000, + "T": 1788152399999, + "o": 60050.0, + "h": 60062.0, + "l": 60046.75, + "c": 60057.5, + "v": 14.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788152400000, + "T": 1788155999999, + "o": 60062.5, + "h": 60074.5, + "l": 60059.25, + "c": 60070.0, + "v": 15.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788156000000, + "T": 1788159599999, + "o": 60075.0, + "h": 60079.5, + "l": 60062.5, + "c": 60065.75, + "v": 16.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788159600000, + "T": 1788163199999, + "o": 60087.5, + "h": 60099.5, + "l": 60084.25, + "c": 60095.0, + "v": 17.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788163200000, + "T": 1788166799999, + "o": 60100.0, + "h": 60112.0, + "l": 60096.75, + "c": 60107.5, + "v": 18.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788166800000, + "T": 1788170399999, + "o": 60112.5, + "h": 60117.0, + "l": 60100.0, + "c": 60103.25, + "v": 19.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788170400000, + "T": 1788173999999, + "o": 60125.0, + "h": 60137.0, + "l": 60121.75, + "c": 60132.5, + "v": 20.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788174000000, + "T": 1788177599999, + "o": 60137.5, + "h": 60149.5, + "l": 60134.25, + "c": 60145.0, + "v": 21.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788177600000, + "T": 1788181199999, + "o": 60150.0, + "h": 60154.5, + "l": 60137.5, + "c": 60140.75, + "v": 22.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788181200000, + "T": 1788184799999, + "o": 60162.5, + "h": 60174.5, + "l": 60159.25, + "c": 60170.0, + "v": 23.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788184800000, + "T": 1788188399999, + "o": 60175.0, + "h": 60187.0, + "l": 60171.75, + "c": 60182.5, + "v": 24.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788188400000, + "T": 1788191999999, + "o": 60187.5, + "h": 60192.0, + "l": 60175.0, + "c": 60178.25, + "v": 25.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788192000000, + "T": 1788195599999, + "o": 60200.0, + "h": 60212.0, + "l": 60196.75, + "c": 60207.5, + "v": 26.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788195600000, + "T": 1788199199999, + "o": 60212.5, + "h": 60224.5, + "l": 60209.25, + "c": 60220.0, + "v": 27.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788199200000, + "T": 1788202799999, + "o": 60225.0, + "h": 60229.5, + "l": 60212.5, + "c": 60215.75, + "v": 28.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788202800000, + "T": 1788206399999, + "o": 60237.5, + "h": 60249.5, + "l": 60234.25, + "c": 60245.0, + "v": 29.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788206400000, + "T": 1788209999999, + "o": 60250.0, + "h": 60262.0, + "l": 60246.75, + "c": 60257.5, + "v": 30.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788210000000, + "T": 1788213599999, + "o": 60262.5, + "h": 60267.0, + "l": 60250.0, + "c": 60253.25, + "v": 31.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788213600000, + "T": 1788217199999, + "o": 60275.0, + "h": 60287.0, + "l": 60271.75, + "c": 60282.5, + "v": 32.0, + "s": "ETH", + "i": "1h", + "n": 5 + }, + { + "t": 1788217200000, + "T": 1788220799999, + "o": 60287.5, + "h": 60299.5, + "l": 60284.25, + "c": 60295.0, + "v": 33.0, + "s": "ETH", + "i": "1h", + "n": 5 + } + ] + } +} diff --git a/ingest/ingest.py b/ingest/ingest.py new file mode 100644 index 0000000..49d9300 --- /dev/null +++ b/ingest/ingest.py @@ -0,0 +1,188 @@ +"""Fetch closed candles from one documented public endpoint and persist them. + +SOURCE: Hyperliquid candleSnapshot, the same endpoint the browser already uses: +https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/info-endpoint + +Run once, or with --every to keep running. Every attempt is recorded, including +failures, so a quiet ingester cannot look like a successful one. +""" +import argparse +import json +import os +import sys +import time +import urllib.error +import urllib.request +from datetime import datetime, timezone + +import psycopg + +from .candles import INTERVAL_MS, CandleError, closed_candles +from .store import apply_schema, read_recent, record_run, upsert_candles + +ENDPOINT = "https://api.hyperliquid.xyz/info" +# SOURCE: the markets already offered by the application's public selector. +SYMBOLS = ("BTC", "ETH", "SOL") +# GUESS: UNCALIBRATED GUESS — request window in bars, matching the existing +# display bound in app/publicSnapshot.ts. Not a trading lookback. +BARS = 300 +# GUESS: UNCALIBRATED GUESS — bounded retry budget and waits for a restarting +# database or a refused upstream connection. Not a measured availability figure. +ATTEMPTS = 4 +BACKOFF_SECONDS = 2.0 +TIMEOUT_SECONDS = 15.0 + + +def _milliseconds(started): + return int(round((time.perf_counter() - started) * 1000)) + + +def fetch_snapshot(symbol, interval, now_ms, opener=urllib.request.urlopen): + """POST one candleSnapshot request, retrying a refused or failing upstream. + + Retries cover connection resets and 5xx replies. A 4xx reply is a request + problem and is not retried. + """ + body = json.dumps({ + "type": "candleSnapshot", + "req": { + "coin": symbol, + "interval": interval, + "startTime": now_ms - INTERVAL_MS[interval] * BARS, + "endTime": now_ms, + }, + }).encode("utf-8") + last_error = None + for attempt in range(ATTEMPTS): + request = urllib.request.Request( + ENDPOINT, data=body, headers={"Content-Type": "application/json"}, method="POST") + try: + with opener(request, timeout=TIMEOUT_SECONDS) as response: + return json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as error: + last_error = error + if error.code < 500: + raise + except (urllib.error.URLError, TimeoutError, ConnectionError, json.JSONDecodeError) as error: + last_error = error + if attempt + 1 < ATTEMPTS: + time.sleep(BACKOFF_SECONDS * (attempt + 1)) + raise RuntimeError(f"The candle endpoint did not answer after {ATTEMPTS} attempts: {last_error}") + + +def connect(dsn, attempts=ATTEMPTS): + """Open a connection, waiting for a database that is still starting up.""" + last_error = None + for attempt in range(attempts): + try: + return psycopg.connect(dsn, connect_timeout=int(TIMEOUT_SECONDS)) + except psycopg.OperationalError as error: + last_error = error + if attempt + 1 < attempts: + time.sleep(BACKOFF_SECONDS * (attempt + 1)) + raise RuntimeError(f"The database did not accept a connection after {attempts} attempts: {last_error}") + + +def load_fixture(path): + """Read recorded upstream payloads keyed by symbol and interval.""" + with open(path, encoding="utf-8") as handle: + return json.load(handle) + + +def ingest_pair(connection, symbol, interval, now_ms, fixture=None, opener=urllib.request.urlopen): + """Fetch, validate and persist one market and interval; always record the attempt.""" + started_at = datetime.now(timezone.utc) + source = "fixture" if fixture is not None else ENDPOINT + fetch_ms = write_ms = None + counts = {"seen": 0, "inserted": 0, "updated": 0, "unchanged": 0} + error = None + try: + fetch_started = time.perf_counter() + if fixture is not None: + payload = fixture.get(symbol, {}).get(interval) + if payload is None: + raise CandleError(f"The fixture has no {symbol} {interval} payload.") + else: + payload = fetch_snapshot(symbol, interval, now_ms, opener) + fetch_ms = _milliseconds(fetch_started) + rows = closed_candles(payload, now_ms, interval) + write_started = time.perf_counter() + counts = upsert_candles(connection, symbol, interval, rows) + connection.commit() + write_ms = _milliseconds(write_started) + except Exception as failure: # recorded, then reported to the caller + connection.rollback() + error = f"{type(failure).__name__}: {failure}" + record_run( + connection, symbol=symbol, interval=interval, source=source, + started_at=started_at, finished_at=datetime.now(timezone.utc), + fetch_ms=fetch_ms, write_ms=write_ms, rows_seen=counts["seen"], + rows_inserted=counts["inserted"], rows_updated=counts["updated"], + rows_unchanged=counts["unchanged"], error=error) + connection.commit() + return { + "symbol": symbol, "interval": interval, "source": source, + "fetch_ms": fetch_ms, "write_ms": write_ms, **counts, "error": error, + } + + +def run_once(connection, symbols, intervals, now_ms, fixture=None, opener=urllib.request.urlopen): + results = [] + for symbol in symbols: + for interval in intervals: + result = ingest_pair(connection, symbol, interval, now_ms, fixture, opener) + results.append(result) + print(json.dumps(result), flush=True) + return results + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--database-url", default=os.environ.get("DATABASE_URL"), + help="PostgreSQL connection string; defaults to DATABASE_URL") + parser.add_argument("--symbols", default=",".join(SYMBOLS), + help=f"comma-separated markets from {', '.join(SYMBOLS)}") + parser.add_argument("--intervals", default="1h,4h,1d", + help=f"comma-separated intervals from {', '.join(INTERVAL_MS)}") + parser.add_argument("--fixture", help="read recorded payloads instead of calling the endpoint") + parser.add_argument("--now", type=int, help="treat this epoch-millisecond value as now") + parser.add_argument("--every", type=float, + help="keep running, waiting this many seconds between passes") + parser.add_argument("--show-stored", action="store_true", + help="after ingesting, print how many candles are stored per pair") + arguments = parser.parse_args(argv) + if not arguments.database_url: + parser.error("set --database-url or DATABASE_URL") + symbols = [value.strip() for value in arguments.symbols.split(",") if value.strip()] + intervals = [value.strip() for value in arguments.intervals.split(",") if value.strip()] + unknown = [value for value in intervals if value not in INTERVAL_MS] + if unknown: + parser.error(f"unsupported intervals: {', '.join(unknown)}") + if arguments.fixture is None and [value for value in symbols if value not in SYMBOLS]: + parser.error(f"live symbols must come from {', '.join(SYMBOLS)}") + + fixture = load_fixture(arguments.fixture) if arguments.fixture else None + connection = connect(arguments.database_url) + apply_schema(connection) + failures = 0 + try: + while True: + now_ms = arguments.now if arguments.now is not None else int(time.time() * 1000) + results = run_once(connection, symbols, intervals, now_ms, fixture) + failures = sum(1 for result in results if result["error"]) + if arguments.show_stored: + for symbol in symbols: + for interval in intervals: + stored = read_recent(connection, symbol, interval, BARS) + print(json.dumps({"symbol": symbol, "interval": interval, + "stored": len(stored)}), flush=True) + if not arguments.every: + break + time.sleep(arguments.every) + finally: + connection.close() + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ingest/requirements.txt b/ingest/requirements.txt new file mode 100644 index 0000000..8b99bee --- /dev/null +++ b/ingest/requirements.txt @@ -0,0 +1,2 @@ +# SOURCE: psycopg 3 with the bundled libpq, so the image needs no build toolchain. +psycopg[binary]==3.3.5 diff --git a/ingest/schema.sql b/ingest/schema.sql new file mode 100644 index 0000000..e861976 --- /dev/null +++ b/ingest/schema.sql @@ -0,0 +1,38 @@ +-- SOURCE: a closed candle is identified by its market, interval and opening time, +-- so that pair is the natural primary key and the duplicate guard. +CREATE TABLE IF NOT EXISTS candles ( + symbol text NOT NULL, + interval text NOT NULL, + open_time bigint NOT NULL, + close_time bigint NOT NULL, + "open" double precision NOT NULL, + "high" double precision NOT NULL, + "low" double precision NOT NULL, + "close" double precision NOT NULL, + volume double precision NOT NULL, + ingested_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (symbol, interval, open_time) +); + +-- SOURCE: the query path reads the most recent candles for one market and +-- interval in opening order, which this index serves directly. +CREATE INDEX IF NOT EXISTS candles_recent + ON candles (symbol, interval, open_time DESC); + +-- Timings recorded here are wall-clock measurements around the fetch and write +-- calls in ingest.py. They are not exchange latency or throughput guarantees. +CREATE TABLE IF NOT EXISTS ingest_runs ( + id bigserial PRIMARY KEY, + symbol text NOT NULL, + interval text NOT NULL, + source text NOT NULL, + started_at timestamptz NOT NULL, + finished_at timestamptz NOT NULL, + fetch_ms integer, + write_ms integer, + rows_seen integer NOT NULL DEFAULT 0, + rows_inserted integer NOT NULL DEFAULT 0, + rows_updated integer NOT NULL DEFAULT 0, + rows_unchanged integer NOT NULL DEFAULT 0, + error text +); diff --git a/ingest/store.py b/ingest/store.py new file mode 100644 index 0000000..e7f8c16 --- /dev/null +++ b/ingest/store.py @@ -0,0 +1,93 @@ +"""PostgreSQL persistence for validated candles. + +Storing a candle twice must not create a second row, and re-reading an +unchanged candle must not be reported as new work. Both are decided by the +database in one statement rather than by a prior read in the ingester. +""" +from pathlib import Path + +SCHEMA_PATH = Path(__file__).with_name("schema.sql") + +# The conflict target is the primary key, so a repeated candle updates in place. +# The WHERE clause suppresses no-op updates, so an unchanged candle returns no +# row at all and is counted as unchanged. +# SOURCE: PostgreSQL exposes xmax = 0 for a row inserted by this statement, +# which is how insert and update are told apart. It is a storage-level detail, +# so the counts below describe this statement only, not the whole table. +UPSERT = """ +INSERT INTO candles (symbol, interval, open_time, close_time, "open", "high", "low", "close", volume) +SELECT %(symbol)s, %(interval)s, u.open_time, u.close_time, u."open", u."high", u."low", u."close", u.volume + FROM unnest( + %(open_time)s::bigint[], %(close_time)s::bigint[], + %(open)s::double precision[], %(high)s::double precision[], + %(low)s::double precision[], %(close)s::double precision[], + %(volume)s::double precision[] + ) AS u(open_time, close_time, "open", "high", "low", "close", volume) + ON CONFLICT (symbol, interval, open_time) DO UPDATE + SET close_time = EXCLUDED.close_time, + "open" = EXCLUDED."open", "high" = EXCLUDED."high", "low" = EXCLUDED."low", + "close" = EXCLUDED."close", volume = EXCLUDED.volume, + ingested_at = now() + WHERE (candles.close_time, candles."open", candles."high", candles."low", candles."close", candles.volume) + IS DISTINCT FROM + (EXCLUDED.close_time, EXCLUDED."open", EXCLUDED."high", EXCLUDED."low", EXCLUDED."close", EXCLUDED.volume) + RETURNING (xmax = 0) AS inserted +""" + +SELECT_RECENT = """ +SELECT open_time, close_time, "open", "high", "low", "close", volume + FROM candles + WHERE symbol = %(symbol)s AND interval = %(interval)s + ORDER BY open_time DESC + LIMIT %(limit)s +""" + +RECORD_RUN = """ +INSERT INTO ingest_runs (symbol, interval, source, started_at, finished_at, fetch_ms, write_ms, + rows_seen, rows_inserted, rows_updated, rows_unchanged, error) +VALUES (%(symbol)s, %(interval)s, %(source)s, %(started_at)s, %(finished_at)s, %(fetch_ms)s, + %(write_ms)s, %(rows_seen)s, %(rows_inserted)s, %(rows_updated)s, %(rows_unchanged)s, %(error)s) +RETURNING id +""" + + +def apply_schema(connection): + """Create the tables and index if they do not exist. Safe to run repeatedly.""" + with connection.cursor() as cursor: + cursor.execute(SCHEMA_PATH.read_text(encoding="utf-8")) + connection.commit() + + +def upsert_candles(connection, symbol, interval, rows): + """Persist validated candles and report what actually changed.""" + if not rows: + return {"seen": 0, "inserted": 0, "updated": 0, "unchanged": 0} + parameters = {"symbol": symbol, "interval": interval} + for field in ("open_time", "close_time", "open", "high", "low", "close", "volume"): + parameters[field] = [row[field] for row in rows] + with connection.cursor() as cursor: + cursor.execute(UPSERT, parameters) + written = [record[0] for record in cursor.fetchall()] + inserted = sum(1 for flag in written if flag) + return { + "seen": len(rows), + "inserted": inserted, + "updated": len(written) - inserted, + "unchanged": len(rows) - len(written), + } + + +def read_recent(connection, symbol, interval, limit): + """Return stored candles in opening order, oldest first.""" + with connection.cursor() as cursor: + cursor.execute(SELECT_RECENT, {"symbol": symbol, "interval": interval, "limit": limit}) + records = cursor.fetchall() + fields = ("open_time", "close_time", "open", "high", "low", "close", "volume") + return [dict(zip(fields, record)) for record in reversed(records)] + + +def record_run(connection, **run): + """Store one ingest attempt, including failures, so runs stay auditable.""" + with connection.cursor() as cursor: + cursor.execute(RECORD_RUN, run) + return cursor.fetchone()[0] diff --git a/ingest/tests/__init__.py b/ingest/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ingest/tests/test_candles.py b/ingest/tests/test_candles.py new file mode 100644 index 0000000..fc7e55a --- /dev/null +++ b/ingest/tests/test_candles.py @@ -0,0 +1,83 @@ +"""Boundary rules for candles entering the database.""" +import json +import unittest +from pathlib import Path + +from ingest.candles import CandleError, closed_candles + +FIXTURE = json.loads((Path(__file__).parents[1] / "fixtures/public-candles.json").read_text(encoding="utf-8")) +NOW_MS = 1_788_220_800_000 +HOUR_MS = 3_600_000 + + +def hourly(open_time, **overrides): + row = {"t": open_time, "T": open_time + HOUR_MS - 1, "o": 100.0, "h": 104.5, + "l": 96.5, "c": 101.0, "v": 12.0} + row.update(overrides) + return row + + +class ClosedCandles(unittest.TestCase): + def test_recorded_payload_drops_forming_and_collapses_duplicates(self): + payload = FIXTURE["BTC"]["1h"] + rows = closed_candles(payload, NOW_MS, "1h") + self.assertEqual(len(payload), 50) + self.assertEqual(len(rows), 48) + self.assertTrue(all(row["close_time"] <= NOW_MS for row in rows)) + + def test_rows_are_sorted_by_opening_time(self): + rows = closed_candles([hourly(NOW_MS - HOUR_MS * 2), hourly(NOW_MS - HOUR_MS * 5)], NOW_MS, "1h") + self.assertEqual([row["open_time"] for row in rows], + [NOW_MS - HOUR_MS * 5, NOW_MS - HOUR_MS * 2]) + + def test_close_time_is_exclusive(self): + opening = NOW_MS - HOUR_MS * 3 + row = closed_candles([hourly(opening)], NOW_MS, "1h")[0] + self.assertEqual(row["close_time"], opening + HOUR_MS) + + def test_numeric_strings_are_accepted(self): + row = closed_candles([hourly(NOW_MS - HOUR_MS, o="100.5", h="104", l="99", c="103", v="7")], + NOW_MS, "1h")[0] + self.assertEqual((row["open"], row["volume"]), (100.5, 7.0)) + + def test_conflicting_rows_sharing_an_opening_time_are_rejected(self): + opening = NOW_MS - HOUR_MS * 4 + with self.assertRaises(CandleError): + closed_candles([hourly(opening), hourly(opening, c=102.0)], NOW_MS, "1h") + + def test_interval_mismatch_is_rejected(self): + opening = NOW_MS - HOUR_MS * 4 + with self.assertRaises(CandleError): + closed_candles([{**hourly(opening), "T": opening + HOUR_MS}], NOW_MS, "1h") + + def test_inconsistent_bounds_are_rejected(self): + opening = NOW_MS - HOUR_MS * 4 + for overrides in ({"h": 99.0}, {"l": 101.5}, {"l": 0.0}, {"v": -1.0}): + with self.subTest(overrides=overrides): + with self.assertRaises(CandleError): + closed_candles([hourly(opening, **overrides)], NOW_MS, "1h") + + def test_non_numeric_values_are_rejected(self): + opening = NOW_MS - HOUR_MS * 4 + for value in (None, True, "", "abc", [], {}): + with self.subTest(value=value): + with self.assertRaises(CandleError): + closed_candles([hourly(opening, c=value)], NOW_MS, "1h") + + def test_malformed_responses_are_rejected(self): + for payload in ({"t": 0}, "candles", None, [None], [[]]): + with self.subTest(payload=payload): + with self.assertRaises(CandleError): + closed_candles(payload, NOW_MS, "1h") + + def test_a_response_with_only_forming_candles_is_rejected(self): + with self.assertRaises(CandleError): + closed_candles([hourly(NOW_MS)], NOW_MS, "1h") + + def test_unsupported_interval_is_rejected(self): + with self.assertRaises(CandleError): + closed_candles([hourly(NOW_MS - HOUR_MS)], NOW_MS, "3h") + + +if __name__ == "__main__": + unittest.main() diff --git a/ingest/tests/test_ingest.py b/ingest/tests/test_ingest.py new file mode 100644 index 0000000..10d985d --- /dev/null +++ b/ingest/tests/test_ingest.py @@ -0,0 +1,158 @@ +"""Retry behaviour and attempt recording. + +These tests cover control flow with a stub connection. They do not prove the SQL +is correct: that is what the PostgreSQL integration job in continuous +integration checks against a real server. +""" +import io +import json +import unittest +import urllib.error +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import patch + +from ingest import ingest as ingest_module +from ingest.ingest import fetch_snapshot, ingest_pair + +NOW_MS = 1_788_220_800_000 +FIXTURE_PATH = Path(ingest_module.__file__).parent / "fixtures/public-candles.json" + + +@contextmanager +def _reply(payload): + yield io.BytesIO(json.dumps(payload).encode("utf-8")) + + +class StubCursor: + def __init__(self, log): + self.log = log + + def __enter__(self): + return self + + def __exit__(self, *_): + return False + + def execute(self, statement, parameters=None): + self.log.append((statement.strip().split()[0].upper(), parameters)) + + def fetchall(self): + return [] + + def fetchone(self): + return (1,) + + +class StubConnection: + """Records the statements and transaction calls the ingester makes.""" + + def __init__(self): + self.log = [] + self.commits = 0 + self.rollbacks = 0 + + def cursor(self): + return StubCursor(self.log) + + def commit(self): + self.commits += 1 + + def rollback(self): + self.rollbacks += 1 + + def recorded_run(self): + for name, parameters in self.log: + if name == "INSERT" and isinstance(parameters, dict) and "started_at" in parameters: + return parameters + return None + + +class FetchSnapshot(unittest.TestCase): + def test_a_server_error_is_retried_and_can_then_succeed(self): + attempts = [] + + def opener(request, timeout=None): + attempts.append(request.full_url) + if len(attempts) < 3: + raise urllib.error.HTTPError(request.full_url, 503, "busy", {}, None) + return _reply([{"t": 1}]) + + with patch.object(ingest_module.time, "sleep") as sleep: + payload = fetch_snapshot("BTC", "1h", NOW_MS, opener) + self.assertEqual(payload, [{"t": 1}]) + self.assertEqual(len(attempts), 3) + self.assertEqual(sleep.call_count, 2) + + def test_a_dropped_connection_is_retried(self): + attempts = [] + + def opener(request, timeout=None): + attempts.append(1) + if len(attempts) < 2: + raise urllib.error.URLError(ConnectionResetError("reset by peer")) + return _reply([]) + + with patch.object(ingest_module.time, "sleep"): + self.assertEqual(fetch_snapshot("BTC", "1h", NOW_MS, opener), []) + self.assertEqual(len(attempts), 2) + + def test_a_request_error_is_not_retried(self): + attempts = [] + + def opener(request, timeout=None): + attempts.append(1) + raise urllib.error.HTTPError(request.full_url, 422, "bad request", {}, None) + + with patch.object(ingest_module.time, "sleep"): + with self.assertRaises(urllib.error.HTTPError): + fetch_snapshot("BTC", "1h", NOW_MS, opener) + self.assertEqual(len(attempts), 1) + + def test_the_retry_budget_is_bounded(self): + def opener(request, timeout=None): + raise urllib.error.URLError("refused") + + with patch.object(ingest_module.time, "sleep"): + with self.assertRaises(RuntimeError): + fetch_snapshot("BTC", "1h", NOW_MS, opener) + + def test_the_request_asks_for_the_selected_market_and_interval(self): + captured = {} + + def opener(request, timeout=None): + captured.update(json.loads(request.data.decode("utf-8"))) + return _reply([]) + + fetch_snapshot("ETH", "4h", NOW_MS, opener) + self.assertEqual(captured["type"], "candleSnapshot") + self.assertEqual(captured["req"]["coin"], "ETH") + self.assertEqual(captured["req"]["interval"], "4h") + self.assertEqual(captured["req"]["endTime"], NOW_MS) + + +class IngestPair(unittest.TestCase): + def test_a_missing_fixture_pair_is_recorded_as_a_failed_attempt(self): + connection = StubConnection() + result = ingest_pair(connection, "BTC", "1d", NOW_MS, fixture={"BTC": {"1h": []}}) + self.assertIn("CandleError", result["error"]) + self.assertEqual(connection.rollbacks, 1) + run = connection.recorded_run() + self.assertIsNotNone(run) + self.assertEqual(run["rows_seen"], 0) + self.assertIn("CandleError", run["error"]) + + def test_a_successful_pass_records_timings_and_counts(self): + connection = StubConnection() + fixture = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) + result = ingest_pair(connection, "BTC", "1h", NOW_MS, fixture=fixture) + self.assertIsNone(result["error"]) + self.assertEqual(result["source"], "fixture") + self.assertIsInstance(result["fetch_ms"], int) + self.assertIsInstance(result["write_ms"], int) + run = connection.recorded_run() + self.assertEqual(run["rows_seen"], 48) + + +if __name__ == "__main__": + unittest.main() diff --git a/ingest/tests/test_persistence.py b/ingest/tests/test_persistence.py new file mode 100644 index 0000000..5c3c681 --- /dev/null +++ b/ingest/tests/test_persistence.py @@ -0,0 +1,130 @@ +"""Integration checks against a real PostgreSQL server. + +Skipped unless DATABASE_URL is set, so the unit suite stays offline. These are +the tests that prove the SQL, not just the control flow: a repeated candle must +not create a second row, a changed candle must update in place, and a new +connection must still see what an earlier process wrote. +""" +import copy +import json +import os +import unittest +from pathlib import Path + +from ingest.ingest import run_once +from ingest.store import apply_schema, read_recent + +DATABASE_URL = os.environ.get("DATABASE_URL") +FIXTURE_PATH = Path(__file__).parents[1] / "fixtures/public-candles.json" +NOW_MS = 1_788_220_800_000 +# The recorded BTC hourly payload holds 50 rows: 48 closed and unique, one still +# forming, and one exact repeat. See ingest/fixtures/build_fixture.py. +BTC_HOURLY_CLOSED = 48 + + +@unittest.skipUnless(DATABASE_URL, "set DATABASE_URL to run the persistence checks") +class Persistence(unittest.TestCase): + @classmethod + def setUpClass(cls): + import psycopg + + cls.psycopg = psycopg + cls.fixture = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) + + def connect(self): + return self.psycopg.connect(DATABASE_URL) + + def setUp(self): + self.connection = self.connect() + apply_schema(self.connection) + with self.connection.cursor() as cursor: + cursor.execute("TRUNCATE candles, ingest_runs") + self.connection.commit() + self.addCleanup(self.connection.close) + + def stored_count(self, connection, symbol="BTC", interval="1h"): + with connection.cursor() as cursor: + cursor.execute("SELECT count(*) FROM candles WHERE symbol = %s AND interval = %s", + (symbol, interval)) + return cursor.fetchone()[0] + + def ingest(self, fixture=None): + return { + (result["symbol"], result["interval"]): result + for result in run_once(self.connection, ["BTC"], ["1h"], NOW_MS, + fixture if fixture is not None else self.fixture) + } + + def test_a_first_pass_stores_only_closed_unique_candles(self): + result = self.ingest()[("BTC", "1h")] + self.assertIsNone(result["error"]) + self.assertEqual(result["seen"], BTC_HOURLY_CLOSED) + self.assertEqual(result["inserted"], BTC_HOURLY_CLOSED) + self.assertEqual((result["updated"], result["unchanged"]), (0, 0)) + self.assertEqual(self.stored_count(self.connection), BTC_HOURLY_CLOSED) + + def test_repeating_the_same_payload_changes_nothing(self): + self.ingest() + repeated = self.ingest()[("BTC", "1h")] + self.assertEqual(repeated["inserted"], 0) + self.assertEqual(repeated["updated"], 0) + self.assertEqual(repeated["unchanged"], BTC_HOURLY_CLOSED) + self.assertEqual(self.stored_count(self.connection), BTC_HOURLY_CLOSED) + + def test_a_corrected_candle_updates_in_place(self): + self.ingest() + corrected = copy.deepcopy(self.fixture) + target = corrected["BTC"]["1h"][10] + target["c"] = target["h"] + result = self.ingest(corrected)[("BTC", "1h")] + self.assertEqual(result["inserted"], 0) + self.assertEqual(result["updated"], 1) + self.assertEqual(result["unchanged"], BTC_HOURLY_CLOSED - 1) + self.assertEqual(self.stored_count(self.connection), BTC_HOURLY_CLOSED) + rows = read_recent(self.connection, "BTC", "1h", BTC_HOURLY_CLOSED) + self.assertEqual(rows[10]["close"], target["h"]) + + def test_every_attempt_is_recorded_with_its_own_timings(self): + self.ingest() + self.ingest({"BTC": {}}) + with self.connection.cursor() as cursor: + cursor.execute(""" + SELECT source, fetch_ms, write_ms, rows_inserted, error + FROM ingest_runs ORDER BY id + """) + runs = cursor.fetchall() + self.assertEqual(len(runs), 2) + succeeded, failed = runs + self.assertEqual(succeeded[0], "fixture") + self.assertIsNotNone(succeeded[1]) + self.assertIsNotNone(succeeded[2]) + self.assertEqual(succeeded[3], BTC_HOURLY_CLOSED) + self.assertIsNone(succeeded[4]) + self.assertIsNone(failed[2]) + self.assertIn("CandleError", failed[4]) + + def test_a_new_connection_reads_what_an_earlier_process_wrote(self): + self.ingest() + self.connection.close() + # A separate connection stands in for a restarted reader: nothing is held + # in the writer's memory, so this can only come from the database. + with self.connect() as reader: + rows = read_recent(reader, "BTC", "1h", 500) + self.assertEqual(len(rows), BTC_HOURLY_CLOSED) + openings = [row["open_time"] for row in rows] + self.assertEqual(openings, sorted(openings)) + self.assertTrue(all(row["close_time"] <= NOW_MS for row in rows)) + + def test_the_read_bound_limits_rows_without_reordering_them(self): + self.ingest() + rows = read_recent(self.connection, "BTC", "1h", 5) + self.assertEqual(len(rows), 5) + openings = [row["open_time"] for row in rows] + self.assertEqual(openings, sorted(openings)) + # A bounded read returns the newest window, still oldest first. + everything = [row["open_time"] for row in read_recent(self.connection, "BTC", "1h", 500)] + self.assertEqual(openings, everything[-5:]) + + +if __name__ == "__main__": + unittest.main() diff --git a/package-lock.json b/package-lock.json index 326e7b6..ab8bbda 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,12 +10,14 @@ "dependencies": { "lightweight-charts": "5.2.0", "next": "16.2.6", + "pg": "8.23.0", "react": "19.2.6", "react-dom": "19.2.6" }, "devDependencies": { "@cloudflare/vite-plugin": "1.37.1", "@types/node": "22.19.19", + "@types/pg": "8.23.1", "@types/react": "19.2.14", "@types/react-dom": "19.2.3", "@vitejs/plugin-react": "6.0.2", @@ -1359,6 +1361,18 @@ "undici-types": "~6.21.0" } }, + "node_modules/@types/pg": { + "version": "8.23.1", + "resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.23.1.tgz", + "integrity": "sha512-fKVHpikPdg4GKks3JuLEhvwSyvwzF23hnabPy6DD8ljVbC7+6J5dQzdv4arV6jqq57djnMgs1HKBxX4P8aBI3A==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*", + "pg-protocol": "*", + "pg-types": "^2.2.0" + } + }, "node_modules/@types/react": { "version": "19.2.14", "resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.14.tgz", @@ -2602,6 +2616,95 @@ "dev": true, "license": "MIT" }, + "node_modules/pg": { + "version": "8.23.0", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.23.0.tgz", + "integrity": "sha512-Ip2EQCngowJLGOfCwkFhPXU7/ljlhn6Rxlmy4XYfL2Y+vyRM59+8uR2xqRWKdYmbXmxCFOAmKxBuSUCdF34qLg==", + "license": "MIT", + "dependencies": { + "pg-connection-string": "^2.14.0", + "pg-pool": "^3.14.0", + "pg-protocol": "^1.16.0", + "pg-types": "2.2.0", + "pgpass": "1.0.5" + }, + "engines": { + "node": ">= 16.0.0" + }, + "optionalDependencies": { + "pg-cloudflare": "^1.4.0" + }, + "peerDependencies": { + "pg-native": ">=3.0.1" + }, + "peerDependenciesMeta": { + "pg-native": { + "optional": true + } + } + }, + "node_modules/pg-cloudflare": { + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz", + "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==", + "license": "MIT", + "optional": true + }, + "node_modules/pg-connection-string": { + "version": "2.14.0", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz", + "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==", + "license": "MIT" + }, + "node_modules/pg-int8": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz", + "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==", + "license": "ISC", + "engines": { + "node": ">=4.0.0" + } + }, + "node_modules/pg-pool": { + "version": "3.14.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz", + "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==", + "license": "MIT", + "peerDependencies": { + "pg": ">=8.0" + } + }, + "node_modules/pg-protocol": { + "version": "1.16.0", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.16.0.tgz", + "integrity": "sha512-sILXutLVjCLjcDuOmvhX5e2Z4cS5qG/6Bu3VkpFwdf/633ElGLpEh9bgmuI5I4sqKqkifQiGyiCcx1HdtrK7tg==", + "license": "MIT" + }, + "node_modules/pg-types": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz", + "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==", + "license": "MIT", + "dependencies": { + "pg-int8": "1.0.1", + "postgres-array": "~2.0.0", + "postgres-bytea": "~1.0.0", + "postgres-date": "~1.0.4", + "postgres-interval": "^1.1.0" + }, + "engines": { + "node": ">=4" + } + }, + "node_modules/pgpass": { + "version": "1.0.5", + "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz", + "integrity": "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==", + "license": "MIT", + "dependencies": { + "split2": "^4.1.0" + } + }, "node_modules/picocolors": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz", @@ -2657,6 +2760,45 @@ "dev": true, "license": "MIT" }, + "node_modules/postgres-array": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz", + "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/postgres-bytea": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.1.tgz", + "integrity": "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-date": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz", + "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==", + "license": "MIT", + "engines": { + "node": ">=0.10.0" + } + }, + "node_modules/postgres-interval": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz", + "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==", + "license": "MIT", + "dependencies": { + "xtend": "^4.0.0" + }, + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/queue-microtask": { "version": "1.2.3", "resolved": "https://registry.npmjs.org/queue-microtask/-/queue-microtask-1.2.3.tgz", @@ -2979,6 +3121,15 @@ "source-map": "^0.6.0" } }, + "node_modules/split2": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/split2/-/split2-4.2.0.tgz", + "integrity": "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==", + "license": "ISC", + "engines": { + "node": ">= 10.x" + } + }, "node_modules/srvx": { "version": "0.11.15", "resolved": "https://registry.npmjs.org/srvx/-/srvx-0.11.15.tgz", @@ -4493,6 +4644,15 @@ } } }, + "node_modules/xtend": { + "version": "4.0.2", + "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", + "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==", + "license": "MIT", + "engines": { + "node": ">=0.4" + } + }, "node_modules/yoga-layout": { "version": "3.2.1", "resolved": "https://registry.npmjs.org/yoga-layout/-/yoga-layout-3.2.1.tgz", diff --git a/package.json b/package.json index 8f2c1d6..179af96 100644 --- a/package.json +++ b/package.json @@ -15,12 +15,14 @@ "dependencies": { "lightweight-charts": "5.2.0", "next": "16.2.6", + "pg": "8.23.0", "react": "19.2.6", "react-dom": "19.2.6" }, "devDependencies": { "@cloudflare/vite-plugin": "1.37.1", "@types/node": "22.19.19", + "@types/pg": "8.23.1", "@types/react": "19.2.14", "@types/react-dom": "19.2.3", "@vitejs/plugin-react": "6.0.2", diff --git a/scripts/check_persistence.mjs b/scripts/check_persistence.mjs new file mode 100644 index 0000000..e8e15e0 --- /dev/null +++ b/scripts/check_persistence.mjs @@ -0,0 +1,61 @@ +/** Read persisted candles through the real route handler, in a fresh process. + * + * The ingester wrote the rows in an earlier process that has already exited, so + * anything this script can see came out of PostgreSQL. It also puts the stored + * frames through the same validation and replay aggregation the chart uses, + * because rows that the chart would reject are not usable evidence. + * + * DATABASE_URL=... node scripts/check_persistence.mjs [expectedHourlyBars] + */ +import assert from 'node:assert/strict'; +import { candleQuery, closeCandlePool } from '../app/db.ts'; +import { createStoredApi } from '../app/storedApi.ts'; +import { aggregateBars, validateBars } from '../app/marketAnalysis.ts'; + +const expectedHourly = Number(process.argv[2] ?? 48); +const handler = createStoredApi(candleQuery); +const request = symbol => new Request(`http://localhost/api/stored/${symbol}`); +const checks = []; +const check = async (name, assertion) => { await assertion(); checks.push(name); }; + +try { + const response = await handler(request('BTC')); + assert.equal(response.status, 200, 'stored BTC candles should be readable'); + const body = await response.json(); + + await check('rows come from the database', () => { + assert.equal(body.source, 'database'); + assert.equal(body.symbol, 'BTC'); + assert.equal(body.stored['1h'], expectedHourly); + }); + + await check('a handler time was measured', () => { + assert.match(response.headers.get('Server-Timing'), /^query;dur=\d+$/); + }); + + const bars = validateBars(body.frames['1h'], '1h', body.asOf); + await check('stored frames pass the chart validation unchanged', () => { + assert.equal(bars.length, expectedHourly); + const openings = bars.map(bar => bar.t); + assert.deepEqual(openings, [...openings].sort((a, b) => a - b)); + }); + + await check('a replay prefix never aggregates a later candle', () => { + const prefix = bars.slice(0, 12); + const cutoff = prefix.at(-1).closeTime; + const aggregated = aggregateBars(prefix, '1h', '4h', cutoff); + assert.ok(aggregated.length > 0, 'the prefix should form at least one higher-timeframe bar'); + assert.ok(aggregated.every(bar => bar.closeTime <= cutoff)); + const wholePeriod = aggregateBars(bars, '1h', '4h', cutoff); + assert.deepEqual(aggregated, wholePeriod, 'the cutoff, not the input length, decides the result'); + }); + + await check('an unlisted market is still refused', async () => { + assert.equal((await handler(request('candles'))).status, 400); + }); + + for (const name of checks) console.log(`ok ${name}`); + console.log(`\n${checks.length} persistence checks passed against PostgreSQL.`); +} finally { + await closeCandlePool(); +} diff --git a/tests/stored-api.test.mjs b/tests/stored-api.test.mjs new file mode 100644 index 0000000..218c675 --- /dev/null +++ b/tests/stored-api.test.mjs @@ -0,0 +1,72 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { createStoredApi } from '../app/storedApi.ts'; +import { STORED_BARS, STORED_INTERVALS } from '../app/storedMarket.ts'; +import { INTERVAL_MS } from '../app/marketAnalysis.ts'; + +// SOURCE: SYNTHETIC rows and a controlled clock; no database is contacted here. +// The SQL itself is exercised by the PostgreSQL job in continuous integration. +const HOUR = INTERVAL_MS['1h']; +const url = symbol => new Request(`http://localhost/api/stored/${symbol}`); +const storedRow = openTime => ({ + interval: '1h', open_time: openTime, close_time: openTime + HOUR, + open: 100, high: 104.5, low: 96.5, close: 101, volume: 12, +}); + +test('SYNTHETIC: rejects unsupported markets without touching the database', async () => { + let calls = 0; + const handler = createStoredApi(async () => { calls++; return []; }); + const response = await handler(url('arbitrary-table')); + assert.equal(response.status, 400); + assert.equal(calls, 0); +}); + +test('SYNTHETIC: asks for the selected market, the served intervals and a bounded count', async () => { + let received; + const handler = createStoredApi(async (text, values) => { received = { text, values }; return [storedRow(HOUR)]; }); + await handler(url('ETH')); + assert.deepEqual(received.values, ['ETH', STORED_INTERVALS, STORED_BARS]); + assert.match(received.text, /FROM candles/); + assert.match(received.text, /WHERE symbol = \$1 AND interval = ANY\(\$2\)/); +}); + +test('SYNTHETIC: an empty store is a 404 with an instruction, not a fault', async () => { + const handler = createStoredApi(async () => []); + const response = await handler(url('BTC')); + assert.equal(response.status, 404); + assert.match((await response.json()).error, /Run the ingester/); + assert.equal(response.headers.get('Cache-Control'), 'no-store'); +}); + +test('SYNTHETIC: stored candles are returned with counts and a measured handler time', async () => { + let clock = 1_000; + const handler = createStoredApi(async () => { clock += 7; return [storedRow(HOUR), storedRow(HOUR * 2)]; }, () => clock); + const response = await handler(url('BTC')); + assert.equal(response.status, 200); + const body = await response.json(); + assert.equal(body.symbol, 'BTC'); + assert.equal(body.source, 'database'); + assert.equal(body.frames['1h'].length, 2); + assert.deepEqual(body.stored, { '1h': 2 }); + assert.equal(body.asOf, HOUR * 3); + assert.equal(response.headers.get('Server-Timing'), 'query;dur=7'); + assert.equal(response.headers.get('Cache-Control'), 'no-store'); +}); + +test('SYNTHETIC: a database failure is 503 and a later request can recover', async () => { + let offline = true; + const handler = createStoredApi(async () => { + if (offline) throw new Error('connection refused'); + return [storedRow(HOUR)]; + }); + assert.equal((await handler(url('SOL'))).status, 503); + offline = false; + const response = await handler(url('SOL')); + assert.equal(response.status, 200); + assert.equal((await response.json()).symbol, 'SOL'); +}); + +test('SYNTHETIC: an unusable stored row is a fault, not a partial chart', async () => { + const handler = createStoredApi(async () => [{ ...storedRow(HOUR), close_time: HOUR + 5 }]); + assert.equal((await handler(url('BTC'))).status, 503); +}); diff --git a/tests/stored-market.test.mjs b/tests/stored-market.test.mjs new file mode 100644 index 0000000..627faef --- /dev/null +++ b/tests/stored-market.test.mjs @@ -0,0 +1,71 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { toFrames, STORED_BARS, STORED_INTERVALS } from '../app/storedMarket.ts'; +import { INTERVAL_MS } from '../app/marketAnalysis.ts'; + +// SOURCE: SYNTHETIC stored rows shaped like a node-postgres result, not observations. +const HOUR = INTERVAL_MS['1h']; +const row = (openTime, overrides = {}) => ({ + interval: '1h', + open_time: openTime, + close_time: openTime + HOUR, + open: 100, high: 104.5, low: 96.5, close: 101, volume: 12, + ...overrides, +}); + +test('SYNTHETIC: groups rows per interval, sorts them and reports counts', () => { + const { frames, counts, asOf } = toFrames([ + row(HOUR * 3), + row(HOUR), + { ...row(0), interval: '4h', close_time: INTERVAL_MS['4h'] }, + ]); + assert.deepEqual(frames['1h'].map(bar => bar.t), [HOUR, HOUR * 3]); + assert.equal(frames['4h'].length, 1); + assert.deepEqual(counts, { '1h': 2, '4h': 1 }); + // The cutoff is the newest stored close, so every stored bar stays visible. + assert.equal(asOf, HOUR * 4); +}); + +test('SYNTHETIC: bigint columns arriving as strings are accepted within safe range', () => { + const openTime = 1_788_220_800_000; + const { frames } = toFrames([row(openTime, { + open_time: String(openTime), close_time: String(openTime + HOUR), open: '100.25', volume: '3', + })]); + assert.equal(frames['1h'][0].t, openTime); + assert.equal(frames['1h'][0].o, 100.25); + assert.equal(frames['1h'][0].v, 3); +}); + +test('SYNTHETIC: an interval the reader does not serve is skipped, not charted', () => { + const { frames, asOf } = toFrames([{ ...row(0), interval: '5m', close_time: INTERVAL_MS['5m'] }]); + assert.deepEqual(frames, {}); + assert.equal(asOf, 0); + assert.ok(!STORED_INTERVALS.includes('5m')); +}); + +test('SYNTHETIC: one row per opening time survives', () => { + const { frames, counts } = toFrames([row(HOUR), row(HOUR, { close: 103 })]); + assert.equal(counts['1h'], 1); + assert.equal(frames['1h'][0].c, 103); +}); + +test('SYNTHETIC: storage faults are reported instead of being drawn', () => { + const cases = { + 'does not span its interval': row(HOUR, { close_time: HOUR + HOUR / 2 }), + 'unusable timestamp': row(HOUR, { open_time: 1.5 }), + 'unsafe timestamp': row(HOUR, { open_time: '9007199254740993' }), + 'unusable price': row(HOUR, { close: 'not-a-number' }), + 'missing volume': row(HOUR, { volume: null }), + 'high below body': row(HOUR, { high: 99 }), + 'low above body': row(HOUR, { low: 100.5 }), + 'non-positive low': row(HOUR, { low: 0, open: 0.5, close: 0.5, high: 1 }), + 'negative volume': row(HOUR, { volume: -1 }), + }; + for (const [name, candidate] of Object.entries(cases)) { + assert.throws(() => toFrames([candidate]), undefined, name); + } +}); + +test('SYNTHETIC: the read bound is a fixed number of bars per interval', () => { + assert.equal(STORED_BARS, 300); +});