Build an anomaly monitoring engine#

Most "anomaly detection" projects fail at the plumbing, not the math: joining the stream to the historical baseline, getting the alert to the right investigator with provenance, and reconstructing later why the system fired. RelataDB already owns that plumbing — bi-temporal storage, streaming windows, commit-time detection, governed alerts, and a tamper-evident audit chain — so you build the detection logic, not the pipeline.

This page walks the anomaly primitives Relata ships, then designs a worked Twitter / social monitoring engine end-to-end, and closes with a statistical-bias / distribution-shift detector (the same machinery, pointed at skew instead of spikes).

Every capability below is verified against source. The one honesty flag: AnomalyRateJob is a library type, not a turnkey scheduled job — Relata's JobRegistry has no scheduler consumer for it yet, so you trigger it from your own cron/scheduler. See the maturity table.

The anomaly primitives#

Worked design — a Twitter / social monitoring engine#

Goal: ingest social posts (and engagement signals), surface spikes (volume, sentiment, coordinated behavior), fuse identities across handles, detect threat patterns, and present an investigator with a defensible, time-travelable case. Relata does the storage / identity / detection / audit; you write the per-feed fetcher.

Architecture#

        fetcher(s) ──► /ingest (HTTP/Kafka/OTLP) ──► governed store
        (your code)         │                          │  SmartIngest canonicalizes
                            │                          │  handles → identity graph
                            ▼                          ▼
        ┌────────────── streaming SQL windows ──────────────┐
        │  TUMBLE / HOP / SESSION over Post/event rows      │
        └──────────────────────┬───────────────────────────┘
                               │
        ┌──────────────────────▼───────────────────────────┐
        │  detection rules (commit-driven) + AnomalyRateJob │
        │  + pattern library (contradictions)               │
        └──────────────────────┬───────────────────────────┘
                               ▼
        governed Alert rows ──► webhook (SOAR) + /alerts/stream (SSE) + audit chain
                               │
                               ▼
        investigator: PATHS_BETWEEN, MCP investigate_entity, EXPLAIN_REPLAY

Step 1 — ingest + canonicalize#

Your fetcher pulls from the social API and pushes governed rows through any door (HTTP /ingest/bulk, Kafka, or the Mongo door if your existing pipeline already writes Mongo docs):

curl -X POST 'http://127.0.0.1:9090/ingest/bulk?object_type=SocialPost&purpose=osint' \
  -H 'Authorization: Bearer <token>' -H 'Content-Type: application/json' \
  -d '{"rows":[
    {"_pk":"tw-1","author_handle":"@suspect_42","body":"...","posted_at_ns":1735490000000000000,
     "like_count":3,"retweet_count":1,"lang":"en"},
    {"_pk":"tw-2","author_handle":"@suspect_42","body":"...","posted_at_ns":1735490060000000000,
     "like_count":4120,"retweet_count":980,"lang":"en"}
  ]}'

Enable the social detector pack so SmartIngest canonicalizes handles into the IdentityIndex:

RELATA_DETECT_PACKS=network,contact,social relata serve

Honest scope on Twitter specifically: Relata ships canonical types for Facebook, Instagram, LinkedIn, Snapchat, Telegram, TikTok, and UPI identifiers, plus a generic social pack. There is no dedicated TwitterHandle canonical type today — for an X/Twitter pipeline, either (a) treat @handle as a generic social identifier resolved by the social pack, or (b) add a one-file canonical type in relata-canonical (it's a typed validator pattern — see tiktok_handle.rs as the template). The rest of the engine is identical.

Step 2 — fuse identities across handles and sources#

The same operator running a Telegram channel, a TikTok account, and a Bitcoin address is auto-linked:

PURPOSE 'osint'
-- Resolve everything known about this identity across all sources
SELECT * FROM RESOLVE_IDENTITY('@suspect_42', MODE => 'cluster');
 
-- Is the @handle the same entity as a wallet we already track?
SELECT SAME_IDENTITY('SocialHandle:suspect_42', 'Wallet:0xabc...') AS same;
 
-- Who is this account connected to, within 5 hops?
SELECT * FROM PATHS_BETWEEN('SocialHandle:suspect_42', 'SocialHandle:target', 5);

Identity fusion is deterministic (no LLM guessing) — see Identity.

Step 3 — detect spikes with streaming windows#

Use TUMBLE / HOP / SESSION to compute rolling baselines in SQL, then alert on deviation:

PURPOSE 'osint'
-- Hourly post volume per author, tumbling window
SELECT
  author_handle,
  tumble_start(posted_at_ns, 3600*1000*000000) AS hour_start,
  COUNT(*) AS posts,
  SUM(like_count) AS total_likes
FROM SocialPost
WHERE posted_at_ns > now() - INTERVAL '7' days
GROUP BY author_handle, hour_start
ORDER BY total_likes DESC;

-- Compare this hour to the trailing 30-day baseline for the same author
WITH baseline AS (
  SELECT author_handle,
         AVG(post_count) AS mean_posts,
         STDDEV(post_count) AS sigma_posts
  FROM (
    SELECT author_handle,
           tumble_start(posted_at_ns, 3600*1000*000000) AS h,
           COUNT(*) AS post_count
    FROM SocialPost
    WHERE posted_at_ns > now() - INTERVAL '30' days
    GROUP BY author_handle, h
  ) GROUP BY author_handle
)
SELECT b.author_handle, mean_posts, sigma_posts,
       (mean_posts + 3 * sigma_posts) AS spike_threshold
FROM baseline b
WHERE sigma_posts > 0;

Step 4 — wire a commit-time detection rule#

A rule fires the moment a matching row lands — no poller lag:

curl -X POST http://127.0.0.1:9090/rules \
  -H 'Authorization: Bearer <token>' -H 'Content-Type: application/json' \
  -d '{
    "name": "viral-spike-coordinated-amplification",
    "target_type": "SocialPost",
    "condition_sql": "retweet_count > 500 AND like_count > 2000 AND lang = '\''en'\''",
    "severity": "medium",
    "mitre_technique": "T1585",
    "purpose": "osint"
  }'

Step 5 — the AnomalyRateJob (2σ incident rate)#

For "did the overall incident rate jump?" — the library job compares the current rolling window's incident count against an EWMA-smoothed historical baseline and emits an AnomalyAlert past a configurable σ threshold.

// crates/relata-intelligence/src/anomaly.rs (paraphrased)
pub struct AnomalyRateJobConfig {
    pub baseline_windows: usize,   // e.g. 24 = last 24h
    pub sigma_threshold: f64,      // default 2.0
    pub ewma_decay: f64,           // 0.0–1.0 — how fast baseline forgets
}

Honest wiring note: JobKind::AnomalyRateDetect is library-only today — Relata's JobRegistry has no scheduler consumer for it, so the binary will not auto-run it on a cron. Wire it from your own scheduler (systemd timer, Kubernetes CronJob, tokio::spawn loop) that calls the job against the running store. This is the one piece of the anomaly story that isn't turnkey out of the box.

Step 6 — alert routing + investigation#

# Register your SOAR / Slack / PagerDuty webhook
client.register_webhook("https://soar.example.com/relata",
                        event_types=["alert.high", "alert.medium"])
 
# Stream alerts to a live dashboard
for ev in client.streaming_client.alerts():
    print(ev["severity"], ev["rule_name"], ev["target_id"])

# Investigate via MCP — natural-language + graph
mcp.call_tool("investigate_entity", {"entity_type": "SocialHandle",
                                     "entity_id": "suspect_42",
                                     "purpose": "osint"})
mcp.call_tool("find_threats", {"entity_type": "SocialPost", "purpose": "osint"})

Every alert is a bi-temporal Alert row — backtest detection quality with SELECT ... FROM Alert AS OF '<ts>' before going live. See Detection Rules.

Building a "bias" / distribution-shift detector#

The same machinery, pointed at skew instead of spikes. A bias detector asks: "has the distribution of a signal drifted from its baseline?" — e.g., a sentiment classifier whose positive/negative ratio suddenly shifts, a recommendation pipeline whose demographic skew crosses a threshold, or a feed whose language mix diverges.

PURPOSE 'ml-observability'
-- Track the sentiment-class distribution per day; flag days where the
-- positive share deviates more than 3σ from the trailing 30-day mean.
WITH daily_dist AS (
  SELECT tumble_start(classified_at_ns, 86400*1000*000000) AS day,
         sentiment,
         COUNT(*) AS c
  FROM ClassifiedPost
  WHERE classified_at_ns > now() - INTERVAL '60' days
  GROUP BY day, sentiment
),
positive_share AS (
  SELECT day,
         1.0 * SUM(CASE WHEN sentiment='positive' THEN c ELSE 0 END)
              / NULLIF(SUM(c), 0) AS pos_share
  FROM daily_dist GROUP BY day
),
baseline AS (
  SELECT AVG(pos_share) AS mu, STDDEV(pos_share) AS sigma
  FROM positive_share
  WHERE day < tumble_start(now(), 86400*1000*000000) - INTERVAL '1' days
)
SELECT p.day, p.pos_share, b.mu, b.sigma,
       ABS(p.pos_share - b.mu) / NULLIF(b.sigma, 0) AS z_score
FROM positive_share p CROSS JOIN baseline b
WHERE ABS(p.pos_share - b.mu) > 3 * b.sigma;

Turn that into a governed Alert via a detection rule or a scheduled job, and you have a bias monitor with the same audit chain, provenance, and time-travel as every other Relata signal — defensible to a regulator or an ML-governance review.

Maturity table (honest)#

Tips & takeaways#

• Pick the right primitive per signal. Single-row pattern → detection rule. Rate/volume spike → streaming-window SQL or AnomalyRateJob. Distribution drift → windowed-stats SQL. Multi-signal correlation → pattern library + graph traversal.
• Backtest before going live. SELECT FROM Alert AS OF '<last-month>' against a new rule tells you what it would have fired — tune σ and thresholds on history, not customers.
• Wire AnomalyRateJob from Kubernetes CronJob or a tokio loop — it won't self-schedule today.
• Add a custom canonical type per social platform (one file, see tiktok_handle.rs) if you need checksum-level validation beyond the generic social pack.
• Pair detection with governance. A bias/anomaly alert that drives an automated action should go through PURPOSE + Cedar — so the action is as auditable as the detection.
• Use the LLM interpretation for triage, not decision. nl_query and the LLM summary explain a spike in prose for the on-call analyst; the governed Alert row is the authoritative record.

See also#

• Detection Rules — the commit-driven rule engine
• Jobs, Workflows & Detection — the typed-Job engine
• Streaming & SSE — WATCH subscriptions + /alerts/stream
• OSINT — Identity Fusion — cross-source handle resolution
• Graph Analytics — PATHS_BETWEEN, community detection, link prediction
• For Security Teams — the SIEM-replacement framing