Build features as they were known at each decision
Reconstruct the information a planner could use when deciding how much stock to order. An order can arrive late, and a later revision can correct its quantity, product or event time. Product attributes also have effective intervals and availability times. A join on event time alone can include information that the planner did not yet have.
Run it
Run all commands below from the repository root (sail-examples/):
uv run pipelines/build-point-in-time-features/main.py
The example prints four tables: features, naive_features, outcomes and
rejected. It uses generated records, requires no credentials and trains no
model.
The early tea decision has 12 known units across three orders. Ignoring availability produces 25 units: it includes a late arrival and a later quantity correction. At the later decision, a new effective product interval supplies a different lead time. The coffee decision has zero observed historical orders and an unavailable future outcome. Missing and overlapping product intervals produce rejected requests rather than arbitrary attribute choices.
Output
The newest capture appears first. Older output stays here when a result changes.
Current output · Sail 0.7.3
Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.
features
+----------+----------+-------------------+--------+---------+-----------+-----+-------------------+
|request_id|product_id|decision_at |category|lead_days|order_count|units|last_order_at |
+----------+----------+-------------------+--------+---------+-----------+-----+-------------------+
|early |tea |2026-01-05 00:00:00|pantry |3 |3 |12 |2026-01-04 00:00:00|
|later |tea |2026-01-08 00:00:00|pantry |2 |2 |16 |2026-01-07 00:00:00|
|zero |coffee |2026-01-05 00:00:00|pantry |5 |0 |0 |NULL |
+----------+----------+-------------------+--------+---------+-----------+-----+-------------------+
naive_features
+----------+----------+-------------------+--------+---------+-----------+-----+-------------------+
|request_id|product_id|decision_at |category|lead_days|order_count|units|last_order_at |
+----------+----------+-------------------+--------+---------+-----------+-----+-------------------+
|early |tea |2026-01-05 00:00:00|pantry |3 |4 |25 |2026-01-04 00:00:00|
|later |tea |2026-01-08 00:00:00|pantry |2 |3 |18 |2026-01-07 00:00:00|
|zero |coffee |2026-01-05 00:00:00|pantry |5 |0 |0 |NULL |
+----------+----------+-------------------+--------+---------+-----------+-----+-------------------+
outcomes
+----------+-------------------+------------+--------------+
|request_id|horizon_end |target_units|outcome_status|
+----------+-------------------+------------+--------------+
|early |2026-01-07 00:00:00|7 |observed |
|later |2026-01-10 00:00:00|NULL |unavailable |
|zero |2026-01-07 00:00:00|NULL |unavailable |
+----------+-------------------+------------+--------------+
rejected
+----------+----------+-------------------+-------------------+-------------------+---------------------------+
|request_id|product_id|history_start |decision_at |horizon_end |reason |
+----------+----------+-------------------+-------------------+-------------------+---------------------------+
|missing |cocoa |2026-01-02 00:00:00|2026-01-05 00:00:00|2026-01-07 00:00:00|missing_product_interval |
|overlap |juice |2026-01-02 00:00:00|2026-01-05 00:00:00|2026-01-07 00:00:00|ambiguous_product_intervals|
+----------+----------+-------------------+-------------------+-------------------+---------------------------+
Python code
main.py497 lines
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "pysail==0.7.3",
# "pyspark-client==4.1.3",
# "pyarrow==26.0.0",
# "pandas==3.0.6",
# ]
# ///
"""Reconstruct demand-planning features using only information available at each decision."""
import argparse
import datetime as dt
import json
from pathlib import Path
REQUEST_SCHEMA = (
"request_id string, product_id string, history_start timestamp_ntz, "
"decision_at timestamp_ntz, horizon_end timestamp_ntz"
)
ORDER_SCHEMA = (
"order_id string, revision bigint, product_id string, event_at timestamp_ntz, "
"available_at timestamp_ntz, units bigint"
)
PRODUCT_SCHEMA = (
"interval_id string, revision bigint, product_id string, valid_from timestamp_ntz, "
"valid_to timestamp_ntz, available_at timestamp_ntz, category string, lead_days bigint"
)
COVERAGE_SCHEMA = "product_id string, observed_through timestamp_ntz, available_at timestamp_ntz"
FIELDS = {
"requests": ("request_id", "product_id", "history_start", "decision_at", "horizon_end"),
"orders": ("order_id", "revision", "product_id", "event_at", "available_at", "units"),
"products": (
"interval_id",
"revision",
"product_id",
"valid_from",
"valid_to",
"available_at",
"category",
"lead_days",
),
"coverage": ("product_id", "observed_through", "available_at"),
}
TIME_FIELDS = {
"history_start",
"decision_at",
"horizon_end",
"event_at",
"available_at",
"valid_from",
"valid_to",
"observed_through",
}
def utc_time(value):
"""Use UTC wall-clock values with TIMESTAMP_NTZ, independent of session time zone."""
if isinstance(value, str):
value = dt.datetime.fromisoformat(value)
if value.tzinfo is None:
raise ValueError("JSON timestamps require a UTC offset or Z")
if not isinstance(value, dt.datetime):
raise ValueError("timestamp required") # noqa: TRY004 - all input-contract failures are ValueError
if value.tzinfo is not None:
if value.utcoffset() != dt.timedelta(0):
raise ValueError("timestamps must use UTC")
value = value.replace(tzinfo=None)
return value
def normalize_inputs(requests, orders, products, coverage, evaluation_at):
"""Validate a bounded input bundle before constructing any query."""
output = {}
for name, rows in zip(FIELDS, (requests, orders, products, coverage)):
if not isinstance(rows, list) or len(rows) > (128 if name == "requests" else 5000):
raise ValueError(f"{name}: expected a bounded list")
normalized, seen, versions = [], set(), {}
for index, source in enumerate(rows):
try:
if not isinstance(source, dict) or set(source) != set(FIELDS[name]):
raise ValueError("unexpected or missing fields")
row = dict(source)
for field, value in row.items():
if field in TIME_FIELDS:
row[field] = utc_time(value)
elif field in {"revision", "units", "lead_days"}:
lower = 1 if field == "revision" else 0
upper = 365 if field == "lead_days" else 1_000_000_000
if type(value) is not int or not lower <= value <= upper:
raise ValueError(f"invalid {field}")
elif not isinstance(value, str) or not value.strip() or len(value) > 200:
raise ValueError(f"invalid {field}")
if name == "requests":
if not row["history_start"] < row["decision_at"] < row["horizon_end"]:
raise ValueError("invalid request interval")
key = row["request_id"]
if key in seen:
raise ValueError("duplicate request_id")
else:
key = tuple(row[field] for field in FIELDS[name])
if name == "orders" and row["available_at"] < row["event_at"]:
raise ValueError("order available before event")
if name == "products" and row["valid_from"] >= row["valid_to"]:
raise ValueError("invalid product interval")
if name == "coverage" and row["observed_through"] > row["available_at"]:
raise ValueError("coverage available before observed_through")
if name in {"orders", "products"}:
identity = (row["order_id" if name == "orders" else "interval_id"], row["revision"])
if identity in versions and versions[identity] != row:
raise ValueError("conflicting source revision")
versions[identity] = row
if key in seen:
continue
seen.add(key)
normalized.append(row)
except (ValueError, TypeError, OverflowError) as error:
raise ValueError(f"{name}[{index}]: {error}; input={source!r}") from error
output[name] = normalized
output["evaluation_at"] = utc_time(evaluation_at)
return output
def build_features(spark, requests, orders, products, coverage, evaluation_at):
"""Return deferred DataFrames bound to these inputs, with no shared temporary views."""
data = normalize_inputs(requests, orders, products, coverage, evaluation_at)
from pyspark.sql import Window
from pyspark.sql import functions as F
requests = spark.createDataFrame(data["requests"], REQUEST_SCHEMA)
orders = spark.createDataFrame(data["orders"], ORDER_SCHEMA)
products = spark.createDataFrame(data["products"], PRODUCT_SCHEMA)
coverage = spark.createDataFrame(data["coverage"], COVERAGE_SCHEMA)
# A Python datetime literal is TIMESTAMP, so compare UTC wall-clock NTZ
# values using an explicitly typed literal instead of a session-zone cast.
evaluation_at = F.lit(data["evaluation_at"].isoformat()).cast("timestamp_ntz")
def latest(source, name, key, cutoff, ignore_availability=False):
r, s = requests.alias("r"), source.alias("s")
eligible = r.join(s, F.lit(True) if ignore_availability else F.col("s.available_at") <= cutoff)
selected = eligible.select("r.request_id", *[F.col(f"s.{field}") for field in FIELDS[name]])
window = Window.partitionBy("request_id", key).orderBy(F.col("revision").desc())
return (
selected.withColumn("position", F.row_number().over(window))
.filter("position = 1")
.drop("position")
)
# Rank all eligible revisions first. Filtering an older revision into a window
# before ranking could resurrect a corrected or cancelled order.
product_versions = latest(products, "products", "interval_id", F.col("r.decision_at"))
candidates = requests.alias("r").join(
product_versions.alias("p"),
(F.col("r.request_id") == F.col("p.request_id"))
& (F.col("r.product_id") == F.col("p.product_id"))
& (F.col("p.valid_from") <= F.col("r.decision_at"))
& (F.col("r.decision_at") < F.col("p.valid_to")),
)
dimensions = candidates.groupBy(F.col("r.request_id").alias("request_id")).agg(
F.count("p.interval_id").alias("interval_count"),
F.min("p.category").alias("category"),
F.min("p.lead_days").alias("lead_days"),
)
classified = requests.join(dimensions, "request_id", "left").fillna({"interval_count": 0})
accepted = classified.filter("interval_count = 1").drop("interval_count")
rejected = (
classified.filter("interval_count != 1")
.select(
*FIELDS["requests"],
F.when(F.col("interval_count") == 0, "missing_product_interval")
.otherwise("ambiguous_product_intervals")
.alias("reason"),
)
.orderBy("request_id")
)
def totals(selected, future=False):
joined = requests.alias("r").join(
selected.alias("o"),
(F.col("r.request_id") == F.col("o.request_id"))
& (F.col("r.product_id") == F.col("o.product_id")),
)
lower = F.col("r.decision_at") if future else F.col("r.history_start")
upper = F.col("r.horizon_end") if future else F.col("r.decision_at")
return (
joined.filter(
(F.col("o.event_at") >= lower) & (F.col("o.event_at") < upper) & (F.col("o.units") > 0)
)
.groupBy(F.col("r.request_id").alias("request_id"))
.agg(
F.count("o.order_id").alias("order_count"),
F.sum("o.units").alias("units"),
F.max("o.event_at").alias("last_order_at"),
)
)
def feature_table(selected):
return (
accepted.join(totals(selected), "request_id", "left")
.fillna({"order_count": 0, "units": 0})
.select(
"request_id",
"product_id",
"decision_at",
"category",
"lead_days",
"order_count",
"units",
"last_order_at",
)
.orderBy("request_id")
)
known = latest(orders, "orders", "order_id", F.col("r.decision_at"))
naive = latest(orders, "orders", "order_id", evaluation_at, ignore_availability=True)
# Outcomes are evaluated at a separate, explicit cutoff and never joined into features.
evaluated = latest(orders, "orders", "order_id", evaluation_at)
watermarks = (
coverage.filter(F.col("available_at") <= evaluation_at)
.groupBy("product_id")
.agg(F.max("observed_through").alias("observed_through"))
)
outcomes = (
accepted.join(watermarks, "product_id", "left")
.join(
totals(evaluated, future=True).select("request_id", F.col("units").alias("target_units")),
"request_id",
"left",
)
.withColumn(
"complete",
(F.col("horizon_end") <= evaluation_at) & (F.col("observed_through") >= F.col("horizon_end")),
)
.select(
"request_id",
"horizon_end",
F.when(F.col("complete"), F.coalesce("target_units", F.lit(0)))
.cast("long")
.alias("target_units"),
F.when(F.col("complete"), "observed").otherwise("unavailable").alias("outcome_status"),
)
.orderBy("request_id")
)
return {
"features": feature_table(known),
"naive_features": feature_table(naive),
"outcomes": outcomes,
"rejected": rejected,
}
def demo_inputs():
def time(day):
return dt.datetime(2026, 1, day)
requests = [
{
"request_id": "early",
"product_id": "tea",
"history_start": time(2),
"decision_at": time(5),
"horizon_end": time(7),
},
{
"request_id": "later",
"product_id": "tea",
"history_start": time(5),
"decision_at": time(8),
"horizon_end": time(10),
},
{
"request_id": "zero",
"product_id": "coffee",
"history_start": time(2),
"decision_at": time(5),
"horizon_end": time(7),
},
{
"request_id": "missing",
"product_id": "cocoa",
"history_start": time(2),
"decision_at": time(5),
"horizon_end": time(7),
},
{
"request_id": "overlap",
"product_id": "juice",
"history_start": time(2),
"decision_at": time(5),
"horizon_end": time(7),
},
]
orders = []
for order_id, revision, event, available, units in [
("left", 1, 2, 2, 2),
("old", 1, 1, 1, 100),
("late-arrival", 1, 3, 6, 10),
("corrected", 1, 4, 4, 4),
("corrected", 2, 4, 6, 7),
("moved", 1, 4, 4, 3),
("moved", 2, 1, 4, 3),
("tie", 1, 4, 4, 1),
("tie", 2, 4, 4, 6),
("at-decision", 1, 5, 5, 5),
("at-horizon", 1, 7, 7, 11),
("outcome-late", 1, 6, 9, 2),
]:
orders.append(
{
"order_id": order_id,
"revision": revision,
"product_id": "tea",
"event_at": time(event),
"available_at": time(available),
"units": units,
}
)
orders.append(dict(orders[0]))
products = []
for interval, revision, product, start, end, available, category, lead in [
("tea-old", 1, "tea", 1, 6, 1, "pantry", 3),
("tea-old", 2, "tea", 1, 6, 6, "pantry", 4),
("tea-new", 1, "tea", 6, 31, 6, "pantry", 2),
("coffee", 1, "coffee", 1, 31, 1, "pantry", 5),
("juice-a", 1, "juice", 1, 8, 1, "chilled", 1),
("juice-b", 1, "juice", 4, 31, 1, "chilled", 2),
]:
products.append(
{
"interval_id": interval,
"revision": revision,
"product_id": product,
"valid_from": time(start),
"valid_to": time(end),
"available_at": time(available),
"category": category,
"lead_days": lead,
}
)
return {
"requests": requests,
"orders": orders,
"products": products,
"coverage": [{"product_id": "tea", "observed_through": time(8), "available_at": time(9)}],
"evaluation_at": time(9),
}
def python_reference(requests, orders, products, coverage, evaluation_at):
"""Independent per-request enumeration, without DataFrame or query helpers."""
output = {name: [] for name in ("features", "naive_features", "outcomes", "rejected")}
for request in sorted(requests, key=lambda item: item["request_id"]):
decision = request["decision_at"]
dimensions = []
for interval in {row["interval_id"] for row in products}:
versions = [
row for row in products if row["interval_id"] == interval and row["available_at"] <= decision
]
if versions:
chosen = max(versions, key=lambda row: row["revision"])
if (
chosen["product_id"] == request["product_id"]
and chosen["valid_from"] <= decision < chosen["valid_to"]
):
dimensions.append(chosen)
if len(dimensions) != 1:
output["rejected"].append(
dict(
request,
reason="missing_product_interval" if not dimensions else "ambiguous_product_intervals",
)
)
continue
dimension = dimensions[0]
for name in ("features", "naive_features", "outcomes"):
chosen_orders = []
for order_id in {row["order_id"] for row in orders}:
versions = [
row
for row in orders
if row["order_id"] == order_id
and (
name == "naive_features"
or row["available_at"] <= (evaluation_at if name == "outcomes" else decision)
)
]
if not versions:
continue
chosen = max(versions, key=lambda row: row["revision"])
lower = decision if name == "outcomes" else request["history_start"]
upper = request["horizon_end"] if name == "outcomes" else decision
if (
chosen["product_id"] == request["product_id"]
and lower <= chosen["event_at"] < upper
and chosen["units"] > 0
):
chosen_orders.append(chosen)
if name == "outcomes":
complete = request["horizon_end"] <= evaluation_at and any(
row["product_id"] == request["product_id"]
and row["available_at"] <= evaluation_at
and row["observed_through"] >= request["horizon_end"]
for row in coverage
)
output[name].append(
{
"request_id": request["request_id"],
"horizon_end": request["horizon_end"],
"target_units": sum(row["units"] for row in chosen_orders) if complete else None,
"outcome_status": "observed" if complete else "unavailable",
}
)
else:
output[name].append(
{
"request_id": request["request_id"],
"product_id": request["product_id"],
"decision_at": decision,
"category": dimension["category"],
"lead_days": dimension["lead_days"],
"order_count": len(chosen_orders),
"units": sum(row["units"] for row in chosen_orders),
"last_order_at": max((row["event_at"] for row in chosen_orders), default=None),
}
)
return output
def run(spark):
return build_features(spark, **demo_inputs())
def reference():
import pyarrow as pa
feature_schema = pa.schema(
[
("request_id", pa.string()),
("product_id", pa.string()),
("decision_at", pa.timestamp("us")),
("category", pa.string()),
("lead_days", pa.int64()),
("order_count", pa.int64()),
("units", pa.int64()),
("last_order_at", pa.timestamp("us")),
]
)
schemas = {
"features": feature_schema,
"naive_features": feature_schema,
"outcomes": pa.schema(
[
("request_id", pa.string()),
("horizon_end", pa.timestamp("us")),
("target_units", pa.int64()),
("outcome_status", pa.string()),
]
),
"rejected": pa.schema(
[
("request_id", pa.string()),
("product_id", pa.string()),
("history_start", pa.timestamp("us")),
("decision_at", pa.timestamp("us")),
("horizon_end", pa.timestamp("us")),
("reason", pa.string()),
]
),
}
return {
name: pa.Table.from_pylist(rows, schema=schemas[name])
for name, rows in python_reference(**demo_inputs()).items()
}
if __name__ == "__main__":
from pysail.spark import SparkConnectServer
from pyspark.sql import SparkSession
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--input", type=Path, help="JSON object with requests, orders, products, coverage, evaluation_at"
)
args = parser.parse_args()
data = normalize_inputs(**json.loads(args.input.read_text())) if args.input else demo_inputs()
server = SparkConnectServer()
server.start()
_, port = server.listening_address
spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
try:
for name, table in build_features(spark, **data).items():
print(name)
table.show(128, truncate=False)
finally:
spark.stop()
server.stop()
Input contract
build_features(spark, requests, orders, products, coverage, evaluation_at)
accepts small lists of complete records. It returns deferred DataFrames and
does not create or stop the session. All timestamps describe UTC instants;
the query uses TIMESTAMP_NTZ UTC values so session time zones cannot change
the boundaries. Python callers may pass naive UTC datetime values. JSON
timestamps require Z or an explicit zero UTC offset.
| Input | Fields |
|---|---|
| Requests | request_id, product_id, history_start, decision_at, horizon_end |
| Order revisions | order_id, revision, product_id, event_at, available_at, units |
| Product interval revisions | interval_id, revision, product_id, valid_from, valid_to, available_at, category, lead_days |
| Coverage watermarks | product_id, observed_through, available_at |
Each request ID is unique, with history_start < decision_at < horizon_end.
Source revisions are positive integers, ordered within an order or interval ID.
They are not arrival sequence numbers. Higher revisions replace the whole
record, including its product and time fields. Units are nonnegative integers;
zero cancels an order. Orders cannot be available before their event time.
Product intervals are half-open, [valid_from, valid_to), with a lead time from
0 to 365 days.
Exact repeated source records collapse. Distinct records at the same source ID and revision fail the whole call, including distinct availability times. This requires a source-assigned immutable revision and a stable first availability time. Availability must mean the record was usable by the planning system, including ingestion delays. Fabricated or backdated availability defeats this check.
Fields cannot be null or omitted; unexpected fields, invalid values and
conflicting revisions raise ValueError with the offending input before any
query is built. Inputs are bounded to 128 requests and 5,000 rows per other
table. Keys and categories are nonblank strings of at most 200 characters;
units and revisions are capped at one billion. This keeps the local input and
per-request joins bounded. Larger ingestion pipelines should validate and load
their own tables rather than collecting them into these lists.
Selection rules
For each request, retain records with available_at <= decision_at, then choose
the highest available source revision per order or product interval. Apply the
product and event-time filters after selecting revisions. Otherwise an old
quantity can reappear when a correction moves an order outside the lookback or
to another product.
Historical orders must lie in [history_start, decision_at). Availability at
the decision is allowed; an event at the decision belongs to the future horizon.
Cancelled orders contribute neither units nor order count. A request with no
qualifying orders receives zero units and count, with a null last-order time.
Product attributes must match the requested product and be effective at the
decision. Exactly one selected interval must match. No match yields
missing_product_interval; multiple matches yield
ambiguous_product_intervals. Rejected requests retain their input fields and
do not enter feature or outcome tables. The example rejects ambiguity even if
overlapping intervals happen to carry identical attributes.
naive_features deliberately uses the latest order revision in the entire
supplied history and ignores order availability. It retains the same product
selection so the difference isolates order leakage. It is a diagnostic output,
not an input for a model.
Outcomes and checks
Future demand is separate from features. outcomes sums order units in
[decision_at, horizon_end) using revisions available by evaluation_at.
A target is available only if the horizon has ended and an available coverage
watermark reaches its end. A watermark asserts that the product's order feed
is complete through that event time as known at evaluation; it does not promise
that future corrections will never occur. Incomplete coverage produces a null
target with unavailable, never a fabricated zero. Complete coverage with no
orders produces an observed zero.
An independent Python implementation enumerates revisions separately for each
request and checks every output through reference(). Additional checks cover
boundary instants, cancellations, equal-time revisions, interval corrections,
empty inputs and delayed collection of earlier query plans. Appending valid
records available strictly after an old decision leaves that decision's
features unchanged. Outcomes and the deliberately naive diagnostic can change.
Use your own records
Put the four input lists and evaluation_at in a JSON object. Use the field
names above, integer revisions and units, and timestamps such as
"2026-01-05T00:00:00Z". Empty lists are supported.
For example, this complete input has one known order and an unavailable outcome:
{
"requests": [{"request_id":"plan-1","product_id":"tea","history_start":"2026-01-01T00:00:00Z","decision_at":"2026-01-05T00:00:00Z","horizon_end":"2026-01-07T00:00:00Z"}],
"orders": [{"order_id":"order-1","revision":1,"product_id":"tea","event_at":"2026-01-03T00:00:00Z","available_at":"2026-01-04T00:00:00Z","units":4}],
"products": [{"interval_id":"tea-jan","revision":1,"product_id":"tea","valid_from":"2026-01-01T00:00:00Z","valid_to":"2026-02-01T00:00:00Z","available_at":"2026-01-01T00:00:00Z","category":"pantry","lead_days":3}],
"coverage": [],
"evaluation_at": "2026-01-08T00:00:00Z"
}
Save the JSON above as pipelines/build-point-in-time-features/planning-inputs.json, then run:
uv run pipelines/build-point-in-time-features/main.py \
--input pipelines/build-point-in-time-features/planning-inputs.json
The console shows all bounded request results. No database or output directory is mutated. This reconstructs a batch feature table; source capture, streaming, model training and online serving are outside its scope.