k4bench.analysis.trend

k4bench.analysis.trend

Pure, Streamlit-free trend aggregation over benchmark run directories.

Each build_*_trend function takes the already date-windowed set of cached run directories (see :func:k4bench.remote.fetch_runs_windowed) and returns a long-form DataFrame keyed by run_id / x_date (nightly tag), or None when no data could be loaded. The dashboard wraps these in @st.cache_data (see dashboard/data.py); the nightly regression report calls them directly from CI, which is why they must not import Streamlit.

upper_trimmed_mean

upper_trimmed_mean(times: ndarray, fraction: float = TRIM_FRACTION) -> float | None

Mean of times with the slowest fraction of events dropped.

A run total is a sum of per-event costs, and where that distribution is heavy-tailed a handful of events carry a disproportionate share of it. That makes the plain mean a statistic about the tail as much as about the software. Dropping the slow tail leaves a statistic about the typical event, which moves when the typical event's cost moves and not otherwise.

Deliberately a complement to the untrimmed mean, never a replacement: the same tail that destabilises the mean is where a tail-confined regression would show up first, so a detector that only watched this would be blind to exactly the most interesting kind of change.

Returns None for fewer than :data:MIN_TRIM_EVENTS events, where the trim would drop nothing and the value would merely duplicate the mean.

Source code in k4bench/analysis/trend.py

def upper_trimmed_mean(times: np.ndarray, fraction: float = TRIM_FRACTION) -> float | None:
    """Mean of *times* with the slowest *fraction* of events dropped.

    A run total is a sum of per-event costs, and where that distribution is
    heavy-tailed a handful of events carry a disproportionate share of it. That
    makes the plain mean a statistic about the tail as much as about the
    software. Dropping the slow tail leaves a statistic about the *typical*
    event, which moves when the typical event's cost moves and not otherwise.

    Deliberately a complement to the untrimmed mean, never a replacement: the
    same tail that destabilises the mean is where a tail-confined regression
    would show up first, so a detector that only watched this would be blind to
    exactly the most interesting kind of change.

    Returns ``None`` for fewer than :data:`MIN_TRIM_EVENTS` events, where the
    trim would drop nothing and the value would merely duplicate the mean.
    """
    n = len(times)
    if n < MIN_TRIM_EVENTS:
        return None
    n_drop = int(n * fraction)
    if n_drop <= 0:
        return None
    kept = np.sort(times)[: n - n_drop]
    return float(kept.mean())

theil_sen_slope

theil_sen_slope(x, y) -> float | None

Median of the pairwise slopes (y_j - y_i) / (x_j - x_i) over i < j.

The Theil–Sen estimator: robust to roughly 29% outlying points, where least squares or last - first follow a single noisy sample. Pairs with equal x carry no slope and are skipped; non-finite points are dropped.

Above :data:MAX_SLOPE_EVENTS points, the estimate is taken over :data:MAX_SLOPE_EVENTS evenly spaced points (first and last included) in x order, each keeping its own x. The cost is then bounded for any input size and the result stays deterministic.

Returns None below :data:MIN_SLOPE_EVENTS finite points or when no pair has distinct x.

Source code in k4bench/analysis/trend.py

def theil_sen_slope(x, y) -> float | None:
    """Median of the pairwise slopes ``(y_j - y_i) / (x_j - x_i)`` over ``i < j``.

    The Theil–Sen estimator: robust to roughly 29% outlying points, where
    least squares or ``last - first`` follow a single noisy sample. Pairs with
    equal ``x`` carry no slope and are skipped; non-finite points are dropped.

    Above :data:`MAX_SLOPE_EVENTS` points, the estimate is taken over
    :data:`MAX_SLOPE_EVENTS` evenly spaced points (first and last included) in
    ``x`` order, each keeping its own ``x``. The cost is then bounded for any
    input size and the result stays deterministic.

    Returns ``None`` below :data:`MIN_SLOPE_EVENTS` finite points or when no
    pair has distinct ``x``.
    """
    x = np.asarray(x, dtype=float)
    y = np.asarray(y, dtype=float)
    finite = np.isfinite(x) & np.isfinite(y)
    x, y = x[finite], y[finite]
    n = len(x)
    if n < MIN_SLOPE_EVENTS:
        return None
    order = np.argsort(x, kind="stable")
    x, y = x[order], y[order]
    if n > MAX_SLOPE_EVENTS:
        keep = np.unique(np.linspace(0, n - 1, MAX_SLOPE_EVENTS).round().astype(int))
        x, y = x[keep], y[keep]
    i, j = np.triu_indices(len(x), k=1)
    dx = x[j] - x[i]
    distinct = dx != 0
    if not distinct.any():
        return None
    return float(np.median((y[j] - y[i])[distinct] / dx[distinct]))

parse_run_dir

parse_run_dir(run_dir: Path) -> dict

Extract run metadata from a date-level run directory.

Expected path structure::

{detector}/{platform}/{stack}/{sample}/{YYYY-MM-DD}/

Prefers run_info.json when present; falls back to inferring fields from the directory path.

Source code in k4bench/analysis/trend.py

def parse_run_dir(run_dir: Path) -> dict:
    """Extract run metadata from a date-level run directory.

    Expected path structure::

        {detector}/{platform}/{stack}/{sample}/{YYYY-MM-DD}/

    Prefers ``run_info.json`` when present; falls back to inferring fields
    from the directory path.
    """
    def _unknown_meta(run_dir: Path) -> dict:
        """Return a minimal metadata dict when no structured info is available."""
        return {
            "run_dir":          str(run_dir),
            "run_date":         pd.to_datetime(run_dir.name[:10], errors="coerce"),
            "platform":         "unknown",
            "k4h_release":      "unknown",
            "k4h_release_date": pd.NaT,
            "k4h_packages":     {},
            "sample":           "unknown",
            "xml_path":         None,
            "github_run_url":   None,
            "commit_sha":       None,
            "n_events":         None,
            "random_seed":      None,
            "status":           None,
            "failed_configs":   [],
            "configured_labels": None,
            "machine_consistent": None,
        }

    info_path = run_dir / "run_info.json"
    if info_path.exists():
        try:
            with open(info_path) as fh:
                info = json.load(fh)
            k4h_release = info.get("k4h_release") or ""
            k4h_release_date_raw = info.get("k4h_release_date")
            if not k4h_release_date_raw and k4h_release:
                m = re.search(r"(\d{4}-\d{2}-\d{2})", k4h_release)
                k4h_release_date_raw = m.group(1) if m else None
            return {
                "run_dir":          str(run_dir),
                "run_date":         pd.to_datetime(info.get("date"), errors="coerce"),
                "platform":         info.get("platform", "unknown") or "unknown",
                "k4h_release":      k4h_release,
                "k4h_release_date": pd.to_datetime(k4h_release_date_raw, errors="coerce"),
                # Per-package upstream commits (see k4bench.provenance). Absent for
                # runs predating provenance capture, and for any night whose stack
                # could not be read — an empty map means "unknown", never "unchanged".
                "k4h_packages":     info.get("k4h_packages") or {},
                "sample":           info.get("sample", "unknown") or "unknown",
                # The compact file this run loaded, relative to $K4GEO. Absent
                # for every run benchmarked before it was captured, which reads
                # as "unknown" — never as "this run loads nothing".
                "xml_path":         info.get("xml_path"),
                "github_run_url":   info.get("github_run_url"),
                "commit_sha":       info.get("commit_sha"),
                "n_events":         info.get("n_events"),
                # The ddsim seed this run simulated with, or ``None``. Two
                # different facts share that ``None``: a run that drew a fresh
                # seed, and a run written before the seed was recorded at all.
                # Neither can be shown to have simulated the same events as any
                # other night, which is the only question downstream asks of
                # this field, so they are deliberately not distinguished.
                "random_seed":      info.get("random_seed"),
                "status":           info.get("status"),
                "failed_configs":   info.get("failed_configs") or [],
                # Exact result labels planned by the expanded benchmark config.
                # Absent on legacy runs; ``None`` tells the regression report to
                # retain its historical-inference fallback.
                "configured_labels": _parse_configured_labels(info),
                "machine_consistent": info.get("machine_consistent"),
            }
        except (OSError, json.JSONDecodeError, TypeError, ValueError):
            _log.warning("parse_run_dir: could not read run_info.json in '%s'", run_dir)
            # Do NOT fall through to path-based parsing: in the temp download
            # directories created by remote.py the parent path components are
            # meaningless tmp names, not the EOS hierarchy.
            return _unknown_meta(run_dir)

    # run_info.json absent — infer only the run date from the directory name.
    # Path-component parsing is intentionally omitted: run dirs live inside
    # temp directories whose parent paths carry no semantic meaning.
    _log.warning("parse_run_dir: run_info.json missing in '%s'", run_dir)
    return _unknown_meta(run_dir)

load_machine_info

load_machine_info(run_dir: str) -> dict | None

Load machine_info.json from a run directory, or return None if absent.

Source code in k4bench/analysis/trend.py

def load_machine_info(run_dir: str) -> dict | None:
    """Load ``machine_info.json`` from a run directory, or return ``None`` if absent."""
    path = Path(run_dir) / "machine_info.json"
    if not path.exists():
        return None
    try:
        with open(path) as fh:
            return json.load(fh)
    except (OSError, json.JSONDecodeError):
        _log.warning("load_machine_info: could not read '%s'", path)
        return None

build_results_trend

build_results_trend(run_dirs: tuple[str, ...]) -> DataFrame | None

Load per-config results across run_dirs into one combined DataFrame.

Each row gets run_id, run_date, platform, k4h_release, k4h_release_date, github_run_url, random_seed and x_date columns.

Source code in k4bench/analysis/trend.py

def build_results_trend(run_dirs: tuple[str, ...]) -> pd.DataFrame | None:
    """Load per-config results across *run_dirs* into one combined DataFrame.

    Each row gets ``run_id``, ``run_date``, ``platform``, ``k4h_release``,
    ``k4h_release_date``, ``github_run_url``, ``random_seed`` and ``x_date``
    columns.
    """
    if not run_dirs:
        return None
    frames = []
    for d in run_dirs:
        run_dir = Path(d)
        if not run_dir.is_dir():
            continue
        meta = parse_run_dir(run_dir)
        try:
            df = load_results(run_dir)
        except EXPECTED_LOAD_ERRORS as exc:
            _log.debug("build_results_trend: skipping '%s': %s", run_dir, exc)
            continue
        except Exception:
            _log.exception("build_results_trend: error loading '%s'", run_dir)
            continue
        df["run_id"]           = run_dir.name
        df["run_date"]         = meta["run_date"]
        df["platform"]         = meta["platform"]
        df["k4h_release"]      = meta["k4h_release"]
        df["k4h_release_date"] = meta["k4h_release_date"]
        df["github_run_url"]   = meta["github_run_url"]
        df["xml_path"]         = meta["xml_path"]
        df["random_seed"]      = meta["random_seed"]
        frames.append(df)
    if not frames:
        return None
    return _finalize_dates(pd.concat(frames, ignore_index=True))

build_region_timing_trend

build_region_timing_trend(run_dirs: tuple[str, ...]) -> DataFrame | None

Load per-region timing summary across run_dirs.

For each run directory, region timing is loaded for all available configs. Per detector per config per run, the median and mean event-level time are computed (event 0 excluded as warmup).

Returns a long-form DataFrame with columns: run_id, run_date, k4h_release_date, label, attribution, detector, median_time_s, mean_time_s or None if no data could be loaded. run_id lets callers join each row with its run's reliability verdict (see k4bench.results.reliability_evidence).

Source code in k4bench/analysis/trend.py

def build_region_timing_trend(run_dirs: tuple[str, ...]) -> pd.DataFrame | None:
    """Load per-region timing summary across *run_dirs*.

    For each run directory, region timing is loaded for all available configs.
    Per detector per config per run, the median and mean event-level time are
    computed (event 0 excluded as warmup).

    Returns a long-form DataFrame with columns:
        run_id, run_date, k4h_release_date, label, attribution, detector,
        median_time_s, mean_time_s
    or ``None`` if no data could be loaded. ``run_id`` lets callers join each row
    with its run's reliability verdict (see ``k4bench.results.reliability_evidence``).
    """
    if not run_dirs:
        return None

    rows: list[dict] = []
    for d in run_dirs:
        run_dir = Path(d)
        if not run_dir.is_dir():
            continue
        meta = parse_run_dir(run_dir)
        try:
            region_data = load_region_timing(run_dir)
        except EXPECTED_LOAD_ERRORS as exc:
            _log.debug("build_region_timing_trend: skipping '%s': %s", run_dir, exc)
            continue
        except Exception:
            _log.exception("build_region_timing_trend: error loading '%s'", run_dir)
            continue

        for label, rdata in region_data.items():
            for attribution in ("at_location", "by_birth"):
                df_attr: pd.DataFrame = rdata.get(attribution)
                if df_attr is None or df_attr.empty:
                    continue
                # Exclude event 0 (warmup) from the index
                df_attr = df_attr[df_attr.index != 0]
                if df_attr.empty:
                    continue
                for detector in df_attr.columns:
                    vals = df_attr[detector].dropna().to_numpy()
                    if len(vals) == 0:
                        continue
                    s = pd.Series(vals)
                    n = len(vals)
                    rows.append({
                        "run_id":           run_dir.name,
                        "run_date":         meta["run_date"],
                        "k4h_release_date": meta["k4h_release_date"],
                        "k4h_release":      meta["k4h_release"],
                        "label":            label,
                        "attribution":      attribution,
                        "detector":         detector,
                        "n_events":         n,
                        "median_time_s":    float(s.median()),
                        "mean_time_s":      float(s.mean()),
                        "std_time_s":       float(s.std()) if n > 1 else 0.0,
                    })

    if not rows:
        return None
    return _finalize_dates(pd.DataFrame(rows))

build_event_timing_trend

build_event_timing_trend(run_dirs: tuple[str, ...]) -> DataFrame | None

Load per-event timing and memory summary across run_dirs.

For each run directory, event timing is loaded for all available configs. Per config per run, summary statistics are computed (event 0 excluded): mean_time_s, median_time_s, p95_time_s, trimmed_mean_time_s, mean_rss_mb, median_rss_mb, p95_rss_mb, max_rss_mb

trimmed_mean_time_s (:func:upper_trimmed_mean) is a view of the typical event, unmoved by how heavy this run's slow tail happened to be. It is absent (NaN) for a config with too few events to support it.

Where the event file records anonymous RSS, rss_anon_slope_mb_per_event is the :func:theil_sen_slope of the valid (non-negative) post-warm-up rss_anon_end_mb samples against their event numbers, in MB/event. It indicates memory growth across the run, which a leak or other accumulating state can cause; it is not proof of either. It is absent (NaN) below :data:MIN_SLOPE_EVENTS valid samples.

Returns a long-form DataFrame with those columns plus run_id, run_date, k4h_release_date, k4h_release, label or None if no data could be loaded. run_id lets callers join each row with its run's reliability verdict (see k4bench.results.reliability_evidence).

Source code in k4bench/analysis/trend.py

def build_event_timing_trend(run_dirs: tuple[str, ...]) -> pd.DataFrame | None:
    """Load per-event timing and memory summary across *run_dirs*.

    For each run directory, event timing is loaded for all available configs.
    Per config per run, summary statistics are computed (event 0 excluded):
        mean_time_s, median_time_s, p95_time_s, trimmed_mean_time_s,
        mean_rss_mb, median_rss_mb, p95_rss_mb, max_rss_mb

    ``trimmed_mean_time_s`` (:func:`upper_trimmed_mean`) is a view of the
    typical event, unmoved by how heavy this run's slow tail happened to be. It
    is absent (NaN) for a config with too few events to support it.

    Where the event file records anonymous RSS, ``rss_anon_slope_mb_per_event``
    is the :func:`theil_sen_slope` of the valid (non-negative) post-warm-up
    ``rss_anon_end_mb`` samples against their event numbers, in MB/event. It
    indicates memory growth across the run, which a leak or other accumulating
    state can cause; it is not proof of either. It is absent (NaN) below
    :data:`MIN_SLOPE_EVENTS` valid samples.

    Returns a long-form DataFrame with those columns plus
        run_id, run_date, k4h_release_date, k4h_release, label
    or ``None`` if no data could be loaded. ``run_id`` lets callers join each row
    with its run's reliability verdict (see ``k4bench.results.reliability_evidence``).
    """
    if not run_dirs:
        return None

    rows: list[dict] = []
    for d in run_dirs:
        run_dir = Path(d)
        if not run_dir.is_dir():
            continue
        meta = parse_run_dir(run_dir)
        try:
            event_data = load_event_timing(run_dir)
        except EXPECTED_LOAD_ERRORS as exc:
            _log.debug("build_event_timing_trend: skipping '%s': %s", run_dir, exc)
            continue
        except Exception:
            _log.exception("build_event_timing_trend: error loading '%s'", run_dir)
            continue

        for label, df_ev in event_data.items():
            # Exclude event 0 (warmup)
            df_ev = df_ev[df_ev["event_number"] != 0]
            if df_ev.empty:
                continue
            row: dict = {
                "run_id":           run_dir.name,
                "run_date":         meta["run_date"],
                "k4h_release_date": meta["k4h_release_date"],
                "k4h_release":      meta["k4h_release"],
                "label":            label,
            }
            if "event_time_s" in df_ev.columns:
                t = df_ev["event_time_s"].dropna()
                nt = len(t)
                if nt:
                    times = t.to_numpy(dtype=float)
                    row["n_events"]      = nt
                    row["mean_time_s"]   = float(t.mean())
                    row["median_time_s"] = float(t.median())
                    row["p95_time_s"]    = float(t.quantile(0.95))
                    row["std_time_s"]    = float(t.std()) if nt > 1 else 0.0
                    trimmed = upper_trimmed_mean(times)
                    if trimmed is not None:
                        row["trimmed_mean_time_s"] = trimmed
            if "rss_end_mb" in df_ev.columns:
                r = df_ev["rss_end_mb"].dropna()
                nr = len(r)
                if nr:
                    row["n_events_rss"]  = nr
                    row["mean_rss_mb"]   = float(r.mean())
                    row["median_rss_mb"] = float(r.median())
                    row["p95_rss_mb"]    = float(r.quantile(0.95))
                    row["max_rss_mb"]    = float(r.max())
                    row["std_rss_mb"]    = float(r.std()) if nr > 1 else 0.0
            if "rss_anon_end_mb" in df_ev.columns:
                r = df_ev["rss_anon_end_mb"].dropna()
                r = r[r >= 0]  # Negative samples indicate a failed /proc read.
                nr = len(r)
                if nr:
                    row["n_events_rss_anon"] = nr
                    row["mean_rss_anon_mb"] = float(r.mean())
                    row["median_rss_anon_mb"] = float(r.median())
                    row["std_rss_anon_mb"] = float(r.std()) if nr > 1 else 0.0
                # Slope against the event number, so a dropped sample leaves a
                # gap in x rather than pulling later events one step earlier.
                slope = theil_sen_slope(df_ev.loc[r.index, "event_number"], r)
                if slope is not None:
                    row["rss_anon_slope_mb_per_event"] = slope
            if "rss_file_end_mb" in df_ev.columns:
                r = df_ev["rss_file_end_mb"].dropna()
                r = r[r >= 0]
                nr = len(r)
                if nr:
                    row["n_events_rss_file"] = nr
                    row["mean_rss_file_mb"] = float(r.mean())
                    row["std_rss_file_mb"] = float(r.std()) if nr > 1 else 0.0
            rows.append(row)

    if not rows:
        return None
    return _finalize_dates(pd.DataFrame(rows))

build_machine_info_trend

build_machine_info_trend(run_dirs: tuple[str, ...]) -> DataFrame | None

Load per-run machine load / memory metrics across run_dirs.

Unlike the other trend builders there is no per-config dimension: each run directory has a single machine_info.json describing the physical machine that executed the benchmark. One row per run captures the load average, available RAM, swap, frequency and throttle counters, so callers can see how the host's condition varied across nightly releases (e.g. a day where the machine was under unusually high load).

Returns a DataFrame with one row per run plus run_id, run_date, k4h_release_date, k4h_release and x_date columns, or None if no machine info could be loaded. run_id lets callers join the machine condition of a run with its per-config results (e.g. to attach a reliability verdict to each run).

Source code in k4bench/analysis/trend.py

def build_machine_info_trend(run_dirs: tuple[str, ...]) -> pd.DataFrame | None:
    """Load per-run machine load / memory metrics across *run_dirs*.

    Unlike the other trend builders there is no per-config dimension: each run
    directory has a single ``machine_info.json`` describing the physical machine
    that executed the benchmark. One row per run captures the load average,
    available RAM, swap, frequency and throttle counters, so callers can see how
    the host's condition varied across nightly releases (e.g. a day where the
    machine was under unusually high load).

    Returns a DataFrame with one row per run plus ``run_id``, ``run_date``,
    ``k4h_release_date``, ``k4h_release`` and ``x_date`` columns, or ``None``
    if no machine info could be loaded. ``run_id`` lets callers join the
    machine condition of a run with its per-config results (e.g. to attach a
    reliability verdict to each run).
    """
    if not run_dirs:
        return None

    rows: list[dict] = []
    for d in run_dirs:
        run_dir = Path(d)
        if not run_dir.is_dir():
            continue
        mi = load_machine_info(str(run_dir))
        if not mi:
            continue
        meta = parse_run_dir(run_dir)
        rows.append({
            "run_id":                  run_dir.name,
            "run_date":                meta["run_date"],
            "k4h_release_date":        meta["k4h_release_date"],
            "k4h_release":             meta["k4h_release"],
            "hostname":                mi.get("hostname"),
            "cpu_logical_cores":       mi.get("cpu_logical_cores"),
            "cpu_physical_cores":      mi.get("cpu_physical_cores"),
            "load_avg_1m_start":       mi.get("load_avg_1m_start"),
            "load_avg_5m_start":       mi.get("load_avg_5m_start"),
            "load_avg_1m_end":         mi.get("load_avg_1m_end"),
            "load_avg_5m_end":         mi.get("load_avg_5m_end"),
            "ram_total_gb":            mi.get("ram_total_gb"),
            "ram_available_gb_start":  mi.get("ram_available_gb_start"),
            "ram_available_gb_end":    mi.get("ram_available_gb_end"),
            "swap_used_gb_start":      mi.get("swap_used_gb_start"),
            "swap_in_pages":           mi.get("swap_in_pages"),
            "swap_out_pages":          mi.get("swap_out_pages"),
            "cpu_freq_mhz_start":      mi.get("cpu_freq_mhz_start"),
            "thermal_throttle_events": mi.get("thermal_throttle_events"),
        })

    if not rows:
        return None
    return _finalize_dates(pd.DataFrame(rows))