///|
/// Operational policy for a monitored numeric stream.
pub struct MonitorPolicy {
window : Int
alert_threshold : Double
drift_threshold : Double
quality_threshold : Double
minimum_batch : Int
}
///|
pub struct MonitorEvent {
timestamp : Int
kind : String
score : Double
severity : String
message : String
}
///|
pub struct MonitorState {
policy : MonitorPolicy
mut baseline : Array[Double]
mut latest : Array[Double]
mut events : Array[MonitorEvent]
mut batches : Int
mut healthy_batches : Int
}
///|
pub struct MonitorSnapshot {
batches : Int
healthy_batches : Int
health_rate : Double
latest_quality : Double
latest_drift : Double
alert_count : Int
events : Array[MonitorEvent]
}
///|
pub fn monitor_policy(
window : Int,
alert_threshold : Double,
drift_threshold : Double,
quality_threshold : Double,
minimum_batch : Int,
) -> MonitorPolicy {
{
window: if window < 2 {
2
} else {
window
},
alert_threshold: if alert_threshold < 0.0 {
-alert_threshold
} else {
alert_threshold
},
drift_threshold: if drift_threshold < 0.0 {
-drift_threshold
} else {
drift_threshold
},
quality_threshold: transform_clip(quality_threshold, 0.0, 1.0),
minimum_batch: if minimum_batch < 1 {
1
} else {
minimum_batch
},
}
}
///|
pub fn monitor_state(
policy : MonitorPolicy,
baseline : Array[Double],
) -> MonitorState {
{
policy,
baseline: baseline.copy(),
latest: [],
events: [],
batches: 0,
healthy_batches: 0,
}
}
///|
fn monitor_event(
timestamp : Int,
kind : String,
score : Double,
severity : String,
message : String,
) -> MonitorEvent {
{ timestamp, kind, score, severity, message }
}
///|
pub fn monitor_event_count(state : MonitorState) -> Int {
state.events.length()
}
///|
pub fn monitor_alert_count(state : MonitorState) -> Int {
let mut count = 0
for event in state.events {
if event.severity == "alert" || event.severity == "critical" {
count += 1
}
}
count
}
///|
pub fn monitor_append_event(state : MonitorState, event : MonitorEvent) -> Unit {
state.events.push(event)
}
///|
pub fn monitor_quality_event(
state : MonitorState,
batch : Array[Double],
timestamp : Int,
) -> Double {
let report = quality_report(batch, quality_default_rule())
let score = report.quality_score
if score < state.policy.quality_threshold {
monitor_append_event(
state,
monitor_event(
timestamp, "quality", score, "alert", "quality score below policy",
),
)
} else {
monitor_append_event(
state,
monitor_event(
timestamp, "quality", score, "info", "quality within policy",
),
)
}
score
}
///|
pub fn monitor_drift_event(
state : MonitorState,
batch : Array[Double],
timestamp : Int,
) -> Double {
if state.baseline.length() == 0 || batch.length() == 0 {
return 0.0
}
let rule = drift_rule(
state.policy.drift_threshold,
state.policy.drift_threshold,
state.policy.drift_threshold,
state.policy.drift_threshold,
10,
)
let report = drift_report(state.baseline, batch, rule)
let score = report.drift_score
if report.drifted {
monitor_append_event(
state,
monitor_event(
timestamp, "drift", score, "alert", "distribution drift detected",
),
)
} else {
monitor_append_event(
state,
monitor_event(timestamp, "drift", score, "info", "distribution stable"),
)
}
score
}
///|
pub fn monitor_outlier_event(
state : MonitorState,
batch : Array[Double],
timestamp : Int,
) -> Double {
let alerts = streaming_alerts(
batch,
state.policy.window,
state.policy.alert_threshold,
)
let rate = streaming_outlier_rate(alerts)
if rate > 0.0 {
monitor_append_event(
state,
monitor_event(
timestamp,
"outlier",
rate,
if rate > 0.1 {
"critical"
} else {
"alert"
},
"online outliers detected",
),
)
}
rate
}
///|
pub fn monitor_batch(
state : MonitorState,
batch : Array[Double],
timestamp : Int,
) -> MonitorSnapshot {
state.latest = batch.copy()
state.batches += 1
let before = state.events.length()
let quality = monitor_quality_event(state, batch, timestamp)
let drift = monitor_drift_event(state, batch, timestamp)
let _ = monitor_outlier_event(state, batch, timestamp)
let new_events = state.events.length() - before
if quality >= state.policy.quality_threshold && drift < 1.0 && new_events <= 1 {
state.healthy_batches += 1
}
monitor_snapshot(state)
}
///|
pub fn monitor_snapshot(state : MonitorState) -> MonitorSnapshot {
let latest_quality = if state.latest.length() == 0 {
1.0
} else {
quality_report(state.latest, quality_default_rule()).quality_score
}
let latest_drift = if state.baseline.length() == 0 ||
state.latest.length() == 0 {
0.0
} else {
drift_report(state.baseline, state.latest, drift_default_rule()).drift_score
}
{
batches: state.batches,
healthy_batches: state.healthy_batches,
health_rate: if state.batches == 0 {
1.0
} else {
state.healthy_batches.to_double() / state.batches.to_double()
},
latest_quality,
latest_drift,
alert_count: monitor_alert_count(state),
events: state.events.copy(),
}
}
///|
pub fn monitor_snapshot_vector(snapshot : MonitorSnapshot) -> Array[Double] {
[
snapshot.batches.to_double(),
snapshot.healthy_batches.to_double(),
snapshot.health_rate,
snapshot.latest_quality,
snapshot.latest_drift,
snapshot.alert_count.to_double(),
]
}
///|
pub fn monitor_events(state : MonitorState) -> Array[MonitorEvent] {
state.events.copy()
}
///|
pub fn monitor_event_kinds(state : MonitorState) -> Array[String] {
let result = []
for event in state.events {
result.push(event.kind)
}
result
}
///|
pub fn monitor_event_scores(state : MonitorState) -> Array[Double] {
let result = []
for event in state.events {
result.push(event.score)
}
result
}
///|
pub fn monitor_event_lines(state : MonitorState) -> Array[String] {
let lines = []
for event in state.events {
lines.push(
event.timestamp.to_string() +
"|" +
event.kind +
"|" +
event.score.to_string() +
"|" +
event.severity +
"|" +
event.message,
)
}
lines
}
///|
pub fn monitor_event_string(state : MonitorState) -> String {
monitor_event_lines(state).join("\n")
}
///|
pub fn monitor_snapshot_lines(snapshot : MonitorSnapshot) -> Array[String] {
[
"batches=" + snapshot.batches.to_string(),
"healthy_batches=" + snapshot.healthy_batches.to_string(),
"health_rate=" + snapshot.health_rate.to_string(),
"latest_quality=" + snapshot.latest_quality.to_string(),
"latest_drift=" + snapshot.latest_drift.to_string(),
"alert_count=" + snapshot.alert_count.to_string(),
]
}
///|
pub fn monitor_snapshot_string(snapshot : MonitorSnapshot) -> String {
monitor_snapshot_lines(snapshot).join("\n")
}
///|
pub fn monitor_is_healthy(
snapshot : MonitorSnapshot,
threshold : Double,
) -> Bool {
snapshot.health_rate >= threshold && snapshot.alert_count == 0
}
///|
pub fn monitor_reset(state : MonitorState) -> Unit {
state.latest = []
state.events = []
state.batches = 0
state.healthy_batches = 0
}
///|
pub fn monitor_update_baseline(
state : MonitorState,
baseline : Array[Double],
) -> Unit {
state.baseline = baseline.copy()
}
///|
pub fn monitor_batch_scores(
state : MonitorState,
batches : Array[Array[Double]],
) -> Array[Double] {
let result = []
for index = 0; index < batches.length(); index = index + 1 {
let snapshot = monitor_batch(state, batches[index], index)
result.push(snapshot.health_rate)
}
result
}
///|
pub fn monitor_summary(state : MonitorState) -> Array[Double] {
monitor_snapshot_vector(monitor_snapshot(state))
}
///|
pub fn monitor_reproducible(
policy : MonitorPolicy,
baseline : Array[Double],
batch : Array[Double],
) -> Bool {
let left = monitor_state(policy, baseline)
let right = monitor_state(policy, baseline)
let left_snapshot = monitor_batch(left, batch, 0)
let right_snapshot = monitor_batch(right, batch, 0)
monitor_snapshot_vector(left_snapshot) ==
monitor_snapshot_vector(right_snapshot) &&
monitor_event_kinds(left) == monitor_event_kinds(right)
}
///|
pub fn monitor_severity_score(state : MonitorState) -> Double {
let mut score = 1.0
for event in state.events {
if event.severity == "critical" {
score -= 0.25
} else if event.severity == "alert" {
score -= 0.1
}
}
transform_clip(score, 0.0, 1.0)
}
///|
pub fn monitor_to_report(
state : MonitorState,
title : String,
) -> AnalyticsReport {
let snapshot = monitor_snapshot(state)
let report = analytics_report(title, "stream")
let metrics = [
report_metric(
"health_rate",
snapshot.health_rate,
"ratio",
snapshot.health_rate >= state.policy.quality_threshold,
),
report_metric(
"quality",
snapshot.latest_quality,
"ratio",
snapshot.latest_quality >= state.policy.quality_threshold,
),
report_metric(
"drift",
snapshot.latest_drift,
"score",
snapshot.latest_drift < 1.0,
),
report_metric(
"alerts",
snapshot.alert_count.to_double(),
"count",
snapshot.alert_count == 0,
),
]
report_add_section(
report,
report_section(
"monitor",
if monitor_severity_score(state) >= 0.8 {
"healthy"
} else {
"review"
},
monitor_severity_score(state),
metrics,
monitor_event_lines(state),
),
)
}