From 8cdb304136275e53f7909985c470b4895b5f642a Mon Sep 17 00:00:00 2001 From: javier Date: Tue, 30 Jun 2026 13:20:15 +0200 Subject: [PATCH 1/2] Make java-qwp dataset-compatible with the Python generator Lets a Python backfill be continued (or taken live) by the Java generator on the same tables, with matching schemas, views, and price continuity. - Cli: add --prefix (verbatim, default qwp_; set "" for the exact Python names), prefix-based table accessors, rename the trades stem to fx_trades so an empty prefix yields fx_trades. Reject --incremental in real-time (real-time always syncs live quotes), and default --create_views to true to match Python. - Views: create the full 11 Python-parity materialized views (core_price OHLC-mid, BBO 1s/1m/1h/1d ladder, market_data OHLC 1m/15m/1d, fx_trades OHLC 1m/1d) with byte-faithful DDL, per-view TTLs, and refresh/deferred clauses; each view gated on its base pool being enabled. - Incremental: seed mid/spread/indicators from the last stored core_price row per symbol (was mid-only from trades) and continue the per-symbol walk from it, so open(seam) == close(prev) for bid/ask/spread/indicators. - Real-time mid-walk drift 5.0 -> 7.0 (faster-than-life stays 5.0), matching the Python real-time path. - Retention parity: base-table STORAGE POLICY under --enterprise and per-view TTLs already mirror Python. - Fix the stale FxUniverse Javadoc (it no longer claims there is no order book / no market_data/core_price tables). - Docs: update both READMEs and add INTERCHANGEABILITY_SPEC.md covering --prefix, the view set, incremental-from-core_price, and the requirement to match EPS/orders rates across engines for a seamless cross-engine dataset. --- README.md | 2 + java-qwp/INTERCHANGEABILITY_SPEC.md | 265 ++++++++++++++++++ java-qwp/README.md | 122 +++++--- .../src/main/java/com/questdb/fxqwp/Cli.java | 31 +- .../java/com/questdb/fxqwp/FxUniverse.java | 11 +- .../com/questdb/fxqwp/QwpTradesGenerator.java | 189 ++++++++++--- 6 files changed, 526 insertions(+), 94 deletions(-) create mode 100644 java-qwp/INTERCHANGEABILITY_SPEC.md diff --git a/README.md b/README.md index acbf68a..b96023f 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,8 @@ This script generates highly realistic, multi-level FX market data and ingests i Designed for **stress testing**, **benchmarking**, and **live demo scenarios**, it supports both wall-clock-paced ("real-time") and maximum-throughput ("faster-than-life") simulation with multi-process orchestration, pip-accurate pricing, WAL backpressure detection, and robust state management. +> **Java sibling (QWP):** [`java-qwp/`](java-qwp/README.md) is a Java generator that ingests the **same three tables and the same materialized views** over QuestDB's QWP (WebSocket) protocol. This Python generator is authoritative; the two are **dataset-compatible** — run the Java one with `--prefix ""` and matching EPS flags and you can backfill with one and continue (or go live) with the other on the same tables. See its README for the interchangeability runbook. + --- ## Features diff --git a/java-qwp/INTERCHANGEABILITY_SPEC.md b/java-qwp/INTERCHANGEABILITY_SPEC.md new file mode 100644 index 0000000..cb44999 --- /dev/null +++ b/java-qwp/INTERCHANGEABILITY_SPEC.md @@ -0,0 +1,265 @@ +# Java ⇄ Python generator interchangeability — change spec + +**Goal.** Make the Java QWP generator and the Python `fx_data_generator.py` produce a +*single consistent dataset* on shared tables, so a demo can ingest with **either** +engine (backfill with one, continue or go live with the other) and downstream +tables, views, and price series stay coherent. + +**Authority.** Python is authoritative. Every divergence is resolved by making Java +match Python, never the reverse. + +**Out of scope.** Identical future price *paths* (RNG differs by design — only the +*character* of the data and *seam continuity* must match). Performance/throughput +parity. No new demo-only or benchmark data modes. + +--- + +## Accepted behavioral contract (not a bug) + +With incremental disabled in real-time on both engines, any **faster-than-life → +real-time** transition snaps to live Yahoo quotes — a one-step price move at that +seam, identical on both engines, by design ("real matches real quotes"). +Seam continuity is preserved for **FTL → FTL** (incremental seed) and +**real-time → real-time** (Yahoo + wall-clock + timestamp resume). + +--- + +## Change 1 — Prefix option + `qwp_trades` → `qwp_fx_trades` + +**Why.** Python's trades table is `fx_trades` (not `trades`); a single prefix can't +reach it unless the trades *stem* is `fx_trades`. A configurable prefix lets the +Java tables collapse onto the exact Python names when empty. + +**Design (verbatim prefix, mirroring the verbatim-suffix rule).** +- New CLI flag `--prefix `, **default `qwp_`** (includes the trailing underscore, + so existing names `qwp_market_data` / `qwp_core_price` are preserved and trades + becomes `qwp_fx_trades`). Empty `--prefix ""` yields the Python names. +- Prefix concatenates verbatim, exactly like `suffix`. The user supplies any + separator. + +**Edits — `Cli.java`:** +- Add field `public String prefix = "qwp_";` next to `suffix` (line ~78). +- Parse `case "prefix": c.prefix = req(args, ++i, raw); break;` next to the `suffix` + case (line ~221). +- Rewrite the three accessors (lines 317–326): + ```java + public String tradesTable() { return prefix + "fx_trades" + suffix; } + public String marketDataTable() { return prefix + "market_data" + suffix; } + public String corePriceTable() { return prefix + "core_price" + suffix; } + ``` +- Update `--help` text (line ~490) and add a `--prefix` line. + +**Name matrix after the change:** + +| stem | `--prefix qwp_` (default) | `--prefix ""` (Python parity) | +|---|---|---| +| trades | `qwp_fx_trades` | `fx_trades` | +| market data | `qwp_market_data` | `market_data` | +| core price | `qwp_core_price` | `core_price` | + +**⚠️ Confirm:** default prefix is `qwp_` *with* the underscore. If you'd rather the +flag value be the bare token `qwp` and have the code insert the underscore, say so — +but verbatim (so the user owns the separator) is consistent with how `suffix` +already works. + +--- + +## Change 2 — Port Python's exact 11 materialized views + +**Why.** Tables/views are `IF NOT EXISTS`; the first generator to run dictates the +definition and the second silently inherits it. Consistency requires the Java DDL to +be **byte-identical** to Python's (modulo prefix/suffix substitution). + +**Current Java (3 views, to be replaced):** `market_data_ohlc_1m`, +`market_data_ohlc_15m`, `bbo_1h` — different refresh policies and cascade bases than +Python. Replace the whole `createViews(...)` set (QwpTradesGenerator.java ~lines +1016–1039) with the 11 below. + +**Target set (names shown with PREFIX/SUFFIX placeholders):** + +From `core_price`: +- `{P}core_price_1s{S}` — `SAMPLE BY 1s`, `PARTITION BY HOUR` +- `{P}core_price_1d{S}` — `REFRESH EVERY 1h DEFERRED START '2025-06-01T00:00:00.000000Z'`, `SAMPLE BY 1d`, `PARTITION BY MONTH` + (both: `first/max/min/last` of `(bid_price+ask_price)/2` as open/high/low/close_mid, + `last(ask_price)-last(bid_price)` as last_spread, plus max/min/avg of bid and ask) + +From `market_data`: +- `{P}bbo_1s{S}` — `last(best_bid) bid, last(best_ask) ask`, `SAMPLE BY 1s`, `PARTITION BY HOUR` +- `{P}bbo_1m{S}` — base `bbo_1s`, `max(bid) bid, min(ask) ask`, `REFRESH EVERY 1m DEFERRED START '2025-06-01…'`, `SAMPLE BY 1m`, `PARTITION BY DAY` +- `{P}bbo_1h{S}` — base `bbo_1m`, same agg, `REFRESH EVERY 10m DEFERRED START …`, `SAMPLE BY 1h`, `PARTITION BY MONTH` +- `{P}bbo_1d{S}` — base `bbo_1h`, same agg, `REFRESH EVERY 1h DEFERRED START …`, `SAMPLE BY 1d`, `PARTITION BY MONTH` +- `{P}market_data_ohlc_1m{S}` — `first/max/min/last(best_bid)`, `SUM(bids[2][1]) total_volume`, `SAMPLE BY 1m`, `PARTITION BY HOUR` +- `{P}market_data_ohlc_15m{S}` — base `…_ohlc_1m`, `first(open)/max(high)/min(low)/last(close)`, `SUM(total_volume)`, `SAMPLE BY 15m`, `PARTITION BY HOUR` +- `{P}market_data_ohlc_1d{S}` — `first/max/min/last(best_bid)`, `SUM(bids[2][1])`, `REFRESH EVERY 1h DEFERRED START …`, `SAMPLE BY 1d`, `PARTITION BY MONTH` + +From `fx_trades`: +- `{P}fx_trades_ohlc_1m{S}` — `first/max/min/last(price)`, `SUM(quantity) total_volume`, `SAMPLE BY 1m`, `PARTITION BY HOUR` +- `{P}fx_trades_ohlc_1d{S}` — base `…_ohlc_1m`, `first(open)/max(high)/min(low)/last(close)`, `SUM(total_volume)`, `REFRESH EVERY 1h DEFERRED START …`, `SAMPLE BY 1d`, `PARTITION BY MONTH` + +**Rules:** +- Reproduce Python's exact column aliases, `REFRESH`/`DEFERRED START` clauses, + `PARTITION BY`, and cascade base tables. The canonical source is + `fx_data_generator.py` lines 454–594 — copy from there, do not paraphrase. +- The bbo and `…_15m`/`…_1d` cascades must reference the *prefixed/suffixed* base + view name, not the raw table. +- Apply TTL exactly as Python does (see Change 7). +- Gate behind the existing `--create_views`; for a clean demo, let **one** engine own + view creation (see Guardrails). + +--- + +## Change 3 — Incremental seed from `core_price`, full state (not mid-only from trades) + +**Why.** Python's `load_initial_state` seeds **bid, ask, spread, indicator1, +indicator2** from `core_price LATEST BY symbol`. Java's `seedInitialMidFromDb` +(QwpTradesGenerator.java lines 1402–1442) reads only `price` from the **trades** +table and restores `initialMid` alone — spread and indicators restart fresh, which +would step at the seam. + +**Edits:** +1. Change the query (line 1411) to: + ```sql + SELECT symbol, bid_price, ask_price, indicator1, indicator2 + FROM LATEST ON timestamp PARTITION BY symbol + ``` + and update the log line (1404) to name `cfg.corePriceTable()`. +2. Add initial-state arrays alongside `initialMid` (line 72/95): + `initialSpreadPips[]`, `initialInd1[]`, `initialInd2[]`. +3. In the batch handler, reconstruct exactly as Python: + - `mid = (bid_price + ask_price) / 2` → `initialMid[i] = quantizeToPip(mid, pip, precision)` + - `spreadPips = round((ask_price - bid_price) / pip)`, clamped 1..8 → `initialSpreadPips[i]` + - `initialInd1[i] = indicator1`, `initialInd2[i] = indicator2` + - keep the bracket reset (`resetBracket(mid*0.99, mid*1.01)`) but base it on the mid. +4. Wire the new arrays into worker init where the walk currently starts from defaults: + - `walkMid[j] = initialMid[g]` already exists (line 406). + - seed the spread walk from `initialSpreadPips[g]` (currently starts at a default). + - seed `walkInd1[j]/walkInd2[j]` from `initialInd1[g]/initialInd2[g]` + (currently start at the default ~0.5). + +**Fallback:** if `core_price` has no rows for a symbol, fall back to the +bracket/default exactly as today (`initialMid[i] <= 0` path, line 137). + +--- + +## Change 4 — Disallow `--incremental` in real-time + +**Why.** Python errors on incremental + real-time and reseeds from Yahoo +(`fx_data_generator.py` line 1304). Java currently *allows* it. Make Java symmetric. + +**Edit (Cli.java validation, near the other `fail(...)` checks ~lines 283–291):** +```java +if (realtime && incremental) { + fail("--incremental is not allowed in real-time mode (real-time syncs from Yahoo)."); +} +``` +Keep the existing `realtime && !noYahoo && !incremental` guard on the Yahoo refresher +(QwpTradesGenerator.java line 162) — it becomes simply `realtime && !noYahoo` once the +above rejects the incompatible combo, but leaving it is harmless and defensive. + +--- + +## Change 5 — Real-time mid-walk drift 7.0 (match Python) + +**Why.** Python evolves the mid with **drift 7.0 pips/sec in real-time** +(`evolve_state_one_tick(..., 7.0)`, line 1076) and **5.0 in FTL**. Java uses **5.0 at +both** call sites, so live Java prices wiggle calmer than live Python. + +**Edit (QwpTradesGenerator.java):** the two `beginSecond(k, 5.0)` calls (lines 441 and +469) — the real-time call site must pass `7.0`; the FTL call site stays `5.0`. +Confirm which of 441/469 is the real-time path and change only that one. Shock prob +(1%) and magnitude (±20 pip) already match — no change there. + +--- + +## Change 6 — Fix stale `FxUniverse` Javadoc + +**Why.** The class header (FxUniverse.java lines 14–18) says "There is no order book +and no `market_data`/`core_price` table." That is outdated — Java creates and emits +both. Rewrite the paragraph to describe the current three-table model so the doc +doesn't mislead future readers. + +--- + +## Change 7 — Retention/TTL parity (so schemas match under all flags) + +**Why.** `IF NOT EXISTS` means the first creator's retention clause wins. To keep +schemas identical regardless of which engine runs first: +- `market_data` / `core_price`: Python applies `STORAGE POLICY(TO REMOTE 1 hour, TO + PARQUET 2 days, DROP LOCAL 3 months)` under `--enterprise`, else `TTL 3 DAYS` under + `--short_ttl`. Java currently uses `TTL 3 DAYS`. Align Java's base-table DDL to the + same conditional. +- `fx_trades`: both `TTL 1 MONTH` (enterprise → storage policy). Confirm Java matches. +- Matviews: Python and Java both use `TTL` (storage policy isn't supported on + matviews). Match Python's per-view TTL. + +**Lower priority** for pure data-content demos (retention tiering doesn't change rows), +but required if "consistent dataset" includes identical table definitions. Flagging +explicitly so it's a conscious choice, not an accident. + +--- + +## Operational guardrails (demo runbook, not code) + +1. **Match the EPS / orders rates across engines (most important).** Per-second row + counts come from `--core_min_eps/--core_max_eps`, + `--market_data_min_eps/--market_data_max_eps`, `--orders_min_per_sec/--orders_max_per_sec` + and `--min_levels/--max_levels` — **not** from the existing data. If the continuing + engine runs at different EPS, row *density* steps at the seam even though prices stay + continuous. (Observed in validation: a low-EPS Python backfill continued by Java at + default EPS jumped ~150× in rows/sec. Prices were continuous; density was not.) Run + both engines with identical rate flags. +2. **One view owner per demo.** Even with byte-identical DDL, let a single engine + create the views in any given environment to remove all risk of a first-creator + mismatch. +3. **Clean-backfill assumption.** The FTL→continuation handoff assumes the backfill + completed so per-table maxes are aligned. The per-pool resume fix absorbs minor + skew; a hard-killed backfill can still leave one table behind. Let backfills finish + before switching engines. +4. **Same `--prefix`/`--suffix`/`--enterprise`/`--short_ttl` on both engines** in a + shared environment, or they won't address the same tables / same schema. + +--- + +## Acceptance criteria + +A run is "interchangeable" when all hold: +1. With `--prefix ""`, Java writes to `fx_trades`, `market_data`, `core_price` and the + 11 views with names identical to Python's. +2. Backfill (FTL) with engine A, continue (FTL, `--incremental`) with engine B → + **zero gaps** and `open == previous close` at the seam for bid/ask/spread/indicators + (verify with `SAMPLE BY 1s FILL(NULL)` + `WHERE col IS NULL`, and a boundary + inspection across the seam second). +3. `--incremental` + real-time is rejected by both engines with the same intent. +4. Live (real-time) data from either engine shows the same per-second volatility + character (drift 7.0). +5. `schema`/`SHOW CREATE TABLE` for all tables and `SHOW CREATE MATERIALIZED VIEW` for + all views are identical (modulo prefix/suffix) whichever engine created them. + +## Verification plan +- Unit/smoke: two-step FTL cap → resume (already exercised for the resume fix), now + also asserting bid/ask/spread/indicator seam continuity, not just timestamps. +- Cross-engine: Python FTL backfill of a fixed window → Java FTL `--incremental` + continuation on the same tables → gap + seam checks above. +- DDL diff: dump `SHOW CREATE …` from a Python-first run and a Java-first run; diff. + +## Resolved decisions +1. Default `--prefix` is `qwp_` (verbatim, like `--suffix`); `--prefix ""` yields the + exact Python names. Trades stem renamed to `fx_trades` so empty prefix gives + `fx_trades` (not `trades`). +2. Retention parity (Change 7) is **in scope** — TTL **and** STORAGE POLICY must match + Python (storage policy on base tables under `--enterprise`; per-view TTL on matviews). +3. `--create_views` default flipped to **true**, matching Python. + +## Validation status — DONE (local QuestDB) +All changes implemented on branch `jv/qwp_python_parity`; compiles clean. Smoke-tested +against a local QuestDB: +- **Isolated DDL:** a throwaway-prefix FTL run created all 3 tables + **all 11 views** + without error, and the views populated. +- **Cross-engine seam:** a real Python backfill (00:00–13:00) continued by Java + (`--prefix "" --incremental`) resumed all three pools contiguously from 13:00:00 with + **zero gap-seconds** in `fx_trades`/`market_data`/`core_price`; `core_price` mid + continuity ≤ 0.5 pip (pip-quantization only); the existing Python matviews + incrementally refreshed across the seam. +- **Findings folded into the runbook:** EPS must be matched across engines (guardrail + 1); the `market_data` best_bid realigns to the `core_price` seed within a few pips at + the seam (same on both engines). diff --git a/java-qwp/README.md b/java-qwp/README.md index 2ca6835..5c79d22 100644 --- a/java-qwp/README.md +++ b/java-qwp/README.md @@ -2,21 +2,27 @@ A synthetic **FX market-data generator** that ingests into QuestDB over **QWP** (the WebSocket binary protocol of the QuestDB Java client), HA-aware across a fleet -of hosts. It can populate three tables, each with its own pool of worker threads: +of hosts. It can populate three tables, each with its own pool of worker threads +(names shown with the default `--prefix qwp_`): -- **`qwp_trades`** — trade prints (schema identical to the Python `fx_trades`). +- **`qwp_fx_trades`** — trade prints (schema identical to the Python `fx_trades`). - **`qwp_market_data`** — full order-book snapshots (identical to the Python `market_data`: `bids`/`asks` as `DOUBLE[][]`, plus `best_bid`/`best_ask`). - **`qwp_core_price`** — top-of-book snapshots with metadata (identical to the Python `core_price`: best `bid_price`/`ask_price` + volumes, `ecn`, `reason`, `indicator1`/`indicator2`). -It is a simplified sibling of the Python FX data generator -(`../fx_data_generator.py`). Differences by design: - -- **Three tables, optional materialized views** — `qwp_trades`, `qwp_market_data`, - `qwp_core_price`; with `--create_views` it also builds three matviews over - `qwp_market_data` (1m + 15m OHLC and an hourly BBO, see below). +It is a sibling of the Python FX data generator (`../fx_data_generator.py`) and is +**dataset-compatible** with it: set `--prefix ""` and the table/view names, schemas, +and the full materialized-view set match the Python generator exactly, so you can +backfill with one and continue with the other on the same tables (see +[Interchangeability with the Python generator](#interchangeability-with-the-python-generator)). +Design points: + +- **Three tables, full Python-parity materialized views** — `fx_trades`, + `market_data`, `core_price`; with `--create_views` (default + **true**) it also builds the same 11 matviews the Python generator creates (BBO + ladder, market-data OHLC, core-price OHLC-mid, and trade OHLC — see below). - **Independent per-table pools:** `--trades_processes` workers feed `qwp_trades`, `--market_data_processes` workers feed `qwp_market_data`, `--core_processes` workers feed `qwp_core_price`. Each pool snake-drafts the symbols across its own @@ -44,7 +50,7 @@ It is a simplified sibling of the Python FX data generator ## Table schemas ```sql -CREATE TABLE IF NOT EXISTS qwp_trades ( +CREATE TABLE IF NOT EXISTS qwp_fx_trades ( timestamp TIMESTAMP_NS, symbol SYMBOL, ecn SYMBOL, trade_id UUID, side SYMBOL, passive BOOLEAN, price DOUBLE, quantity DOUBLE, counterparty SYMBOL, order_id UUID @@ -66,27 +72,42 @@ CREATE TABLE IF NOT EXISTS qwp_core_price ( (Real DDL also carries the Python PARQUET column encodings.) The generator creates whichever tables its enabled pools need over QWP. Retention is attached only with -`--short_ttl` (`TTL 1 MONTH`/`3 DAYS`, or `STORAGE POLICY(...)` with `--enterprise`). -`--suffix` applies to all three names (`qwp_trades`, `qwp_market_data`, -`qwp_core_price`). - -### Materialized views (`--create_views`) - -With `--create_views true` the generator also creates three matviews over -`qwp_market_data` (suffix threaded through, the 15m view cascades off the 1m): - -| view | base | refresh | content | +`--short_ttl` (`TTL 1 MONTH`/`3 DAYS`, or `STORAGE POLICY(...)` with `--enterprise` +on the base tables — matviews always take TTL). Table/view names are +``: `--prefix` (default `qwp_`, set `""` to match the Python +names) and `--suffix` (default empty) both concatenate **verbatim**, so the trades +table is `fx_trades`, e.g. `qwp_fx_trades` by default or `fx_trades` +with `--prefix ""`. + +### Materialized views (`--create_views`, default on) + +With `--create_views true` (the default, matching Python) the generator creates the +**same 11 matviews as the Python generator** — prefix/suffix threaded through every +name, and each view gated on its base pool being enabled (a disabled pool's views are +skipped, not errored). The cascades and per-view TTL/partitions are byte-faithful to +the Python DDL: + +| view (`…`) | base | refresh | content | | --- | --- | --- | --- | -| `qwp_market_data_ohlc_1m` | `qwp_market_data` | IMMEDIATE | 1m OHLC of `best_bid` + Σ best-bid volume | -| `qwp_market_data_ohlc_15m` | the 1m view | EVERY 5m | 15m OHLC rolled up from the 1m view | -| `qwp_bbo_1h` | `qwp_market_data` | EVERY 10m | hourly max bid / min ask | - -All use `CREATE ... IF NOT EXISTS` (idempotent — to redefine one, drop it first). -Retention follows Python: matviews always take **TTL** (not STORAGE POLICY, which -QuestDB doesn't yet support on views) when `--short_ttl` is set, via a helper that's -ready to switch to storage policies once enterprise matviews support them. The views -are owned by the connecting (ingest) user — no `OWNED BY` clause — so creation never -requires admin/superuser privileges. +| `core_price_1s` | `core_price` | immediate | 1s OHLC-mid, spread, bid/ask stats | +| `core_price_1d` | `core_price` | EVERY 1h (deferred) | 1d OHLC-mid, spread, bid/ask stats | +| `bbo_1s` | `market_data` | immediate | 1s last bid / last ask | +| `bbo_1m` | `bbo_1s` | EVERY 1m (deferred) | 1m max bid / min ask | +| `bbo_1h` | `bbo_1m` | EVERY 10m (deferred) | 1h max bid / min ask | +| `bbo_1d` | `bbo_1h` | EVERY 1h (deferred) | 1d max bid / min ask | +| `market_data_ohlc_1m` | `market_data` | immediate | 1m OHLC of `best_bid` + Σ best-bid volume | +| `market_data_ohlc_15m` | the 1m view | immediate | 15m OHLC rolled up from the 1m view | +| `market_data_ohlc_1d` | `market_data` | EVERY 1h (deferred) | 1d OHLC of `best_bid` + Σ volume | +| `fx_trades_ohlc_1m` | `fx_trades` | immediate | 1m OHLC of traded `price` + Σ quantity | +| `fx_trades_ohlc_1d` | the 1m view | EVERY 1h (deferred) | 1d OHLC rolled up from the 1m view | + +All use `CREATE ... IF NOT EXISTS` (idempotent — to redefine one, drop it first; the +**first** generator to run fixes the definition, so keep both engines on the same +build for byte-identical DDL). Retention follows Python: matviews always take **TTL** +(not STORAGE POLICY, which QuestDB doesn't yet support on views) when `--short_ttl` +is set, with per-view TTLs matching the Python generator. The views are owned by the +connecting (ingest) user — no `OWNED BY` clause — so creation never requires +admin/superuser privileges. ## Prerequisites @@ -220,7 +241,7 @@ mvn -q -f ./pom.xml compile exec:java -Dexec.args="--mode real-time \ Verify (HTTP query endpoint, any node — add the `--suffix` for the throughput run): ```bash -curl -s "http://172.31.42.41:9000/exec?query=SELECT%20count()%20FROM%20qwp_trades" +curl -s "http://172.31.42.41:9000/exec?query=SELECT%20count()%20FROM%20qwp_fx_trades" curl -s "http://172.31.42.41:9000/exec?query=SELECT%20count()%20FROM%20qwp_market_data" curl -s "http://172.31.42.41:9000/exec?query=SELECT%20count()%20FROM%20qwp_core_price" ``` @@ -273,6 +294,38 @@ state, with the **first and last event pinned exactly to open/close**, so `close(second t) == open(second t+1)` holds in the emitted rows — OHLC candles join cleanly with no false gaps. +## Interchangeability with the Python generator + +This generator and the Python `fx_data_generator.py` are **dataset-compatible**: you +can backfill with one and continue (or go live) with the other on the **same tables**, +and downstream views stay consistent. To share tables, run this generator with +`--prefix ""` so the names match the Python generator exactly (`fx_trades`, +`market_data`, `core_price`, and the 11 views). The Python generator is authoritative; +schemas, view DDL, the symbol universe, ECN/reason/counterparty pools, the volume +ladder, and the price/spread/indicator walks all mirror it. + +For a genuinely seamless dataset, also align: + +- **Event density (most important).** Per-second row counts come from the EPS / orders + flags, **not** from the existing data. Run both engines with the **same** + `--core_min_eps/--core_max_eps`, `--market_data_min_eps/--market_data_max_eps` and + `--orders_min_per_sec/--orders_max_per_sec` (and `--min_levels/--max_levels`), or the + row density will step at the hand-off even though prices stay continuous. +- **Continuation seed.** Use `--incremental true` (faster-than-life) on the continuing + run: it seeds mid/spread/indicators from the last stored `core_price` row per symbol + — exactly like the Python `--incremental` — so `open(seam) == close(prev)` holds and + OHLC candles join cleanly. Real-time never uses incremental on either engine; it + reseeds from live Yahoo quotes, so a faster-than-life → live transition deliberately + snaps to the real market (identical behaviour on both). +- **Same build / flags.** Views are `IF NOT EXISTS` (the first creator fixes the + definition), and `--enterprise/--short_ttl/--prefix/--suffix` must match across the + engines so both address the same tables with the same schema. + +Because both engines reconstruct `market_data` and `core_price` from a **single** +`core_price` seed, the order-book `best_bid` can step a few pips at the seam to realign +with `core_price`. This is the same on either engine (it is how the Python generator's +own incremental restart behaves) and is bounded by the spread. + ## Parameters Option names accept Python underscore form (`--start_ts`) or kebab form @@ -329,10 +382,11 @@ depends on the mode: | `--yahoo_refresh_secs ` | 300 | real-time Yahoo refresh interval | | `--realtime_lookahead_secs ` | 2 | real-time: stamp rows n s ahead of wall-clock so the live dashboard stays ahead of WAL apply lag | | `--no_yahoo` | off | skip Yahoo, use template brackets (offline) | -| `--incremental [true\|false]` | false | seed mids from last stored trade, skip Yahoo | -| `--short_ttl` / `--enterprise [true\|false]` | off | retention (TTL, or STORAGE POLICY with enterprise) | -| `--create_views [true\|false]` | false | build the market_data OHLC (1m/15m) + hourly BBO matviews | -| `--suffix ` | none | tables become `qwp_trades` / `qwp_market_data` / `qwp_core_price` | +| `--incremental [true\|false]` | false | seed full state (mid, spread, indicators) from the last stored `core_price` row, skip Yahoo; **faster-than-life only** (rejected in real-time, which always syncs live quotes) | +| `--short_ttl` / `--enterprise [true\|false]` | off | retention (TTL, or STORAGE POLICY with enterprise on the base tables) | +| `--create_views [true\|false]` | **true** | build the full Python-parity matview set (11 views) | +| `--prefix ` | `qwp_` | table-name prefix (verbatim); set `""` to match the Python names exactly | +| `--suffix ` | none | suffix (verbatim); tables become `fx_trades` / `market_data` / `core_price` | | `--lei_pool_size ` | 2000 | distinct counterparties | ### Accepted but unused / unsupported diff --git a/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java b/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java index 8c7af51..f160843 100644 --- a/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java +++ b/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java @@ -74,7 +74,8 @@ public final class Cli { public boolean incremental = false; // seed state from last stored prices, skip Yahoo public boolean shortTtl = false; public boolean enterprise = false; - public boolean createViews = false; // create market_data OHLC/BBO matviews + public boolean createViews = true; // create the Python-parity matviews (default on, like Python) + public String prefix = "qwp_"; // table-name prefix; set "" to match the Python table names exactly public String suffix = ""; // table-name suffix (parity with Python) public int leiPoolSize = 2000; @@ -218,6 +219,9 @@ public static Cli parse(String[] args) { case "create_views": i = boolFlag(args, i, v -> c.createViews = v); break; + case "prefix": + c.prefix = req(args, ++i, raw); + break; case "suffix": c.suffix = req(args, ++i, raw); break; @@ -312,18 +316,26 @@ private void validate() { if ("faster-than-life".equals(mode) && totalTrades <= 0 && endTs == null && runSecs <= 0) { fail("faster-than-life requires a bound: set --total_market_data_events > 0, --end_ts, or --run_secs"); } + if ("real-time".equals(mode) && incremental) { + fail("--incremental is not allowed in real-time mode (real-time syncs live quotes from Yahoo)"); + } } public String tradesTable() { - return "qwp_trades" + suffix; + return prefix + "fx_trades" + suffix; } public String marketDataTable() { - return "qwp_market_data" + suffix; + return prefix + "market_data" + suffix; } public String corePriceTable() { - return "qwp_core_price" + suffix; + return prefix + "core_price" + suffix; + } + + /** View/base name with the same prefix+suffix scheme as the tables (stem has no leading prefix). */ + public String viewName(String stem) { + return prefix + stem + suffix; } public int tradesCommitIntervalMs() { @@ -483,11 +495,12 @@ public static void printUsage() { " --yahoo_refresh_secs real-time Yahoo refresh interval (default 300)", " --realtime_lookahead_secs real-time: stamp events n s ahead of wall-clock (default 2)", " --no_yahoo skip Yahoo, use template brackets (offline)", - " --incremental [true|false] seed prices from last stored row, skip Yahoo", - " --short_ttl [true|false] attach retention to the table", - " --enterprise [true|false] with --short_ttl, use STORAGE POLICY instead of TTL", - " --create_views [true|false] create market_data OHLC (1m/15m) + hourly BBO matviews", - " --suffix append suffix to the table name (-> qwp_trades)", + " --incremental [true|false] seed state from last stored core_price row, skip Yahoo (faster-than-life only)", + " --short_ttl [true|false] attach retention to the tables/views", + " --enterprise [true|false] with --short_ttl, use STORAGE POLICY instead of TTL on base tables", + " --create_views [true|false] create the full Python-parity matview set (default true)", + " --prefix table-name prefix (default 'qwp_'); set '' to match the Python names exactly", + " --suffix append suffix to the table name (-> fx_trades)", " --lei_pool_size distinct counterparties (default 2000)", " --chunk_seconds accepted but unused (state is streamed per-second)"))); } diff --git a/java-qwp/src/main/java/com/questdb/fxqwp/FxUniverse.java b/java-qwp/src/main/java/com/questdb/fxqwp/FxUniverse.java index ec55741..9764781 100644 --- a/java-qwp/src/main/java/com/questdb/fxqwp/FxUniverse.java +++ b/java-qwp/src/main/java/com/questdb/fxqwp/FxUniverse.java @@ -11,11 +11,12 @@ * The simulated FX market: the pair universe, the ECN/counterparty pools, and * the price-continuity model. * - *

This is the simplified sibling of the Python generator. There is no order - * book and no {@code market_data}/{@code core_price} table: trades are sampled - * directly off an evolving mid/spread. The continuity guarantees that matter for - * downstream OHLC still hold, because the mid follows the same controlled random - * walk and {@code close(second t) == open(second t+1)}. + *

This mirrors the Python generator's market model and feeds all three tables + * ({@code fx_trades}, {@code market_data}, {@code core_price}): trades, the order + * book, and top-of-book are all derived from the same per-symbol mid/spread walk, + * seeded identically across the worker pools. The continuity guarantees that matter + * for downstream OHLC hold because the mid follows the same controlled random walk + * and {@code close(second t) == open(second t+1)}. */ public final class FxUniverse { diff --git a/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java b/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java index e0235c9..fa207dd 100644 --- a/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java +++ b/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java @@ -67,9 +67,15 @@ private enum Kind { TRADES, MARKET_DATA, CORE_PRICE } private final String[] leiPool; private final long[] ladder = FxUniverse.makeVolumeLadder(); - // Read-only initial mid per global pair index (computed once from Yahoo/template/ + // Read-only initial state per global pair index (computed once from Yahoo/template/ // incremental). Both pools seed their deterministic walk from the same values. + // initialSpreadPips/initialInd* stay at their sentinels unless incremental seeding + // fills them from core_price, so the non-incremental defaults (4 pips, 0.2, 0.5) are + // untouched. See seedInitialStateFromDb and the worker constructor. private final double[] initialMid; + private final int[] initialSpreadPips; // 0 => use the default (4 pips) + private final double[] initialInd1; // < 0 => use the default (0.2) + private final double[] initialInd2; // < 0 => use the default (0.5) private final int totalAllWeight; // sum of (11 - rank) across all pairs private final AtomicBoolean running = new AtomicBoolean(true); @@ -93,6 +99,11 @@ private QwpTradesGenerator(Cli cfg) { this.leiPool = FxUniverse.generateLeiPool(cfg.leiPoolSize); int n = pairs.size(); this.initialMid = new double[n]; + this.initialSpreadPips = new int[n]; // all 0 => default spread until seeded + this.initialInd1 = new double[n]; + this.initialInd2 = new double[n]; + java.util.Arrays.fill(this.initialInd1, -1.0); // < 0 => default until seeded + java.util.Arrays.fill(this.initialInd2, -1.0); int acc = 0; for (int i = 0; i < n; i++) { acc += (11 - pairs.get(i).rank); @@ -128,7 +139,7 @@ private void run() throws Exception { // 1) Reference data → initial mid per symbol (shared, read-only). if (cfg.incremental) { - seedInitialMidFromDb(); + seedInitialStateFromDb(); } else if (!cfg.noYahoo) { new YahooFinance().refreshBrackets(pairs, 1.0, Long.MIN_VALUE); // startup: reset timeline } @@ -404,12 +415,15 @@ private final class Worker implements Runnable { int g = owned[j]; walkRng[j] = new SplittableRandom(WALK_SEED_BASE + GOLDEN * g); walkMid[j] = initialMid[g]; - walkSp[j] = 4; + // Spread/indicators default to the Python template (4 pips, 0.2, 0.5) unless + // incremental seeding filled the initial* arrays from core_price, in which + // case continue from the last stored state so the seam stays continuous. + walkSp[j] = initialSpreadPips[g] > 0 ? initialSpreadPips[g] : 4; // Distinct seed from walkRng so the indicator draws never shift the - // mid/spread stream. Initial values match the Python template (0.2, 0.5). + // mid/spread stream. indRng[j] = new SplittableRandom((WALK_SEED_BASE * 31 + 0x5DEECE66DL) + GOLDEN * g); - walkInd1[j] = 0.2; - walkInd2[j] = 0.5; + walkInd1[j] = initialInd1[g] >= 0 ? initialInd1[g] : 0.2; + walkInd2[j] = initialInd2[g] >= 0 ? initialInd2[g] : 0.5; } } @@ -438,7 +452,7 @@ public void run() { // Data-second index, identical across pools for the same // secondStartNs -> deterministic bracket lookup -> consistent walk. long k = (secondStartNs - base) / NANOS_PER_SEC; - beginSecond(k, 5.0); + beginSecond(k, 7.0); // real-time drift, matches the Python real-time path (FTL uses 5.0) long sliceEnd = sliceNs; while (true) { emitSlice(sender, table, secondStartNs, sliceEnd); @@ -466,7 +480,7 @@ public void run() { break; } long k = (secondStartNs - base) / NANOS_PER_SEC; - beginSecond(k, 5.0); + beginSecond(k, 5.0); // faster-than-life drift, matches the Python precompute path emitSlice(sender, table, secondStartNs, NANOS_PER_SEC); endSecond(); secondStartNs += NANOS_PER_SEC; @@ -997,46 +1011,114 @@ private void createTables() { } } + // Deferred-start anchor shared by every REFRESH EVERY view, matching the Python DDL. + private static final String MV_DEFERRED = "DEFERRED START '2025-06-01T00:00:00.000000Z'"; + /** - * Optional materialized views over market_data (Python parity, suffix-aware): - * a 1m OHLC (REFRESH IMMEDIATE), a 15m OHLC cascading off the 1m view - * (REFRESH EVERY 5m), and an hourly BBO (REFRESH EVERY 10m). Retention is TTL via - * {@link #mvRetentionClause(String)}. The views are owned by the connecting user - * (no OWNED BY clause), so creation never needs admin/superuser privileges. + * The full Python-parity materialized-view set (prefix/suffix-aware), byte-faithful + * to {@code ensure_materialized_views_exist} in the Python generator: same column + * expressions, SAMPLE BY cadences, REFRESH/DEFERRED-START clauses, PARTITION BY, and + * per-view TTL (via {@link #mvRetentionClause(String)} — matviews take TTL only, never + * STORAGE POLICY). Views are gated on their base pool being enabled so a disabled pool + * never triggers a create against a missing table. Owned by the connecting user (no + * OWNED BY), so creation needs no admin/superuser privileges. */ private void createViews() { if (!cfg.createViews) { return; } - if (cfg.marketDataProcesses <= 0) { - System.out.println("[INFO] --create_views set but the market_data pool is off; skipping views " - + "(they read from " + cfg.marketDataTable() + ")."); - return; + // core_price views (open/high/low/close mid, spread, bid/ask stats). + if (cfg.coreProcesses > 0) { + String cp = cfg.corePriceTable(); + execDdl(cfg.viewName("core_price_1s"), corePriceView(cfg.viewName("core_price_1s"), + "", cp, "1s", "HOUR", "4 HOURS")); + execDdl(cfg.viewName("core_price_1d"), corePriceView(cfg.viewName("core_price_1d"), + "REFRESH EVERY 1h " + MV_DEFERRED + " ", cp, "1d", "MONTH", "1 MONTH")); + } else { + System.out.println("[INFO] --create_views: core_price pool off; skipping core_price_1s/1d."); + } + // market_data views: BBO ladder (1s -> 1m -> 1h -> 1d) and OHLC (1m, 15m, 1d). + if (cfg.marketDataProcesses > 0) { + String md = cfg.marketDataTable(); + String bbo1s = cfg.viewName("bbo_1s"); + String bbo1m = cfg.viewName("bbo_1m"); + String bbo1h = cfg.viewName("bbo_1h"); + String bbo1d = cfg.viewName("bbo_1d"); + execDdl(bbo1s, "CREATE MATERIALIZED VIEW IF NOT EXISTS " + bbo1s + " AS (" + + "SELECT timestamp, symbol, last(best_bid) AS bid, last(best_ask) AS ask " + + "FROM " + md + " SAMPLE BY 1s) PARTITION BY HOUR" + mvRetentionClause("3 DAYS")); + execDdl(bbo1m, bboCascade(bbo1m, bbo1s, "1m", "DAY", "REFRESH EVERY 1m " + MV_DEFERRED + " ", "3 DAYS")); + execDdl(bbo1h, bboCascade(bbo1h, bbo1m, "1h", "MONTH", "REFRESH EVERY 10m " + MV_DEFERRED + " ", "1 MONTH")); + execDdl(bbo1d, bboCascade(bbo1d, bbo1h, "1d", "MONTH", "REFRESH EVERY 1h " + MV_DEFERRED + " ", "1 MONTH")); + + String mdOhlc1m = cfg.viewName("market_data_ohlc_1m"); + String mdOhlc15m = cfg.viewName("market_data_ohlc_15m"); + String mdOhlc1d = cfg.viewName("market_data_ohlc_1d"); + execDdl(mdOhlc1m, mdOhlcFromBook(mdOhlc1m, "", md, "1m", "HOUR", "1 DAY")); + execDdl(mdOhlc15m, ohlcRollup(mdOhlc15m, mdOhlc1m, "15m", "HOUR", "", "2 DAYS")); + execDdl(mdOhlc1d, mdOhlcFromBook(mdOhlc1d, "REFRESH EVERY 1h " + MV_DEFERRED + " ", md, "1d", "MONTH", "1 MONTH")); + } else { + System.out.println("[INFO] --create_views: market_data pool off; skipping bbo_* and market_data_ohlc_* views."); + } + // fx_trades views: traded-price OHLC (1m, 1d). + if (cfg.tradesProcesses > 0) { + String tr = cfg.tradesTable(); + String trOhlc1m = cfg.viewName("fx_trades_ohlc_1m"); + String trOhlc1d = cfg.viewName("fx_trades_ohlc_1d"); + execDdl(trOhlc1m, "CREATE MATERIALIZED VIEW IF NOT EXISTS " + trOhlc1m + " AS (" + + "SELECT timestamp, symbol, first(price) AS open, max(price) AS high, " + + "min(price) AS low, last(price) AS close, SUM(quantity) AS total_volume " + + "FROM " + tr + " SAMPLE BY 1m) PARTITION BY HOUR" + mvRetentionClause("2 DAYS")); + execDdl(trOhlc1d, ohlcRollup(trOhlc1d, trOhlc1m, "1d", "MONTH", + "REFRESH EVERY 1h " + MV_DEFERRED + " ", "1 MONTH")); + } else { + System.out.println("[INFO] --create_views: trades pool off; skipping fx_trades_ohlc_* views."); } - String md = cfg.marketDataTable(); - String md1m = "qwp_market_data_ohlc_1m" + cfg.suffix; - String md15m = "qwp_market_data_ohlc_15m" + cfg.suffix; - String bbo1h = "qwp_bbo_1h" + cfg.suffix; - String ttl = mvRetentionClause("3 DAYS"); + } - execDdl(md1m, "CREATE MATERIALIZED VIEW IF NOT EXISTS '" + md1m + "' WITH BASE '" + md + "' REFRESH IMMEDIATE AS (" + /** core_price OHLC-mid + spread + bid/ask stats view (Python core_price_1s/1d). */ + private String corePriceView(String name, String refresh, String base, String sampleBy, + String partitionBy, String ossTtl) { + return "CREATE MATERIALIZED VIEW IF NOT EXISTS " + name + " " + refresh + "AS (" + "SELECT timestamp, symbol, " - + "first(best_bid) AS open, max(best_bid) AS high, min(best_bid) AS low, last(best_bid) AS close, " - + "SUM(bids[2][1]) AS total_volume " - + "FROM " + md + " SAMPLE BY 1m" - + ") PARTITION BY HOUR" + ttl); + + "first((bid_price + ask_price)/2) AS open_mid, " + + "max((bid_price + ask_price)/2) AS high_mid, " + + "min((bid_price + ask_price)/2) AS low_mid, " + + "last((bid_price + ask_price)/2) AS close_mid, " + + "last(ask_price) - last(bid_price) AS last_spread, " + + "max(bid_price) AS max_bid, min(bid_price) AS min_bid, avg(bid_price) AS avg_bid, " + + "max(ask_price) AS max_ask, min(ask_price) AS min_ask, avg(ask_price) AS avg_ask " + + "FROM " + base + " SAMPLE BY " + sampleBy + + ") PARTITION BY " + partitionBy + mvRetentionClause(ossTtl); + } - execDdl(md15m, "CREATE MATERIALIZED VIEW IF NOT EXISTS '" + md15m + "' WITH BASE '" + md1m + "' REFRESH EVERY 5m AS (" - + "SELECT timestamp, symbol, " - + "first(open) AS open, max(high) AS high, min(low) AS low, last(close) AS close, " - + "SUM(total_volume) AS total_volume " - + "FROM " + md1m + " SAMPLE BY 15m" - + ") PARTITION BY HOUR" + ttl); - - execDdl(bbo1h, "CREATE MATERIALIZED VIEW IF NOT EXISTS '" + bbo1h + "' REFRESH EVERY 10m AS (" - + "SELECT timestamp, symbol, max(best_bid) AS bid, min(best_ask) AS ask " - + "FROM " + md + " SAMPLE BY 1h" - + ") PARTITION BY DAY" + ttl); + /** BBO rollup cascading off the previous BBO view (Python bbo_1m/1h/1d). */ + private String bboCascade(String name, String base, String sampleBy, String partitionBy, + String refresh, String ossTtl) { + return "CREATE MATERIALIZED VIEW IF NOT EXISTS " + name + " " + refresh + "AS (" + + "SELECT timestamp, symbol, max(bid) AS bid, min(ask) AS ask " + + "FROM " + base + " SAMPLE BY " + sampleBy + + ") PARTITION BY " + partitionBy + mvRetentionClause(ossTtl); + } + + /** OHLC straight off the order book best_bid (Python market_data_ohlc_1m/1d). */ + private String mdOhlcFromBook(String name, String refresh, String base, String sampleBy, + String partitionBy, String ossTtl) { + return "CREATE MATERIALIZED VIEW IF NOT EXISTS " + name + " " + refresh + "AS (" + + "SELECT timestamp, symbol, first(best_bid) AS open, max(best_bid) AS high, " + + "min(best_bid) AS low, last(best_bid) AS close, SUM(bids[2][1]) AS total_volume " + + "FROM " + base + " SAMPLE BY " + sampleBy + + ") PARTITION BY " + partitionBy + mvRetentionClause(ossTtl); + } + + /** OHLC rollup off a finer OHLC view (Python market_data_ohlc_15m, fx_trades_ohlc_1d). */ + private String ohlcRollup(String name, String base, String sampleBy, String partitionBy, + String refresh, String ossTtl) { + return "CREATE MATERIALIZED VIEW IF NOT EXISTS " + name + " " + refresh + "AS (" + + "SELECT timestamp, symbol, first(open) AS open, max(high) AS high, " + + "min(low) AS low, last(close) AS close, SUM(total_volume) AS total_volume " + + "FROM " + base + " SAMPLE BY " + sampleBy + + ") PARTITION BY " + partitionBy + mvRetentionClause(ossTtl); } private void execDdl(String table, String ddl) { @@ -1399,16 +1481,22 @@ public void onError(byte status, String message) { return ns; } - /** Incremental: seed initial mids from the last stored trade prices. */ - private void seedInitialMidFromDb() { - System.out.println("[INFO] incremental: seeding mids from last stored prices in " + cfg.tradesTable()); + /** + * Incremental: seed the full initial state (mid, spread, indicators) from the last + * stored core_price row per symbol. Mirrors the Python {@code load_initial_state}, + * which reads {@code symbol, bid_price, ask_price, indicator1, indicator2 ... LATEST + * BY symbol} from core_price so the resumed walk continues bid/ask/spread/indicators + * without a step at the seam. Symbols with no prior row keep the template defaults. + */ + private void seedInitialStateFromDb() { + System.out.println("[INFO] incremental: seeding state from last stored core_price rows in " + cfg.corePriceTable()); Map idx = new HashMap<>(); for (int i = 0; i < pairs.size(); i++) { idx.put(pairs.get(i).symbol, i); } try (QwpQueryClient client = QwpQueryClient.fromConfig(cfg.queryClientConfig())) { client.connect(); - client.execute("SELECT symbol, price FROM " + cfg.tradesTable() + client.execute("SELECT symbol, bid_price, ask_price, indicator1, indicator2 FROM " + cfg.corePriceTable() + " LATEST ON timestamp PARTITION BY symbol", new QwpColumnBatchHandler() { @Override @@ -1418,11 +1506,20 @@ public void onBatch(QwpColumnBatch batch) { if (i == null) { return; } - double px = row.getDoubleValue(1); - if (px > 0) { + double bid = row.getDoubleValue(1); + double ask = row.getDoubleValue(2); + double ind1 = row.getDoubleValue(3); + double ind2 = row.getDoubleValue(4); + if (bid > 0 && ask > 0) { FxPair p = pairs.get(i); - initialMid[i] = FxUniverse.quantizeToPip(px, p.pip, p.precision); - p.resetBracket(px * 0.99, px * 1.01); + double mid = (bid + ask) / 2.0; + initialMid[i] = FxUniverse.quantizeToPip(mid, p.pip, p.precision); + // Spread in whole pips, clamped to the same 1..8 band the walk uses. + int sp = (int) Math.round((ask - bid) / p.pip); + initialSpreadPips[i] = Math.max(1, Math.min(8, sp)); + initialInd1[i] = Math.max(0.0, Math.min(1.0, ind1)); + initialInd2[i] = Math.max(0.0, Math.min(1.0, ind2)); + p.resetBracket(mid * 0.99, mid * 1.01); } }); } From 59e64835154c922915867e7ecb0b98e719f72090 Mon Sep 17 00:00:00 2001 From: javier Date: Mon, 6 Jul 2026 14:02:37 +0200 Subject: [PATCH 2/2] Add --durable_ack flag and default the client to 1.3.6-SNAPSHOT - Cli: new --durable_ack [true|false] (default false), independent of --enterprise. OSS rejects durable ack during the WS upgrade, so it stays opt-in; enable it for no-loss Enterprise HA. - buildSender: call requestDurableAck(true) when --durable_ack, so a failover cannot lose rows the old primary accepted but had not yet replicated to the node that gets promoted. - pom: default questdb.client.version to 1.3.6-SNAPSHOT (the build with the QWP HA role-failover + durable-ack lifecycle). Not on Maven Central; install the java-questdb-client locally, or override with -Dquestdb.client.version=... - README: document --durable_ack and the 1.3.6-SNAPSHOT requirement. --- java-qwp/README.md | 20 +++++++++++++++---- java-qwp/pom.xml | 9 +++++---- .../src/main/java/com/questdb/fxqwp/Cli.java | 5 +++++ .../com/questdb/fxqwp/QwpTradesGenerator.java | 7 +++++++ 4 files changed, 33 insertions(+), 8 deletions(-) diff --git a/java-qwp/README.md b/java-qwp/README.md index 5c79d22..b96e179 100644 --- a/java-qwp/README.md +++ b/java-qwp/README.md @@ -113,9 +113,12 @@ admin/superuser privileges. - **Java 17+** and **Maven 3+**. - A running **QuestDB** that speaks **QWP over WebSocket**, protocol-compatible - with client `1.3.2`. QWP shares the HTTP port (default `9000`). -- The **`org.questdb:questdb-client:1.3.2`** dependency — resolved automatically - from **Maven Central**, no local build needed. + with client `1.3.6-SNAPSHOT`. QWP shares the HTTP port (default `9000`). +- The **`org.questdb:questdb-client:1.3.6-SNAPSHOT`** dependency. This build carries + the QWP HA/role-failover and durable-ack support used here and is **not on Maven + Central**, so install it locally first (`mvn clean install` from the + `java-questdb-client` checkout). Override with `-Dquestdb.client.version=...` to pin + another build. ## Build @@ -263,6 +266,15 @@ All hosts share **one** credential set and **one** transport scheme: Spill files are purged automatically once the data is acknowledged (only tiny `.lock` stubs remain). Size the S&F volume for your worst outage: at ~1M rows/sec a 30s stop buffers a few GB. +- **Durable ack (`--durable_ack`, default off).** By default spilled frames are + released on a normal ack (the node received them). With `--durable_ack true` the + sender instead holds each frame until a **durable, replicated** ack, so a failover + cannot lose rows the old primary accepted but had not yet replicated to the node + that gets promoted. This is **Enterprise-only** — OSS servers reject it during the + WebSocket upgrade — and is independent of `--enterprise` (which only drives table + retention), so enable it explicitly for no-loss HA. On promotion the client detects + the new writable primary via its role header, reconnects, and replays the still-held + frames. - **WAL backpressure:** a monitor polls `wal_tables()` for each enabled table and pauses **only that table's pool** when its `sequencerTxn - writerTxn` lag exceeds the high-water threshold — `3 × processes` above 2 workers, `5 × processes` at or @@ -410,6 +422,6 @@ ILP/PG-transport flags (`--protocol`, `--pg_port`, `--ilp_user`, `--token_x`, keep an eye on the WAL lag. Separate tables/pools and modest per-table worker counts keep O3 low. - **Protocol compatibility:** QWP is a development wire protocol; the server must - be protocol-compatible with client `1.3.2`. + be protocol-compatible with client `1.3.6-SNAPSHOT`. - Java uses its own truststore — no macOS certifi workaround needed. ``` diff --git a/java-qwp/pom.xml b/java-qwp/pom.xml index e22fb7b..021cb27 100644 --- a/java-qwp/pom.xml +++ b/java-qwp/pom.xml @@ -14,10 +14,11 @@ 17 UTF-8 com.questdb.fxqwp.QwpTradesGenerator - - 1.3.2 + + 1.3.6-SNAPSHOT diff --git a/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java b/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java index f160843..90dbdac 100644 --- a/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java +++ b/java-qwp/src/main/java/com/questdb/fxqwp/Cli.java @@ -36,6 +36,7 @@ public final class Cli { public String sfDir = "/tmp/qwp_trades_sf"; public String senderId = "qwp-fx-trades"; public int autoFlushBytes = 524288; // QWP sender auto-flush size (bytes); 512 KiB, safely under the ~1MB WS frame cap + public boolean durableAck = false; // QWP: hold store-and-forward frames until a durable (replicated) ack. Enterprise-only (OSS rejects it during the WS upgrade). Independent of --enterprise. // --- mode / volume / time ------------------------------------------------ public String mode = null; // real-time | faster-than-life (required) @@ -129,6 +130,9 @@ public static Cli parse(String[] args) { case "autoflush_bytes": c.autoFlushBytes = Integer.parseInt(req(args, ++i, raw)); break; + case "durable_ack": + i = boolFlag(args, i, v -> c.durableAck = v); + break; // ---- mode / volume / time ---- case "mode": @@ -468,6 +472,7 @@ public static void printUsage() { " --sf_dir

store-and-forward dir (default /tmp/qwp_trades_sf)", " --sender_id store-and-forward sender id (default qwp-fx-trades)", " --auto_flush_bytes QWP sender auto-flush size in bytes (default 524288 = 512 KiB)", + " --durable_ack [true|false] hold store-and-forward frames until a durable (replicated) ack; Enterprise-only, default false (OSS rejects it)", "", "Pools (one thread set per table; symbols snake-drafted across each pool):", " --trades_processes worker threads for qwp_trades, 0-30 (default 1; 0 = off)", diff --git a/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java b/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java index fa207dd..e227db3 100644 --- a/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java +++ b/java-qwp/src/main/java/com/questdb/fxqwp/QwpTradesGenerator.java @@ -922,6 +922,13 @@ private Sender buildSender(Kind kind, int workerId) { } final AtomicLong lastConnLogMs = new AtomicLong(0); final String who = tag(kind) + workerId; + // Enterprise-only: hold spilled frames until a durable (replicated) ack, so a failover + // cannot lose rows the old primary accepted but had not yet replicated. OSS rejects this + // during the WS upgrade, so it is off by default and gated behind its own --durable_ack + // flag (independent of --enterprise). + if (cfg.durableAck) { + b.requestDurableAck(true); + } b.storeAndForwardDir(sfPath) .senderId(cfg.senderId + "-" + who) .transactional(true)