src/mimir_lens/change_series.rs
//! **The per-node change TIME-SERIES over `git_heat_facts`** — the shared substrate
//! the change/anomaly lens family folds.
//!
//! `git_heat_facts` is append-only and carries a `snapshot_id` + `ts_micros` per scan
//! ([`crate::warehouse::iceberg_schema::git_heat_facts`]): every `nornir knowledge scan`
//! re-walks the repo's git history and appends a fresh generation of per-file rows. So
//! the rows of ONE `(repo, file)` node across snapshots ARE its change time-series —
//! the signal-processing substrate the [`super::churn_filter`] frequency filter and the
//! [`super::anomaly`] outlier detector read.
//!
//! This is deliberately NOT run through [`super::rows::latest_snapshot_rows`] (which the
//! other lenses use): those lenses want the newest state, these want the whole history.
//!
//! **Honest about the signal.** `git_heat_facts` aggregates PER FILE, per scan. It
//! records, for each file, `commits_total` (lifetime), `commits_30d`, `commits_90d`
//! (rolling windows anchored at the scan `ts`). It does NOT retain the per-commit
//! file-SET, so true same-commit co-change is not recoverable from this table — the
//! coupling signal in [`super::anomaly`] is a documented churn-correlation PROXY, not
//! same-commit coupling. The per-interval change count we build here is the delta of
//! `commits_total` between consecutive snapshots — the commits added to that file's
//! history in the interval between two scans.
use std::collections::BTreeMap;
use super::rows::{Table, ts_micros};
/// One `(repo, file)` node's churn AS OF one warehouse snapshot.
#[derive(Debug, Clone, PartialEq)]
pub struct HeatPoint {
pub snapshot_id: String,
pub ts_micros: i64,
/// Lifetime commits touching the file, as of this scan (cumulative, monotone).
pub commits_total: i64,
pub commits_30d: i64,
pub commits_90d: i64,
pub authors_total: i64,
}
/// The full change-series of ONE node `(repo, file)`: its churn at each snapshot,
/// ascending by `ts_micros` (ties broken by `snapshot_id`).
#[derive(Debug, Clone, PartialEq)]
pub struct NodeSeries {
pub repo: String,
pub file: String,
pub points: Vec<HeatPoint>,
}
impl NodeSeries {
/// The newest snapshot's point — the node's current churn.
pub fn latest(&self) -> Option<&HeatPoint> {
self.points.last()
}
/// The recent change frequency: the latest snapshot's `commits_90d` (the
/// convention the risk lens uses). `0` when the node has no points.
pub fn recent_frequency(&self) -> i64 {
self.latest().map(|p| p.commits_90d).unwrap_or(0)
}
/// **Per-interval change count** — the number of commits added to this file's
/// history between consecutive snapshots (`Δ commits_total`). The first
/// snapshot's increment is its own `commits_total` (the history it first
/// captured). A negative delta (a re-scan that saw fewer commits, e.g. a rebase)
/// is clamped to 0. This is the discrete change-rate series the anomaly detector
/// runs z-score / spike math over.
pub fn increments(&self) -> Vec<i64> {
let mut out = Vec::with_capacity(self.points.len());
let mut prev = 0i64;
for (i, p) in self.points.iter().enumerate() {
let inc = if i == 0 {
p.commits_total
} else {
(p.commits_total - prev).max(0)
};
out.push(inc);
prev = p.commits_total;
}
out
}
}
/// **Parse `git_heat_facts` (ALL snapshots) into per-node change series.**
///
/// Groups rows by `(repo, file)` and sorts each node's points ascending by
/// `(ts_micros, snapshot_id)` so the series is a total temporal order. Rows missing
/// the required columns are skipped (an empty table yields an empty vec — the caller
/// reports that via the evidence note, never as "no change").
pub fn parse_series(t: &Table) -> Vec<NodeSeries> {
let c = t.cursor(&[
"repo",
"file",
"snapshot_id",
"ts_micros",
"commits_total",
"commits_30d",
"commits_90d",
"authors_total",
]);
let mut by_node: BTreeMap<(String, String), Vec<HeatPoint>> = BTreeMap::new();
for r in c.rows() {
let repo = c.get(r, 0).to_string();
let file = c.get(r, 1).to_string();
if repo.is_empty() && file.is_empty() {
continue;
}
let point = HeatPoint {
snapshot_id: c.get(r, 2).to_string(),
ts_micros: ts_micros(c.get(r, 3)),
commits_total: c.get(r, 4).trim().parse().unwrap_or(0),
commits_30d: c.get(r, 5).trim().parse().unwrap_or(0),
commits_90d: c.get(r, 6).trim().parse().unwrap_or(0),
authors_total: c.get(r, 7).trim().parse().unwrap_or(0),
};
by_node.entry((repo, file)).or_default().push(point);
}
by_node
.into_iter()
.map(|((repo, file), mut points)| {
points.sort_by(|a, b| {
a.ts_micros
.cmp(&b.ts_micros)
.then_with(|| a.snapshot_id.cmp(&b.snapshot_id))
});
NodeSeries { repo, file, points }
})
.collect()
}
/// Population mean + standard deviation of a sample. `(0.0, 0.0)` for an empty slice.
/// Population (not sample) std so a two-point baseline is still defined.
pub fn mean_std(xs: &[f64]) -> (f64, f64) {
if xs.is_empty() {
return (0.0, 0.0);
}
let n = xs.len() as f64;
let mean = xs.iter().sum::<f64>() / n;
let var = xs.iter().map(|x| (x - mean) * (x - mean)).sum::<f64>() / n;
(mean, var.sqrt())
}
/// **EWMA (exponentially-weighted moving average) of a series.** The classic
/// change-frequency anomaly-detection baseline: recent history weighs more than old.
/// `alpha` in `(0,1]` — higher = more responsive. Returns the EWMA after the last
/// point (the baseline the newest observation is judged against). Empty ⇒ `0.0`.
pub fn ewma(xs: &[f64], alpha: f64) -> f64 {
let mut it = xs.iter();
let Some(&first) = it.next() else { return 0.0 };
let mut s = first;
for &x in it {
s = alpha * x + (1.0 - alpha) * s;
}
s
}
#[cfg(test)]
mod tests {
use super::*;
fn heat_table(rows: &[&[&str]]) -> Table {
let cols = [
"repo",
"file",
"snapshot_id",
"ts_micros",
"commits_total",
"commits_30d",
"commits_90d",
"authors_total",
];
Table::new(
cols.iter().map(|s| s.to_string()).collect(),
rows.iter()
.map(|r| r.iter().map(|s| s.to_string()).collect())
.collect(),
)
}
#[test]
fn series_are_grouped_by_node_and_sorted_by_time() {
// Two snapshots of one file, appended out of order.
let t = heat_table(&[
&["korp", "src/a.rs", "s2", "200", "10", "2", "5", "1"],
&["korp", "src/a.rs", "s1", "100", "6", "1", "3", "1"],
&["korp", "src/b.rs", "s1", "100", "1", "0", "0", "1"],
]);
let series = parse_series(&t);
assert_eq!(series.len(), 2, "two distinct nodes");
let a = series.iter().find(|s| s.file == "src/a.rs").unwrap();
assert_eq!(a.points.len(), 2);
// ascending by ts — s1 (100) before s2 (200).
assert_eq!(a.points[0].snapshot_id, "s1");
assert_eq!(a.points[1].snapshot_id, "s2");
assert_eq!(a.latest().unwrap().commits_total, 10);
assert_eq!(a.recent_frequency(), 5); // latest commits_90d
}
#[test]
fn increments_are_the_delta_of_lifetime_commits() {
let t = heat_table(&[
&["korp", "f", "s1", "100", "6", "0", "0", "1"],
&["korp", "f", "s2", "200", "10", "0", "0", "1"],
&["korp", "f", "s3", "300", "11", "0", "0", "1"],
]);
let a = &parse_series(&t)[0];
// first = its own total (6), then Δ (10-6=4), then Δ (11-10=1).
assert_eq!(a.increments(), vec![6, 4, 1]);
}
#[test]
fn a_rebased_smaller_total_clamps_to_zero_not_negative() {
let t = heat_table(&[
&["korp", "f", "s1", "100", "10", "0", "0", "1"],
&["korp", "f", "s2", "200", "7", "0", "0", "1"],
]);
let a = &parse_series(&t)[0];
assert_eq!(
a.increments(),
vec![10, 0],
"a shrinking history is not a negative change"
);
}
#[test]
fn mean_std_and_ewma_are_the_expected_stats() {
let (m, s) = mean_std(&[2.0, 4.0, 6.0]);
assert!((m - 4.0).abs() < 1e-9);
// population std of {2,4,6} = sqrt(8/3) ≈ 1.632993
assert!((s - (8.0f64 / 3.0).sqrt()).abs() < 1e-9);
// EWMA of a flat series is the flat value.
assert!((ewma(&[3.0, 3.0, 3.0], 0.5) - 3.0).abs() < 1e-9);
}
}