CSV to live forecast

The full production loop — load history from a CSV, define a segment, train a model, deploy the winner, serve forecasts.

This is the canonical predictAI story, end to end: two years of daily sales history goes in as a CSV, and a continuously improving forecast comes out over HTTP. Every block builds on the previous one — run them top to bottom.

You'll need: an API token and a CSV of daily history with columns date,sales,foot_traffic,promo. No CSV handy? The first block generates a realistic one.

0. Demo data (skip if you have a CSV)

import csv, math, random
from datetime import date, timedelta

random.seed(42)
start = date.today() - timedelta(days=730)

with open("daily_sales.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["date", "sales", "foot_traffic", "promo"])
    for i in range(730):
        d = start + timedelta(days=i)
        promo = 1 if random.random() < 0.15 else 0
        season = 1 + 0.25 * math.sin(2 * math.pi * i / 365)
        weekday = 1.15 if d.weekday() < 5 else 0.8
        traffic = int(1800 * season * weekday * random.uniform(0.9, 1.1))
        sales = round(traffic * (7.0 + 1.5 * promo) * random.uniform(0.95, 1.05), 2)
        w.writerow([d.isoformat(), sales, traffic, promo])

import { writeFileSync } from "node:fs";

const lines = ["date,sales,foot_traffic,promo"];
const start = Date.now() - 730 * 86_400_000;
for (let i = 0; i < 730; i++) {
  const d = new Date(start + i * 86_400_000);
  const promo = Math.random() < 0.15 ? 1 : 0;
  const season = 1 + 0.25 * Math.sin((2 * Math.PI * i) / 365);
  const weekday = d.getUTCDay() >= 1 && d.getUTCDay() <= 5 ? 1.15 : 0.8;
  const traffic = Math.round(1800 * season * weekday * (0.9 + Math.random() * 0.2));
  const sales = +(traffic * (7.0 + 1.5 * promo) * (0.95 + Math.random() * 0.1)).toFixed(2);
  lines.push(`${d.toISOString().slice(0, 10)},${sales},${traffic},${promo}`);
}
writeFileSync("daily_sales.csv", lines.join("\n"));

1. Workspace

from predictai import PredictAI

client = PredictAI(token="pa_live_...")

workspace = client.workspaces.post_workspace(json={
    "name": "Coffee chain demand",
    "goal": "Forecast daily sales 14 days out",
    "data_granularity": "day",
})
WORKSPACE_ID = workspace["data"]["uid"]
client.workspace_id = WORKSPACE_ID

import { PredictAI } from "@predictai/sdk";

const client = new PredictAI({ token: "pa_live_..." });

const workspace = await client.workspaces.postWorkspace({
  json: {
    name: "Coffee chain demand",
    goal: "Forecast daily sales 14 days out",
    data_granularity: "day",
  },
});
const WORKSPACE_ID: string = workspace.data.uid;
(client as { workspaceId?: string }).workspaceId = WORKSPACE_ID;

2. CSV in — signals out

Each CSV column becomes a signal; each row, one value per signal:

import csv

with open("daily_sales.csv") as f:
    for row in csv.DictReader(f):
        ts = f"{row['date']}T00:00:00Z"
        client.signals.post_signal_push(json={"key": "daily_sales",  "value": float(row["sales"]),        "timestamp": ts})
        client.signals.post_signal_push(json={"key": "foot_traffic", "value": float(row["foot_traffic"]), "timestamp": ts})
        client.signals.post_signal_push(json={"key": "promo_active", "value": int(row["promo"]),          "timestamp": ts})

import { readFileSync } from "node:fs";

const [, ...rows] = readFileSync("daily_sales.csv", "utf8").trim().split("\n");
for (const row of rows) {
  const [date, sales, traffic, promo] = row.split(",");
  const timestamp = `${date}T00:00:00Z`;
  await client.signals.postSignalPush({ json: { key: "daily_sales", value: Number(sales), timestamp } });
  await client.signals.postSignalPush({ json: { key: "foot_traffic", value: Number(traffic), timestamp } });
  await client.signals.postSignalPush({ json: { key: "promo_active", value: Number(promo), timestamp } });
}

For big files, skip the loop: the ingestion API uploads the whole CSV in one call and maps columns to signals server-side.

3. The segment — your training table

A segment pins down exactly what a model trains on: which columns, on what time grid, with gaps filled how. Training and serving both read from it, so the model always sees the same data shape:

segment = client.segments.post_segment(json={
    "name": "Daily sales view",
    "workspace_id": WORKSPACE_ID,
    "features": ["foot_traffic", "promo_active", "daily_sales"],
    "labels": ["daily_sales"],
    "interval": 86400,
    "tolerance": 43200,
    "live": True,
})
SEGMENT_ID = segment["data"]["uid"]

const segment = await client.segments.postSegment({
  json: {
    name: "Daily sales view",
    workspace_id: WORKSPACE_ID,
    features: ["foot_traffic", "promo_active", "daily_sales"],
    labels: ["daily_sales"],
    interval: 86400,
    tolerance: 43200,
    live: true,
  },
});
const SEGMENT_ID: string = segment.data.uid;

4. A model and a pipeline

The model describes an architecture (this LSTM is the documented reference — input units matches the segment's three features). The pipeline binds it to the segment; with no schedule set, one training run queues immediately:

model = client.models.post_models(json={
    "kind": "custom",
    "name": "LSTM demand forecaster",
    "purpose": "Forecast daily sales from store drivers",
    "description": "Stacked LSTM for daily demand forecasting",
    "visibility": "private",
    "categories": ["retail"],
    "tags": ["demand"],
    "metadata": {"framework": "TensorFlow", "framework_version": "2.x", "training_environment": "cloud"},
    "model_blueprint": {
        "model_type": "sequential",
        "layers": [
            {"type": "Input", "units": 3},
            {"type": "LSTM", "units": 64, "activation": "tanh", "return_sequences": False},
            {"type": "Dropout", "rate": 0.2},
            {"type": "Dense", "units": 1, "activation": "linear"},
        ],
    },
    "compilation_details": {
        "loss": "huber",
        "optimizer": {"type": "adamw", "learning_rate": 0.001},
        "metrics": ["mae", "rmse"],
    },
    "training_details": {"epochs": 200, "batch_size": 64, "sequence_length": 20},
})
MODEL_ID = model["data"]["uid"]

pipeline = client.pipelines.post_pipelines(json={
    "name": "Daily sales forecaster",
    "segment": SEGMENT_ID,
    "workspace_id": WORKSPACE_ID,
    "models": [{"model": MODEL_ID}],
    "forecast_horizon": 14,
    "promotion_policy": {"mode": "auto"},
})
PIPELINE_ID = pipeline["data"]["uid"]

const model = await client.models.postModels({
  json: {
    kind: "custom",
    name: "LSTM demand forecaster",
    purpose: "Forecast daily sales from store drivers",
    description: "Stacked LSTM for daily demand forecasting",
    visibility: "private",
    categories: ["retail"],
    tags: ["demand"],
    metadata: { framework: "TensorFlow", framework_version: "2.x", training_environment: "cloud" },
    model_blueprint: {
      model_type: "sequential",
      layers: [
        { type: "Input", units: 3 },
        { type: "LSTM", units: 64, activation: "tanh", return_sequences: false },
        { type: "Dropout", rate: 0.2 },
        { type: "Dense", units: 1, activation: "linear" },
      ],
    },
    compilation_details: {
      loss: "huber",
      optimizer: { type: "adamw", learning_rate: 0.001 },
      metrics: ["mae", "rmse"],
    },
    training_details: { epochs: 200, batch_size: 64, sequence_length: 20 },
  },
});
const MODEL_ID: string = model.data.uid;

const pipeline = await client.pipelines.postPipelines({
  json: {
    name: "Daily sales forecaster",
    segment: SEGMENT_ID,
    workspace_id: WORKSPACE_ID,
    models: [{ model: MODEL_ID }],
    forecast_horizon: 14,
    promotion_policy: { mode: "auto" },
  },
});
const PIPELINE_ID: string = pipeline.data.uid;

5. Wait for training and deployment

The run takes a few minutes. promotion_policy: auto means the winner deploys itself — wait for both:

import time

while True:
    runs = client.trainings.get_training(params={"model_id": PIPELINE_ID, "per_page": 1})
    latest = runs["data"]["trainings"][0]
    print("training:", latest["last_status"])
    if latest["last_status"] in ("completed", "failed"):
        break
    time.sleep(30)

while True:
    status = client.deployments.get_deployments_pipelines_by_pipeline_id_status(PIPELINE_ID)
    if status["is_promoted"] and status["current_deployment"]["status"] == "deployed":
        break
    time.sleep(10)

print("accuracy:", latest["accuracy_score"], "- serving")

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

let latest: any;
for (;;) {
  const runs = await client.trainings.getTraining({ params: { model_id: PIPELINE_ID, per_page: 1 } });
  latest = runs.data.trainings[0];
  console.log("training:", latest.last_status);
  if (latest.last_status === "completed" || latest.last_status === "failed") break;
  await sleep(30_000);
}

for (;;) {
  const status = await client.deployments.getDeploymentsPipelinesByPipelineIdStatus(PIPELINE_ID);
  if (status.is_promoted && status.current_deployment.status === "deployed") break;
  await sleep(10_000);
}

console.log("accuracy:", latest.accuracy_score, "- serving");

6. Serve forecasts

Segment-based inference rebuilds the model's input from the segment's live data on every request — an empty body is a complete request. The pipeline URL survives re-trainings and re-deployments:

forecast = client.inference.post_inference_pipelines_by_pipeline_id_infer_segment(
    PIPELINE_ID, json={},
)

print(forecast["predictions"][0]["label"], "->",
      [round(v, 1) for v in forecast["predictions"][0]["point"]])
print("built from:", forecast["data_period"])

const forecast = await client.inference.postInferencePipelinesByPipelineIdInferSegment(
  PIPELINE_ID, { json: {} },
);

console.log(forecast.predictions[0].label, "->", forecast.predictions[0].point);
console.log("built from:", forecast.data_period);

daily_sales -> [13102.4, 13350.9, 13571.2, ...]
built from: {'start': '2026-06-30T00:00:00Z', 'end': '2026-07-16T00:00:00Z'}

The loop is closed

Keep pushing signals as they happen — the live segment keeps model input fresh, and a training_schedule or retrain trigger retrains on your cadence. With promotion_policy: auto, a new training only replaces the serving model when it beats it, so accuracy ratchets upward.

Race more architectures per run — Race a model pool.
Prefer outcomes over infrastructure? Hand the whole loop to a goal on autopilot.

On this page