"""
Evolutionary Optimization for Fuzzy Rule Base Learning
This module implements genetic algorithm-based optimization for learning fuzzy rule bases.
It provides automatic rule discovery, parameter tuning, and structure optimization for
fuzzy inference systems using evolutionary computation techniques.
Main Components:
- FitRuleBase: Core optimization problem class for genetic algorithms
- Fitness functions: Multiple objective functions for rule quality assessment
- Genetic operators: Specialized crossover, mutation, and selection for fuzzy rules
- Multi-objective optimization: Support for accuracy vs. complexity trade-offs
- Parallel evaluation: Efficient fitness evaluation using multiple threads
- Integration with Pymoo: Leverages the Pymoo optimization framework
The module supports automatic learning of:
- Rule antecedents (which variables and linguistic terms to use)
- Rule consequents (output class assignments)
- Rule structure (number of rules, complexity constraints)
- Membership function parameters (when combined with other modules)
Key Features:
- Stratified cross-validation for robust fitness evaluation
- Multiple fitness metrics (accuracy, MCC, F1-score, etc.)
- Support for Type-1, Type-2, and General Type-2 fuzzy systems
- Automatic handling of imbalanced datasets
- Configurable complexity penalties to avoid overfitting
"""
import os
import time
import warnings
from functools import wraps
from typing import Callable, Any, Optional, Union
import numpy as np
import pandas as pd
from sklearn.model_selection import StratifiedKFold
from sklearn.base import BaseEstimator, ClassifierMixin
from multiprocessing.pool import ThreadPool
# Import backend abstraction
# Optimizer-independent problem base: importing this module imports no pymoo.
from ._problem import Integer, Problem, StarmapParallelization
from . import evolutionary_backends as ev_backends
from . import fuzzy_sets as fs
from . import rules
from . import eval_rules as evr
from . import vis_rules
from .evolutionary_search import ExploreRuleBases
class _ConstructorValue:
"""Marker for fit arguments that default to the classifier's constructor setting."""
def __repr__(self) -> str:
return 'CONSTRUCTOR'
#: Default of the search settings of ``fit``: use the value given to the constructor.
CONSTRUCTOR = _ConstructorValue()
def _resolve(value, default):
return default if value is CONSTRUCTOR else value
def _check_no_missing(X, name: str = 'X') -> None:
"""
Reject missing or infinite feature values, which have no membership degree.
A NaN feature would give NaN firing strengths, and argmax would then pick the
first rule silently. Impute the data first, or use FERL with an observed mask.
"""
values = X.values if hasattr(X, 'values') else np.asarray(X)
if values.dtype.kind in 'fc':
invalid = ~np.isfinite(values)
elif values.dtype.kind in 'biu':
return
else:
invalid = np.asarray(pd.isna(values), dtype=bool)
if np.any(invalid):
columns = np.flatnonzero(np.any(np.atleast_2d(invalid), axis=0)).tolist()
raise ValueError(f'{name} contains missing or infinite values in columns {columns}. Fuzzy rule '
'classifiers need complete feature values: impute them first, or use FERL with an '
'observed mask for missing features.')
def _fit_scoped_thread_runner(fit_method):
"""
Create a classifier-owned PyMoo runner for one ``fit`` call only.
A caller can manually assign ``thread_runner``; that runner is external and
must never be closed here. Constructor-configured workers, however, are
owned by the classifier for precisely the duration of one fit.
"""
@wraps(fit_method)
def wrapped(self, *args, **kwargs):
external_runner = self.thread_runner
runner_count = getattr(self, 'runner', 1)
owns_pool = (
external_runner is None
and isinstance(runner_count, (int, np.integer))
and runner_count > 1
)
if not owns_pool:
return fit_method(self, *args, **kwargs)
pool = ThreadPool(runner_count)
try:
self.thread_runner = StarmapParallelization(pool.starmap)
return fit_method(self, *args, **kwargs)
finally:
# The wrapper owns a bound pool method. Drop it and join workers
# on all exits, including optimizer, callback, and finalization
# exceptions.
self.thread_runner = external_runner
pool.close()
pool.join()
return wrapped
[docs]
class BaseFuzzyRulesClassifier(ClassifierMixin, BaseEstimator):
"""
Class that is used as a classifier for a fuzzy rule based system. Supports precomputed and optimization of the linguistic variables.
"""
[docs]
def __init__(self, nRules: int = 30, nAnts: int = 4, fuzzy_type: fs.FUZZY_SETS = fs.FUZZY_SETS.t1, tolerance: float = 0.0, class_names: list[str] = None,
n_linguistic_variables: Union[list, int] = 3, verbose=False, linguistic_variables: list[fs.fuzzyVariable] = None, categorical_mask: list[int] = None,
domain: list[float] = None, n_class: int=None, precomputed_rules: rules.MasterRuleBase=None, runner: int=1, ds_mode: Union[int, str] = 0, allow_unknown:bool=False, backend: str='pymoo',
detect_categorical: bool = True, n_gen: int = 70, pop_size: int = 30, patience: Optional[int] = 10,
min_delta: float = 1e-4, random_state: int = 33, var_prob: float = 0.3, sbx_eta: float = 3.0,
mutation_eta: float = 7.0, tournament_size: int = 3) -> None:
"""
Inits the optimizer with the corresponding parameters.
Args:
nRules: number of rules to optimize.
nAnts: max number of antecedents to use.
fuzzy_type: FUZZY_SET enum type in fuzzy_sets module. The kind of fuzzy set used.
tolerance: tolerance for the dominance score of the rules.
n_linguist_variables: number of linguistic variables per antecedent.
verbose: if True, prints the progress of the optimization.
linguistic_variables: list of fuzzyVariables type. If None (default) the optimization process will init+optimize them.
domain: list of the limits for each variable. If None (default) the classifier will compute them empirically.
n_class: number of classes in the problem. If None (default) the classifier will compute it empirically.
precomputed_rules: MasterRuleBase object. If not None, the classifier will use the rules in the object and ignore the conflicting parameters.
runner: number of threads used to evaluate candidates. Threads disable the fit-local fitness and firing caches, so a serial fit (1, the default) is usually faster; more threads only pay off for an expensive custom loss.
ds_mode: inference weighting mode: 0 or 'dominance' weights rules by their dominance score, 1 or 'unweighted' uses the firing strengths alone, 2 or 'optimized' lets the genetic search set a weight per rule.
allow_unknown: if True, the classifier will allow the unknown class in the classification process. (Which would be a -1 value)
backend: evolutionary backend to use. Options: 'pymoo' (default, CPU) or 'evox' (GPU-accelerated). Install with: pip install ex-fuzzy[evox]
detect_categorical: if True (default) and no categorical_mask is given, the categorical variables are detected from the data with utils.detect_categorical_mask. Ignored when categorical_mask is given or when linguistic_variables are precomputed.
n_gen: number of generations of the genetic search. fit can override it for one call.
pop_size: population size of the genetic search. fit can override it for one call.
patience: generations without improvement before the search stops early; None runs every generation.
min_delta: minimum fitness improvement that resets the patience.
random_state: random seed of the genetic search.
var_prob: crossover probability.
sbx_eta: eta parameter of the SBX crossover.
mutation_eta: eta parameter of the polynomial mutation.
tournament_size: size of the selection tournament.
"""
rules.resolve_ds_mode(ds_mode) # Reject unknown modes early; fit resolves the code.
self.n_gen = n_gen
self.pop_size = pop_size
self.patience = patience
self.min_delta = min_delta
self.random_state = random_state
self.var_prob = var_prob
self.sbx_eta = sbx_eta
self.mutation_eta = mutation_eta
self.tournament_size = tournament_size
# Every constructor argument is kept under its own name, as
# scikit-learn's get_params, clone and repr require. The attributes
# below them are the derived state the rest of the library reads.
self.class_names = class_names
self.n_linguistic_variables = n_linguistic_variables
self.linguistic_variables = linguistic_variables
self.n_class = n_class
self.precomputed_rules = precomputed_rules
if precomputed_rules is not None:
self.nRules = len(precomputed_rules.get_rules())
self.nAnts = len(precomputed_rules.get_rules()[0].antecedents)
self.n_class = len(precomputed_rules)
self.nclasses_ = len(precomputed_rules.consequent_names)
self.classes_names = precomputed_rules.consequent_names
self.rule_base = precomputed_rules
else:
self.nRules = nRules
self.nAnts = nAnts
self.nclasses_ = n_class
if not (class_names is None):
if isinstance(class_names, np.ndarray):
self.classes_names = list(class_names)
else:
self.classes_names = class_names
else:
self.classes_names = class_names
self.categorical_mask = categorical_mask
self.custom_loss = None
self.verbose = verbose
self.tolerance = tolerance
self.ds_mode = ds_mode
self.allow_unknown = allow_unknown
self.detect_categorical = detect_categorical
# Initialize evolutionary backend. A backend instance is kept as is,
# which is how a clone receives this parameter.
try:
self.backend = ev_backends.get_backend(backend)
if verbose:
print(f"Using evolutionary backend: {self.backend.name()}")
except ValueError as e:
warnings.warn(f"{e} Falling back to the pymoo backend.", stacklevel=2)
self.backend = ev_backends.get_backend('pymoo')
self.runner = runner
# A wrapper around ``pool.starmap`` retains the pool. Keep no pool on
# an unfitted estimator; the fit decorator owns the temporary runner.
self.thread_runner = None
if linguistic_variables is not None:
# If the linguistic variables are precomputed then we act accordingly
self.lvs = linguistic_variables
self.n_linguist_variables = [len(lv.linguistic_variable_names()) for lv in self.lvs]
self.domain = None
self.fuzzy_type = self.lvs[0].fuzzy_type()
if self.nAnts > len(linguistic_variables):
self.nAnts = len(linguistic_variables)
warnings.warn('The number of antecedents is higher than the number of variables. '
'Setting nAnts to the number of linguistic variables. (%d)' % len(linguistic_variables),
stacklevel=2)
else:
# If not, then we need the parameters sumistered by the user.
self.lvs = None
self.fuzzy_type = fuzzy_type
self.n_linguist_variables = n_linguistic_variables
self.domain = domain
self.alpha_ = 0.0
self.beta_ = 0.0
[docs]
def customized_loss(self, loss_function):
"""
Function to customize the loss function used for the optimization.
Args:
loss_function: function that takes as input the true labels and the predicted labels and returns a float.
Returns:
None
"""
self.custom_loss = loss_function
[docs]
@_fit_scoped_thread_runner
def fit(self, X: np.array, y: np.array, n_gen: int = CONSTRUCTOR, pop_size: int = CONSTRUCTOR,
checkpoints:int=0, candidate_rules:rules.MasterRuleBase=None, initial_rules:rules.MasterRuleBase=None, random_state: int = CONSTRUCTOR,
var_prob: float = CONSTRUCTOR, sbx_eta: float = CONSTRUCTOR, mutation_eta: float = CONSTRUCTOR, tournament_size: int = CONSTRUCTOR, bootstrap_size:int=1000, checkpoint_path:str='',
p_value_compute:bool=False, checkpoint_callback: Callable[[int, rules.MasterRuleBase], None] = None,
patience: Optional[int] = CONSTRUCTOR, min_delta: float = CONSTRUCTOR):
"""
Fits a fuzzy rule based classifier using a genetic algorithm to the given data.
The search settings (n_gen, pop_size, patience, min_delta, random_state, var_prob, sbx_eta,
mutation_eta and tournament_size) default to the values given to the constructor; passing
them here overrides them for this fit only.
Args:
X: numpy array samples x features
y: labels. integer array samples (x 1)
n_gen: integer. Number of generations to run the genetic algorithm.
pop_size: integer. Population size for each gneration.
checkpoints: integer. Number of checkpoints to save the best rulebase found so far.
candidate_rules: if these rules exist, the optimization process will choose the best rules from this set. If None (default) the rules will be generated from scratch.
initial_rules: if these rules exist, the optimization process will start from this set. If None (default) the rules will be generated from scratch.
random_state: integer. Random seed for the optimization process.
var_prob: float. Probability of crossover for the genetic algorithm.
sbx_eta: float. Eta parameter for the SBX crossover.
checkpoint_path: string. Path to save the checkpoints. If None (default) the checkpoints will be saved in the current directory.
mutation_eta: float. Eta parameter for the polynomial mutation.
tournament_size: integer. Size of the tournament for the genetic algorithm.
checkpoint_callback: function. Callback function that get executed at each checkpoint ('checkpoints' must be greater than 0), its arguments are the generation number and the rule_base of the checkpoint.
patience: integer. Stop early when the best fitness does not improve for this many generations. Default is 10. Use None to run all generations.
min_delta: float. Minimum fitness improvement required to reset patience. Default is 1e-4.
Returns:
the fitted classifier.
"""
n_gen, pop_size = _resolve(n_gen, self.n_gen), _resolve(pop_size, self.pop_size)
random_state = _resolve(random_state, self.random_state)
var_prob, sbx_eta = _resolve(var_prob, self.var_prob), _resolve(sbx_eta, self.sbx_eta)
mutation_eta = _resolve(mutation_eta, self.mutation_eta)
tournament_size = _resolve(tournament_size, self.tournament_size)
patience, min_delta = _resolve(patience, self.patience), _resolve(min_delta, self.min_delta)
_check_no_missing(X)
ds_mode = rules.resolve_ds_mode(self.ds_mode)
if patience is not None and patience <= 0:
patience = None
min_delta = max(0.0, float(min_delta))
# Detected before X loses its column dtypes below.
categorical_mask = self.categorical_mask
if categorical_mask is None and self.detect_categorical and self.lvs is None:
from . import utils
detected = utils.detect_categorical_mask(X)
if np.any(detected > 0):
categorical_mask = detected
if self.verbose:
print('Detected categorical variables: ' + str(np.flatnonzero(detected).tolist()))
if isinstance(X, pd.DataFrame):
lvs_names = list(X.columns)
X = X.values
else:
lvs_names = [str(ix) for ix in range(X.shape[1])]
self.n_features_in_ = X.shape[1]
y = np.asarray(y)
if y.ndim > 1:
y = np.ravel(y)
if self.classes_names is None:
self.classes_names = [aux for aux in np.unique(y)]
if self.nclasses_ is None:
self.nclasses_ = len(self.classes_names)
# The search works on consequent indexes; predict maps them back.
y = self._encode_labels(y)
if candidate_rules is None:
if initial_rules is not None:
self.fuzzy_type = initial_rules.fuzzy_type()
self.n_linguist_variables = initial_rules.n_linguistic_variables()
self.domain = [fv.domain() for fv in initial_rules[0].antecedents]
self.nRules = len(initial_rules.get_rules())
self.nAnts = len(initial_rules.get_rules()[0].antecedents)
# Use linguistic variables from the initial rules (don't optimize memberships)
if self.lvs is None:
self.lvs = initial_rules[0].antecedents
if self.lvs is None:
# Check if self.n_linguist_variables is a list or a single value.
if isinstance(self.n_linguist_variables, int):
self.n_linguist_variables = [self.n_linguist_variables for _ in range(X.shape[1])]
if self.nAnts > X.shape[1]:
self.nAnts = X.shape[1]
warnings.warn('The number of antecedents is higher than the number of variables. '
'Setting nAnts to the number of variables. (%d)' % X.shape[1], stacklevel=2)
# If Fuzzy variables need to be optimized.
problem = FitRuleBase(X, y, nRules=self.nRules, nAnts=self.nAnts, tolerance=self.tolerance, n_classes=len(np.unique(y)),
n_linguistic_variables=self.n_linguist_variables, fuzzy_type=self.fuzzy_type, domain=self.domain, thread_runner=self.thread_runner,
alpha=self.alpha_, beta=self.beta_, ds_mode=ds_mode, categorical_mask=categorical_mask,
allow_unknown=self.allow_unknown, backend_name=self.backend.name(), var_names=lvs_names)
else:
# If Fuzzy variables are already precomputed.
problem = FitRuleBase(X, y, nRules=self.nRules, nAnts=self.nAnts, n_classes=len(np.unique(y)),
linguistic_variables=self.lvs, domain=self.domain, tolerance=self.tolerance, thread_runner=self.thread_runner,
alpha=self.alpha_, beta=self.beta_, ds_mode=ds_mode,
allow_unknown=self.allow_unknown, backend_name=self.backend.name(), var_names=lvs_names)
else:
self.fuzzy_type = candidate_rules.fuzzy_type()
self.n_linguist_variables = candidate_rules.n_linguistic_variables()
problem = ExploreRuleBases(X, y, n_classes=len(np.unique(y)), candidate_rules=candidate_rules, thread_runner=self.thread_runner, nRules=self.nRules)
if self.custom_loss is not None:
problem.fitness_func = self.custom_loss
# Prepare initial population
if initial_rules is None:
rules_gene = None # Will use default random sampling
else:
rules_gene = problem.encode_rulebase(initial_rules, self.lvs is None)
rules_gene = (np.ones((pop_size, len(rules_gene))) * rules_gene).astype(int)
from ._fitness import _fitness_cache_scope
# Only a built-in serial search has a private, stable fit context for
# the fit-local caches. Keep callbacks, custom losses/backends and
# workers on their existing evaluation path. EvoX scores whole
# generations through the same cached routes, so it qualifies as well.
cache_enabled = (type(problem) is FitRuleBase and self.custom_loss is None
and type(self.backend) in (ev_backends.PyMooBackend,
ev_backends.EvoXBackend)
and self.thread_runner is None)
# Use backend for optimization
if checkpoints > 0:
# Checkpoint mode - delegate to backend if supported
if self.backend.name() == 'pymoo':
# Define checkpoint handler
def handle_checkpoint(gen: int, best_individual: np.array):
rule_base = problem._construct_ruleBase(best_individual, self.fuzzy_type)
eval_performance = evr.evalRuleBase(rule_base, np.array(X), y)
eval_performance.add_full_evaluation()
rule_base.purge_rules(self.tolerance)
rule_base.rename_cons(self.classes_names)
checkpoint_rules = rule_base.print_rules(True, bootstrap_results=True)
if checkpoint_callback is None:
with open(os.path.join(checkpoint_path, "checkpoint_" + str(gen)), "w") as f:
f.write(checkpoint_rules)
else:
checkpoint_callback(gen, rule_base)
# Call backend's checkpoint optimization
result = self.backend.optimize_with_checkpoints(
problem=problem,
n_gen=n_gen,
pop_size=pop_size,
random_state=random_state,
verbose=self.verbose,
checkpoint_freq=checkpoints,
checkpoint_callback=handle_checkpoint,
var_prob=var_prob,
sbx_eta=sbx_eta,
mutation_eta=mutation_eta,
tournament_size=tournament_size,
sampling=rules_gene,
patience=patience,
min_delta=min_delta
)
best_individual = result['X']
self.performance = 1 - result['F']
else:
# EvoX or other backends: checkpoints not supported
warnings.warn(f"Checkpoints are not yet supported with {self.backend.name()} backend. "
"Running without checkpoints.", stacklevel=2)
with _fitness_cache_scope(problem, cache_enabled, pop_size):
result = self.backend.optimize(
problem=problem,
n_gen=n_gen,
pop_size=pop_size,
random_state=random_state,
verbose=self.verbose,
var_prob=var_prob,
sbx_eta=sbx_eta,
mutation_eta=mutation_eta,
tournament_size=tournament_size,
sampling=rules_gene,
patience=patience,
min_delta=min_delta
)
best_individual = result['X']
self.performance = 1 - result['F']
else:
# Normal optimization without checkpoints
with _fitness_cache_scope(problem, cache_enabled, pop_size):
result = self.backend.optimize(
problem=problem,
n_gen=n_gen,
pop_size=pop_size,
random_state=random_state,
verbose=self.verbose,
var_prob=var_prob,
sbx_eta=sbx_eta,
mutation_eta=mutation_eta,
tournament_size=tournament_size,
sampling=rules_gene,
patience=patience,
min_delta=min_delta
)
best_individual = result['X']
self.performance = 1 - result['F']
self.optimization_result_ = result
self.n_generations_run_ = result.get('n_gen_run', n_gen)
self.stopped_early_ = result.get('stopped_early', False)
# X was converted to an array above, keeping the column names.
self.X = X
self.var_names = lvs_names
self.rule_base = problem._construct_ruleBase(
best_individual, self.fuzzy_type)
self.lvs = self.rule_base.rule_bases[0].antecedents if self.lvs is None else self.lvs
# Finalization requests the same firing strengths repeatedly while it
# computes rule weights, pruning accuracy, and the public metrics. Reuse
# fixed-partition memberships from the problem; for the one selected
# optimized partition, compute them once. Both temporary memberships and
# firing matrices are released before optional resampling or fit return.
finalization_truth = getattr(problem, '_precomputed_truth', None)
if (finalization_truth is None and type(problem) is FitRuleBase
and self.rule_base.get_rules()):
finalization_truth = rules.compute_antecedents_memberships(
self.rule_base.antecedents, np.asarray(X))
self.eval_performance = evr.evalRuleBase(
self.rule_base, np.array(X), y,
precomputed_truth=finalization_truth)
try:
with self.rule_base._firing_cache_scope():
# Pruning needs per-rule dominance scores and winning-rule
# accuracy, but not the global MCC/accuracy. The latter is
# computed after pruning and is the public final evaluation.
self.eval_performance.add_rule_weights()
self.eval_performance.add_classification_metrics()
self.rule_base.purge_rules(self.tolerance)
self.eval_performance.add_full_evaluation()
finally:
self.eval_performance.precomputed_truth = None
finalization_truth = None
if p_value_compute:
self.p_value_validation(bootstrap_size)
self.rule_base.rename_cons(self.classes_names)
return self
def _encode_labels(self, y: np.ndarray) -> np.ndarray:
"""
Maps the training labels to consequent indexes, in the order of classes_names, and sets classes_.
Args:
y: array with the labels.
Returns:
integer array with the consequent index of each label.
Raises:
ValueError: if a label is neither a class name nor a consequent index.
"""
names = self.classes_names
lookup = {}
for index, name in enumerate(names):
lookup.setdefault(name, index)
try:
encoded = np.fromiter((lookup[label] for label in y), dtype=int, count=len(y))
except KeyError:
pass
else:
self.classes_ = np.asarray(names)
return encoded
# Integer labels with class_names given as their display names.
numeric = y.dtype.kind in 'biuf' and (y.size == 0 or (np.all(np.floor(y) == y)
and y.min() >= 0 and y.max() < len(names)))
if numeric:
self.classes_ = np.arange(len(names))
return y.astype(int)
unknown = sorted({str(label) for label in y if label not in lookup})
raise ValueError(f'Labels {unknown} are not among the class names {list(names)}.')
def _decode_predictions(self, indexes: np.ndarray) -> np.ndarray:
"""
Maps consequent indexes to the labels the classifier was fitted with.
Samples without a firing rule (-1) stay -1 for numeric labels and become 'Unknown' otherwise.
A classifier without fitted classes_ (built from precomputed rules) returns the indexes unchanged.
Args:
indexes: array with the consequent index of each sample.
Returns:
array with the label of each sample.
"""
classes = getattr(self, 'classes_', None)
indexes = np.asarray(indexes)
if classes is None or indexes.dtype.kind not in 'biuf':
return indexes
indexes = indexes.astype(int)
known = indexes >= 0
labels = classes[np.where(known, indexes, 0)]
if known.all():
return labels
unknown = -1 if classes.dtype.kind in 'biuf' else 'Unknown'
return np.where(known, labels, unknown)
[docs]
def print_rule_bootstrap_results(self) -> None:
"""
Prints the bootstrap results for each rule.
"""
self.rule_base.print_rule_bootstrap_results()
[docs]
def p_value_validation(self, bootstrap_size:int=100):
"""
Computes the permutation and bootstrapping p-values for the classifier and its rules.
Args:
bootstrap_size: integer. Number of bootstraps samples to use.
"""
self.p_value_class_structure, self.p_value_feature_coalitions = self.eval_performance.p_permutation_classifier_validation()
self.eval_performance.p_bootstrapping_rules_validation(bootstrap_size)
[docs]
def load_master_rule_base(self, rule_base: rules.MasterRuleBase) -> None:
"""
Loads a master rule base to be used in the prediction process.
Args:
rule_base: ruleBase object.
Returns:
None
"""
self.rule_base = rule_base
self.nRules = len(rule_base.get_rules())
self.nAnts = len(rule_base.get_rules()[0].antecedents)
self.nclasses_ = len(rule_base)
[docs]
def explainable_predict(self, X: np.array, out_class_names=False) -> np.array:
"""
Returns the predicted class for each sample, with the winning rule, its association degree and confidence interval.
Args:
X: np array samples x features.
out_class_names: if True, the predictions are the consequent names instead of the fitted labels.
Returns:
an ExplainedPrediction named tuple: predictions, winning rules, winning association degrees and confidence intervals.
"""
try:
X = X.values # If X was a pandas dataframe
except AttributeError:
pass
_check_no_missing(X)
prediction, winners, association, confidence = self.rule_base.explainable_predict(
X, out_class_names=out_class_names)
if not out_class_names:
prediction = self._decode_predictions(prediction)
return rules.ExplainedPrediction(prediction, winners, association, confidence)
[docs]
def forward(self, X: np.array, out_class_names=False) -> np.array:
"""
Returns the predicted class for each sample.
Args:
X: np array samples x features.
out_class_names: if True, the output will be the consequent names instead of the fitted labels.
Returns:
np array samples (x 1) with the predicted class.
"""
try:
X = X.values # If X was a pandas dataframe
except AttributeError:
pass
_check_no_missing(X)
if out_class_names:
return self.rule_base.winning_rule_predict(X, out_class_names=True)
return self._decode_predictions(self.rule_base.winning_rule_predict(X))
[docs]
def predict(self, X: np.array, out_class_names=False) -> np.array:
"""
Returns the predicted class for each sample.
A fitted classifier predicts the labels it was fitted with (see ``classes_``); samples where
no rule fires are -1 for numeric labels and 'Unknown' otherwise. A classifier built from
precomputed rules predicts consequent indexes.
Args:
X: np array samples x features.
out_class_names: if True, the output will be the consequent names instead of the fitted labels.
Returns:
np array samples (x 1) with the predicted class.
"""
return self.forward(X, out_class_names=out_class_names)
[docs]
def predict_proba_rules(self, X: np.array, truth_degrees:bool=True) -> np.array:
"""
Returns the predicted class probabilities for each sample.
Args:
X: np array samples x features.
truth_degrees: if True, the output will be the truth degrees of the rules. If false, will return the association degrees i.e. the truth degree multiplied by the weights/dominance of the rules. (depending on the inference mode chosen)
Returns:
np array samples x classes with the predicted class probabilities.
"""
try:
X = X.values # If X was a pandas dataframe
except AttributeError:
pass
_check_no_missing(X)
if truth_degrees:
return self.rule_base.compute_firing_strengths(X)
else:
return self.rule_base.compute_association_degrees(X)
[docs]
def predict_membership_class(self, X: np.array) -> np.array:
"""
Returns the predicted class memberships for each sample.
Args:
X: np array samples x features.
Returns:
np array samples x classes with the predicted class probabilities.
"""
try:
X = X.values # If X was a pandas dataframe
except AttributeError:
pass
_check_no_missing(X)
rule_predict_proba = self.rule_base.compute_association_degrees(X)
rule_consequents = self.rule_base.get_consequents()
res = np.zeros((X.shape[0], self.nclasses_))
for jx in range(rule_predict_proba.shape[1]):
consequent = rule_consequents[jx]
res[:, consequent] = np.maximum(res[:, consequent], rule_predict_proba[:, jx])
return res
[docs]
def predict_proba(self, X:np.array) -> np.array:
"""
Returns the predicted class probabilities for each sample.
Args:
X: np array samples x features.
Returns:
np array samples x classes with the predicted class probabilities.
"""
beliefs = self.predict_membership_class(X)
# Normalize beliefs to sum to 1; handle zero-sum cases with uniform distribution
row_sums = np.sum(beliefs, axis=1, keepdims=True)
zero_mask = row_sums == 0
row_sums[zero_mask] = 1 # Avoid division by zero
beliefs = beliefs / row_sums
# For samples with no rule activation, assign uniform probability
if np.any(zero_mask):
beliefs[zero_mask.flatten()] = 1.0 / beliefs.shape[1]
return beliefs
[docs]
def print_rules(self, return_rules:bool=False, bootstrap_results:bool=False) -> None:
"""
Print the rules contained in the fitted rulebase.
"""
return self.rule_base.print_rules(return_rules, bootstrap_results)
[docs]
def plot_fuzzy_variables(self) -> None:
"""
Plot the fuzzy partitions in each fuzzy variable.
"""
fuzzy_variables = self.rule_base.rule_bases[0].antecedents
for ix, fv in enumerate(fuzzy_variables):
vis_rules.plot_fuzzy_variable(fv)
[docs]
def rename_fuzzy_variables(self) -> None:
"""
Renames the linguist labels so that high, low and so on are consistent. It does so usually after an optimization process.
Returns:
None. Names are sorted accorded to the central point of the fuzzy memberships.
"""
for ix in range(len(self.rule_base)):
fuzzy_variables = self.rule_base.rule_bases[ix].antecedents
for jx, fv in enumerate(fuzzy_variables):
if fv[0].shape() != 'categorical':
new_order_values = []
possible_names = FitRuleBase.vl_names[self.n_linguist_variables[jx]]
for zx, fuzzy_set in enumerate(fv.linguistic_variables):
studied_fz = fuzzy_set.type()
if studied_fz == fs.FUZZY_SETS.temporal:
studied_fz = fuzzy_set.inside_type()
if studied_fz == fs.FUZZY_SETS.t1:
f1 = np.mean(
fuzzy_set.membership_parameters[0] + fuzzy_set.membership_parameters[1])
elif (studied_fz == fs.FUZZY_SETS.t2):
f1 = np.mean(
fuzzy_set.secondMF_upper[0] + fuzzy_set.secondMF_upper[1])
elif studied_fz == fs.FUZZY_SETS.gt2: # pragma: no branch - supported types are exhaustive
sec_memberships = fuzzy_set.secondary_memberships.values()
f1 = float(list(fuzzy_set.secondary_memberships.keys())[np.argmax(
[fzm.membership_parameters[2] for ix, fzm in enumerate(sec_memberships)])])
new_order_values.append(f1)
new_order = np.argsort(np.array(new_order_values))
fuzzy_sets_vl = fv.linguistic_variables
for jx, x in enumerate(new_order):
fuzzy_sets_vl[x].name = possible_names[jx]
[docs]
def get_rulebase(self) -> list[np.array]:
"""
Get the rulebase obtained after fitting the classifier to the data.
Returns:
a matrix format for the rulebase.
"""
return self.rule_base.get_rulebase_matrix()
[docs]
def reparametrize_loss(self, alpha:float, beta:float) -> None:
"""
Changes the parameters in the loss function.
Args:
alpha: controls the MCC term.
beta: controls the average rule size loss.
Note:
Does not check for convexity preservation. The user can play with these parameters as it wills.
"""
self.alpha_ = alpha
self.beta_ = beta
[docs]
def reparametrice_loss(self, alpha: float, beta: float) -> None:
"""
Deprecated spelling of reparametrize_loss.
"""
warnings.warn('reparametrice_loss is deprecated; use reparametrize_loss.', DeprecationWarning, stacklevel=2)
self.reparametrize_loss(alpha, beta)
[docs]
def __call__(self, X:np.array) -> np.array:
"""
Returns the predicted class for each sample.
Args:
X: np array samples x features.
Returns:
np array samples (x 1) with the predicted class.
"""
return self.predict(X)
[docs]
class FitRuleBase(Problem):
"""
Class to model, independently of the optimizer, the fitting of a rulebase for a classification problem using Evolutionary strategies.
Supports type 1 and iv fs (iv-type 2)
"""
def _init_optimize_vl(self, fuzzy_type: fs.FUZZY_SETS, n_linguist_variables: int, domain: list[(float, float)] = None, categorical_variables: list[int] = None, X=None):
"""
Inits the corresponding fields if no linguistic partitions were given.
Args:
fuzzy_type: FUZZY_SET enum type in fuzzy_sets module. The kind of fuzzy set used.
n_linguistic_variables: number of linguistic variables per antecedent.
domain: list of the limits for each variable. If None (default) the classifier will compute them empirically.
"""
from . import utils
self.lvs = None
self.vl_names = [FitRuleBase.vl_names[n_vars] if n_vars < len(FitRuleBase.vl_names) else [str(ix) for ix in range(n_vars)] for n_vars in n_linguist_variables]
self.fuzzy_type = fuzzy_type
self.domain = domain
self._precomputed_truth = None
self.categorical_mask = categorical_variables
self.categorical_boolean_mask = np.array(categorical_variables) > 0 if categorical_variables is not None else None
self.categorical_variables = {}
for ix, cat in enumerate(categorical_variables):
if cat > 0:
self.categorical_variables[ix] = utils.construct_crisp_categorical_partition(np.array(X)[:, ix], self.var_names[ix], fuzzy_type)
self.n_lv_possible = []
for ix in range(len(self.categorical_mask)):
if self.categorical_mask[ix] > 0:
self.n_lv_possible.append(len(self.categorical_variables[ix]))
else:
self.n_lv_possible.append(n_linguist_variables[ix])
def _init_precomputed_vl(self, linguist_variables: list[fs.fuzzyVariable], X: np.array):
"""
Inits the corresponding fields if linguistic partitions for each variable are given.
Args:
linguistic_variables: list of fuzzyVariables type.
X: np array samples x features.
"""
self.lvs = linguist_variables
self.vl_names = [lv.linguistic_variable_names() for lv in self.lvs]
self.n_lv_possible = [len(lv.linguistic_variable_names()) for lv in self.lvs]
self.fuzzy_type = self.lvs[0].fs_type
self.domain = None
self._precomputed_truth = rules.compute_antecedents_memberships(linguist_variables, X)
vl_names = [ # Linguistic variable names prenamed for some specific cases.
[],
[],
['Low', 'High'],
['Low', 'Medium', 'High'],
['Low', 'Medium', 'High', 'Very High'],
['Very Low', 'Low', 'Medium', 'High', 'Very High']
]
[docs]
def __init__(self, X: np.array, y: np.array, nRules: int, nAnts: int, n_classes: int, thread_runner: Optional[Any]=None,
linguistic_variables:list[fs.fuzzyVariable]=None, n_linguistic_variables:int=3, fuzzy_type=fs.FUZZY_SETS.t1, domain:list=None, categorical_mask: np.array=None,
tolerance:float=0.01, alpha:float=0.0, beta:float=0.0, ds_mode: int =0, allow_unknown:bool=False, backend_name:str='pymoo', var_names:list=None) -> None:
"""
Cosntructor method. Initializes the classifier with the number of antecedents, linguist variables and the kind of fuzzy set desired.
Args:
X: np array or pandas dataframe samples x features.
y: np vector containing the target classes. vector sample
nRules: number of rules to optimize.
nAnts: max number of antecedents to use.
n_class: number of classes in the problem. If None (as default) it will be computed from the data.
linguistic_variables: list of linguistic variables precomputed. If given, the rest of conflicting arguments are ignored.
n_linguistic_variables: number of linguistic variables per antecedent.
fuzzy_type: Define the fuzzy set or fuzzy set extension used as linguistic variable.
domain: list with the upper and lower domains of each input variable. If None (as default) it will stablish the empirical min/max as the limits.
tolerance: float. Tolerance for the size evaluation.
alpha: float. Weight for the rulebase size term in the fitness function. (Penalizes number of rules)
beta: float. Weight for the average rule size term in the fitness function.
ds_mode: int. Mode for the dominance score. 0: normal dominance score, 1: rules without weights, 2: weights optimized for each rule based on the data.
allow_unknown: if True, the classifier will allow the unknown class in the classification process. (Which would be a -1 value)
var_names: list of variable names. If None, extracted from DataFrame columns or auto-generated.
"""
if var_names is not None:
self.var_names = var_names
self.X = np.array(X) if not isinstance(X, np.ndarray) else X
else:
try:
self.var_names = list(X.columns)
self.X = X.values
except AttributeError:
self.X = X
self.var_names = [str(ix) for ix in range(X.shape[1])]
self.tolerance = tolerance
self.y = y
self.classes_names = np.unique(y)
self.nRules = nRules
self.nAnts = nAnts
self.nCons = 1 # This is fixed to MISO rules.
self.ds_mode = rules.resolve_ds_mode(ds_mode)
self.allow_unknown = allow_unknown
if n_classes is not None:
self.n_classes = n_classes
else:
self.n_classes = len(np.unique(y))
if categorical_mask is None:
self.categorical_mask = np.zeros(X.shape[1])
categorical_mask = self.categorical_mask
if linguistic_variables is not None:
self._init_precomputed_vl(linguistic_variables, X)
else:
if isinstance(n_linguistic_variables, int):
n_linguistic_variables = [n_linguistic_variables] * self.X.shape[1]
self._init_optimize_vl(
fuzzy_type=fuzzy_type, n_linguist_variables=n_linguistic_variables, categorical_variables=categorical_mask, domain=domain, X=X)
possible_antecedent_bounds = np.array(
[[0, self.X.shape[1] - 1]] * self.nAnts * self.nRules)
vl_antecedent_bounds = np.array(
[[-1, self.n_lv_possible[ax] - 1] for ax in range(self.nAnts)] * self.nRules) # -1 means not caring
antecedent_bounds = np.concatenate(
(possible_antecedent_bounds, vl_antecedent_bounds))
vars_antecedent = {ix: Integer(
bounds=antecedent_bounds[ix]) for ix in range(len(antecedent_bounds))}
aux_counter = len(vars_antecedent)
if self.lvs is None:
self.feature_domain_bounds = np.array(
[[0, 99] for ix in range(self.X.shape[1])])
if self.fuzzy_type == fs.FUZZY_SETS.t1:
correct_size = [(self.n_lv_possible[ixx]-1) * 4 + 3 for ixx in range(len(self.n_lv_possible))]
elif self.fuzzy_type == fs.FUZZY_SETS.t2:
correct_size = [(self.n_lv_possible[ixx]-1) * 6 + 2 for ixx in range(len(self.n_lv_possible))]
else:
raise ValueError(f"Fuzzy type {self.fuzzy_type} not supported for dynamic membership optimization. "
"Please provide precomputed linguistic_variables.")
membership_bounds = np.concatenate(
[[self.feature_domain_bounds[ixx]] * correct_size[ixx] for ixx in range(len(self.n_lv_possible))])
vars_memberships = {
aux_counter + ix: Integer(bounds=membership_bounds[ix]) for ix in range(len(membership_bounds))}
aux_counter += len(vars_memberships)
final_consequent_bounds = np.array(
[[-1, self.n_classes - 1]] * self.nRules)
vars_consequent = {aux_counter + ix: Integer(
bounds=final_consequent_bounds[ix]) for ix in range(len(final_consequent_bounds))}
if self.lvs is None:
vars = {key: val for d in [
vars_antecedent, vars_memberships, vars_consequent] for key, val in d.items()}
varbound = np.concatenate(
(antecedent_bounds, membership_bounds, final_consequent_bounds), axis=0)
else:
vars = {key: val for d in [vars_antecedent,
vars_consequent] for key, val in d.items()}
varbound = np.concatenate(
(antecedent_bounds, final_consequent_bounds), axis=0)
if self.ds_mode == 2:
weights_bounds = np.array([[0, 99] for ix in range(self.nRules)])
vars_weights = {max(vars.keys()) + 1 + ix: Integer(
bounds=weights_bounds[ix]) for ix in range(len(weights_bounds))}
vars = {key: val for d in [vars, vars_weights] for key, val in d.items()}
varbound = np.concatenate((varbound, weights_bounds), axis=0)
nVar = len(varbound)
self.single_gen_size = nVar
self.alpha_ = alpha
self.beta_ = beta
self.backend_name = backend_name
if self.lvs is None:
self._normalization_domain()
if thread_runner is not None:
self._external_elementwise_runner = True
super().__init__(
vars=vars,
n_var=nVar,
n_obj=1,
elementwise=True,
vtype=int,
xl=varbound[:, 0],
xu=varbound[:, 1],
elementwise_runner=thread_runner)
else:
self._external_elementwise_runner = False
super().__init__(
vars=vars,
n_var=nVar,
n_obj=1,
elementwise=True,
vtype=int,
xl=varbound[:, 0],
xu=varbound[:, 1])
def _normalization_domain(self) -> tuple:
"""
Return fit-local empirical bounds for chromosome normalization.
These deliberately use the decoder's nan-aware empirical bounds, not
the user-supplied sampling domain. A new problem is created on refit.
"""
if not hasattr(self, '_normalization_domain_cache'):
minimum = np.zeros(self.X.shape[1])
maximum = np.zeros(self.X.shape[1])
for ix in range(self.X.shape[1]):
column = self.X[:, ix]
# A numerical column keeps its numbers even inside an object
# array (a mixed DataFrame): only categorical columns count
# their categories.
mask = getattr(self, 'categorical_mask', None)
categorical = mask is not None and mask[ix] > 0
numeric = None
if not categorical:
try:
numeric = column.astype(float)
except (TypeError, ValueError):
numeric = None
if numeric is not None:
minimum[ix] = np.nanmin(numeric)
maximum[ix] = np.nanmax(numeric)
else:
maximum[ix] = len(np.unique(column[~pd.isna(column)]))
span = maximum - minimum
for values in (minimum, maximum, span):
values.flags.writeable = False
self._normalization_domain_cache = minimum, maximum, span
return self._normalization_domain_cache
def _decode_membership_functions(self, x: np.array, fuzzy_type: fs.FUZZY_SETS) -> list[fs.fuzzyVariable]:
"""
Decode membership function parameters from gene encoding.
Args:
x: gene array containing encoded membership function parameters
fuzzy_type: type of fuzzy set (t1 or t2)
Returns:
list of fuzzyVariable objects with decoded membership functions
"""
third_pointer = 2 * self.nAnts * self.nRules
aux_pointer = 0
antecedents = []
for fuzzy_variable in range(self.X.shape[1]):
linguistic_variables = []
lv_FS = []
for lx in range(self.n_lv_possible[fuzzy_variable]):
parameter_pointer = third_pointer + aux_pointer
if fuzzy_type == fs.FUZZY_SETS.t1:
if lx == 0:
fz_parameters_idx0 = x[parameter_pointer]
fz_parameters_idx1 = x[parameter_pointer + 1]
fz_parameters_idx2 = x[parameter_pointer + 2]
fz_parameters_idx3 = x[parameter_pointer + 3]
fz0 = fz_parameters_idx0
fz1 = fz_parameters_idx0
fz2 = fz1 + fz_parameters_idx1
next_fz0 = fz2 + fz_parameters_idx2
fz3 = next_fz0 + fz_parameters_idx3
fz_parameters = np.array([fz0, fz1, fz2, fz3])
aux_pointer += 4
elif lx == self.n_lv_possible[fuzzy_variable] - 1:
fz_parameters_idx1 = x[parameter_pointer]
fz_parameters_idx2 = x[parameter_pointer + 1]
fz0 = next_fz0
fz1 = fz3 + fz_parameters_idx1
fz2 = fz1 + fz_parameters_idx2
fz3 = fz2
fz_parameters = np.array([fz0, fz1, fz2, fz3])
aux_pointer += 3
else:
fz_parameters_idx1 = x[parameter_pointer]
fz_parameters_idx2 = x[parameter_pointer + 1]
fz_parameters_idx3 = x[parameter_pointer + 2]
fz_parameters_idx4 = x[parameter_pointer + 3]
fz0 = next_fz0
fz1 = fz3 + fz_parameters_idx1
fz2 = fz1 + fz_parameters_idx2
next_fz0 = fz2 + fz_parameters_idx3
fz3 = next_fz0 + fz_parameters_idx4
aux_pointer += 4
fz_parameters = np.array([fz0, fz1, fz2, fz3])
lv_FS.append(fz_parameters)
elif fuzzy_type == fs.FUZZY_SETS.t2: # pragma: no branch - callers restrict decoding to T1/T2
if lx == 0:
fz_parameters_idx0 = x[parameter_pointer]
fz_parameters_idx1 = x[parameter_pointer + 1]
fz_parameters_idx2 = x[parameter_pointer + 2]
fz_parameters_idx3 = x[parameter_pointer + 3]
fz_parameters_idx4 = x[parameter_pointer + 4]
fz_parameters_idx5 = x[parameter_pointer + 5]
l_fz0 = fz_parameters_idx0
l_fz1 = l_fz0
l_fz2 = l_fz1 + fz_parameters_idx1
next_ufz0 = l_fz2 + fz_parameters_idx2
next_lfz0 = next_ufz0 + fz_parameters_idx3
l_fz3 = next_lfz0 + fz_parameters_idx4
u_fz0 = l_fz0
u_fz1 = u_fz0
u_fz2 = l_fz2
u_fz3 = l_fz3 + fz_parameters_idx5
l_fz_parameters = np.array([l_fz0, l_fz1, l_fz2, l_fz3])
u_fz_parameters = np.array([u_fz0, u_fz1, u_fz2, u_fz3])
next_init = l_fz2 + fz_parameters_idx4
aux_pointer += 6
elif lx == self.n_lv_possible[fuzzy_variable] - 1:
fz_parameters_idx0 = x[parameter_pointer]
fz_parameters_idx1 = x[parameter_pointer + 1]
u_fz0 = next_ufz0
l_fz0 = next_lfz0
u_fz1 = u_fz3 + fz_parameters_idx0
l_fz1 = u_fz1
u_fz2 = l_fz1 + fz_parameters_idx1
l_fz2 = u_fz2
l_fz3 = l_fz2
u_fz3 = l_fz3
l_fz_parameters = np.array([l_fz0, l_fz1, l_fz2, l_fz3])
u_fz_parameters = np.array([u_fz0, u_fz1, u_fz2, u_fz3])
aux_pointer += 2
else:
fz_parameters_idx0 = x[parameter_pointer]
fz_parameters_idx1 = x[parameter_pointer + 1]
fz_parameters_idx2 = x[parameter_pointer + 2]
fz_parameters_idx3 = x[parameter_pointer + 3]
fz_parameters_idx4 = x[parameter_pointer + 4]
fz_parameters_idx5 = x[parameter_pointer + 5]
u_fz0 = next_ufz0
l_fz0 = next_lfz0
l_fz1 = u_fz3 + fz_parameters_idx0
u_fz1 = l_fz1
l_fz2 = l_fz1 + fz_parameters_idx1
u_fz2 = l_fz2
next_ufz0 = l_fz2 + fz_parameters_idx2
next_lfz0 = next_ufz0 + fz_parameters_idx3
l_fz3 = next_lfz0 + fz_parameters_idx4
u_fz3 = l_fz3 + fz_parameters_idx5
l_fz_parameters = np.array([l_fz0, l_fz1, l_fz2, l_fz3])
u_fz_parameters = np.array([u_fz0, u_fz1, u_fz2, u_fz3])
aux_pointer += 6
lv_FS.append((l_fz_parameters, u_fz_parameters))
# Build fuzzy variable from the decoded parameters
if self.categorical_boolean_mask is not None and self.categorical_boolean_mask[fuzzy_variable]:
linguistic_variable = self.categorical_variables[fuzzy_variable]
else:
for lx, relevant_lv in enumerate(lv_FS):
if fuzzy_type == fs.FUZZY_SETS.t1:
proper_FS = fs.FS(self.vl_names[fuzzy_variable][lx], relevant_lv, None)
elif fuzzy_type == fs.FUZZY_SETS.t2: # pragma: no branch - lv_FS contains only decoded T1/T2 sets
proper_FS = fs.IVFS(self.vl_names[fuzzy_variable][lx], relevant_lv[0], relevant_lv[1], None)
linguistic_variables.append(proper_FS)
linguistic_variable = fs.fuzzyVariable(self.var_names[fuzzy_variable], linguistic_variables)
antecedents.append(linguistic_variable)
return antecedents
[docs]
def encode_rulebase(self, rule_base: rules.MasterRuleBase, optimize_lv: bool) -> np.array:
"""
Given a rule base, constructs the corresponding gene associated with that rule base.
GENE STRUCTURE
First: antecedents chosen by each rule. Size: nAnts * nRules (index of the antecedent)
Second: Variable linguistics used. Size: nAnts * nRules
Third: Parameters for the fuzzy partitions of the chosen variables. Size: nAnts * self.n_linguistic_variables * 8|4 (2 trapezoidal memberships if t2)
Four: Consequent classes. Size: nRules
Args:
rule_base: rule base object.
optimize_lv: must be False: only genes over fixed linguistic variables can be encoded.
Returns:
np array of size self.single_gen_size.
Raises:
NotImplementedError: if optimize_lv is True.
ValueError: if the problem does not have one antecedent slot per feature.
"""
if optimize_lv:
raise NotImplementedError('Encoding optimized membership functions is not supported. '
'Encode rule bases over fixed linguistic variables instead.')
if self.nAnts != self.X.shape[1]:
raise ValueError('Encoding a rule base needs one antecedent slot per feature (nAnts equal to the number of features).')
gene = np.zeros((self.single_gen_size,))
rule_consequents = rule_base.get_consequents()
nreal_rules = len(rule_consequents)
# Pointer to the fourth section of the gene: consequents
fourth_pointer = 2 * self.nAnts * self.nRules
# Pointer to the fifth section of the gene: weights, which the decoder divides by 100
fifth_pointer = fourth_pointer + self.nRules
if self.ds_mode == 2:
for ix, rule in enumerate(rule_base.get_rules()):
gene[fifth_pointer + ix] = min(round(getattr(rule, 'weight', 1.0) * 100), 99)
# First and second sections of the gene: antecedents and linguistic variables
for i0, rule in enumerate(rule_base.get_rules()): # Reconstruct the rules
first_pointer = i0 * self.nAnts
second_pointer = (self.nRules * self.nAnts) + i0 * self.nAnts
for ax, linguistic_variable in enumerate(rule.antecedents):
gene[first_pointer + ax] = ax
gene[second_pointer + ax] = linguistic_variable
# Update the fourth section of the gene: consequents using the fourth pointer
gene[fourth_pointer + i0] = rule_consequents[i0]
# Fill the rest of the rules with don't care values and no consequent
for vx in range(nreal_rules, self.nRules):
first_pointer = vx * self.nAnts
second_pointer = (self.nRules * self.nAnts) + vx * self.nAnts
for ax in range(self.nAnts):
gene[first_pointer + ax] = ax
gene[second_pointer + ax] = -1
gene[fourth_pointer + vx] = -1
return np.array(list(map(int, gene)))
def _consequent_pointer(self, fuzzy_type: fs.FUZZY_SETS) -> int:
"""
Returns the gene offset at which the consequent classes start.
Args:
fuzzy_type: a enum type. Check fuzzy_sets for complete specification.
Returns:
int. Index of the first consequent gene.
"""
if self.lvs is not None:
# If no memberships are optimized.
return 2 * self.nAnts * self.nRules
# If memberships are optimized.
mf_size = 4 if fuzzy_type == fs.FUZZY_SETS.t1 else 6
if fuzzy_type == fs.FUZZY_SETS.t1:
return 2 * self.nAnts * self.nRules + \
len(self.n_lv_possible) * 3 + sum(np.array(self.n_lv_possible)-1) * 4 # 4 is the size of the membership function, 3 is the size of the first (and last) membership function
else: # Only Type-1 and Type-2 partitions are optimized.
return 2 * self.nAnts * self.nRules + \
len(self.n_lv_possible) * 2 + sum(np.array(self.n_lv_possible)-1) * mf_size
def _decode_antecedents(self, x: np.array, fuzzy_type: fs.FUZZY_SETS, **kwargs) -> list:
"""
Decodes the fuzzy variables a subject uses as antecedents.
Shared by the rule-object decoder and the array evaluation path so that
partition normalization has a single implementation.
Args:
x: integer gene of a rulebase.
fuzzy_type: enum type, see fuzzy_sets.
Returns:
list of fuzzy variables.
"""
if self.lvs is None:
min_domain, max_domain, range_domain = self._normalization_domain()
antecedents_raw = self._decode_membership_functions(x, fuzzy_type)
# Normalize the membership functions to the data domain
antecedents = []
for fuzzy_variable, fv_raw in enumerate(antecedents_raw):
if self.categorical_boolean_mask is not None and self.categorical_boolean_mask[fuzzy_variable]:
antecedents.append(fv_raw)
else:
# Extract raw parameters and normalize based on fuzzy type
if fuzzy_type == fs.FUZZY_SETS.t1:
lv_FS = [lv.membership_parameters for lv in fv_raw.linguistic_variables]
min_lv = np.min(np.array(lv_FS))
max_lv = np.max(np.array(lv_FS))
linguistic_variables = []
for lx, relevant_lv in enumerate(lv_FS):
relevant_lv = np.array(relevant_lv)
relevant_lv = (relevant_lv - min_lv) / (max_lv - min_lv) * range_domain[fuzzy_variable] + min_domain[fuzzy_variable]
proper_FS = fs.FS(self.vl_names[fuzzy_variable][lx], relevant_lv.tolist(), (min_domain[fuzzy_variable], max_domain[fuzzy_variable]))
linguistic_variables.append(proper_FS)
else:
# For T2/IVFS, extract both lower and upper parameters
lv_lower = [lv.secondMF_lower for lv in fv_raw.linguistic_variables]
lv_upper = [lv.secondMF_upper for lv in fv_raw.linguistic_variables]
all_params = lv_lower + lv_upper
min_lv = np.min(np.array(all_params))
max_lv = np.max(np.array(all_params))
linguistic_variables = []
for lx in range(len(lv_lower)):
lower = np.array(lv_lower[lx])
upper = np.array(lv_upper[lx])
lower = (lower - min_lv) / (max_lv - min_lv) * range_domain[fuzzy_variable] + min_domain[fuzzy_variable]
upper = (upper - min_lv) / (max_lv - min_lv) * range_domain[fuzzy_variable] + min_domain[fuzzy_variable]
proper_FS = fs.IVFS(self.vl_names[fuzzy_variable][lx], lower.tolist(), upper.tolist(), (min_domain[fuzzy_variable], max_domain[fuzzy_variable]))
linguistic_variables.append(proper_FS)
linguistic_variable = fs.fuzzyVariable(self.var_names[fuzzy_variable], linguistic_variables)
antecedents.append(linguistic_variable)
else:
try:
antecedents = self.lvs[kwargs['time_moment']]
except (KeyError, TypeError, IndexError):
antecedents = self.lvs
return antecedents
def _construct_ruleBase(self, x: np.array, fuzzy_type: fs.FUZZY_SETS, **kwargs) -> rules.MasterRuleBase:
"""
Given a subject, it creates a rulebase according to its specification.
kwargs:
- time_moment: if temporal fuzzy sets are used with different partitions for each time interval,
then this parameter is used to specify which time moment is being used.
Args:
x: gen of a rulebase. type: dict.
fuzzy_type: a enum type. Check fuzzy_sets for complete specification (two fields, t1 and t2, to mark which fs you want to use)
kwargs: additional parameters to pass to the rule
Returns:
a rulebase object.
"""
rule_list = [[] for _ in range(self.n_classes)]
'''
GEN STRUCTURE
First: features chosen by each rule. Size: nAnts * nRules
Second: linguistic labels used. Size: nAnts * nRules
Third: Parameters for the fuzzy partitions of the chosen variables. Size: X.shape[1] * ((self.n_linguistic_variables-1) * mf_size + 2)
Four: Consequent classes. Size: nRules
Five: Weights for each rule. Size: nRules (only if ds_mode == 2)
Sixth: Modifiers for the membership functions. Size: len(self.lvs) * nAnts * nRules
'''
fourth_pointer = self._consequent_pointer(fuzzy_type)
if self.ds_mode == 2:
fifth_pointer = fourth_pointer + self.nRules
else:
fifth_pointer = fourth_pointer
if self.ds_mode == 2:
sixth_pointer = fifth_pointer + self.nRules
else:
sixth_pointer = fifth_pointer
aux_pointer = 0
# Integer sampling doesnt work fine in pymoo, so we do this (which is btw what pymoo is really doing if you just set integer optimization)
try:
# subject might come as a dict.
x = np.array(list(x.values())).astype(int)
except AttributeError:
x = x.astype(int)
for i0 in range(self.nRules): # Reconstruct the rules
first_pointer = i0 * self.nAnts
chosen_ants = x[first_pointer:first_pointer + self.nAnts]
second_pointer = (i0 * self.nAnts) + (self.nAnts * self.nRules)
# Shape: self.nAnts + self.n_lv_possible + 1
antecedent_parameters = x[second_pointer:second_pointer+self.nAnts]
init_rule_antecedents = np.zeros(
(self.X.shape[1],)) - 1 # -1 is dont care
for jx, ant in enumerate(chosen_ants):
if self.lvs is not None:
antecedent_parameters[jx] = min(antecedent_parameters[jx], len(self.lvs[ant]) - 1)
else:
antecedent_parameters[jx] = min(antecedent_parameters[jx], self.n_lv_possible[ant] - 1)
init_rule_antecedents[ant] = antecedent_parameters[jx]
consequent_idx = x[fourth_pointer + aux_pointer]
assert consequent_idx < self.n_classes, "Consequent class is not valid. Something in the gene is wrong."
aux_pointer += 1
if self.ds_mode == 2:
rule_weight = x[fifth_pointer + i0] / 100
else:
rule_weight = 1.0
if consequent_idx != -1 and np.any(init_rule_antecedents != -1):
rs_instance = rules.RuleSimple(init_rule_antecedents, 0, None)
if self.ds_mode == 1 or self.ds_mode == 2:
rs_instance.weight = rule_weight
rule_list[consequent_idx].append(
rs_instance)
antecedents = self._decode_antecedents(x, fuzzy_type, **kwargs)
for i in range(self.n_classes):
if fuzzy_type == fs.FUZZY_SETS.temporal:
fuzzy_type = self.lvs[0][0].inside_type()
if fuzzy_type == fs.FUZZY_SETS.t1:
rule_base = rules.RuleBaseT1(antecedents, rule_list[i])
elif fuzzy_type == fs.FUZZY_SETS.t2:
rule_base = rules.RuleBaseT2(antecedents, rule_list[i])
elif fuzzy_type == fs.FUZZY_SETS.gt2: # pragma: no branch - temporal types are unwrapped above
rule_base = rules.RuleBaseGT2(antecedents, rule_list[i])
if i == 0:
res = rules.MasterRuleBase([rule_base], self.classes_names, ds_mode=self.ds_mode, allow_unknown=self.allow_unknown)
else:
res.add_rule_base(rule_base)
res.rename_cons(self.classes_names)
return res
def _evaluate_slow(self, x: np.array, out: dict, *args, **kwargs):
"""
Args:
x: array of train samples. x shape = features
those features are the parameters to optimize.
out: dict where the F field is the fitness. It is used from the outside.
"""
ruleBase = self._construct_ruleBase(x, self.fuzzy_type)
if len(ruleBase.get_rules()) > 0:
if getattr(self.fitness_func, '__func__', None) is not FitRuleBase.fitness_func:
evaluator = evr.evalRuleBase(
ruleBase, self.X, self.y, precomputed_truth=self._precomputed_truth)
evaluator.add_rule_weights()
score = self.fitness_func(ruleBase, self.X, self.y, self.tolerance, self.alpha_, self.beta_, self._precomputed_truth)
else:
score = 0.0
out["F"] = 1 - score
#: Set to False to force the object decoder, for benchmarks and parity tests.
array_evaluation = True
#: Device types on which populations may be scored by the exact PyTorch
#: objective. Scoring CPU tensors gains nothing over the NumPy routes, so
#: only CUDA is enabled; parity tests add 'cpu' to run the same code.
torch_devices = ('cuda',)
def _standard_population_objective(self, *args, **kwargs) -> bool:
"""
Whether whole populations have an exact batched Type-1 objective.
True for the unmodified built-in objective with ``ds_mode`` 0 or 1,
numeric labels and no external runner, with fixed or optimized
partitions alike. The individual routes add their own requirements.
"""
if type(self) is not FitRuleBase or not self.array_evaluation or args or kwargs:
return False
if (getattr(self._evaluate, '__func__', None) is not _BATCH_SCALAR_EVALUATE
or getattr(self._array_score, '__func__', None) is not _BATCH_ARRAY_SCORE):
return False
standard_loss = getattr(self.fitness_func, '__func__', None) is FitRuleBase.fitness_func
return (standard_loss and np.asarray(self.y).dtype.kind in 'biuf'
and self.fuzzy_type == fs.FUZZY_SETS.t1
and self.ds_mode in (0, 1)
and not self._external_elementwise_runner)
def _can_batch_population(self, *args, **kwargs) -> bool:
"""Whether this fit context can use the private C01 evaluator."""
return (self.lvs is not None and hasattr(self, '_fitness_cache')
and self._standard_population_objective(*args, **kwargs))
def _population_scores(self, X: np.ndarray, *args, **kwargs) -> Optional[np.ndarray]:
"""Score an eligible small fixed-partition population, or return None."""
if not self._can_batch_population(*args, **kwargs):
return None
packed = self._packed_memberships()
if packed is None:
return None
popfit = _population_module()
values = np.asarray(X)
if values.ndim != 2 or values.dtype.kind not in 'biuf':
return None
integer_values = values.astype(int)
term_counts = np.asarray([len(lv) for lv in self.lvs])
return popfit.score_population(
integer_values, packed, np.asarray(self.y), self.nRules, self.nAnts,
self.X.shape[1], self.n_classes, term_counts,
self._consequent_pointer(self.fuzzy_type), self.ds_mode,
self.allow_unknown, self.tolerance, self.alpha_, self.beta_,
self._label_domain(), getattr(self, '_firing_cache', None))
def _population_chunks(self, population: int):
"""
Split a population into chunks whose intermediates fit the budget.
Candidates are scored independently, so chunking cannot change a
result. A trailing single candidate is merged into the previous chunk
because the batched route needs at least two; that overshoots the
budget by one candidate at most.
"""
popfit = _population_module()
chunk = popfit.chunk_size(len(self.X), self.nRules, self.X.shape[1])
if chunk < 2 or population < 2:
return None
bounds = list(range(0, population, chunk))
if len(bounds) > 1 and population - bounds[-1] == 1:
bounds.pop()
return [slice(start, min(start + chunk, population))
for start in bounds[:-1]] + [slice(bounds[-1], population)]
def _batched_scores(self, genes: np.ndarray, *args, **kwargs):
"""Score candidates through the batched route, chunked, or None."""
chunks = self._population_chunks(len(genes))
if chunks is None:
return None
parts = []
for piece in chunks:
scored = self._population_scores(genes[piece], *args, **kwargs)
if scored is None:
return None
parts.append(scored)
return parts[0] if len(parts) == 1 else np.concatenate(parts)
def _scalar_scores(self, genes: np.ndarray) -> np.ndarray:
"""Score candidates one at a time, exactly as the scalar route does."""
scored = np.empty(len(genes))
target = {}
for index, gene in enumerate(genes):
self._evaluate(gene, target)
scored[index] = target['F']
return scored
def _evaluate_elementwise(self, X, out, *args, **kwargs):
"""Use population scoring when it measures faster, else stay scalar."""
values = np.asarray(X)
cache = getattr(self, '_fitness_cache', None)
if (cache is None or values.ndim != 2
or not self._can_batch_population(*args, **kwargs)
or self._population_chunks(len(values)) is None):
return super()._evaluate_elementwise(X, out, *args, **kwargs)
fitness = self._cached_population(
values, cache,
lambda fresh: self._score_fresh(fresh, len(values), *args, **kwargs))
if fitness is None:
return super()._evaluate_elementwise(X, out, *args, **kwargs)
out['F'] = fitness
def _cached_population(self, values: np.ndarray, cache, score: Callable) -> Optional[np.ndarray]:
"""
Fitness of a population, scoring each uncached genotype only once.
Genotypes in the fitness cache are not rescored, and a genotype repeated
within the population is scored at its first occurrence and copied to
the others. ``score`` receives the remaining genotypes in population
order; if it returns None nothing is cached and None is returned.
"""
fitness = np.empty(len(values))
keys, fresh, copies, first = [], [], [], {}
for index, gene in enumerate(values):
key = cache.key(gene)
keys.append(key)
cached = cache.get(key)
if cached is not None:
fitness[index] = cached
elif key is not None and key in first:
copies.append((index, first[key]))
else:
if key is not None:
first[key] = index
fresh.append(index)
# Calling a scorer with an empty population would decline the batched
# route's ``population >= 2`` contract and needlessly send the whole
# population through scalar evaluation again.
if fresh:
scored = score(values[fresh])
if scored is None:
return None
fitness[fresh] = scored
for index, value in zip(fresh, scored):
cache.put(keys[index], value)
for index, source in copies:
fitness[index] = fitness[source]
return fitness
def _score_fresh(self, fresh: np.ndarray, population: int, *args, **kwargs):
"""
Score uncached candidates on the chosen route, or None to fall back.
While the probe is undecided each generation runs wholly on the route it
nominates, and its cost per candidate is recorded. ``_scalar_scores``
fills the fitness cache itself, so its results are re-cached harmlessly
by the caller with identical values.
"""
popfit = _population_module()
probe = self._route_probe_state(population)
route = probe.route(len(fresh)) if probe is not None else None
if route is None:
if probe is not None and probe.decision == popfit._RouteProbe.SCALAR:
return self._scalar_scores(fresh)
batched = self._batched_scores(fresh, *args, **kwargs)
return None if batched is None else 1 - batched
start = time.perf_counter()
if route == popfit._RouteProbe.SCALAR:
scored = self._scalar_scores(fresh)
else:
batched = self._batched_scores(fresh, *args, **kwargs)
if batched is None:
return None
scored = 1 - batched
probe.record(route, time.perf_counter() - start, len(fresh))
return scored
def _route_probe_state(self, population: int):
"""
The fit-local route probe, created on first use, or None.
A stored calibration profile, if the user has run the offline campaign
on this machine and it covers this workload, settles the probe before it
measures anything. Both routes give identical results either way, so
this only decides which one runs.
"""
if not hasattr(self, '_route_probe'):
return None
if self._route_probe is None:
popfit = _population_module()
probe = popfit._RouteProbe()
from . import _dispatch_profile
stored = _dispatch_profile.decision_for(
len(self.X), self.nRules, self.X.shape[1], population)
if stored is not None:
probe.decision = (popfit._RouteProbe.BATCH if stored == 'batch'
else popfit._RouteProbe.SCALAR)
self._route_probe = probe
return self._route_probe
def _evaluate_gene_population(self, genes: np.ndarray, device=None) -> tuple:
"""
Fitness of a whole population, for backends that evaluate populations.
EvoX hands over a generation at once. Inside a fit's cache scope, cached
and repeated genotypes are scored once and the rest go to the fastest
exact route: scalar, batched or, on ``device``, the PyTorch objective.
Outside that scope each candidate is evaluated in turn. Every value is
what ``_evaluate`` computes for that chromosome.
Args:
genes: integer chromosomes, one per row.
device: the backend's torch device, or None.
Returns:
``(fitness, on_device)``, where ``on_device`` tells whether the
device objective scored any candidate.
"""
values = np.asarray(genes)
cache = getattr(self, '_fitness_cache', None)
if cache is None or values.ndim != 2 or values.dtype.kind not in 'biu':
return self._scalar_scores(values), False
on_device = False
def score(fresh):
nonlocal on_device
fitness, used = self._score_on_best_route(fresh, len(values), device)
on_device = on_device or used
return fitness
return self._cached_population(values, cache, score), on_device
def _score_on_best_route(self, fresh: np.ndarray, population: int, device=None) -> tuple:
"""
Score uncached candidates on the CPU or on a verified device.
The device objective is trusted only once it has reproduced the CPU
scores of a sample of candidates. The first generation large enough is
scored on the device while that sample is also scored on the CPU; a
single difference gives the generation the CPU's scores and keeps the
fit on the CPU for good. Once trusted, which route runs is a speed
question settled like the scalar/batched choice.
Returns:
``(fitness, on_device)``.
"""
route = self._device_route_state(device)
if route is None or route.rejected:
return self._score_fresh_on_cpu(fresh, population), False
if not route.verified:
if len(fresh) < route.MIN_CANDIDATES:
return self._score_fresh_on_cpu(fresh, population), False
sample = fresh[:route.VERIFY_CANDIDATES]
# The scalar route keeps this short sample out of the scalar/batched
# probe; every CPU route computes identical values.
start = time.perf_counter()
expected = self._scalar_scores(sample)
cpu_seconds = (time.perf_counter() - start) / len(sample)
start = time.perf_counter()
fitness = self._device_fitness(route.objective, fresh)
device_seconds = (time.perf_counter() - start) / len(fresh)
if route.verify(expected, fitness[:len(sample)], cpu_seconds, device_seconds):
return fitness, True
fitness[:len(sample)] = expected
if len(fresh) > len(sample):
fitness[len(sample):] = self._score_fresh_on_cpu(fresh[len(sample):], population)
return fitness, False
choice = route.probe.route(len(fresh))
use_device = (route.probe.decision if choice is None else choice) == route.DEVICE
start = time.perf_counter()
if use_device:
fitness = self._device_fitness(route.objective, fresh)
else:
fitness = self._score_fresh_on_cpu(fresh, population)
if choice is not None:
route.probe.record(choice, time.perf_counter() - start, len(fresh))
return fitness, use_device
def _score_fresh_on_cpu(self, fresh: np.ndarray, population: int) -> np.ndarray:
"""Score candidates on the fastest exact CPU route."""
if self._can_batch_population() and self._population_chunks(population) is not None:
scored = self._score_fresh(fresh, population)
if scored is not None:
return scored
return self._scalar_scores(fresh)
def _device_fitness(self, objective, genes: np.ndarray) -> np.ndarray:
"""Device objective fitness, with declined candidates scored on the CPU."""
scores, declined = objective.score(genes)
fitness = 1 - scores
if np.any(declined):
fitness[declined] = self._scalar_scores(genes[declined])
return fitness
def _device_route_state(self, device):
"""
The fit-local device route, created on first use, or None.
Like the route probe it exists only inside a fit's cache scope, and only
for device types in ``torch_devices`` and problems the exact PyTorch
objective can represent.
"""
if device is None or not hasattr(self, '_torch_route'):
return None
if self._torch_route is None:
self._torch_route = False
if self._standard_population_objective():
try:
import torch
torchfit = _torch_module()
except ImportError:
return None
if torch.device(device).type in self.torch_devices:
objective = torchfit.TorchObjective.build(self, device)
if objective is not None:
self._torch_route = torchfit.DeviceRoute(objective)
return self._torch_route or None
def _label_domain(self):
"""
Returns the fit-local integer label layout used by the MCC, or None.
The training labels of a problem do not change during its optimization,
exactly like its precomputed memberships.
"""
if not hasattr(self, '_label_domain_cache'):
from ._fitness import _LabelDomain
self._label_domain_cache = _LabelDomain.build(self.y, self.n_classes)
return self._label_domain_cache
def _packed_memberships(self) -> Optional[tuple]:
"""
Returns the packed gather table for fixed partitions, or None.
Precomputed partitions do not change during a problem's optimization, so
the table is built once instead of once per candidate. Optimized
partitions get None: their memberships differ per candidate.
"""
if not hasattr(self, '_packed_memberships_cache'):
table = None
if self.lvs is not None and self._precomputed_truth is not None:
table = rules.pack_membership_table(
self._precomputed_truth, len(self.X),
(2,) if self.fuzzy_type == fs.FUZZY_SETS.t2 else ())
self._packed_memberships_cache = table
return self._packed_memberships_cache
def _array_score(self, x: np.array, **kwargs) -> Optional[float]:
"""
Score a chromosome without building rule objects, or None if unsupported.
Returning None means the candidate leaves the supported case and the
ordinary object decoder must evaluate it instead.
"""
if not self.array_evaluation or type(self) is not FitRuleBase or kwargs:
return None
if self.X.shape[1] == 0 or self.fuzzy_type not in (fs.FUZZY_SETS.t1, fs.FUZZY_SETS.t2):
return None
from . import _array_fitness as arrfit
if self.lvs is None:
term_counts = np.asarray(self.n_lv_possible)
elif isinstance(self.lvs, (list, tuple)):
term_counts = np.asarray([len(lv) for lv in self.lvs])
else:
return None # Temporal partitions keep the object decoder.
try:
x = np.array(list(x.values())).astype(int)
except AttributeError:
x = x.astype(int)
decoded = arrfit.decode_rule_arrays(
x, self.nRules, self.nAnts, self.X.shape[1], self.n_classes,
term_counts, self.ds_mode, self._consequent_pointer(self.fuzzy_type))
if decoded is None:
return None
packed = None
if self.lvs is None:
antecedents = self._decode_antecedents(x, self.fuzzy_type)
tail = (2,) if self.fuzzy_type == fs.FUZZY_SETS.t2 else ()
packed = rules.pack_membership_table_from_variables(
antecedents, self.X, tail)
truth = (self._precomputed_truth if packed is not None
else rules.compute_antecedents_memberships(antecedents, self.X))
else:
truth = self._precomputed_truth
packed = self._packed_memberships()
if truth is None and packed is None:
return None
return arrfit.score_candidate(
decoded, truth, self.X, self.y, self.n_classes, self.ds_mode,
self.allow_unknown, self.tolerance, self.alpha_, self.beta_,
self.fuzzy_type == fs.FUZZY_SETS.t2, self._label_domain(),
getattr(self, '_firing_cache', None), packed)
def _evaluate(self, x: np.array, out: dict, *args, **kwargs):
"""Use reusable T1/T2 fitness primitives for the built-in objective only."""
standard_loss = getattr(self.fitness_func, '__func__', None) is FitRuleBase.fitness_func
numeric_labels = np.asarray(self.y).dtype.kind in 'biuf'
if standard_loss and numeric_labels and self.fuzzy_type in (fs.FUZZY_SETS.t1, fs.FUZZY_SETS.t2):
from ._fitness import score_rulebase
cache = getattr(self, '_fitness_cache', None)
key = cache.key(x) if cache is not None else None
cached = cache.get(key) if cache is not None else None
if cached is not None:
out['F'] = cached
return
score = self._array_score(x, **kwargs)
if score is None:
rulebase = self._construct_ruleBase(x, self.fuzzy_type)
score = score_rulebase(
rulebase, self.X, self.y, self.tolerance,
self.alpha_, self.beta_, self._precomputed_truth)
out['F'] = 1 - score
if cache is not None:
cache.put(key, out['F'])
else:
self._evaluate_slow(x, out, *args, **kwargs)
[docs]
def fitness_func(self, ruleBase: rules.RuleBase, X:np.array, y:np.array, tolerance:float, alpha:float=0.0, beta:float=0.0, precomputed_truth:np.array=None) -> float:
"""
Fitness function for the optimization problem.
Args:
ruleBase: RuleBase object
X: array of train samples. X shape = (n_samples, n_features)
y: array of train labels. y shape = (n_samples,)
tolerance: float. Tolerance for the size evaluation.
alpha: float. Weight for the accuracy term.
beta: float. Weight for the average rule size term.
precomputed_truth: np array. If given, it will be used as the truth values for the evaluation.
Returns:
float. Fitness value.
"""
if precomputed_truth is None:
precomputed_truth = rules.compute_antecedents_memberships(ruleBase.antecedents, X)
ev_object = evr.evalRuleBase(ruleBase, X, y, precomputed_truth=precomputed_truth)
ev_object.add_full_evaluation()
ruleBase.purge_rules(tolerance)
if len(ruleBase.get_rules()) > 0:
score_acc = ev_object.classification_eval()
score_rules_size = ev_object.size_antecedents_eval(tolerance)
score_nrules = ev_object.effective_rulesize_eval(tolerance)
score = score_acc + score_rules_size * alpha + score_nrules * beta
else:
score = 0.0
return score
def _population_module():
"""Import the private population evaluator lazily."""
from . import _population_fitness as popfit
return popfit
def _torch_module():
"""Import the private PyTorch objective lazily."""
from . import _torch_fitness as torchfit
return torchfit
# Population evaluation declines when either scalar oracle is monkeypatched.
# Besides keeping instrumentation meaningful, this makes benchmarks and parity
# tests able to select the old path without a public configuration flag.
_BATCH_SCALAR_EVALUATE = FitRuleBase._evaluate
_BATCH_ARRAY_SCORE = FitRuleBase._array_score