diff --git a/README.md b/README.md index 305ef18..50eab26 100644 --- a/README.md +++ b/README.md @@ -163,7 +163,7 @@ bootstrap_notebook(add_src_root=True) Run `task setup` first so VS Code can select the registered `Python (joint-client-python)` notebook kernel backed by this repository's `.venv`. -The bootstrap helper resolves the nearest src-layout Python project root, switches the working directory there, and prepends that project's local `src` tree during development. The examples cover hosted health checks, low-level JSON prediction, mean forecasts, sample forecasts, quantile forecasts, conditional forecasts, pandas/NumPy result conversion, and CSV forecast workflows. They use `.env.sample` placeholders and checked-in fixture payloads; no real tokens or deployment IDs are stored in notebooks. +The bootstrap helper resolves the nearest src-layout Python project root, switches the working directory there, and prepends that project's local `src` tree during development. The examples cover hosted health checks, low-level JSON prediction, mean forecasts, sample forecasts, quantile forecasts, conditional forecasts (one conditional read as a mean, as draws, as quantiles inside a bounded band, and as the log density of observed values, plus the ranking of candidate conditions), pandas/NumPy result conversion, and CSV forecast workflows. They use `.env.sample` placeholders and checked-in fixture payloads; no real tokens or deployment IDs are stored in notebooks. The current forecast request contract is: diff --git a/docs/api-reference.md b/docs/api-reference.md index 1f05497..e8de616 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -6,7 +6,7 @@ This reference covers the supported public Python surface exported by `jointfm_c | Name | Purpose | | --- | --- | -| `JointFMClient` | Synchronous client for hosted or local JointFM endpoints. Use `from_env()` for `.env` and `config.yaml` backed hosted settings, `health()` for consensus typed service metadata, `health_instances()` for per-deployment probe results and pooled sample topology, `predict(payload)` for low-level JSON prediction, `forecast(...)` for validated tabular forecasts, and the `forecast_mean(...)`, `forecast_samples(...)`, and `forecast_quantiles(...)` convenience methods for typed forecast results. Each forecast method accepts `condition=ConditionBlock(...)`, which switches the request to the `condition` query mode and, before anything is sent, checks that the deployment's health metadata advertises the mode and every condition kind the block uses. `health()` probes `GET /healthz` for local deployments and POSTs `{"request_type": "health"}` to `predict_url` for hosted DataRobot deployments because the DataRobot deployment gateway only proxies the unstructured prediction route. `feature_importance(...)` runs permutation feature importance: one baseline `forecast_samples` call plus one per shuffled feature column, returning a list of `{"feature", "mean", "distance"}` dicts, each holding that feature's absolute forecast-mean shift and centered squared 2-Wasserstein distance indexed by target and horizon. | +| `JointFMClient` | Synchronous client for hosted or local JointFM endpoints. Use `from_env()` for `.env` and `config.yaml` backed hosted settings, `health()` for consensus typed service metadata, `health_instances()` for per-deployment probe results and pooled sample topology, `predict(payload)` for low-level JSON prediction, `forecast(...)` for validated tabular forecasts, and the `forecast_mean(...)`, `forecast_samples(...)`, `forecast_quantiles(...)`, and `forecast_log_prob(...)` convenience methods for typed forecast results. `forecast_log_prob(...)` is the one that asks about values the caller already holds: it takes `query_rows`, one observed row per entry of `query_times` carrying every declared column, and returns their log density under the model's joint. Each forecast method accepts `condition=ConditionBlock(...)`, which switches the request to the `condition` query mode and, before anything is sent, checks that the deployment's health metadata advertises the mode and every condition kind the block uses. `health()` probes `GET /healthz` for local deployments and POSTs `{"request_type": "health"}` to `predict_url` for hosted DataRobot deployments because the DataRobot deployment gateway only proxies the unstructured prediction route. `feature_importance(...)` runs permutation feature importance: one baseline `forecast_samples` call plus one per shuffled feature column, returning a list of `{"feature", "mean", "distance"}` dicts, each holding that feature's absolute forecast-mean shift and centered squared 2-Wasserstein distance indexed by target and horizon. | `JointFMClient.from_env()` loads `config.yaml`, optional `.env` values, and process environment variables. `JointFMClient.health(cache=True)` caches health metadata only when requested. `JointFMClient.health_instances()` returns the same probe as a `HealthInstances` object: one `InstanceHealth` per configured deployment (including failures), `max_sample_count` as the sum of reachable caps (overall parallel capacity), and `topology` / `topology_label` grouping those caps (unavailable peers are listed but excluded from the sum and topology). Each endpoint's health payload describes only that endpoint; the client aggregates by calling each configured peer. `health()` still exposes the minimum reachable `max_sample_count`, which is the sample-batch cap used by forecast helpers. `JointFMClient.predict(payload)` requires `payload["model_version"]`; high-level forecast helpers resolve the configured model version when the caller does not pass one explicitly. When `forecast_samples(...)` requests an explicit `n_samples`, the client learns the deployment's `max_sample_count` from health metadata before the first prediction, splits oversized requests into capped prediction batches, and returns one merged `SampleForecastResult`. Clients configured without a reachable health route fall back to discovering the cap from the structured service error. @@ -17,7 +17,7 @@ This reference covers the supported public Python surface exported by `jointfm_c | `ColumnSpec` | Describes one modeled request column. Fields are `name`, `modality`, `role`, `nullable`, `vocabulary_size`, `level_count`, `mapping`, `lower_bound`, `upper_bound`, `time_value_kind`, `time_value_scale_seconds`, `time_value_use_local_normalized_time`, `time_value_calendar_id`, and `time_value_timezone`. | | `DataFrameSchema` | Describes tabular history layout. Fields are `columns`, `time_index_mode`, `time_column`, `time_scale_seconds`, `use_local_normalized_time`, `calendar_id`, and `timezone`. | | `ForecastRequestMetadata` | Holds `schema_version`, `model_version`, `query_mode` (`forecast` or `condition`), and `return_mode` for one forecast request. | -| `ForecastRequest` | Validated request object that combines metadata, schema, history rows, query times, requested columns, sample or quantile controls, `seed`, and an optional `condition` block, then emits a JSON-compatible payload with `to_payload()`. `query_mode="condition"` and a `condition` block must appear together, and the block is validated against the request's own schema and `query_times` before any round trip. | +| `ForecastRequest` | Validated request object that combines metadata, schema, history rows, query times, requested columns, sample or quantile controls, `seed`, an optional `condition` block, and the optional `query_rows` a scored request supplies, then emits a JSON-compatible payload with `to_payload()`. `query_mode="condition"` and a `condition` block must appear together, as must `return_mode="log_prob"` and `query_rows`, and both are validated against the request's own schema and `query_times` before any round trip. | | `EqualityCondition` | One column of the conditioned position pinned to a finite `value`. The pinned column leaves the read-out set, so it must not appear in `requested_columns`. | | `IntervalCondition` | One column of the conditioned position confined to `[lower, upper]`; `None` leaves a side open, at least one side must be bounded, and `lower < upper`. The column stays readable and the response describes its distribution inside the range. | | `ConditionBlock` | Every condition of one request: `query_time_index` into the request's `query_times` and a sequence of `EqualityCondition` or `IntervalCondition` values, at most one per column. `kinds`, `pinned_columns`, and `conditioned_columns` expose what the capability gate and the request validation need. | @@ -31,11 +31,13 @@ This reference covers the supported public Python surface exported by `jointfm_c | `StructuredError` | One structured JointFM service error with `code`, `message`, and optional `field`. | | `ForecastDiagnostics` | Response diagnostics containing `history_rows`, `horizon_count`, optional `seed`, and, on condition responses, `condition_draws` and an optional `interval_estimator`. | | `QuantileForecast` | One quantile surface with `quantile` and `values`; `to_numpy()` returns axis order `(horizon, column)`. | -| `ForecastOutputs` | Legacy nested view of parsed output arrays, including `query_times`, `requested_columns`, and exactly one of `mean`, `samples`, or `quantiles`. | +| `ForecastOutputs` | Legacy nested view of parsed output arrays, including `query_times`, `requested_columns`, and exactly one of `mean`, `samples`, `quantiles`, or `log_prob`. A response whose `return_mode` no parser handles is refused rather than read as another mode's shape. | +| `LogProbScores` | One `outputs.log_prob` block: per-horizon `values` and `nll_values`, plus the `total`, `mean`, `nll_total`, and `nll_mean` summaries. The summaries are redundant, so parsing recomputes them from `values` and rejects a payload that disagrees with itself. A log density is joint over the scored columns, so the block has no column axis. | | `ForecastResponse` | Shared base for parsed forecast results. It preserves schema, image, model, checkpoint, head, mode, query-time, requested-column, diagnostic, and error metadata, plus `plausibility` (`None` on forecast responses). | | `MeanForecastResult` | Parsed mean forecast. `to_numpy()` returns `(horizon, column)` and pandas helpers return tidy or wide frames. | | `SampleForecastResult` | Parsed sample forecasts. `to_numpy()` returns `(sample, horizon, column)` and pandas helpers return tidy or wide frames. | | `QuantileForecastResult` | Parsed quantile forecasts. `to_numpy()` returns `(quantile, horizon, column)`, `quantile_levels` exposes the ordered levels, and pandas helpers return tidy or wide frames. | +| `LogProbResult` | Parsed log densities of the values the request supplied; it carries no forecast values. `to_numpy()` returns `(horizon,)`, `to_pandas_tidy()` one row per scored position, and `to_pandas_wide()` the one-row request summary, since a log density has no column axis to widen. | ## Configuration Classes @@ -158,13 +160,14 @@ The string literals are exposed as `PREDICT_REQUEST_TYPE`, `HEALTH_REQUEST_TYPE` | `schema_version` | Yes | Must be `"v3"`. | | `model_version` | Yes | Exact deployed model version expected by the caller. | | `query_mode` | Yes | `"forecast"` for the unconditional forecast or `"condition"` for a conditional query at one future position. | -| `return_mode` | Yes | One of `"mean"`, `"samples"`, `"quantiles"`, or `"log_prob"`. The high-level `forecast_mean`, `forecast_samples`, and `forecast_quantiles` helpers cover the first three; `"log_prob"` is reachable through the low-level `predict(payload)` path. | +| `return_mode` | Yes | One of `"mean"`, `"samples"`, `"quantiles"`, or `"log_prob"`, covered by the `forecast_mean`, `forecast_samples`, `forecast_quantiles`, and `forecast_log_prob` helpers. | | `time_index_mode` | Yes | One of `"ordinal"`, `"continuous_float"`, or `"absolute_datetime"`. | | `columns` | Yes | Non-empty array of column descriptors for modeled columns. | | `history_rows` | Yes | Non-empty array of history row objects in the declared schema. | | `query_times` | Yes | Non-empty future forecast horizon values. Absolute datetimes are encoded timezone-stably. | | `time_column` | For absolute datetime, optional otherwise | Name of the history time column. It must not duplicate a modeled column name. | -| `requested_columns` | Optional | Output column names or integer indices. Duplicates are rejected. Defaults to all modeled columns. | +| `requested_columns` | Optional | Output column names or integer indices. Duplicates are rejected. Defaults to all modeled columns, minus the columns an equality condition pinned. Under `return_mode="log_prob"` an explicit list must name every readable column in schema order, because the score is joint over them. | +| `query_rows` | `log_prob` mode | Observed rows to score, one per entry of `query_times`, each carrying a value for every declared column. Required by `return_mode="log_prob"` and rejected for every other mode. A row that contradicts a condition — a pinned column given another value, or a bounded column outside its range — is refused by the service instead of scored. | | `n_samples` | Samples and quantiles controls | Positive sample count when sampling controls are needed. Oversized sample forecasts are batched automatically against the cap advertised in health metadata. | | `quantiles` | Quantiles mode | Quantile levels in `(0, 1)`, required for `return_mode="quantiles"`. | | `seed` | Optional | Integer random seed for reproducible stochastic outputs. | @@ -212,6 +215,7 @@ The string literals are exposed as `PREDICT_REQUEST_TYPE`, `HEALTH_REQUEST_TYPE` | `diagnostics.history_rows` | Number of history rows processed. | | `diagnostics.horizon_count` | Number of forecast horizon steps returned. | | `diagnostics.seed` | Optional seed used by the service. | +| `outputs.log_prob` | `log_prob` responses only: per-horizon `values` and `nll_values` with their `total`, `mean`, `nll_total`, and `nll_mean` summaries. | | `diagnostics.condition_draws` | Condition responses only: number of draws behind sampled outputs. Batched sample requests report the merged count. | | `diagnostics.interval_estimator` | Condition responses only, and only when more than one column carries an interval: `points` and `effective_sample_size` of the numerical region-probability estimate. | | `plausibility` | `null` on forecast responses. On condition responses an object with `equality_log_density` (log density of the pinned values) and `region_log_probability` (log probability of the interval region), each `null` when the request carried no condition of that kind. | diff --git a/notebooks/forecast_condition.ipynb b/notebooks/forecast_condition.ipynb index 521c81e..d3fc9f9 100644 --- a/notebooks/forecast_condition.ipynb +++ b/notebooks/forecast_condition.ipynb @@ -54,8 +54,12 @@ "- An **equality condition** pins a column to a value (`EqualityCondition`). The pinned column leaves the read-out set, because reading it back would only repeat the request.\n", "- An **interval condition** confines a column to a range whose bounds may be open on either side (`IntervalCondition`). The column stays readable: what comes back is its distribution inside the range.\n", "\n", + "The cell below asks for the same forecast twice, once without the block and once with it, because a conditional mean is only readable next to the unconditional one: what the what-if is worth is the shift between them, not the level of either.\n", + "\n", "Every column without a condition is a read-out column, and the response describes the conditional distribution of those columns at the conditioned position only, so `outputs.query_times` has exactly one entry however many `query_times` the request carried.\n", "\n", + "Every read-out the service serves reads that same conditional, and the sections below take it in turn: as a mean beside the unconditional one, as coherent joint draws, as quantiles inside a bounded band, and as the log density of values you already hold. The last two sections use the plausibility numbers to rank candidate what-ifs against each other.\n", + "\n", "Whether a deployment can condition depends on the checkpoint's head, so `/healthz` advertises `condition` in `supported_query_modes` and the kinds it answers in `supported_condition_kinds`. The client checks that advertisement before sending, and this notebook reads it explicitly so the check is visible." ] }, @@ -83,12 +87,16 @@ "HISTORY_PATH = Path(\"notebooks/history.csv\")\n", "FEATURE_COLUMNS = [\"equity_index_level\", \"treasury_10y_yield\", \"eur_usd_rate\"]\n", "TARGET_COLUMNS = [\"portfolio_nav\", \"realized_volatility\"]\n", + "PINNED_COLUMN = \"equity_index_level\"\n", "INPUT_STEPS = 100\n", "OUTPUT_HORIZONS = 10\n", "CONDITIONED_STEP = 0\n", "EQUITY_RALLY = 1.02\n", "EXPECTED_COLUMNS = FEATURE_COLUMNS + TARGET_COLUMNS\n", "QUERY_TIMES = list(range(INPUT_STEPS, INPUT_STEPS + OUTPUT_HORIZONS))\n", + "# Every column a request may read back: an equality condition removes its own\n", + "# column from the read-out set, and nothing else does.\n", + "READABLE_COLUMNS = [name for name in EXPECTED_COLUMNS if name != PINNED_COLUMN]\n", "\n", "history = pd.read_csv(HISTORY_PATH, dtype=float)\n", "if list(history.columns) != EXPECTED_COLUMNS:\n", @@ -115,14 +123,18 @@ " query_times_length=len(QUERY_TIMES),\n", ")\n", "\n", - "last_equity_level = float(history[\"equity_index_level\"].iloc[-1])\n", + "last_equity_level = float(history[PINNED_COLUMN].iloc[-1])\n", + "rallied_equity_level = last_equity_level * EQUITY_RALLY\n", "rally = ConditionBlock(\n", " query_time_index=CONDITIONED_STEP,\n", - " conditions=[\n", - " EqualityCondition(\n", - " column=\"equity_index_level\", value=last_equity_level * EQUITY_RALLY\n", - " )\n", - " ],\n", + " conditions=[EqualityCondition(column=PINNED_COLUMN, value=rallied_equity_level)],\n", + ")\n", + "baseline = client.forecast_mean(\n", + " history,\n", + " query_times=QUERY_TIMES,\n", + " requested_columns=plan.requested_columns,\n", + " columns=plan.columns,\n", + " seed=7,\n", ")\n", "result = client.forecast_mean(\n", " history,\n", @@ -145,7 +157,18 @@ " raise ValueError(\n", " f\"Expected {expected_forecast_rows} forecast rows, got {len(forecast)}\"\n", " )\n", - "forecast" + "\n", + "# The conditional answers one position, so joining on it keeps exactly the rows\n", + "# the two requests have in common.\n", + "comparison = baseline.to_pandas_tidy().merge(\n", + " forecast,\n", + " on=[\"query_time\", \"requested_column\"],\n", + " suffixes=(\"_unconditional\", \"_given_the_rally\"),\n", + ")\n", + "comparison[\"shift\"] = (\n", + " comparison[\"value_given_the_rally\"] - comparison[\"value_unconditional\"]\n", + ")\n", + "comparison" ] }, { @@ -157,27 +180,76 @@ }, "source": [ "## Reading the plausibility\n", - "`result.plausibility.equality_log_density` is the log density the model assigns to the pinned value before conditioning. It separates *the model is confident about NAV given this rally* from *the model finds a rally of this size absurd and is extrapolating*. The service reports the number and never refuses on it; comparing it across candidate pins, or against the density of a pin at the model's own unconditional mean, is the caller's decision." + "`result.plausibility.equality_log_density` is the log density the model assigns to the pinned value before conditioning. It separates *the model is confident about NAV given this rally* from *the model finds a rally of this size absurd and is extrapolating*. The service reports the number and never refuses on it; comparing it across candidate pins, or against the density of a pin at the model's own unconditional mean, is the caller's decision.\n", + "\n", + "It is a *density*, not a probability, and `exp()` of it is not one either: an exact value of a continuous column has probability zero, and a density carries the reciprocal units of the column it scores, so quoting the equity index in thousands of points instead of points shifts this number by `log(1000)` and can push its exponential above one. Only differences are unit-free, which is why the sentence above says compare rather than threshold: the gap between two pins on the same column is a log likelihood ratio, and that is what ranks candidate scenarios. When a pin covers several columns at once the number stays a single joint density over all of them — never a per-column value to be multiplied back together, which would assume the columns move independently." ] }, { "cell_type": "markdown", "id": "5", + "metadata": { + "id": "condition-draws-description", + "language": "markdown" + }, + "source": [ + "## Scenarios, not summaries\n", + "A mean answers *where*, and a band answers *how wide*, but a portfolio question usually needs whole futures: draws. Every row below is one coherent joint scenario, because the service draws the read-out columns together from the same conditional mixture — the NAV and the volatility in one row belong to each other, so a function of several columns can be evaluated row by row. A quantile table cannot answer that: the 90th percentile of NAV and the 90th percentile of volatility need not describe any single future.\n", + "\n", + "`diagnostics.condition_draws` reports how many draws stand behind the answer. Oversized sample requests are split into batches against the deployment's advertised cap and merged locally, and the merged count is what this field reports." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6", + "metadata": { + "id": "condition-draws", + "language": "python" + }, + "outputs": [], + "source": [ + "N_SAMPLES = 8\n", + "\n", + "scenarios = client.forecast_samples(\n", + " history,\n", + " query_times=QUERY_TIMES,\n", + " requested_columns=plan.requested_columns,\n", + " columns=plan.columns,\n", + " n_samples=N_SAMPLES,\n", + " seed=7,\n", + " condition=rally,\n", + ")\n", + "if scenarios.query_times != (QUERY_TIMES[CONDITIONED_STEP],):\n", + " raise ValueError(\n", + " f\"Expected the conditioned position alone, got {scenarios.query_times!r}\"\n", + " )\n", + "print(\"draws behind the answer:\", scenarios.diagnostics.condition_draws)\n", + "scenarios.to_pandas_wide()" + ] + }, + { + "cell_type": "markdown", + "id": "7", "metadata": { "id": "condition-interval-description", "language": "markdown" }, "source": [ "## Interval condition\n", - "Now confine the 10-year yield to a band around its last observed value instead of pinning it, and pin the equity index at the same time: a request may mix both kinds across the columns of one position. The response then carries `region_log_probability`, the log probability the model gives the yield band, and the yield column itself stays readable because its distribution inside the band is a genuine answer.\n", + "Now confine the 10-year yield to a band around its last observed value instead of pinning it, and pin the equity index at the same time: a request may mix both kinds across the columns of one position. The response then carries `region_log_probability`, and the yield column itself stays readable because its distribution inside the band is a genuine answer.\n", + "\n", + "That number is the log probability of the band **given the pinned equity index**, not the band's own probability, because the two condition kinds compose in a fixed order and the band is measured on the distribution the pin has already reduced. The two numbers therefore chain rather than describe separate things, and adding them gives the plausibility of the whole request — `log(density(pin) * P(band | pin))` — with no independence assumed anywhere. A request carrying only interval conditions has nothing to combine, and its `region_log_probability` alone is the joint probability of everything it asked about.\n", + "\n", + "With one interval column, as here, the region probability is exact — one difference of distribution functions per mixture component — and `diagnostics.interval_estimator` stays empty, because there is no estimate to characterize. Bounding a second column makes the region a box with no closed form, which the service estimates numerically and then reports the accounting for — the next section does exactly that.\n", "\n", - "With one interval column the region probability is exact. When several columns carry intervals the service estimates the probability of the box numerically and reports the accounting in `diagnostics.interval_estimator`, so the caller can judge the estimate." + "A band is also where the quantile read-out earns its place over the mean: what the banded column comes back with is a distribution inside its own range, so the cell asserts every quantile of it lands there." ] }, { "cell_type": "code", "execution_count": null, - "id": "6", + "id": "8", "metadata": { "id": "condition-interval-example", "language": "python" @@ -186,27 +258,31 @@ "source": [ "from jointfm_client import IntervalCondition\n", "\n", + "QUANTILES = [0.1, 0.5, 0.9]\n", "YIELD_BAND_HALF_WIDTH = 0.002\n", + "# A quantile of the truncated draws can sit on the band edge, and the round trip\n", + "# through the response's float encoding may move it by an ulp.\n", + "BAND_TOLERANCE = 1e-9\n", "\n", "last_yield = float(history[\"treasury_10y_yield\"].iloc[-1])\n", + "yield_lower = last_yield - YIELD_BAND_HALF_WIDTH\n", + "yield_upper = last_yield + YIELD_BAND_HALF_WIDTH\n", + "yield_band = IntervalCondition(\n", + " column=\"treasury_10y_yield\", lower=yield_lower, upper=yield_upper\n", + ")\n", "rally_with_yield_band = ConditionBlock(\n", " query_time_index=CONDITIONED_STEP,\n", " conditions=[\n", - " EqualityCondition(\n", - " column=\"equity_index_level\", value=last_equity_level * EQUITY_RALLY\n", - " ),\n", - " IntervalCondition(\n", - " column=\"treasury_10y_yield\",\n", - " lower=last_yield - YIELD_BAND_HALF_WIDTH,\n", - " upper=last_yield + YIELD_BAND_HALF_WIDTH,\n", - " ),\n", + " EqualityCondition(column=PINNED_COLUMN, value=rallied_equity_level),\n", + " yield_band,\n", " ],\n", ")\n", - "banded = client.forecast_mean(\n", + "banded = client.forecast_quantiles(\n", " history,\n", " query_times=QUERY_TIMES,\n", " requested_columns=[\"treasury_10y_yield\", *plan.requested_columns],\n", " columns=plan.columns,\n", + " quantiles=QUANTILES,\n", " seed=7,\n", " condition=rally_with_yield_band,\n", ")\n", @@ -214,16 +290,215 @@ " raise ValueError(\"A condition response must carry its plausibility block\")\n", "print(\"log density of the pinned value:\", banded.plausibility.equality_log_density)\n", "print(\"log probability of the yield band:\", banded.plausibility.region_log_probability)\n", - "print(\"interval estimator accounting:\", banded.diagnostics.interval_estimator)\n", - "banded.to_pandas_tidy()" + "if banded.diagnostics.interval_estimator is not None:\n", + " raise ValueError(\"A single interval column is exact and reports no estimator\")\n", + "band = banded.to_pandas_wide()\n", + "outside_band = band[\n", + " (band[\"treasury_10y_yield\"] < yield_lower - BAND_TOLERANCE)\n", + " | (band[\"treasury_10y_yield\"] > yield_upper + BAND_TOLERANCE)\n", + "]\n", + "if not outside_band.empty:\n", + " raise ValueError(f\"A banded column left its own interval:\\n{outside_band}\")\n", + "band" + ] + }, + { + "cell_type": "markdown", + "id": "9", + "metadata": { + "id": "condition-box-description", + "language": "markdown" + }, + "source": [ + "## When the region has to be estimated\n", + "Bounding the FX rate as well turns the region into a box. That has no closed form, so the service estimates its probability with a quasi-random walk and only then reports the accounting in `diagnostics.interval_estimator`: `points` is how many quasi-random points went into the estimate and `effective_sample_size` is Kish's effective sample size of the weights behind them. A small effective sample size against a large point count means a deep-tail box where few points carry the answer. The cell asserts both sides of that rule — absent for the single band above, present here.\n", + "\n", + "Drawing from a boxed conditional also shows what the box does to the draws. Each row pairs one truncated draw of the bounded columns with a read-out drawn from *that* draw's conditional, so the rows are exact draws from the joint conditional rather than separately summarized margins, and every one of them lands inside both bands." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "10", + "metadata": { + "id": "condition-box", + "language": "python" + }, + "outputs": [], + "source": [ + "FX_BAND_HALF_WIDTH = 0.004\n", + "\n", + "last_fx_rate = float(history[\"eur_usd_rate\"].iloc[-1])\n", + "fx_lower = last_fx_rate - FX_BAND_HALF_WIDTH\n", + "fx_upper = last_fx_rate + FX_BAND_HALF_WIDTH\n", + "fx_band = IntervalCondition(column=\"eur_usd_rate\", lower=fx_lower, upper=fx_upper)\n", + "rally_with_two_bands = ConditionBlock(\n", + " query_time_index=CONDITIONED_STEP,\n", + " conditions=[\n", + " EqualityCondition(column=PINNED_COLUMN, value=rallied_equity_level),\n", + " yield_band,\n", + " fx_band,\n", + " ],\n", + ")\n", + "boxed = client.forecast_samples(\n", + " history,\n", + " query_times=QUERY_TIMES,\n", + " requested_columns=[\"treasury_10y_yield\", \"eur_usd_rate\", *plan.requested_columns],\n", + " columns=plan.columns,\n", + " n_samples=N_SAMPLES,\n", + " seed=7,\n", + " condition=rally_with_two_bands,\n", + ")\n", + "if boxed.plausibility is None:\n", + " raise ValueError(\"A condition response must carry its plausibility block\")\n", + "estimator = boxed.diagnostics.interval_estimator\n", + "if estimator is None:\n", + " raise ValueError(\n", + " \"A multi-column region is estimated and must report its accounting\"\n", + " )\n", + "print(\n", + " \"log probability of the box given the pin:\",\n", + " boxed.plausibility.region_log_probability,\n", + ")\n", + "print(\"estimator points:\", estimator.points)\n", + "print(\"effective sample size:\", estimator.effective_sample_size)\n", + "box_draws = boxed.to_pandas_wide()\n", + "outside_box = box_draws[\n", + " (box_draws[\"treasury_10y_yield\"] < yield_lower - BAND_TOLERANCE)\n", + " | (box_draws[\"treasury_10y_yield\"] > yield_upper + BAND_TOLERANCE)\n", + " | (box_draws[\"eur_usd_rate\"] < fx_lower - BAND_TOLERANCE)\n", + " | (box_draws[\"eur_usd_rate\"] > fx_upper + BAND_TOLERANCE)\n", + "]\n", + "if not outside_box.empty:\n", + " raise ValueError(f\"A draw left the box it was conditioned into:\\n{outside_box}\")\n", + "box_draws" + ] + }, + { + "cell_type": "markdown", + "id": "11", + "metadata": { + "id": "condition-ranking-description", + "language": "markdown" + }, + "source": [ + "## Ranking candidate scenarios\n", + "The plausibility section above ended on the rule that only differences of `equality_log_density` are unit-free. This is what that buys: the gap between two pins on the same column is a log likelihood ratio, so candidate what-ifs can be ordered even though no single one of them can be thresholded.\n", + "\n", + "The table asks the model for the conditional forecast under several rallies and reports each pin's plausibility relative to the most plausible one, so a scenario the model finds far-fetched is visible next to the answer it produced. The mean is the right read-out here precisely because the question is not about one scenario's shape: it needs one comparable number per candidate." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "12", + "metadata": { + "id": "condition-ranking", + "language": "python" + }, + "outputs": [], + "source": [ + "CANDIDATE_RALLIES = [0.94, 0.98, 1.00, 1.02, 1.06]\n", + "\n", + "rankings = []\n", + "for candidate in CANDIDATE_RALLIES:\n", + " candidate_level = last_equity_level * candidate\n", + " candidate_result = client.forecast_mean(\n", + " history,\n", + " query_times=QUERY_TIMES,\n", + " requested_columns=plan.requested_columns,\n", + " columns=plan.columns,\n", + " seed=7,\n", + " condition=ConditionBlock(\n", + " query_time_index=CONDITIONED_STEP,\n", + " conditions=[EqualityCondition(column=PINNED_COLUMN, value=candidate_level)],\n", + " ),\n", + " )\n", + " if candidate_result.plausibility is None:\n", + " raise ValueError(\"A condition response must carry its plausibility block\")\n", + " conditional_mean = candidate_result.to_pandas_wide().iloc[0]\n", + " rankings.append(\n", + " {\n", + " \"rally\": candidate,\n", + " PINNED_COLUMN: candidate_level,\n", + " \"equality_log_density\": candidate_result.plausibility.equality_log_density,\n", + " **{column: conditional_mean[column] for column in plan.requested_columns},\n", + " }\n", + " )\n", + "\n", + "ranking = pd.DataFrame.from_records(rankings)\n", + "ranking[\"log_ratio_vs_best\"] = (\n", + " ranking[\"equality_log_density\"] - ranking[\"equality_log_density\"].max()\n", + ")\n", + "ranking.sort_values(\"log_ratio_vs_best\", ascending=False, ignore_index=True)" + ] + }, + { + "cell_type": "markdown", + "id": "13", + "metadata": { + "id": "condition-score-description", + "language": "markdown" + }, + "source": [ + "## Scoring an outcome you already have\n", + "Every read-out so far answers *what does the model expect*. `log_prob` asks the opposite — how plausible are values I already hold — and is the only return mode where the caller supplies the future instead of receiving it. Under a condition it scores those values against the conditional, which is how two candidate outcomes of the same what-if are compared.\n", + "\n", + "Three rules follow from scoring a joint rather than a projection of one, and the client enforces all three before anything is sent. `query_rows` carries one observed row per entry of `query_times`, each with a value for every declared column. `requested_columns` must name every *readable* column in schema order — a narrower projection would score a different distribution than the caller means to ask about — and the pinned column is the one exception, since conditioning fixed its value. The deployment refuses a row that contradicts the condition, a pinned column given another value or a banded column outside its range, instead of scoring it.\n", + "\n", + "The scores below are log densities of whole rows, so the unit caveat from the plausibility section applies unchanged: read their difference, not their level." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "14", + "metadata": { + "id": "condition-score", + "language": "python" + }, + "outputs": [], + "source": [ + "SCORED_QUERY_TIMES = QUERY_TIMES[: CONDITIONED_STEP + 1]\n", + "STRESS_NAV_DROP = 0.9\n", + "\n", + "last_row = history.iloc[-1]\n", + "continuation = {\n", + " PINNED_COLUMN: rallied_equity_level,\n", + " \"treasury_10y_yield\": last_yield,\n", + " \"eur_usd_rate\": last_fx_rate,\n", + " \"portfolio_nav\": float(last_row[\"portfolio_nav\"]),\n", + " \"realized_volatility\": float(last_row[\"realized_volatility\"]),\n", + "}\n", + "stressed = {\n", + " **continuation,\n", + " \"portfolio_nav\": continuation[\"portfolio_nav\"] * STRESS_NAV_DROP,\n", + "}\n", + "\n", + "scores = {}\n", + "for label, outcome in {\"continuation\": continuation, \"stressed\": stressed}.items():\n", + " scored = client.forecast_log_prob(\n", + " history,\n", + " query_times=SCORED_QUERY_TIMES,\n", + " query_rows=pd.DataFrame([outcome], columns=EXPECTED_COLUMNS),\n", + " requested_columns=READABLE_COLUMNS,\n", + " columns=plan.columns,\n", + " seed=7,\n", + " condition=rally,\n", + " )\n", + " scores[label] = scored.log_prob.total\n", + "\n", + "print(\"log density of the continuation:\", scores[\"continuation\"])\n", + "print(\"log density of the stressed outcome:\", scores[\"stressed\"])\n", + "print(\"log likelihood ratio:\", scores[\"continuation\"] - scores[\"stressed\"])" ] } ], "metadata": { "kernelspec": { - "display_name": "joint-client-python (3.13.3)", + "display_name": "Python (joint-client-python)", "language": "python", - "name": "python3" + "name": "joint-client-python" }, "language_info": { "codemirror_mode": { diff --git a/src/jointfm_client/__init__.py b/src/jointfm_client/__init__.py index 5972c76..ffe9a94 100644 --- a/src/jointfm_client/__init__.py +++ b/src/jointfm_client/__init__.py @@ -64,6 +64,8 @@ IMPORT_NAMESPACE, LOCAL_HEALTH_ROUTE, LOCAL_PREDICT_ROUTE, + LogProbResult, + LogProbScores, PACKAGE_VERSION, PREDICT_REQUEST_TYPE, SCHEMA_VERSION, @@ -214,6 +216,8 @@ "JSONTransport", "LOCAL_HEALTH_ROUTE", "LOCAL_PREDICT_ROUTE", + "LogProbResult", + "LogProbScores", "QuantileForecast", "QuantileForecastResult", "PathConfig", diff --git a/src/jointfm_client/adapters.py b/src/jointfm_client/adapters.py index 988cb22..d1aa57d 100644 --- a/src/jointfm_client/adapters.py +++ b/src/jointfm_client/adapters.py @@ -171,8 +171,15 @@ def infer_column_specs_from_dataframe( def dataframe_to_history_rows( frame: Any, schema: DataFrameSchema, + *, + field: str = "history_rows", ) -> list[dict[str, Any]]: - """Convert a pandas ``DataFrame`` into ordered JointFM ``history_rows``.""" + """Convert a pandas ``DataFrame`` into ordered JointFM row payloads. + + ``field`` names the payload array being built, so a bad value in the + observed rows of a scored request reports ``query_rows[...]`` rather than + pointing the caller at their history. + """ pandas_module = _require_pandas() numpy_module = _require_numpy() if not isinstance(frame, pandas_module.DataFrame): @@ -198,7 +205,7 @@ def dataframe_to_history_rows( row_payload[column_name] = _time_index_value_to_json( value, time_index_mode=schema.time_index_mode, - field=f"history_rows[{row_index}].{column_name}", + field=f"{field}[{row_index}].{column_name}", pandas_module=pandas_module, numpy_module=numpy_module, ) @@ -207,7 +214,7 @@ def dataframe_to_history_rows( row_payload[column_name] = _column_value_to_json( value, column_spec=column_spec, - field=f"history_rows[{row_index}].{column_name}", + field=f"{field}[{row_index}].{column_name}", pandas_module=pandas_module, numpy_module=numpy_module, ) @@ -298,12 +305,18 @@ def build_forecast_payload_from_dataframe( nullable_columns: Sequence[str] | None = None, bounds: ColumnBounds | None = None, condition: ConditionBlock | None = None, + query_rows: Any | None = None, ) -> dict[str, Any]: """Build a validated forecast payload from a pandas ``DataFrame``. Passing ``condition`` makes this a conditioning request: the payload then carries ``query_mode='condition'`` and the block, and the deployment answers the conditional at the one future position the block names. + + ``query_rows`` carries the *observed* values at ``query_times`` that + ``return_mode='log_prob'`` scores — a ``DataFrame`` shaped like ``frame``, + or a sequence of row mappings. The two go together: neither is accepted + without the other. """ column_specs = ( infer_column_specs_from_dataframe( @@ -337,6 +350,9 @@ def build_forecast_payload_from_dataframe( timezone=timezone, ) history_rows = dataframe_to_history_rows(frame, schema) + query_row_payloads = ( + None if query_rows is None else _query_rows_payload(query_rows, schema) + ) normalized_query_times = validate_forecast_horizon( _history_times_from_dataframe(frame, schema), query_times, @@ -355,9 +371,27 @@ def build_forecast_payload_from_dataframe( schema_version=schema_version, query_mode="forecast" if condition is None else "condition", condition=condition, + query_rows=query_row_payloads, ) +def _query_rows_payload( + query_rows: Any, + schema: DataFrameSchema, +) -> list[dict[str, Any]]: + """Normalize observed scoring rows into row payloads. + + A ``DataFrame`` goes through the same column encoding as the history, so a + categorical label or a timestamp is spelled identically in both arrays; + anything else is already a sequence of row mappings and is passed through + for ``ForecastRequest`` to validate. + """ + pandas_module = _require_pandas() + if isinstance(query_rows, pandas_module.DataFrame): + return dataframe_to_history_rows(query_rows, schema, field="query_rows") + return [dict(row) for row in query_rows] + + def build_forecast_payload_from_arrays( values: Any, *, diff --git a/src/jointfm_client/client.py b/src/jointfm_client/client.py index 2b969a8..c15b1f9 100644 --- a/src/jointfm_client/client.py +++ b/src/jointfm_client/client.py @@ -64,6 +64,7 @@ from jointfm_client.contract import ( ForecastDiagnostics, ForecastResponse, + LogProbResult, MeanForecastResult, QuantileForecastResult, SampleForecastResult, @@ -313,6 +314,7 @@ def forecast( bounds: Mapping[str, tuple[float | int | None, float | int | None]] | None = None, condition: ConditionBlock | None = None, + query_rows: Any | None = None, ) -> ForecastResponse: """Build and submit a forecast request from tabular history inputs. @@ -320,6 +322,9 @@ def forecast( future position the block names, instead of the unconditional forecast. The deployment's advertised capability is checked first, so a deployment that cannot condition is refused here rather than after a round trip. + + ``query_rows`` carries the observed values at ``query_times`` that + ``return_mode='log_prob'`` scores, in the same shape as ``history``. """ self._require_predict_url("forecast") if condition is not None: @@ -348,6 +353,7 @@ def forecast( calendar_id=calendar_id, timezone=timezone, condition=condition, + query_rows=query_rows, ) else: payload = build_forecast_payload_from_dataframe( @@ -382,6 +388,7 @@ def forecast( nullable_columns=nullable_columns, bounds=bounds, condition=condition, + query_rows=query_rows, ) sample_cap = self._resolve_sample_batch_cap(payload) if sample_cap is not None: @@ -512,6 +519,52 @@ def forecast_quantiles( ), ) + def forecast_log_prob( + self, + history: Any, + *, + query_times: Sequence[Any], + query_rows: Any, + schema: DataFrameSchema | None = None, + time_index_mode: TimeIndexMode = "ordinal", + columns: Sequence[ColumnSpec] | None = None, + time_column: str | None = None, + requested_columns: Sequence[str | int] | None = None, + model_version: str | None = None, + seed: int | None = None, + condition: ConditionBlock | None = None, + ) -> LogProbResult: + """Score observed future values through the shared forecast validation path. + + This is the one return mode that answers a question about values the + caller already has: ``query_rows`` holds the observed row at each entry + of ``query_times``, and the result is the model's log density of those + values under its joint at that position. It therefore scores the whole + joint, and ``requested_columns`` — when given at all — must name every + readable column in schema order. + + With ``condition`` the score is taken under the conditional at the one + future position the block names, and the service refuses rows that + contradict the condition instead of scoring them; see :meth:`forecast`. + """ + return cast( + LogProbResult, + self.forecast( + history, + query_times=query_times, + schema=schema, + time_index_mode=time_index_mode, + columns=columns, + time_column=time_column, + requested_columns=requested_columns, + return_mode="log_prob", + model_version=model_version, + seed=seed, + condition=condition, + query_rows=query_rows, + ), + ) + def feature_importance( self, history: Any, @@ -668,6 +721,7 @@ def _forecast_payload_from_rows( calendar_id: str, timezone: str | None, condition: ConditionBlock | None = None, + query_rows: Any | None = None, ) -> dict[str, Any]: if schema is None: if columns is None: @@ -696,6 +750,7 @@ def _forecast_payload_from_rows( schema_version=schema_version, query_mode="forecast" if condition is None else "condition", condition=condition, + query_rows=query_rows, ) def _resolve_sample_batch_cap(self, payload: Mapping[str, Any]) -> int | None: diff --git a/src/jointfm_client/contract.py b/src/jointfm_client/contract.py index 1675796..208a1d9 100644 --- a/src/jointfm_client/contract.py +++ b/src/jointfm_client/contract.py @@ -38,6 +38,9 @@ PACKAGE_VERSION: Final = importlib.metadata.version(DISTRIBUTION_NAME) SCHEMA_VERSION: Final = "v3" +# The service sums and averages its log densities in the model's own precision, +# so its summaries differ from a recomputation in the last few digits. +DERIVED_SCORE_TOLERANCE: Final = 1e-6 DATAROBOT_UNSTRUCTURED_PREDICTION_ROUTE_TEMPLATE: Final = ( "deployments/{deployment_id}/predictionsUnstructured" ) @@ -503,6 +506,7 @@ class ForecastRequest: seed: int | None = None query_row_ids: Sequence[int] | None = None condition: ConditionBlock | None = None + query_rows: Sequence[Mapping[str, Any]] | None = None def __post_init__(self) -> None: """Validate payload controls and JSON-facing request arrays.""" @@ -539,6 +543,14 @@ def __post_init__(self) -> None: "quantiles may be provided only when return_mode='quantiles'" ) + if (self.metadata.return_mode == "log_prob") != (self.query_rows is not None): + raise ValueError( + "return_mode='log_prob' and query_rows go together: scoring needs " + "the observed values, and no other return mode reads them" + ) + if self.query_rows is not None: + self._validate_query_rows() + def _validate_condition(self) -> None: """Check the condition block against this request's schema and horizons. @@ -583,11 +595,73 @@ def _validate_condition(self) -> None: "the request supplied" ) + def _validate_query_rows(self) -> None: + """Check the observed rows a log-density request scores. + + The service scores the whole joint at each future position, so a row + must carry every declared column and ``requested_columns`` must name + every readable one in schema order: a narrower projection would score a + different distribution than the caller believes it asked about. An + equality condition is the one exception, since it fixes its column's + value and the conditioning contract forbids reading it back. + + Only two per-value failures are decidable without the deployment's own + encoding and are checked here — a missing value and a non-finite number. + A categorical label passes, because the service maps it through the + column's declared mapping before scoring it. + """ + rows = _require_sequence(self.query_rows, field="query_rows") + query_times = _require_sequence(self.query_times, field="query_times") + if len(rows) != len(query_times): + raise ValueError( + f"query_rows must carry one row per query time: got {len(rows)} " + f"rows for {len(query_times)} query_times" + ) + declared = [column.name for column in self.schema.columns] + required = ( + declared + if self.schema.time_column is None + else [ + *declared, + self.schema.time_column, + ] + ) + for index, row in enumerate(rows): + row_mapping = _require_mapping(row, field=f"query_rows[{index}]") + missing = [name for name in required if name not in row_mapping] + if missing: + raise ValueError( + f"query_rows[{index}] is missing declared columns: {missing}" + ) + for name in declared: + _require_scored_value( + row_mapping[name], field=f"query_rows[{index}].{name}" + ) + + requested = _resolve_requested_columns( + self.schema.columns, self.requested_columns + ) + if requested is None: + return + pinned = set(() if self.condition is None else self.condition.pinned_columns) + readable = [name for name in declared if name not in pinned] + if requested != readable: + excused = ( + f", the pinned columns {sorted(pinned)} excepted" if pinned else "" + ) + raise ValueError( + "return_mode='log_prob' scores the whole joint, so " + "requested_columns must list every declared column in schema " + f"order{excused}: expected {readable}, got {requested}" + ) + def to_payload(self) -> dict[str, Any]: """Return a JSON-compatible forecast request without mutating inputs.""" payload = self.metadata.to_payload() payload.update(self.schema.to_payload()) - payload["history_rows"] = _serialize_history_rows(self.history_rows) + payload["history_rows"] = _serialize_rows( + self.history_rows, field="history_rows" + ) payload["query_times"] = _serialize_query_times( self.query_times, time_index_mode=self.schema.time_index_mode, @@ -606,6 +680,8 @@ def to_payload(self) -> dict[str, Any]: payload["seed"] = self.seed if self.condition is not None: payload["condition"] = self.condition.to_payload() + if self.query_rows is not None: + payload["query_rows"] = _serialize_rows(self.query_rows, field="query_rows") return payload @@ -822,6 +898,28 @@ class ConditionPlausibility: space, and each is ``None`` when the request carried no condition of that kind. The service reports them and never refuses on them, so acting on them is the caller's decision. + + Each number is joint over its whole condition block rather than one value + per column: ``equality_log_density`` covers every pinned value at once, and + ``region_log_probability`` covers the whole box at once. Do not rebuild + either by multiplying per-column numbers, which would assume the columns are + independent and so discard what a joint model is for. + + ``region_log_probability`` is measured *after* the pinned columns are + applied, so on a request carrying both kinds it is the log probability of + the box **given the pins**, not the box's own. The two therefore chain, and + their sum is the plausibility of the whole condition set:: + + equality_log_density + region_log_probability + == log( density(pins) * P(box | pins) ) + + That sum is a density times a probability. Its ``exp`` is not a + probability -- it carries the reciprocal units of the pinned columns and can + exceed one -- so compare it, never threshold it: the difference between two + scenarios pinning the same columns is a log likelihood ratio and is + unit-free. For a request carrying only interval conditions there is nothing + to combine and ``region_log_probability`` alone is the joint probability of + the conditioned event. """ equality_log_density: float | None = None @@ -903,6 +1001,86 @@ def to_numpy(self) -> Any: return numpy_module.asarray(self.values, dtype=float) +@dataclass(frozen=True, slots=True) +class LogProbScores: + """Per-horizon log densities of one scored request, with its summaries. + + The service reports the same numbers several times over: ``values`` holds + the log density of each scored future position, ``nll_values`` their + negatives, and the four scalars the sum and the mean of each. Parsing + recomputes every redundant field from ``values`` instead of trusting it, so + a truncated or mismatched payload fails here rather than reading as a score. + + A log density is joint over the columns that were scored, so the block has + no column axis however wide the request was: one number per horizon. + """ + + values: tuple[float, ...] + nll_values: tuple[float, ...] + total: float + mean: float + nll_total: float + nll_mean: float + + @classmethod + def from_payload( + cls, + payload: Mapping[str, Any], + *, + expected_horizon_count: int | None = None, + ) -> Self: + """Parse one ``outputs.log_prob`` block and check its derived fields.""" + values = _require_float_vector( + payload.get("values"), + field="outputs.log_prob.values", + expected_length=expected_horizon_count, + ) + nll_values = _require_float_vector( + payload.get("nll_values"), + field="outputs.log_prob.nll_values", + expected_length=len(values), + ) + for index, (score, negated) in enumerate(zip(values, nll_values, strict=True)): + _require_derived_score( + negated, + expected=-score, + field=f"outputs.log_prob.nll_values[{index}]", + ) + summed = math.fsum(values) + total = _require_derived_score( + _require_output_float(payload.get("total"), field="outputs.log_prob.total"), + expected=summed, + field="outputs.log_prob.total", + ) + mean = _require_derived_score( + _require_output_float(payload.get("mean"), field="outputs.log_prob.mean"), + expected=summed / len(values), + field="outputs.log_prob.mean", + ) + nll_total = _require_derived_score( + _require_output_float( + payload.get("nll_total"), field="outputs.log_prob.nll_total" + ), + expected=-total, + field="outputs.log_prob.nll_total", + ) + nll_mean = _require_derived_score( + _require_output_float( + payload.get("nll_mean"), field="outputs.log_prob.nll_mean" + ), + expected=-mean, + field="outputs.log_prob.nll_mean", + ) + return cls( + values=values, + nll_values=nll_values, + total=total, + mean=mean, + nll_total=nll_total, + nll_mean=nll_mean, + ) + + @dataclass(frozen=True, slots=True) class ForecastOutputs: """Legacy-shaped forecast output arrays for one parsed forecast result.""" @@ -912,6 +1090,7 @@ class ForecastOutputs: mean: tuple[tuple[float, ...], ...] | None = None samples: tuple[tuple[tuple[float, ...], ...], ...] | None = None quantiles: tuple[QuantileForecast, ...] | None = None + log_prob: LogProbScores | None = None @classmethod def from_payload( @@ -948,6 +1127,7 @@ def from_payload( mean: tuple[tuple[float, ...], ...] | None = None samples: tuple[tuple[tuple[float, ...], ...], ...] | None = None quantiles: tuple[QuantileForecast, ...] | None = None + log_prob: LogProbScores | None = None if return_mode == "mean": mean = _require_float_matrix( @@ -958,6 +1138,7 @@ def from_payload( ) _require_none(payload.get("samples"), field="outputs.samples") _require_none(payload.get("quantiles"), field="outputs.quantiles") + _require_none(payload.get("log_prob"), field="outputs.log_prob") elif return_mode == "samples": samples = _require_float_tensor3( payload.get("samples"), @@ -968,7 +1149,8 @@ def from_payload( ) _require_none(payload.get("mean"), field="outputs.mean") _require_none(payload.get("quantiles"), field="outputs.quantiles") - else: + _require_none(payload.get("log_prob"), field="outputs.log_prob") + elif return_mode == "quantiles": quantiles = _require_quantile_forecasts( payload.get("quantiles"), field="outputs.quantiles", @@ -978,6 +1160,21 @@ def from_payload( ) _require_none(payload.get("mean"), field="outputs.mean") _require_none(payload.get("samples"), field="outputs.samples") + _require_none(payload.get("log_prob"), field="outputs.log_prob") + elif return_mode == "log_prob": + log_prob = LogProbScores.from_payload( + _require_mapping(payload.get("log_prob"), field="outputs.log_prob"), + expected_horizon_count=horizon_count, + ) + _require_none(payload.get("mean"), field="outputs.mean") + _require_none(payload.get("samples"), field="outputs.samples") + _require_none(payload.get("quantiles"), field="outputs.quantiles") + else: + # A mode nobody parses must fail here rather than fall through to + # another mode's branch and be read as that mode's shape. + raise ValueError( + f"outputs cannot be parsed for return_mode {return_mode!r}" + ) return cls( query_times=query_times, @@ -985,6 +1182,7 @@ def from_payload( mean=mean, samples=samples, quantiles=quantiles, + log_prob=log_prob, ) @@ -1168,8 +1366,11 @@ def from_payload( if return_mode == "samples": assert outputs.samples is not None return SampleForecastResult(samples=outputs.samples, **shared_fields) - assert outputs.quantiles is not None - return QuantileForecastResult(quantiles=outputs.quantiles, **shared_fields) + if return_mode == "quantiles": + assert outputs.quantiles is not None + return QuantileForecastResult(quantiles=outputs.quantiles, **shared_fields) + assert outputs.log_prob is not None + return LogProbResult(log_prob=outputs.log_prob, **shared_fields) @dataclass(frozen=True, slots=True) @@ -1370,6 +1571,66 @@ def to_pandas_wide(self) -> Any: return pandas_module.DataFrame.from_records(rows) +@dataclass(frozen=True, slots=True) +class LogProbResult(ForecastResponse): + """Parsed log densities of observed values, with shared metadata. + + This result carries no forecast values, because a scored request supplies + them itself as ``query_rows``: what comes back is how plausible the model + finds them. Each score is joint over the scored columns, so the conversions + have a horizon axis and no column axis. + """ + + log_prob: LogProbScores + + @property + def outputs(self) -> ForecastOutputs: + """Return the legacy nested outputs view for scored requests.""" + return ForecastOutputs( + query_times=self.query_times, + requested_columns=self.requested_columns, + log_prob=self.log_prob, + ) + + def to_numpy(self) -> Any: + """Return NumPy log densities with axis order ``(horizon,)``.""" + numpy_module = _require_numpy_module() + return numpy_module.asarray(self.log_prob.values, dtype=float) + + def to_pandas_tidy(self) -> Any: + """Return a tidy DataFrame with ``query_time``, ``log_prob``, and ``nll``.""" + pandas_module = _require_pandas_module() + rows = [ + {"query_time": query_time, "log_prob": score, "nll": negated} + for query_time, score, negated in zip( + self.query_times, + self.log_prob.values, + self.log_prob.nll_values, + strict=True, + ) + ] + return pandas_module.DataFrame.from_records(rows) + + def to_pandas_wide(self) -> Any: + """Return the one-row request summary over the scored horizons. + + The other result classes widen the column axis, which a log density does + not have. The wide view is therefore the request-level summary the + service reports, and the per-horizon scores stay in the tidy view. + """ + pandas_module = _require_pandas_module() + return pandas_module.DataFrame.from_records( + [ + { + "total": self.log_prob.total, + "mean": self.log_prob.mean, + "nll_total": self.log_prob.nll_total, + "nll_mean": self.log_prob.nll_mean, + } + ] + ) + + def build_forecast_payload( *, model_version: str, @@ -1384,6 +1645,7 @@ def build_forecast_payload( schema_version: str = SCHEMA_VERSION, query_mode: QueryMode = "forecast", condition: ConditionBlock | None = None, + query_rows: Sequence[Mapping[str, Any]] | None = None, ) -> dict[str, Any]: """Build a validated JSON-compatible forecast request payload.""" return ForecastRequest( @@ -1401,6 +1663,7 @@ def build_forecast_payload( quantiles=quantiles, seed=seed, condition=condition, + query_rows=query_rows, ).to_payload() @@ -1630,18 +1893,27 @@ def _validate_history_declared_columns( raise ValueError(f"history_rows are missing time_column {schema.time_column!r}") -def _serialize_history_rows( - history_rows: Sequence[Mapping[str, Any]], +def _serialize_rows( + rows: Sequence[Mapping[str, Any]], + *, + field: str, ) -> list[dict[str, Any]]: return [ { - key: _to_json_compatible(value, field=f"history_rows[{index}].{key}") - for key, value in history_row.items() + key: _to_json_compatible(value, field=f"{field}[{index}].{key}") + for key, value in row.items() } - for index, history_row in enumerate(history_rows) + for index, row in enumerate(rows) ] +def _require_scored_value(value: Any, *, field: str) -> None: + if value is None: + raise ValueError(f"{field} must carry an observed value to score") + if isinstance(value, float) and not math.isfinite(value): + raise ValueError(f"{field} must be finite") + + def _serialize_query_times( query_times: Sequence[Any], *, @@ -1979,6 +2251,38 @@ def _require_output_float(value: Any, *, field: str) -> float: return parsed +def _require_float_vector( + value: Any, + *, + field: str, + expected_length: int | None = None, +) -> tuple[float, ...]: + values = _require_sequence(value, field=field) + _validate_length( + actual_length=len(values), + expected_length=expected_length, + field=field, + ) + return tuple( + _require_output_float(item, field=f"{field}[{index}]") + for index, item in enumerate(values) + ) + + +def _require_derived_score(value: float, *, expected: float, field: str) -> float: + if not math.isclose( + value, + expected, + rel_tol=DERIVED_SCORE_TOLERANCE, + abs_tol=DERIVED_SCORE_TOLERANCE, + ): + raise ValueError( + f"{field} disagrees with outputs.log_prob.values: " + f"expected {expected}, got {value}" + ) + return value + + def _require_float_matrix( value: Any, *, diff --git a/tests/fixtures/condition_log_prob_request.json b/tests/fixtures/condition_log_prob_request.json new file mode 100644 index 0000000..2407223 --- /dev/null +++ b/tests/fixtures/condition_log_prob_request.json @@ -0,0 +1,51 @@ +{ + "schema_version": "v3", + "model_version": "jointfm-inference:0.3.0+ckpt.sdk-test", + "query_mode": "condition", + "return_mode": "log_prob", + "time_index_mode": "ordinal", + "columns": [ + { + "name": "driver", + "modality": "numeric" + }, + { + "name": "target", + "modality": "numeric", + "role": "target" + } + ], + "history_rows": [ + { + "driver": 1.0, + "target": 10.0 + }, + { + "driver": 1.1, + "target": 11.0 + } + ], + "query_times": [2, 3], + "requested_columns": ["target"], + "condition": { + "query_time_index": 1, + "conditions": [ + { + "column": "driver", + "kind": "equality", + "value": 1.5 + } + ] + }, + "query_rows": [ + { + "driver": 1.2, + "target": 12.0 + }, + { + "driver": 1.5, + "target": 13.0 + } + ], + "seed": 7 +} diff --git a/tests/fixtures/condition_log_prob_response.json b/tests/fixtures/condition_log_prob_response.json new file mode 100644 index 0000000..21d9f62 --- /dev/null +++ b/tests/fixtures/condition_log_prob_response.json @@ -0,0 +1,36 @@ +{ + "schema_version": "v3", + "image_version": "0.3.0", + "model_version": "jointfm-inference:0.3.0+ckpt.sdk-test", + "checkpoint_version": "sdk-test", + "head": "gmm", + "query_mode": "condition", + "return_mode": "log_prob", + "outputs": { + "query_times": [3], + "requested_columns": ["target"], + "mean": null, + "samples": null, + "quantiles": null, + "log_prob": { + "values": [-1.75], + "nll_values": [1.75], + "total": -1.75, + "mean": -1.75, + "nll_total": 1.75, + "nll_mean": 1.75 + } + }, + "plausibility": { + "equality_log_density": -1.27, + "region_log_probability": null + }, + "diagnostics": { + "history_rows": 2, + "horizon_count": 1, + "seed": 7, + "condition_draws": 500, + "interval_estimator": null + }, + "errors": [] +} diff --git a/tests/fixtures/forecast_log_prob_request.json b/tests/fixtures/forecast_log_prob_request.json new file mode 100644 index 0000000..f843ce4 --- /dev/null +++ b/tests/fixtures/forecast_log_prob_request.json @@ -0,0 +1,41 @@ +{ + "schema_version": "v3", + "model_version": "jointfm-inference:0.3.0+ckpt.sdk-test", + "query_mode": "forecast", + "return_mode": "log_prob", + "time_index_mode": "ordinal", + "columns": [ + { + "name": "driver", + "modality": "numeric" + }, + { + "name": "target", + "modality": "numeric", + "role": "target" + } + ], + "history_rows": [ + { + "driver": 1.0, + "target": 10.0 + }, + { + "driver": 1.1, + "target": 11.0 + } + ], + "query_times": [2, 3], + "requested_columns": ["driver", "target"], + "query_rows": [ + { + "driver": 1.2, + "target": 12.0 + }, + { + "driver": 1.3, + "target": 13.0 + } + ], + "seed": 7 +} diff --git a/tests/fixtures/forecast_log_prob_response.json b/tests/fixtures/forecast_log_prob_response.json new file mode 100644 index 0000000..f6ad94a --- /dev/null +++ b/tests/fixtures/forecast_log_prob_response.json @@ -0,0 +1,33 @@ +{ + "schema_version": "v3", + "image_version": "0.3.0", + "model_version": "jointfm-inference:0.3.0+ckpt.sdk-test", + "checkpoint_version": "sdk-test", + "head": "gmm", + "query_mode": "forecast", + "return_mode": "log_prob", + "outputs": { + "query_times": [2, 3], + "requested_columns": ["driver", "target"], + "mean": null, + "samples": null, + "quantiles": null, + "log_prob": { + "values": [-2.5, -3.25], + "nll_values": [2.5, 3.25], + "total": -5.75, + "mean": -2.875, + "nll_total": 5.75, + "nll_mean": 2.875 + } + }, + "plausibility": null, + "diagnostics": { + "history_rows": 2, + "horizon_count": 2, + "seed": 7, + "condition_draws": null, + "interval_estimator": null + }, + "errors": [] +} diff --git a/tests/test_fixture_compatibility.py b/tests/test_fixture_compatibility.py index e699ef8..f7a79cf 100644 --- a/tests/test_fixture_compatibility.py +++ b/tests/test_fixture_compatibility.py @@ -26,6 +26,7 @@ HealthMetadata, SCHEMA_VERSION, JointFMServiceError, + LogProbResult, MeanForecastResult, QuantileForecastResult, SampleForecastResult, @@ -78,6 +79,18 @@ def test_health_fixture_matches_current_service_contract( QuantileForecastResult, "quantiles", ), + ( + "forecast_log_prob_request", + "forecast_log_prob_response", + LogProbResult, + "log_prob", + ), + ( + "condition_log_prob_request", + "condition_log_prob_response", + LogProbResult, + "log_prob", + ), ], ) def test_checked_in_fixture_payloads_parse_as_forecast_results( diff --git a/tests/test_log_prob_mode.py b/tests/test_log_prob_mode.py new file mode 100644 index 0000000..cdb6e01 --- /dev/null +++ b/tests/test_log_prob_mode.py @@ -0,0 +1,363 @@ +# Copyright 2026 DataRobot, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for the ``log_prob`` return mode on the client side. + +Scoring inverts the direction of every other return mode: the caller supplies +the values and the deployment answers how plausible they are. That shows up in +three places, and the split below follows them. The request carries observed +rows nothing else carries, and a projection narrower than the scored joint asks +a different question than the caller believes they asked. The response carries +a block with no column axis whose summary fields only repeat what ``values`` +already says. Under a condition the two meet: a pinned column leaves the +read-out set but still has to be supplied, because the service refuses a row +contradicting the pin rather than scoring it. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from typing import Any + +import pytest + +from jointfm_client import ( + ColumnSpec, + ConditionBlock, + ConditionPlausibility, + DataFrameSchema, + EqualityCondition, + ForecastRequest, + ForecastRequestMetadata, + ForecastResponse, + JointFMClient, + LogProbResult, + build_forecast_payload_from_dataframe, +) +from jointfm_client.contract import ReturnMode + +_MODEL_VERSION = "jointfm-inference:0.3.0+ckpt.sdk-test" +_HISTORY_ROWS = ( + {"driver": 1.0, "target": 10.0}, + {"driver": 1.1, "target": 11.0}, +) +_QUERY_ROWS = ( + {"driver": 1.2, "target": 12.0}, + {"driver": 1.3, "target": 13.0}, +) +_PINNED_QUERY_ROWS = ( + {"driver": 1.2, "target": 12.0}, + {"driver": 1.5, "target": 13.0}, +) +_PIN_DRIVER = ConditionBlock( + query_time_index=1, + conditions=(EqualityCondition(column="driver", value=1.5),), +) + + +def _schema() -> DataFrameSchema: + """Build one two-column ordinal schema for request-level tests.""" + return DataFrameSchema( + columns=( + ColumnSpec(name="driver", modality="numeric"), + ColumnSpec(name="target", modality="numeric", role="target"), + ), + time_index_mode="ordinal", + ) + + +def _request( + *, + return_mode: ReturnMode = "log_prob", + query_rows: Any = _QUERY_ROWS, + requested_columns: tuple[str, ...] | None = ("driver", "target"), + condition: ConditionBlock | None = None, +) -> ForecastRequest: + """Build one scoring request against the two-column schema.""" + return ForecastRequest( + metadata=ForecastRequestMetadata( + model_version=_MODEL_VERSION, + query_mode="forecast" if condition is None else "condition", + return_mode=return_mode, + ), + schema=_schema(), + history_rows=_HISTORY_ROWS, + query_times=(2, 3), + requested_columns=requested_columns, + condition=condition, + query_rows=query_rows, + seed=7, + ) + + +class _ScoringTransport: + """Fake JSON transport for a deployment that serves one canned score.""" + + def __init__( + self, + *, + health_payload: Mapping[str, Any], + predict_payload: Mapping[str, Any], + ) -> None: + """Remember the advertisement and the canned score to serve.""" + self.health_payload = health_payload + self.predict_payload = predict_payload + self.payloads: list[dict[str, Any]] = [] + + def get_json(self, url: str) -> Mapping[str, Any]: + """Serve the health advertisement on the local health route.""" + assert url == "http://127.0.0.1:8080/healthz" + return self.health_payload + + def post_json(self, url: str, payload: Mapping[str, Any]) -> Mapping[str, Any]: + """Record the scoring payload and answer it.""" + assert url == "http://127.0.0.1:8080/predict" + self.payloads.append(dict(payload)) + return self.predict_payload + + +def _client(transport: _ScoringTransport) -> JointFMClient: + """Build a local-service client over the fake transport.""" + return JointFMClient( + health_url="http://127.0.0.1:8080/healthz", + predict_url="http://127.0.0.1:8080/predict", + transport=transport, + ) + + +def test_scoring_needs_the_values_it_scores() -> None: + """The mode and the observed rows are one request, not two independent options.""" + with pytest.raises(ValueError, match="go together"): + _request(query_rows=None) + + with pytest.raises(ValueError, match="go together"): + _request(return_mode="mean", requested_columns=("target",)) + + +def test_every_future_position_needs_its_observed_row() -> None: + """One score exists per query time, so a missing row has no answer to give.""" + with pytest.raises(ValueError, match="one row per query time"): + _request(query_rows=(_QUERY_ROWS[0],)) + + +@pytest.mark.parametrize( + ("row", "message"), + [ + ({"driver": 1.3}, "missing declared columns"), + ({"driver": 1.3, "target": None}, "must carry an observed value"), + ({"driver": 1.3, "target": float("nan")}, "must be finite"), + ], +) +def test_a_row_that_cannot_be_scored_is_refused_before_any_round_trip( + row: Mapping[str, Any], + message: str, +) -> None: + """The score is joint over every declared column, so each one needs a value.""" + with pytest.raises(ValueError, match=message): + _request(query_rows=(_QUERY_ROWS[0], row)) + + +def test_a_partial_projection_is_refused_because_it_scores_something_else() -> None: + """Dropping a column would score a narrower joint than the caller believes.""" + with pytest.raises(ValueError, match="every declared column in schema order"): + _request(requested_columns=("target",)) + + +def test_a_pinned_column_is_the_one_column_a_score_may_leave_out() -> None: + """Conditioning fixes the column, and the contract forbids reading it back.""" + pinned = _request( + requested_columns=("target",), + query_rows=_PINNED_QUERY_ROWS, + condition=_PIN_DRIVER, + ) + assert pinned.to_payload()["requested_columns"] == ["target"] + + default_projection = _request( + requested_columns=None, + query_rows=_PINNED_QUERY_ROWS, + condition=_PIN_DRIVER, + ) + assert "requested_columns" not in default_projection.to_payload() + + with pytest.raises(ValueError, match="lists pinned columns"): + _request(query_rows=_PINNED_QUERY_ROWS, condition=_PIN_DRIVER) + + +@pytest.mark.parametrize( + ("fixture_name", "condition", "requested_columns", "query_rows"), + [ + ("forecast_log_prob_request", None, ("driver", "target"), _QUERY_ROWS), + ( + "condition_log_prob_request", + _PIN_DRIVER, + ("target",), + _PINNED_QUERY_ROWS, + ), + ], +) +def test_the_payload_carries_the_rows_the_service_scores( + json_fixture_loader: Callable[[str], dict[str, Any]], + fixture_name: str, + condition: ConditionBlock | None, + requested_columns: tuple[str, ...], + query_rows: tuple[Mapping[str, Any], ...], +) -> None: + """What the request builds is what the checked-in service fixture contains.""" + payload = _request( + requested_columns=requested_columns, + condition=condition, + query_rows=query_rows, + ).to_payload() + + assert payload == json_fixture_loader(fixture_name) + + +def test_the_scored_response_parses_into_its_own_result( + json_fixture_loader: Callable[[str], dict[str, Any]], +) -> None: + """A score has a horizon axis and no column axis, and says so in its result.""" + result = ForecastResponse.from_payload( + json_fixture_loader("forecast_log_prob_response"), + request_payload=json_fixture_loader("forecast_log_prob_request"), + ) + + assert isinstance(result, LogProbResult) + assert result.log_prob.values == (-2.5, -3.25) + assert result.log_prob.nll_values == (2.5, 3.25) + assert result.log_prob.total == pytest.approx(-5.75) + assert result.log_prob.mean == pytest.approx(-2.875) + assert result.plausibility is None + assert result.outputs.log_prob is result.log_prob + + +def test_the_scored_conversions_keep_the_horizon_axis( + json_fixture_loader: Callable[[str], dict[str, Any]], +) -> None: + """Tidy carries one row per scored position; wide carries the request summary.""" + pytest.importorskip("pandas") + pytest.importorskip("numpy") + result = ForecastResponse.from_payload( + json_fixture_loader("forecast_log_prob_response"), + request_payload=json_fixture_loader("forecast_log_prob_request"), + ) + assert isinstance(result, LogProbResult) + + assert result.to_numpy().tolist() == [-2.5, -3.25] + tidy = result.to_pandas_tidy() + assert list(tidy.columns) == ["query_time", "log_prob", "nll"] + assert tidy["query_time"].tolist() == [2, 3] + wide = result.to_pandas_wide() + assert list(wide.columns) == ["total", "mean", "nll_total", "nll_mean"] + assert len(wide) == 1 + + +@pytest.mark.parametrize("field", ["total", "mean", "nll_total", "nll_mean"]) +def test_a_summary_that_disagrees_with_its_scores_is_refused( + json_fixture_loader: Callable[[str], dict[str, Any]], + field: str, +) -> None: + """The summaries are redundant, so a payload that disagrees with itself is broken.""" + payload = json_fixture_loader("forecast_log_prob_response") + payload["outputs"]["log_prob"][field] += 1.0 + + with pytest.raises(ValueError, match=f"outputs.log_prob.{field} disagrees"): + ForecastResponse.from_payload(payload) + + +def test_a_scored_response_is_never_read_as_another_mode( + json_fixture_loader: Callable[[str], dict[str, Any]], +) -> None: + """A mode whose block is absent must fail on its own field, not on another's.""" + payload = json_fixture_loader("forecast_mean_response") + payload["return_mode"] = "log_prob" + + with pytest.raises(ValueError, match="outputs.log_prob"): + ForecastResponse.from_payload(payload) + + +def test_the_client_scores_observed_rows_and_types_the_answer( + json_fixture_loader: Callable[[str], dict[str, Any]], +) -> None: + """The typed helper puts the rows on the wire and types what comes back.""" + transport = _ScoringTransport( + health_payload=json_fixture_loader("health_metadata"), + predict_payload=json_fixture_loader("forecast_log_prob_response"), + ) + + result = _client(transport).forecast_log_prob( + list(_HISTORY_ROWS), + schema=_schema(), + query_times=[2, 3], + query_rows=list(_QUERY_ROWS), + requested_columns=["driver", "target"], + model_version=_MODEL_VERSION, + seed=7, + ) + + assert isinstance(result, LogProbResult) + assert result.log_prob.values == (-2.5, -3.25) + sent = transport.payloads[0] + assert sent["return_mode"] == "log_prob" + assert sent["query_rows"] == [dict(row) for row in _QUERY_ROWS] + + +def test_the_client_scores_under_a_condition( + json_fixture_loader: Callable[[str], dict[str, Any]], +) -> None: + """A conditioned score answers at the conditioned position and reports its pin.""" + health_payload = json_fixture_loader("health_metadata") + health_payload["supported_query_modes"] = ["forecast", "condition"] + health_payload["supported_condition_kinds"] = ["equality", "interval"] + transport = _ScoringTransport( + health_payload=health_payload, + predict_payload=json_fixture_loader("condition_log_prob_response"), + ) + + result = _client(transport).forecast_log_prob( + list(_HISTORY_ROWS), + schema=_schema(), + query_times=[2, 3], + query_rows=list(_PINNED_QUERY_ROWS), + requested_columns=["target"], + model_version=_MODEL_VERSION, + seed=7, + condition=_PIN_DRIVER, + ) + + assert isinstance(result, LogProbResult) + assert result.query_times == (3,) + assert result.log_prob.values == (-1.75,) + assert result.plausibility == ConditionPlausibility(equality_log_density=-1.27) + sent = transport.payloads[0] + assert sent["query_mode"] == "condition" + assert sent["query_rows"] == [dict(row) for row in _PINNED_QUERY_ROWS] + + +def test_the_dataframe_adapter_encodes_observed_rows_like_history() -> None: + """A pandas caller gets both arrays through the same column encoding.""" + pandas = pytest.importorskip("pandas") + + payload = build_forecast_payload_from_dataframe( + pandas.DataFrame(list(_HISTORY_ROWS)), + model_version=_MODEL_VERSION, + time_index_mode="ordinal", + query_times=[2, 3], + target_columns=["target"], + requested_columns=["driver", "target"], + return_mode="log_prob", + query_rows=pandas.DataFrame(list(_QUERY_ROWS)), + ) + + assert payload["return_mode"] == "log_prob" + assert payload["query_rows"] == [dict(row) for row in _QUERY_ROWS]