An anomaly goal, end to end

Upload a CSV in one call, create an anomaly goal, watch everything happen live over WebSockets, and wire alerts — the whole platform in one sitting.

The capstone. Everything the other examples did one piece at a time happens here in a single flow — and instead of polling, you watch it all happen live over WebSockets: one CSV upload brings the history in, an anomaly goal learns what "normal" looks like, and alerts page you when the probability of an anomaly spikes.

The story: an ops team wants to know when checkout errors start behaving abnormally — before the support tickets arrive.

You'll need: an API token, and for realtime: pip install "predictai[realtime]" (Python) or Node 22+ / npm install ws (Node).

1. Workspace and a live connection

Open the WebSocket before doing anything else — then you get to watch every following step as it happens:

from predictai import PredictAI

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

workspace = client.workspaces.post_workspace(json={"name": "Checkout health"})
WORKSPACE_ID = workspace["data"]["uid"]
client.workspace_id = WORKSPACE_ID

rt = client.realtime.connect(channels=[f"workspace:{WORKSPACE_ID}"])
print("connected:", rt.welcome["connection_id"], "in cell", rt.welcome["cell"])

import { PredictAI } from "@predictai/sdk";
import WS from "ws"; // Node 22+ has a global WebSocket - drop this and the option below

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

const workspace = await client.workspaces.postWorkspace({ json: { name: "Checkout health" } });
const WORKSPACE_ID: string = workspace.data.uid;
(client as { workspaceId?: string }).workspaceId = WORKSPACE_ID;

const rt = await client.realtime.connect({
  channels: [`workspace:${WORKSPACE_ID}`],
  webSocket: WS,
});
console.log("connected:", rt.welcome?.connection_id, "in cell", rt.welcome?.cell);

The workspace:ID channel is the firehose — ingestion, trainings, deployments, and every goal event in the workspace flow through it.

2. Ninety days of history, one file

No push loops this time. Generate an hourly ops CSV — three metrics, with a few genuine anomaly bursts baked in so the goal has something to learn from:

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

random.seed(11)
start = datetime.now(timezone.utc) - timedelta(days=90)
burst_hours = {random.randrange(90 * 24) for _ in range(6)}   # 6 incident bursts

with open("checkout_metrics.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["timestamp", "checkout_errors", "gateway_latency_ms", "checkout_traffic"])
    for h in range(90 * 24):
        ts = start + timedelta(hours=h)
        daily = 1 + 0.6 * math.sin(2 * math.pi * (h % 24) / 24 - 1.5)  # day/night cycle
        traffic = int(1200 * daily * random.uniform(0.9, 1.1))
        latency = round(180 * random.uniform(0.85, 1.25), 1)
        errors = traffic * 0.004 * random.uniform(0.6, 1.4)
        if h in burst_hours or (h - 1) in burst_hours:                 # incidents: errors x8, latency x3
            errors *= 8
            latency *= 3
        w.writerow([ts.isoformat(), round(errors, 1), latency, traffic])

import { writeFileSync } from "node:fs";

const lines = ["timestamp,checkout_errors,gateway_latency_ms,checkout_traffic"];
const start = Date.now() - 90 * 24 * 3_600_000;
const burstHours = new Set(Array.from({ length: 6 }, () => Math.floor(Math.random() * 90 * 24)));

for (let h = 0; h < 90 * 24; h++) {
  const ts = new Date(start + h * 3_600_000).toISOString();
  const daily = 1 + 0.6 * Math.sin((2 * Math.PI * (h % 24)) / 24 - 1.5); // day/night cycle
  const traffic = Math.round(1200 * daily * (0.9 + Math.random() * 0.2));
  let latency = 180 * (0.85 + Math.random() * 0.4);
  let errors = traffic * 0.004 * (0.6 + Math.random() * 0.8);
  if (burstHours.has(h) || burstHours.has(h - 1)) {                     // incidents: errors x8, latency x3
    errors *= 8;
    latency *= 3;
  }
  lines.push(`${ts},${errors.toFixed(1)},${latency.toFixed(1)},${traffic}`);
}
writeFileSync("checkout_metrics.csv", lines.join("\n"));

Upload it in one multipart call — format detection and column typing happen server-side (Connect & import) — then watch the import complete on the socket instead of polling:

with open("checkout_metrics.csv", "rb") as f:
    upload = client.ingestion.post_ingestion_connect_upload(files={"file": f})

execution_id = upload["data"]["execution_id"]
print("import queued:", execution_id)

for frame in rt.events():
    event, data = frame["event"], frame["data"]
    if not event.startswith("ingestion.execution."):
        continue
    print(event, data.get("percentage", ""))
    if data.get("execution_id") != execution_id:
        continue
    if event == "ingestion.execution.completed":
        break
    if event == "ingestion.execution.failed":
        raise RuntimeError(data.get("error"))

print("history is in - 3 signals created from the CSV columns")

import { openAsBlob } from "node:fs";

const form = new FormData();
form.append("file", await openAsBlob("checkout_metrics.csv"), "checkout_metrics.csv");
const upload = await client.ingestion.postIngestionConnectUpload({ form });

const executionId: string = upload.data.execution_id;
console.log("import queued:", executionId);

for await (const frame of rt.events()) {
  const { event, data } = frame;
  if (!event!.startsWith("ingestion.execution.")) continue;
  console.log(event, data.percentage ?? "");
  if (data.execution_id !== executionId) continue;
  if (event === "ingestion.execution.completed") break;
  if (event === "ingestion.execution.failed") throw new Error(data.error);
}

console.log("history is in - 3 signals created from the CSV columns");

import queued: e5b2c9d4-...
ingestion.execution.started
ingestion.execution.progress 42
ingestion.execution.progress 87
ingestion.execution.completed
history is in - 3 signals created from the CSV columns

3. Create the anomaly goal

An anomaly goal doesn't forecast a level — it learns what normal looks like for the target and serves the probability that current behavior is anomalous, refreshed on the data's cadence. Two fields; cadence, horizon, and drivers are resolved from your data:

goal = client.goals.post_goals(json={
    "target_key": "checkout_errors",
    "goal_type": "anomaly",
})
GOAL_ID = goal["data"]["goal_id"]

rt.subscribe(f"goal:{GOAL_ID}")   # alerts.evaluated flows only on the goal channel

const goal = await client.goals.postGoals({
  json: { target_key: "checkout_errors", goal_type: "anomaly" },
});
const GOAL_ID: string = goal.data.goal_id;

rt.subscribe(`goal:${GOAL_ID}`);  // alerts.evaluated flows only on the goal channel

Now watch the goal build itself — discovery stages, the model tournament, and the moment it starts serving, all as events:

for frame in rt.events():
    event, data = frame["event"], frame["data"]
    if event == "goal.run.stage":
        print(f'discovery · {data["stage"]:16} {data["status"]}')
    elif event == "goal.tournament.status":
        print("tournament:", data["status"])
    elif event == "goal.status":
        print("goal:", data["status"])
        if data["status"] == "serving":
            break
    elif event in ("goal.run.failed",):
        raise RuntimeError(data.get("error"))

for await (const frame of rt.events()) {
  const { event, data } = frame;
  if (event === "goal.run.stage") {
    console.log(`discovery · ${data.stage} ${data.status}`);
  } else if (event === "goal.tournament.status") {
    console.log("tournament:", data.status);
  } else if (event === "goal.status") {
    console.log("goal:", data.status);
    if (data.status === "serving") break;
  } else if (event === "goal.run.failed") {
    throw new Error(data.error);
  }
}

discovery · profile          completed
discovery · represent        completed
discovery · generate         completed
...
tournament: running
tournament: completed
goal: serving

4. Wire the alert

Anomaly goals serve a probability, so alert conditions use the probability metric (0–100, percent likelihood). Preview it first — free, and it tells you whether it would fire right now — then save it:

condition = {"metric": "probability", "scope": "any", "operator": "gte", "threshold": 80}

preview = client.goals.post_goals_by_goal_id_alerts_preview(GOAL_ID, json={"condition": condition})
print("would fire now:", preview["data"]["triggered"], "-", preview["data"]["detail"])

alert = client.goals.post_goals_by_goal_id_alerts(GOAL_ID, json={
    "name": "Checkout anomaly",
    "condition": condition,
    "severity": "critical",
    "cooldown_minutes": 60,
    "channels": ["in_app", "email", "websocket"],
})["data"]["alert"]
print("alert armed:", alert["alert_id"])

const condition = { metric: "probability", scope: "any", operator: "gte", threshold: 80 };

const { data: preview } = await client.goals.postGoalsByGoalIdAlertsPreview(GOAL_ID, {
  json: { condition },
});
console.log("would fire now:", preview.triggered, "-", preview.detail);

const { data: created } = await client.goals.postGoalsByGoalIdAlerts(GOAL_ID, {
  json: {
    name: "Checkout anomaly",
    condition,
    severity: "critical",
    cooldown_minutes: 60,
    channels: ["in_app", "email", "websocket"],
  },
});
console.log("alert armed:", created.alert.alert_id);

Add slack or pagerduty to channels once you've registered a channel endpoint — the alerts example shows that flow.

5. The watchdog loop

That's the whole setup. What remains is the part that runs forever: keep streaming live metrics in (your real pipeline replaces this loop), and react when the goal sees something wrong:

for frame in rt.events():
    event, data = frame["event"], frame["data"]

    if event == "goal.anomaly":
        print("ANOMALY:", data)                 # detection details

    elif event == "alerts.evaluated" and data["fired"] > 0:
        print(f'{data["fired"]} alert(s) fired - inference {data["inference_uid"]}')
        forecast = client.goals.get_goals_by_goal_id_forecast(GOAL_ID)["data"]["forecast"]
        print("anomaly probability now:", forecast["point"][0])

    elif event == "goal.status" and data.get("shock_state") not in (None, "normal"):
        print("shock state:", data["shock_state"])   # see shock analogues in the goal docs

for await (const frame of rt.events()) {
  const { event, data } = frame;

  if (event === "goal.anomaly") {
    console.log("ANOMALY:", data);              // detection details

  } else if (event === "alerts.evaluated" && data.fired > 0) {
    console.log(`${data.fired} alert(s) fired - inference ${data.inference_uid}`);
    const { data: g } = await client.goals.getGoalsByGoalIdForecast(GOAL_ID);
    console.log("anomaly probability now:", g.forecast.point[0]);

  } else if (event === "goal.status" && data.shock_state && data.shock_state !== "normal") {
    console.log("shock state:", data.shock_state);  // see shock analogues in the goal docs
  }
}

Two delivery paths, on purpose: alerts.evaluated on the socket reaches your systems within the same second, while the alert's channels reach people through Notifications with the human-readable condition label.

Realtime delivery is best-effort and nothing is replayed after a disconnect — on reconnect, resubscribe and reconcile via REST (the goal's activity feed and the alert's last_result are your source of truth). A 4401 close means the token was rejected: replace it before reconnecting.

What you built

One CSV upload, two API calls of configuration, and a socket:

Ingestion turned a file into three typed signals — no schema, no push loops.
The anomaly goal profiled the target, discovered its drivers (latency and traffic), trained a tournament, and is now serving an anomaly probability that refreshes as data arrives.
Alerts watch that probability with a threshold you previewed before arming.
WebSockets replaced every polling loop — import progress, discovery stages, tournament, serving, and firings, all pushed to you.

Keep the socket loop running in a small worker, keep your metrics flowing in through signals or a scheduled ingestion sync, and the watchdog runs itself.