from owlapy.vocab import OWLFacet, XSDVocabulary
from owlapy.owl_property import OWLDataProperty
from owlapy.owl_literal import Decimal, OWLLiteral
from owlapy.owl_datatype import OWLDatatype
from owlapy import owl_expression_to_dl
from sklearn.inspection import permutation_importance
from .tree_learner import concepts_reducer
from ontolearn.learning_problem import PosNegLPStandard
from owlapy.class_expression import (
OWLClassExpression,
OWLObjectUnionOf,
OWLDataSomeValuesFrom,
)
from ..utils.static_funcs import make_iterable_verbose
from owlapy.class_expression.restriction import (
OWLFacetRestriction,
OWLDatatypeRestriction,
)
from owlapy.owl_individual import OWLNamedIndividual
import numpy as np
import pandas as pd
from sklearn import tree
import itertools
from shap import TreeExplainer
import matplotlib.pyplot as plt
from typing import Dict, Set, Tuple, List, Union
from fitter import Fitter
import logging
import scipy as sc
from scipy.stats._distn_infrastructure import rv_frozen
from ontolearn.learners import TDL
logging.getLogger("fitter").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)
[docs]
class TDL_refinement(TDL):
def __init__(
self,
knowledge_base,
use_inverse: bool = False,
use_data_properties_numeric: bool = True,
use_data_properties_boolean: bool = True,
use_data_properties_string: bool = True,
use_data_properties_date: bool = True,
use_nominals: bool = False,
use_card_restrictions: bool = True,
kwargs_classifier: dict = None,
max_runtime: int = None,
grid_search_over: dict = None,
grid_search_apply: bool = False,
kwargs_grid_search: dict = None,
report_classification: bool = False,
plot_tree: bool = False,
plot_embeddings: bool = False,
plot_feature_importance: bool = False,
verbose: int = 10,
verbalize: bool = False,
feature_refinement: bool = True,
plot_importance_evo: bool = False,
refine_iterations: int = 3,
topk: int = 30,
):
super().__init__(
knowledge_base,
use_inverse,
use_data_properties_boolean,
use_data_properties_string,
use_data_properties_date,
use_data_properties_numeric,
use_nominals,
use_card_restrictions,
kwargs_classifier,
max_runtime,
grid_search_over,
grid_search_apply,
kwargs_grid_search,
report_classification,
plot_tree,
plot_embeddings,
plot_feature_importance,
verbose,
verbalize,
)
self.feature_refinement = feature_refinement
self.plot_importance_evo = plot_importance_evo
self.refine_iterations = refine_iterations
self.topk = topk
self.concept_per_iteration: List = list()
self.top_feature_dicts: List = list()
self.initial_importance_dict: Dict = dict()
self.use_data_properties = any([use_data_properties_numeric, use_data_properties_boolean, use_data_properties_string, use_data_properties_date])
self.data_property_datatype_dict: Dict[OWLDataProperty, OWLDatatype] = dict()
def _pack_data_property_with_range_to_dl_concept(self, property: OWLDataProperty, literal_range: tuple, type) -> str:
"""Repack the ranged data property into an DL Concept String"""
xsdtype: OWLDatatype = self.data_property_datatype_dict.get(property)
if xsdtype == OWLDatatype(XSDVocabulary.INTEGER):
literal_range = (int(literal_range[0]), int(literal_range[1]))
min_restr = OWLFacetRestriction(
OWLFacet.MIN_INCLUSIVE,
OWLLiteral(literal_range[0], xsdtype),
)
if literal_range[1] is None:
max_restr = OWLFacetRestriction(
OWLFacet.MAX_INCLUSIVE,
OWLLiteral(literal_range[0], xsdtype),
)
else:
max_restr = OWLFacetRestriction(
OWLFacet.MAX_INCLUSIVE,
OWLLiteral(literal_range[1], xsdtype),
)
dt_range = OWLDatatypeRestriction(
type_=xsdtype,
facet_restrictions=[min_restr, max_restr],
)
packed_owl_class_expression = OWLDataSomeValuesFrom(property, dt_range)
return packed_owl_class_expression
def _compute_dt_pdf_ranges(self, best_dist: dict) -> List[tuple]:
"""Compute ranges based on the fitted distribution."""
frozen_dist: rv_frozen
dist_name: list
dist_name = list(best_dist.keys())[0]
distribution = getattr(sc.stats, dist_name)
frozen_dist = distribution(**best_dist[dist_name])
mean: np.float64
std: np.float64
mean = frozen_dist.stats(moments="m")
std = frozen_dist.std()
ranges = [(mean - std, mean), (mean, mean + std), (mean - 2 * std, mean - 1 * std), (mean + std, mean + 2 * std), (mean - 3 * std, mean - 2 * std), (mean + 2 * std, mean + 3 * std)]
return ranges
def _find_best_distribution_for_dp(self, dt_values: List[float], plot: bool = False) -> dict:
"""Iteratively fit distributions on the data and choose the best one based on K-S statistic."""
assert len(dt_values) > 0, "Data property values list cannot be empty for distribution fitting."
f = Fitter(
dt_values,
distributions=[
"cauchy",
"chi2",
"expon",
"exponpow",
"gamma",
"lognorm",
"norm",
"powerlaw",
"rayleigh",
],
)
f.fit()
best_dist = f.get_best(method="ks_statistic")
logger.debug(best_dist)
if plot:
# distribution fit plot
f.summary(Nbest=1, method="ks_statistic")
plt.legend(["function", " data (Histogram)"])
plt.xlabel(" values")
plt.ylabel("Density (Data & Probability)")
plt.title("Best Distribution Fit (Ranked by K-S)")
plt.show()
return best_dist
def _extract_refined_ranges_from_data_properties(self, data_property_values: dict, range_values):
"""Extract refined ranges from data property values based on the original range values."""
lb, ub = range_values
prop_values = np.sort(np.array(data_property_values))
# where can maybe be runtime optimized
lb_index = np.where(prop_values[np.abs(prop_values - lb).argmin()] == prop_values)
ub_index = np.where(prop_values[np.abs(prop_values - ub).argmin()] == prop_values)
# if self.verbose>10:
logger.debug("lb: %s ub: %s", lb_index[0][0], ub_index[0][0])
if lb_index[0][0] >= ub_index[0][0]:
if self.verbose > 0:
logger.warning("Lower bound and upper bound indices are the same. Refinement may not be effective.")
return None
dist = self._find_best_distribution_for_dp(prop_values[lb_index[0][0] : ub_index[0][0]], plot=False)
return self._compute_dt_pdf_ranges(dist)
def _extract_ranges_from_data_properties(
self,
features: Dict[str, OWLClassExpression],
individuals_to_feature_mapping,
data_properties_dict: dict[OWLDataProperty, List[float]],
per_individual_data_properties: dict[OWLNamedIndividual, dict[OWLDataProperty, float]],
):
"""Extract refined ranges from data properties and create new DL concepts based on these ranges."""
if len(data_properties_dict) == 0:
return
ranges_dict: dict[OWLDataProperty, List[Tuple[float, float]]] = dict()
generated_dt_class_expressions = dict()
best_dists = dict()
for prop in data_properties_dict:
best_dist = self._find_best_distribution_for_dp(data_properties_dict[prop])
logger.debug("prop%r", prop)
logger.debug(best_dist)
best_dists[prop._iri] = best_dist
ranges_dict.setdefault(prop, self._compute_dt_pdf_ranges(best_dist))
for r in ranges_dict:
for interval in range(len(ranges_dict[r])):
if (np.isnan(ranges_dict[r][interval][0]) or np.isinf(ranges_dict[r][interval][0])) or (np.isnan(ranges_dict[r][interval][1]) or np.isinf(ranges_dict[r][interval][1])):
continue
new_class_expression = self._pack_data_property_with_range_to_dl_concept(r, ranges_dict[r][interval], data_properties_dict[r][0])
str_dl_concept = owl_expression_to_dl(new_class_expression)
# add concept to features
if str_dl_concept not in features:
features[str_dl_concept] = new_class_expression
# assign concept per individual, if satisfied
for ind, prop in itertools.product(per_individual_data_properties.keys(), data_properties_dict.keys()):
if prop in per_individual_data_properties.get(ind, {}):
for v in per_individual_data_properties[ind][prop]:
for interval in ranges_dict[prop]:
if (np.isnan(interval[0]) or np.isinf(interval[0])) or (np.isnan(interval[1]) or np.isinf(interval[1])):
continue
if interval[0] <= v and v <= interval[1]:
new_class_expression = self._pack_data_property_with_range_to_dl_concept(prop, interval, v)
str_dl_concept = owl_expression_to_dl(new_class_expression)
individuals_to_feature_mapping[ind.str].add(str_dl_concept)
if new_class_expression not in generated_dt_class_expressions:
generated_dt_class_expressions[new_class_expression] = set()
generated_dt_class_expressions[new_class_expression].add(ind)
break
return generated_dt_class_expressions
def _extract_data_property_features(
self,
individual: OWLNamedIndividual,
features: Dict[str, OWLClassExpression],
individuals_to_feature_mapping: Dict[str, Set[str]],
data_properties_dict,
per_individual_data_properties,
numeric_ranges: bool = True,
):
"""Extract features based on data properties."""
try:
# Get data properties for this individual
for data_prop in self.knowledge_base.get_data_properties_for_ind(individual):
# Get data property values
data_values = list(self.knowledge_base.get_data_property_values(individual, data_prop))
if data_values:
# For each data value, we already have features from abox(mode="expression")
if numeric_ranges:
for v in data_values:
self.data_property_datatype_dict[data_prop] = v.get_datatype()
# check if literal is a numeric
if v.is_decimal() or v.is_float() or v.is_integer() or v.is_double():
literal = v.get_literal()
if v.is_decimal():
literal = Decimal(v.get_literal())
elif v.is_float():
literal = float(v.get_literal())
elif v.is_integer():
literal = int(v.get_literal())
elif v.is_double():
literal = float(v.get_literal())
# save data_properties per individual
if individual not in per_individual_data_properties:
per_individual_data_properties.setdefault(individual, {})
if data_prop not in per_individual_data_properties[individual]:
per_individual_data_properties[individual].setdefault(data_prop, [])
per_individual_data_properties[individual][data_prop].append(literal)
# save global data property values per property
if data_prop not in data_properties_dict:
data_properties_dict[data_prop] = list()
data_properties_dict[data_prop].append(literal)
except Exception as e:
if self.verbose > 0:
logger.warning(f"Error extracting data property features: {e}")
[docs]
def plot_perm_feature_importances(self, perm_importance_result, feat_name):
"""bar plot the permutation feature importance"""
feat_name = np.array(feat_name)
fig, ax = plt.subplots()
indices = perm_importance_result["importances_mean"].argsort()
perm_importance = perm_importance_result["importances_mean"][indices[-40:]]
perm_err = perm_importance_result["importances_std"][indices[-40:]]
plt.barh(
range(len(indices[-40:])),
perm_importance,
xerr=perm_err,
)
ax.set_yticks(range(len(indices[-40:])))
_ = ax.set_yticklabels(feat_name[indices[-40:]])
logger.debug("show plot")
plt.show()
[docs]
def plot_shap_feature_importances(self, shap_vals, feat_name):
"""Bar plot the mean absolute SHAP feature importance"""
feat_name = np.array(feat_name)
indices = np.argsort(shap_vals)
num_to_plot = min(len(indices), 40)
top_indices = indices[-num_to_plot:]
fig, ax = plt.subplots(figsize=(10, 8))
plt.barh(range(num_to_plot), shap_vals[top_indices], color="skyblue")
ax.set_yticks(range(num_to_plot))
ax.set_yticklabels(feat_name[top_indices])
ax.set_xlabel("Global feature importance")
ax.set_title("Top DL Feature Importance (SHAP)")
plt.tight_layout()
logger.debug("Displaying SHAP Importance Plot")
plt.show()
[docs]
def refine_numerical_features(self, topk_expressions: list[OWLClassExpression], iteration: int):
"""Refine numerical features based on data property values and fitted distributions."""
individuals_to_refined_feature_mapping = dict()
refined_features = dict()
topk_expressions = set(topk_expressions)
for e in topk_expressions:
if type(e) is OWLDataSomeValuesFrom:
filler = e.get_filler()
if type(filler) is OWLDatatypeRestriction:
borders = tuple(float(r.get_facet_value().get_literal()) for r in filler.get_facet_restrictions())
if self.verbose > 10:
logger.debug("Range to be refined : %s", borders)
logger.debug(e)
refined_ranges = self._extract_refined_ranges_from_data_properties(self.data_properties_dict[e.get_property()], borders)
if refined_ranges is None:
continue
for r in refined_ranges:
if (np.isnan(r[0]) or np.isinf(r[0])) or (np.isnan(r[1]) or np.isinf(r[1])):
continue
new_class_expression = self._pack_data_property_with_range_to_dl_concept(e.get_property(), r, e.get_filler().get_facet_restrictions()[0].get_facet_value().get_datatype())
str_dl_concept = owl_expression_to_dl(new_class_expression)
refined_features[str_dl_concept] = new_class_expression
individuals_for_e = self.generated_dt_classexpressions_per_individual.get(e)
if individuals_for_e is None:
prop = e.get_property()
lb, ub = borders
individuals_for_e = {ind for ind, prop_dict in self.per_individual_data_properties.items() if prop in prop_dict and any(lb <= v <= ub for v in prop_dict[prop])}
if self.verbose > 0 and not individuals_for_e:
logger.warning(f"No individuals found for expression {owl_expression_to_dl(e)}, skipping.")
for ind in individuals_for_e:
for v in self.per_individual_data_properties[ind][e.get_property()]:
for interval in refined_ranges:
if (np.isnan(interval[0]) or np.isinf(interval[0])) or (np.isnan(interval[1]) or np.isinf(interval[1])):
continue
if interval[0] <= v <= interval[1]:
new_class_expression = self._pack_data_property_with_range_to_dl_concept(
e.get_property(), interval, e.get_filler().get_facet_restrictions()[0].get_facet_value().get_datatype()
)
string_dl_new_concept = owl_expression_to_dl(new_class_expression)
individuals_to_refined_feature_mapping.setdefault(ind.str, set()).add(string_dl_new_concept)
self.individuals_to_feature_mapping[ind.str].add(string_dl_new_concept)
if new_class_expression not in self.generated_dt_classexpressions_per_individual:
self.generated_dt_classexpressions_per_individual[new_class_expression] = set()
self.generated_dt_classexpressions_per_individual[new_class_expression].add(ind)
return refined_features, individuals_to_refined_feature_mapping
# create_training_data() is inherited from TDL: identical logic aside from
# which extract_expressions_from_owl_individuals() override it dispatches to.
[docs]
def plot_importance_evolution(self, initial_dict, refined_dicts, top_k=10):
# prepare data
def to_str(obj):
return str(obj).strip()
initial_set = {to_str(k) for k in initial_dict.keys()}
feature_birthdays = {feat: 0 for feat in initial_set}
for i, d in enumerate(refined_dicts):
iteration_idx = i + 1
for feat_obj in d.keys():
feat_str = to_str(feat_obj)
if feat_str not in feature_birthdays:
feature_birthdays[feat_str] = iteration_idx
last_dict = refined_dicts[-1]
sorted_feats = sorted(last_dict.items(), key=lambda x: x[1], reverse=True)
features_to_plot = [to_str(f[0]) for f in sorted_feats[: int(top_k)]]
plt.figure(figsize=(12, 8))
labels = ["Initial"] + [f"Iteration {i + 1}" for i in range(len(refined_dicts))]
cmap = plt.cm.get_cmap("tab20", len(features_to_plot))
feature_colors = {feat: cmap(i) for i, feat in enumerate(features_to_plot)}
for feat in features_to_plot:
birth_idx = feature_birthdays.get(feat, 0)
v = []
color = feature_colors[feat] # unique color per feature
is_new = birth_idx > 0
if birth_idx == 0:
val = next((_v for _k, _v in initial_dict.items() if to_str(_k) == feat), 0)
v.append(val)
else:
v.append(None)
for i, d in enumerate(refined_dicts):
current_step = i + 1
if current_step < birth_idx:
v.append(None)
else:
step_val = next((_v for _k, _v in d.items() if to_str(_k) == feat), 0)
v.append(step_val)
plt.plot(
labels,
v,
marker="o",
color=color,
alpha=0.9 if is_new else 0.6,
linewidth=3 if is_new else 1.5,
linestyle="--" if is_new else "-", # dashes for new features
zorder=5 if is_new else 2,
)
if v[-1] is not None:
plt.text(len(labels) - 1, v[-1], f" {feat}", va="center", fontsize=9, color=color, fontweight="bold" if is_new else "normal")
plt.title(f"Feature Importance Shift (Top {top_k})")
plt.ylabel("Global SHAP Importance")
plt.grid(axis="y", linestyle="--", alpha=0.3)
plt.tight_layout()
plt.show()
def _get_shap_importance_dict(self, df):
"""Calculate mean absolute SHAP values for each feature and return as a dictionary.(Per iteration)"""
tree_explainer = TreeExplainer(self.clf)
shap_values = tree_explainer.shap_values(df.values)
if isinstance(shap_values, np.ndarray) and shap_values.ndim == 3:
importances = np.abs(shap_values[:, :, 1]).mean(axis=0)
else:
importances = np.abs(shap_values).mean(axis=0)
# Map to names: self.features must be synced with the current df columns
return dict(zip([owl_expression_to_dl(f) for f in self.features], importances))
def _merge_binary_feature_matrices(self, lp: PosNegLPStandard, X: pd.DataFrame, new_features: list, new_individuals_to_feature_mapping) -> pd.DataFrame:
"""Merge the original binary feature matrix with new features extracted from data properties."""
new_features_list = [v for k, v in new_features.items()]
dl_by_feature = {f: owl_expression_to_dl(f) for f in new_features_list}
positive_examples = [i for i in lp.pos]
negative_examples = [i for i in lp.neg]
examples = positive_examples + negative_examples
X_nf = []
for owl_named_individual in make_iterable_verbose(examples, verbose=self.verbose, desc="Constructing Training Data for Refined Features"):
binary_sparse_representation = []
features_of_owl_named_individual = None
if owl_named_individual.str in new_individuals_to_feature_mapping:
features_of_owl_named_individual = new_individuals_to_feature_mapping[owl_named_individual.str]
for owl_class_expression in new_features_list:
if features_of_owl_named_individual is not None and dl_by_feature[owl_class_expression] in features_of_owl_named_individual:
binary_sparse_representation.append(1.0)
else:
binary_sparse_representation.append(0.0)
X_nf.append(binary_sparse_representation)
X_nf = np.array(X_nf)
X_new = pd.DataFrame(data=X_nf, index=examples, columns=new_features_list)
X = pd.concat([X, X_new], axis=1)
self.features = X.columns.tolist()
same_value_columns = X.apply(lambda col: col.nunique() == 1)
X = X.loc[:, ~same_value_columns]
return X
def _refine_features_with_shap(self, X: pd.DataFrame, y: pd.DataFrame, learning_problem: PosNegLPStandard) -> pd.DataFrame:
"""Iteratively extend the feature set with SHAP-guided, distribution-based numeric data-property ranges.
Retrains self.clf after every feature-set extension and returns the final binary feature matrix.
"""
tree_explainer = TreeExplainer(self.clf)
sVal = tree_explainer.shap_values(self.X.values)
if self.verbose >= 10:
self.initial_importance_dict = self._get_shap_importance_dict(X)
# map individuals to additional data property features
# refine base features here ( extract from data properties)
dp_features: dict = dict()
generated_dt_class_expressions = self._extract_ranges_from_data_properties(dp_features, self.individuals_to_feature_mapping, self.data_properties_dict, self.per_individual_data_properties)
self.generated_dt_classexpressions_per_individual = generated_dt_class_expressions
X = self._merge_binary_feature_matrices(learning_problem, X, dp_features, self.individuals_to_feature_mapping)
self.X = X
self.clf = tree.DecisionTreeClassifier(**self.kwargs_classifier).fit(X=self.X.values, y=self.y.values)
if self.verbose >= 10:
self.concept_per_iteration.append(concepts_reducer(concepts=self.construct_owl_expression_from_tree(X, y), reduced_cls=OWLObjectUnionOf))
self.top_feature_dicts.append(self._get_shap_importance_dict(self.X))
refined_expressions: Set[OWLClassExpression] = set()
for i in range(1, self.refine_iterations):
if self.verbose > 0:
logger.info(f"Refinement iteration {i + 1}/{self.refine_iterations}")
# calculate SHAP global feature importance
tree_explainer = TreeExplainer(self.clf)
sVal = tree_explainer.shap_values(self.X.values)
if isinstance(sVal, np.ndarray):
if sVal.ndim == 3:
global_importance = np.abs(sVal[:, :, 1]).mean(axis=0)
else:
global_importance = np.abs(sVal).mean(axis=0)
topk_id = np.argsort(global_importance)
# top expressions sorted low to high
top_expressions: list[OWLClassExpression]
top_expressions = [self.features[i] for i in topk_id[-self.topk :].tolist()]
# print(top_expressions[-5:])
features_to_refine: set[OWLClassExpression] = (set(top_expressions) - refined_expressions)
refined_features, refined_individuals_to_feature_mapping = self.refine_numerical_features(features_to_refine, iteration=i + 1)
if self.verbose >0:
logger.info("Expressions to be refined: %d", len(features_to_refine))
if(len(features_to_refine) == 0):
if self.verbose > 0:
logger.info("No new expressions to refine. Stopping refinement.")
break
refined_expressions = refined_expressions.union(set(top_expressions))
X = self._merge_binary_feature_matrices(learning_problem, X, refined_features, refined_individuals_to_feature_mapping)
self.X = X
self.clf = tree.DecisionTreeClassifier(**self.kwargs_classifier).fit(X=self.X.values, y=self.y.values)
if self.verbose >= 10:
self.top_feature_dicts.append(self._get_shap_importance_dict(self.X))
self.concept_per_iteration.append(concepts_reducer(concepts=self.construct_owl_expression_from_tree(X, y), reduced_cls=OWLObjectUnionOf))
if self.plot_importance_evo and self.verbose >= 10:
self.plot_importance_evolution(self.initial_importance_dict, self.top_feature_dicts, top_k=10)
return X
[docs]
def fit(self, learning_problem: PosNegLPStandard = None, max_runtime: int = None):
"""Fit the learner to the given learning problem
(1) Extract multi-hop information about E^+ and E^-.
(2) Create OWL Class Expressions from (1)
(3) Build a binary sparse training data X where
first |E+| rows denote the binary representations of positives
Remaining rows denote the binary representations of E⁻
(4) Create binary labels.
(5) Train a decision tree classifier on (3) and (4)
(6) If feature refinement is enabled, refine the numerical dp features based on SHAP values and data property ranges.
(4) Construct a set of DL concept for each e \in E^+
(5) Union (4)
:param learning_problem: The learning problem
:param max_runtime:total runtime of the learning
"""
assert learning_problem is not None, "Learning problem cannot be None."
assert isinstance(learning_problem, PosNegLPStandard), f"Learning problem must be PosNegLPStandard. Currently:{learning_problem}."
if max_runtime is not None:
self.max_runtime = max_runtime
if self.verbose >= 10:
self.concept_per_iteration = []
self.top_feature_dicts = []
self.initial_importance_dict = {}
X: pd.DataFrame
y: Union[pd.DataFrame, pd.Series]
X, y = self.create_training_data(learning_problem=learning_problem)
# CD: Remember so that if user wants to use them
self.X, self.y = X, y
self._maybe_plot_embeddings(X, y)
self._fit_decision_tree(X, y)
if self.verbose >= 10:
self.concept_per_iteration.append(concepts_reducer(concepts=self.construct_owl_expression_from_tree(X, y), reduced_cls=OWLObjectUnionOf))
if self.feature_refinement:
X = self._refine_features_with_shap(X, y, learning_problem)
self._report_and_plot(X, y)
perm_feature_topk = False
if perm_feature_topk:
# plot permutation feature importance
res = permutation_importance(self.clf, X.values, y.values, n_repeats=10, random_state=0, n_jobs=12)
self.plot_perm_feature_importances(res, [owl_expression_to_dl(f) for f in self.features])
shap_feature_topk = False
# calculate SHAP values, to get topk features
if shap_feature_topk:
tree_explainer = TreeExplainer(self.clf)
sVal = tree_explainer.shap_values(X.values)
if isinstance(sVal, np.ndarray):
if sVal.ndim == 3:
global_importance = np.abs(sVal[:, :, 1]).mean(axis=0)
else:
global_importance = np.abs(sVal).mean(axis=0)
self.plot_shap_feature_importances(global_importance, [owl_expression_to_dl(f) for f in self.features])
self._finalize_predictions(X, y)
return self