Skip to content

Your first model

In this walkthrough, you fit a demand model for three stores and two products. You forecast the next four weeks, and the forecast includes a store that is not open yet. You also see the errors that stop dishonest answers. You run all the code in a notebook, one cell at a time. The complete script is docs/snippets/quickstart.py. The test suite runs it on each change.

flowchart LR
    D["1. Labelled data<br/><code>xr.Dataset</code>"] --> S["2. Schema<br/>dims, roles, policies"]
    S --> P["3. Spec<br/>pipeline + model"]
    P --> V["4. Static check<br/>no data touched"]
    V --> F["5. Fit<br/>evaluate, then refit"]
    F --> SC["6. Score<br/>forecast, new store"]
    SC --> B["7. Bundle<br/>write, read, verify"]

1. Labelled data

Your data is an xr.Dataset. Each variable has named dims, and each dim has coordinates (labels). No later part of the process uses the sequence of stores or products.

import numpy as np
import pandas as pd
import xarray as xr

rng = np.random.default_rng(7)
stores = ["north", "south", "east"]
skus = ["apples", "pears"]
weeks = pd.date_range("2025-01-05", periods=52, freq="W-SUN")
shape = (len(stores), len(skus), len(weeks))

price = 3.0 * np.exp(rng.normal(0.0, 0.1, shape))
promo = (rng.uniform(size=shape) < 0.2).astype(float)
units = rng.poisson(np.exp(2.5 - 1.2 * np.log(price / 3.0) + 0.5 * promo)).astype(float)

dims = ("store", "sku", "week")
data = xr.Dataset(
    {"units_sold": (dims, units), "price": (dims, price), "promo": (dims, promo)},
    coords={"store": stores, "sku": skus, "week": weeks},
)

Coming from pandas?

A long DataFrame with one row for each store, sku and week becomes this dataset with df.set_index(["store", "sku", "week"]).to_xarray().

2. Describe it with a schema

A DataSchema tells what each dim is. It also tells what can occur to the dim when you score:

from mapi_spec import DataSchema

schema = DataSchema.model_validate(
    {
        "dims": {
            # a store we have never seen may be scored: it gets the pooled prediction
            "store": {"kind": "categorical", "policy": "new_levels", "new_level_strategy": "pooled"},
            # the product range is fixed: an unknown sku is an error
            "sku": {"kind": "categorical", "policy": "fixed"},
            # time extends into the future, one Sunday-ending week at a time
            "week": {"kind": "time", "policy": "extend", "freq": "W-SUN"},
        },
        "vars": {
            "units_sold": {"dims": list(dims), "dtype": "float64", "role": "target"},
            "price": {
                "dims": list(dims),
                "dtype": "float64",
                "role": "covariate",
                "units": "EUR",
                "future_required": True,  # a forecast must say what the price will be
            },
            "promo": {
                "dims": list(dims),
                "dtype": "float64",
                "role": "covariate",
                "future_required": True,
            },
        },
    }
)
Dim Policy At score time
store new_levels (pooled) Stores that the fit did not see are permitted. They get the pooled prediction.
sku fixed Only the fitted products. All other products are UNSEEN_LEVELS.
week extend New weeks must continue the fitted range. Gaps are not permitted.

If you do not write the schema, mapi_sdk.infer_schema(data, targets=[...]) makes a draft. In the draft, time dims become extend and all other dims become fixed.

3. Write the spec

A ModelSpec does these things:

  • It names the dataset.
  • It binds your names to the canonical names of the model.
  • It selects a train/holdout split.
  • It lists the preprocessing steps.
  • It selects the model.
from mapi_spec import load_model_spec

spec = load_model_spec("""
apiVersion: mapi/v1
kind: ModelSpec
metadata: {name: fruit-demand, project: quickstart}
data:
  dataset: dataset://quickstart/fruit_weekly@latest
  bind:                                   # canonical name: your name
    dims: {store: store, sku: sku, time: week}
    vars: {y: units_sold, price: price, promo: promo}
  split: {kind: time_holdout, dim: time, last: 8}
pipeline:
  - {id: cal,  use: core.calendar@^1, with: {along: time, features: [week_of_year]}}
  - {id: logy, use: core.log1p@^1,    with: {vars: [y]}}
model:
  use: tabular.regressor@^1
  settings:
    estimator: {kind: xgboost.XGBRegressor, params: {n_estimators: 60, max_depth: 3}}
    rows: [store, sku, time]
    coord_features: {store: category, sku: category}
  fit: {seed: 7}
evaluate: {metrics: [rmse, mape, coverage_80]}
""")
  • bind maps canonical names onto the names in your dataset. Steps and models use the canonical names. Outputs come back with your names.
  • core.calendar adds week-of-year features. core.log1p models log(1 + units). The pipeline inverts it automatically on the way out.
  • tabular.regressor is a built-in quantile model (xgboost underneath). It stacks its rows from the labelled dims. Store and sku become categorical features.

4. Check it before you fit it

This is the same static check that the API does before it accepts a job. It reads no data, and it takes milliseconds.

from mapi_sdk import default_registry
from mapi_spec import validate_spec

registry = default_registry()  # every installed step, model and plot (entry points)
check = validate_spec(spec, registry.catalog(), schema)
print(check.ok)  # True: names, dims, settings, split stage and leakage all check out
print(check.resolved.model.resolved)  # tabular.regressor@1.0.0 (the semver range, pinned)

The check finds these errors, among others:

  • An incorrect variable name.
  • A split on a dim that does not exist yet.
  • A stateful step before the split.

Each error has a stable code and a JSON pointer, for example /pipeline/1/with/vars/0.

5. Fit

from mapi_sdk import fit_spec

bundle = fit_spec(spec, data, registry=registry, raw_schema=schema)
print(bundle.metrics.values)  # holdout metrics, in units sold: {'rmse': ..., 'mape': ..., ...}
print(bundle.metrics.n_holdout)  # 48 cells: 3 stores x 2 skus x 8 weeks

With the default fit.mode: evaluate_then_refit, fit_spec does these steps:

  1. It fits each stateful step and the model on the first 44 weeks.
  2. It scores the last 8 weeks through the score path, exactly as a forecast. It calculates rmse, mape and coverage_80 in units sold (it undoes the log1p).
  3. It fits again on all 52 weeks, so that a forecast starts from the latest data. It links the evaluation fit in the bundle.

Metrics and params of the fit

The bundle keeps the metrics and params that the model logs in the model group metrics_history. To also get the params of the spec in memory, give the fit a recorder: fit_spec(..., hooks=FitHooks(tracker=MetricRecorder())), with FitHooks from mapi_sdk and MetricRecorder from mapi_sdk.tracking. On the platform, the run page shows these values. See Track metrics, params and tags.

6. Forecast, with a new store

Supply the plan for the next four weeks. Prices and promotions are future_required. The target is not necessary. The fit did not see west.

from mapi_sdk import score_bundle

future_weeks = pd.date_range(weeks[-1] + pd.Timedelta(weeks=1), periods=4, freq="W-SUN")
plan = xr.Dataset(
    {
        "price": (dims, np.full((4, 2, 4), 2.9)),
        "promo": (dims, np.zeros((4, 2, 4))),
    },
    # "west" opens next month: the model has never seen it
    coords={"store": [*stores, "west"], "sku": skus, "week": future_weeks},
)

forecast = score_bundle(
    bundle,
    """
    apiVersion: mapi/v1
    kind: ScoreRequest
    model: endpoint://quickstart/fruit-demand
    mode: forecast
    horizon: {dim: time, periods: 4}
    output: {summary: [median, q10, q90]}
    """,
    plan,
    registry=registry,
)
summary = forecast["summary"].to_dataset()
print(summary["y_hat"].dims)  # ('stat', 'store', 'sku', 'week'), in your names
print(summary["y_hat"].sel(stat="median", store="west").values.round(1))
print(forecast.attrs["representation"])  # quantiles: what this model's uncertainty is
print(forecast.attrs["output_space"])  # {"y_hat": "raw"}: log1p undone, back in units sold
print(forecast.attrs["issues"])  # an info notice: 1 new level of 'store' uses 'pooled'

The result is an xr.DataTree. summary holds the statistics that you asked for, along a stat dim, with your dim names. Its attributes tell you how to read it:

Attribute Value here Meaning
representation quantiles Quantiles for each cell. You can read them, but you cannot add them.
output_space {"y_hat": "raw"} The pipeline inverted the values through each value step, back to the space of the raw data.
issues UNSEEN_LEVELS (info) The model scored west with the pooled strategy.

7. The errors that protect you

These requests fail before model code runs. Each failure has a stable code:

from mapi_spec import ErrorCode, MapiError


def try_score(request_yaml: str, payload: xr.Dataset) -> ErrorCode:
    try:
        score_bundle(bundle, request_yaml, payload, registry=registry)
    except MapiError as err:
        print(err.code, "-", err)
        return err.code
    raise AssertionError("expected a typed error")


FORECAST = """
apiVersion: mapi/v1
kind: ScoreRequest
model: endpoint://quickstart/fruit-demand
mode: forecast
horizon: {dim: time, periods: 4}
"""

# 1. a forecast without next month's promotions: promo is future_required
code_1 = try_score(FORECAST, plan.drop_vars("promo"))

# 2. quantiles are per cell, so "total over all stores" would be a made-up number
code_2 = try_score(FORECAST + "output: {aggregate: {over: [store]}}\n", plan)

# 3. a product outside the fitted range: sku is a fixed dim
code_3 = try_score(FORECAST, plan.assign_coords(sku=["apples", "kiwis"]))

MISSING_FUTURE_COVARIATE - forecasting needs future values of ['promo'] at stage 'raw'
CAPABILITY_UNSUPPORTED - cannot aggregate 'quantiles' output: Quantiles are marginal per cell; summing quantiles is not a quantile of the sum.
UNSEEN_LEVELS - 1 level(s) of 'sku' were not seen at fit and the dim's policy is 'fixed'

So how do I get a total over stores?

Use a model that has a representation that supports totals. sklearn.tabular_bootstrap returns joint draws, and you can add them for each draw. Bayesian models return posterior draws. See Representations.

8. Save and reload the bundle

import tempfile
from pathlib import Path

from mapi_sdk import Bundle

out = Path(tempfile.mkdtemp()) / "fruit-demand-v1"
manifest = bundle.write(out)
print(sorted(p.name for p in out.iterdir()))
# ['bundle.zarr.zip', 'diagnostics.json', 'evaluation', 'manifest.json', 'metrics.json',
#  'plots.json']  - `evaluation/` is the train-only fit that produced the holdout metrics

again = Bundle.read(out, registry)  # every file's size and hashes are verified here
assert again.bundle_id == bundle.bundle_id

A bundle holds these items:

  • The spec and the binding.
  • The state and the context tail of each step.
  • The fitted groups and the native artifacts of the model (here, the xgboost model as JSON).
  • A manifest with the size, CRC32C, MD5 and SHA-256 of each file.

The bundle contains no pickled objects. See Bundles.

The whole script

docs/snippets/quickstart.py 
"""Getting started: fit and score a labelled demand model in a notebook.

Every section between ``--8<--`` markers is included verbatim in the docs, and
``conformance/tests/test_docs_snippets.py`` runs this file top to bottom.
"""

import numpy as np
import pandas as pd
import xarray as xr

rng = np.random.default_rng(7)
stores = ["north", "south", "east"]
skus = ["apples", "pears"]
weeks = pd.date_range("2025-01-05", periods=52, freq="W-SUN")
shape = (len(stores), len(skus), len(weeks))

price = 3.0 * np.exp(rng.normal(0.0, 0.1, shape))
promo = (rng.uniform(size=shape) < 0.2).astype(float)
units = rng.poisson(np.exp(2.5 - 1.2 * np.log(price / 3.0) + 0.5 * promo)).astype(float)

dims = ("store", "sku", "week")
data = xr.Dataset(
    {"units_sold": (dims, units), "price": (dims, price), "promo": (dims, promo)},
    coords={"store": stores, "sku": skus, "week": weeks},
)

from mapi_spec import DataSchema

schema = DataSchema.model_validate(
    {
        "dims": {
            # a store we have never seen may be scored: it gets the pooled prediction
            "store": {"kind": "categorical", "policy": "new_levels", "new_level_strategy": "pooled"},
            # the product range is fixed: an unknown sku is an error
            "sku": {"kind": "categorical", "policy": "fixed"},
            # time extends into the future, one Sunday-ending week at a time
            "week": {"kind": "time", "policy": "extend", "freq": "W-SUN"},
        },
        "vars": {
            "units_sold": {"dims": list(dims), "dtype": "float64", "role": "target"},
            "price": {
                "dims": list(dims),
                "dtype": "float64",
                "role": "covariate",
                "units": "EUR",
                "future_required": True,  # a forecast must say what the price will be
            },
            "promo": {
                "dims": list(dims),
                "dtype": "float64",
                "role": "covariate",
                "future_required": True,
            },
        },
    }
)

from mapi_spec import load_model_spec

spec = load_model_spec("""
apiVersion: mapi/v1
kind: ModelSpec
metadata: {name: fruit-demand, project: quickstart}
data:
  dataset: dataset://quickstart/fruit_weekly@latest
  bind:                                   # canonical name: your name
    dims: {store: store, sku: sku, time: week}
    vars: {y: units_sold, price: price, promo: promo}
  split: {kind: time_holdout, dim: time, last: 8}
pipeline:
  - {id: cal,  use: core.calendar@^1, with: {along: time, features: [week_of_year]}}
  - {id: logy, use: core.log1p@^1,    with: {vars: [y]}}
model:
  use: tabular.regressor@^1
  settings:
    estimator: {kind: xgboost.XGBRegressor, params: {n_estimators: 60, max_depth: 3}}
    rows: [store, sku, time]
    coord_features: {store: category, sku: category}
  fit: {seed: 7}
evaluate: {metrics: [rmse, mape, coverage_80]}
""")

from mapi_sdk import default_registry
from mapi_spec import validate_spec

registry = default_registry()  # every installed step, model and plot (entry points)
check = validate_spec(spec, registry.catalog(), schema)
print(check.ok)  # True: names, dims, settings, split stage and leakage all check out
print(check.resolved.model.resolved)  # tabular.regressor@1.0.0 (the semver range, pinned)
assert check.ok, check.errors

from mapi_sdk import fit_spec

bundle = fit_spec(spec, data, registry=registry, raw_schema=schema)
print(bundle.metrics.values)  # holdout metrics, in units sold: {'rmse': ..., 'mape': ..., ...}
print(bundle.metrics.n_holdout)  # 48 cells: 3 stores x 2 skus x 8 weeks
assert set(bundle.metrics.values) == {"rmse", "mape", "coverage_80"}
assert bundle.metrics.n_holdout == 48

from mapi_sdk import score_bundle

future_weeks = pd.date_range(weeks[-1] + pd.Timedelta(weeks=1), periods=4, freq="W-SUN")
plan = xr.Dataset(
    {
        "price": (dims, np.full((4, 2, 4), 2.9)),
        "promo": (dims, np.zeros((4, 2, 4))),
    },
    # "west" opens next month: the model has never seen it
    coords={"store": [*stores, "west"], "sku": skus, "week": future_weeks},
)

forecast = score_bundle(
    bundle,
    """
    apiVersion: mapi/v1
    kind: ScoreRequest
    model: endpoint://quickstart/fruit-demand
    mode: forecast
    horizon: {dim: time, periods: 4}
    output: {summary: [median, q10, q90]}
    """,
    plan,
    registry=registry,
)
summary = forecast["summary"].to_dataset()
print(summary["y_hat"].dims)  # ('stat', 'store', 'sku', 'week'), in your names
print(summary["y_hat"].sel(stat="median", store="west").values.round(1))
print(forecast.attrs["representation"])  # quantiles: what this model's uncertainty is
print(forecast.attrs["output_space"])  # {"y_hat": "raw"}: log1p undone, back in units sold
print(forecast.attrs["issues"])  # an info notice: 1 new level of 'store' uses 'pooled'
assert set(summary["store"].values) == {"north", "south", "east", "west"}
assert summary.sizes["week"] == 4

from mapi_spec import ErrorCode, MapiError


def try_score(request_yaml: str, payload: xr.Dataset) -> ErrorCode:
    try:
        score_bundle(bundle, request_yaml, payload, registry=registry)
    except MapiError as err:
        print(err.code, "-", err)
        return err.code
    raise AssertionError("expected a typed error")


FORECAST = """
apiVersion: mapi/v1
kind: ScoreRequest
model: endpoint://quickstart/fruit-demand
mode: forecast
horizon: {dim: time, periods: 4}
"""

# 1. a forecast without next month's promotions: promo is future_required
code_1 = try_score(FORECAST, plan.drop_vars("promo"))

# 2. quantiles are per cell, so "total over all stores" would be a made-up number
code_2 = try_score(FORECAST + "output: {aggregate: {over: [store]}}\n", plan)

# 3. a product outside the fitted range: sku is a fixed dim
code_3 = try_score(FORECAST, plan.assign_coords(sku=["apples", "kiwis"]))
assert code_1 is ErrorCode.MISSING_FUTURE_COVARIATE
assert code_2 is ErrorCode.CAPABILITY_UNSUPPORTED
assert code_3 is ErrorCode.UNSEEN_LEVELS

from mapi_sdk import CoordPlan
from mapi_spec import load_score_request, validate_score_request

# What the API stores next to a model version: JSON only, no arrays, no plugin code.
meta = CoordPlan(load_score_request(FORECAST), bundle.resolved, bundle.fit_coords).metadata()
print(f"{len(meta.model_dump_json()) / 1024:.0f} KiB of metadata")

request = load_score_request(FORECAST + "output: {summary: [hdi_90], draws: 100}\n")
for issue in validate_score_request(request, meta):  # no data, no model: milliseconds
    print(issue.code, issue.pointer, "-", issue.message)
# MISSING_FUTURE_COVARIATE /data - forecasting needs future values of ['price', 'promo'] ...
# CAPABILITY_UNSUPPORTED /output/draws - draws were requested but the model's
#                                        representation is 'quantiles'
# CAPABILITY_UNSUPPORTED /output/summary/0 - summary 'hdi_90' cannot be computed from
#                                            'quantiles' output
assert any(
    i.code is ErrorCode.CAPABILITY_UNSUPPORTED for i in validate_score_request(request, meta)
)

import tempfile
from pathlib import Path

from mapi_sdk import Bundle

out = Path(tempfile.mkdtemp()) / "fruit-demand-v1"
manifest = bundle.write(out)
print(sorted(p.name for p in out.iterdir()))
# ['bundle.zarr.zip', 'diagnostics.json', 'evaluation', 'manifest.json', 'metrics.json',
#  'plots.json']  - `evaluation/` is the train-only fit that produced the holdout metrics

again = Bundle.read(out, registry)  # every file's size and hashes are verified here
assert again.bundle_id == bundle.bundle_id

print(manifest.role, manifest.plugin["id"], manifest.spec_hash[:19])
# production tabular.regressor sha256:...
print(manifest.files["bundle.zarr.zip"])  # size, crc32c, md5 (GCS forms) and sha256
print(manifest.lineage["evaluation_bundle"])  # the train-only fit behind the metrics

tree = again.to_tree()  # the whole bundle as one xr.DataTree
print(sorted(tree.children))
# ['binding', 'fit_coords', 'model', 'native', 'preprocess', 'schema', 'spec']
print(sorted(tree["preprocess"].children))  # ['cal', 'logy']: one node per step
assert {"model", "preprocess", "native", "spec"} <= set(tree.children)

Next steps