Docs/audience language review

docs/advanced/asset-health-score.mdx

@@ -20,23 +20,22 @@ can rank a whole fleet at a glance and a dashboard can show green/amber/red. It'
 lightweight cousin of [predictive maintenance](/advanced/predictive-maintenance): no
 model to train, just a transparent, tunable index.
 
-:::tip New to machine learning?
-No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
-ideas in plain language — model, feature, training, and the algorithm itself — and use
-[Generate sample data](/advanced/generate-sample-data) for a sandbox to run this against.
-:::
-
-:::tip Need data to run this?
+:::info Needs a sandbox
 This reads several pump signals. [Generate a sandbox](/advanced/generate-sample-data)
 first — section A ingests `pump_07_bearing_temp_c`, `pump_07_oil_pressure_kpa` and a
 stand-in `pump_07_vibration_anomaly` (or run [predictive maintenance](/advanced/predictive-maintenance)
 to produce the real one).
 :::
 
+:::tip New to machine learning?
+No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
+ideas in plain language — model, feature, training, and the algorithm itself.
+:::
+
 ## 1. Pull the latest value of each signal
 
-Take the most recent reading (or short average) of each contributing series for the
-asset.
+Take the most recent reading of each contributing series for the asset. The seed writes
+these hourly, so ask for the latest datapoint rather than a short window that may be empty.
 
 <Tabs groupId="lang">
 <TabItem value="python" label="Python">
@@ -47,13 +46,8 @@ import intellistream_datahub_sdk, pandas as pd
 client = intellistream_datahub_sdk.DataHubClient.from_env()
 
 def latest(external_id):
-    rf = intellistream_datahub_sdk.RetrieveFilter(
-        ts=external_id,
-        start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(minutes=15),
-        end=pd.Timestamp.now(tz="UTC"),
-        aggregates=["avg"], granularity="15m")
-    pts = client.timeseries.retrieve_datapoints(rf)[0].get_datapoints()
-    return float(pts[-1].average)
+    pts = client.timeseries.retrieve_latest_datapoints([external_id])[0].get_datapoints()
+    return float(pts[-1].value)
 
 signals = {
     "vibration": latest("pump_07_vibration_anomaly"),   # 0..~1, from the anomaly model
@@ -66,13 +60,13 @@ signals = {
 <TabItem value="java" label="Java">
 
 ```java
-// The same retrieve, per signal; the scoring arithmetic below is shown in Python.
+// Java has no latest-datapoint call: read the last day and take the newest point.
+// The scoring arithmetic below is shown in Python.
 var filter = new RetrieveFilter();
 filter.setExternalId("pump_07_bearing_temp_c");
-filter.setStart(ZonedDateTime.now().minusMinutes(15));
+filter.setStart(ZonedDateTime.now().minusHours(24));
 filter.setEnd(ZonedDateTime.now());
-filter.setAggregates(List.of("avg"));
-filter.setGranularity("15m");
+filter.setLimit(100);
 
 var request = new DataRetriever<RetrieveFilter>();
 request.setItems(List.of(filter));
@@ -84,21 +78,14 @@ double bearingTemp = Double.parseDouble(pts.get(pts.size() - 1).getValue());
 <TabItem value="rust" label="Rust">
 
 ```rust
-// The same retrieve, per signal; the scoring arithmetic below is shown in Python.
-use intellistream_datahub_sdk::generic::{DataWrapper, RetrieveFilter};
-use chrono::Utc;
-
-let filter = RetrieveFilter {
-    external_id: Some("pump_07_bearing_temp_c".into()),
-    start: Some(Utc::now() - chrono::Duration::minutes(15)),
-    end: Some(Utc::now()),
-    aggregates: Some(vec!["avg".into()]),
-    granularity: Some("15m".into()),
-    ..Default::default()
-};
-let s = api.time_series
-    .retrieve_datapoints(&DataWrapper::from(vec![filter])).await?
+// The latest datapoint, per signal; the scoring arithmetic below is shown in Python.
+use intellistream_datahub_sdk::generic::{DataWrapper, IdAndExtId};
+
+let latest = api.time_series
+    .retrieve_latest_datapoint(&DataWrapper::from(vec![
+        IdAndExtId::from_external_id("pump_07_bearing_temp_c")])).await?
     .get_items().remove(0);
+let bearing_temp = latest.datapoints.last().and_then(|p| p.value).unwrap_or(f64::NAN);
 ```
 
 </TabItem>
@@ -156,6 +143,11 @@ if band == "critical":
         event_time=pd.Timestamp.now(tz="UTC"),
         metadata={"asset": "pump_07", "score": str(score),
                   "worst_signal": max(parts, key=parts.get)})])
+
+# read the score back
+stored = client.timeseries.retrieve_latest_datapoints(["pump_07_health_score"])[0].get_datapoints()
+print(f"health {stored[-1].value:.1f} ({band})")
+assert abs(stored[-1].value - score) < 0.01, "the score was not written"
 ```
 
 Run it across the fleet on a schedule and you have a single ranked health view —

docs/advanced/data-cleaning-lineage.mdx

@@ -14,10 +14,14 @@ linked in the graph, so the whole `raw → … → result` chain is traceable bo
 **Stack:** the SDK for data and graph, plus `pandas`, `numpy`.
 :::
 
-:::tip New to this?
-No data-science background needed — see the [gentle primer](/advanced/machine-learning-gently)
-for any unfamiliar term, and [generate a sandbox](/advanced/generate-sample-data) to run
-this against.
+:::info Needs a sandbox
+This reads two raw sensors that already exist. [Generate a sandbox](/advanced/generate-sample-data)
+first — section K ingests `engine_temperature_raw` and `engine_vibration_raw`, defects and all.
+:::
+
+:::tip New to machine learning?
+No data-science background needed. Skim the [gentle primer](/advanced/machine-learning-gently)
+for any unfamiliar term.
 :::
 
 **The idea in one paragraph.** Real data never goes straight from sensor to model. It's
@@ -90,8 +94,9 @@ how they *relate*. That's step 2.
 
 Model each transformation as a **function node** and link it up: each input series is
 `processed_by` the function, which `produces` an output series. Done across the pipeline,
-this builds a branching, ten-deep lineage DAG in one call. (`resources.create` is the
-same call in Java and Rust — see [model assets as a graph](/guides/model-assets-graph).)
+this builds a branching lineage DAG of 18 nodes (8 functions and 10 series) in one call.
+(`resources.create` is the same call in Java and Rust — see
+[model assets as a graph](/guides/model-assets-graph).)
 
 ```python
 def fn(ext_id, name):
@@ -127,7 +132,10 @@ client.resources.create(functions, edges)
 ```
 
 That graph branches (each cleaned signal feeds two features), converges (four features
-into one vector), and runs ten nodes deep from raw sensor to health score.
+into one vector), and runs nine nodes deep from raw sensor to health score:
+`engine_temperature_raw` → `clean_temp_fn` → `engine_temperature_clean` → `roll_mean_fn` →
+`engine_temp_roll_mean` → `assemble_fn` → `engine_feature_vector` → `score_fn` →
+`engine_health_score`.
 
 ## 3. The payoff — trace it backward
 
@@ -137,8 +145,9 @@ ultimately came from. No guessing, no stale wiki page.
 
 ```python
 back = client.resources.fetch_related(external_id="engine_health_score", depth=12)
-raw_sources = [n.external_id for n in back.nodes if n.external_id.endswith("_raw")]
+raw_sources = {n.external_id for n in back.nodes if n.external_id.endswith("_raw")}
 print("this score derives from:", raw_sources)   # engine_temperature_raw, engine_vibration_raw
+assert raw_sources == {"engine_temperature_raw", "engine_vibration_raw"}, raw_sources
 ```
 
 ## 4. The payoff — trace it forward (impact analysis)
@@ -155,6 +164,7 @@ affected = [n.external_id for n in fwd.nodes
 print("affected by the bad temperature sensor:", affected)
 # engine_temperature_clean, engine_temp_roll_mean, engine_temp_roc,
 # engine_feature_vector, engine_health_score   — but NOT the vibration features
+assert "engine_health_score" in affected and "engine_vib_rms" not in affected, affected
 ```
 
 This is the same [blast-radius traversal](/guides/correlate-alarms) used for asset

docs/advanced/demand-forecasting.mdx

@@ -19,17 +19,16 @@ hour. The pattern is the same regardless of domain — past values plus calendar
 predict the next ones — and the forecast becomes a **new series** you can chart against
 actuals and alert on.
 
-:::tip New to machine learning?
-No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
-ideas in plain language — model, feature, training, and the algorithm itself — and use
-[Generate sample data](/advanced/generate-sample-data) for a sandbox to run this against.
-:::
-
-:::tip Need data to run this?
+:::info Needs a sandbox
 This reads a load history that already exists. [Generate a sandbox](/advanced/generate-sample-data)
 first — section G ingests the `feeder_f12_load_mw` curve this build forecasts.
 :::
 
+:::tip New to machine learning?
+No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
+ideas in plain language — model, feature, training, and the algorithm itself.
+:::
+
 ## 1. Load the history
 
 Pull a long, regularly-sampled history of the quantity you want to forecast.
@@ -50,7 +49,7 @@ rf = intellistream_datahub_sdk.RetrieveFilter(
 
 points = client.timeseries.retrieve_datapoints(rf)[0].get_datapoints()
 demand = pd.Series([float(p.average) for p in points],
-                   index=pd.to_datetime([p.timestamp for p in points])).sort_index()
+                   index=pd.to_datetime([p.timestamp for p in points], utc=True)).sort_index()
 demand = demand.asfreq("1h").interpolate()        # regular hourly grid
 ```
 
@@ -166,6 +165,14 @@ client.timeseries.create([intellistream_datahub_sdk.TimeSeries(
     external_id="feeder_f12_load_mw_forecast", name="Feeder F12 load — 48h forecast", unit="mw", value_type="float")])
 client.timeseries.insert_from_lists(
     timestamps=forecast.index, values=forecast.to_numpy(), ts="feeder_f12_load_mw_forecast")
+
+# read the horizon back: one point per forecast hour
+stored = client.timeseries.retrieve_datapoints(intellistream_datahub_sdk.RetrieveFilter(
+    ts="feeder_f12_load_mw_forecast",
+    start=forecast.index[0], end=forecast.index[-1] + pd.Timedelta(minutes=1),
+    limit=1000))[0].get_datapoints()
+print(f"{len(stored)} forecast hours stored")
+assert len(stored) == HORIZON, f"expected {HORIZON}, got {len(stored)}"
 ```
 
 With the forecast stored as a series, a [threshold rule](/guides/detect-events) on it
@@ -184,7 +191,7 @@ actual, is set to exceed a limit, hours before it happens.
 ## Further reading
 
 - **Gradient boosting** — [Wikipedia](https://en.wikipedia.org/wiki/Gradient_boosting) · [scikit-learn](https://scikit-learn.org/stable/modules/ensemble.html#histogram-based-gradient-boosting)
-- **Time-series forecasting** — [Wikipedia](https://en.wikipedia.org/wiki/Time_series)
+- **Time series forecasting** — [Wikipedia](https://en.wikipedia.org/wiki/Time_series)
 - **The ideas in plain language** — [Machine learning, gently](/advanced/machine-learning-gently)
 
 ## See also

docs/advanced/fraud-classification.mdx

@@ -13,8 +13,7 @@ combined with behavioural features, feeding a supervised classifier · **Stack:*
 for data and traversal, plus `networkx`, `pandas`, `scikit-learn`.
 :::
 
-This is the capstone: it fuses the two things the rest of the docs treat separately — the
-**knowledge graph** and **machine learning**. Money laundering doesn't look suspicious one
+Money laundering doesn't look suspicious one
 payment at a time; it looks suspicious in the *shape* of the network — funds fanning out
 through mules and looping back — combined with behaviour like rapid pass-through. A rules
 engine flags thousands of alerts a day; a classifier that scores each one by its network
@@ -24,17 +23,16 @@ The technique generalises to any "score an entity by its connections plus its be
 problem — [wafer-lot risk](/industries/manufacturing-process/semiconductor), [insurance rings](/industries/financial-services/insurance-fraud),
 [telecom abuse](/industries/technology-operations/network).
 
-:::tip New to machine learning?
-No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
-ideas in plain language — model, feature, training, and the algorithm itself — and use
-[Generate sample data](/advanced/generate-sample-data) for a sandbox to run this against.
-:::
-
-:::tip Need data to run this?
+:::info Needs a sandbox
 This walks a transfer graph that already exists. [Generate a sandbox](/advanced/generate-sample-data)
 first — section F creates the flagged ring around `account_77310`.
 :::
 
+:::tip New to machine learning?
+No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
+ideas in plain language — model, feature, training, and the algorithm itself.
+:::
+
 ## 1. Pull the account's network
 
 For a given account, [walk the money-flow graph](/guides/correlate-alarms) to get its
@@ -210,6 +208,14 @@ if risk > 0.7:
             "in_loop": str(feat["in_loop"]),
             "pass_through": f"{feat['pass_through']:.2f}",
         })])
+
+# read the score back
+pts = client.timeseries.retrieve_datapoints(intellistream_datahub_sdk.RetrieveFilter(
+    ts=f"{alert}_aml_risk",
+    start=pd.Timestamp.now(tz="UTC") - pd.Timedelta(minutes=5),
+    end=pd.Timestamp.now(tz="UTC")))[0].get_datapoints()
+print(f"{len(pts)} risk score(s) stored, latest {pts[-1].value:.2f}")
+assert len(pts) >= 1, "the risk score was not written"
 ```
 
 The risk score is now a live series the investigations dashboard ranks on, and each

docs/advanced/generate-sample-data.mdx

@@ -23,8 +23,12 @@ import intellistream_datahub_sdk, numpy as np, pandas as pd
 client = intellistream_datahub_sdk.DataHubClient.from_env()
 
 def ingest(external_id, index, values, unit=None, name=None):
-    client.timeseries.create([intellistream_datahub_sdk.TimeSeries(
-        external_id=external_id, name=name or external_id, unit=unit or "value", value_type="float")])
+    try:
+        client.timeseries.create([intellistream_datahub_sdk.TimeSeries(
+            external_id=external_id, name=name or external_id, unit=unit or "value", value_type="float")])
+    except intellistream_datahub_sdk.DataHubException as e:
+        if e.status_code != 409:          # a second run finds the series already there
+            raise
     client.timeseries.insert_from_lists(timestamps=index, values=np.asarray(values), ts=external_id)
     print(f"ingested {len(values):,} points → {external_id}")
 ```
@@ -34,11 +38,13 @@ def ingest(external_id, index, values, unit=None, name=None):
 For [predictive maintenance](/advanced/predictive-maintenance),
 [LSTM anomaly detection](/advanced/lstm-anomaly-detection) and
 [health scoring](/advanced/asset-health-score): a vibration signal that runs healthy,
-then develops a fault — rising amplitude and impulsiveness over the final stretch.
+then develops a fault — rising amplitude and impulsiveness over the final stretch. It is
+written at **one reading a second**, which is the rate the predictive-maintenance page's
+`SAMPLE_HZ` and window length assume.
 
 ```python
 def degrading_vibration(hours=24, fs_per_hour=3600):
-    n = hours * fs_per_hour
+    n = hours * fs_per_hour                    # one reading a second
     idx = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="1s")
     t = np.linspace(0, 1, n)
     healthy = 0.6 * np.sin(2 * np.pi * 50 * np.arange(n) / fs_per_hour)   # base tone

docs/advanced/kmeans-clustering.mdx

@@ -19,15 +19,14 @@ centre. That one idea answers three different operational questions, all covered
 (cohorts), and *which parts of the network belong together?* (communities). The trick is
 always the same: turn the thing you want to group into a **feature vector**.
 
-:::tip Need data to run this?
+:::info Needs a sandbox
 These steps read series (and a graph for section 3) that already exist.
-[Generate a sandbox](/advanced/generate-sample-data) first.
+[Generate a sandbox](/advanced/generate-sample-data) first: sections I and F.
 :::
 
 :::tip New to machine learning?
 No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
-ideas in plain language — model, feature, training, and the algorithm itself — and use
-[Generate sample data](/advanced/generate-sample-data) for a sandbox to run this against.
+ideas in plain language — model, feature, training, and the algorithm itself.
 :::
 
 ## 1. Asset cohorts — group assets that behave alike
@@ -60,7 +59,8 @@ assets = ["pump_07", "pump_08", "pump_11", "pump_19"]   # ...your fleet
 X = pd.DataFrame([signature(a) for a in assets], index=assets)
 Xs = StandardScaler().fit_transform(X)
 
-km = KMeans(n_clusters=4, n_init=10, random_state=0).fit(Xs)
+# k must be well below the number of assets, or every asset is its own cluster
+km = KMeans(n_clusters=2, n_init=10, random_state=0).fit(Xs)
 X["cohort"] = km.labels_
 ```
 
@@ -119,6 +119,10 @@ for asset, d in zip(assets, dist):
             type="peer_outlier", status="open",
             event_time=pd.Timestamp.now(tz="UTC"),
             metadata={"asset": asset, "cohort": str(int(km.labels_[assets.index(asset)]))})])
+
+outliers = list(client.events.filter(intellistream_datahub_sdk.EventFilter(
+    basic_filter=intellistream_datahub_sdk.BasicEventFilter(type="peer_outlier"), limit=100)))
+print(f"{len(outliers)} peer outlier(s) on record")
 ```
 
 ## 2. Operating regimes — group an asset's *states*
@@ -191,7 +195,7 @@ the k that separates clusters best:
 ```python
 from sklearn.metrics import silhouette_score
 scores = {k: silhouette_score(Xs, KMeans(k, n_init=10, random_state=0).fit_predict(Xs))
-          for k in range(2, 8)}
+          for k in range(2, len(Xs))}          # the silhouette needs fewer clusters than points
 best_k = max(scores, key=scores.get)
 ```
 

docs/advanced/lstm-anomaly-detection.mdx

@@ -20,15 +20,14 @@ unmistakable. An **LSTM autoencoder** learns to reconstruct that normal joint be
 when a real event arrives, it can't reconstruct it well, and the reconstruction error
 spikes. It needs **no examples of failure** — only normal operation.
 
-:::tip Need data to run this?
+:::info Needs a sandbox
 These steps read series that already exist. [Generate a sandbox](/advanced/generate-sample-data)
 first — section J ingests the correlated drilling channels (with an injected kick) this build watches.
 :::
 
 :::tip New to machine learning?
 No background needed. Skim the [gentle primer](/advanced/machine-learning-gently) for the
-ideas in plain language — model, feature, training, and the algorithm itself — and use
-[Generate sample data](/advanced/generate-sample-data) for a sandbox to run this against.
+ideas in plain language — model, feature, training, and the algorithm itself.
 :::
 
 ## 1. Load the normal multivariate window
@@ -51,7 +50,7 @@ def load(external_id, start, end):
     rf = intellistream_datahub_sdk.RetrieveFilter(ts=external_id, start=start, end=end, limit=100_000)
     pts = client.timeseries.retrieve_datapoints(rf)[0].get_datapoints()
     return pd.Series([float(p.value) for p in pts],
-                     index=pd.to_datetime([p.timestamp for p in pts])).sort_index()
+                     index=pd.to_datetime([p.timestamp for p in pts], utc=True)).sort_index()
 
 start = pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=1)      # a stretch of normal running
 end   = pd.Timestamp.now(tz="UTC") - pd.Timedelta(minutes=12)   # ends before the recent kick
@@ -164,6 +163,13 @@ if error[-3:].mean() > THRESHOLD:
         type="kick_detected", status="critical",
         event_time=score_index[-1],
         metadata={"rig": "rig_deepwater_1", "score": f"{error[-1]:.3f}", "model": "lstm_ae_v1"})])
+
+# read the score back: one point per scored second
+stored = client.timeseries.retrieve_datapoints(intellistream_datahub_sdk.RetrieveFilter(
+    ts="rig_dw1_anomaly_score", start=score_index[0], end=pd.Timestamp.now(tz="UTC"),
+    limit=10_000))[0].get_datapoints()
+print(f"{len(stored)} scores stored, latest {stored[-1].value:.3f} against a limit of {THRESHOLD:.3f}")
+assert len(stored) == len(error), f"expected {len(error)}, got {len(stored)}"
 ```
 
 ## Where to take it further