08 · A workflow in plain Python: build it, use it, share it

You will learn

• Write a workflow as four plain functions
• Save fitted state as JSON plus .npy arrays
• Get the same labels as the scikit-learn version

You need

• Python 3.12 and uv
• Section 3 is recorded; rerunning it needs the reference deployment

Run this notebook Offline, section 3 recorded

notebooks/08_workflow_plain_python.ipynb

Runs offline. Section 3 shows a recorded session against the reference deployment and cannot be run without it.

About 20 minutes

Access needed. The packages and notebooks are not yet publicly distributed. The steps below need a checkout of the UMP repository. Pages show recorded outputs, so you can follow along without running them.

cd notebooks && uv run jupyter lab

Runs offline; section 3 shows a recorded session against the reference deployment. Uses babelfin and NumPy, and no modelling library at all. The data is SYNTHETIC, generated with the same known truth as notebook 06, so every answer can be checked.

Priya’s media mix model for Northwind started life as forty lines of NumPy in a scratch notebook. It worked, and nobody else could run it: the adstock lived in one cell, the bootstrap in another, and “the TV ROI” was whatever the last cell printed. This notebook turns those same lines into a workflow that she can save, rerun on next quarter’s weeks, hand to a colleague and serve as an app.

The point to take away: a workflow is just functions. Four of them (fit, to_state, from_state and one per question) in an ordinary module. Notebook 07 does the same with scikit-learn and notebook 09 with JAX.

In [1]:

import os
import shutil
import socket
import subprocess
import sys
import tempfile
import time
from pathlib import Path

import babelfin as fin
import httpx
import numpy as np
import pandas as pd
from finprint import trust
from ump_sdk.ports import base_url

OUT = Path("out/workflows")
OUT.mkdir(parents=True, exist_ok=True)
CHANNELS = ["tv", "search", "social"]
SEED = 11

1. Build

A synthetic world with a known truth

The truth of notebook 06 (carryover, diminishing returns, a yearly cycle and promotions), run for 180 weeks: Priya fits on the first 156 and keeps the rest back as “next quarter”, whose spend plan stays inside the levels she has already seen. The feed is clean here; notebook 06 shows how to record the clean-up of a messy one.

In [2]:

TRUE = {"decay": {"tv": 0.5, "search": 0.0, "social": 0.3}, "slope": {"tv": 1.8, "search": 3.0, "social": 2.4}}


def adstock(x, decay):
    out = np.zeros(len(x))
    for t, v in enumerate(x):
        out[t] = v + (decay * out[t - 1] if t else 0.0)
    return out


def make_world(seed=SEED, weeks=180):
    rng = np.random.default_rng(seed)
    spend = pd.DataFrame({"tv": rng.gamma(4, 9000, weeks), "search": rng.gamma(5, 2500, weeks),
                          "social": rng.gamma(3, 4000, weeks)}).round(2)
    spend.iloc[156:] = spend.iloc[156:].clip(spend.iloc[:156].min(), spend.iloc[:156].max(), axis=1)   # plan in range
    promo, t = (np.arange(weeks) % 9 == 3).astype(int), np.arange(weeks)
    ads = {c: adstock(spend[c].to_numpy(), TRUE["decay"][c]) for c in CHANNELS}
    contrib = pd.DataFrame({c: TRUE["slope"][c] * ads[c].mean() * np.log1p(ads[c] / ads[c].mean()) for c in CHANNELS})
    sales = 200_000 + 30_000 * np.sin(2 * np.pi * t / 52) + 15_000 * promo + contrib.sum(axis=1).to_numpy()
    week = pd.date_range("2023-01-02", periods=weeks, freq="W-MON")
    df = pd.DataFrame({"week": week, "sales": (sales + rng.normal(0, 8_000, weeks)).round(2), **spend, "promo": promo})
    return df, contrib


world, contrib = make_world()
train, later = world.iloc[:156].reset_index(drop=True), world.iloc[156:].reset_index(drop=True)
TRUTH_ROI = {c: contrib[c].iloc[:156].sum() / train[c].sum() for c in CHANNELS}   # average ROI over the 156 weeks
train.tail(3)

Out [2]:

	week	sales	tv	search	social	promo
153	2025-12-08	340485.22	40334.67	18926.91	4013.52	0
154	2025-12-15	303867.70	15504.88	8004.43	22994.64	0
155	2025-12-22	346880.23	27929.58	16846.43	19090.68	0

The input contract

What “the same necessary shape” means for this model, read off the training frame: these columns, these types, one row per Monday week. More weeks are fine; a missing column is not.

In [3]:

contract = fin.Contract.from_frame(train, time="week", target="sales", freq="W-MON")
contract

Out [3]:

Contract: 4 required, 0 optional columns, time week, target sales

ump.input_contract/0.1

details

key	week
extra columns	drop
categories	none

column	dtype	required	missing
tv	float64	True	forbid
search	float64	True	forbid
social	float64	True	forbid
promo	int64	True	forbid

The model, as a module

The code goes in a file, because a bundle saves source text (never pickled objects), and only a file has source to save. Everything is plain NumPy: adstock and saturation are functions, the fit is numpy.linalg.lstsq, and a block bootstrap gives joint draws. Each draw picks its own decay and saturation from a small grid by error on the weeks it left out, so the intervals carry that tuning uncertainty too.

In [4]:

%%writefile out/workflows/plain_mmm.py
"""Northwind weekly sales: adstock, saturation and least squares in plain NumPy, with block-bootstrap draws."""
import numpy as np
import pandas as pd

CHANNELS = ["tv", "search", "social"]
GRID = [(tv, so, k) for tv in (0.3, 0.5, 0.7) for so in (0.0, 0.3, 0.6) for k in (0.5, 1.0, 2.0)]  # tv, social decay; saturation
HISTORY = 26   # weeks of spend kept so new weeks start with their carryover (0.7 ** 26 is about 0.0001)


def adstock(x, decay):
    out = np.zeros(len(x))
    for t, v in enumerate(x):
        out[t] = v + (decay * out[t - 1] if t else 0.0)
    return out


def days(week):
    return pd.to_datetime(week).to_numpy().astype("datetime64[D]").astype(np.int64)


def design(model, df, g):
    """Features for grid point g, with the spend of the weeks before df prepended for carryover."""
    earlier = model["hist_days"] < days(df["week"]).min()
    spend, n0 = np.vstack([model["hist_spend"][earlier], df[CHANNELS].to_numpy(float)]), int(earlier.sum())
    tv, so, _ = GRID[g]
    media = [model["k"][g, j] * np.log1p(adstock(spend[:, j], d) / model["k"][g, j])[n0:]
             for j, d in enumerate((tv, 0.0, so))]
    t = (days(df["week"]) - model["origin"]) / 7.0
    return np.column_stack([np.ones(len(df)), *media, np.sin(2 * np.pi * t / 52), np.cos(2 * np.pi * t / 52),
                            t / 52, df["promo"].to_numpy(float)])


def fit(df, *, seed, params):
    model = {"origin": int(days(df["week"]).min()), "hist_days": np.empty(0, np.int64), "hist_spend": np.empty((0, 3))}
    model["k"] = np.array([[k * adstock(df[c].to_numpy(float), d).mean() for c, d in zip(CHANNELS, (tv, 0.0, so))]
                           for tv, so, k in GRID])
    F, y = [design(model, df, g) for g in range(len(GRID))], df["sales"].to_numpy(float)
    n, block, rng = len(y), int(params["block"]), np.random.default_rng(seed)
    picks, betas = [], []
    for _ in range(int(params["draws"])):
        rows = np.concatenate([np.arange(s, s + block) for s in rng.integers(0, n - block + 1, n // block + 1)])[:n]
        oob = np.setdiff1d(np.arange(n), rows)
        oob = oob if oob.size else rows
        fits = [np.linalg.lstsq(X[rows], y[rows], rcond=None)[0] for X in F]
        g = int(np.argmin([np.abs(X[oob] @ b - y[oob]).mean() for X, b in zip(F, fits)]))
        picks.append(g)
        betas.append(fits[g])
    model.update(grid=np.array(picks), beta=np.array(betas), hist_days=days(df["week"])[-HISTORY:],
                 hist_spend=df[CHANNELS].to_numpy(float)[-HISTORY:])
    return model


def to_state(model):
    return {"origin": model["origin"]}, {k: v for k, v in model.items() if k != "origin"}


def from_state(meta, arrays):
    return {"origin": int(meta["origin"]), **{k: np.asarray(v) for k, v in arrays.items()}}


def sales(model, df, **spend):
    """Weekly sales per draw (draws x weeks), with some channels' spend in df's weeks replaced."""
    df = df.assign(**spend)
    X = {g: design(model, df, g) for g in set(model["grid"].tolist())}
    return np.stack([X[g] @ b for g, b in zip(model["grid"].tolist(), model["beta"])])


def roi(model, df, ch):
    return (sales(model, df) - sales(model, df, **{ch: 0.0})).sum(axis=1) / df[ch].sum()


def roi_tv(model, df):
    return roi(model, df, "tv")


def roi_search(model, df):
    return roi(model, df, "search")


def roi_social(model, df):
    return roi(model, df, "social")


def mroi_tv(model, df):
    return (sales(model, df, tv=df["tv"] * 1.1) - sales(model, df)).sum(axis=1) / (0.1 * df["tv"].sum())


def cut_tv_20pct(model, df):
    return (sales(model, df, tv=df["tv"] * 0.8) - sales(model, df)).sum(axis=1)

Out [4]:

Overwriting out/workflows/plain_mmm.py

Named questions

Each question names a function in the module and says, with its estimand, exactly what the number means. The horizon follows the rows each run receives, so “ROI over these weeks” stays true on next quarter’s data. The ROI and mROI questions also name their Tier D summary number: the labelled fallback when the code cannot run.

In [5]:

COMMON = {"kpi": "incremental_sales", "cost_basis": "gross", "point_rule": "posterior_median", "interval": "bootstrap_pct"}
wf = fin.Workflow("Northwind weekly sales, plain NumPy (synthetic)", contract=contract, code=OUT / "plain_mmm.py",
                  seed=SEED, params={"draws": 300, "block": 8}, dependence="joint_draws")
for ch in CHANNELS:
    wf.question(f"roi_{ch}", ask=f"What did each dollar of {ch} return over these weeks?",
                estimand=fin.estimand("roi", channels=ch, **COMMON), summary=("roi", ch))
wf.question("mroi_tv", ask="What would the next dollar of TV return?",
            estimand=fin.estimand("mroi", channels="tv", **COMMON), summary=("mroi", "tv"))
wf.question("cut_tv_20pct", ask="What happens to sales if we cut TV by 20%?",
            estimand=fin.estimand("roi", channels="tv", intervention="scale", size=-0.2, baseline="current_plan",
                                  per="absolute", **COMMON))
wf

Out [5]:

Northwind weekly sales, plain NumPy (synthetic)

workflow, not fitted · 5 questions

details

module	plain_mmm.py
seed	11
prep recipe	none
contract	<Contract: 4 required, 0 optional columns, time week, target sales>

question	ask
roi_tv	What did each dollar of tv return over these weeks?
roi_search	What did each dollar of search return over these weeks?
roi_social	What did each dollar of social return over these weeks?
mroi_tv	What would the next dollar of TV return?
cut_tv_20pct	What happens to sales if we cut TV by 20%?

Fit, and check against the truth

Every answer is a fin.Number: the median of 300 bootstrap draws with its 90% interval and its estimand.

In [6]:

fitted = wf.fit(train)
fitted

Out [6]:

Northwind weekly sales, plain NumPy (synthetic)

fitted on 156 rows · 5 answers

details

module	plain_mmm.py
code sha256	6344172b8335

question	value	lo	hi	level
roi_tv	3.00	2.29	3.24	0.9
roi_search	1.91	1.51	2.35	0.9
roi_social	2.52	1.98	2.93	0.9
mroi_tv	1.82	1.69	1.92	0.9
cut_tv_20pct	-2,295,419	-2,419,080	-2,137,262	0.9

In [7]:

answers = fitted.answers()
check = pd.DataFrame([{"channel": c, "ROI estimate": answers[f"roi_{c}"].value, "lo": answers[f"roi_{c}"].lo,
                       "hi": answers[f"roi_{c}"].hi, "true ROI": TRUTH_ROI[c]} for c in CHANNELS])
check["covered"] = (check.lo <= check["true ROI"]) & (check["true ROI"] <= check.hi)
check.round(2)   # average ROI, incremental sales per gross dollar, 156 weeks, 90% block-bootstrap interval

Out [7]:

	channel	ROI estimate	lo	hi	true ROI	covered
0	tv	3.00	2.29	3.24	2.45	True
1	search	1.91	1.51	2.35	2.02	True
2	social	2.52	1.98	2.93	2.27	True

All three intervals cover the true average ROI. TV’s point estimate sits well above the truth, near the top of its interval: with 156 weeks, carryover and saturation are hard to tell apart, and the interval is what carries that.

2. Save and use

Saving writes a folder: the module’s source, the fitted state as JSON plus .npy arrays, the contract, the questions and a Tier D summary, every file hashed. Nothing is pickled.

In [8]:

BUNDLE = OUT / "plain_mmm.bundle"
shutil.rmtree(BUNDLE, ignore_errors=True)
print("saved", fitted.save(BUNDLE).name)
print(sorted(p.name for p in BUNDLE.rglob("*") if p.is_file()))

Out [8]:

saved plain_mmm.bundle
['README.md', 'SHA256SUMS', 'beta.npy', 'contract.json', 'grid.npy', 'hist_days.npy', 'hist_spend.npy', 'k.npy', 'plain_mmm.py', 'questions.json', 'state.json', 'summary.json', 'workflow.json']

A new machine

Next quarter Priya opens the bundle on a fresh laptop. Loading checks every hash and runs nothing; she can read the code first. (The first two lines stand in for a new machine: an empty trust store and no trust left over from saving.)

In [9]:

os.environ["FINPRINT_HOME"] = tempfile.mkdtemp(prefix="priya_laptop_")
trust._SESSION.clear()
lw = fin.Workflow.load(BUNDLE)
print("trusted:", lw.trusted)
print(lw.code[:400])

Out [9]:

trusted: False
"""Northwind weekly sales: adstock, saturation and least squares in plain NumPy, with block-bootstrap draws."""
import numpy as np
import pandas as pd

CHANNELS = ["tv", "search", "social"]
GRID = [(tv, so, k) for tv in (0.3, 0.5, 0.7) for so in (0.0, 0.3, 0.6) for k in (0.5, 1.0, 2.0)]  # tv, social decay; saturation
HISTORY = 26   # weeks of spend kept so new weeks start with their carryover (0.

Ask before trusting, and the answers come from the bundle’s summary, labelled as such. cut_tv_20pct has no summary number, so it is not available.

In [10]:

new = later.iloc[:12]   # the 12 weeks right after training
print(lw.score(new).explain())

Out [10]:

Northwind weekly sales, plain NumPy (synthetic): 4 answers from summary on 12 rows, route not_available (WORKFLOW_CODE_UNTRUSTED, SUMMARY_FALLBACK)
- roi_tv: 3.00, likely between 2.29 and 3.24
- roi_search: 1.91, likely between 1.51 and 2.35
- roi_social: 2.52, likely between 1.98 and 2.93
- mroi_tv: 1.82, likely between 1.69 and 1.92
warn WORKFLOW_CODE_UNTRUSTED: code code/plain_mmm.py is not trusted on this machine, so it was not run
warn SUMMARY_FALLBACK: answers are the bundle's fit-time Tier D summary because the code is not trusted; no summary number for cut_tv_20pct (not available)

She has read the code. Trust it, and score the 12 new weeks with the saved fit (score by default: nothing refits).

In [11]:

run = lw.trust().score(new)
print(run.explain())

Out [11]:

Northwind weekly sales, plain NumPy (synthetic): 5 answers from score on 12 rows, route score
- roi_tv: 2.76, likely between 2.10 and 2.98
- roi_search: 1.81, likely between 1.46 and 2.23
- roi_social: 2.63, likely between 2.04 and 3.03
- mroi_tv: 1.76, likely between 1.64 and 1.86
- cut_tv_20pct: -151,396, likely between -160K and -140K

When the new data does not line up

The next file skips two weeks. Carryover cannot be scored across a hole, so the run refits on the data it was given with the saved seed, and says so on every number.

In [12]:

gap = later.iloc[2:24]
gapped = lw.score(gap)
print(gapped.route, [n["code"] for n in gapped.notes])
print(gapped.answer("roi_tv").explain())

Out [12]:

refit ['INPUT_HISTORY_GAP', 'HISTORY_ROUTED_TO_REFIT']
2.37, likely between 1.57 and 4.26
means: incremental sales per dollar, gross from removing all spend, geo_set, versus zero spend, over 22 weeks, posterior median, 90% bootstrap_pct
confidence: low; trust level 2
fallback G-DATA-07 INPUT_HISTORY_GAP: new input starts 2026-01-12, after 2025-12-29 (one step past the history); the rerun refits; answer refitted on this input with the saved seed and params, not the saved state
fallback G-DATA-07 HISTORY_ROUTED_TO_REFIT: the input does not follow on from the saved history, so the saved state is not scored; the rerun refits; answer refitted on this input with the saved seed and params, not the saved state

An extra column is dropped with a note, and the score goes ahead.

In [13]:

extra = lw.score(new.assign(weather_index=1.0))
print(extra.route, [(n["code"], n["severity"]) for n in extra.notes])

Out [13]:

score [('INPUT_EXTRA_DROPPED', 'info')]

A missing required column cannot be scored or refit, so the run falls back to the summary, labelled.

In [14]:

missing = lw.score(new.drop(columns="promo"))
print(missing.route, missing.source, [n["code"] for n in missing.notes])
print(missing.answer("roi_tv").explain())

Out [14]:

not_available summary ['INPUT_REQUIRED_MISSING', 'SUMMARY_FALLBACK']
3.00, likely between 2.29 and 3.24
means: incremental sales per dollar, gross from removing all spend, geo_set, versus zero spend, over 156 weeks, posterior median, 90% bootstrap_pct
confidence: low; trust level 1
fallback G-DATA-07 INPUT_REQUIRED_MISSING: required column promo is missing; answer is the bundle's fit-time Tier D summary, not a run on this input
fallback G-DATA-07 SUMMARY_FALLBACK: answers are the bundle's fit-time Tier D summary because the input does not meet the contract; no summary number for cut_tv_20pct (not available); answer is the bundle's fit-time Tier D summary, not a run on this input

And inspect() describes the bundle without running it. The environment pin names only Python, NumPy and pandas (plus UMP’s own packages): no modelling library to install on the next machine.

In [15]:

info = lw.inspect()
{k: info[k] for k in ("entry", "code_sha256", "training", "env_pin")}

Out [15]:

{'entry': 'code/plain_mmm.py',
 'code_sha256': '6344172b8335c107e49e2a1440528a773b66a4b050ddc8f9a81c8b45c73f78fe',
 'training': {'rows': 156,
  'time_min': '2023-01-02',
  'time_max': '2025-12-22',
  'history_end': '2025-12-22',
  'channel_ranges': {'tv': [8164.73, 84982.69],
   'search': [3101.53, 30427.92],
   'social': [1846.89, 38013.34],
   'promo': [0.0, 1.0]}},
 'env_pin': {'python': '3.12.8',
  'numpy': '2.5.4',
  'pandas': '3.1.0',
  'finprint': '0.2.0',
  'ump_spec': '0.1.0'}}

3. Share and serve (recorded session)

Priya (POC login maya.chen) publishes the bundle to the labelled test client. Sam approves it; then Li, on her own machine with an empty trust store, runs it with no manual trust, because Sam’s approval vouches for the code’s hash. Finally the bundle runs as an app.

Recorded run

This section shows a recorded session against the reference deployment. It cannot be run without that deployment; the offline sections above run anywhere. The identity stub (as_=) and the localhost address belong to the reference deployment, not to the standard; a conforming core authenticates callers as section 19 requires.

In [16]:

CLIENT = "northwind-e2e-live"
try:
    CORE_UP = httpx.get(base_url("core") + "/healthz", timeout=2).status_code == 200
except httpx.HTTPError:
    CORE_UP = False
print("core is up" if CORE_UP else "Section 3 needs the POC (scripts/dev.sh); skipped.")

Out [16]:

core is up

Publish files a draft Tier A model manifest that pins every file by hash. Priya submits it, and cannot approve her own work (G-INT-02).

In [17]:

if CORE_UP:
    with fin.connect(as_="maya.chen") as s:
        art = s.workflows.publish(fitted, client=CLIENT, title="Notebook 08: plain NumPy workflow (synthetic)")
        art.submit(reason="notebook 08")
        try:
            art.approve(accept=art.fallbacks)
        except fin.Forbidden as e:
            print("refused:", e)
        print(art.id, art.doc["tier"], art.doc["engine"])

Out [17]:

refused: SELF_APPROVAL_FORBIDDEN: an approver cannot approve their own work (owner or submitter)
mdl_247cfef766c14c9982ec A ump-workflow

Sam reads the code and approves, accepting the fallbacks by name.

In [18]:

if CORE_UP:
    with fin.connect(as_="sam.patel") as sam:
        mine = sam.artifact(art.id)
        print("accepting:", [f.code for f in mine.fallbacks])
        mine.approve(accept=mine.fallbacks, comment="read plain_mmm.py; synthetic notebook run")
        print("state:", sam.artifact(art.id).state)

Out [18]:

accepting: ['UNVALIDATED_METHOD']
state: validated

Li gets the bundle on her own machine. Every blob is checked, and she scores with no trust step of her own.

In [19]:

if CORE_UP:
    os.environ["FINPRINT_HOME"] = tempfile.mkdtemp(prefix="li_laptop_")
    trust._SESSION.clear()
    with fin.connect(as_="li.wong") as li:
        li_wf = li.workflows.get(art.id, dest=Path(tempfile.mkdtemp()) / "li_copy")
    print("trusted:", li_wf.trusted, "| untrusted reason:", li_wf.approval.reason or "none")
    li_run = li_wf.score(new)
    print(li_run.route, li_run.source)
    print(li_run.answer("roi_tv").explain())

Out [19]:

trusted: True | untrusted reason: none
score score
2.76, likely between 2.10 and 2.98
means: incremental sales per dollar, gross from removing all spend, geo_set, versus zero spend, over 12 weeks, posterior median, 90% bootstrap_pct
confidence: medium; trust level 2

Serve it as an app

ump-workflow serve puts the approved bundle behind UMP’s adapter contract on a free port, with an empty trust store of its own: it runs the code only because the approval vouches for it. Li asks it questions over HTTP.

In [20]:

if CORE_UP:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    server = subprocess.Popen(
        [sys.executable, "-m", "ump_workflow", "serve", "--artifact", art.id, "--principal", "li.wong",
         "--port", str(port), "--dest", str(Path(tempfile.mkdtemp()) / "served")],
        env={**os.environ, "FINPRINT_HOME": tempfile.mkdtemp()}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    url = f"http://127.0.0.1:{port}"
    for _ in range(200):
        try:
            if httpx.get(url + "/healthz", timeout=1).status_code == 200:
                break
        except httpx.HTTPError:
            time.sleep(0.3)
    with fin.connect(as_="li.wong") as li:
        served = li.workflows.answer(url, new)
    print(served.route, served.source, served.rows)
    print(served.answer("mroi_tv").explain())

Out [20]:

score score 12
1.76, likely between 1.64 and 1.86
means: incremental sales per dollar, gross from a small increase in spend (+10%), geo_set, versus the current plan, over 12 weeks, posterior median, 90% bootstrap_pct
confidence: medium; trust level 2

In [21]:

if CORE_UP:
    server.terminate()
    server.wait(timeout=15)
    with fin.connect(as_="maya.chen") as s:   # a validated Tier A model without curves should not linger
        s.core.transition(art.id, "deprecated", reason="notebook 08 cleanup (synthetic shared workflow)")
        print("server stopped; manifest", s.artifact(art.id).state)

Out [21]:

server stopped; manifest deprecated

What you have now

• out/workflows/plain_mmm.py: four kinds of plain function (fit, to_state, from_state, one per question), and no modelling library.
• out/workflows/plain_mmm.bundle: a hashed folder that reruns on new weeks by scoring, refits with a label when the weeks do not line up, and falls back to its labelled summary when it cannot run.
• With the POC: the same bundle published, approved by a second person, run by a colleague and served as an app.