TDL: opt-in pruned_leaves extraction (not part of the published TDL)

CHANGELOG.md

@@ -12,6 +12,15 @@ date) and use that section as the basis for the GitHub release notes.
 ## [Unreleased]
 
 ### Added
+- `TDL(hypothesis_extraction="pruned_leaves", prune_tolerance=0.0)`: the `predicted_leaves` expression derefined on
+  the training examples by reduced-error pruning (`ontolearn.learners.tdl_pruning.prune_rules`; drops literals and
+  whole conjunctions while the training F1 stays within `prune_tolerance`). **Not part of the published,
+  peer-reviewed TDL algorithm**: it is an Ontolearn addition, stated as such in the code and docstrings, opt-in, and
+  the default is unchanged. Evaluated in nested CV on 20 problems (issue #672, after the `∀ r⁻.⊤` fix): with
+  `prune_tolerance=0` the expressions are 1.2x shorter at the geometric mean (38 to 32 characters), their instances
+  are retrieved 1.6x faster (faster on 20/20 problems; Mutagenesis median 3.6 s to 0.3 s with a tuned tolerance
+  and a length penalty), and held-out F1 changes by -0.0006 (Holm p=0.35; non-inferiority was not tested). 18 of
+  the 20 problems are one synthetic KB, so treat the efficiency gain as measured and the "no loss" as unproven.
 - `EvoLearner(selection='lexicase')`: lexicase parent selection (`ontolearn.ea_semantic.lexicase_select`), which
   filters the population on randomly ordered examples and breaks ties toward the shortest tree, so specialists
   that solve different subsets of the examples survive. Every evaluated individual now carries a per-example

ontolearn/learners/tdl_pruning.py

@@ -0,0 +1,77 @@
+"""Reduced-error pruning of the class expression that :class:`ontolearn.learners.tree_learner.TDL` reads off its tree.
+
+.. warning::
+   **This is not part of the published TDL algorithm.** TDL (Tree-based Description Logic learner, the peer-reviewed
+   method) fits a decision tree on binary class-expression features and reads the hypothesis off the tree. The
+   pruning here was added afterwards, is *not* from a peer-reviewed publication, and has only been evaluated in the
+   Ontolearn repository (issue #672). It is opt-in (``TDL(hypothesis_extraction="pruned_leaves")``); the default
+   behaviour of TDL is unchanged. Do not cite it as part of TDL.
+
+A *rule* is a tuple of ``(column, holds)`` literals: the conjunction of the column expressions that must hold
+(``holds=True``) or must not hold (``holds=False``, the complement). The hypothesis is the union of the rules. Everything
+works on plain numpy arrays (no reasoner).
+"""
+from typing import List, Sequence, Tuple
+
+import numpy as np
+
+Rule = Tuple[Tuple[int, bool], ...]
+
+
+def rule_mask(X: np.ndarray, rule: Rule) -> np.ndarray:
+    """Boolean vector: which rows of `X` satisfy the conjunction `rule` (an empty rule covers every row)."""
+    mask = np.ones(X.shape[0], dtype=bool)
+    for column, holds in rule:
+        mask &= (X[:, column] > 0.5) == holds
+    return mask
+
+
+def prune_rules(rules: Sequence[Rule], X: np.ndarray, y: np.ndarray, tolerance: float = 0.0) -> List[Rule]:
+    """Derefine a union of rules on the rows `(X, y)` (reduced-error pruning of a class expression).
+
+    Two kinds of edit make the hypothesis simpler and more general: dropping one literal from a conjunction (a
+    conjunction keeps at least one literal) and dropping a whole conjunction (at least one stays). Each round applies
+    the edit with the best F1 on `(X, y)` (the first of equally good edits, so the result is deterministic). It is
+    accepted if the F1 stays within `tolerance` of the F1 of the *unpruned* rules, so the F1 cannot drift down over
+    many rounds. With ``tolerance=0`` ties are resolved in favour of the simpler hypothesis, which removes
+    redundancy without lowering the F1 on `(X, y)`. Pruning stops when no edit is acceptable.
+
+    If `y` has no positive row there is nothing to measure and the rules are returned unchanged.
+    """
+    rules = [tuple(r) for r in rules]
+    if not rules or int((y == 1).sum()) == 0:
+        return rules
+    positive = y == 1
+    n_positive = int(positive.sum())
+
+    def f1_of(covered):
+        tp = int((covered & positive).sum())
+        return 0.0 if tp == 0 else 2 * tp / (int(covered.sum()) + n_positive)
+
+    masks = [rule_mask(X, r) for r in rules]
+    count = np.sum(masks, axis=0)  # how many rules cover each row: the union without rule i is count - masks[i] > 0
+    floor = f1_of(count > 0) - tolerance - 1e-12
+    while True:
+        best_score, best = -1.0, None
+        for i, rule in enumerate(rules):
+            rest = count - masks[i]
+            if len(rules) > 1:  # drop the whole conjunction
+                score = f1_of(rest > 0)
+                if score > best_score + 1e-12:
+                    best_score, best = score, (i, None, None)
+            if len(rule) > 1:  # drop one literal
+                for j in range(len(rule)):
+                    shorter = rule[:j] + rule[j + 1:]
+                    mask = rule_mask(X, shorter)
+                    score = f1_of((rest + mask) > 0)
+                    if score > best_score + 1e-12:
+                        best_score, best = score, (i, shorter, mask)
+        if best is None or best_score < floor:
+            return rules
+        i, shorter, mask = best
+        if shorter is None:
+            count = count - masks[i]
+            del rules[i], masks[i]
+        else:
+            count = count - masks[i] + mask
+            rules[i], masks[i] = shorter, mask

ontolearn/learners/tree_learner.py

@@ -26,6 +26,7 @@
 import numpy as np
 import pandas as pd
 from ontolearn.verbalizer import verbalize_learner_prediction
+from .tdl_pruning import prune_rules
 from owlapy.class_expression import (
     OWLObjectIntersectionOf,
     OWLClassExpression,
@@ -271,7 +272,7 @@ def contains_data_property(expr: OWLClassExpression, include_boolean:bool = Fals
     return found
 
 
-HYPOTHESIS_EXTRACTION_MODES = ("positive_paths", "predicted_leaves", "factored_tree")
+HYPOTHESIS_EXTRACTION_MODES = ("positive_paths", "predicted_leaves", "factored_tree", "pruned_leaves")
 r"""Ways of turning the trained decision tree into one OWL class expression (``TDL(hypothesis_extraction=...)``).
 
 Every split of the tree tests one *binary* feature, i.e. one OWL class expression ``C`` that an individual either
@@ -320,6 +321,16 @@ def contains_data_property(expr: OWLClassExpression, include_boolean:bool = Fals
     (e.g. ``¬(¬C ⊓ B)`` instead of a union of three conjunctions). It makes the same predictions as
     ``predicted_leaves``; only the notation is shorter.
 
+``"pruned_leaves"``  (NOT part of the published TDL algorithm)
+    The ``predicted_leaves`` expression, derefined on the training examples by reduced-error pruning
+    (:func:`ontolearn.learners.tdl_pruning.prune_rules`): literals and whole conjunctions are dropped as long as the
+    training F1 does not fall by more than ``prune_tolerance`` (default 0: the training F1 is kept and only
+    redundancy is removed). Example: in the tree above, if every training example that satisfies ``¬C`` is positive
+    whatever ``B`` says, ``¬C ⊓ ¬B`` is shortened to ``¬C``. This is an addition made in Ontolearn after TDL was published. It is
+    not from a peer-reviewed work and has only been evaluated in Ontolearn (issue #672), where it gave expressions
+    about 1.2x shorter at the geometric mean and retrieved about 1.6x faster (``prune_tolerance=0``), with no
+    significant change in held-out F1 (-0.0006, 20 problems). Opt-in; the default mode is unchanged.
+
 Held-out F1 observed in 5-fold cross-validation (see ``tests/test_tdl_hypothesis_extraction.py``)::
 
                                     positive_paths  predicted_leaves  factored_tree
@@ -355,12 +366,15 @@ def __init__(self, knowledge_base,
                  plot_feature_importance: bool = False,
                  verbose: int = 10,
                  verbalize: bool = False,
-                 hypothesis_extraction: str = "positive_paths"):
+                 hypothesis_extraction: str = "positive_paths",
+                 prune_tolerance: float = 0.0):
 
         if hypothesis_extraction not in HYPOTHESIS_EXTRACTION_MODES:
             raise ValueError(f"hypothesis_extraction must be one of {HYPOTHESIS_EXTRACTION_MODES}. "
                              f"Currently: {hypothesis_extraction!r}")
         self.hypothesis_extraction = hypothesis_extraction
+        # Only used by hypothesis_extraction="pruned_leaves" (not part of the published TDL, see HYPOTHESIS_EXTRACTION_MODES)
+        self.prune_tolerance = prune_tolerance
         self.use_inverse = use_inverse
         self.use_data_properties_boolean = use_data_properties_boolean
         self.use_data_properties_string = use_data_properties_string
@@ -709,6 +723,8 @@ def construct_owl_expression_from_tree(self, X: pd.DataFrame, y: pd.DataFrame) -
             return self._expressions_from_predicted_leaves(X)
         if self.hypothesis_extraction == "factored_tree":
             return [self._factored_expression_from_tree(X)]
+        if self.hypothesis_extraction == "pruned_leaves":
+            return self._expressions_from_pruned_leaves(X, y)
         return self._expressions_from_positive_paths(X, y)
 
     def _tree_feature_expression(self, X: pd.DataFrame, node_id: int) -> OWLClassExpression:
@@ -760,6 +776,30 @@ def visit(node_id: int, path: List[OWLClassExpression]):
         # No positive leaf at all: the hypothesis is the empty class.
         return conjunctions if conjunctions else [nothing]
 
+    def _expressions_from_pruned_leaves(self, X: pd.DataFrame, y: pd.DataFrame) -> List[OWLClassExpression]:
+        """Mode ``pruned_leaves`` (an Ontolearn addition, not part of the published TDL): the positive leaves of the
+        tree as in ``predicted_leaves``, derefined on the training rows with
+        :func:`ontolearn.learners.tdl_pruning.prune_rules` (tolerance ``self.prune_tolerance``)."""
+        tree_ = self.clf.tree_
+        thing, nothing = self.knowledge_base.generator.thing, self.knowledge_base.generator.nothing
+        rules = []
+
+        def visit(node_id: int, path: tuple):
+            if tree_.children_left[node_id] == -1:
+                if self._is_positive_leaf(node_id):
+                    rules.append(path)
+                return
+            feature = int(tree_.feature[node_id])
+            visit(tree_.children_left[node_id], path + ((feature, False),))
+            visit(tree_.children_right[node_id], path + ((feature, True),))
+
+        visit(0, ())
+        if not rules:
+            return [nothing]
+        rules = prune_rules(rules, X.values, np.asarray(y).ravel().astype(int), self.prune_tolerance)
+        return [concepts_reducer([X.columns[c] if holds else X.columns[c].get_object_complement_of() for c, holds in rule],
+                                 OWLObjectIntersectionOf) if rule else thing for rule in rules]
+
     def _factored_expression_from_tree(self, X: pd.DataFrame) -> OWLClassExpression:
         r"""Mode ``factored_tree``: fold the tree into one simplified, nested class expression.
 

tests/test_tdl_hypothesis_extraction.py

@@ -36,7 +36,9 @@
 from ontolearn.learning_problem import PosNegLPStandard
 
 NS = "http://example.com/tdl#"
-POSITIVE_PATHS, PREDICTED_LEAVES, FACTORED_TREE = HYPOTHESIS_EXTRACTION_MODES
+# The three modes of the published TDL; "pruned_leaves" is an Ontolearn addition with its own tests (test_tdl_pruning.py)
+MODES = HYPOTHESIS_EXTRACTION_MODES[:3]
+POSITIVE_PATHS, PREDICTED_LEAVES, FACTORED_TREE = MODES
 
 
 # ----------------------------------------------------------------------------------------------------------------------
@@ -72,13 +74,13 @@ def cross_validate(kb, pos, neg, kwargs_classifier=None, folds=5):
     pos, neg = np.array(pos), np.array(neg)
     examples = np.concatenate([pos, neg])
     labels = np.array([1] * len(pos) + [0] * len(neg))
-    scores = {m: [] for m in HYPOTHESIS_EXTRACTION_MODES}
-    sizes = {m: [] for m in HYPOTHESIS_EXTRACTION_MODES}
+    scores = {m: [] for m in MODES}
+    sizes = {m: [] for m in MODES}
     for train, test in StratifiedKFold(folds, shuffle=True, random_state=0).split(examples, labels):
         train_pos, train_neg = examples[train][labels[train] == 1], examples[train][labels[train] == 0]
         test_pos, test_neg = set(examples[test][labels[test] == 1]), set(examples[test][labels[test] == 0])
-        model = fit(kb, train_pos, train_neg, HYPOTHESIS_EXTRACTION_MODES[0], kwargs_classifier)
-        for m in HYPOTHESIS_EXTRACTION_MODES:
+        model = fit(kb, train_pos, train_neg, MODES[0], kwargs_classifier)
+        for m in MODES:
             model.hypothesis_extraction = m
             model._finalize_predictions(model.X, model.y)
             h = model.best_hypotheses()
@@ -226,7 +228,7 @@ def tearDownClass(cls):
 
     def scores(self, **kwargs_classifier):
         out = {m: evaluate(self.kb, self.pos, self.neg, (self.test_pos, self.test_neg), m, kwargs_classifier)
-               for m in HYPOTHESIS_EXTRACTION_MODES}
+               for m in MODES}
         report(f"synthetic KB {kwargs_classifier or 'fully grown tree'}", out)
         return out
 
@@ -268,7 +270,7 @@ def test_noise_free_fully_grown_tree_modes_agree(self):
             path = os.path.join(d, "kb.owl")
             pos, neg, test_pos, test_neg = write_kb(path, noise=0.0)
             kb = KnowledgeBase(path=path)
-            out = {m: evaluate(kb, pos, neg, (test_pos, test_neg), m) for m in HYPOTHESIS_EXTRACTION_MODES}
+            out = {m: evaluate(kb, pos, neg, (test_pos, test_neg), m) for m in MODES}
             for m in (PREDICTED_LEAVES, FACTORED_TREE):
                 self.assertAlmostEqual(out[m][0], out[POSITIVE_PATHS][0], places=9)
             self.assertLessEqual(out[FACTORED_TREE][1], out[POSITIVE_PATHS][1])
@@ -380,7 +382,7 @@ def setUpClass(cls):
     def test_full_tree(self):
         out = cross_validate(self.kb, self.pos, self.neg)
         report("Carcinogenesis full tree", out)
-        for m in HYPOTHESIS_EXTRACTION_MODES:
+        for m in MODES:
             self.assertAlmostEqual(out[m][0], 0.620, places=3)
         self.assertLess(out[FACTORED_TREE][1], 0.5 * out[POSITIVE_PATHS][1])  # 560 vs 1632 characters
 

tests/test_tdl_pruning.py

@@ -0,0 +1,109 @@
+"""Tests for ``TDL(hypothesis_extraction="pruned_leaves")``.
+
+This mode is an Ontolearn addition that is NOT part of the published, peer-reviewed TDL algorithm (see
+``ontolearn/learners/tdl_pruning.py``). The tests check what it promises: it never changes the training F1 at
+``prune_tolerance=0``, never makes the expression longer, and leaves the default behaviour of TDL untouched.
+"""
+import json
+import os
+import unittest
+
+import numpy as np
+
+from ontolearn.learners.tdl_pruning import prune_rules, rule_mask
+
+FAMILY_KB = "KGs/Family/family-benchmark_rich_background.owl"
+FAMILY_LPS = "LPs/Family/lps.json"
+
+
+class TestPruneRules(unittest.TestCase):
+    # columns: 0 = A, 1 = B, 2 = C. The positives are exactly the rows with A.
+    X = np.array([[1, 0, 0], [1, 1, 0], [1, 0, 1], [1, 1, 1], [0, 0, 0], [0, 1, 0], [0, 0, 1], [0, 1, 1]])
+    y = np.array([1, 1, 1, 1, 0, 0, 0, 0])
+
+    def test_redundant_literals_are_dropped_without_changing_the_cover(self):
+        pruned = prune_rules([((0, True), (1, False)), ((0, True), (1, True))], self.X, self.y)
+        self.assertEqual(pruned, [((0, True),)])
+
+    def test_a_literal_that_is_needed_is_kept(self):
+        rules = [((0, True), (1, False))]  # covers 2 of the 4 positives and no negative
+        self.assertEqual(prune_rules(rules, self.X, self.y), [((0, True),)])  # dropping B only improves F1
+        rules = [((1, False), (2, False))]  # covers one positive and one negative
+        pruned = prune_rules(rules, self.X, self.y)
+        self.assertEqual(len(pruned), 1)
+        self.assertGreaterEqual(len(pruned[0]), 1)
+
+    def test_pruning_never_lowers_the_f1_at_zero_tolerance(self):
+        rng = np.random.RandomState(0)
+        for _ in range(30):
+            X = rng.randint(0, 2, size=(40, 6))
+            y = rng.randint(0, 2, size=40)
+            rules = [tuple((int(c), bool(rng.randint(2))) for c in rng.choice(6, size=3, replace=False))
+                     for _ in range(4)]
+            pruned = prune_rules(rules, X, y)
+
+            def f1(rs):
+                covered = np.any([rule_mask(X, r) for r in rs], axis=0)
+                tp = int((covered & (y == 1)).sum())
+                return 0.0 if tp == 0 else 2 * tp / (int(covered.sum()) + int((y == 1).sum()))
+            self.assertGreaterEqual(f1(pruned) + 1e-12, f1(rules))
+            self.assertLessEqual(sum(map(len, pruned)), sum(map(len, rules)))
+            self.assertTrue(all(len(r) >= 1 for r in pruned) and len(pruned) >= 1)
+
+    def test_no_positives_returns_the_rules_unchanged(self):
+        rules = [((0, True), (1, False))]
+        self.assertEqual(prune_rules(rules, self.X, np.zeros(8, dtype=int)), rules)
+
+    def test_tolerance_is_measured_against_the_unpruned_rules(self):
+        rules = [((0, True),), ((2, True), (1, True))]
+        strict = prune_rules(rules, self.X, self.y, 0.0)
+        loose = prune_rules(rules, self.X, self.y, 0.5)
+        self.assertLessEqual(sum(map(len, loose)), sum(map(len, strict)))
+
+
+@unittest.skipUnless(os.path.exists(FAMILY_KB) and os.path.exists(FAMILY_LPS), "Family data not downloaded")
+class TestPrunedLeavesOnFamily(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        from ontolearn.knowledge_base import KnowledgeBase
+        cls.kb = KnowledgeBase(path=FAMILY_KB)
+        with open(FAMILY_LPS) as f:
+            cls.problems = json.load(f)["problems"]
+
+    def fit(self, problem, mode, **kwargs):
+        from owlapy.owl_individual import OWLNamedIndividual
+        from ontolearn.learners import TDL
+        from ontolearn.learning_problem import PosNegLPStandard
+        ex = self.problems[problem]
+        lp = PosNegLPStandard(pos={OWLNamedIndividual(i) for i in ex["positive_examples"]},
+                              neg={OWLNamedIndividual(i) for i in ex["negative_examples"]})
+        return TDL(knowledge_base=self.kb, kwargs_classifier={"random_state": 1}, verbose=0,
+                   report_classification=False, hypothesis_extraction=mode, **kwargs).fit(lp), lp
+
+    def train_f1(self, model, lp):
+        covered = set(self.kb.individuals(model.best_hypotheses(1)))
+        tp, fp = len(covered & lp.pos), len(covered & lp.neg)
+        fn = len(lp.pos) - tp
+        return 0.0 if tp == 0 else 2 * tp / (2 * tp + fp + fn)
+
+    def test_zero_tolerance_keeps_the_training_f1_and_is_not_longer(self):
+        from owlapy import owl_expression_to_dl
+        for problem in ("Uncle", "Cousin", "Aunt"):
+            base, lp = self.fit(problem, "predicted_leaves")
+            pruned, _ = self.fit(problem, "pruned_leaves")
+            self.assertAlmostEqual(self.train_f1(pruned, lp), self.train_f1(base, lp), places=9)
+            self.assertLessEqual(len(owl_expression_to_dl(pruned.best_hypotheses(1))),
+                                 len(owl_expression_to_dl(base.best_hypotheses(1))))
+
+    def test_default_mode_is_unchanged(self):
+        from ontolearn.learners import TDL
+        self.assertEqual(TDL(knowledge_base=self.kb).hypothesis_extraction, "positive_paths")
+
+    def test_unknown_mode_is_rejected(self):
+        from ontolearn.learners import TDL
+        with self.assertRaises(ValueError):
+            TDL(knowledge_base=self.kb, hypothesis_extraction="nonsense")
+
+
+if __name__ == "__main__":
+    unittest.main()