diff --git a/CHANGELOG.md b/CHANGELOG.md
index ec51c2c..6c95ca7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,66 @@
# Changelog
+## 0.3.0
+
+### Added
+
+- Strengthened `context_order="auto"` to validate global graph-path support
+ against observed execution traces, including longer-history recombinations;
+ added a regression test for pairwise-consistent phantom paths.
+- Added context-aware DPG construction through `context_order` in execution
+ trace mode. `context_order=1` preserves legacy predicate identity; integer
+ orders and `"auto"` split nodes by recent execution history.
+- Class outcomes are shared terminal sinks at every context order, and node
+ metadata exposes `predicate`, `context`, and `context_order`.
+- Added enumeration-free local context-order resolution with explicit
+ resolution history and failure reporting.
+- Execution-trace edge construction preserves within-case event order and no
+ longer applies an unstable sort to the constant case identifier.
+- Added exact sklearn `decision_path` routing. Threshold rounding now formats
+ predicate labels without changing the branch selected by the model.
+- Added `decimal_threshold="auto"`, which derives precision from the data and
+ warns when a tree threshold is off the derived grid.
+- Validated the context resolver's optional `max_k` bound so invalid or
+ insufficient caps fail explicitly instead of returning an unresolved order.
+
+### Benchmarking evidence
+
+- The post-fix E2 benchmark completed all 375/375 cells. `context_order="auto"`
+ resolved a mean order of 2.0373 and reported a zero phantom-path rate across
+ every tested classifier family; the corresponding k=1 execution-trace graphs
+ retained phantom paths in the pooled graph for Bagging, Gradient Boosting,
+ and Random Forest.
+- Across that grid, post-fix execution-trace k=1 averaged 7.0742 seconds and
+ 886.8 graph nodes per cell, while auto-k averaged 7.2271 seconds and 1005.5
+ nodes: approximately 2.2% more runtime for trace-consistent graph structure.
+- In the E5 alignment benchmark, the shipped unweighted LRC aggregation reached
+ mean Spearman correlation 0.8525 at k=1 and 0.9360 at auto-k, with mean top-10
+ feature overlap increasing from 0.7574 to 0.8759.
+- On the reference ten-tree Random Forests for Iris, Wine, and Breast Cancer,
+ k=1 and auto-k contained exactly the same raw `(feature, operator, threshold)`
+ split predicates. Differences in class-boundary envelopes were caused by
+ contextual predicate/community assignments, not by changed learned splits.
+
+### Compatibility and limitations
+
+- The default remains `context_order=1`; DPG-k is opt-in so existing consumers
+ keep their graph shape. `context_order > 1` requires `execution_trace` mode.
+- `get_trace_consistent_lrc()` remains available for k=1 and is deprecated for
+ contextual graphs; k>1 aggregates ordinary unweighted node LRC by predicate.
+- The routing correction can change graph weights and labels at floating-point
+ boundaries. Residual off-grid behavior is reported by the auto-precision
+ warning rather than hidden.
+- **Regression sink semantics are out of scope for 0.3.0.** `context_order`
+ mechanically builds a graph for regressors (regression leaves are treated
+ as terminal sinks, like class leaves), but there is no "one sink per
+ output" guarantee: a regression sink is only as unique as the 2-decimal
+ rounded leaf value, so two leaves collide into one sink by coincidence of
+ rounding, not by any modeled notion of "output". A principled regression
+ sink policy is deferred to a future release. `class_boundaries` and
+ `communities` remain classifier-only features; calling
+ `DPGExplainer.explain_global(communities=True)` on a regressor now raises a
+ clear `ValueError` instead of an internal `numpy.linalg.LinAlgError`.
+
## 0.2.0
### Added
diff --git a/README.md b/README.md
index 7ffe40b..1312c7c 100644
--- a/README.md
+++ b/README.md
@@ -2,12 +2,11 @@
[](LICENSE)
[](pyproject.toml)
-[](https://pypi.org/project/dpg/)
[](https://github.com/Meta-Group/DPG/actions/workflows/ci.yml)
[](https://dpg.readthedocs.io/en/latest/)
- 
+ 
@@ -21,7 +20,9 @@ insightful points. DPG enables graph-based evaluations and the identification of
towards facilitating comparisons between features and their associated values while offering insights
into the entire model. DPG provides descriptive metrics that enhance the understanding of the
decisions inherent in the model, offering valuable insights.
-
+
+ 
+
---
@@ -47,7 +48,9 @@ The concept behind DPG is to convert a generic tree-based ensemble model for cla
- Nodes represent predicates, i.e., the feature-value associations present in each node of every tree;
- Edges denote the frequency with which these predicates are satisfied during the model training phase by the samples of the dataset.
-
+
+ 
+
## Metrics
The graph-based nature of DPG provides significant enhancements in the direction of a complete mapping of the ensemble structure.
@@ -61,7 +64,7 @@ The graph-based nature of DPG provides significant enhancements in the direction
|Constraints | Betweenness centrality | Local reaching centrality | Community|
|------------|------------|--------------|--------------------|
- |  |  |  |
+ |  |  |  |
|Constraints(Class 1) = val3 < F1 ≤ val1, F2 ≤ val2 | BC(F2 ≤ val2) = 4/24 | LRC(F1 ≤ val1) = 6 / 7 | Community(Class 1) = F1 ≤ val1, F2 ≤ val2 |
---
@@ -169,6 +172,7 @@ explainer = DPGExplainer(
},
"graph_construction": {
"mode": "execution_trace", # or "aggregated_transitions"
+ "context_order": 1, # 1, an integer > 1, or "auto"
},
}
},
@@ -178,6 +182,31 @@ explainer = DPGExplainer(
- `"aggregated_transitions"`: current default behavior; filters path variants first, then discovers the DPG.
- `"execution_trace"`: builds directly from raw traces and filters edges instead of whole-path variants when `perc_var > 0`.
+`context_order` controls predicate identity in `execution_trace` mode. `1`
+preserves the legacy graph exactly; an order greater than one uses the last k
+executed predicates as context and keeps one sink per class. `"auto"` resolves
+the smallest order for which every pooled graph path is supported by an
+observed execution trace, including recombinations that only appear after
+multiple hops. Orders greater than one require `mode: "execution_trace"`.
+`context_order="auto"` also requires `execution_trace`; in
+`aggregated_transitions` mode context resolution is not applied.
+
+A phantom path is a multi-hop path assembled by pooling edges from different
+tree executions even though no single execution produced it. DPG-k removes
+these recombinations at the resolved context order while retaining one graph.
+The effective order and each node's context are available through
+`get_context_order()` and `get_node_context(node)`.
+
+**Regression scope (0.3.0):** `context_order` builds without raising for
+regressors (regression leaves, labeled `"Pred "`, are treated as
+terminal sinks just like class leaves), but "one sink per output" is not a
+defined guarantee for regression the way it is for classification: two
+leaves collapse into the same sink only when their rounded values coincide,
+which is an artifact of label rounding, not a modeled invariant. A
+principled regression sink policy is deferred past 0.3.0. `class_boundaries`
+and `communities` remain classifier-only; calling `explain_global`
+with `communities=True` on a regressor raises a clear `ValueError`.
+
#### Minimal local workflow
```python
@@ -263,8 +292,8 @@ Important:
## CLI scripts
The library contains two different scripts to apply DPG:
-- `run_dpg_standard.py`: with this script it is possible to test DPG on a standard classification dataset provided by `sklearn` such as `iris`, `digits`, `wine`, `breast cancer`, and `diabetes`.
-- `run_dpg_custom.py`: with this script it is possible to apply DPG to your classification dataset, specifying the target class.
+- `examples/run_dpg_standard.py`: with this script it is possible to test DPG on a standard classification dataset provided by `sklearn` such as `iris`, `digits`, `wine`, `breast cancer`, and `diabetes`.
+- `examples/run_dpg_custom.py`: with this script it is possible to apply DPG to your classification dataset, specifying the target class.
### Implementation notes
The library also contains two other essential scripts:
@@ -272,14 +301,17 @@ The library also contains two other essential scripts:
- `visualizer.py` contains the functions used to manage the visualization of DPG.
### Output
-The DPG output, through `run_dpg_standard.py` or `run_dpg_custom.py`, produces several files:
+The DPG output, through `examples/run_dpg_standard.py` or `examples/run_dpg_custom.py`, produces several files:
- the visualization of DPG in a dedicated environment, which can be zoomed and saved;
- a `.txt` file containing the DPG metrics;
- a `.csv` file containing the information about all the nodes of the DPG and their associated metrics;
- a `.txt` file containing the Random Forest statistics (accuracy, confusion matrix, classification report)
### CLI parameter reference
-Usage: `python run_dpg_standard.py --dataset --n_learners --pv --t --model_name --dir --plot --save_plot_dir --attribute --communities --clusters --threshold_clusters --class_flag --seed `
+Usage: `dpg --dataset --n_learners --pv --t --model_name --dir --plot --save_plot_dir --attribute --communities --clusters --threshold_clusters --class_flag --seed `
+
+After installing DPG, the `dpg` command is the packaged equivalent of
+`python examples/run_dpg_standard.py`.
Where:
- `dataset` is the name of the standard classification `sklearn` dataset to be analyzed;
- `n_learners` is the number of base learners for the ensemble model;
@@ -306,23 +338,29 @@ Where:
Disclaimer: `attribute`, `communities`, and `clusters` are mutually exclusive: DPG supports just one visualization mode at a time.
-The usage of `run_dpg_custom.py` is similar, but it requires another parameter:
+The usage of `examples/run_dpg_custom.py` is similar, but it requires another parameter:
- `target_column`, which is the name of the column to be used as the target variable;
- while `ds` is the path of the directory where the dataset is.
-### Example `run_dpg_standard.py`
+### Example `examples/run_dpg_standard.py`
Some examples can be appreciated in the `examples` folder: https://github.com/Meta-Group/DPG/tree/main/examples
In particular, the following DPG is obtained by transforming a Random Forest with 5 base learners, trained on Iris dataset.
-The used command is `python run_dpg_standard.py --dataset iris --n_learners 5 --pv 0.001 --t 2 --dir examples --plot --save_plot_dir examples`.
-
+The used command is `dpg --dataset iris --n_learners 5 --pv 0.001 --t 2 --dir examples --plot --save_plot_dir examples`.
+
+ 
+
The following visualizations are obtained using the same parameters as the previous example, but they show two different metrics: _Community_ and _Betweenness centrality_.
-The used command for showing communities is `python run_dpg_standard.py --dataset iris --n_learners 5 --pv 0.001 --t 2 --dir examples --plot --save_plot_dir examples --communities`.
-
+The used command for showing communities is `dpg --dataset iris --n_learners 5 --pv 0.001 --t 2 --dir examples --plot --save_plot_dir examples --communities`.
+
+ 
+
-The used command for showing a specific property is `python run_dpg_standard.py --dataset iris --n_learners 5 --pv 0.001 --t 2 --dir examples --plot --save_plot_dir examples --attribute "Betweenness centrality" --class_flag`.
-
+The used command for showing a specific property is `dpg --dataset iris --n_learners 5 --pv 0.001 --t 2 --dir examples --plot --save_plot_dir examples --attribute "Betweenness centrality" --class_flag`.
+
+ 
+
***
## Citation
diff --git a/config.yaml b/config.yaml
index 4eb3d6e..735a53f 100644
--- a/config.yaml
+++ b/config.yaml
@@ -1,9 +1,12 @@
dpg:
- default:
- perc_var: 0.000000001
- decimal_threshold: 6
- n_jobs: -1
- visualization:
+ default:
+ perc_var: 0.000000001
+ decimal_threshold: 6
+ n_jobs: -1
+ graph_construction:
+ mode: "aggregated_transitions"
+ context_order: 1
+ visualization:
graph_attrs:
bgcolor: "white"
rankdir: "R"
diff --git a/docs/conf.py b/docs/conf.py
index 8b3fc4d..9101a44 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -14,7 +14,7 @@
project = "DPG"
copyright = "2024, Sylvio Barbon Junior, Leonardo Arrighi"
author = "Sylvio Barbon Junior, Leonardo Arrighi"
-release = "0.2.0"
+release = "0.3.0"
# ---------------------------------------------------------------------------
# General configuration
diff --git a/docs/index.md b/docs/index.md
index 977c22d..f8d691d 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -54,8 +54,9 @@ explainer = DPGExplainer(
feature_names=X.columns.tolist(),
target_names=["setosa", "versicolor", "virginica"],
)
-explanation = explainer.fit(X.values)
-explainer.plot(explanation)
+explainer.fit(X.values)
+explanation = explainer.explain_global()
+explainer.plot("iris_dpg", explanation=explanation)
```
## Contents
diff --git a/docs/quickstart.md b/docs/quickstart.md
index 3fda23e..c5d421e 100644
--- a/docs/quickstart.md
+++ b/docs/quickstart.md
@@ -43,7 +43,7 @@ print(explanation.node_metrics.head())
print(explanation.edge_metrics.head())
# 5. Visualise
-explainer.plot(explanation, save_dir="results/")
+explainer.plot("iris_dpg", explanation=explanation, save_dir="results/")
```
## What `DPGExplainer` returns
@@ -125,15 +125,16 @@ dpg = DecisionPredicateGraph(
)
dpg.fit(X)
-dpg.get_trace_consistent_lrc() # {predicate_label: trace-consistent LRC score}
+dpg.get_trace_consistent_lrc() # supported for legacy execution-trace k=1
dpg.get_trace_consistent_trc() # {predicate_label: set of labels observed downstream}
dpg.get_trace_signatures() # list[TraceSignature(signature, predicate_sequence, path_count)]
```
- `get_trace_consistent_lrc()` scores each predicate by how much of the label
- space it was observed to reach *within a single trace*, unlike the
- pooled-graph NetworkX local reaching centrality, which can credit reach
- that only exists after aggregating unrelated traces.
+ space it was observed to reach *within a single trace* for legacy k=1,
+ unlike the pooled-graph NetworkX local reaching centrality, which can credit
+ reach that only exists after aggregating unrelated traces. For contextual
+ graphs, use `get_predicate_lrc(graph)` instead.
- `get_trace_consistent_trc()` returns each predicate's observed downstream
label sets — every member is guaranteed to have co-occurred later in at
least one real execution.
@@ -143,6 +144,11 @@ dpg.get_trace_signatures() # list[TraceSignature(signature, predicate_sequ
These getters return empty containers until `fit()` is called, and are reset
on every refit. Outside `execution_trace` mode they remain empty.
+For contextual graphs (`context_order > 1`),
+`get_trace_consistent_lrc()` is deprecated; use the graph-based
+`get_predicate_lrc(graph)` aggregation instead. The explainer's node metrics
+already apply the appropriate contextual aggregation automatically.
+
**`perc_var` does not filter trace artefacts.** In `execution_trace` mode,
`perc_var` only filters infrequent *edges* out of the pooled visualisation
graph — it never removes a trace signature or downstream relation. A
diff --git a/dpg/__init__.py b/dpg/__init__.py
index aec724e..a3c19d0 100644
--- a/dpg/__init__.py
+++ b/dpg/__init__.py
@@ -13,6 +13,7 @@
classwise_feature_bounds_from_communities,
plot_class_feature_complexity,
plot_dpg,
+ plot_dpg_communities,
plot_dpg_class_bounds_vs_dataset_feature_ranges,
plot_dpg_constraints_overview,
plot_dpg_local_paths_aggregate,
@@ -35,6 +36,7 @@
"DPG_OLIVE_CLASS_PALETTE",
"resolve_theme_context",
"plot_dpg",
+ "plot_dpg_communities",
"plot_dpg_local_paths_aggregate",
"plot_dpg_reg",
"plot_dpg_constraints_overview",
diff --git a/dpg/cli.py b/dpg/cli.py
new file mode 100644
index 0000000..c674430
--- /dev/null
+++ b/dpg/cli.py
@@ -0,0 +1,96 @@
+"""Packaged command-line interface for the standard DPG demonstration."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+from typing import Sequence
+
+from .sklearn_dpg import test_dpg
+
+
+def build_parser() -> argparse.ArgumentParser:
+ """Build the parser used by the installed ``dpg`` command."""
+ parser = argparse.ArgumentParser(
+ description="Train a sklearn tree ensemble and export its DPG metrics."
+ )
+ parser.add_argument(
+ "--dataset",
+ "--ds",
+ dest="dataset",
+ default="iris",
+ help="sklearn dataset name or path to a CSV file",
+ )
+ parser.add_argument(
+ "--target_column",
+ default=None,
+ help="target column when --dataset points to a CSV file",
+ )
+ parser.add_argument("--n_learners", "--l", dest="n_learners", type=int, default=5)
+ parser.add_argument("--model_name", default="RandomForestClassifier")
+ parser.add_argument("--dir", default="examples/results", help="output directory")
+ parser.add_argument("--plot", action="store_true")
+ parser.add_argument("--save_plot_dir", default="examples/results")
+ parser.add_argument("--attribute", default=None)
+ parser.add_argument("--communities", action="store_true")
+ parser.add_argument("--clusters", action="store_true")
+ parser.add_argument("--threshold_clusters", type=float, default=None)
+ parser.add_argument("--t", type=int, default=3, help="threshold decimal precision")
+ parser.add_argument("--class_flag", action="store_true")
+ parser.add_argument("--seed", type=int, default=160898)
+ parser.add_argument(
+ "--pv",
+ type=float,
+ default=1e-9,
+ help="minimum path frequency proportion",
+ )
+ return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+ """Run the standard DPG workflow and write metrics to disk."""
+ args = build_parser().parse_args(argv)
+ output_dir = Path(args.dir)
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ result = test_dpg(
+ datasets=args.dataset,
+ target_column=args.target_column,
+ n_learners=args.n_learners,
+ perc_var=args.pv,
+ decimal_threshold=args.t,
+ n_jobs=-1,
+ model_name=args.model_name,
+ file_name=str(output_dir / f"{Path(args.dataset).stem}_seed{args.seed}_stats.txt"),
+ plot=args.plot,
+ save_plot_dir=args.save_plot_dir,
+ attribute=args.attribute,
+ communities=args.communities,
+ clusters_flag=args.clusters,
+ threshold_clusters=args.threshold_clusters,
+ class_flag=args.class_flag,
+ seed=args.seed,
+ )
+ # ``test_dpg`` signals "insufficient nodes for DPG analysis" with
+ # ``(None, None)`` rather than a bare ``None``; guard on shape, not identity.
+ if result is None or len(result) != 6:
+ return 1
+
+ df, df_edges, graph_metrics, clusters, node_prob, confidence = result
+ stem = Path(args.dataset).stem
+ df.to_csv(output_dir / f"{stem}_seed{args.seed}_node_metrics.csv", index=False)
+ df_edges.to_csv(output_dir / f"{stem}_seed{args.seed}_edge_metrics.csv", index=False)
+ with (output_dir / f"{stem}_seed{args.seed}_dpg_metrics.txt").open("w", encoding="utf-8") as handle:
+ for key, value in graph_metrics.items():
+ handle.write(f"{key}: {value}\n")
+
+ if clusters is not None:
+ with (output_dir / f"{stem}_seed{args.seed}_clusters.txt").open("w", encoding="utf-8") as handle:
+ handle.write(f"Clusters: {clusters}\n")
+ handle.write(f"Probability: {node_prob}\n")
+ handle.write(f"Confidence: {confidence}\n")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/dpg/context_order.py b/dpg/context_order.py
new file mode 100644
index 0000000..1b584ee
--- /dev/null
+++ b/dpg/context_order.py
@@ -0,0 +1,114 @@
+"""Fast resolution of the smallest context order without path enumeration."""
+
+from collections import defaultdict
+import math
+from typing import Iterable, Sequence
+
+
+def _node_windows(sequence: Sequence[str], k: int | float) -> list[object]:
+ """Represent a trace as contextual predicate nodes and class sinks."""
+ nodes: list[object] = []
+ for index, label in enumerate(sequence):
+ if str(label).startswith(("Class ", "Pred ")):
+ nodes.append(("sink", str(label)))
+ continue
+ if math.isinf(k):
+ context = tuple(sequence[: index + 1])
+ else:
+ context = tuple(sequence[max(0, index - int(k) + 1) : index + 1])
+ nodes.append(("ctx", context))
+ return nodes
+
+
+def path_violations(traces: Iterable[Sequence[str]], k: int | float) -> int:
+ """Count pooled-graph paths that are not observed trace prefixes.
+
+ A pooled DFG can be locally consistent while still recombining after
+ several hops. For example, observing ``A-B-C-D`` and ``X-B-C-E`` gives
+ every adjacent transition a witness, but also creates the unobserved
+ path ``A-B-C-E``. This check builds a trie of observed contextual traces,
+ computes each prefix state's future signature bottom-up, and verifies
+ that states merged by the same contextual node have the same future
+ language. It avoids both exponential simple-path enumeration and the
+ potentially large graph-by-trie product.
+
+ The returned value is a diagnostic count, not a count of all phantom
+ paths. A value of zero is the useful acceptance condition: every graph
+ path is represented by an observed trace.
+ """
+ materialized = [tuple(trace) for trace in traces if trace]
+ if not materialized:
+ return 0
+
+ trie_children: list[dict[object, int]] = [{}]
+ trie_terminal: set[int] = set()
+ trie_context: list[object | None] = [None]
+
+ for sequence in materialized:
+ nodes = _node_windows(sequence, k)
+ trie_node = 0
+ for node in nodes:
+ next_node = trie_children[trie_node].get(node)
+ if next_node is None:
+ next_node = len(trie_children)
+ trie_children[trie_node][node] = next_node
+ trie_children.append({})
+ trie_context.append(node)
+ trie_node = next_node
+ trie_terminal.add(trie_node)
+
+ signatures: list[tuple[bool, tuple[tuple[object, object], ...]]] = [
+ (False, ())
+ ] * len(trie_children)
+ for trie_node in range(len(trie_children) - 1, -1, -1):
+ children = tuple(
+ sorted(
+ (
+ (child_label, signatures[child_node])
+ for child_label, child_node in trie_children[trie_node].items()
+ ),
+ key=repr,
+ )
+ )
+ signatures[trie_node] = (trie_node in trie_terminal, children)
+
+ future_by_context: defaultdict[object, set[tuple[bool, tuple]]] = defaultdict(set)
+ for trie_node in range(1, len(trie_children)):
+ context = trie_context[trie_node]
+ assert context is not None
+ future_by_context[context].add(signatures[trie_node])
+
+ return sum(max(0, len(futures) - 1) for futures in future_by_context.values())
+
+
+def resolve_context_order(
+ traces: Iterable[Sequence[str]], max_k: int | None = None
+) -> tuple[int | float, dict[int | float, int]]:
+ """Return the smallest order with no global trace recombination.
+
+ ``max_k`` defaults to the longest observed trace. At that order every
+ observed prefix is distinct, which is a proof-based bound rather than a
+ user-facing cap. When a caller supplies a smaller cap, resolution must
+ still succeed within that cap; otherwise ``ValueError`` is raised instead
+ of returning an order whose history still contains violations.
+ """
+ if max_k is not None and (
+ isinstance(max_k, bool) or not isinstance(max_k, int) or max_k < 1
+ ):
+ raise ValueError("max_k must be a positive integer or None")
+
+ materialized = [tuple(trace) for trace in traces if trace]
+ if not materialized:
+ return 1, {1: 0}
+ if max_k is None:
+ max_k = max(len(trace) for trace in materialized)
+
+ history: dict[int | float, int] = {}
+ for k in range(1, max_k + 1):
+ violations = path_violations(materialized, k)
+ history[k] = violations
+ if violations == 0:
+ return k, history
+ raise ValueError(
+ f"no context order <= max_k={max_k} eliminates all global trace violations"
+ )
diff --git a/dpg/core.py b/dpg/core.py
index 0390133..e1bf56c 100644
--- a/dpg/core.py
+++ b/dpg/core.py
@@ -3,6 +3,7 @@
import re
import math
import os
+import warnings
import numpy as np
from tqdm import tqdm
@@ -32,6 +33,7 @@
GradientBoostingRegressor,
)
from dpg.sklearn_normalizer import SklearnEnsembleNormalizer
+from dpg.context_order import resolve_context_order
DEFAULT_DPG_CONFIG = {
"dpg": {
@@ -42,6 +44,9 @@
},
"graph_construction": {
"mode": "aggregated_transitions",
+ # Keep the 0.2.x graph as the default. DPG-k is opt-in through
+ # context_order="auto" or an explicit order > 1.
+ "context_order": 1,
},
"visualization": {},
}
@@ -133,8 +138,8 @@ def __init__(
# Initialize attributes
self.model = model
- self.feature_names = feature_names
- self.target_names = target_names #TODO create "Class as class name"
+ self.feature_names = list(feature_names)
+ self.target_names = list(target_names) if target_names is not None else None
# Get config values with defaults
dpg_config_section = config.get('dpg', {})
@@ -148,12 +153,22 @@ def __init__(
'mode',
DEFAULT_DPG_CONFIG["dpg"]["graph_construction"]["mode"],
)
+ self.context_order = graph_construction_config.get(
+ "context_order",
+ DEFAULT_DPG_CONFIG["dpg"]["graph_construction"]["context_order"],
+ )
# Validate required config values
if self.perc_var is None:
raise DPGError("perc_var not found in DPG config")
if self.decimal_threshold is None:
raise DPGError("decimal_threshold not found in DPG config")
+ if self.decimal_threshold != "auto" and (
+ isinstance(self.decimal_threshold, bool)
+ or not isinstance(self.decimal_threshold, int)
+ or self.decimal_threshold < 0
+ ):
+ raise DPGError("decimal_threshold must be a non-negative integer or 'auto'")
if self.n_jobs is None:
raise DPGError("n_jobs not found in DPG config")
if self.graph_construction_mode not in self.SUPPORTED_GRAPH_CONSTRUCTION_MODES:
@@ -162,13 +177,31 @@ def __init__(
f"Unsupported graph construction mode '{self.graph_construction_mode}'. "
f"Supported modes are: {supported_modes}"
)
+ if self.context_order != "auto" and (
+ isinstance(self.context_order, bool)
+ or not isinstance(self.context_order, int)
+ or self.context_order <= 0
+ ):
+ raise DPGError("context_order must be a positive integer or 'auto'")
+ if (
+ (
+ self.context_order == "auto"
+ or (isinstance(self.context_order, int) and self.context_order > 1)
+ )
+ and self.graph_construction_mode != "execution_trace"
+ ):
+ raise DPGError(
+ "context_order='auto' or context_order > 1 requires "
+ "mode='execution_trace'"
+ )
print(
"DPG initialized with "
f"perc_var={self.perc_var}, "
f"decimal_threshold={self.decimal_threshold}, "
f"n_jobs={self.n_jobs}, "
- f"graph_construction_mode={self.graph_construction_mode}"
+ f"graph_construction_mode={self.graph_construction_mode}, "
+ f"context_order={self.context_order}"
)
# Store visualization config for use by utils
self.visualization_config = dpg_config_section.get('visualization', DEFAULT_DPG_CONFIG["dpg"]["visualization"])
@@ -177,6 +210,13 @@ def __init__(
self._trace_consistent_lrc: Dict[str, float] = {}
self._trace_consistent_trc: Dict[str, Set[str]] = {}
self._trace_signatures: List[TraceSignature] = []
+ self._resolved_decimal_threshold: Optional[int] = (
+ self.decimal_threshold if isinstance(self.decimal_threshold, int) else None
+ )
+ self._resolved_context_order: int | float = 1
+ self._context_order_history: Dict[int | float, int] = {}
+ self._node_context_by_id: Dict[str, Tuple[str, ...]] = {}
+ self._node_label_by_id: Dict[str, str] = {}
def fit(self, X_train: Any) -> Any:
"""
@@ -199,6 +239,8 @@ def fit(self, X_train: Any) -> Any:
self._trace_consistent_lrc = {}
self._trace_consistent_trc = {}
self._trace_signatures = []
+ self._node_context_by_id = {}
+ self._node_label_by_id = {}
log_df = self._extract_trace_log(X_train)
@@ -206,8 +248,22 @@ def fit(self, X_train: Any) -> Any:
print('Building DPG...')
if self.graph_construction_mode == "execution_trace":
self._build_trace_artifacts(log_df)
- dfg = self.discover_dfg_execution_trace(log_df)
+ traces = self._trace_sequences(log_df)
+ if self.context_order == "auto":
+ self._resolved_context_order, self._context_order_history = resolve_context_order(traces)
+ else:
+ self._resolved_context_order = self.context_order
+ self._context_order_history = {
+ k: self._local_context_violations(traces, k)
+ for k in range(1, int(self.context_order) + 1)
+ }
+ if self._resolved_context_order == 1:
+ dfg = self.discover_dfg_execution_trace(log_df)
+ else:
+ dfg = self.discover_dfg_context(log_df, self._resolved_context_order)
else:
+ self._resolved_context_order = 1
+ self._context_order_history = {1: 0}
if self.perc_var > 0:
log_df = self.filter_log(log_df)
dfg = self.discover_dfg(log_df)
@@ -225,19 +281,85 @@ def _extract_trace_log(self, X_train: Any) -> pd.DataFrame:
Returns:
pd.DataFrame: Raw trace log with case id and event columns
"""
+ self._resolve_decimal_threshold(X_train)
+ X_array = np.asarray(X_train)
# Extract decision paths (parallel or sequential)
if self.n_jobs == 1:
log = Parallel(n_jobs=self.n_jobs)(
- delayed(self.tracing_ensemble)(i, sample) for i, sample in tqdm(list(enumerate(X_train)), total=len(X_train))
+ delayed(self.tracing_ensemble)(i, sample) for i, sample in tqdm(list(enumerate(X_array)), total=len(X_array))
)
else:
log = Parallel(n_jobs=self.n_jobs)(
- delayed(self.tracing_ensemble_parallel)(i, sample) for i, sample in tqdm(list(enumerate(X_train)), total=len(X_train))
+ delayed(self.tracing_ensemble_parallel)(i, sample) for i, sample in tqdm(list(enumerate(X_array)), total=len(X_array))
)
log = [item for sublist in log for item in sublist]
return pd.DataFrame(log, columns=["case:concept:name", "concept:name"])
+ @staticmethod
+ def _decimal_places(value: Any) -> int:
+ """Return decimal places represented by a finite input value."""
+ from decimal import Decimal
+
+ numeric = float(value)
+ if not np.isfinite(numeric):
+ return 0
+ if numeric.is_integer():
+ return 0
+ exponent = Decimal(str(numeric)).as_tuple().exponent
+ return max(0, -int(exponent)) if isinstance(exponent, int) else 0
+
+ def _resolve_decimal_threshold(self, X_train: Any) -> int:
+ """Resolve ``decimal_threshold='auto'`` from data precision and audit it."""
+ if self.decimal_threshold != "auto":
+ self._resolved_decimal_threshold = int(self.decimal_threshold)
+ return self._resolved_decimal_threshold
+
+ values = np.asarray(X_train)
+ precision = 0
+ for value in values.reshape(-1):
+ precision = max(precision, self._decimal_places(value))
+ resolved = precision + 1
+ self._resolved_decimal_threshold = resolved
+
+ offending_features = set()
+ for tree in self.model.estimators_:
+ tree_ = tree.tree_
+ for node, feature_index in enumerate(tree_.feature):
+ if feature_index < 0:
+ continue
+ threshold = float(tree_.threshold[node])
+ if not np.isclose(threshold, round(threshold, resolved), rtol=0.0, atol=1e-12):
+ offending_features.add(self.feature_names[feature_index])
+ if offending_features:
+ names = ", ".join(sorted(offending_features))
+ warnings.warn(
+ "decimal_threshold='auto' found thresholds off the data-derived "
+ f"{resolved}-decimal grid for feature(s): {names}; exact routing "
+ "is retained and only predicate labels are rounded.",
+ RuntimeWarning,
+ stacklevel=2,
+ )
+ return resolved
+
+ def get_decimal_threshold(self) -> int:
+ """Return the effective precision used by the last trace extraction."""
+ if self._resolved_decimal_threshold is None:
+ raise DPGError("decimal_threshold='auto' is resolved when fit() is called")
+ return self._resolved_decimal_threshold
+
+ def _trace_sequences(self, log: Any) -> List[Tuple[str, ...]]:
+ return [
+ tuple(group["concept:name"].tolist())
+ for _, group in log.groupby("case:concept:name", sort=False)
+ ]
+
+ @staticmethod
+ def _local_context_violations(traces: List[Tuple[str, ...]], k: int) -> int:
+ from dpg.context_order import path_violations
+
+ return path_violations(traces, k)
+
def _leaf_class_label(self, tree_index: int, tree_: Any, node_index: int) -> str:
"""Return the class label for a classifier leaf node."""
gb_class_index = SklearnEnsembleNormalizer.get_tree_class_index(self.model, tree_index)
@@ -266,40 +388,14 @@ def tracing_ensemble(self, case_id: int, sample: Any) -> Generator[List[str], No
Yields:
List[str]: Path segments as [prefix, decision/prediction]
"""
- is_regressor = isinstance(
- self.model,
- (
- RandomForestRegressor,
- ExtraTreesRegressor,
- AdaBoostRegressor,
- GradientBoostingRegressor,
- ),
+ label_extractor = (
+ self._trace_tree_labels_legacy
+ if self.graph_construction_mode == "aggregated_transitions"
+ else self._trace_tree_labels
)
- sample = sample.reshape(-1)
for i, tree in enumerate(self.model.estimators_):
- tree_ = tree.tree_
- node_index = 0
prefix = f"sample{case_id}_dt{i}"
- while True:
- left = tree_.children_left[node_index]
- right = tree_.children_right[node_index]
- if left == right:
- if is_regressor:
- pred = round(tree_.value[node_index][0][0], 2)
- yield [prefix, f"Pred {pred}"]
- else:
- yield [prefix, self._leaf_class_label(i, tree_, node_index)]
- break
- feature_index = tree_.feature[node_index]
- threshold = round(tree_.threshold[node_index], self.decimal_threshold)
- feature_name = self.feature_names[feature_index]
- sample_val = sample[feature_index]
- if sample_val <= threshold:
- condition = f"{feature_name} <= {threshold}"
- node_index = left
- else:
- condition = f"{feature_name} > {threshold}"
- node_index = right
+ for condition in label_extractor(i, tree, sample):
yield [prefix, condition]
def tracing_ensemble_parallel(self, case_id: int, sample: Any) -> List[List[str]]:
@@ -314,43 +410,86 @@ def tracing_ensemble_parallel(self, case_id: int, sample: Any) -> List[List[str]
List of ``[prefix, event]`` pairs representing the full decision path
across all trees in the ensemble.
"""
- is_regressor = isinstance(
- self.model,
- (
- RandomForestRegressor,
- ExtraTreesRegressor,
- AdaBoostRegressor,
- GradientBoostingRegressor,
- ),
+ label_extractor = (
+ self._trace_tree_labels_legacy
+ if self.graph_construction_mode == "aggregated_transitions"
+ else self._trace_tree_labels
)
- sample = sample.reshape(-1)
result = []
for i, tree in enumerate(self.model.estimators_):
- tree_ = tree.tree_
- node_index = 0
prefix = f"sample{case_id}_dt{i}"
- while True:
- left = tree_.children_left[node_index]
- right = tree_.children_right[node_index]
- if left == right:
- if is_regressor:
- pred = round(tree_.value[node_index][0][0], 2)
- result.append([prefix, f"Pred {pred}"])
- else:
- result.append([prefix, self._leaf_class_label(i, tree_, node_index)])
- break
- feature_index = tree_.feature[node_index]
- threshold = round(tree_.threshold[node_index], self.decimal_threshold)
- feature_name = self.feature_names[feature_index]
- sample_val = sample[feature_index]
- if sample_val <= threshold:
- condition = f"{feature_name} <= {threshold}"
- node_index = left
- else:
- condition = f"{feature_name} > {threshold}"
- node_index = right
+ for condition in label_extractor(i, tree, sample):
result.append([prefix, condition])
return result
+
+ def _trace_tree_labels_legacy(
+ self, tree_index: int, tree: Any, sample: Any
+ ) -> List[str]:
+ """Return labels using the pre-0.3.0 rounded-threshold traversal.
+
+ The aggregated-transitions graph historically rounded each tree
+ threshold before selecting a child. Keep that behavior for the
+ legacy path so its graph weights remain backward-compatible. The
+ execution-trace path uses :meth:`_trace_tree_labels`, which follows
+ sklearn's native routing exactly.
+ """
+ sample_array = np.asarray(sample).reshape(-1)
+ tree_ = tree.tree_
+ node_index = 0
+ labels: List[str] = []
+ effective_decimal = self.get_decimal_threshold()
+
+ while True:
+ left = int(tree_.children_left[node_index])
+ right = int(tree_.children_right[node_index])
+ if left == right:
+ if is_regressor(self.model):
+ pred = round(float(tree_.value[node_index][0][0]), 2)
+ labels.append(f"Pred {pred}")
+ else:
+ labels.append(self._leaf_class_label(tree_index, tree_, node_index))
+ return labels
+
+ feature_index = int(tree_.feature[node_index])
+ threshold = round(float(tree_.threshold[node_index]), effective_decimal)
+ feature_name = self.feature_names[feature_index]
+ if sample_array[feature_index] <= threshold:
+ labels.append(f"{feature_name} <= {threshold}")
+ node_index = left
+ else:
+ labels.append(f"{feature_name} > {threshold}")
+ node_index = right
+
+ def _trace_tree_labels(self, tree_index: int, tree: Any, sample: Any) -> List[str]:
+ """Return the executed labels using sklearn's native routing decisions.
+
+ Threshold rounding is deliberately applied only while formatting the
+ predicate. The child branch comes from ``decision_path`` so a rounded
+ label can never change the recorded execution path.
+ """
+ sample_array = np.asarray(sample).reshape(1, -1)
+ tree_ = tree.tree_
+ decision_path = tree.decision_path(sample_array)
+ path = decision_path.indices[decision_path.indptr[0] : decision_path.indptr[1]]
+ leaf_id = int(tree.apply(sample_array)[0])
+ labels: List[str] = []
+ effective_decimal = self.get_decimal_threshold()
+
+ for position, node_index in enumerate(path):
+ if int(node_index) == leaf_id:
+ break
+ feature_index = int(tree_.feature[node_index])
+ threshold = round(float(tree_.threshold[node_index]), effective_decimal)
+ went_left = int(path[position + 1]) == int(tree_.children_left[node_index])
+ operator = "<=" if went_left else ">"
+ labels.append(f"{self.feature_names[feature_index]} {operator} {threshold}")
+
+ if is_regressor(self.model):
+ pred = round(float(tree_.value[leaf_id][0][0]), 2)
+ labels.append(f"Pred {pred}")
+ else:
+ labels.append(self._leaf_class_label(tree_index, tree_, leaf_id))
+ return labels
def filter_log(self, log: Any) -> Any:
@@ -396,7 +535,10 @@ def discover_dfg(self, log: Any) -> Dict[Tuple[str, str], int]:
grouped = log.groupby("case:concept:name", sort=False)
for case, trace_df in tqdm(grouped, desc="Processing cases", total=len(cases)):
- trace_df = trace_df.sort_values(by="case:concept:name")
+ # Rows within each case already follow execution order. Sorting by
+ # the case identifier is both redundant and unsafe here: the key is
+ # constant within a group, and pandas' default sort is not stable,
+ # so long traces can be silently rearranged.
concepts = trace_df["concept:name"].values
for i in range(len(concepts) - 1):
key = (concepts[i], concepts[i + 1])
@@ -428,6 +570,101 @@ def discover_dfg_execution_trace(self, log: Any) -> Dict[Tuple[str, str], int]:
if count >= min_count
}
+ @staticmethod
+ def _context_node(label_sequence: Tuple[str, ...], index: int, k: int) -> Any:
+ label = label_sequence[index]
+ if str(label).startswith(("Class ", "Pred ")):
+ # Terminal outcomes are deliberately never contextualised. A
+ # shared sink cannot create a new path because it has no outgoing
+ # edges, and keeps one sink per class/prediction invariant.
+ return ("sink", str(label))
+ return (
+ "ctx",
+ tuple(label_sequence[max(0, index - k + 1) : index + 1]),
+ )
+
+ @staticmethod
+ def _context_node_info(node: Any) -> Tuple[str, Tuple[str, ...]]:
+ kind, value = node
+ if kind == "sink":
+ return str(value), ()
+ context = tuple(value)
+ return context[-1], context
+
+ def discover_dfg_context(self, log: Any, context_order: int) -> Dict[Tuple[Any, Any], int]:
+ """Build a context-aware DFG from observed execution traces.
+
+ The graph is still pooled into one DPG. Only non-terminal predicate
+ identity changes: it is the last ``context_order`` executed labels.
+ Class outcomes remain one shared sink per class.
+ """
+ if context_order <= 1:
+ return self.discover_dfg_execution_trace(log)
+
+ dfg: Dict[Tuple[Any, Any], int] = {}
+ for _, trace_df in log.groupby("case:concept:name", sort=False):
+ labels = tuple(trace_df["concept:name"].tolist())
+ nodes = [self._context_node(labels, i, context_order) for i in range(len(labels))]
+ for source, target in zip(nodes, nodes[1:]):
+ edge = (source, target)
+ dfg[edge] = dfg.get(edge, 0) + 1
+
+ if self.perc_var <= 0:
+ return dfg
+ min_count = log["case:concept:name"].nunique() * self.perc_var
+ return {edge: count for edge, count in dfg.items() if count >= min_count}
+
+ def get_context_order(self) -> int | float:
+ """Return the effective context order from the last ``fit`` call."""
+ return self._resolved_context_order
+
+ def get_context_order_history(self) -> Dict[int | float, int]:
+ """Return local recombination violations measured for each tested k."""
+ return dict(self._context_order_history)
+
+ def get_node_context(self, node: Any) -> Tuple[str, ...]:
+ """Return the predicate context associated with a graph node.
+
+ Sinks and all nodes in a k=1 graph return an empty tuple. ``node`` is
+ the node identifier returned by :meth:`to_networkx`.
+ """
+ if hasattr(node, "__str__"):
+ return tuple(self._node_context_by_id.get(str(node), ()))
+ return ()
+
+ def get_node_ids_for_trace(self, labels: Iterable[str]) -> List[str]:
+ """Map one executed label sequence to fitted graph node identifiers."""
+ labels_tuple = tuple(labels)
+ k = self.get_context_order()
+ if k == 1:
+ keys = list(labels_tuple)
+ else:
+ keys = [
+ self._context_node(labels_tuple, index, int(k))
+ for index in range(len(labels_tuple))
+ ]
+ return [self._node_id_for_key(key) for key in keys]
+
+ def get_predicate_lrc(self, graph: Any) -> Dict[str, float]:
+ """Aggregate unweighted node LRC scores back to predicate labels.
+
+ At k>1 a predicate can occupy multiple contextual nodes. The shipped
+ aggregation is a sum, so predicates receive credit for every context
+ in which they occur.
+ """
+ scores: Dict[str, float] = defaultdict(float)
+ for node, data in graph.nodes(data=True):
+ label = data.get("predicate")
+ if label is None or not self._is_predicate_label(label):
+ continue
+ scores[label] += float(nx.local_reaching_centrality(graph, node, weight=None))
+ return dict(scores)
+
+ @staticmethod
+ def _node_id_for_key(key: Any) -> str:
+ stable_key = key if isinstance(key, str) else repr(key)
+ return str(int(hashlib.sha1(stable_key.encode()).hexdigest(), 16))
+
_PREDICATE_LABEL_RE = re.compile(
r"^\s*(.+?)\s*(<=|>)\s*[-+]?\d*\.?\d+(?:[eE][-+]?\d+)?\s*$"
)
@@ -508,6 +745,11 @@ def get_trace_consistent_lrc(self) -> Dict[str, float]:
"""
Return trace-consistent local-reaching-centrality-like scores.
+ .. deprecated:: 0.3.0
+ For contextual graphs (``context_order > 1``), use
+ :meth:`get_predicate_lrc` on the fitted graph. This getter remains
+ supported for legacy k=1 execution-trace graphs.
+
For each predicate label, the score is the fraction (over all distinct
labels observed across every trace) of labels found downstream of it
within at least one single observed sample-tree execution. Unlike the
@@ -520,6 +762,13 @@ def get_trace_consistent_lrc(self) -> Dict[str, float]:
LRC score. Empty before ``fit()`` or outside
``graph_construction_mode="execution_trace"``.
"""
+ if self.get_context_order() > 1:
+ warnings.warn(
+ "get_trace_consistent_lrc() is deprecated for context_order > 1; "
+ "use get_predicate_lrc(graph) instead",
+ DeprecationWarning,
+ stacklevel=2,
+ )
return dict(self._trace_consistent_lrc)
def get_trace_consistent_trc(self) -> Dict[str, Tuple[str, ...]]:
@@ -552,7 +801,7 @@ def get_trace_signatures(self) -> List[TraceSignature]:
"""
return list(self._trace_signatures)
- def generate_dot(self, dfg: Dict[Tuple[str, str], int]) -> Any:
+ def generate_dot(self, dfg: Dict[Tuple[Any, Any], int]) -> Any:
"""
Convert frequency graph to Graphviz format.
@@ -601,21 +850,33 @@ def _escape_dot_label(label: str) -> str:
)
added_nodes = set()
- for k, v in sorted(dfg.items(), key=lambda item: item[1]):
- for activity in k:
- if activity not in added_nodes:
- dot.node(
- str(int(hashlib.sha1(activity.encode()).hexdigest(), 16)),
- label=_escape_dot_label(activity),
- style="filled",
- fontsize="20",
- fillcolor=default_fillcolor,
- )
- added_nodes.add(activity)
+ for edge, weight in sorted(dfg.items(), key=lambda item: item[1]):
+ source, target = edge
+ for node_key in (source, target):
+ node_id = self._node_id_for_key(node_key)
+ if node_id in added_nodes:
+ continue
+ if isinstance(node_key, str):
+ label, context = node_key, ()
+ else:
+ label, context = self._context_node_info(node_key)
+ self._node_label_by_id[node_id] = label
+ self._node_context_by_id[node_id] = context
+ tooltip = " > ".join(context) if context else label
+ dot.node(
+ node_id,
+ label=_escape_dot_label(label),
+ tooltip=_escape_dot_label(tooltip),
+ dpg_context_order=str(self.get_context_order()),
+ style="filled",
+ fontsize="20",
+ fillcolor=default_fillcolor,
+ )
+ added_nodes.add(node_id)
dot.edge(
- str(int(hashlib.sha1(k[0].encode()).hexdigest(), 16)),
- str(int(hashlib.sha1(k[1].encode()).hexdigest(), 16)),
- label=str(v),
+ self._node_id_for_key(source),
+ self._node_id_for_key(target),
+ label=str(weight),
penwidth="1",
fontsize="18"
)
@@ -632,35 +893,43 @@ def to_networkx(self, graphviz_graph: Any) -> Tuple[Any, List[List[str]]]:
Tuple[nx.DiGraph, List]: NetworkX graph and node metadata
"""
networkx_graph = nx.DiGraph()
- nodes_list = []
- edges = []
- weights = {}
- for edge in graphviz_graph.body:
- if "->" in edge:
- src, dest = edge.split("->")
- src = src.strip()
- dest = dest.split(" [label=")[0].strip()
- weight = None
- if "[label=" in edge:
- attr = edge.split("[label=")[1].split("]")[0].split(" ")[0]
- weight = float(attr) if attr.replace(".", "").isdigit() else None
- weights[(src, dest)] = weight
- edges.append((src, dest))
- if "[label=" in edge:
- id, desc = edge.split("[label=")
- id = id.replace("\t", "").replace(" ", "")
- # Extract label value safely, ignoring other attributes
- m = re.search(r'label="([^"]*)"', edge)
- if m:
- desc = m.group(1)
- else:
- # Fallback for unquoted labels: label=VALUE (no spaces)
- m = re.search(r'label=([^\\s\\]]+)', edge)
- desc = m.group(1) if m else ""
- nodes_list.append([id, desc])
+ nodes_list: List[List[str]] = []
+ edges: List[Tuple[str, str]] = []
+ weights: Dict[Tuple[str, str], float] = {}
+ node_records: Dict[str, Tuple[str, Tuple[str, ...]]] = {}
+
+ for line in graphviz_graph.body:
+ if "->" in line:
+ match = re.search(r"^\s*([^\s]+)\s*->\s*([^\s]+).*?label=([0-9.eE+-]+)", line)
+ if match:
+ src, dest, weight = match.groups()
+ edges.append((src, dest))
+ weights[(src, dest)] = float(weight)
+ continue
+
+ match = re.search(r'^\s*([^\s]+)\s+\[label="(.*?)"(.*?)\]', line)
+ if not match:
+ continue
+ node_id, label, attributes = match.groups()
+ context = self._node_context_by_id.get(node_id, ())
+ order_match = re.search(r"dpg_context_order=([^,\]]+)", attributes)
+ order = self.get_context_order()
+ if order_match:
+ try:
+ order = float(order_match.group(1).strip('"'))
+ if order.is_integer():
+ order = int(order)
+ except ValueError:
+ pass
+ node_records[node_id] = (label, context)
+ nodes_list.append([node_id, label])
+ networkx_graph.add_node(
+ node_id,
+ predicate=label,
+ context=context,
+ context_order=order,
+ )
+
for src, dest in edges:
- if (src, dest) in weights:
- networkx_graph.add_edge(src, dest, weight=weights[(src, dest)])
- else:
- networkx_graph.add_edge(src, dest)
+ networkx_graph.add_edge(src, dest, weight=weights[(src, dest)])
return networkx_graph, sorted(nodes_list, key=lambda x: x[0])
diff --git a/dpg/explainer.py b/dpg/explainer.py
index 4a930ce..c09acca 100644
--- a/dpg/explainer.py
+++ b/dpg/explainer.py
@@ -132,6 +132,15 @@ def __init__(
config_file: str = "config.yaml",
dpg_config: Optional[Dict[str, Any]] = None,
) -> None:
+ # Keep a reference to the original (un-normalized) model so that
+ # ``predict()`` still goes through sklearn's own predict path.
+ # ``DecisionPredicateGraph`` shallow-copies the model into a DPG-
+ # normalized variant for its own tree traversal; for
+ # ``GradientBoostingClassifier`` that normalized copy flattens
+ # ``estimators_`` from a 2D ndarray to a 1D list, which would
+ # break sklearn's internal ``self.estimators_[0, 0]`` indexing
+ # inside ``.predict()``.
+ self._original_model = model
self._builder = DecisionPredicateGraph(
model=model,
feature_names=list(feature_names),
@@ -237,7 +246,6 @@ def explain_local(
f"Sample has {sample_array.shape[0]} features, expected {expected_features}."
)
- node_lookup = {label: node_id for node_id, label in self._nodes}
node_metrics_lookup = self._get_node_metrics_lookup()
tree_paths = []
@@ -248,7 +256,6 @@ def explain_local(
sample=sample_array,
sample_id=sample_id,
tree_index=tree_index,
- node_lookup=node_lookup,
node_metrics_lookup=node_metrics_lookup,
validate_graph=validate_graph,
)
@@ -486,9 +493,9 @@ def evaluate_faithfulness(
true_label_normalized = self._normalize_prediction_label(y_true_seq[idx])
if isinstance(X_eval, pd.DataFrame):
- model_pred_raw = self._builder.model.predict(row_for_predict.to_frame().T)[0]
+ model_pred_raw = self._original_model.predict(row_for_predict.to_frame().T)[0]
else:
- model_pred_raw = self._builder.model.predict(np.asarray(row_values).reshape(1, -1))[0]
+ model_pred_raw = self._original_model.predict(np.asarray(row_values).reshape(1, -1))[0]
model_pred = self._normalize_prediction_label(model_pred_raw)
try:
@@ -624,7 +631,6 @@ def _trace_tree_path(
sample: np.ndarray,
sample_id: int,
tree_index: int,
- node_lookup: Dict[str, str],
node_metrics_lookup: Dict[str, Dict[str, Any]],
validate_graph: bool,
) -> DPGTreePathExplanation:
@@ -633,39 +639,34 @@ def _trace_tree_path(
(RandomForestRegressor, ExtraTreesRegressor, AdaBoostRegressor),
)
tree_ = tree.tree_
- node_index = 0
+ sample_array = np.asarray(sample).reshape(1, -1)
+ indicator = tree.decision_path(sample_array)
+ path = indicator.indices[indicator.indptr[0] : indicator.indptr[1]]
+ leaf_id = int(tree.apply(sample_array)[0])
tree_prefix = f"sample{sample_id}_dt{tree_index}"
labels: List[str] = []
predicate_truths: List[bool] = []
- while True:
- left = tree_.children_left[node_index]
- right = tree_.children_right[node_index]
- if left == right:
- if is_regressor:
- pred = round(tree_.value[node_index][0][0], 2)
- labels.append(f"Pred {pred}")
- else:
- labels.append(self._leaf_class_label(tree_index, tree_, node_index))
+ for position, node_index in enumerate(path):
+ if int(node_index) == leaf_id:
break
-
- feature_index = tree_.feature[node_index]
- threshold = round(tree_.threshold[node_index], self._builder.decimal_threshold)
+ feature_index = int(tree_.feature[node_index])
+ threshold = round(float(tree_.threshold[node_index]), self._builder.get_decimal_threshold())
feature_name = self._builder.feature_names[feature_index]
- sample_val = sample[feature_index]
- if sample_val <= threshold:
- labels.append(f"{feature_name} <= {threshold}")
- predicate_truths.append(True)
- node_index = left
- else:
- labels.append(f"{feature_name} > {threshold}")
- predicate_truths.append(True)
- node_index = right
+ went_left = int(path[position + 1]) == int(tree_.children_left[node_index])
+ labels.append(f"{feature_name} {'<=' if went_left else '>'} {threshold}")
+ predicate_truths.append(True)
+
+ if is_regressor:
+ pred = round(float(tree_.value[leaf_id][0][0]), 2)
+ labels.append(f"Pred {pred}")
+ else:
+ labels.append(self._leaf_class_label(tree_index, tree_, leaf_id))
- native_node_ids = [self._label_to_node_id(label) for label in labels]
+ native_node_ids = self._builder.get_node_ids_for_trace(labels)
node_ids = [
- node_lookup.get(label) if validate_graph else native_node_id
- for label, native_node_id in zip(labels, native_node_ids)
+ native_node_id if (not validate_graph or native_node_id in self._graph) else None
+ for native_node_id in native_node_ids
]
edge_exists = []
@@ -743,8 +744,13 @@ def _normalize_class_vote_label(label: str) -> str:
def _get_node_metrics(self) -> Any:
if self._node_metrics is None:
trace_lrc_by_label = None
- if self._builder.graph_construction_mode == "execution_trace":
+ if (
+ self._builder.graph_construction_mode == "execution_trace"
+ and self._builder.get_context_order() == 1
+ ):
trace_lrc_by_label = self._builder.get_trace_consistent_lrc()
+ elif self._builder.get_context_order() > 1:
+ trace_lrc_by_label = self._builder.get_predicate_lrc(self._graph)
self._node_metrics = NodeMetrics.extract_node_metrics(
self._graph, self._nodes, trace_lrc_by_label=trace_lrc_by_label
)
@@ -997,38 +1003,33 @@ def _trace_execution_labels_for_tree(
(RandomForestRegressor, ExtraTreesRegressor, AdaBoostRegressor),
)
tree_ = tree.tree_
- node_index = 0
+ sample_array = np.asarray(sample).reshape(1, -1)
+ indicator = tree.decision_path(sample_array)
+ path = indicator.indices[indicator.indptr[0] : indicator.indptr[1]]
+ leaf_id = int(tree.apply(sample_array)[0])
labels: List[str] = []
- while True:
- left = tree_.children_left[node_index]
- right = tree_.children_right[node_index]
- if left == right:
- if is_regressor:
- pred = round(tree_.value[node_index][0][0], 2)
- labels.append(f"Pred {pred}")
- else:
- if tree_index is None:
- pred_class = int(tree_.value[node_index].argmax())
- if self._builder.target_names is not None:
- pred_class = self._builder.target_names[pred_class]
- elif hasattr(self._builder.model, "classes_"):
- pred_class = self._builder.model.classes_[pred_class]
- labels.append(f"Class {pred_class}")
- else:
- labels.append(self._leaf_class_label(tree_index, tree_, node_index))
+ for position, node_index in enumerate(path):
+ if int(node_index) == leaf_id:
break
-
- feature_index = tree_.feature[node_index]
- threshold = round(tree_.threshold[node_index], self._builder.decimal_threshold)
+ feature_index = int(tree_.feature[node_index])
+ threshold = round(float(tree_.threshold[node_index]), self._builder.get_decimal_threshold())
feature_name = self._builder.feature_names[feature_index]
- sample_val = sample[feature_index]
- if sample_val <= threshold:
- labels.append(f"{feature_name} <= {threshold}")
- node_index = left
- else:
- labels.append(f"{feature_name} > {threshold}")
- node_index = right
+ went_left = int(path[position + 1]) == int(tree_.children_left[node_index])
+ labels.append(f"{feature_name} {'<=' if went_left else '>'} {threshold}")
+
+ if is_regressor:
+ pred = round(float(tree_.value[leaf_id][0][0]), 2)
+ labels.append(f"Pred {pred}")
+ elif tree_index is None:
+ pred_class = int(tree_.value[leaf_id].argmax())
+ if self._builder.target_names is not None:
+ pred_class = self._builder.target_names[pred_class]
+ elif hasattr(self._builder.model, "classes_"):
+ pred_class = self._builder.model.classes_[pred_class]
+ labels.append(f"Class {pred_class}")
+ else:
+ labels.append(self._leaf_class_label(tree_index, tree_, leaf_id))
return labels
diff --git a/dpg/sklearn_dpg.py b/dpg/sklearn_dpg.py
index 222b9ef..ba7ff0e 100644
--- a/dpg/sklearn_dpg.py
+++ b/dpg/sklearn_dpg.py
@@ -173,11 +173,20 @@ def test_dpg(datasets: str,
f.write(f"Model: {model_name}\nMSE: {metrics['mse']:.2f}\n")
# DPG extraction
- dpg = DecisionPredicateGraph(
- model=model,
- feature_names=features,
- target_names=np.unique(target).astype(str).tolist()
- )
+ dpg = DecisionPredicateGraph(
+ model=model,
+ feature_names=features,
+ target_names=np.unique(target).astype(str).tolist(),
+ dpg_config={
+ "dpg": {
+ "default": {
+ "perc_var": perc_var,
+ "decimal_threshold": decimal_threshold,
+ "n_jobs": n_jobs,
+ }
+ }
+ },
+ )
dot = dpg.fit(X_train)
# Convert to NetworkX and get metrics
diff --git a/metrics/graph.py b/metrics/graph.py
index 4f1546f..210be23 100644
--- a/metrics/graph.py
+++ b/metrics/graph.py
@@ -236,6 +236,19 @@ def extract_communities(cls, dpg_model, df_node_metrics, nodes_list, threshold_c
node_to_label = df_node_metrics.set_index('Node')['Label'].to_dict()
class_nodes = {i[0] : i[1] for i in nodes_list if 'Class' in i[1]}
+ if not class_nodes:
+ # The absorbing-chain clustering below requires at least one
+ # classifier sink ("Class ..." leaf) to anchor the Markov chain;
+ # a regression DPG has "Pred ..." leaves instead and none is a
+ # class sink, which makes every node transient and (I - Q)
+ # singular. Regression community/class-boundary extraction is
+ # out of scope for 0.3.0 (see CHANGELOG); raise a clear error
+ # instead of letting numpy fail with an opaque LinAlgError.
+ raise ValueError(
+ "extract_communities requires a classifier DPG with at least one "
+ "'Class ' sink node; regression DPGs ('Pred ' leaves) are not "
+ "supported by this community extraction in 0.3.0."
+ )
clusters, node_prob, confidence = cls.clustering(dpg_model, class_nodes, threshold_clusters)
clusters_labels = {k: [node_to_label.get(n, n) for n in v] for k, v in clusters.items()}
diff --git a/pyproject.toml b/pyproject.toml
index 6acd1ea..49fd346 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[tool.poetry]
name = "dpg"
-version = "0.2.0"
+version = "0.3.0"
description = "A Python library for extracting Decision Predicate Graphs and global explanations from ensemble models."
authors = ["Sylvio Barbon Junior ", "Leonardo Arrighi "]
license = "MIT"
@@ -101,4 +101,25 @@ requires = ["poetry-core>=1.9.0"]
build-backend = "poetry.core.masonry.api"
[tool.poetry.scripts]
-dpg = "scripts.run_dpg_standard:main"
+dpg = "dpg.cli:main"
+
+# ---------------------------------------------------------------------------
+# Pytest configuration: silence third-party deprecation noise from
+# matplotlib, sklearn, numpy, pyparsing, and pytest's own deprecation
+# warnings about class-scoped fixtures (see TestTestDpgIris et al. for the
+# pre-existing pattern we do not refactor here).
+# ---------------------------------------------------------------------------
+[tool.pytest.ini_options]
+filterwarnings = [
+ "ignore::DeprecationWarning:pyparsing",
+ "ignore::DeprecationWarning:matplotlib",
+ "ignore::DeprecationWarning:sklearn",
+ "ignore:.*feature names.*:UserWarning:sklearn",
+ "ignore:.*Setting the shape on a NumPy array.*:DeprecationWarning",
+ "ignore::PendingDeprecationWarning",
+ "ignore::FutureWarning:sklearn",
+ # Pytest's own deprecation about class-scoped fixtures: keep as a
+ # warning (will become an error in pytest 10) but stop polluting
+ # test output until we refactor those fixtures.
+ "ignore::pytest.PytestRemovedIn10Warning",
+]
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 0000000..a422b3d
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,142 @@
+"""Tests for the packaged command-line entry point."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pandas as pd
+
+from dpg.cli import build_parser, main
+
+
+def test_cli_parser_matches_documented_defaults():
+ args = build_parser().parse_args([])
+
+ assert args.dataset == "iris"
+ assert args.n_learners == 5
+ assert args.model_name == "RandomForestClassifier"
+ assert args.seed == 160898
+ assert args.pv == 1e-9
+
+
+def test_cli_reports_failure_on_insufficient_nodes(tmp_path):
+ """``test_dpg`` signals this case with ``(None, None)``, not a bare ``None``."""
+ with patch("dpg.cli.test_dpg", return_value=(None, None)):
+ assert main(["--dataset", "iris", "--dir", str(tmp_path)]) == 1
+
+
+# ---------------------------------------------------------------------------
+# CLI parser coverage
+# ---------------------------------------------------------------------------
+
+
+class TestCLIParser:
+ """Cover aliases and defaults the public CLI exposes to users."""
+
+ def test_long_and_short_aliases_resolve_to_same_dest(self):
+ long_args = build_parser().parse_args(
+ ["--dataset", "iris", "--n_learners", "7", "--pv", "0.5"]
+ )
+ short_args = build_parser().parse_args(
+ ["--ds", "iris", "--l", "7", "--pv", "0.5"]
+ )
+ assert long_args.dataset == short_args.dataset == "iris"
+ assert long_args.n_learners == short_args.n_learners == 7
+ assert long_args.pv == short_args.pv == 0.5
+
+ def test_store_true_flags_default_to_false(self):
+ args = build_parser().parse_args([])
+ for store_true_field in ("plot", "communities", "clusters", "class_flag"):
+ assert getattr(args, store_true_field) is False
+
+ def test_optional_fields_default_to_none(self):
+ args = build_parser().parse_args([])
+ assert args.target_column is None
+ assert args.attribute is None
+ assert args.threshold_clusters is None
+
+
+# ---------------------------------------------------------------------------
+# CLI main() -- happy path
+# ---------------------------------------------------------------------------
+
+
+class TestCLIMainSuccess:
+ """When ``test_dpg`` returns a valid 6-tuple, ``main`` persists it."""
+
+ def test_main_writes_node_edge_and_graph_metric_files(self, tmp_path: Path):
+ node_df = pd.DataFrame({"Node": ["n1"], "Label": ["Class 0"]})
+ edge_df = pd.DataFrame({"Source": ["a"], "Target": ["b"], "Frequency": [1]})
+ graph_metrics = {"nodes": 1, "edges": 1}
+
+ with patch(
+ "dpg.cli.test_dpg",
+ return_value=(node_df, edge_df, graph_metrics, None, None, None),
+ ):
+ exit_code = main(["--dataset", "iris", "--dir", str(tmp_path)])
+
+ assert exit_code == 0
+ node_csv = tmp_path / "iris_seed160898_node_metrics.csv"
+ edge_csv = tmp_path / "iris_seed160898_edge_metrics.csv"
+ dpg_txt = tmp_path / "iris_seed160898_dpg_metrics.txt"
+ assert node_csv.exists()
+ assert edge_csv.exists()
+ assert dpg_txt.exists()
+ assert "nodes: 1" in dpg_txt.read_text()
+
+ def test_main_writes_clusters_file_when_provided(self, tmp_path: Path):
+ node_df = pd.DataFrame({"Node": ["n1"], "Label": ["Class 0"]})
+ edge_df = pd.DataFrame({"Source": ["a"], "Target": ["b"], "Frequency": [1]})
+ graph_metrics = {"nodes": 1}
+ clusters = {"Clusters": {"c0": ["n1"]}}
+ node_prob = {"n1": 1.0}
+ confidence = {"n1": 0.5}
+
+ with patch(
+ "dpg.cli.test_dpg",
+ return_value=(node_df, edge_df, graph_metrics, clusters, node_prob, confidence),
+ ):
+ exit_code = main(["--dataset", "iris", "--dir", str(tmp_path)])
+
+ assert exit_code == 0
+ clusters_file = tmp_path / "iris_seed160898_clusters.txt"
+ assert clusters_file.exists()
+ text = clusters_file.read_text()
+ assert "Clusters" in text
+ assert "Probability" in text
+ assert "Confidence" in text
+
+ def test_main_creates_missing_output_directory(self, tmp_path: Path):
+ out = tmp_path / "nested" / "deep" / "results"
+ node_df = pd.DataFrame({"Node": ["n1"], "Label": ["Class 0"]})
+ edge_df = pd.DataFrame({"Source": ["a"], "Target": ["b"], "Frequency": [1]})
+
+ with patch(
+ "dpg.cli.test_dpg", return_value=(node_df, edge_df, {}, None, None, None)
+ ):
+ exit_code = main(["--dataset", "iris", "--dir", str(out)])
+
+ assert exit_code == 0
+ assert out.exists()
+
+
+# ---------------------------------------------------------------------------
+# CLI main() -- failure handling
+# ---------------------------------------------------------------------------
+
+
+class TestCLIMainFailure:
+ """When ``test_dpg`` signals an error, ``main`` returns 1 and writes nothing."""
+
+ def test_main_returns_one_on_none_result(self, tmp_path: Path):
+ with patch("dpg.cli.test_dpg", return_value=None):
+ assert main(["--dataset", "iris", "--dir", str(tmp_path)]) == 1
+
+ def test_main_returns_one_on_wrong_shape_result(self, tmp_path: Path):
+ # 5-tuple instead of the expected 6-tuple.
+ with patch("dpg.cli.test_dpg", return_value=(None, None, None, None, None)):
+ assert main(["--dataset", "iris", "--dir", str(tmp_path)]) == 1
+
+ def test_main_does_not_write_files_on_failure(self, tmp_path: Path):
+ with patch("dpg.cli.test_dpg", return_value=(None, None)):
+ main(["--dataset", "iris", "--dir", str(tmp_path)])
+ assert list(tmp_path.iterdir()) == []
diff --git a/tests/test_context_order_downstream.py b/tests/test_context_order_downstream.py
new file mode 100644
index 0000000..8b03d06
--- /dev/null
+++ b/tests/test_context_order_downstream.py
@@ -0,0 +1,167 @@
+"""E7 -- downstream compatibility of context_order > 1 / "auto".
+
+Exercises the consumers named in the DPG 0.3.0 execution plan (plot_dpg,
+class-boundary extraction, communities, DPGExplainer global/local
+explanations, faithfulness evaluation) at context_order 2 and "auto", to
+confirm DPG-k does not silently break them. No DPG-IF/DPG-CF consumers exist
+in this codebase, so there is nothing to test for those.
+
+Two pre-existing issues surfaced during this investigation are documented
+here rather than fixed, because they are independent of context_order (both
+reproduce identically at the k=1 default) and are therefore out of E7's
+scope ("fix compatibility issues only within the requested scope"):
+
+- `evaluate_faithfulness()` raises inside sklearn's own GradientBoosting
+ internals because `SklearnEnsembleNormalizer.normalize()` flattens
+ `estimators_` from a 2D array to a list for DPG's own tree iteration, and
+ `evaluate_faithfulness` calls `.predict()` on that same normalized copy.
+- `plot_dpg` does not visually disambiguate two nodes that share a predicate
+ label but differ in context (the DOT `tooltip` carries context, but static
+ PNG/PDF renders do not show tooltips). `get_node_context(node)` remains the
+ documented way to recover per-node context.
+"""
+
+import os
+import shutil
+
+os.environ.setdefault("MPLBACKEND", "Agg")
+
+import networkx as nx
+import numpy as np
+import pytest
+from sklearn.datasets import load_iris
+from sklearn.ensemble import GradientBoostingClassifier, RandomForestClassifier
+
+from dpg import DPGExplainer
+
+
+def _require_graphviz_dot():
+ if shutil.which("dot") is None:
+ pytest.skip("Graphviz 'dot' executable is unavailable")
+
+
+def _iris():
+ iris = load_iris()
+ target_names = np.unique(iris.target).astype(str).tolist()
+ return iris, target_names
+
+
+def _config(context_order):
+ return {
+ "dpg": {
+ "default": {"perc_var": 1e-9, "decimal_threshold": 6, "n_jobs": 1},
+ "graph_construction": {"mode": "execution_trace", "context_order": context_order},
+ }
+ }
+
+
+@pytest.mark.parametrize("context_order", [1, 2, "auto"])
+def test_to_networkx_two_tuple_shape_preserved(context_order):
+ """Public API contract: to_networkx always returns (graph, nodes)."""
+ iris, target_names = _iris()
+ model = RandomForestClassifier(n_estimators=10, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names, dpg_config=_config(context_order))
+ explainer.fit(iris.data)
+
+ result = explainer.builder.to_networkx(explainer._dot)
+ assert isinstance(result, tuple)
+ assert len(result) == 2
+ graph, nodes = result
+ assert isinstance(graph, nx.DiGraph)
+ assert isinstance(nodes, list)
+
+
+@pytest.mark.parametrize("context_order", [2, "auto"])
+def test_node_attributes_preserved_at_higher_context_order(context_order):
+ iris, target_names = _iris()
+ model = RandomForestClassifier(n_estimators=10, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names, dpg_config=_config(context_order))
+ explanation = explainer.explain_global(iris.data)
+
+ assert all(
+ {"predicate", "context", "context_order"} <= data.keys()
+ for _, data in explanation.graph.nodes(data=True)
+ )
+ resolved_k = explainer.builder.get_context_order()
+ assert resolved_k >= 2
+
+
+@pytest.mark.parametrize("context_order", [1, 2, "auto"])
+def test_class_boundaries_and_communities_at_context_order(context_order):
+ iris, target_names = _iris()
+ model = RandomForestClassifier(n_estimators=10, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names, dpg_config=_config(context_order))
+ explanation = explainer.explain_global(iris.data, communities=True)
+
+ assert sorted(explanation.class_boundaries["Class Bounds"].keys()) == ["Class 0", "Class 1", "Class 2"]
+ assert explanation.communities is not None
+ assert set(explanation.communities.keys()) == {"Clusters", "Probability", "Confidence Interval"}
+
+
+def test_local_explanation_class_votes_are_stable_across_context_order():
+ """context_order changes predicate *identity*, not routing: the same
+ sample must still land on the same per-tree leaf and the same class
+ votes at k=1, k=2, and k='auto'."""
+ iris, target_names = _iris()
+ model = RandomForestClassifier(n_estimators=10, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ sample = iris.data[0]
+
+ votes_by_k = {}
+ for context_order in (1, 2, "auto"):
+ explainer = DPGExplainer(
+ model, iris.feature_names, target_names=target_names, dpg_config=_config(context_order)
+ )
+ explainer.fit(iris.data)
+ local = explainer.explain_local(sample=sample, sample_id=0)
+ votes_by_k[context_order] = (local.majority_vote, dict(local.class_votes))
+
+ assert votes_by_k[1] == votes_by_k[2] == votes_by_k["auto"]
+
+
+@pytest.mark.parametrize("context_order", [1, 2, "auto"])
+def test_faithfulness_evaluation_at_context_order_random_forest(context_order):
+ iris, target_names = _iris()
+ model = RandomForestClassifier(n_estimators=10, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names, dpg_config=_config(context_order))
+ explainer.fit(iris.data)
+
+ details = explainer.evaluate_faithfulness(iris.data[:20], y_true=iris.target[:20], return_details=True)
+
+ assert 0.0 <= details["faithfulness_score"] <= 1.0
+ assert 0.0 <= details["output_fidelity"] <= 1.0
+ assert "mean_trace_coverage_score" in details
+ assert "mean_recombination_rate" in details
+
+
+@pytest.mark.parametrize("context_order", [2, "auto"])
+def test_plot_dpg_and_communities_render_at_context_order(tmp_path, context_order):
+ _require_graphviz_dot()
+ iris, target_names = _iris()
+ model = RandomForestClassifier(n_estimators=5, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names, dpg_config=_config(context_order))
+ explanation = explainer.explain_global(iris.data, communities=True)
+
+ explainer.plot(f"dpg_k_{context_order}", explanation, save_dir=str(tmp_path), show=False)
+ assert (tmp_path / f"dpg_k_{context_order}.png").exists()
+
+ explainer.plot_communities(f"dpg_k_{context_order}_communities", explanation, save_dir=str(tmp_path), show=False)
+ assert (tmp_path / f"dpg_k_{context_order}_communities_communities.png").exists()
+
+
+def test_gradient_boosting_faithfulness_runs_at_k1():
+ """Regression: SklearnEnsembleNormalizer flattens
+ GradientBoostingClassifier.estimators_ from (n_stages, n_classes) to a
+ flat list for DPG's own traversal, and ``evaluate_faithfulness()``
+ used to call ``.predict()`` on that same normalized copy, breaking
+ sklearn's internal ``estimators_[0, 0]`` indexing. The explainer
+ now keeps the original model around for ``predict()``, so this code
+ path must run end-to-end at the k=1 default."""
+ iris, target_names = _iris()
+ model = GradientBoostingClassifier(n_estimators=10, random_state=0).fit(iris.data, iris.target)
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names)
+ explainer.fit(iris.data)
+ details = explainer.evaluate_faithfulness(
+ iris.data[:20], y_true=iris.target[:20], return_details=True
+ )
+ assert 0.0 <= details["faithfulness_score"] <= 1.0
+ assert 0.0 <= details["output_fidelity"] <= 1.0
diff --git a/tests/test_context_order_review.py b/tests/test_context_order_review.py
new file mode 100644
index 0000000..51759e8
--- /dev/null
+++ b/tests/test_context_order_review.py
@@ -0,0 +1,176 @@
+"""Independent edge-case review tests for the DPG-k context resolver."""
+
+import pytest
+
+import dpg.context_order as context_order_module
+from dpg.context_order import path_violations, resolve_context_order
+
+
+def test_empty_and_duplicate_traces_resolve_at_k_one():
+ assert resolve_context_order([]) == (1, {1: 0})
+ assert resolve_context_order(
+ [("A", "Class 0"), ("A", "Class 0")]
+ ) == (1, {1: 0})
+
+
+@pytest.mark.parametrize("invalid_max_k", [0, -1, True, 1.5])
+def test_max_k_requires_a_positive_integer(invalid_max_k):
+ with pytest.raises(ValueError, match="max_k must be a positive integer"):
+ resolve_context_order([("A", "Class 0")], max_k=invalid_max_k)
+
+
+def test_insufficient_max_k_does_not_return_unresolved_order():
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+
+ with pytest.raises(ValueError, match="eliminates all global trace violations"):
+ resolve_context_order(traces, max_k=2)
+
+
+def test_sufficient_max_k_returns_zero_violation_order_and_history():
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+
+ resolved, history = resolve_context_order(traces, max_k=3)
+
+ assert resolved == 3
+ assert history[resolved] == 0
+
+
+# ---------------------------------------------------------------------------
+# path_violations -- the diagnostic at the heart of ``auto`` resolution
+# ---------------------------------------------------------------------------
+
+
+class TestPathViolations:
+ """Edge-case coverage for ``path_violations``.
+
+ The motivating example from the docstring is also exercised end-to-end
+ via ``resolve_context_order`` -- here we test the diagnostic primitive
+ in isolation so a regression can point at the right call site.
+ """
+
+ def test_empty_traces_return_zero(self):
+ assert path_violations([], 1) == 0
+ # Falsy traces are dropped before the trie is built.
+ assert path_violations([(), ("a",)], 1) == 0
+
+ def test_single_trace_has_no_violations_at_any_k(self):
+ # A single trace has no recombination to detect.
+ assert path_violations([("A", "B", "C", "Class 0")], 1) == 0
+ assert path_violations([("A", "B", "C", "Class 0")], 3) == 0
+
+ def test_pairwise_consistent_traces_become_consistent_at_higher_k(self):
+ """The motivating phantom-path example from the docstring.
+
+ ``(B)`` and ``(C)`` appear in both traces with *different* futures
+ (``D`` vs ``E``), so even at k=1 there are violations: the pooled
+ graph lets a sample reach ``D`` from a state that was observed to
+ lead only to ``E``. k=3 disambiguates by prefix and the trace
+ language is recovered exactly.
+ """
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+ # k=1: (B) and (C) each merge across traces with different futures.
+ assert path_violations(traces, 1) > 0
+ # k=2: only the (B,C) context still merges with diverging futures.
+ assert path_violations(traces, 2) == 1
+ # k=3 keys every branch by its full prefix and the language is exact.
+ assert path_violations(traces, 3) == 0
+
+ def test_infinite_k_uses_full_history_so_no_violations(self):
+ """``k=math.inf`` is the documented identity bound -- it should
+ recover every observed prefix as a distinct context and so never
+ merge states with different futures."""
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+ assert path_violations(traces, float("inf")) == 0
+
+ def test_sink_nodes_are_never_contextualised(self):
+ """Sinks collapse to a single shared node, which cannot create a
+ new path (it has no outgoing edges)."""
+ traces = [
+ ("A", "B", "Class 0"),
+ ("X", "B", "Class 0"),
+ ]
+ # Both traces terminate in the same Class 0 sink, so no recombination.
+ assert path_violations(traces, 1) == 0
+
+ def test_shared_prefix_branch_is_seen_by_the_trie(self):
+ # Both traces share the (A,B,C) prefix, so the trie at ctx(C) has
+ # *both* ctx(D) and ctx(E) as children. The contextual signature
+ # of (C) therefore captures both children as a single future
+ # language -- there is no separate contextual node with two
+ # futures, hence path_violations sees zero violations.
+ assert path_violations(
+ [("A", "B", "C", "D", "Class 0"), ("A", "B", "C", "E", "Class 1")], 1
+ ) == 0
+ assert path_violations(
+ [("A", "B", "C", "D", "Class 0"), ("A", "B", "C", "E", "Class 1")], 3
+ ) == 0
+
+ def test_resolved_by_distinguishing_first_symbol(self):
+ # The first symbol (A vs X) splits (B) into two distinct contexts
+ # at k=2; only the (B,C) window still merges and the violation
+ # drops to 1, then to 0 at k=3 once every node carries a unique
+ # context.
+ assert path_violations(
+ [("A", "B", "C", "D", "Class 0"), ("X", "B", "C", "E", "Class 1")], 2
+ ) == 1
+ assert path_violations(
+ [("A", "B", "C", "D", "Class 0"), ("X", "B", "C", "E", "Class 1")], 3
+ ) == 0
+
+
+# ---------------------------------------------------------------------------
+# Module surface
+# ---------------------------------------------------------------------------
+
+
+class TestContextOrderModule:
+ """The module surface is intentionally small but stable."""
+
+ def test_public_exports_are_callable(self):
+ # Module surface is small but stable: both helpers are reachable.
+ assert callable(context_order_module.path_violations)
+ assert callable(context_order_module.resolve_context_order)
+
+ def test_history_is_a_dict_not_a_list(self):
+ resolved, history = resolve_context_order([("A", "Class 0")])
+ assert isinstance(history, dict)
+
+ def test_long_history_traces_walk_k_until_zero_violations(self):
+ """``resolve_context_order`` walks k=1..max_k until violations vanish,
+ so the returned history contains every k actually evaluated."""
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+ resolved, history = resolve_context_order(traces)
+ assert resolved == 3
+ assert history[resolved] == 0
+ # Every intermediate k that we evaluated is recorded.
+ assert sorted(history) == [1, 2, 3]
+
+ def test_explicit_max_k_caps_the_evaluation_window(self):
+ """``max_k`` is the upper bound *and* the upper bound on the
+ recorded history; k is walked only up to the supplied cap."""
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+ resolved, history = resolve_context_order(traces, max_k=5)
+ assert resolved == 3
+ # The function stops as soon as a k yields zero violations, so
+ # the resolved key's value is zero and no larger k is recorded.
+ assert history[resolved] == 0
+ assert max(history.keys()) == resolved
+ assert max(history.keys()) <= 5
diff --git a/tests/test_dpg_core.py b/tests/test_dpg_core.py
index 6784a51..4abf830 100644
--- a/tests/test_dpg_core.py
+++ b/tests/test_dpg_core.py
@@ -693,6 +693,8 @@ def fit_and_rank(mode):
assert not aggregated_ranking.empty
assert not trace_ranking.empty
+ # Monotonicity is forced by ``sort_values(ascending=False)``; these
+ # assertions exist to catch the case where sort produced an empty frame.
assert aggregated_ranking["Local reaching centrality"].is_monotonic_decreasing
assert trace_ranking["Local reaching centrality"].is_monotonic_decreasing
@@ -707,3 +709,599 @@ def fit_and_rank(mode):
for rank, row in trace_ranking.iterrows():
print(f"{rank + 1:>2}. {row['Label']:<35} {row['Local reaching centrality']:.4f}")
print()
+
+ def test_pooled_lrc_dominates_trace_consistent_lrc_per_predicate(self, iris_rf, iris_split):
+ """The two LRC implementations measure related but distinct
+ quantities:
+
+ - ``aggregated_transitions`` mode: pooled-graph NetworkX
+ ``local_reaching_centrality`` (weighted) on the merged
+ transitions graph.
+ - ``execution_trace`` mode: trace-consistent LRC computed as the
+ fraction of distinct labels downstream of the predicate within
+ a single observed sample-tree execution.
+
+ These two metrics need not satisfy a strict inequality for every
+ predicate (NetworkX's weighted LRC rewards edges with high
+ weight, while the trace-consistent score is a unit-interval
+ fraction), but they must disagree on at least some predicates:
+ if they were identical across the board, the
+ ``execution_trace`` implementation would be silently returning
+ the pooled mode's answer and the whole point of having a second
+ mode would collapse.
+
+ A regression that, say, made ``execution_trace`` delegate to
+ ``aggregated_transitions`` for every label would be caught here.
+ """
+ from dpg.explainer import DPGExplainer
+
+ X_train, _, _, _, feature_names, target_names = iris_split
+
+ def fit_explainer(mode):
+ explainer = DPGExplainer(
+ model=iris_rf,
+ feature_names=feature_names,
+ target_names=target_names,
+ dpg_config={
+ "dpg": {
+ "default": {"perc_var": 1e-9, "decimal_threshold": 6, "n_jobs": 1},
+ "graph_construction": {"mode": mode},
+ }
+ },
+ )
+ explainer.fit(X_train)
+ return explainer
+
+ pooled = fit_explainer("aggregated_transitions")
+ traced = fit_explainer("execution_trace")
+
+ pooled_metrics = pooled._get_node_metrics()
+ traced_metrics = traced._get_node_metrics()
+
+ predicate_mask = pooled_metrics["Label"].apply(
+ DecisionPredicateGraph._is_predicate_label
+ )
+ pooled_predicates = pooled_metrics[predicate_mask]
+ traced_predicates = traced_metrics[predicate_mask]
+
+ # Join on label so we compare the same predicate in both modes.
+ merged = pooled_predicates.merge(
+ traced_predicates,
+ on="Label",
+ suffixes=("_pooled", "_traced"),
+ )
+ assert not merged.empty, "Expected at least one predicate label."
+
+ pooled_scores = merged["Local reaching centrality_pooled"]
+ traced_scores = merged["Local reaching centrality_traced"]
+
+ # Both implementations must produce scores in the documented
+ # range; this catches gross regressions (e.g. NaNs or unbounded
+ # values) regardless of whether the two rankings happen to agree.
+ assert (pooled_scores >= 0.0).all() and (pooled_scores <= 1.0).all()
+ assert (traced_scores >= 0.0).all() and (traced_scores <= 1.0).all()
+
+ # The substantive comparison: at least some predicates must
+ # disagree -- identical rankings would mean the second mode is
+ # not contributing any distinguishing signal.
+ differing = (pooled_scores - traced_scores).abs() > 1e-12
+ assert differing.any(), (
+ "Pooled and trace-consistent LRC are identical for every "
+ "predicate; the execution_trace implementation is not "
+ "providing any distinguishing signal."
+ )
+
+ # And the set of "top-5" predicates must differ -- the whole
+ # point of trace-consistent LRC is to surface predicates that
+ # pooled-graph LRC misses because of edge pooling.
+ top_k = min(5, len(merged))
+ pooled_top = set(
+ merged.sort_values("Local reaching centrality_pooled", ascending=False)
+ .head(top_k)["Label"]
+ )
+ traced_top = set(
+ merged.sort_values("Local reaching centrality_traced", ascending=False)
+ .head(top_k)["Label"]
+ )
+ assert pooled_top != traced_top, (
+ "Top-{} predicates are identical across modes; the trace-"
+ "consistent ranking does not surface any predicates the "
+ "pooled ranking missed.".format(top_k)
+ )
+
+
+# ---------------------------------------------------------------------------
+# PR #32 additions: context_order validation, decimal_threshold="auto",
+# sink invariants, DOT/NetworkX round-trip for context-aware graphs.
+# ---------------------------------------------------------------------------
+
+import hashlib
+import warnings
+
+from dpg import DPGExplainer
+
+
+def _config(mode="execution_trace", context_order=1, decimal_threshold=6, perc_var=1e-9, n_jobs=1):
+ """Tiny helper to build a dpg_config for the new graph_construction key."""
+ return {
+ "dpg": {
+ "default": {
+ "perc_var": perc_var,
+ "decimal_threshold": decimal_threshold,
+ "n_jobs": n_jobs,
+ },
+ "graph_construction": {
+ "mode": mode,
+ "context_order": context_order,
+ },
+ }
+ }
+
+
+class TestDecimalThresholdValidation:
+ """``decimal_threshold`` accepts a non-negative int or ``"auto"``."""
+
+ @pytest.mark.parametrize("bad_value", [-1, -10, True, False, 1.5, "six"])
+ def test_invalid_decimal_threshold_is_rejected(self, iris_rf, iris_split, bad_value):
+ _, _, _, _, feature_names, _ = iris_split
+ with pytest.raises(
+ DPGError, match="decimal_threshold must be a non-negative integer or 'auto'"
+ ):
+ DecisionPredicateGraph(
+ iris_rf,
+ feature_names,
+ dpg_config=_config(decimal_threshold=bad_value),
+ )
+
+ def test_zero_decimal_threshold_is_accepted(self, iris_rf, iris_split):
+ _, _, _, _, feature_names, _ = iris_split
+ dpg = DecisionPredicateGraph(
+ iris_rf, feature_names, dpg_config=_config(decimal_threshold=0)
+ )
+ assert dpg.decimal_threshold == 0
+
+
+class TestContextOrderValidation:
+ """``context_order`` accepts a positive int or ``"auto"``."""
+
+ @pytest.mark.parametrize("bad_value", [0, -1, -2, True, False, 1.5])
+ def test_invalid_context_order_is_rejected(self, iris_rf, iris_split, bad_value):
+ _, _, _, _, feature_names, _ = iris_split
+ with pytest.raises(
+ DPGError, match="context_order must be a positive integer or 'auto'"
+ ):
+ DecisionPredicateGraph(
+ iris_rf,
+ feature_names,
+ dpg_config=_config(context_order=bad_value),
+ )
+
+ def test_string_context_order_is_rejected(self, iris_rf, iris_split):
+ _, _, _, _, feature_names, _ = iris_split
+ with pytest.raises(
+ DPGError, match="context_order must be a positive integer or 'auto'"
+ ):
+ DecisionPredicateGraph(
+ iris_rf,
+ feature_names,
+ dpg_config=_config(context_order="two"),
+ )
+
+
+class TestDecimalPlacesHelper:
+ """``_decimal_places`` is the building block of ``decimal_threshold='auto'``."""
+
+ @pytest.mark.parametrize(
+ "value, expected",
+ [
+ (0, 0),
+ (1, 0),
+ (-3, 0),
+ (0.5, 1),
+ (0.125, 3),
+ (1.25, 2),
+ (1.0001, 4),
+ ],
+ )
+ def test_decimal_places_for_finite_values(self, value, expected):
+ assert DecisionPredicateGraph._decimal_places(value) == expected
+
+ @pytest.mark.parametrize("bad_value", [float("nan"), float("inf"), float("-inf")])
+ def test_decimal_places_for_non_finite_values_returns_zero(self, bad_value):
+ assert DecisionPredicateGraph._decimal_places(bad_value) == 0
+
+
+class TestResolveDecimalThreshold:
+ """``decimal_threshold='auto'`` derives precision from data and audits tree thresholds."""
+
+ def test_non_auto_passes_through(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(decimal_threshold=4)
+ )
+ dpg._resolve_decimal_threshold(iris.data)
+ assert dpg.get_decimal_threshold() == 4
+
+ def test_auto_with_integer_data_returns_one(self):
+ # 0 decimal places in data => precision 0, +1 = 1.
+ rng = np.random.default_rng(0)
+ X = rng.integers(0, 50, size=(40, 2)).astype(float)
+ y = (X[:, 0] > 25).astype(int)
+ model = RandomForestClassifier(n_estimators=3, random_state=0, n_jobs=1).fit(X, y)
+ dpg = DecisionPredicateGraph(
+ model, ["a", "b"], dpg_config=_config(decimal_threshold="auto")
+ )
+ dpg._resolve_decimal_threshold(X)
+ assert dpg.get_decimal_threshold() == 1
+
+ def test_auto_with_fractional_data_uses_max_precision(self):
+ rng = np.random.default_rng(1)
+ X = np.round(rng.random((40, 2)), 3)
+ y = (X[:, 0] > 0.5).astype(int)
+ model = RandomForestClassifier(n_estimators=3, random_state=0, n_jobs=1).fit(X, y)
+ dpg = DecisionPredicateGraph(
+ model, ["a", "b"], dpg_config=_config(decimal_threshold="auto")
+ )
+ # The synthetic random forest will almost certainly learn thresholds
+ # that are off the data-derived 4-decimal grid -- this is the
+ # exact behaviour ``decimal_threshold='auto'`` is supposed to flag.
+ with pytest.warns(RuntimeWarning, match="off the data-derived"):
+ dpg._resolve_decimal_threshold(X)
+ # 3 decimal places in data => precision 3, +1 = 4.
+ assert dpg.get_decimal_threshold() == 4
+
+ def test_get_decimal_threshold_before_fit_raises(self, iris_rf):
+ """Before ``fit()`` the public getter must fail loudly rather than
+ silently return ``None``."""
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(decimal_threshold="auto")
+ )
+ dpg._resolved_decimal_threshold = None
+ with pytest.raises(DPGError, match="decimal_threshold='auto' is resolved when fit"):
+ dpg.get_decimal_threshold()
+
+
+class TestAutoDecimalThresholdWarning:
+ """Audit step in ``_resolve_decimal_threshold`` must warn on off-grid tree thresholds."""
+
+ def test_off_grid_tree_threshold_emits_warning(self):
+ """Tree thresholds off the data-derived grid warn -- but exact
+ routing is preserved (only labels are rounded)."""
+ rng = np.random.default_rng(42)
+ X = rng.integers(0, 100, size=(60, 2)).astype(float)
+ y = (X[:, 0] > 50).astype(int)
+
+ model = RandomForestClassifier(n_estimators=2, random_state=0, n_jobs=1)
+ model.fit(X, y)
+ # Force one tree's threshold off the 1-decimal grid by a clear margin.
+ first_tree = model.estimators_[0].tree_
+ first_tree.threshold[0] = 50.123
+
+ dpg = DecisionPredicateGraph(
+ model, ["a", "b"], dpg_config=_config(decimal_threshold="auto")
+ )
+ with pytest.warns(RuntimeWarning, match="off the data-derived"):
+ dpg._resolve_decimal_threshold(X)
+
+ def test_on_grid_thresholds_emit_no_warning(self):
+ """A tree whose thresholds already fit the 1-decimal grid must not warn."""
+ rng = np.random.default_rng(123)
+ X = rng.integers(0, 50, size=(40, 2)).astype(float)
+ y = (X[:, 0] > 25).astype(int)
+ model = RandomForestClassifier(n_estimators=2, random_state=0, n_jobs=1).fit(X, y)
+ # Snap every threshold to the 1-decimal grid so the audit step finds nothing.
+ for tree in model.estimators_:
+ tree_ = tree.tree_
+ for idx, feature_index in enumerate(tree_.feature):
+ if feature_index < 0:
+ continue
+ tree_.threshold[idx] = round(float(tree_.threshold[idx]), 1)
+ dpg = DecisionPredicateGraph(
+ model, ["a", "b"], dpg_config=_config(decimal_threshold="auto")
+ )
+ with warnings.catch_warnings():
+ warnings.simplefilter("error", RuntimeWarning)
+ dpg._resolve_decimal_threshold(X)
+ assert dpg.get_decimal_threshold() == 1
+
+
+class TestContextNodeHelpers:
+ """The static ``_context_node`` helpers are pure functions."""
+
+ def test_context_node_returns_sink_for_class_label(self):
+ node = DecisionPredicateGraph._context_node(("a", "Class 0"), index=1, k=2)
+ assert node == ("sink", "Class 0")
+
+ def test_context_node_returns_sink_for_pred_label(self):
+ node = DecisionPredicateGraph._context_node(("a", "Pred 1.5"), index=1, k=2)
+ assert node == ("sink", "Pred 1.5")
+
+ def test_context_node_window_at_start_truncates_to_seen_prefix(self):
+ node = DecisionPredicateGraph._context_node(("A", "B"), index=0, k=3)
+ assert node == ("ctx", ("A",))
+
+ def test_context_node_full_window_after_k_steps(self):
+ node = DecisionPredicateGraph._context_node(("A", "B", "C"), index=2, k=3)
+ assert node == ("ctx", ("A", "B", "C"))
+
+ def test_context_node_window_respects_k(self):
+ node = DecisionPredicateGraph._context_node(("A", "B", "C"), index=2, k=2)
+ assert node == ("ctx", ("B", "C"))
+
+ @pytest.mark.parametrize(
+ "node, expected",
+ [
+ (("sink", "Class 0"), ("Class 0", ())),
+ (("ctx", ("A", "B")), ("B", ("A", "B"))),
+ ],
+ )
+ def test_context_node_info_round_trip(self, node, expected):
+ assert DecisionPredicateGraph._context_node_info(node) == expected
+
+
+class TestDiscoverDFGContext:
+ """Context-aware DFG construction must fall back at k=1 and split at k>1."""
+
+ def test_k_one_falls_through_to_execution_trace(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=1)
+ )
+ dpg.fit(iris.data)
+ log = dpg._extract_trace_log(iris.data)
+
+ ctx = dpg.discover_dfg_context(log, 1)
+ legacy = dpg.discover_dfg_execution_trace(log)
+ assert ctx == legacy
+
+ def test_k_two_splits_diverging_traces(self):
+ """A hand-built trace log exposes the contextual split: the shared
+ predicate ``p`` yields a (``ctx``, (``p``,)) node at k=2."""
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ model = RandomForestClassifier(n_estimators=2, random_state=0, n_jobs=1).fit(
+ iris.data, iris.target
+ )
+ dpg = DecisionPredicateGraph(
+ model, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+
+ log = pd.DataFrame(
+ {
+ "case:concept:name": ["c1", "c1", "c2", "c2"],
+ "concept:name": ["p", "Class 0", "p", "Class 1"],
+ }
+ )
+
+ dfg = dpg.discover_dfg_context(log, 2)
+ ctx_nodes = [
+ node for edge in dfg for node in edge
+ if isinstance(node, tuple) and node[0] == "ctx"
+ ]
+ assert ("ctx", ("p",)) in ctx_nodes
+
+
+class TestNodeLookupHelpers:
+ """Public lookup helpers for graph nodes and traces."""
+
+ def test_get_node_context_returns_empty_for_k1(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=1)
+ )
+ dpg.fit(iris.data)
+ assert dpg.get_node_context("any-id") == ()
+
+ def test_get_node_context_returns_empty_for_non_string_node(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ dpg.fit(iris.data)
+
+ class Weird:
+ def __str__(self):
+ return ""
+
+ assert dpg.get_node_context(Weird()) == ()
+
+ def test_get_node_ids_for_trace_k1_matches_string_keys(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=1)
+ )
+ dpg.fit(iris.data)
+
+ labels = ("a", "b", "Class 0")
+ ids = dpg.get_node_ids_for_trace(labels)
+ assert len(ids) == 3
+ # k=1 keys are just the label strings, hashed the same way the dot
+ # generator hashes them.
+ for label, node_id in zip(labels, ids):
+ assert node_id == str(int(hashlib.sha1(label.encode()).hexdigest(), 16))
+
+ def test_get_node_ids_for_trace_k_gt_1_uses_context_keys(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ dpg.fit(iris.data)
+
+ labels = ("a", "b", "Class 0")
+ ids = dpg.get_node_ids_for_trace(labels)
+ # The two non-sink labels live in distinct contexts, so they
+ # must hash to different node ids.
+ assert ids[0] != ids[1]
+
+
+class TestGetPredicateLrc:
+ """``get_predicate_lrc`` aggregates per-node LRC scores back to predicate labels."""
+
+ def test_returns_scores_in_unit_interval(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ graph, _ = dpg.to_networkx(dpg.fit(iris.data))
+ scores = dpg.get_predicate_lrc(graph)
+ assert isinstance(scores, dict)
+ assert all(0.0 <= s <= 1.0 for s in scores.values())
+ assert scores # at least one predicate for the iris fit
+
+ def test_aggregation_sums_node_lrcs(self, iris_rf):
+ """A predicate appearing in two contexts must have score ==
+ sum of its two node LRCs."""
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ graph, _ = dpg.to_networkx(dpg.fit(iris.data))
+
+ # Build the per-predicate sum ourselves and compare.
+ expected = {}
+ for node, data in graph.nodes(data=True):
+ label = data.get("predicate")
+ if label is None or not dpg._is_predicate_label(label):
+ continue
+ score = float(nx.local_reaching_centrality(graph, node, weight=None))
+ expected[label] = expected.get(label, 0.0) + score
+
+ actual = dpg.get_predicate_lrc(graph)
+ assert set(actual) == set(expected)
+ for label in actual:
+ assert actual[label] == pytest.approx(expected[label], rel=1e-9)
+
+
+class TestTraceConsistentLRCBackwardCompat:
+ """``get_trace_consistent_lrc`` is kept for k=1 and emits a deprecation warning at k>1."""
+
+ def test_k1_still_silently_returns_lrc(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=1)
+ )
+ dpg.fit(iris.data)
+ with warnings.catch_warnings():
+ warnings.simplefilter("error", DeprecationWarning)
+ dpg.get_trace_consistent_lrc() # must not warn
+
+ def test_k_gt_1_emits_deprecation_warning(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ dpg.fit(iris.data)
+ with pytest.warns(DeprecationWarning, match="get_predicate_lrc"):
+ dpg.get_trace_consistent_lrc()
+
+
+class TestDOTAndNetworkXRoundTrip:
+ """The new DOT attribute ``dpg_context_order`` survives the NetworkX round trip."""
+
+ def test_dot_attaches_context_order_to_every_node(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ dot = dpg.fit(iris.data)
+ body = "\n".join(dot.body)
+ assert "dpg_context_order=" in body
+
+ def test_to_networkx_parses_context_order_from_dot(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ dot = dpg.fit(iris.data)
+ graph, _ = dpg.to_networkx(dot)
+
+ assert all(
+ data["context_order"] == dpg.get_context_order() == 2
+ for _, data in graph.nodes(data=True)
+ )
+
+ def test_node_records_attach_predicate_and_context(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ dot = dpg.fit(iris.data)
+ graph, _ = dpg.to_networkx(dot)
+ for _, data in graph.nodes(data=True):
+ assert "predicate" in data
+ assert "context" in data
+ assert isinstance(data["context"], tuple)
+
+
+class TestTraceTreeLabels:
+ """Both label extractors must produce the same leaf and the same length."""
+
+ def test_legacy_and_native_have_same_length_and_same_leaf(self, iris_rf):
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=1)
+ )
+ dpg.fit(iris.data)
+ sample = iris.data[0]
+ tree = dpg.model.estimators_[0]
+
+ legacy = dpg._trace_tree_labels_legacy(0, tree, sample)
+ native = dpg._trace_tree_labels(0, tree, sample)
+
+ assert legacy[-1] == native[-1]
+ assert len(legacy) == len(native)
+
+ def test_native_uses_decision_path_for_routing(self, iris_rf):
+ """The native extractor must follow sklearn's ``decision_path``
+ exactly -- both branches must come from sklearn, not from our own
+ rounding comparison."""
+ from sklearn.datasets import load_iris
+
+ iris = load_iris()
+ dpg = DecisionPredicateGraph(
+ iris_rf, iris.feature_names, dpg_config=_config(context_order=1)
+ )
+ dpg.fit(iris.data)
+
+ sample = iris.data[0].reshape(1, -1)
+ tree = dpg.model.estimators_[0]
+ indicator = tree.decision_path(sample)
+ path = indicator.indices[indicator.indptr[0] : indicator.indptr[1]]
+ leaf_id = int(tree.apply(sample)[0])
+ # Number of predicates in the native trace == number of internal
+ # nodes visited (path excludes the leaf itself).
+ native = dpg._trace_tree_labels(0, tree, iris.data[0])
+ predicates = [label for label in native if dpg._is_predicate_label(label)]
+ assert len(predicates) == sum(1 for n in path if int(n) != leaf_id)
diff --git a/tests/test_dpg_k.py b/tests/test_dpg_k.py
new file mode 100644
index 0000000..4dbdf51
--- /dev/null
+++ b/tests/test_dpg_k.py
@@ -0,0 +1,121 @@
+import numpy as np
+import pandas as pd
+import pytest
+from sklearn.datasets import load_iris
+from sklearn.ensemble import RandomForestClassifier
+
+from dpg.context_order import resolve_context_order
+from dpg.core import DPGError, DecisionPredicateGraph
+
+
+def _forest():
+ iris = load_iris()
+ model = RandomForestClassifier(n_estimators=5, random_state=7, n_jobs=1)
+ model.fit(iris.data, iris.target)
+ return iris, model
+
+
+def _config(mode="execution_trace", context_order=1, decimal_threshold=6):
+ return {
+ "dpg": {
+ "default": {
+ "perc_var": 1e-9,
+ "decimal_threshold": decimal_threshold,
+ "n_jobs": 1,
+ },
+ "graph_construction": {
+ "mode": mode,
+ "context_order": context_order,
+ },
+ }
+ }
+
+
+@pytest.mark.parametrize("context_order", [2, "auto"])
+def test_context_order_requires_execution_trace(context_order):
+ iris, model = _forest()
+ with pytest.raises(DPGError, match="requires mode='execution_trace'"):
+ DecisionPredicateGraph(
+ model,
+ iris.feature_names,
+ dpg_config=_config("aggregated_transitions", context_order),
+ )
+
+
+def test_auto_context_has_one_sink_per_class_and_no_local_violations():
+ iris, model = _forest()
+ dpg = DecisionPredicateGraph(
+ model,
+ iris.feature_names,
+ target_names=["0", "1", "2"],
+ dpg_config=_config(context_order="auto"),
+ )
+ graph, nodes = dpg.to_networkx(dpg.fit(iris.data))
+
+ assert dpg.get_context_order() >= 1
+ assert dpg.get_context_order_history()[dpg.get_context_order()] == 0
+ sinks = [label for _, label in nodes if label.startswith("Class ")]
+ assert sorted(sinks) == ["Class 0", "Class 1", "Class 2"]
+ assert all("context" in data and "predicate" in data for _, data in graph.nodes(data=True))
+
+
+def test_auto_context_detects_longer_history_recombination():
+ """Pairwise transitions can be consistent while a longer path is not."""
+ traces = [
+ ("A", "B", "C", "D", "Class 0"),
+ ("X", "B", "C", "E", "Class 1"),
+ ]
+
+ resolved, history = resolve_context_order(traces)
+
+ assert history[1] > 0
+ assert resolved >= 3
+ assert history[resolved] == 0
+
+
+def test_execution_trace_graph_preserves_long_case_order():
+ """A case-id sort must not scramble a long execution trace."""
+ iris, model = _forest()
+ dpg = DecisionPredicateGraph(
+ model,
+ iris.feature_names,
+ dpg_config=_config(context_order=1),
+ )
+ labels = tuple(f"step-{index}" for index in range(20))
+ log = pd.DataFrame(
+ {
+ "case:concept:name": ["case-0"] * len(labels),
+ "concept:name": labels,
+ }
+ )
+
+ assert dpg.discover_dfg(log) == {
+ (source, target): 1 for source, target in zip(labels, labels[1:])
+ }
+
+
+def test_k1_execution_trace_matches_legacy_edge_weights():
+ iris, model = _forest()
+ dpg = DecisionPredicateGraph(
+ model,
+ iris.feature_names,
+ target_names=["0", "1", "2"],
+ dpg_config=_config(context_order=1),
+ )
+ log = dpg._extract_trace_log(iris.data)
+ assert dpg.discover_dfg(log) == dpg.discover_dfg_execution_trace(log)
+
+
+def test_integer_data_auto_precision_is_lossless():
+ rng = np.random.default_rng(3)
+ X = rng.integers(0, 20, size=(200, 3)).astype(float)
+ y = (X[:, 0] > 9).astype(int)
+ model = RandomForestClassifier(n_estimators=3, random_state=3, n_jobs=1).fit(X, y)
+ dpg = DecisionPredicateGraph(
+ model,
+ ["a", "b", "c"],
+ target_names=["0", "1"],
+ dpg_config=_config(context_order=1, decimal_threshold="auto"),
+ )
+ dpg.fit(X)
+ assert dpg.get_decimal_threshold() == 1
diff --git a/tests/test_e6_regression_scope.py b/tests/test_e6_regression_scope.py
new file mode 100644
index 0000000..6a44c7e
--- /dev/null
+++ b/tests/test_e6_regression_scope.py
@@ -0,0 +1,135 @@
+"""E6 — regression scope: sink behavior and downstream guards.
+
+Regression leaves are labeled ``"Pred "`` instead of ``"Class "``.
+``_context_node`` (dpg/core.py) already treats any ``"Pred "``-prefixed label
+as a terminal sink, so DPG-k mechanically builds a graph for regressors at
+``context_order`` 2 and "auto" without raising. But unlike classification,
+where sink count equals the known, finite number of classes, a regression
+sink is only as unique as the 2-decimal-rounded leaf value: two leaves collide
+into one sink purely by coincidence of rounding, not because of any modeled
+notion of "output". There is therefore no "one sink per output" invariant for
+regression in 0.3.0 -- these tests document the current, explicitly
+out-of-scope behavior rather than assert a guarantee that does not exist.
+
+``GraphMetrics.extract_communities`` separately assumed a classifier (it
+partitions the graph into absorbing "Class " states); called on a regression
+DPG it used to crash with an opaque ``numpy.linalg.LinAlgError: Singular
+matrix`` because no node was absorbing. It now raises a clear ``ValueError``
+instead. This is a defensive fix, not a semantic change: no classifier
+behavior is touched, and no regression numeric output changes.
+"""
+
+import numpy as np
+import pytest
+from sklearn.datasets import load_diabetes
+from sklearn.ensemble import RandomForestRegressor
+
+from dpg import DPGExplainer
+from dpg.core import DecisionPredicateGraph
+
+
+def _diabetes():
+ X, y = load_diabetes(return_X_y=True)
+ feature_names = [f"feature_{i}" for i in range(X.shape[1])]
+ return X, y, feature_names
+
+
+def _config(context_order):
+ return {
+ "dpg": {
+ "default": {"perc_var": 1e-9, "decimal_threshold": 6, "n_jobs": 1},
+ "graph_construction": {"mode": "execution_trace", "context_order": context_order},
+ }
+ }
+
+
+@pytest.mark.parametrize("context_order", [1, 2, "auto"])
+def test_regressor_execution_trace_builds_without_crashing(context_order):
+ """DPG-k mechanically supports regressors: no invariant is claimed here."""
+ X, y, feature_names = _diabetes()
+ model = RandomForestRegressor(n_estimators=5, max_depth=3, random_state=0, n_jobs=1).fit(X, y)
+ dpg = DecisionPredicateGraph(model, feature_names, dpg_config=_config(context_order))
+ graph, nodes = dpg.to_networkx(dpg.fit(X))
+
+ assert graph.number_of_nodes() > 0
+ leaf_labels = [label for _, label in nodes if str(label).startswith("Pred ")]
+ assert leaf_labels, "regression leaves must be labeled 'Pred ...', not 'Class ...'"
+ assert not any(str(label).startswith("Class ") for _, label in nodes)
+
+
+def test_regressor_sink_count_is_not_a_fixed_output_count():
+ """Sink count for regression tracks rounding collisions, not a fixed
+ 'number of outputs' the way classification tracks class count.
+
+ The earlier version of this test only asserted
+ ``0 < n_sinks <= n_leaves``, which is trivially true for any
+ non-empty label set and would not catch a regression that silently
+ collapsed every regression sink to a single label. This version
+ additionally asserts that the sinks are exactly the set of leaves
+ rounded to 2 decimals (the implementation's hard-coded leaf-rounding
+ precision, see ``_trace_execution_labels_for_tree`` in dpg/core.py):
+ a regression that changed that rounding or dropped it altogether
+ would change the set of strings, and this test would catch it.
+ """
+ X, y, feature_names = _diabetes()
+ model = RandomForestRegressor(n_estimators=10, random_state=0, n_jobs=1).fit(X, y)
+ dpg = DecisionPredicateGraph(model, feature_names, dpg_config=_config(1))
+ _, nodes = dpg.to_networkx(dpg.fit(X))
+
+ distinct_sinks = {label for _, label in nodes if str(label).startswith("Pred ")}
+ total_leaves = sum(int(tree.tree_.n_leaves) for tree in model.estimators_)
+
+ # Sanity: at least one sink, never more than total leaves.
+ assert 0 < len(distinct_sinks) <= total_leaves
+
+ # The sinks are exactly ``Pred ``.
+ # A regression that changed the leaf-rounding precision or dropped
+ # the rounding altogether would change the set of strings.
+ expected_sinks = {
+ f"Pred {round(float(tree.tree_.value[leaf_id][0][0]), 2)}"
+ for tree in model.estimators_
+ for leaf_id in range(tree.tree_.node_count)
+ if tree.tree_.children_left[leaf_id] == -1 # -1 children_left marks a leaf
+ }
+ assert distinct_sinks == expected_sinks, (
+ "Regression sinks must equal the set of 2-decimal-rounded leaf "
+ "values; a mismatch means the rounding pipeline changed."
+ )
+
+
+def test_regressor_class_boundaries_are_empty_by_design():
+ """extract_class_boundaries only recognizes 'Class ' sinks; for a
+ regression DPG it returns an empty mapping rather than raising. This is
+ existing, documented behavior -- guard against a silent regression."""
+ X, y, feature_names = _diabetes()
+ model = RandomForestRegressor(n_estimators=5, max_depth=3, random_state=0, n_jobs=1).fit(X, y)
+ explainer = DPGExplainer(model, feature_names, target_names=["prediction"])
+ explanation = explainer.explain_global(X)
+
+ assert explanation.class_boundaries == {"Class Bounds": {}}
+
+
+def test_regressor_communities_raises_clear_error_not_linalg_crash():
+ """explain_global(communities=True) on a regressor must fail loudly and
+ clearly, not with an internal numpy.linalg.LinAlgError."""
+ X, y, feature_names = _diabetes()
+ model = RandomForestRegressor(n_estimators=5, max_depth=3, random_state=0, n_jobs=1).fit(X, y)
+ explainer = DPGExplainer(model, feature_names, target_names=["prediction"])
+
+ with pytest.raises(ValueError, match="requires a classifier DPG"):
+ explainer.explain_global(X, communities=True)
+
+
+def test_classifier_communities_unaffected_by_regression_guard():
+ """The new guard must not change classifier behavior at all."""
+ from sklearn.datasets import load_iris
+ from sklearn.ensemble import RandomForestClassifier
+
+ iris = load_iris()
+ model = RandomForestClassifier(n_estimators=5, random_state=0, n_jobs=1).fit(iris.data, iris.target)
+ target_names = np.unique(iris.target).astype(str).tolist()
+ explainer = DPGExplainer(model, iris.feature_names, target_names=target_names)
+ explanation = explainer.explain_global(iris.data, communities=True)
+
+ assert explanation.communities is not None
+ assert "Clusters" in explanation.communities
diff --git a/tests/test_explainer.py b/tests/test_explainer.py
index 2c1075c..91289f7 100644
--- a/tests/test_explainer.py
+++ b/tests/test_explainer.py
@@ -6,6 +6,7 @@
"""
import re
+from collections import Counter
import numpy as np
import pandas as pd
@@ -242,7 +243,10 @@ def test_sample_bc_weights(self, explainer, explanation, iris_model):
def test_plot_sample_using_bc_weights(self, explainer, explanation, iris_model, tmp_path):
_, X, _, target_names = iris_model
X_df = pd.DataFrame(X, columns=explainer.builder.feature_names)
- y = explainer.builder.model.predict(X_df)
+ # ``iris_model`` is fitted on the bare ``iris.data`` numpy array
+ # (no feature names), so predict on the numpy view of the same
+ # shape to avoid the sklearn feature-name mismatch warning.
+ y = explainer.builder.model.predict(np.asarray(X_df))
fig = explainer.plot_sample_using_bc_weights(
X_df=X_df,
y=y,
@@ -360,36 +364,147 @@ def test_sample_confidence_contains_required_keys(self, explainer, iris_model):
assert explanation.sample_confidence is not None
assert expected_keys.issubset(explanation.sample_confidence.keys())
- def test_vote_confidence_and_class_scores_match_class_votes(self, explainer, iris_model):
- _, X, _, _ = iris_model
- explainer.fit(X)
- explanation = explainer.explain_local(sample=X[0])
+ def test_vote_confidence_and_class_scores_match_class_votes(self, explainer):
+ """Independent oracle: ``class_scores`` must equal the normalised
+ ``class_votes`` dict, ``vote_confidence`` must equal its max, and
+ ``score_margin`` must equal the top-vs-second margin.
+
+ The previous version of this test built the expected values from
+ ``explanation.class_votes`` (i.e. the very dict the production
+ code consumes), so it could not detect a regression that broke
+ the relationship between votes and scores. This version uses a
+ hand-built ``DPGTreePathExplanation`` set whose class votes are
+ independently verifiable.
+ """
+ paths = [
+ DPGTreePathExplanation(
+ tree_index=0,
+ tree_prefix="sample0_dt0",
+ labels=["f0 <= 0.5", "Class 0"],
+ node_ids=["1", "2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.5,
+ mean_bc=0.25,
+ path_confidence=0.7,
+ ),
+ DPGTreePathExplanation(
+ tree_index=1,
+ tree_prefix="sample0_dt1",
+ labels=["f0 > 0.5", "Class 0"],
+ node_ids=["3", "2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.4,
+ mean_bc=0.15,
+ path_confidence=0.5,
+ ),
+ DPGTreePathExplanation(
+ tree_index=2,
+ tree_prefix="sample0_dt2",
+ labels=["f1 <= 1.5", "Class 1"],
+ node_ids=["4", "5"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.3,
+ mean_bc=0.1,
+ path_confidence=0.4,
+ ),
+ ]
- total_votes = sum(explanation.class_votes.values())
- expected_scores = {
- label: votes / total_votes
- for label, votes in explanation.class_votes.items()
- }
- assert explanation.sample_confidence["class_scores"] == expected_scores
- assert explanation.sample_confidence["vote_confidence"] == max(expected_scores.values())
+ # class_votes is what the production code will compute from
+ # tree_paths' terminal labels. Asserting on it independently
+ # before passing it in keeps the test honest if the production
+ # vote-counting code itself regresses.
+ expected_votes = Counter(leaf.rsplit(" ", 1)[-1] for path in paths
+ for leaf in path.labels[-1:] if leaf.startswith("Class "))
+ assert dict(expected_votes) == {"0": 2, "1": 1}
+
+ sample_confidence = explainer._compute_sample_confidence(
+ paths,
+ {"0": 2, "1": 1},
+ np.asarray([5.1, 3.5, 1.4, 0.2]),
+ )
+
+ total_votes = 3
+ expected_scores = {"0": 2 / total_votes, "1": 1 / total_votes}
+ assert sample_confidence["class_scores"] == pytest.approx(expected_scores)
+ assert sample_confidence["vote_confidence"] == pytest.approx(2 / total_votes)
sorted_scores = sorted(expected_scores.values(), reverse=True)
- expected_margin = sorted_scores[0] - sorted_scores[1] if len(sorted_scores) > 1 else sorted_scores[0]
- assert explanation.sample_confidence["score_margin"] == expected_margin
+ expected_margin = sorted_scores[0] - sorted_scores[1]
+ assert sample_confidence["score_margin"] == pytest.approx(expected_margin)
- def test_class_support_equals_summed_path_confidence_by_class(self, explainer, iris_model):
- _, X, _, _ = iris_model
- explainer.fit(X)
- explanation = explainer.explain_local(sample=X[0])
+ def test_class_support_equals_summed_path_confidence_by_class(self, explainer):
+ """Independent oracle for ``class_support`` aggregation.
- expected_support = {}
- for path in explanation.tree_paths:
- leaf = path.labels[-1]
- if leaf.startswith("Class "):
- class_name = leaf[len("Class ") :]
- expected_support[class_name] = expected_support.get(class_name, 0.0) + path.path_confidence
+ Hand-built paths carry known path_confidence values; their sum,
+ grouped by leaf label, must equal the dict the explainer emits.
+ """
+ paths = [
+ DPGTreePathExplanation(
+ tree_index=0,
+ tree_prefix="sample0_dt0",
+ labels=["f0 <= 0.5", "Class 0"],
+ node_ids=["1", "2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.5,
+ mean_bc=0.25,
+ path_confidence=0.7,
+ ),
+ DPGTreePathExplanation(
+ tree_index=1,
+ tree_prefix="sample0_dt1",
+ labels=["f0 > 0.5", "Class 0"],
+ node_ids=["3", "2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.4,
+ mean_bc=0.15,
+ path_confidence=0.5,
+ ),
+ DPGTreePathExplanation(
+ tree_index=2,
+ tree_prefix="sample0_dt2",
+ labels=["f1 <= 1.5", "Class 1"],
+ node_ids=["4", "5"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.3,
+ mean_bc=0.1,
+ path_confidence=0.4,
+ ),
+ ]
+
+ sample_confidence = explainer._compute_sample_confidence(
+ paths,
+ {"0": 2, "1": 1},
+ np.asarray([5.1, 3.5, 1.4, 0.2]),
+ )
- assert explanation.sample_confidence["class_support"] == expected_support
+ # Aggregate from the (known) path_confidence values, not from
+ # the explanation object under test.
+ expected_support = {"0": 0.7 + 0.5, "1": 0.4}
+ assert sample_confidence["class_support"] == pytest.approx(expected_support)
def test_evidence_scores_sum_to_one_when_support_exists(self, explainer, iris_model):
_, X, _, _ = iris_model
@@ -665,13 +780,59 @@ def test_node_ids_and_edge_exists_handle_pruned_graph(self, iris_model):
assert len(path.edge_exists) == max(0, len(path.labels) - 1)
assert path.path_confidence is not None
- def test_local_path_dataframe_returns_one_row_per_path_label(self, explainer, iris_model):
- _, X, _, _ = iris_model
- explainer.fit(X)
- explanation = explainer.explain_local(sample=X[0], sample_id=3)
- df = explainer.local_path_dataframe(explanation)
+ def test_local_path_dataframe_returns_one_row_per_path_label(self, explainer):
+ """Independent oracle: with hand-built paths of known length, the
+ DataFrame's row count must equal ``sum(len(path.labels))``.
+
+ The previous version of this test built the expected count from
+ the same ``explanation.tree_paths`` it was passing in, so the
+ assertion was true by construction.
+ """
+ paths = [
+ DPGTreePathExplanation(
+ tree_index=0,
+ tree_prefix="sample0_dt0",
+ labels=["f0 <= 0.5", "Class 0"],
+ node_ids=["1", "2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.5,
+ mean_bc=0.25,
+ path_confidence=0.7,
+ ),
+ DPGTreePathExplanation(
+ tree_index=1,
+ tree_prefix="sample0_dt1",
+ labels=["f0 > 0.5", "f1 <= 1.0", "Class 0"],
+ node_ids=["3", "4", "5"],
+ predicate_truths=[True, True],
+ edge_exists=[True, True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.4,
+ mean_bc=0.15,
+ path_confidence=0.5,
+ ),
+ ]
+ hand_built = DPGLocalExplanation(
+ sample_id=42,
+ sample=[0.1, 0.2],
+ tree_paths=paths,
+ graph_validated=True,
+ all_trees_valid=True,
+ majority_vote="0",
+ class_votes={"0": 2},
+ path_mode="execution_trace",
+ sample_confidence={},
+ )
+ df = explainer.local_path_dataframe(hand_built)
- assert len(df) == sum(len(path.labels) for path in explanation.tree_paths)
+ expected_rows = sum(len(path.labels) for path in paths)
+ assert len(df) == expected_rows
def test_local_path_dataframe_has_required_columns(self, explainer, iris_model):
_, X, _, _ = iris_model
@@ -697,33 +858,136 @@ def test_local_path_dataframe_has_required_columns(self, explainer, iris_model):
]
assert list(df.columns) == expected_columns
- def test_local_path_dataframe_rows_are_sorted(self, explainer, iris_model):
- _, X, _, _ = iris_model
- explainer.fit(X)
- explanation = explainer.explain_local(sample=X[0])
- df = explainer.local_path_dataframe(explanation)
+ def test_local_path_dataframe_rows_are_sorted(self, explainer):
+ """Independent oracle: tree_paths given out of order must come
+ out sorted by ``(tree_index, step_index)`` in the DataFrame.
+ The previous version of this test sorted the same ``tree_paths``
+ it then passed to the function, so it could not catch a
+ regression that dropped or scrambled the sort.
+ """
+ paths = [
+ DPGTreePathExplanation(
+ tree_index=2,
+ tree_prefix="sample0_dt2",
+ labels=["f1 <= 1.5", "Class 1"],
+ node_ids=["4", "5"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.3,
+ mean_bc=0.1,
+ path_confidence=0.4,
+ ),
+ DPGTreePathExplanation(
+ tree_index=0,
+ tree_prefix="sample0_dt0",
+ labels=["f0 <= 0.5", "Class 0"],
+ node_ids=["1", "2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.5,
+ mean_bc=0.25,
+ path_confidence=0.7,
+ ),
+ DPGTreePathExplanation(
+ tree_index=1,
+ tree_prefix="sample0_dt1",
+ labels=["f0 > 0.5", "f1 <= 1.0", "Class 0"],
+ node_ids=["3", "4", "5"],
+ predicate_truths=[True, True],
+ edge_exists=[True, True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.4,
+ mean_bc=0.15,
+ path_confidence=0.5,
+ ),
+ ]
+ hand_built = DPGLocalExplanation(
+ sample_id=42,
+ sample=[0.1, 0.2],
+ tree_paths=paths,
+ graph_validated=True,
+ all_trees_valid=True,
+ majority_vote="0",
+ class_votes={"0": 2, "1": 1},
+ path_mode="execution_trace",
+ sample_confidence={},
+ )
+ df = explainer.local_path_dataframe(hand_built)
+
+ pairs = list(zip(df["tree_index"], df["step_index"]))
+ # The function should sort by (tree_index, step_index) ascending.
+ assert pairs == sorted(pairs)
+ # And cover every (tree_index, step_index) pair from the input.
+ # ``paths`` is intentionally given out of order (2, 0, 1), so we
+ # look up by tree_index rather than by list position.
+ labels_by_tree = {path.tree_index: path.labels for path in paths}
expected_pairs = [
- (path.tree_index, step_index)
- for path in sorted(explanation.tree_paths, key=lambda path: path.tree_index)
- for step_index in range(len(path.labels))
+ (tree_index, step_index)
+ for tree_index in sorted(labels_by_tree)
+ for step_index in range(len(labels_by_tree[tree_index]))
]
- assert list(zip(df["tree_index"], df["step_index"])) == expected_pairs
-
- def test_local_path_dataframe_edge_alignment(self, explainer, iris_model):
- _, X, _, _ = iris_model
- explainer.fit(X)
- explanation = explainer.explain_local(sample=X[0])
- df = explainer.local_path_dataframe(explanation)
-
- for path in explanation.tree_paths:
- path_df = df[df["tree_index"] == path.tree_index].sort_values("step_index")
- assert path_df.iloc[0]["edge_exists_from_prev"]
- for step_index in range(1, len(path.labels)):
- assert (
- path_df.iloc[step_index]["edge_exists_from_prev"]
- == path.edge_exists[step_index - 1]
- )
+ assert pairs == expected_pairs
+
+ def test_local_path_dataframe_edge_alignment(self, explainer):
+ """Independent oracle for ``edge_exists_from_prev``.
+
+ The previous version of this test was tautological for two
+ reasons: the first-row ``edge_exists_from_prev`` is hard-coded
+ ``True`` by the implementation, and subsequent rows were
+ compared to ``path.edge_exists`` from the same explanation
+ object. This version uses a hand-built set of paths with
+ deliberately mixed edge-presence to verify that the column
+ propagates the ``edge_exists`` array faithfully.
+ """
+ paths = [
+ DPGTreePathExplanation(
+ tree_index=0,
+ tree_prefix="sample0_dt0",
+ labels=["f0 <= 0.5", "f1 > 0.5", "Class 0"],
+ node_ids=["1", "2", "3"],
+ predicate_truths=[True, True],
+ # First edge present, second missing (e.g. pruned by
+ # perc_var); this is what the implementation must echo.
+ edge_exists=[True, False],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=False,
+ mean_lrc=0.5,
+ mean_bc=0.25,
+ path_confidence=0.7,
+ ),
+ ]
+ hand_built = DPGLocalExplanation(
+ sample_id=42,
+ sample=[0.1, 0.2],
+ tree_paths=paths,
+ graph_validated=True,
+ all_trees_valid=False,
+ majority_vote="0",
+ class_votes={"0": 1},
+ path_mode="execution_trace",
+ sample_confidence={},
+ )
+ df = explainer.local_path_dataframe(hand_built)
+
+ # The first row of each path is the root, with no predecessor,
+ # so ``edge_exists_from_prev`` is hard-coded True by the
+ # implementation. Use ``bool(...)`` to avoid ``is`` comparison
+ # pitfalls with numpy scalar ``np.True_``.
+ path_df = df[df["tree_index"] == 0].sort_values("step_index").reset_index(drop=True)
+ assert bool(path_df.iloc[0]["edge_exists_from_prev"]) is True
+ # Subsequent rows must echo ``path.edge_exists`` step by step.
+ assert bool(path_df.iloc[1]["edge_exists_from_prev"]) == bool(paths[0].edge_exists[0])
+ assert bool(path_df.iloc[2]["edge_exists_from_prev"]) == bool(paths[0].edge_exists[1])
def test_local_path_dataframe_empty_explanation(self):
exp = DPGExplainer(
@@ -766,21 +1030,53 @@ def test_local_path_dataframe_empty_explanation(self):
"path_confidence",
]
- def test_local_path_dataframe_values_match_explanation(self, explainer, iris_model):
- _, X, _, _ = iris_model
- explainer.fit(X)
- explanation = explainer.explain_local(sample=X[0], sample_id=11)
- df = explainer.local_path_dataframe(explanation)
-
- first_path = min(explanation.tree_paths, key=lambda path: path.tree_index)
- first_row = df[df["tree_index"] == first_path.tree_index].sort_values("step_index").iloc[0]
+ def test_local_path_dataframe_values_match_explanation(self, explainer):
+ """Independent oracle: a path with a known set of fields must
+ produce rows with those same field values, regardless of where
+ it appears in the input list.
+
+ The previous version of this test pulled its expected values
+ out of the same ``DPGTreePathExplanation`` it was checking, so
+ the assertions were true by construction.
+ """
+ path = DPGTreePathExplanation(
+ tree_index=0,
+ tree_prefix="sample0_dt0",
+ labels=["f0 <= 0.5", "Class 0"],
+ node_ids=["n1", "n2"],
+ predicate_truths=[True],
+ edge_exists=[True],
+ starts_from_root=True,
+ ends_in_leaf=True,
+ graph_path_valid=True,
+ mean_lrc=0.5,
+ mean_bc=0.25,
+ path_confidence=0.7,
+ )
+ hand_built = DPGLocalExplanation(
+ sample_id=42,
+ sample=[0.1, 0.2],
+ tree_paths=[path],
+ graph_validated=True,
+ all_trees_valid=True,
+ majority_vote="0",
+ class_votes={"0": 1},
+ path_mode="execution_trace",
+ sample_confidence={},
+ )
+ df = explainer.local_path_dataframe(hand_built)
- assert first_row["sample_id"] == explanation.sample_id
- assert first_row["label"] == first_path.labels[0]
- assert first_row["node_id"] == first_path.node_ids[0]
- assert first_row["mean_lrc"] == first_path.mean_lrc
- assert first_row["mean_bc"] == first_path.mean_bc
- assert first_row["path_confidence"] == first_path.path_confidence
+ # One row per label, in order.
+ path_df = df[df["tree_index"] == 0].sort_values("step_index").reset_index(drop=True)
+ assert len(path_df) == len(path.labels)
+ for step_index, label in enumerate(path.labels):
+ row = path_df.iloc[step_index]
+ assert row["sample_id"] == hand_built.sample_id
+ assert row["label"] == label
+ assert row["node_id"] == path.node_ids[step_index]
+ assert row["mean_lrc"] == path.mean_lrc
+ assert row["mean_bc"] == path.mean_bc
+ assert row["path_confidence"] == path.path_confidence
class TestFaithfulnessEvaluation:
@@ -925,3 +1221,74 @@ def maybe_fail(sample, sample_id=0, X=None, validate_graph=True):
assert details["n_successful"] == 2
assert len(details["per_sample"]) == 3
assert details["per_sample"]["error"].notna().sum() == 1
+
+
+# ---------------------------------------------------------------------------
+# PR #32 additions: explainer stays consistent across context orders, and
+# uses ``builder.get_node_ids_for_trace`` for local node-id mapping.
+# ---------------------------------------------------------------------------
+
+
+def _config(mode="execution_trace", context_order=1, decimal_threshold=6, perc_var=1e-9, n_jobs=1):
+ return {
+ "dpg": {
+ "default": {
+ "perc_var": perc_var,
+ "decimal_threshold": decimal_threshold,
+ "n_jobs": n_jobs,
+ },
+ "graph_construction": {
+ "mode": mode,
+ "context_order": context_order,
+ },
+ }
+ }
+
+
+class TestContextOrderIntegration:
+ """``DPGExplainer.explain_local`` must produce the same routing at any k."""
+
+ def test_local_node_ids_match_builder_helper(self):
+ """Each tree path's node ids must come from
+ ``builder.get_node_ids_for_trace`` (same length as labels)."""
+ from sklearn.datasets import load_iris
+
+ from dpg import DPGExplainer
+
+ iris = load_iris()
+ model = RandomForestClassifier(n_estimators=5, random_state=0, n_jobs=1).fit(
+ iris.data, iris.target
+ )
+ explainer = DPGExplainer(
+ model, iris.feature_names, dpg_config=_config(context_order=2)
+ )
+ explainer.fit(iris.data)
+
+ local = explainer.explain_local(iris.data[0], sample_id=0)
+ for tree_path in local.tree_paths:
+ node_ids = explainer.builder.get_node_ids_for_trace(tree_path.labels)
+ assert len(node_ids) == len(tree_path.labels)
+
+ def test_explain_local_returns_same_votes_across_context_order(self):
+ """Routing is independent of context order -- so local votes
+ (which depend on routing) must not change between k=1 and k=2."""
+ from sklearn.datasets import load_iris
+
+ from dpg import DPGExplainer
+
+ iris = load_iris()
+ model = RandomForestClassifier(n_estimators=5, random_state=0, n_jobs=1).fit(
+ iris.data, iris.target
+ )
+ sample = iris.data[0]
+ truth = {}
+
+ for k in (1, 2):
+ explainer = DPGExplainer(
+ model, iris.feature_names, dpg_config=_config(context_order=k)
+ )
+ explainer.fit(iris.data)
+ local = explainer.explain_local(sample, sample_id=0)
+ truth[k] = (local.majority_vote, dict(local.class_votes))
+
+ assert truth[1] == truth[2]
diff --git a/tests/test_metrics.py b/tests/test_metrics.py
index fec0690..04f0dcb 100644
--- a/tests/test_metrics.py
+++ b/tests/test_metrics.py
@@ -12,7 +12,7 @@
import pandas as pd
import pytest
from sklearn.datasets import load_iris, load_wine
-from sklearn.ensemble import RandomForestClassifier
+from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
from sklearn.model_selection import train_test_split
from dpg.core import DecisionPredicateGraph
@@ -375,3 +375,72 @@ def test_class_boundaries_per_class(self, wine_dpg):
assert len(bounds["Class 0"]) == 8
assert len(bounds["Class 1"]) == 11
assert len(bounds["Class 2"]) == 9
+
+
+# ---------------------------------------------------------------------------
+# PR #32 addition: extract_communities defensive guard against regression DPGs
+# ---------------------------------------------------------------------------
+
+
+class TestExtractCommunitiesClassifierGuard:
+ """``GraphMetrics.extract_communities`` must still work on classifiers
+ and raise a clear error on regression DPGs (which have only ``Pred ``
+ sinks, no ``Class `` sink)."""
+
+ def test_classifier_dpg_still_returns_communities(self):
+ from sklearn.datasets import load_iris
+
+ from dpg import DPGExplainer
+
+ iris = load_iris()
+ model = RandomForestClassifier(n_estimators=5, random_state=0, n_jobs=1).fit(
+ iris.data, iris.target
+ )
+ explainer = DPGExplainer(
+ model,
+ iris.feature_names,
+ target_names=["0", "1", "2"],
+ dpg_config=_config_for_test(),
+ )
+ explanation = explainer.explain_global(iris.data, communities=True)
+
+ assert explanation.communities is not None
+ assert "Clusters" in explanation.communities
+ assert "Probability" in explanation.communities
+ assert "Confidence Interval" in explanation.communities
+
+ def test_direct_call_on_regression_dpg_raises_clear_error(self):
+ """``GraphMetrics.extract_communities`` must raise a clear
+ ``ValueError`` instead of an opaque numpy LinAlgError when no
+ ``Class ...`` sink node is present."""
+ from sklearn.datasets import load_diabetes
+
+ from dpg.core import DecisionPredicateGraph
+
+ X, y = load_diabetes(return_X_y=True)
+ feature_names = [f"f_{i}" for i in range(X.shape[1])]
+ model = RandomForestRegressor(
+ n_estimators=3, max_depth=3, random_state=0, n_jobs=1
+ ).fit(X, y)
+
+ dpg = DecisionPredicateGraph(
+ model, feature_names, dpg_config=_config_for_test()
+ )
+ graph, nodes = dpg.to_networkx(dpg.fit(X))
+
+ df_node_metrics = pd.DataFrame(
+ {"Node": [nid for nid, _ in nodes], "Label": [lbl for _, lbl in nodes]}
+ )
+
+ with pytest.raises(ValueError, match="requires a classifier DPG"):
+ GraphMetrics.extract_communities(graph, df_node_metrics, nodes)
+
+
+def _config_for_test():
+ """Minimal dpg_config for CLI-style tests."""
+ return {
+ "dpg": {
+ "default": {"perc_var": 1e-9, "decimal_threshold": 6, "n_jobs": 1},
+ "graph_construction": {"mode": "execution_trace", "context_order": 1},
+ }
+ }
diff --git a/tests/test_sklearn_dpg.py b/tests/test_sklearn_dpg.py
index c6c144d..2bf8aa4 100644
--- a/tests/test_sklearn_dpg.py
+++ b/tests/test_sklearn_dpg.py
@@ -8,6 +8,7 @@
import numpy as np
import pandas as pd
import pytest
+from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
import dpg.sklearn_dpg as sklearn_dpg
@@ -269,3 +270,106 @@ def test_negative_learners_raises(self):
def test_zero_learners_raises(self):
with pytest.raises(ValueError, match="positive"):
sklearn_dpg.test_dpg(datasets="iris", n_learners=0)
+
+
+# ---------------------------------------------------------------------------
+# PR #32 addition: DecisionPredicateGraph must receive dpg_config from
+# sklearn_dpg.test_dpg so CLI perc_var and decimal_threshold take effect.
+# ---------------------------------------------------------------------------
+
+
+class TestDpgConfigPropagation:
+ """``sklearn_dpg.test_dpg`` must pass the user-supplied dpg_config to
+ ``DecisionPredicateGraph``; otherwise the CLI's perc_var and
+ decimal_threshold would silently be ignored."""
+
+ def test_test_dpg_propagates_dpg_config_to_decision_predicate_graph(self, monkeypatch):
+ """Spy on ``DecisionPredicateGraph.__init__`` and assert that the
+ ``dpg_config`` ``test_dpg`` builds internally (matching the CLI's
+ ``--pv`` / ``--t`` flags) is what reaches the constructor.
+
+ Earlier versions of this test only instantiated
+ ``DecisionPredicateGraph`` directly, which exercised the spy but
+ bypassed ``test_dpg`` entirely -- a regression in the latter would
+ not have been caught.
+ """
+ from sklearn.datasets import load_iris
+
+ from dpg.core import DecisionPredicateGraph
+
+ iris = load_iris()
+
+ captured_kwargs = {}
+ original_init = DecisionPredicateGraph.__init__
+
+ def spy_init(self, *args, **kwargs):
+ captured_kwargs.update(kwargs)
+ original_init(self, *args, **kwargs)
+
+ monkeypatch.setattr(DecisionPredicateGraph, "__init__", spy_init)
+
+ # These values intentionally differ from test_dpg's defaults
+ # (perc_var=1e-9, decimal_threshold=6); if test_dpg silently drops
+ # its own dpg_config or hard-codes the defaults, the spy will see
+ # the wrong values.
+ sklearn_dpg.test_dpg(
+ datasets="iris",
+ n_learners=3,
+ seed=0,
+ perc_var=1e-6,
+ decimal_threshold=4,
+ n_jobs=1,
+ )
+
+ assert "dpg_config" in captured_kwargs, (
+ "test_dpg must pass a dpg_config to DecisionPredicateGraph; "
+ "it called the constructor without one."
+ )
+ assert captured_kwargs["dpg_config"]["dpg"]["default"]["perc_var"] == 1e-6
+ assert (
+ captured_kwargs["dpg_config"]["dpg"]["default"]["decimal_threshold"] == 4
+ )
+
+ def test_decision_predicate_graph_constructor_accepts_dpg_config(self, monkeypatch):
+ """Smoke test that ``DecisionPredicateGraph`` itself accepts and
+ stores a user-supplied ``dpg_config``. This is the unit-level
+ contract that ``TestDpgConfigPropagation::test_test_dpg_propagates_dpg_config_to_decision_predicate_graph``
+ relies on.
+ """
+ from dpg.core import DecisionPredicateGraph
+
+ captured_kwargs = {}
+ original_init = DecisionPredicateGraph.__init__
+
+ def spy_init(self, *args, **kwargs):
+ captured_kwargs.update(kwargs)
+ original_init(self, *args, **kwargs)
+
+ monkeypatch.setattr(DecisionPredicateGraph, "__init__", spy_init)
+
+ from sklearn.datasets import load_iris as _load_iris
+
+ iris = _load_iris()
+ model = RandomForestClassifier(n_estimators=3, random_state=0, n_jobs=1).fit(
+ iris.data, iris.target
+ )
+
+ DecisionPredicateGraph(
+ model,
+ iris.feature_names,
+ target_names=["0", "1", "2"],
+ dpg_config={
+ "dpg": {
+ "default": {
+ "perc_var": 1e-6,
+ "decimal_threshold": 4,
+ "n_jobs": 1,
+ }
+ }
+ },
+ )
+
+ assert captured_kwargs["dpg_config"]["dpg"]["default"]["perc_var"] == 1e-6
+ assert (
+ captured_kwargs["dpg_config"]["dpg"]["default"]["decimal_threshold"] == 4
+ )
diff --git a/tests/test_smoke.py b/tests/test_smoke.py
index 34d12ad..d91778f 100644
--- a/tests/test_smoke.py
+++ b/tests/test_smoke.py
@@ -9,45 +9,50 @@
from sklearn.ensemble import RandomForestClassifier
-def test_import_dpg():
- import dpg
-
-
-def test_import_metrics():
- import metrics
-
-
-def test_import_core_classes():
- from dpg.core import DecisionPredicateGraph, DPGError
-
-
-def test_import_explainer():
- from dpg.explainer import (
- DPGExplainer,
- DPGExplanation,
- DPGLocalExplanation,
- DPGTreePathExplanation,
- )
-
-
-def test_import_node_metrics():
- from metrics.nodes import NodeMetrics
-
-
-def test_import_edge_metrics():
- from metrics.edges import EdgeMetrics
-
-
-def test_import_graph_metrics():
- from metrics.graph import GraphMetrics
-
-
-def test_import_sklearn_dpg():
- from dpg.sklearn_dpg import test_dpg, select_dataset
-
-
-def test_import_visualizer():
- from dpg.visualizer import plot_dpg, plot_dpg_communities
+# Public symbols that must remain importable. Keeping them in one place
+# makes it cheap to extend the surface area and easy to spot when a
+# rename slips through.
+PUBLIC_SYMBOLS = {
+ "dpg.core": ["DecisionPredicateGraph", "DPGError"],
+ "dpg.explainer": [
+ "DPGExplainer",
+ "DPGExplanation",
+ "DPGLocalExplanation",
+ "DPGTreePathExplanation",
+ ],
+ "dpg.sklearn_dpg": ["test_dpg", "select_dataset"],
+ "dpg.visualizer": ["plot_dpg", "plot_dpg_communities"],
+ "metrics.nodes": ["NodeMetrics"],
+ "metrics.edges": ["EdgeMetrics"],
+ "metrics.graph": ["GraphMetrics"],
+}
+
+
+def test_public_api_imports():
+ """Every documented public symbol must import without error.
+
+ A bare ``import dpg`` only proves the package is importable, not that
+ its public surface is intact. This single test asserts the full
+ documented surface area so a renamed/removed symbol causes a single
+ targeted failure rather than silently passing.
+ """
+ import importlib
+
+ for module_name, symbols in PUBLIC_SYMBOLS.items():
+ module = importlib.import_module(module_name)
+ for symbol in symbols:
+ assert hasattr(module, symbol), (
+ f"Public symbol {symbol!r} missing from {module_name!r}"
+ )
+ assert callable(getattr(module, symbol)) or not callable(getattr(module, symbol)), (
+ f"Public symbol {symbol!r} from {module_name!r} is not accessible"
+ )
+
+
+def test_top_level_packages_importable():
+ """The top-level ``dpg`` and ``metrics`` packages must import."""
+ import dpg # noqa: F401
+ import metrics # noqa: F401
def test_local_explanation_public_workflow_smoke():