diff --git a/.github/workflows/checks.main-pr.yaml b/.github/workflows/checks.main-pr.yaml
index 623c0ad..f2430fa 100644
--- a/.github/workflows/checks.main-pr.yaml
+++ b/.github/workflows/checks.main-pr.yaml
@@ -35,7 +35,8 @@ jobs:
run: |
set -euo pipefail
# sdist: README lives as an actual file
- tar -tzf dist/*.tar.gz | grep -qE "/README\.md$" || (echo "README.md missing from sdist"; exit 1)
+ tar -tzf dist/*.tar.gz > /tmp/sdist-listing
+ grep -qE "/README\.md$" /tmp/sdist-listing || (echo "README.md missing from sdist"; exit 1)
# wheel: README content is embedded in dist-info/METADATA, not as a file
python -m zipfile -e dist/*.whl /tmp/wheel-check
META=/tmp/wheel-check/*.dist-info/METADATA
diff --git a/README.md b/README.md
index c3afde1..8733a86 100644
--- a/README.md
+++ b/README.md
@@ -328,6 +328,75 @@ ethibench compare exp-extra/ --parent-dir all-experiments/ --output-dir comparis
(back to top)
+### `ethibench temporal`
+
+Generates a temporal evaluation plot showing how findings accumulate over elapsed time. Requires that findings include a `timestamp` field and that `ethibench evaluate` has already been run.
+
+```bash
+ethibench temporal --dataset [options]
+```
+
+**Arguments:**
+- `experiment_dir` — Experiment directory (with `run_*` subdirectories or a single run).
+
+**Options:**
+- `--dataset, -d` — (required) Path to dataset YAML file.
+- `--gt-dir, -g` — Ground truth directory. Defaults to `gt/` next to the dataset YAML.
+- `--output, -o` — Output PNG path. Defaults to `evaluation_outputs/plots/temporal_evaluation.png`.
+
+**What it does:**
+
+Produces a 4×N subplot figure (4 metric rows × N target columns) showing cumulative metrics over elapsed time for each run:
+- **True Positives** — cumulative TP count over time.
+- **False Positives** — cumulative FP count over time.
+- **Severity Score** — cumulative CVSS-weighted severity points.
+- **CWE Coverage** — number of unique CWEs discovered over time.
+
+**Timestamp handling:**
+- Findings without a `timestamp` field are skipped with a warning indicating the count and affected targets.
+- If entire runs lack timestamps, they are excluded from the plot.
+- If no finding in any run has a timestamp, the command exits with an error.
+- Start time is read from `metrics.json`; if unavailable, the earliest finding timestamp is used as t=0.
+
+(back to top)
+
+### `ethibench subset`
+
+Finds the subset of targets whose F1 score best correlates with the full benchmark, subject to a cost constraint. Useful for selecting a reduced evaluation suite that is cheaper to run while still being representative. No LLM API calls are made — only pre-computed evaluation results are used.
+
+```bash
+# Find subset costing at most 50% of the full benchmark
+ethibench subset --parent-dir experiments/ --dataset examples/dataset.yaml --max-ratio 0.5
+
+# Use Spearman correlation instead of Pearson
+ethibench subset -p experiments/ -d examples/dataset.yaml -r 0.5 -m spearman
+```
+
+**Options:**
+- `--parent-dir, -p` — (required) Parent folder containing experiment directories (each with `evaluation_outputs/`).
+- `--dataset, -d` — (required) Path to dataset YAML file.
+- `--metric, -m` — Correlation metric: `pearson` (default) or `spearman`.
+- `--max-ratio, -r` — (required) Maximum cost ratio threshold (0–1]. The selected subset's average cost must not exceed this fraction of the full benchmark cost.
+
+**What it does:**
+
+1. Discovers all experiments in the parent directory that have evaluation outputs.
+2. For each experiment, loads per-target TP/FP/FN from averaged results and per-target cost from `metrics_summary.json`.
+3. Enumerates all possible target subsets and computes the F1 score of each subset across experiments.
+4. Selects the subset with the highest correlation (Pearson or Spearman) to the full benchmark F1, subject to the cost ratio constraint.
+5. Prints the selected targets, correlation value, and cost breakdown.
+
+**Requirements:**
+- `ethibench evaluate` must have been run on all experiments.
+- Each experiment needs a `metrics_summary.json` with per-target cost data.
+- At least 3 experiments are recommended for reliable correlation estimates (a warning is shown otherwise).
+
+**Notes:**
+- With more than 15 targets, the exhaustive enumeration may be slow (2^N subsets). A warning is shown in this case.
+- Cost ratio is computed as: `average_subset_cost / average_full_benchmark_cost` across experiments.
+
+(back to top)
+
## File Formats
### Findings (`findings.jsonl`)
@@ -543,7 +612,7 @@ evaluation_outputs/cumulative-analysis/
```
src/ethibench/
-├── cli.py # Click CLI entry points (evaluate, convert-report, analyze, compare)
+├── cli.py # Click CLI entry points (evaluate, convert-report, analyze, compare, temporal, subset)
├── config.py # Environment variable configuration
├── models.py # Pydantic data models
├── datasets.py # Dataset/target YAML management
@@ -553,6 +622,8 @@ src/ethibench/
├── metrics.py # Per-target cost/token/duration metrics
├── convert_report.py # Report → findings conversion
├── cumulative_analysis.py # Cross-run cumulative analysis + overlap
+├── temporal.py # Temporal evaluation (cumulative metrics over time)
+├── subset.py # Reduced-suite selection via F1 correlation
├── pairwise.py # Pairwise A/B statistical comparison (t-test, Cohen's d)
├── plots.py # PNG chart generation (eval, cumulative, comparison)
├── report.py # Markdown summary generation
diff --git a/experiments/claude-code-sonnet/temporal_evaluation.png b/experiments/claude-code-sonnet/temporal_evaluation.png
deleted file mode 100644
index b143a11..0000000
Binary files a/experiments/claude-code-sonnet/temporal_evaluation.png and /dev/null differ
diff --git a/experiments/claude-code-sonnet/temporal_evaluation.py b/experiments/claude-code-sonnet/temporal_evaluation.py
deleted file mode 100644
index 768502e..0000000
--- a/experiments/claude-code-sonnet/temporal_evaluation.py
+++ /dev/null
@@ -1,267 +0,0 @@
-"""Temporal evaluation of claude-code-sonnet findings.
-
-Produces a 4×3 subplot figure showing cumulative TP, FP, severity score, and
-CWE coverage over elapsed time for each target (paygoat, vuln-bank, xben-090)
-across 3 replicates (run_1, run_2, run_3).
-"""
-
-import json
-import re
-from datetime import datetime, timezone
-from pathlib import Path
-
-import matplotlib.pyplot as plt
-
-# ---------------------------------------------------------------------------
-# Configuration
-# ---------------------------------------------------------------------------
-
-SCRIPT_DIR = Path(__file__).resolve().parent
-PROJECT_ROOT = SCRIPT_DIR.parents[1]
-
-RUNS = ["run_1", "run_2", "run_3"]
-TARGETS = [
- {"target_id": "paygoat", "subset": "PAYGoat", "gt_file": "paygoat_gt.jsonl"},
- {"target_id": "vuln-bank", "subset": "Vulnbank", "gt_file": "vulnbank_gt.jsonl"},
- {"target_id": "xben-090", "subset": "XBEN", "gt_file": "xben_gt.jsonl"},
-]
-
-GT_DIR = PROJECT_ROOT / "examples" / "gt"
-EXPERIMENT_DIR = SCRIPT_DIR
-EVAL_OUTPUTS_DIR = SCRIPT_DIR / "evaluation_outputs"
-OUTPUT_PNG = SCRIPT_DIR / "temporal_evaluation.png"
-
-RUN_COLORS = ["#2E6E3E", "#D35400", "#2C3E80"]
-RUN_LABELS = ["Run 1", "Run 2", "Run 3"]
-
-# ---------------------------------------------------------------------------
-# Helpers (replicated from evaluate.py to keep standalone)
-# ---------------------------------------------------------------------------
-
-
-def cvss_to_severity_points(cvss: float | str | None) -> int:
- if cvss is None:
- return 0
- cvss = float(cvss)
- if cvss == 0.0:
- return 0
- elif cvss <= 3.9:
- return 3
- elif cvss <= 6.9:
- return 15
- elif cvss <= 8.9:
- return 30
- else:
- return 50
-
-
-def extract_cwe_id(additional_info: str) -> str | None:
- m = re.search(r"CWE-(\d+)", additional_info)
- return m.group(1) if m else None
-
-
-def parse_timestamp(ts: str) -> datetime:
- """Parse ISO 8601 timestamps (supports both Z suffix and +00:00 offset).
-
- Always returns a timezone-aware datetime in UTC.
- """
- ts = ts.replace("Z", "+00:00")
- dt = datetime.fromisoformat(ts)
- if dt.tzinfo is None:
- dt = dt.replace(tzinfo=timezone.utc)
- return dt
-
-
-# ---------------------------------------------------------------------------
-# Data loading
-# ---------------------------------------------------------------------------
-
-
-def load_gt(gt_file: Path) -> tuple[dict[str, float | None], dict[str, str | None]]:
- """Load ground truth file, return (id→cvss, id→cwe) mappings."""
- gt_id_to_cvss: dict[str, float | None] = {}
- gt_id_to_cwe: dict[str, str | None] = {}
- with open(gt_file) as f:
- for line in f:
- if not line.strip():
- continue
- entry = json.loads(line)
- gid = entry["id"]
- gt_id_to_cvss[gid] = entry.get("cvss")
- gt_id_to_cwe[gid] = extract_cwe_id(entry.get("additional_info", ""))
- return gt_id_to_cvss, gt_id_to_cwe
-
-
-def load_findings_for_target(run_name: str, target_id: str) -> list[dict]:
- """Load findings from findings_parsed.jsonl filtered by target_id."""
- parsed_file = EVAL_OUTPUTS_DIR / run_name / "findings_parsed.jsonl"
- findings = []
- with open(parsed_file) as f:
- for line in f:
- if not line.strip():
- continue
- entry = json.loads(line)
- if entry.get("target_id") == target_id:
- findings.append(entry)
- return findings
-
-
-def load_matchings(run_name: str, subset: str) -> dict[str, list[str]]:
- """Load bipartite matchings for a subset. Returns uuid → list of matched GT IDs."""
- matchings_file = EVAL_OUTPUTS_DIR / run_name / "matchings" / f"matchings_{subset}.json"
- with open(matchings_file) as f:
- data = json.load(f)
- # Format is a JSON array with one dict element
- matchings_dict = data[0] if isinstance(data, list) else data
- return {
- uuid: mdata.get("selected_gt_ids", [])
- for uuid, mdata in matchings_dict.items()
- }
-
-
-def load_start_time(run_name: str, target_id: str) -> datetime:
- """Load start_time from metrics.json for a given run/target."""
- metrics_file = EXPERIMENT_DIR / run_name / target_id / "metrics.json"
- with open(metrics_file) as f:
- data = json.load(f)
- return parse_timestamp(data["start_time"])
-
-
-# ---------------------------------------------------------------------------
-# Temporal metric computation
-# ---------------------------------------------------------------------------
-
-
-def compute_temporal_metrics(
- findings: list[dict],
- matchings: dict[str, list[str]],
- start_time: datetime,
- gt_id_to_cvss: dict[str, float | None],
- gt_id_to_cwe: dict[str, str | None],
-) -> dict[str, list[float]]:
- """Compute cumulative TP, FP, severity, CWE coverage over time.
-
- Returns dict with keys: elapsed_min, tp, fp, severity, cwe_coverage.
- Each value is a list aligned by finding index (sorted by timestamp).
- Includes a t=0 origin point.
- """
- # Sort findings by timestamp
- findings_sorted = sorted(findings, key=lambda f: parse_timestamp(f["timestamp"]))
-
- elapsed_min = [0.0]
- tp = [0]
- fp = [0]
- severity = [0]
- cwe_coverage = [0]
-
- cum_tp = 0
- cum_fp = 0
- cum_severity = 0
- seen_cwes: set[str] = set()
-
- for finding in findings_sorted:
- ts = parse_timestamp(finding["timestamp"])
- minutes = (ts - start_time).total_seconds() / 60.0
-
- uuid = finding["uuid"]
- matched_gt_ids = matchings.get(uuid, [])
-
- if matched_gt_ids:
- cum_tp += 1
- for gt_id in matched_gt_ids:
- cum_severity += cvss_to_severity_points(gt_id_to_cvss.get(gt_id))
- cwe = gt_id_to_cwe.get(gt_id)
- if cwe:
- seen_cwes.add(cwe)
- else:
- cum_fp += 1
-
- elapsed_min.append(minutes)
- tp.append(cum_tp)
- fp.append(cum_fp)
- severity.append(cum_severity)
- cwe_coverage.append(len(seen_cwes))
-
- return {
- "elapsed_min": elapsed_min,
- "tp": tp,
- "fp": fp,
- "severity": severity,
- "cwe_coverage": cwe_coverage,
- }
-
-
-# ---------------------------------------------------------------------------
-# Plotting
-# ---------------------------------------------------------------------------
-
-METRICS_CONFIG = [
- {"key": "tp", "label": "True Positives"},
- {"key": "fp", "label": "False Positives"},
- {"key": "severity", "label": "Severity Score"},
- {"key": "cwe_coverage", "label": "CWE Coverage"},
-]
-
-
-def plot_temporal_evaluation():
- """Generate the 4×3 temporal evaluation figure."""
- fig, axes = plt.subplots(4, 3, figsize=(15, 12))
-
- for col_idx, target_cfg in enumerate(TARGETS):
- target_id = target_cfg["target_id"]
- subset = target_cfg["subset"]
- gt_file = GT_DIR / target_cfg["gt_file"]
-
- gt_id_to_cvss, gt_id_to_cwe = load_gt(gt_file)
-
- for run_idx, run_name in enumerate(RUNS):
- start_time = load_start_time(run_name, target_id)
- findings = load_findings_for_target(run_name, target_id)
- matchings = load_matchings(run_name, subset)
-
- metrics = compute_temporal_metrics(
- findings, matchings, start_time, gt_id_to_cvss, gt_id_to_cwe
- )
-
- for row_idx, metric_cfg in enumerate(METRICS_CONFIG):
- ax = axes[row_idx, col_idx]
- ax.plot(
- metrics["elapsed_min"],
- metrics[metric_cfg["key"]],
- color=RUN_COLORS[run_idx],
- label=RUN_LABELS[run_idx],
- linewidth=1.4,
- alpha=0.8,
- marker="o",
- markersize=3,
- )
-
- # Configure axes
- for row_idx, metric_cfg in enumerate(METRICS_CONFIG):
- ax = axes[row_idx, col_idx]
- if row_idx == 0:
- ax.set_title(target_id, fontsize=11, fontweight="bold")
- if col_idx == 0:
- ax.set_ylabel(metric_cfg["label"], fontsize=10)
- if row_idx == len(METRICS_CONFIG) - 1:
- ax.set_xlabel("Elapsed Time (min)", fontsize=9)
- ax.grid(True, alpha=0.3)
- ax.set_xlim(left=0)
- ax.set_ylim(bottom=0)
- if row_idx == 0 and col_idx == 2:
- ax.legend(loc="lower right", fontsize=8)
-
- fig.suptitle(
- "Temporal Evaluation \u2014 claude-code-sonnet",
- fontsize=13,
- fontweight="bold",
- y=0.98,
- )
- plt.tight_layout(rect=[0, 0, 1, 0.96])
- plt.savefig(OUTPUT_PNG, dpi=150, bbox_inches="tight")
- plt.close()
- print(f"Figure saved to: {OUTPUT_PNG}")
-
-
-if __name__ == "__main__":
- plot_temporal_evaluation()
diff --git a/src/ethibench/cli.py b/src/ethibench/cli.py
index ee67832..cafac23 100644
--- a/src/ethibench/cli.py
+++ b/src/ethibench/cli.py
@@ -770,3 +770,173 @@ def compare(experiment_dirs: tuple[Path, ...], output_dir: Path, parent_dir: Pat
logger.info(f"Pairwise comparison written to {output_dir / 'pairwise_comparison.md'}")
logger.info(f"Comparison complete. Results in {output_dir}")
+
+
+@cli.command()
+@click.argument(
+ "experiment_dir",
+ type=click.Path(exists=True, file_okay=False, path_type=Path),
+)
+@click.option(
+ "--dataset",
+ "-d",
+ required=True,
+ type=click.Path(exists=True, path_type=Path),
+ help="Dataset YAML file.",
+)
+@click.option(
+ "--gt-dir",
+ "-g",
+ type=click.Path(exists=True, file_okay=False, path_type=Path),
+ default=None,
+ help="Ground truth directory. Defaults to gt/ next to dataset YAML.",
+)
+@click.option(
+ "--output",
+ "-o",
+ type=click.Path(path_type=Path),
+ default=None,
+ help="Output PNG path. Defaults to evaluation_outputs/plots/temporal_evaluation.png.",
+)
+def temporal(
+ experiment_dir: Path,
+ dataset: Path,
+ gt_dir: Path | None,
+ output: Path | None,
+):
+ """Generate a temporal evaluation plot for EXPERIMENT_DIR.
+
+ Shows cumulative true positives, false positives, severity score, and CWE
+ coverage over elapsed time for each target across runs. Requires that
+ ``ethibench evaluate`` has already been run on the experiment.
+
+ Findings without a ``timestamp`` field are skipped with a warning.
+ """
+ from ethibench.temporal import run_temporal_evaluation
+
+ dc = DatasetCollection()
+ dc.init_from_yaml(dataset)
+
+ if gt_dir is None:
+ gt_dir = dataset.parent / "gt"
+ if not gt_dir.is_dir():
+ logger.error(f"No gt/ directory found next to {dataset}. Pass --gt-dir explicitly.")
+ sys.exit(1)
+
+ run_temporal_evaluation(experiment_dir, dc, gt_dir, output_path=output)
+
+
+@cli.command()
+@click.option(
+ "--parent-dir",
+ "-p",
+ required=True,
+ type=click.Path(exists=True, file_okay=False, path_type=Path),
+ help="Parent folder containing experiment directories (each with evaluation_outputs/).",
+)
+@click.option(
+ "--dataset",
+ "-d",
+ required=True,
+ type=click.Path(exists=True, path_type=Path),
+ help="Dataset YAML file.",
+)
+@click.option(
+ "--metric",
+ "-m",
+ type=click.Choice(["pearson", "spearman"], case_sensitive=False),
+ default="pearson",
+ help="Correlation metric to use.",
+)
+@click.option(
+ "--max-ratio",
+ "-r",
+ required=True,
+ type=click.FloatRange(min=0.0, max=1.0, min_open=True),
+ help="Maximum cost ratio threshold (0–1]. The selected subset's average cost "
+ "must not exceed this fraction of the full benchmark cost.",
+)
+def subset(parent_dir: Path, dataset: Path, metric: str, max_ratio: float):
+ """Find the target subset with highest F1 correlation to the full benchmark.
+
+ Enumerates all possible subsets of targets defined in the dataset YAML and
+ selects the one whose F1 score (across experiments) correlates best with
+ the full benchmark F1, subject to a cost ratio constraint.
+
+ Requires that ``ethibench evaluate`` has already been run on the experiments.
+ No LLM API calls are made — only pre-computed results are used.
+ """
+ from ethibench.subset import find_best_subset, load_all_experiments
+
+ dc = DatasetCollection()
+ dc.init_from_yaml(dataset)
+
+ experiment_dirs = _discover_experiment_dirs(parent_dir)
+ if not experiment_dirs:
+ logger.error(
+ f"No experiment directories found in {parent_dir}. "
+ "Subdirectories must contain an evaluation_outputs/ folder."
+ )
+ sys.exit(1)
+
+ logger.info(f"Found {len(experiment_dirs)} experiment(s) in {parent_dir}")
+
+ per_target_data, avg_costs, labels = load_all_experiments(experiment_dirs, dc)
+
+ if len(labels) < 3:
+ logger.warning(
+ f"Only {len(labels)} experiment(s) with valid data. "
+ "Correlation estimates may be unreliable with fewer than 3 data points."
+ )
+
+ if not per_target_data:
+ logger.error("No valid evaluation data found across experiments.")
+ sys.exit(1)
+
+ # Verify all targets have the same number of observations
+ counts = {tid: len(v) for tid, v in per_target_data.items()}
+ expected = len(labels)
+ missing = {tid for tid, c in counts.items() if c != expected}
+ if missing:
+ logger.warning(
+ f"Targets {missing} have inconsistent data across experiments — "
+ "dropping them from the analysis."
+ )
+ for tid in missing:
+ del per_target_data[tid]
+
+ if not per_target_data:
+ logger.error("No targets with consistent data across all experiments.")
+ sys.exit(1)
+
+ if not avg_costs:
+ logger.error(
+ "No cost data available. Each experiment needs a "
+ "metrics_summary.json file with a 'per_target' section containing "
+ "'total_cost' for each target. Run 'ethibench evaluate' first."
+ )
+ sys.exit(1)
+
+ best = find_best_subset(per_target_data, avg_costs, max_ratio, metric)
+
+ if best is None:
+ logger.error(
+ f"No subset found within cost ratio {max_ratio:.2f}. "
+ "Try increasing --max-ratio."
+ )
+ sys.exit(1)
+
+ # Print results
+ click.echo()
+ click.echo("=== Subset selection ===")
+ click.echo(f"Experiments used: {len(labels)}")
+ click.echo(f"Correlation metric: {best['metric']}")
+ click.echo(f"Correlation: {best['correlation']:.4f}")
+ click.echo(f"Full benchmark cost: {best['full_cost']:.4f}")
+ click.echo(f"Subset cost: {best['subset_cost']:.4f}")
+ click.echo(f"Cost ratio: {best['cost_ratio']:.4f}")
+ click.echo(f"Max allowed ratio: {max_ratio:.4f}")
+ click.echo()
+ click.echo("Selected targets:")
+ for t in best["subset"]:
+ click.echo(f" - {t}")
diff --git a/src/ethibench/subset.py b/src/ethibench/subset.py
new file mode 100644
index 0000000..deae872
--- /dev/null
+++ b/src/ethibench/subset.py
@@ -0,0 +1,226 @@
+"""Subset selection — find the target subset that best correlates with the full benchmark."""
+
+import itertools
+import json
+from pathlib import Path
+
+import numpy as np
+from loguru import logger
+from scipy.stats import spearmanr
+
+from ethibench.datasets import DatasetCollection
+
+
+def _f1_from_counts(tp: float, fp: float, fn: float) -> float:
+ denom = 2 * tp + fp + fn
+ return 0.0 if denom == 0 else 2 * tp / denom
+
+
+def load_experiment_data(
+ experiment_dir: Path,
+ dc: DatasetCollection,
+) -> tuple[dict[str, dict], dict[str, float]] | None:
+ """Load per-target evaluation results and costs for a single experiment.
+
+ Returns:
+ Tuple of (target_results, target_costs) or None if data is unavailable.
+ - target_results: ``{target_id: {"tp": ..., "fp": ..., "fn": ...}}``
+ - target_costs: ``{target_id: total_cost}``
+ """
+ eval_dir = experiment_dir / "evaluation_outputs"
+
+ # Find results folder (multi-run first, then single-run)
+ results_folder = None
+ for candidate in ("results_avg_all", "results_avg"):
+ path = eval_dir / candidate
+ if path.is_dir() and any(path.glob("evaluation_results_*.json")):
+ results_folder = path
+ break
+
+ if results_folder is None:
+ logger.warning(f"No evaluation results found in {eval_dir}")
+ return None
+
+ # Load per-subset results and map back to target_ids
+ target_results: dict[str, dict] = {}
+ for json_file in results_folder.glob("evaluation_results_*.json"):
+ name = json_file.stem.replace("evaluation_results_", "")
+ if name in ("unweighted", "weighted"):
+ continue
+ with open(json_file) as f:
+ data = json.load(f)
+ if not isinstance(data, dict):
+ continue
+ subset_name = data.get("subset_name", name)
+ # Map subset_name back to target_ids
+ target_ids = dc.get_target_ids_for_subset(subset_name)
+ if len(target_ids) == 1:
+ tid = next(iter(target_ids))
+ target_results[tid] = {
+ "tp": data["true_positives"],
+ "fp": data["false_positives"],
+ "fn": data["false_negatives"],
+ }
+ else:
+ # Multiple targets per subset — attribute equally (best effort)
+ for tid in target_ids:
+ n = len(target_ids)
+ target_results[tid] = {
+ "tp": data["true_positives"] / n,
+ "fp": data["false_positives"] / n,
+ "fn": data["false_negatives"] / n,
+ }
+
+ # Load cost from metrics_summary.json
+ target_costs: dict[str, float] = {}
+ metrics_file = eval_dir / "metrics_summary.json"
+ if metrics_file.exists():
+ with open(metrics_file) as f:
+ metrics = json.load(f)
+ per_target = metrics.get("per_target", {})
+ for tid in target_results:
+ if tid in per_target:
+ target_costs[tid] = per_target[tid].get("total_cost", 0.0)
+
+ return target_results, target_costs
+
+
+def load_all_experiments(
+ experiment_dirs: list[Path],
+ dc: DatasetCollection,
+) -> tuple[dict[str, list[dict]], dict[str, float], list[str]]:
+ """Load data from all experiments.
+
+ Returns:
+ - per_target_data: ``{target_id: [{"tp", "fp", "fn"}, ...]}`` one entry per experiment
+ - avg_costs: ``{target_id: average_cost}`` averaged across experiments
+ - experiment_labels: list of experiment names that were successfully loaded
+ """
+ per_target_data: dict[str, list[dict]] = {}
+ cost_accum: dict[str, list[float]] = {}
+ experiment_labels: list[str] = []
+ missing_cost_experiments: list[str] = []
+
+ for exp_dir in experiment_dirs:
+ result = load_experiment_data(exp_dir, dc)
+ if result is None:
+ continue
+
+ target_results, target_costs = result
+ if not target_results:
+ continue
+
+ experiment_labels.append(exp_dir.name)
+
+ for tid, counts in target_results.items():
+ per_target_data.setdefault(tid, []).append(counts)
+
+ if target_costs:
+ for tid, cost in target_costs.items():
+ cost_accum.setdefault(tid, []).append(cost)
+ else:
+ missing_cost_experiments.append(exp_dir.name)
+
+ if missing_cost_experiments:
+ logger.warning(
+ f"No cost data (metrics_summary.json with per_target costs) for: "
+ f"{', '.join(missing_cost_experiments)}. "
+ f"Cost ratio filtering will use data from the remaining experiments only."
+ )
+
+ # Average costs across experiments
+ avg_costs = {tid: np.mean(costs) for tid, costs in cost_accum.items()}
+
+ return per_target_data, avg_costs, experiment_labels
+
+
+def _compute_f1_vector(
+ per_target_data: dict[str, list[dict]],
+ subset: tuple[str, ...],
+ n_experiments: int,
+) -> np.ndarray:
+ """Compute F1 for a subset of targets, one value per experiment."""
+ f1s = []
+ for i in range(n_experiments):
+ tp = sum(per_target_data[t][i]["tp"] for t in subset)
+ fp = sum(per_target_data[t][i]["fp"] for t in subset)
+ fn = sum(per_target_data[t][i]["fn"] for t in subset)
+ f1s.append(_f1_from_counts(tp, fp, fn))
+ return np.array(f1s)
+
+
+def _pearson(x: np.ndarray, y: np.ndarray) -> float:
+ if np.std(x) == 0 or np.std(y) == 0:
+ return 0.0
+ return float(np.corrcoef(x, y)[0, 1])
+
+
+def _spearman(x: np.ndarray, y: np.ndarray) -> float:
+ if np.std(x) == 0 or np.std(y) == 0:
+ return 0.0
+ result = spearmanr(x, y)
+ return float(result.correlation)
+
+
+def find_best_subset(
+ per_target_data: dict[str, list[dict]],
+ avg_costs: dict[str, float],
+ max_ratio: float,
+ metric: str = "pearson",
+) -> dict | None:
+ """Find the target subset with highest correlation to the full benchmark F1.
+
+ Args:
+ per_target_data: ``{target_id: [{"tp", "fp", "fn"}, ...]}``
+ avg_costs: ``{target_id: average_cost}``
+ max_ratio: Maximum allowed cost ratio (0–1).
+ metric: ``"pearson"`` or ``"spearman"``.
+
+ Returns:
+ Dict with ``subset``, ``correlation``, ``cost_ratio``, ``subset_cost``,
+ ``full_cost``, ``metric`` — or ``None`` if no valid subset is found.
+ """
+ targets = sorted(per_target_data.keys())
+ n_experiments = len(next(iter(per_target_data.values())))
+
+ corr_fn = _pearson if metric == "pearson" else _spearman
+
+ full_f1 = _compute_f1_vector(per_target_data, tuple(targets), n_experiments)
+ full_cost = sum(avg_costs.get(t, 0.0) for t in targets)
+
+ if full_cost <= 0:
+ logger.warning("Total cost is zero — cost ratio filtering is disabled.")
+ max_allowed_cost = float("inf")
+ else:
+ max_allowed_cost = max_ratio * full_cost
+
+ n_targets = len(targets)
+ if n_targets > 15:
+ total_combos = 2**n_targets - 1
+ logger.warning(
+ f"{n_targets} targets → {total_combos} subsets to evaluate. "
+ f"This may take a while."
+ )
+
+ best: dict | None = None
+
+ for k in range(1, n_targets + 1):
+ for subset in itertools.combinations(targets, k):
+ subset_cost = sum(avg_costs.get(t, 0.0) for t in subset)
+ if subset_cost > max_allowed_cost:
+ continue
+
+ subset_f1 = _compute_f1_vector(per_target_data, subset, n_experiments)
+ corr = corr_fn(full_f1, subset_f1)
+
+ if best is None or corr > best["correlation"]:
+ best = {
+ "subset": list(subset),
+ "correlation": corr,
+ "cost_ratio": subset_cost / full_cost if full_cost > 0 else 0.0,
+ "subset_cost": subset_cost,
+ "full_cost": full_cost,
+ "metric": metric,
+ }
+
+ return best
diff --git a/src/ethibench/temporal.py b/src/ethibench/temporal.py
new file mode 100644
index 0000000..d918186
--- /dev/null
+++ b/src/ethibench/temporal.py
@@ -0,0 +1,409 @@
+"""Temporal evaluation — cumulative metrics over elapsed time.
+
+Produces a 4×N subplot figure showing cumulative TP, FP, severity score, and
+CWE coverage over elapsed time for each target across runs.
+"""
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+from loguru import logger
+
+from ethibench.datasets import DatasetCollection
+from ethibench.evaluate import cvss_to_severity_points, extract_cwe_id, load_jsonl
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_RUN_COLORS = [
+ "#2E6E3E",
+ "#D35400",
+ "#2C3E80",
+ "#8E44AD",
+ "#C0392B",
+ "#16A085",
+ "#F39C12",
+ "#2980B9",
+]
+
+_METRICS_CONFIG = [
+ {"key": "tp", "label": "True Positives"},
+ {"key": "fp", "label": "False Positives"},
+ {"key": "severity", "label": "Severity Score"},
+ {"key": "cwe_coverage", "label": "CWE Coverage"},
+]
+
+
+def _parse_timestamp(ts: str) -> datetime:
+ """Parse ISO 8601 timestamps (supports both Z suffix and +00:00 offset).
+
+ Always returns a timezone-aware datetime in UTC.
+ """
+ ts = ts.replace("Z", "+00:00")
+ dt = datetime.fromisoformat(ts)
+ if dt.tzinfo is None:
+ dt = dt.replace(tzinfo=timezone.utc)
+ return dt
+
+
+# ---------------------------------------------------------------------------
+# Data loading
+# ---------------------------------------------------------------------------
+
+
+def _load_gt_mappings(
+ gt_dir: Path, subset_name: str
+) -> tuple[dict[str, float | None], dict[str, str | None]]:
+ """Load ground truth entries for a subset, return (id→cvss, id→cwe) mappings."""
+ entries = load_jsonl(gt_dir, subset_filter=subset_name)
+ gt_id_to_cvss: dict[str, float | None] = {}
+ gt_id_to_cwe: dict[str, str | None] = {}
+ for entry in entries:
+ gid = entry["id"]
+ gt_id_to_cvss[gid] = entry.get("cvss")
+ gt_id_to_cwe[gid] = extract_cwe_id(entry.get("additional_info", ""))
+ return gt_id_to_cvss, gt_id_to_cwe
+
+
+def _load_findings_for_target(
+ eval_output_dir: Path, target_id: str
+) -> tuple[list[dict], int]:
+ """Load findings from findings_parsed.jsonl filtered by target_id.
+
+ Returns (findings_with_timestamp, count_without_timestamp).
+ """
+ parsed_file = eval_output_dir / "findings_parsed.jsonl"
+ if not parsed_file.exists():
+ return [], 0
+
+ with_ts: list[dict] = []
+ without_ts = 0
+ with open(parsed_file) as f:
+ for line in f:
+ if not line.strip():
+ continue
+ entry = json.loads(line)
+ if entry.get("target_id") != target_id:
+ continue
+ if "timestamp" in entry and entry["timestamp"]:
+ with_ts.append(entry)
+ else:
+ without_ts += 1
+ return with_ts, without_ts
+
+
+def _load_matchings(eval_output_dir: Path, subset: str) -> dict[str, list[str]]:
+ """Load bipartite matchings for a subset. Returns uuid → list of matched GT IDs."""
+ matchings_file = eval_output_dir / "matchings" / f"matchings_{subset}.json"
+ if not matchings_file.exists():
+ return {}
+ with open(matchings_file) as f:
+ data = json.load(f)
+ matchings_dict = data[0] if isinstance(data, list) else data
+ return {
+ uuid: mdata.get("selected_gt_ids", [])
+ for uuid, mdata in matchings_dict.items()
+ }
+
+
+def _load_start_time(
+ experiment_dir: Path,
+ run_dir_name: str | None,
+ target_id: str,
+ findings: list[dict],
+) -> datetime | None:
+ """Load start_time from metrics.json, falling back to earliest finding timestamp.
+
+ For multi-run experiments: experiment_dir/run_dir_name/target_id/metrics.json
+ For single-run experiments: experiment_dir/target_id/metrics.json
+ """
+ if run_dir_name is not None:
+ metrics_file = experiment_dir / run_dir_name / target_id / "metrics.json"
+ else:
+ metrics_file = experiment_dir / target_id / "metrics.json"
+
+ if metrics_file.exists():
+ with open(metrics_file) as f:
+ data = json.load(f)
+ if "start_time" in data:
+ return _parse_timestamp(data["start_time"])
+
+ # Fallback to earliest finding timestamp
+ if findings:
+ logger.warning(
+ f"No metrics.json start_time for {target_id}"
+ + (f" in {run_dir_name}" if run_dir_name else "")
+ + "; using earliest finding timestamp as t=0."
+ )
+ return min(_parse_timestamp(f["timestamp"]) for f in findings)
+
+ return None
+
+
+# ---------------------------------------------------------------------------
+# Temporal metric computation
+# ---------------------------------------------------------------------------
+
+
+def _compute_temporal_metrics(
+ findings: list[dict],
+ matchings: dict[str, list[str]],
+ start_time: datetime,
+ gt_id_to_cvss: dict[str, float | None],
+ gt_id_to_cwe: dict[str, str | None],
+) -> dict[str, list[float]]:
+ """Compute cumulative TP, FP, severity, CWE coverage over time.
+
+ Returns dict with keys: elapsed_min, tp, fp, severity, cwe_coverage.
+ Each value is a list aligned by finding index (sorted by timestamp).
+ Includes a t=0 origin point.
+ """
+ findings_sorted = sorted(findings, key=lambda f: _parse_timestamp(f["timestamp"]))
+
+ elapsed_min: list[float] = [0.0]
+ tp: list[int] = [0]
+ fp: list[int] = [0]
+ severity: list[int] = [0]
+ cwe_coverage: list[int] = [0]
+
+ cum_tp = 0
+ cum_fp = 0
+ cum_severity = 0
+ seen_cwes: set[str] = set()
+
+ for finding in findings_sorted:
+ ts = _parse_timestamp(finding["timestamp"])
+ minutes = (ts - start_time).total_seconds() / 60.0
+
+ uuid = finding["uuid"]
+ matched_gt_ids = matchings.get(uuid, [])
+
+ if matched_gt_ids:
+ cum_tp += 1
+ for gt_id in matched_gt_ids:
+ cum_severity += cvss_to_severity_points(gt_id_to_cvss.get(gt_id))
+ cwe = gt_id_to_cwe.get(gt_id)
+ if cwe:
+ seen_cwes.add(cwe)
+ else:
+ cum_fp += 1
+
+ elapsed_min.append(minutes)
+ tp.append(cum_tp)
+ fp.append(cum_fp)
+ severity.append(cum_severity)
+ cwe_coverage.append(len(seen_cwes))
+
+ return {
+ "elapsed_min": elapsed_min,
+ "tp": tp,
+ "fp": fp,
+ "severity": severity,
+ "cwe_coverage": cwe_coverage,
+ }
+
+
+# ---------------------------------------------------------------------------
+# Main entry point
+# ---------------------------------------------------------------------------
+
+
+def run_temporal_evaluation(
+ experiment_dir: Path,
+ dc: DatasetCollection,
+ gt_dir: Path,
+ output_path: Path | None = None,
+) -> None:
+ """Generate a temporal evaluation plot for an experiment.
+
+ Args:
+ experiment_dir: Root experiment directory (contains run_* subdirs or
+ target subdirs directly).
+ dc: Loaded DatasetCollection.
+ gt_dir: Path to ground truth JSONL files directory.
+ output_path: Where to save the PNG. Defaults to
+ ``experiment_dir/evaluation_outputs/plots/temporal_evaluation.png``.
+ """
+ eval_outputs_dir = experiment_dir / "evaluation_outputs"
+ if not eval_outputs_dir.is_dir():
+ logger.error(
+ f"No evaluation_outputs/ directory found in {experiment_dir}. "
+ "Run `ethibench evaluate` first."
+ )
+ raise SystemExit(1)
+
+ # Detect runs
+ run_candidates = sorted(
+ [d for d in experiment_dir.iterdir() if d.is_dir() and d.name.startswith("run")],
+ key=lambda p: p.name,
+ )
+ multi_run = bool(run_candidates)
+
+ if multi_run:
+ run_names = [d.name for d in run_candidates]
+ else:
+ run_names = [None] # type: ignore[list-item]
+
+ # Build target list from the dataset
+ targets: list[dict[str, str]] = []
+ for dataset in dc.datasets:
+ for target in dataset.targets:
+ targets.append(
+ {"target_id": target.target_id, "subset": dataset.subset}
+ )
+
+ if not targets:
+ logger.error("No targets found in the dataset.")
+ raise SystemExit(1)
+
+ # --- Phase 1: check timestamp availability per run ---
+ # run_name → total findings with timestamp across all targets
+ run_ts_counts: dict[str | None, int] = {}
+ run_no_ts_counts: dict[str | None, int] = {}
+ # (run_name, target_id) → count without timestamp
+ per_run_target_no_ts: dict[tuple[str | None, str], int] = {}
+
+ for run_name in run_names:
+ if multi_run:
+ eval_dir = eval_outputs_dir / run_name
+ else:
+ eval_dir = eval_outputs_dir
+
+ total_with = 0
+ total_without = 0
+ for t in targets:
+ with_ts, without_ts = _load_findings_for_target(eval_dir, t["target_id"])
+ total_with += len(with_ts)
+ total_without += without_ts
+ if without_ts > 0:
+ per_run_target_no_ts[(run_name, t["target_id"])] = without_ts
+
+ run_ts_counts[run_name] = total_with
+ run_no_ts_counts[run_name] = total_without
+
+ # Determine which runs are usable
+ runs_without_ts = [r for r in run_names if run_ts_counts[r] == 0]
+ runs_with_ts = [r for r in run_names if run_ts_counts[r] > 0]
+
+ if not runs_with_ts:
+ logger.error("No findings have timestamp in this experiment.")
+ raise SystemExit(1)
+
+ # Print warnings for missing timestamps
+ if multi_run:
+ if runs_without_ts:
+ labels = ", ".join(r for r in runs_without_ts)
+ logger.warning(f"No timestamps available for {labels}.")
+ for run_name in runs_with_ts:
+ no_ts = run_no_ts_counts[run_name]
+ if no_ts > 0:
+ affected = [
+ f"{cnt} in {tid}"
+ for (rn, tid), cnt in per_run_target_no_ts.items()
+ if rn == run_name
+ ]
+ logger.warning(
+ f"{no_ts} findings in {run_name} do not have timestamp "
+ f"({', '.join(affected)})."
+ )
+ else:
+ no_ts = run_no_ts_counts[None]
+ if no_ts > 0:
+ affected = [
+ f"{cnt} in {tid}"
+ for (rn, tid), cnt in per_run_target_no_ts.items()
+ ]
+ logger.warning(
+ f"{no_ts} findings do not have timestamp "
+ f"({', '.join(affected)})."
+ )
+
+ # --- Phase 2: compute metrics and plot ---
+ n_targets = len(targets)
+ n_runs = len(runs_with_ts)
+
+ fig, axes = plt.subplots(
+ len(_METRICS_CONFIG), n_targets, figsize=(5 * n_targets, 12), squeeze=False
+ )
+
+ run_colors = _RUN_COLORS
+ run_labels = (
+ [r.replace("_", " ").title() for r in runs_with_ts]
+ if multi_run
+ else [""]
+ )
+
+ for col_idx, target_cfg in enumerate(targets):
+ target_id = target_cfg["target_id"]
+ subset = target_cfg["subset"]
+
+ gt_id_to_cvss, gt_id_to_cwe = _load_gt_mappings(gt_dir, subset)
+
+ for run_idx, run_name in enumerate(runs_with_ts):
+ if multi_run:
+ eval_dir = eval_outputs_dir / run_name
+ else:
+ eval_dir = eval_outputs_dir
+
+ findings, _ = _load_findings_for_target(eval_dir, target_id)
+ if not findings:
+ continue
+
+ matchings = _load_matchings(eval_dir, subset)
+ start_time = _load_start_time(
+ experiment_dir, run_name, target_id, findings
+ )
+ if start_time is None:
+ continue
+
+ metrics = _compute_temporal_metrics(
+ findings, matchings, start_time, gt_id_to_cvss, gt_id_to_cwe
+ )
+
+ color = run_colors[run_idx % len(run_colors)]
+ for row_idx, metric_cfg in enumerate(_METRICS_CONFIG):
+ ax = axes[row_idx, col_idx]
+ ax.plot(
+ metrics["elapsed_min"],
+ metrics[metric_cfg["key"]],
+ color=color,
+ label=run_labels[run_idx] if run_labels[run_idx] else None,
+ linewidth=1.4,
+ alpha=0.8,
+ marker="o",
+ markersize=3,
+ )
+
+ # Configure axes
+ for row_idx, metric_cfg in enumerate(_METRICS_CONFIG):
+ ax = axes[row_idx, col_idx]
+ if row_idx == 0:
+ ax.set_title(target_id, fontsize=11, fontweight="bold")
+ if col_idx == 0:
+ ax.set_ylabel(metric_cfg["label"], fontsize=10)
+ if row_idx == len(_METRICS_CONFIG) - 1:
+ ax.set_xlabel("Elapsed Time (min)", fontsize=9)
+ ax.grid(True, alpha=0.3)
+ ax.set_xlim(left=0)
+ ax.set_ylim(bottom=0)
+ if row_idx == 0 and col_idx == n_targets - 1 and multi_run:
+ ax.legend(loc="lower right", fontsize=8)
+
+ experiment_name = experiment_dir.name
+ fig.suptitle(
+ f"Temporal Evaluation \u2014 {experiment_name}",
+ fontsize=13,
+ fontweight="bold",
+ y=0.98,
+ )
+ plt.tight_layout(rect=[0, 0, 1, 0.96])
+
+ if output_path is None:
+ output_path = eval_outputs_dir / "plots" / "temporal_evaluation.png"
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+ plt.savefig(output_path, dpi=150, bbox_inches="tight")
+ plt.close()
+ logger.info(f"Temporal evaluation figure saved to {output_path}")