Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Detect fraud, unusual behavior, and anomalies in events or transactions using AI. Use when detecting fraud, flagging suspicious transactions, anomaly detection in logs, spotting unusual user behavior, abuse detection, identifying outliers in data, suspicious activity monitoring, fraud scoring, unusual pattern detection, flagging account takeover attempts, detecting bot traffic, abnormal usage patterns, security event triage, risk scoring with AI.
Build an AI Anomaly Detector
Build an AI anomaly detector with DSPy - define what normal looks like, score events for severity, route by risk level, and explain findings to human reviewers.
Step 1: Understand the detection task
Ask the user:
What events are you analyzing? (transactions, logins, API calls, server logs, user actions, etc.)
What does "normal" look like? (Do you have historical baselines? Average values? Known-good patterns?)
What counts as suspicious? (Frequency spikes, unusual amounts, geographic outliers, time-of-day mismatches, etc.)
What action fires on detection, and how strict is your false-positive tolerance? (Actions: alert, block, escalate, log for review. Tolerance: low = only flag high-confidence anomalies; high = cast a wide net and filter later.)
The answers determine severity thresholds, routing logic, and how much baseline context to include.
When NOT to use AI anomaly detection
High-volume numeric time series — millions of events/second with simple numeric signals (CPU, latency, request rate). Use statistical methods instead - z-score, EWMA, isolation forest, or Prometheus alerting rules.
Simple threshold rules — "flag any transaction over $10,000" does not need an LM. Write a rule.
Real-time sub-10ms requirements — LM calls add latency. Use rule-based pre-filters and only invoke the LM on candidates.
When you have millions of events per second — LM calls cost money. Pre-filter with cheap heuristics, then use AI only on flagged candidates.
Step 2: Build baseline construction
Summarize historical "normal" behavior into a compact string the LM can reason against. A good baseline summary describes typical patterns, ranges, and context.
import dspy
lm = dspy.LM("openai/gpt-4o-mini") # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)
classSummarizeBaseline(dspy.Signature):
"""Summarize the normal behavior pattern from historical event data.
Produce a concise description of what typical events look like - include
typical values, frequencies, time patterns, and common attributes."""
historical_events: str = dspy.InputField(
desc="Sample of recent normal events, formatted as JSON or CSV"
)
baseline_summary: str = dspy.OutputField(
desc="Concise description of normal behavior - typical ranges, patterns, and context"
)
baseline_summarizer = dspy.ChainOfThought(SummarizeBaseline)
For many use cases, the baseline summary can be constructed once per period (daily, hourly) and cached rather than recomputed per event.
Step 3: Build the anomaly scorer
The core signature takes an event and the baseline summary, and outputs a severity level with a specific explanation.
from typing importLiteral
SEVERITY_LEVELS = ["normal", "low", "medium", "high", "critical"]
classScoreAnomaly(dspy.Signature):
"""Analyze an event against the normal baseline and determine if it is anomalous.
Consider all dimensions - amount, timing, location, frequency, sequence, and context.
Severity reflects both how unusual the event is AND the potential impact if malicious."""
event: str = dspy.InputField(
desc="The event to analyze, as a JSON object or structured text"
)
baseline_summary: str = dspy.InputField(
desc="Description of normal behavior for this type of event"
)
severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField(
desc="Anomaly severity - normal (expected), low (minor deviation), ""medium (notable deviation), high (strong anomaly signal), ""critical (likely fraud or attack, needs immediate action)"
)
explanation: str = dspy.OutputField(
desc="Specific explanation citing concrete details - what exactly is unusual, ""how it deviates from the baseline, and what the risk is. ""Not vague ('looks suspicious') but specific ('transaction amount $4,800 ""is 12x the users 30-day average of $400, combined with a new device ""and 3am local time')."
)
anomaly_score: float = dspy.OutputField(
desc="Confidence score from 0.0 (clearly normal) to 1.0 (clearly anomalous)"
)
anomaly_scorer = dspy.ChainOfThought(ScoreAnomaly)
Step 4: Full detection pipeline module
Combine baseline construction and anomaly scoring into a single reusable module.
classAnomalyDetector(dspy.Module):
def__init__(self):
self.baseline_summarizer = dspy.ChainOfThought(SummarizeBaseline)
self.scorer = dspy.ChainOfThought(ScoreAnomaly)
self._cached_baseline = Nonedefset_baseline(self, historical_events: str):
"""Pre-compute the baseline summary from historical data."""
result = self.baseline_summarizer(historical_events=historical_events)
self._cached_baseline = result.baseline_summary
returnself._cached_baseline
defforward(self, event: str, baseline_summary: str = None):
baseline = baseline_summary orself._cached_baseline
ifnot baseline:
raise ValueError("No baseline set. Call set_baseline() first or pass baseline_summary.")
returnself.scorer(event=event, baseline_summary=baseline)
detector = AnomalyDetector()
# Set baseline once from recent history
detector.set_baseline("""
Recent 30-day transactions:
- Average amount: $412, std dev: $180, max: $1,200
- Typical locations: New York, Chicago, LA
- Typical hours: 9am-9pm local time
- Devices: 1-2 known devices per user
- Frequency: 3-8 transactions/week per user
""")
result = detector(event='{"amount": 4800, "location": "Lagos", "hour": 3, "device": "unknown"}')
print(f"Severity: {result.severity}")
print(f"Score: {result.anomaly_score:.2f}")
print(f"Explanation: {result.explanation}")
Step 5: Severity scoring with confidence-based routing
Route events automatically based on severity, and escalate only when confidence is high enough.
defroute_anomaly(result) -> dict:
"""Route a scored anomaly to the appropriate action."""
routing = {
"normal": {"action": "dismiss", "notify": False, "block": False},
"low": {"action": "log", "notify": False, "block": False},
"medium": {"action": "queue", "notify": True, "block": False},
"high": {"action": "alert", "notify": True, "block": False},
"critical": {"action": "escalate", "notify": True, "block": True},
}
route = routing[result.severity]
# Downgrade routing if confidence is lowif result.anomaly_score < 0.6and result.severity in ("high", "critical"):
route = routing["medium"] # reduce to queue for human review
route["confidence_downgraded"] = Truereturn {**route, "severity": result.severity, "score": result.anomaly_score}
Severity to action mapping
Severity
Score range
Default action
Blocks transaction
normal
0.0 - 0.2
Dismiss silently
No
low
0.2 - 0.4
Log for review
No
medium
0.4 - 0.6
Queue for analyst
No
high
0.6 - 0.8
Alert on-call
No
critical
0.8 - 1.0
Escalate + block
Yes
Adjust these thresholds based on your false-positive tolerance.
Step 6: Explanation generation for human reviewers
Explanations are only useful if they are specific. Force the LM to cite concrete numbers and deviations by adding a dedicated explanation signature for high-severity events.
classExplainAnomaly(dspy.Signature):
"""Generate a reviewer-ready explanation of why this event is anomalous.
Write for a human analyst who needs to decide quickly. Cite specific numbers,
list each anomalous dimension separately, and state the risk clearly."""
event: str = dspy.InputField(desc="The flagged event")
baseline_summary: str = dspy.InputField(desc="Normal behavior baseline")
severity: str = dspy.InputField(desc="Assigned severity level")
reviewer_explanation: str = dspy.OutputField(
desc="Bullet-point explanation for human reviewer: what deviates, by how much, ""and what action is recommended. Must cite specific values from the event."
)
explainer = dspy.ChainOfThought(ExplainAnomaly)
# Only invoke for high/critical — saves costif result.severity in ("high", "critical"):
detail = explainer(
event=event,
baseline_summary=baseline,
severity=result.severity,
)
print(detail.reviewer_explanation)
Step 7: Alert routing by confidence
Batch multiple events together (same user, same time window) before scoring to give the LM session-level context.
import json
classScoreSession(dspy.Signature):
"""Analyze a sequence of events from the same user session for anomalies.
Consider the pattern across events, not just individual events in isolation.
Rapid-fire actions, escalating amounts, or device switches mid-session
are signals invisible when events are scored individually."""
session_events: str = dspy.InputField(
desc="JSON array of events from the same user/session, in chronological order"
)
baseline_summary: str = dspy.InputField(desc="Normal behavior baseline for this user")
severity: Literal[tuple(SEVERITY_LEVELS)] = dspy.OutputField()
explanation: str = dspy.OutputField(
desc="What pattern across the session is anomalous, not just individual events"
)
anomaly_score: float = dspy.OutputField()
session_scorer = dspy.ChainOfThought(ScoreSession)
defscore_user_window(events: list[dict], baseline: str, window_minutes: int = 15):
"""Group events into time windows and score as sessions."""# Sort by timestamp, group into windows
events_json = json.dumps(events, indent=2)
return session_scorer(session_events=events_json, baseline_summary=baseline)
Step 8: Evaluate and optimize
from dspy.evaluate import Evaluate
# Build a labeled dataset - events with known ground truth# label: 0 = normal, 1 = anomalous
labeled_events = [
dspy.Example(
event='{"amount": 4800, "location": "Lagos", "hour": 3, "device": "unknown"}',
baseline_summary="Avg $400, NY/Chicago, 9am-9pm, known devices",
severity="critical",
).with_inputs("event", "baseline_summary"),
dspy.Example(
event='{"amount": 380, "location": "New York", "hour": 14, "device": "iPhone-known"}',
baseline_summary="Avg $400, NY/Chicago, 9am-9pm, known devices",
severity="normal",
).with_inputs("event", "baseline_summary"),
# ... more examples
]
trainset = labeled_events[:int(len(labeled_events) * 0.8)]
devset = labeled_events[int(len(labeled_events) * 0.8):]
defanomaly_metric(example, prediction, trace=None):
"""Measure exact severity match, or partial credit for adjacent severities."""if prediction.severity == example.severity:
return1.0# Adjacent severity is a partial match (e.g., predicted high vs actual critical)
order = {s: i for i, s inenumerate(SEVERITY_LEVELS)}
distance = abs(order[prediction.severity] - order[example.severity])
returnmax(0.0, 1.0 - distance * 0.3)
evaluator = Evaluate(
devset=devset,
metric=anomaly_metric,
num_threads=4,
display_progress=True,
display_table=5,
)
score = evaluator(anomaly_scorer)
print(f"Baseline accuracy: {score:.1f}%")
Optimize with BootstrapFewShot
optimizer = dspy.BootstrapFewShot(
metric=anomaly_metric,
max_bootstrapped_demos=4,
)
optimized_detector = optimizer.compile(anomaly_scorer, trainset=trainset)
# Re-evaluate
score = evaluator(optimized_detector)
print(f"Optimized accuracy: {score:.1f}%")
# Typical result: ~65% exact severity match baseline → ~82% after BootstrapFewShot with 4 demos# Save
optimized_detector.save("anomaly_scorer.json")
False positive rate metric
deffalse_positive_rate(examples, predictions):
"""Compute FPR - normal events flagged as anomalous."""
normals = [(e, p) for e, p inzip(examples, predictions) if e.severity == "normal"]
ifnot normals:
return0.0
flagged = sum(1for _, p in normals if p.severity != "normal")
return flagged / len(normals)
Key patterns
Pattern
When to use
Single event + baseline
Simple fraud scoring, log triage
Session window scoring
Account takeover, multi-step attacks
Two-stage (pre-filter + LM)
High-volume streams - rule-based pre-filter, LM on candidates
Cached baseline
Baselines are stable - build once per period, reuse
Explanation-on-demand
Cost savings - only generate detailed explanations for high/critical
Gotchas
Claude flags every unusual event as anomalous without baseline context — always provide a baseline_summary in the signature. Without it, the model has no reference point and defaults to flagging anything non-trivial.
Claude outputs binary anomaly/not-anomaly instead of severity levels — use Literal with five graduated severity levels (normal, low, medium, high, critical) so downstream routing can take proportional action.
Claude uses dspy.Assert/dspy.Suggest for severity validation — use dspy.Refine with a reward function that checks severity is one of the valid values and the explanation cites specific numbers.
Claude generates vague explanations such as "this looks suspicious" or "unusual activity detected" — add an explicit desc on the explanation field requiring concrete deviations, specific values, and a risk statement.
Claude processes events independently and misses session-level patterns — batch related events (same user, same time window) into a session before scoring. Account takeover and credential stuffing are only visible at the session level.
Cross-references
Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>
Need fraud scores instead of severity categories? See /ai-scoring
Measure and improve detection accuracy - see /ai-improving-accuracy
Generate labeled training data for anomalies - see /ai-generating-data
Add reasoning before severity classification - see /dspy-chain-of-thought
Iterative refinement with feedback to fix wrong severity outputs - see /dspy-refine
Sample N severity predictions and pick the best one - see /dspy-best-of-n
Install /ai-do if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do
Additional resources
For worked examples (transaction fraud, user behavior, log anomalies), see examples.md
For DSPy API reference (constructor signatures, parameter tables, BestOfN/Refine/Evaluate), see reference.md
