Modules

Top-level package for Action Rules.

ActionRules

Generate action rules from tabular data using one-hot encoding and bitset support counting.

Attributes:

Name	Type	Description
min_stable_attributes	int	The minimum number of stable attributes required.
min_flexible_attributes	int	The minimum number of flexible attributes required.
min_undesired_support	int	The minimum support for the undesired state.
min_undesired_confidence	float	The minimum confidence for the undesired state.
min_desired_support	int	The minimum support for the desired state.
min_desired_confidence	float	The minimum confidence for the desired state.
verbose	(bool, optional)	If True, enables verbose output.
rules	(Optional[Rules], optional)	Stores the generated rules.
output	(Optional[Output], optional)	Stores the generated action rules.
np	(Optional[ModuleType], optional)	The numpy or cupy module used for array operations.
pd	(Optional[ModuleType], optional)	The pandas or cudf module used for DataFrame operations.
is_gpu_np	bool	Indicates whether GPU-accelerated numpy (cupy) is used.
is_gpu_pd	bool	Indicates whether GPU-accelerated pandas (cudf) is used.
intrinsic_utility_table	(dict, optional)	(attribute, value) -> float A lookup table for the intrinsic utility of each attribute-value pair. If None, no intrinsic utility is considered.
transition_utility_table	(dict, optional)	(attribute, from_value, to_value) -> float A lookup table for cost/gain of transitions between values. If None, no transition utility is considered.

Methods:

Name	Description
fit	Generates action rules based on the provided dataset and parameters.
get_bindings	Binds attributes to corresponding columns in the dataset.
get_stop_list	Generates a stop list to prevent certain combinations of attributes.
get_rules	Returns the generated action rules if available.
predict	Predicts recommended actions based on the provided row of data.

Source code in src/action_rules/action_rules.py

class ActionRules:
    """
    Generate action rules from tabular data using one-hot encoding and bitset support counting.

    Attributes
    ----------
    min_stable_attributes : int
        The minimum number of stable attributes required.
    min_flexible_attributes : int
        The minimum number of flexible attributes required.
    min_undesired_support : int
        The minimum support for the undesired state.
    min_undesired_confidence : float
        The minimum confidence for the undesired state.
    min_desired_support : int
        The minimum support for the desired state.
    min_desired_confidence : float
        The minimum confidence for the desired state.
    verbose : bool, optional
        If True, enables verbose output.
    rules : Optional[Rules], optional
        Stores the generated rules.
    output : Optional[Output], optional
        Stores the generated action rules.
    np : Optional[ModuleType], optional
        The numpy or cupy module used for array operations.
    pd : Optional[ModuleType], optional
        The pandas or cudf module used for DataFrame operations.
    is_gpu_np : bool
        Indicates whether GPU-accelerated numpy (cupy) is used.
    is_gpu_pd : bool
        Indicates whether GPU-accelerated pandas (cudf) is used.
    intrinsic_utility_table : dict, optional
        (attribute, value) -> float
        A lookup table for the intrinsic utility of each attribute-value pair.
        If None, no intrinsic utility is considered.
    transition_utility_table : dict, optional
        (attribute, from_value, to_value) -> float
        A lookup table for cost/gain of transitions between values.
        If None, no transition utility is considered.

    Methods
    -------
    fit(data, stable_attributes, flexible_attributes, target, undesired_state, desired_state, use_gpu=False)
        Generates action rules based on the provided dataset and parameters.
    get_bindings(data, stable_attributes, flexible_attributes, target)
        Binds attributes to corresponding columns in the dataset.
    get_stop_list(stable_items_binding, flexible_items_binding)
        Generates a stop list to prevent certain combinations of attributes.
    get_rules()
        Returns the generated action rules if available.
    predict(frame_row)
        Predicts recommended actions based on the provided row of data.
    """

    def __init__(
        self,
        min_stable_attributes: int,
        min_flexible_attributes: int,
        min_undesired_support: int,
        min_undesired_confidence: float,
        min_desired_support: int,
        min_desired_confidence: float,
        verbose=False,
        intrinsic_utility_table: Optional[dict] = None,
        transition_utility_table: Optional[dict] = None,
    ):
        """
        Initialize the ActionRules class with the specified parameters.

        Parameters
        ----------
        min_stable_attributes : int
            The minimum number of stable attributes required.
        min_flexible_attributes : int
            The minimum number of flexible attributes required.
        min_undesired_support : int
            The minimum support for the undesired state.
        min_undesired_confidence : float
            The minimum confidence for the undesired state.
        min_desired_support : int
            The minimum support for the desired state.
        min_desired_confidence : float
            The minimum confidence for the desired state.
        verbose : bool, optional
            If True, enables verbose output. Default is False.
        intrinsic_utility_table : dict, optional
            (attribute, value) -> float
            A lookup table for the intrinsic utility of each attribute-value pair.
            If None, no intrinsic utility is considered.
        transition_utility_table : dict, optional
            (attribute, from_value, to_value) -> float
            A lookup table for cost/gain of transitions between values.
            If None, no transition utility is considered.

        Notes
        -----
        The `verbose` parameter can be used to enable detailed output during the rule generation process.
        """
        self.min_stable_attributes = min_stable_attributes
        self.min_flexible_attributes = min_flexible_attributes
        self.min_undesired_support = min_undesired_support
        self.min_desired_support = min_desired_support
        self.min_undesired_confidence = min_undesired_confidence
        self.min_desired_confidence = min_desired_confidence
        self.verbose = verbose
        self.rules = None  # type: Optional[Rules]
        self.output = None  # type: Optional[Output]
        self.np = None  # type: Optional[ModuleType]
        self.pd = None  # type: Optional[ModuleType]
        self.is_gpu_np = False
        self.is_gpu_pd = False
        self.is_onehot = False
        self.bit_masks = None  # type: Optional['numpy.ndarray']
        self.target_state_bit_masks = None  # type: Optional[dict]
        self.frames_bit_masks = None  # type: Optional[dict]
        self.intrinsic_utility_table = intrinsic_utility_table or {}
        self.transition_utility_table = transition_utility_table or {}
        self._original_intrinsic_utility_table = {}  # type: dict
        self._original_transition_utility_table = {}  # type: dict
        self._column_values = None  # type: Optional[dict]

    def count_max_nodes(self, stable_items_binding: dict, flexible_items_binding: dict) -> int:
        """
        Calculate the maximum number of nodes based on the given item bindings.

        This function takes two dictionaries, `stable_items_binding` and `flexible_items_binding`,
        which map attributes to lists of items. It calculates the total number of nodes by considering
        all possible combinations of the lengths of these item lists and summing the product of each combination.

        Parameters
        ----------
        stable_items_binding : dict
            A dictionary where keys are attributes and values are lists of stable items.
        flexible_items_binding : dict
            A dictionary where keys are attributes and values are lists of flexible items.

        Returns
        -------
        int
            The total number of nodes calculated by summing the product of lengths of all combinations of item lists.

        Notes
        -----
        - The function first combines the lengths of item lists from both dictionaries.
        - It then calculates the sum of the products of all possible combinations of these lengths.
        """
        import numpy

        values_in_attribute = []
        for items in list(stable_items_binding.values()) + list(flexible_items_binding.values()):
            values_in_attribute.append(len(items))

        sum_nodes = 0
        for i in range(len(values_in_attribute)):
            for comb in itertools.combinations(values_in_attribute, i + 1):
                sum_nodes += int(numpy.prod(comb))
        return sum_nodes

    def set_array_library(self, use_gpu: bool, df: Union['cudf.DataFrame', 'pandas.DataFrame']):
        """
        Set the appropriate array and DataFrame libraries (cuDF or pandas) based on the user's preference.

        Parameters
        ----------
        use_gpu : bool
            Indicates whether to use GPU (cuDF) for data processing if available.
        df : Union[cudf.DataFrame, pandas.DataFrame]
            The DataFrame to convert.

        Raises
        ------
        ImportError
            If `use_gpu` is True but cuDF is not available and pandas cannot be imported as fallback.

        Warnings
        --------
        UserWarning
            If `use_gpu` is True but cuDF is not available, a warning is issued indicating fallback to pandas.

        Notes
        -----
        This method determines whether to use GPU-accelerated libraries for processing data, falling back to CPU-based
        libraries if necessary.
        """
        if use_gpu:
            try:
                import cupy as np

                is_gpu_np = True
            except ImportError:
                warnings.warn("CuPy is not available. Falling back to Numpy.")
                import numpy as np

                is_gpu_np = False
        else:
            import numpy as np

            is_gpu_np = False

        df_library_imported = False
        try:
            import pandas as pd

            if isinstance(df, pd.DataFrame):
                is_gpu_pd = False
                df_library_imported = True
        except ImportError:
            df_library_imported = False

        if not df_library_imported:
            try:
                import cudf as pd

                if isinstance(df, pd.DataFrame):
                    is_gpu_pd = True
                    df_library_imported = True
            except ImportError:
                df_library_imported = False

        if not df_library_imported:
            raise ImportError('Just Pandas or cuDF dataframes are supported.')

        self.np = np
        self.pd = pd
        self.is_gpu_np = is_gpu_np
        self.is_gpu_pd = is_gpu_pd

    def df_to_array(self, df: Union['cudf.DataFrame', 'pandas.DataFrame']) -> tuple:
        """
        Convert a one-hot DataFrame to a binary array.

        Parameters
        ----------
        df : Union[cudf.DataFrame, pandas.DataFrame]
            The DataFrame to convert.

        Returns
        -------
        tuple
            A tuple containing the transposed array and the DataFrame columns.

        Notes
        -----
        The data is converted to an unsigned 8-bit array (`np.uint8`), backed by
        NumPy or CuPy depending on the selected cpu/gpu backend.
        """
        columns = list(df.columns)
        if self.is_gpu_np:
            data = self.np.asarray(df.values, dtype=self.np.uint8).T  # type: ignore
        elif self.is_gpu_pd:
            data = df.to_numpy().T  # type: ignore
        else:
            data = df.to_numpy(dtype=self.np.uint8).T  # type: ignore
        return data, columns

    def build_bit_masks(
        self,
        data: Union['numpy.ndarray', 'cupy.ndarray'],
    ) -> Union['numpy.ndarray', 'cupy.ndarray']:
        """
        Pack a binary feature matrix into 64-bit masks for fast intersection.

        Parameters
        ----------
        data : Union[numpy.ndarray, cupy.ndarray]
            Dense matrix produced by `df_to_array`, shaped (num_attributes, num_transactions)
            and containing 0/1 values.

        Returns
        -------
        Union[numpy.ndarray, cupy.ndarray]
            bit_masks is a uint64 array with shape (num_attributes, num_words)
            holding 64 bits/transactions for each item.

        Notes
        -----
        - The packing uses 64-bit little-endian words (bit 0 corresponds to the
          first transaction in each chunk).
        - Sparse inputs are not supported; callers should densify before packing.
        """
        if self.np is None:
            raise RuntimeError("Array library is not initialised. Call set_array_library first.")
        # Shape is (num_attributes, num_transactions).
        num_attributes, num_transactions = data.shape
        num_words = (num_transactions + 63) // 64
        padded_transactions = num_words * 64
        padding = padded_transactions - num_transactions

        if padding > 0:
            pad_block = self.np.zeros((num_attributes, padding), dtype=data.dtype)
            padded_data = self.np.concatenate((data, pad_block), axis=1)
        else:
            padded_data = data

        # Group transactions into 64-bit chunks: (num_attributes, num_words, 64).
        chunks = padded_data.reshape(num_attributes, num_words, 64).astype(self.np.uint64, copy=False)
        bit_offsets = self.np.arange(64, dtype=self.np.uint64)
        bit_weights = self.np.left_shift(self.np.uint64(1), bit_offsets)

        # Pack each 64-sized transaction chunk into one uint64 word.
        bit_masks = self.np.tensordot(chunks, bit_weights, axes=([2], [0])).astype(self.np.uint64, copy=False)
        return bit_masks

    def _cache_bitset_structures(
        self,
        bit_masks: Union['numpy.ndarray', 'cupy.ndarray'],
        target_items_binding: dict,
        target: str,
    ) -> None:
        """
        Save all column masks; extract target-state rows into a separate dict.

        Parameters
        ----------
        bit_masks : Union[numpy.ndarray, cupy.ndarray]
            Packed transaction masks for every attribute/value.
        target_items_binding : dict
            Mapping from target attribute name to indices of its one-hot columns.
        target : str
            Name of the target attribute.
        """
        target_state_indices = target_items_binding.get(target, [])
        target_state_bit_masks = {index: bit_masks[index] for index in target_state_indices}

        self.bit_masks = bit_masks
        self.target_state_bit_masks = target_state_bit_masks

    def one_hot_encode(
        self,
        data: Union['cudf.DataFrame', 'pandas.DataFrame'],
        stable_attributes: list,
        flexible_attributes: list,
        target: str,
    ) -> Union['cudf.DataFrame', 'pandas.DataFrame']:
        """
        Perform one-hot encoding on the attributes of the DataFrame.

        Parameters
        ----------
        data : Union[cudf.DataFrame, pandas.DataFrame]
            The input DataFrame containing the data to be encoded.
        stable_attributes : list
            List of stable attributes to be one-hot encoded.
        flexible_attributes : list
            List of flexible attributes to be one-hot encoded.
        target : str
            The target attribute to be one-hot encoded.

        Returns
        -------
        Union[cudf.DataFrame, pandas.DataFrame]
            A DataFrame with the specified attributes one-hot encoded.

        Notes
        -----
        Stable and flexible (antecedent) columns are cast to strings only for non-missing values; ``NaN`` is
        preserved so that ``pd.get_dummies`` skips it instead of creating a phantom ``<attr>_<item_*>_nan``
        category.  This implements the *pessimistic* interpretation of null values in incomplete information
        systems --- a missing antecedent does not match any value-specific itemset and therefore cannot appear
        in a discovered rule --- as defined for action-rule mining by Dardzinska, *Action Rules Mining*
        (Springer 2013, Section 2.3.2).  The target column is cast to strings in full so that any ``NaN``
        target value becomes its own explicit category (downstream ``get_split_bit_masks`` will exclude it
        from both the undesired and desired splits, which is the intended behaviour when callers want to
        ignore unlabelled rows).
        """

        def _prepare_antecedent_frame(frame, attributes):
            """Stringify non-missing antecedent cells while keeping ``NaN`` as ``NaN``.

            Letting ``get_dummies`` see a real ``NaN`` is the documented way to make it skip the value;
            calling ``astype(str)`` first would convert ``np.nan`` into the literal string ``'nan'`` and
            spawn a spurious one-hot column.
            """
            antecedent = frame[attributes].copy()
            return antecedent.where(antecedent.isna(), antecedent.astype(str))

        to_concat = []
        if len(stable_attributes) > 0:
            stable_frame = _prepare_antecedent_frame(data, stable_attributes)
            data_stable = self.pd.get_dummies(stable_frame, sparse=False, prefix_sep='_<item_stable>_')  # type: ignore
            to_concat.append(data_stable)
        if len(flexible_attributes) > 0:
            flexible_frame = _prepare_antecedent_frame(data, flexible_attributes)
            data_flexible = self.pd.get_dummies(  # type: ignore
                flexible_frame, sparse=False, prefix_sep='_<item_flexible>_'
            )
            to_concat.append(data_flexible)
        data_target = self.pd.get_dummies(  # type: ignore
            data[[target]].astype(str), sparse=False, prefix_sep='_<item_target>_'
        )
        to_concat.append(data_target)
        data = self.pd.concat(to_concat, axis=1)  # type: ignore
        return data

    def fit_onehot(
        self,
        data: Union['cudf.DataFrame', 'pandas.DataFrame'],
        stable_attributes: dict,
        flexible_attributes: dict,
        target: dict,
        target_undesired_state: str,
        target_desired_state: str,
        use_sparse_matrix: bool = False,
        use_gpu: bool = False,
        **kwargs,
    ):
        """
        Fit the model when input data is already one-hot encoded.

        The method remaps one-hot columns to the internal naming convention
        (`_<item_stable>_`, `_<item_flexible>_`, `_<item_target>_`), drops
        unrelated columns, and forwards execution to `fit`.

        Parameters
        ----------
        data : Union[cudf.DataFrame, pandas.DataFrame]
            The dataset to be processed and used for fitting the model.
        stable_attributes : dict
            A dictionary mapping stable attribute names to lists of column
            names corresponding to those attributes.
        flexible_attributes : dict
            A dictionary mapping flexible attribute names to lists of column
            names corresponding to those attributes.
        target : dict
            A dictionary mapping the target attribute name to a list of
            column names corresponding to that attribute.
        target_undesired_state : str
            The undesired state of the target attribute, used in action rule generation.
        target_desired_state : str
            The desired state of the target attribute, used in action rule generation.
        use_sparse_matrix : bool, optional
            Kept for backward compatibility with action-rules <= 1.0.11. The bitset
            backend supersedes sparse matrices, so this flag is accepted and ignored.
            Other unrecognized keyword arguments (``**kwargs``) are likewise accepted
            and ignored for backward compatibility with older call signatures.
        use_gpu : bool, optional
            If True, the GPU (cuDF) is used for data processing if available.
            Default is False.

        Notes
        -----
        This method expects boolean/binary one-hot columns.
        """
        self.is_onehot = True
        data = data.copy()
        data = data.astype('bool')
        new_labels = []
        attributes_stable = set([])
        attribtes_flexible = set([])
        attribute_target = ''
        remove_cols = []
        for label in data.columns:
            to_remove = True
            for attribute, columns in stable_attributes.items():
                if label in columns:
                    new_labels.append(attribute + '_<item_stable>_' + label)
                    attributes_stable.add(attribute)
                    to_remove = False
            for attribute, columns in flexible_attributes.items():
                if label in columns:
                    new_labels.append(attribute + '_<item_flexible>_' + label)
                    attribtes_flexible.add(attribute)
                    to_remove = False
            for attribute, columns in target.items():
                if label in columns:
                    new_labels.append(attribute + '_<item_target>_' + label)
                    attribute_target = attribute
                    to_remove = False
            if to_remove:
                new_labels.append(label)
                remove_cols.append(label)
        data.columns = new_labels
        data = data.drop(columns=remove_cols)
        self.fit(
            data,
            list(attributes_stable),
            list(attribtes_flexible),
            attribute_target,
            target_undesired_state,
            target_desired_state,
            use_sparse_matrix=use_sparse_matrix,
            use_gpu=use_gpu,
        )

    def fit(
        self,
        data: Union['cudf.DataFrame', 'pandas.DataFrame'],
        stable_attributes: list,
        flexible_attributes: list,
        target: str,
        target_undesired_state: str,
        target_desired_state: str,
        use_sparse_matrix: bool = False,
        use_gpu: bool = False,
        **kwargs,
    ):
        """
        Generate action rules for the provided dataset.

        Parameters
        ----------
        data : Union[cudf.DataFrame, pandas.DataFrame]
            The dataset to generate action rules from.
        stable_attributes : list
            List of stable attributes.
        flexible_attributes : list
            List of flexible attributes.
        target : str
            The target attribute.
        target_undesired_state : str
            The undesired state of the target attribute.
        target_desired_state : str
            The desired state of the target attribute.
        use_sparse_matrix : bool, optional
            Kept for backward compatibility with action-rules <= 1.0.11. The bitset
            backend supersedes sparse matrices, so this flag is accepted and ignored.
            Other unrecognized keyword arguments (``**kwargs``) are likewise accepted
            and ignored for backward compatibility with older call signatures.
        use_gpu : bool, optional
            Use GPU (cuDF) for data processing if available. Default is False.

        Raises
        ------
        RuntimeError
            If the model has already been fitted.

        Notes
        -----
        The method runs one-hot encoding (when needed), packs bit masks, explores
        candidate branches, prunes classification rules by depth, and finally
        materializes action rules.
        """
        if self.output is not None:
            raise RuntimeError("The model is already fit.")
        if use_sparse_matrix:
            warnings.warn(
                "The 'use_sparse_matrix' parameter is obsolete and has no effect: action-rules now "
                "always uses the packed-bitset backend.",
                UserWarning,
                stacklevel=2,
            )
        # Forward tolerance: legacy callers passed use_gpu="auto" for backend
        # autoselection. That harness lives outside the package now, so treat any
        # truthy string as a plain GPU request instead of raising.
        if isinstance(use_gpu, str):
            use_gpu = use_gpu.strip().lower() not in ("", "false", "cpu", "no", "0")

        # reset cached bitset structures before fitting a new model
        self.bit_masks = None
        self.target_state_bit_masks = None
        self.frames_bit_masks = None
        self.set_array_library(use_gpu, data)
        if not self.is_onehot:
            data = self.one_hot_encode(data, stable_attributes, flexible_attributes, target)
        data, columns = self.df_to_array(data)

        stable_items_binding, flexible_items_binding, target_items_binding, column_values = self.get_bindings(
            columns, stable_attributes, flexible_attributes, target
        )

        # Preserve original string-keyed tables before remapping to integer indices.
        # confidence_intervals() needs the originals to pass to inference engines.
        self._original_intrinsic_utility_table = dict(self.intrinsic_utility_table)
        self._original_transition_utility_table = dict(self.transition_utility_table)
        self._column_values = column_values
        self.intrinsic_utility_table, self.transition_utility_table = self.remap_utility_tables(column_values)

        local_bit_masks = self.build_bit_masks(data)
        self._cache_bitset_structures(local_bit_masks, target_items_binding, target)
        self.frames_bit_masks = self.get_split_bit_masks(target_items_binding, target)

        if self.verbose:
            print('Maximum number of nodes to check for support:')
            print('_____________________________________________')
            print(self.count_max_nodes(stable_items_binding, flexible_items_binding))
            print('')
        use_gpu_batching = bool(self.is_gpu_np and self.bit_masks is not None and self.frames_bit_masks)

        # Set membership is hot in candidate pruning; use a set internally for O(1) lookups.
        stop_list = set(self.get_stop_list(stable_items_binding, flexible_items_binding))
        undesired_state = columns.index(target + '_<item_target>_' + str(target_undesired_state))
        desired_state = columns.index(target + '_<item_target>_' + str(target_desired_state))

        stop_list_itemset = set()  # type: set

        initial_candidate = {
            'ar_prefix': tuple(),
            'itemset_prefix': tuple(),
            'stable_items_binding': stable_items_binding,
            'flexible_items_binding': flexible_items_binding,
            'actionable_attributes': 0,
        }
        candidates_pool = deque([initial_candidate])
        pending_depth_counts = {0: 1}
        min_pending_depth: Optional[int] = 0
        max_depth_seen = 0
        next_prune_depth = 1
        self.rules = Rules(
            undesired_state,
            desired_state,
            columns,
            data.shape[1],
            self.intrinsic_utility_table,
            self.transition_utility_table,
        )
        candidate_generator = CandidateGenerator(
            frames_bit_masks=self.frames_bit_masks,
            bit_masks=self.bit_masks,
            min_stable_attributes=self.min_stable_attributes,
            min_flexible_attributes=self.min_flexible_attributes,
            min_undesired_support=self.min_undesired_support,
            min_desired_support=self.min_desired_support,
            min_undesired_confidence=self.min_undesired_confidence,
            min_desired_confidence=self.min_desired_confidence,
            undesired_state=undesired_state,
            desired_state=desired_state,
            rules=self.rules,
            verbose=self.verbose,
        )
        # Default GPU node batch; the adaptive VRAM budgeting in CandidateGenerator
        # shrinks this automatically to fit available device memory.
        effective_gpu_node_batch_size = 32

        def pop_next_candidate() -> dict:
            """Pop one pending candidate and keep pending-depth bookkeeping in sync."""
            nonlocal min_pending_depth
            candidate_to_expand = candidates_pool.popleft()
            depth = len(candidate_to_expand['ar_prefix'])
            pending_depth_counts[depth] -= 1
            if pending_depth_counts[depth] <= 0:
                pending_depth_counts.pop(depth, None)
                if depth == min_pending_depth:
                    min_pending_depth = min(pending_depth_counts.keys(), default=None)
            return candidate_to_expand

        while len(candidates_pool) > 0:
            if use_gpu_batching:  # pragma: no cover
                batch: list = []
                while candidates_pool and len(batch) < effective_gpu_node_batch_size:
                    batch.append(pop_next_candidate())
                new_candidates = candidate_generator.generate_candidates_batch(
                    batch,
                    stop_list=stop_list,
                    stop_list_itemset=stop_list_itemset,
                    batch_size=effective_gpu_node_batch_size,
                )
            else:
                candidate = pop_next_candidate()
                new_candidates = candidate_generator.generate_candidates(
                    **candidate,
                    stop_list=stop_list,
                    stop_list_itemset=stop_list_itemset,
                )
            if new_candidates:
                candidates_pool.extend(new_candidates)
                for new_candidate in new_candidates:
                    new_depth = len(new_candidate['ar_prefix'])
                    pending_depth_counts[new_depth] = pending_depth_counts.get(new_depth, 0) + 1
                    if min_pending_depth is None or new_depth < min_pending_depth:
                        min_pending_depth = new_depth
                    if new_depth > max_depth_seen:
                        max_depth_seen = new_depth
            while next_prune_depth <= max_depth_seen and (
                min_pending_depth is None or min_pending_depth >= next_prune_depth
            ):
                self.rules.prune_classification_rules(next_prune_depth, stop_list)
                next_prune_depth += 1
        self.rules.generate_action_rules()
        self.output = Output(
            self.rules.action_rules, target, stable_items_binding, flexible_items_binding, column_values
        )
        del data
        if self.is_gpu_np:  # pragma: no cover
            gpu_pool = self.np.get_default_memory_pool()  # type: ignore[union-attr, attr-defined]
            gpu_pool.free_all_blocks()

    def get_bindings(
        self,
        columns: list,
        stable_attributes: list,
        flexible_attributes: list,
        target: str,
    ) -> tuple:
        """
        Bind stable/flexible/target attribute to corresponding column in the dataset.

        Parameters
        ----------
        columns : list
            List of column names in the dataset.
        stable_attributes : list
            List of stable attributes.
        flexible_attributes : list
            List of flexible attributes.
        target : str
            The target attribute.

        Returns
        -------
        tuple
            A tuple containing the bindings for stable attributes, flexible attributes, and target items.

        Notes
        -----
        The method generates mappings from column indices to attribute values for stable, flexible, and target
        attributes.
        """
        stable_items_binding = defaultdict(lambda: [])
        flexible_items_binding = defaultdict(lambda: [])
        target_items_binding = defaultdict(lambda: [])
        column_values = {}

        for i, col in enumerate(columns):
            is_continue = False
            # stable
            for attribute in stable_attributes:
                if col.startswith(attribute + '_<item_stable>_'):
                    stable_items_binding[attribute].append(i)
                    column_values[i] = (attribute, col.split('_<item_stable>_', 1)[1])
                    is_continue = True
                    break
            if is_continue is True:
                continue
            # flexible
            for attribute in flexible_attributes:
                if col.startswith(attribute + '_<item_flexible>_'):
                    flexible_items_binding[attribute].append(i)
                    column_values[i] = (attribute, col.split('_<item_flexible>_', 1)[1])
                    is_continue = True
                    break
            if is_continue is True:
                continue
            # target
            if col.startswith(target + '_<item_target>_'):
                target_items_binding[target].append(i)
                column_values[i] = (target, col.split('_<item_target>_', 1)[1])
        return stable_items_binding, flexible_items_binding, target_items_binding, column_values

    def get_stop_list(self, stable_items_binding: dict, flexible_items_binding: dict) -> list:
        """
        Generate a stop list to prevent certain combinations of attributes.

        Parameters
        ----------
        stable_items_binding : dict
            Dictionary containing bindings for stable items.
        flexible_items_binding : dict
            Dictionary containing bindings for flexible items.

        Returns
        -------
        list
            A list of stop combinations.

        Notes
        -----
        The stop list is generated by creating pairs of stable item indices and ensuring flexible items do not repeat.
        """
        stop_list = []
        for items in stable_items_binding.values():
            for stop_couple in itertools.product(items, repeat=2):
                stop_list.append(tuple(stop_couple))
        for item in flexible_items_binding.keys():
            stop_list.append(tuple([item, item]))
        return stop_list

    def get_split_bit_masks(self, target_items_binding: dict, target: str) -> dict:
        """
        Return packed bit-mask rows for each target state.

        Parameters
        ----------
        target_items_binding : dict
            Indexes of target attributes columns in one-hot table.
        target : str
            Name of the target attribute.

        Returns
        -------
        dict
            Dictionary mapping target attributes to the corresponding packed mask rows.

        Notes
        -----
        Requires that `build_bit_masks` has been executed beforehand.
        """
        if self.bit_masks is None:
            raise RuntimeError("Bit masks are not available. Ensure fit() was run first.")

        target_state_masks = {}
        for item_index in target_items_binding.get(target, []):
            target_state_masks[item_index] = self.bit_masks[item_index]
        return target_state_masks

    def get_rules(self) -> Output:
        """
        Return the generated action rules if available.

        Raises
        ------
        RuntimeError
            If the model has not been fitted.

        Returns
        -------
        Output
            The generated action rules.

        Notes
        -----
        This method returns the `Output` object containing the generated action rules.
        """
        if self.output is None:
            raise RuntimeError("The model is not fit.")
        return self.output

    def predict(self, frame_row: Union['cudf.Series', 'pandas.Series']) -> Union['cudf.DataFrame', 'pandas.DataFrame']:
        """
        Predict recommended actions based on the provided row of data.

        This method applies the fitted action rules to the given row of data and generates
        a DataFrame with recommended actions if any of the action rules are triggered.

        Parameters
        ----------
        frame_row : Union['cudf.Series', 'pandas.Series']
            A row of data in the form of a cuDF or pandas Series. The Series should
            contain the features required by the action rules.

        Returns
        -------
        Union['cudf.DataFrame', 'pandas.DataFrame']
            A DataFrame with the recommended actions. The DataFrame includes the following columns:
            - The original attributes with recommended changes.
            - 'ActionRules_RuleIndex': Index of the action rule applied.
            - 'ActionRules_UndesiredSupport': Support of the undesired part of the rule.
            - 'ActionRules_DesiredSupport': Support of the desired part of the rule.
            - 'ActionRules_UndesiredConfidence': Confidence of the undesired part of the rule.
            - 'ActionRules_DesiredConfidence': Confidence of the desired part of the rule.
            - 'ActionRules_Uplift': Uplift value of the rule.

        Raises
        ------
        RuntimeError
            If the model has not been fitted.

        Notes
        -----
        The method compares the given row of data against the undesired itemsets of the action rules.
        If a match is found, it applies the desired itemset changes and records the action rule's
        metadata. The result is a DataFrame with one or more rows representing the recommended actions
        for the given data.
        """
        if self.output is None:
            raise RuntimeError("The model is not fit.")
        index_value_tuples = list(zip(frame_row.index, frame_row))
        values = []
        column_values = self.output.column_values
        for index_value_tuple in index_value_tuples:
            values.append(list(column_values.keys())[list(column_values.values()).index(index_value_tuple)])
        new_values = tuple(values)
        predicted = []
        for i, action_rule in enumerate(self.output.action_rules):
            if set(action_rule['undesired']['itemset']) <= set(new_values):
                predicted_row = frame_row.copy()
                for recommended in set(action_rule['desired']['itemset']) - set(new_values):
                    attribute, value = column_values[recommended]
                    predicted_row[attribute + ' (Recommended)'] = value
                predicted_row['ActionRules_RuleIndex'] = i
                predicted_row['ActionRules_UndesiredSupport'] = action_rule['undesired']['support']
                predicted_row['ActionRules_DesiredSupport'] = action_rule['desired']['support']
                predicted_row['ActionRules_UndesiredConfidence'] = action_rule['undesired']['confidence']
                predicted_row['ActionRules_DesiredConfidence'] = action_rule['desired']['confidence']
                predicted_row['ActionRules_Uplift'] = action_rule['uplift']
                predicted.append(predicted_row)
        return self.pd.DataFrame(predicted)  # type: ignore

    def confidence_intervals(
        self,
        data,
        method: str = "bootstrap",
        confidence_level: float = 0.95,
        threshold: Optional[float] = None,
        metric: str = "uplift",
        n_bootstrap: int = 1000,
        n_mc: int = 10000,
        random_state: Optional[int] = None,
        analytic_type: str = "wald",
        bootstrap_type: str = "percentile",
    ):
        r"""Compute confidence intervals for all fitted action rules.

        Applies a statistical inference engine to the provided dataset and
        attaches confidence interval results to the output object.  Results
        are also returned directly for immediate inspection.

        Parameters
        ----------
        data : Union[cudf.DataFrame, pandas.DataFrame]
            The original (pre-encoding) dataset used for inference.  Columns
            must match the attribute names supplied during ``fit()``.
        method : str, optional
            CI method to use.  One of:

            - ``'bootstrap'`` — non-parametric percentile bootstrap
              (default).
            - ``'analytic'`` or ``'wald'`` — closed-form Wald interval via
              the delta method (requires ``scipy``).
            - ``'bayesian'`` — Beta-Binomial conjugate model with Monte
              Carlo posterior sampling.
        confidence_level : float, optional
            Nominal coverage probability, e.g. ``0.95`` (default).
        threshold : float, optional
            Decision boundary used to categorise rules after computing
            intervals.  When ``None`` (default), categorisation is skipped.
        metric : str, optional
            Metric to use for categorisation when *threshold* is provided.
            One of ``'uplift'`` (default) or ``'realistic_rule_gain'``.
        n_bootstrap : int, optional
            Number of bootstrap resamples.  Only used when
            ``method='bootstrap'``.  Default ``1000``.
        n_mc : int, optional
            Number of Monte Carlo samples.  Only used when
            ``method='bayesian'``.  Default ``10000``.
        random_state : int, optional
            Seed for reproducibility.  Passed to the engine when applicable.
            ``None`` uses the global NumPy random state.
        analytic_type : str, optional
            Sub-type of the analytic method.  Only used when
            ``method='analytic'`` or ``method='wald'``.  One of:

            - ``'wald'`` — standard Wald normal approximation (default).
            - ``'newcombe_wilson'`` (preferred) or ``'wilson'`` (alias) —
              Newcombe-Wilson interval (Newcombe, 1998) for the unscaled
              rule contrast :math:`\\delta = p_d + p_u - 1`, built by
              combining two single-proportion Wilson score intervals; the
              resulting interval is asymmetric.  ``'wilson'`` is retained
              as a backward-compatible alias.
            - ``'auto'`` — Newcombe-Wilson when sample is small (``n < 40``)
              or proportion is extreme (``< 0.05`` or ``> 0.95``), Wald
              otherwise (following Agresti & Coull, 1998).
        bootstrap_type : str, optional
            Sub-type of the bootstrap method.  Only used when
            ``method='bootstrap'``.  One of:

            - ``'percentile'`` — standard percentile bootstrap (default).
            - ``'bca'`` — bias-corrected and accelerated (BCa) interval,
              which adjusts for bias and skewness using jackknife
              acceleration (Efron, 1987).

        Returns
        -------
        list
            List of :class:`~action_rules.inference.base.ConfidenceIntervalResult`
            objects, one per action rule, in the same order as
            ``self.output.action_rules``.

        Raises
        ------
        RuntimeError
            If the model has not been fitted yet (``self.output is None``).
        ValueError
            If *method* is not one of the supported values.

        Notes
        -----
        Results are also stored on the output object via
        ``self.output.set_confidence_intervals(results)`` so that subsequent
        calls to ``get_ar_notation()``, ``get_pretty_ar_notation()``, and
        ``get_export_notation()`` include the CI information.
        """
        if self.output is None:
            raise RuntimeError("The model is not fit.")

        if not (0 < confidence_level < 1):
            raise ValueError("confidence_level must be strictly between 0 and 1.")
        if n_bootstrap < 1:
            raise ValueError("n_bootstrap must be >= 1.")
        if n_mc < 1:
            raise ValueError("n_mc must be >= 1.")
        valid_analytic_types = {"wald", "wilson", "newcombe_wilson", "auto"}
        if analytic_type not in valid_analytic_types:
            raise ValueError(f"Unknown analytic_type '{analytic_type}'. Choose from {valid_analytic_types}.")
        valid_bootstrap_types = {"percentile", "bca"}
        if bootstrap_type not in valid_bootstrap_types:
            raise ValueError(f"Unknown bootstrap_type '{bootstrap_type}'. Choose from {valid_bootstrap_types}.")
        valid_metrics = {"uplift", "realistic_rule_gain"}
        if metric not in valid_metrics:
            raise ValueError(f"Unknown metric '{metric}'. Choose from {valid_metrics}.")

        from .inference.base import categorize_rule, extract_rule_masks

        masks = extract_rule_masks(self.output)

        engine: Any
        if method == "bootstrap":
            from .inference.bootstrap import BootstrapEngine

            engine = BootstrapEngine(n_bootstrap, random_state, bootstrap_type=bootstrap_type)
        elif method in ("analytic", "wald"):
            from .inference.analytic import AnalyticEngine

            engine = AnalyticEngine(analytic_type=analytic_type)
        elif method == "bayesian":
            from .inference.bayesian import BayesianEngine

            engine = BayesianEngine(n_mc, random_state=random_state)
        else:
            raise ValueError(
                f"Unknown method '{method}'. Supported methods: 'bootstrap', 'analytic', 'wald', 'bayesian'."
            )

        results = engine.compute(
            data=data,
            rules=masks,
            confidence_level=confidence_level,
            intrinsic_utility_table=self._original_intrinsic_utility_table or None,
            transition_utility_table=self._original_transition_utility_table or None,
            column_values=self._column_values,
        )

        if threshold is not None:
            for result in results:
                if metric == "uplift":
                    result.category = categorize_rule(result.uplift_ci_lower, result.uplift_ci_upper, threshold)
                else:
                    if (
                        result.realistic_rule_gain_ci_lower is not None
                        and result.realistic_rule_gain_ci_upper is not None
                    ):
                        result.category = categorize_rule(
                            result.realistic_rule_gain_ci_lower,
                            result.realistic_rule_gain_ci_upper,
                            threshold,
                        )

        self.output.set_confidence_intervals(results)
        return results

    def cross_validate(
        self,
        data,
        stable_attributes: list,
        flexible_attributes: list,
        target: str,
        target_undesired_state: str,
        target_desired_state: str,
        *,
        n_splits: int = 5,
        stratify: bool = True,
        strategies=None,
        metrics=None,
        k_fraction: float = 0.2,
        ci_method: str = 'bootstrap',
        n_bootstrap: int = 500,
        risk_lambda: float = 1.96,
        confidence_level: float = 0.95,
        random_state: Optional[int] = None,
        n_bootstrap_oof: int = 1000,
        bootstrap_design: str = 'cluster_fold',
        track_stability: bool = True,
        use_sparse_matrix: bool = False,
        scale_support_thresholds: bool = True,
        compute_insample_baseline: bool = False,
    ):
        """Run stratified K-fold cross-validation on the action-rule pipeline.

        Each fold receives a fresh :class:`ActionRules` instance configured
        with the same hyperparameters and utility tables as ``self``.  Per
        fold, rules are mined on the train split, confidence intervals are
        computed on the train split, and every discovered rule is re-scored
        on the held-out test split (``test_uplift``, ``test_realistic_gain``).
        Targeting metrics (``uplift@k``, Qini, AUUC, ``profit@k``) are
        evaluated under several targeting strategies on the test split.

        Parameters
        ----------
        data : pandas.DataFrame
            The full dataset, pre-encoding.
        stable_attributes, flexible_attributes : list of str
        target : str
        target_undesired_state, target_desired_state : str
        n_splits : int, optional
            Number of folds (default ``5``).  Must be ≥ 2.
        stratify : bool, optional
            Whether to stratify folds by ``target`` value (default ``True``).
        strategies : sequence of str, optional
            Subset of ``('point', 'lower', 'lower_positive', 'risk_adjusted')``.
            Defaults to all four.
        metrics : sequence of str, optional
            Subset of ``('uplift_at_k', 'qini', 'auuc', 'profit_at_k')``.
            Defaults to all four.
        k_fraction : float, optional
            Top-k cutoff used by the ``*_at_k`` metrics (default ``0.2``).
        ci_method, n_bootstrap, confidence_level, risk_lambda : forwarded to
            :class:`~action_rules.evaluation.cv.CrossValidator`.
        random_state : int, optional
            Seed for fold splitting and bootstrap CIs.
        n_bootstrap_oof : int, optional
            Bootstrap replicates for the across-fold rule-resampling CI.
            Set to ``0`` to disable bootstrap CIs.
        bootstrap_design : str, optional
            ``'cluster_fold'`` (default) resamples rules within each fold,
            computes the metric per fold, and averages — so the bootstrap CI
            estimates the same fold-mean quantity as the ``mean`` column.
            ``'oof_pool'`` (legacy) resamples within fold then concatenates
            into one pool before computing the metric; estimates a pool-level
            statistic that differs from the fold mean by roughly a factor of
            K.
        track_stability : bool, optional
            Compute pairwise Jaccard overlap of discovered rule sets across
            folds (default ``True``).
        compute_insample_baseline : bool, optional
            When ``True``, additionally mine on the full dataset and score on
            the full dataset to compute an apparent (in-sample) performance
            baseline; the result is stored on
            ``CrossValidationResult.insample_summary``.  Default ``False``
            preserves the existing return shape.

        Returns
        -------
        action_rules.evaluation.cv.CrossValidationResult

        Notes
        -----
        - Naive K-fold CIs based on ``mean ± 1.96·std/√K`` over folds have
          below-nominal coverage (Bates, Hastie & Tibshirani, 2021,
          arXiv:2104.00673).  This method therefore reports fold spread
          (``std``) as a stability indicator and a stratified bootstrap CI
          over OOF rule records as the inferential interval.
        - Calling :meth:`cross_validate` does **not** require the model to
          be fitted on the full data first.  It does not mutate ``self``;
          each fold operates on a fresh internal instance.
        """
        from .evaluation.cv import METRICS, STRATEGIES, CrossValidator

        # Snapshot the hyperparameters needed to build pristine per-fold instances.
        min_stable_attributes = self.min_stable_attributes
        min_flexible_attributes = self.min_flexible_attributes
        # Support thresholds are absolute counts.  When mining on a train fold
        # that is ``(n_splits-1)/n_splits`` of the full data, scale them down
        # proportionally so the same prevalence requirements apply on each fold.
        scale = (n_splits - 1) / n_splits if scale_support_thresholds else 1.0
        min_undesired_support = max(1, int(round(self.min_undesired_support * scale)))
        min_desired_support = max(1, int(round(self.min_desired_support * scale)))
        min_undesired_confidence = self.min_undesired_confidence
        min_desired_confidence = self.min_desired_confidence
        verbose = self.verbose
        intrinsic = self._original_intrinsic_utility_table or dict(self.intrinsic_utility_table)
        transition = self._original_transition_utility_table or dict(self.transition_utility_table)

        def _factory():
            return ActionRules(
                min_stable_attributes=min_stable_attributes,
                min_flexible_attributes=min_flexible_attributes,
                min_undesired_support=min_undesired_support,
                min_undesired_confidence=min_undesired_confidence,
                min_desired_support=min_desired_support,
                min_desired_confidence=min_desired_confidence,
                verbose=verbose,
                intrinsic_utility_table=intrinsic or None,
                transition_utility_table=transition or None,
            )

        validator = CrossValidator(
            _factory,
            stable_attributes=stable_attributes,
            flexible_attributes=flexible_attributes,
            target=target,
            target_undesired_state=target_undesired_state,
            target_desired_state=target_desired_state,
            n_splits=n_splits,
            stratify=stratify,
            intrinsic_utility_table=intrinsic,
            transition_utility_table=transition,
            strategies=STRATEGIES if strategies is None else strategies,
            metrics=METRICS if metrics is None else metrics,
            k_fraction=k_fraction,
            ci_method=ci_method,
            n_bootstrap=n_bootstrap,
            risk_lambda=risk_lambda,
            confidence_level=confidence_level,
            random_state=random_state,
            n_bootstrap_oof=n_bootstrap_oof,
            bootstrap_design=bootstrap_design,
            track_stability=track_stability,
            use_sparse_matrix=use_sparse_matrix,
            compute_insample_baseline=compute_insample_baseline,
        )
        return validator.run(data)

    def remap_utility_tables(self, column_values):
        """
        Remap the keys of intrinsic and transition utility tables using the provided column mapping.

        The function uses `column_values`, a dictionary mapping internal column indices to
        (attribute, value) tuples, to invert the mapping so that utility table keys are replaced
        with the corresponding integer index (for intrinsic utilities) or a tuple of integer indices
        (for transition utilities).

        Parameters
        ----------
        column_values : dict
            Dictionary mapping integer column indices to (attribute, value) pairs.
            Example: {0: ('Age', 'O'), 1: ('Age', 'Y'), 2: ('Sex', 'F'), ...}

        Returns
        -------
        tuple
            A tuple (remapped_intrinsic, remapped_transition) where:
              - remapped_intrinsic is a dict mapping integer column index to utility value.
              - remapped_transition is a dict mapping (from_index, to_index) to utility value.

        Notes
        -----
        - The method performs case-insensitive matching by converting attribute names and values to lowercase.
        - If a key in a utility table does not have a corresponding entry in column_values, it is skipped.
        """
        # Invert column_values to map (attribute.lower(), value.lower()) -> column index.
        inv_map = {(attr.lower(), val.lower()): idx for idx, (attr, val) in column_values.items()}

        remapped_intrinsic = {}
        # Remap intrinsic utility table keys: ('Attribute', 'Value') -> utility
        for key, utility in self.intrinsic_utility_table.items():
            # Normalize key to lowercase
            attr, val = key
            lookup_key = (attr.lower(), val.lower())
            # Look up the corresponding column index; if not found, skip this key.
            if lookup_key in inv_map:
                col_index = inv_map[lookup_key]
                remapped_intrinsic[col_index] = utility
            # Else: optionally, one could log or warn about a missing mapping.

        remapped_transition = {}
        # Remap transition utility table keys: ('Attribute', from_value, to_value) -> utility
        for key, utility in self.transition_utility_table.items():
            attr, from_val, to_val = key
            lookup_from = (attr.lower(), from_val.lower())
            lookup_to = (attr.lower(), to_val.lower())
            # Only remap if both the from and to values exist in inv_map.
            if lookup_from in inv_map and lookup_to in inv_map:
                from_index = inv_map[lookup_from]
                to_index = inv_map[lookup_to]
                remapped_transition[(from_index, to_index)] = utility
            # Else: skip or log missing mapping.

        return remapped_intrinsic, remapped_transition

__init__(min_stable_attributes, min_flexible_attributes, min_undesired_support, min_undesired_confidence, min_desired_support, min_desired_confidence, verbose=False, intrinsic_utility_table=None, transition_utility_table=None)

Initialize the ActionRules class with the specified parameters.

Parameters:

Name	Type	Description	Default
min_stable_attributes	int	The minimum number of stable attributes required.	required
min_flexible_attributes	int	The minimum number of flexible attributes required.	required
min_undesired_support	int	The minimum support for the undesired state.	required
min_undesired_confidence	float	The minimum confidence for the undesired state.	required
min_desired_support	int	The minimum support for the desired state.	required
min_desired_confidence	float	The minimum confidence for the desired state.	required
verbose	bool	If True, enables verbose output. Default is False.	False
intrinsic_utility_table	dict	(attribute, value) -> float A lookup table for the intrinsic utility of each attribute-value pair. If None, no intrinsic utility is considered.	None
transition_utility_table	dict	(attribute, from_value, to_value) -> float A lookup table for cost/gain of transitions between values. If None, no transition utility is considered.	None

Notes

The verbose parameter can be used to enable detailed output during the rule generation process.

Source code in src/action_rules/action_rules.py

def __init__(
    self,
    min_stable_attributes: int,
    min_flexible_attributes: int,
    min_undesired_support: int,
    min_undesired_confidence: float,
    min_desired_support: int,
    min_desired_confidence: float,
    verbose=False,
    intrinsic_utility_table: Optional[dict] = None,
    transition_utility_table: Optional[dict] = None,
):
    """
    Initialize the ActionRules class with the specified parameters.

    Parameters
    ----------
    min_stable_attributes : int
        The minimum number of stable attributes required.
    min_flexible_attributes : int
        The minimum number of flexible attributes required.
    min_undesired_support : int
        The minimum support for the undesired state.
    min_undesired_confidence : float
        The minimum confidence for the undesired state.
    min_desired_support : int
        The minimum support for the desired state.
    min_desired_confidence : float
        The minimum confidence for the desired state.
    verbose : bool, optional
        If True, enables verbose output. Default is False.
    intrinsic_utility_table : dict, optional
        (attribute, value) -> float
        A lookup table for the intrinsic utility of each attribute-value pair.
        If None, no intrinsic utility is considered.
    transition_utility_table : dict, optional
        (attribute, from_value, to_value) -> float
        A lookup table for cost/gain of transitions between values.
        If None, no transition utility is considered.

    Notes
    -----
    The `verbose` parameter can be used to enable detailed output during the rule generation process.
    """
    self.min_stable_attributes = min_stable_attributes
    self.min_flexible_attributes = min_flexible_attributes
    self.min_undesired_support = min_undesired_support
    self.min_desired_support = min_desired_support
    self.min_undesired_confidence = min_undesired_confidence
    self.min_desired_confidence = min_desired_confidence
    self.verbose = verbose
    self.rules = None  # type: Optional[Rules]
    self.output = None  # type: Optional[Output]
    self.np = None  # type: Optional[ModuleType]
    self.pd = None  # type: Optional[ModuleType]
    self.is_gpu_np = False
    self.is_gpu_pd = False
    self.is_onehot = False
    self.bit_masks = None  # type: Optional['numpy.ndarray']
    self.target_state_bit_masks = None  # type: Optional[dict]
    self.frames_bit_masks = None  # type: Optional[dict]
    self.intrinsic_utility_table = intrinsic_utility_table or {}
    self.transition_utility_table = transition_utility_table or {}
    self._original_intrinsic_utility_table = {}  # type: dict
    self._original_transition_utility_table = {}  # type: dict
    self._column_values = None  # type: Optional[dict]

build_bit_masks(data)

Pack a binary feature matrix into 64-bit masks for fast intersection.

Parameters:

Name	Type	Description	Default
data	Union[ndarray, ndarray]	Dense matrix produced by df_to_array, shaped (num_attributes, num_transactions) and containing 0/1 values.	required

Returns:

Type	Description
Union[ndarray, ndarray]	bit_masks is a uint64 array with shape (num_attributes, num_words) holding 64 bits/transactions for each item.

Notes

• The packing uses 64-bit little-endian words (bit 0 corresponds to the first transaction in each chunk).
• Sparse inputs are not supported; callers should densify before packing.

Source code in src/action_rules/action_rules.py

def build_bit_masks(
    self,
    data: Union['numpy.ndarray', 'cupy.ndarray'],
) -> Union['numpy.ndarray', 'cupy.ndarray']:
    """
    Pack a binary feature matrix into 64-bit masks for fast intersection.

    Parameters
    ----------
    data : Union[numpy.ndarray, cupy.ndarray]
        Dense matrix produced by `df_to_array`, shaped (num_attributes, num_transactions)
        and containing 0/1 values.

    Returns
    -------
    Union[numpy.ndarray, cupy.ndarray]
        bit_masks is a uint64 array with shape (num_attributes, num_words)
        holding 64 bits/transactions for each item.

    Notes
    -----
    - The packing uses 64-bit little-endian words (bit 0 corresponds to the
      first transaction in each chunk).
    - Sparse inputs are not supported; callers should densify before packing.
    """
    if self.np is None:
        raise RuntimeError("Array library is not initialised. Call set_array_library first.")
    # Shape is (num_attributes, num_transactions).
    num_attributes, num_transactions = data.shape
    num_words = (num_transactions + 63) // 64
    padded_transactions = num_words * 64
    padding = padded_transactions - num_transactions

    if padding > 0:
        pad_block = self.np.zeros((num_attributes, padding), dtype=data.dtype)
        padded_data = self.np.concatenate((data, pad_block), axis=1)
    else:
        padded_data = data

    # Group transactions into 64-bit chunks: (num_attributes, num_words, 64).
    chunks = padded_data.reshape(num_attributes, num_words, 64).astype(self.np.uint64, copy=False)
    bit_offsets = self.np.arange(64, dtype=self.np.uint64)
    bit_weights = self.np.left_shift(self.np.uint64(1), bit_offsets)

    # Pack each 64-sized transaction chunk into one uint64 word.
    bit_masks = self.np.tensordot(chunks, bit_weights, axes=([2], [0])).astype(self.np.uint64, copy=False)
    return bit_masks

confidence_intervals(data, method='bootstrap', confidence_level=0.95, threshold=None, metric='uplift', n_bootstrap=1000, n_mc=10000, random_state=None, analytic_type='wald', bootstrap_type='percentile')

Compute confidence intervals for all fitted action rules.

Applies a statistical inference engine to the provided dataset and attaches confidence interval results to the output object. Results are also returned directly for immediate inspection.

Parameters:

Name	Type	Description	Default
data	Union[DataFrame, DataFrame]	The original (pre-encoding) dataset used for inference. Columns must match the attribute names supplied during fit().	required
method	str	CI method to use. One of: 'bootstrap' — non-parametric percentile bootstrap (default). 'analytic' or 'wald' — closed-form Wald interval via the delta method (requires scipy). 'bayesian' — Beta-Binomial conjugate model with Monte Carlo posterior sampling.	'bootstrap'
confidence_level	float	Nominal coverage probability, e.g. 0.95 (default).	0.95
threshold	float	Decision boundary used to categorise rules after computing intervals. When None (default), categorisation is skipped.	None
metric	str	Metric to use for categorisation when threshold is provided. One of 'uplift' (default) or 'realistic_rule_gain'.	'uplift'
n_bootstrap	int	Number of bootstrap resamples. Only used when method='bootstrap'. Default 1000.	1000
n_mc	int	Number of Monte Carlo samples. Only used when method='bayesian'. Default 10000.	10000
random_state	int	Seed for reproducibility. Passed to the engine when applicable. None uses the global NumPy random state.	None
analytic_type	str	Sub-type of the analytic method. Only used when method='analytic' or method='wald'. One of: 'wald' — standard Wald normal approximation (default). 'newcombe_wilson' (preferred) or 'wilson' (alias) — Newcombe-Wilson interval (Newcombe, 1998) for the unscaled rule contrast :math:\\delta = p_d + p_u - 1, built by combining two single-proportion Wilson score intervals; the resulting interval is asymmetric. 'wilson' is retained as a backward-compatible alias. 'auto' — Newcombe-Wilson when sample is small (n < 40) or proportion is extreme (< 0.05 or > 0.95), Wald otherwise (following Agresti & Coull, 1998).	'wald'
bootstrap_type	str	Sub-type of the bootstrap method. Only used when method='bootstrap'. One of: 'percentile' — standard percentile bootstrap (default). 'bca' — bias-corrected and accelerated (BCa) interval, which adjusts for bias and skewness using jackknife acceleration (Efron, 1987).	'percentile'

Returns:

Type	Description
list	List of :class:~action_rules.inference.base.ConfidenceIntervalResult objects, one per action rule, in the same order as self.output.action_rules.

Raises:

Type	Description
RuntimeError	If the model has not been fitted yet (self.output is None).
ValueError	If method is not one of the supported values.

Notes

Results are also stored on the output object via self.output.set_confidence_intervals(results) so that subsequent calls to get_ar_notation(), get_pretty_ar_notation(), and get_export_notation() include the CI information.

Source code in src/action_rules/action_rules.py

def confidence_intervals(
    self,
    data,
    method: str = "bootstrap",
    confidence_level: float = 0.95,
    threshold: Optional[float] = None,
    metric: str = "uplift",
    n_bootstrap: int = 1000,
    n_mc: int = 10000,
    random_state: Optional[int] = None,
    analytic_type: str = "wald",
    bootstrap_type: str = "percentile",
):
    r"""Compute confidence intervals for all fitted action rules.

    Applies a statistical inference engine to the provided dataset and
    attaches confidence interval results to the output object.  Results
    are also returned directly for immediate inspection.

    Parameters
    ----------
    data : Union[cudf.DataFrame, pandas.DataFrame]
        The original (pre-encoding) dataset used for inference.  Columns
        must match the attribute names supplied during ``fit()``.
    method : str, optional
        CI method to use.  One of:

        - ``'bootstrap'`` — non-parametric percentile bootstrap
          (default).
        - ``'analytic'`` or ``'wald'`` — closed-form Wald interval via
          the delta method (requires ``scipy``).
        - ``'bayesian'`` — Beta-Binomial conjugate model with Monte
          Carlo posterior sampling.
    confidence_level : float, optional
        Nominal coverage probability, e.g. ``0.95`` (default).
    threshold : float, optional
        Decision boundary used to categorise rules after computing
        intervals.  When ``None`` (default), categorisation is skipped.
    metric : str, optional
        Metric to use for categorisation when *threshold* is provided.
        One of ``'uplift'`` (default) or ``'realistic_rule_gain'``.
    n_bootstrap : int, optional
        Number of bootstrap resamples.  Only used when
        ``method='bootstrap'``.  Default ``1000``.
    n_mc : int, optional
        Number of Monte Carlo samples.  Only used when
        ``method='bayesian'``.  Default ``10000``.
    random_state : int, optional
        Seed for reproducibility.  Passed to the engine when applicable.
        ``None`` uses the global NumPy random state.
    analytic_type : str, optional
        Sub-type of the analytic method.  Only used when
        ``method='analytic'`` or ``method='wald'``.  One of:

        - ``'wald'`` — standard Wald normal approximation (default).
        - ``'newcombe_wilson'`` (preferred) or ``'wilson'`` (alias) —
          Newcombe-Wilson interval (Newcombe, 1998) for the unscaled
          rule contrast :math:`\\delta = p_d + p_u - 1`, built by
          combining two single-proportion Wilson score intervals; the
          resulting interval is asymmetric.  ``'wilson'`` is retained
          as a backward-compatible alias.
        - ``'auto'`` — Newcombe-Wilson when sample is small (``n < 40``)
          or proportion is extreme (``< 0.05`` or ``> 0.95``), Wald
          otherwise (following Agresti & Coull, 1998).
    bootstrap_type : str, optional
        Sub-type of the bootstrap method.  Only used when
        ``method='bootstrap'``.  One of:

        - ``'percentile'`` — standard percentile bootstrap (default).
        - ``'bca'`` — bias-corrected and accelerated (BCa) interval,
          which adjusts for bias and skewness using jackknife
          acceleration (Efron, 1987).

    Returns
    -------
    list
        List of :class:`~action_rules.inference.base.ConfidenceIntervalResult`
        objects, one per action rule, in the same order as
        ``self.output.action_rules``.

    Raises
    ------
    RuntimeError
        If the model has not been fitted yet (``self.output is None``).
    ValueError
        If *method* is not one of the supported values.

    Notes
    -----
    Results are also stored on the output object via
    ``self.output.set_confidence_intervals(results)`` so that subsequent
    calls to ``get_ar_notation()``, ``get_pretty_ar_notation()``, and
    ``get_export_notation()`` include the CI information.
    """
    if self.output is None:
        raise RuntimeError("The model is not fit.")

    if not (0 < confidence_level < 1):
        raise ValueError("confidence_level must be strictly between 0 and 1.")
    if n_bootstrap < 1:
        raise ValueError("n_bootstrap must be >= 1.")
    if n_mc < 1:
        raise ValueError("n_mc must be >= 1.")
    valid_analytic_types = {"wald", "wilson", "newcombe_wilson", "auto"}
    if analytic_type not in valid_analytic_types:
        raise ValueError(f"Unknown analytic_type '{analytic_type}'. Choose from {valid_analytic_types}.")
    valid_bootstrap_types = {"percentile", "bca"}
    if bootstrap_type not in valid_bootstrap_types:
        raise ValueError(f"Unknown bootstrap_type '{bootstrap_type}'. Choose from {valid_bootstrap_types}.")
    valid_metrics = {"uplift", "realistic_rule_gain"}
    if metric not in valid_metrics:
        raise ValueError(f"Unknown metric '{metric}'. Choose from {valid_metrics}.")

    from .inference.base import categorize_rule, extract_rule_masks

    masks = extract_rule_masks(self.output)

    engine: Any
    if method == "bootstrap":
        from .inference.bootstrap import BootstrapEngine

        engine = BootstrapEngine(n_bootstrap, random_state, bootstrap_type=bootstrap_type)
    elif method in ("analytic", "wald"):
        from .inference.analytic import AnalyticEngine

        engine = AnalyticEngine(analytic_type=analytic_type)
    elif method == "bayesian":
        from .inference.bayesian import BayesianEngine

        engine = BayesianEngine(n_mc, random_state=random_state)
    else:
        raise ValueError(
            f"Unknown method '{method}'. Supported methods: 'bootstrap', 'analytic', 'wald', 'bayesian'."
        )

    results = engine.compute(
        data=data,
        rules=masks,
        confidence_level=confidence_level,
        intrinsic_utility_table=self._original_intrinsic_utility_table or None,
        transition_utility_table=self._original_transition_utility_table or None,
        column_values=self._column_values,
    )

    if threshold is not None:
        for result in results:
            if metric == "uplift":
                result.category = categorize_rule(result.uplift_ci_lower, result.uplift_ci_upper, threshold)
            else:
                if (
                    result.realistic_rule_gain_ci_lower is not None
                    and result.realistic_rule_gain_ci_upper is not None
                ):
                    result.category = categorize_rule(
                        result.realistic_rule_gain_ci_lower,
                        result.realistic_rule_gain_ci_upper,
                        threshold,
                    )

    self.output.set_confidence_intervals(results)
    return results

count_max_nodes(stable_items_binding, flexible_items_binding)

Calculate the maximum number of nodes based on the given item bindings.

This function takes two dictionaries, stable_items_binding and flexible_items_binding, which map attributes to lists of items. It calculates the total number of nodes by considering all possible combinations of the lengths of these item lists and summing the product of each combination.

Parameters:

Name	Type	Description	Default
stable_items_binding	dict	A dictionary where keys are attributes and values are lists of stable items.	required
flexible_items_binding	dict	A dictionary where keys are attributes and values are lists of flexible items.	required

Returns:

Type	Description
int	The total number of nodes calculated by summing the product of lengths of all combinations of item lists.

Notes

• The function first combines the lengths of item lists from both dictionaries.
• It then calculates the sum of the products of all possible combinations of these lengths.

Source code in src/action_rules/action_rules.py

def count_max_nodes(self, stable_items_binding: dict, flexible_items_binding: dict) -> int:
    """
    Calculate the maximum number of nodes based on the given item bindings.

    This function takes two dictionaries, `stable_items_binding` and `flexible_items_binding`,
    which map attributes to lists of items. It calculates the total number of nodes by considering
    all possible combinations of the lengths of these item lists and summing the product of each combination.

    Parameters
    ----------
    stable_items_binding : dict
        A dictionary where keys are attributes and values are lists of stable items.
    flexible_items_binding : dict
        A dictionary where keys are attributes and values are lists of flexible items.

    Returns
    -------
    int
        The total number of nodes calculated by summing the product of lengths of all combinations of item lists.

    Notes
    -----
    - The function first combines the lengths of item lists from both dictionaries.
    - It then calculates the sum of the products of all possible combinations of these lengths.
    """
    import numpy

    values_in_attribute = []
    for items in list(stable_items_binding.values()) + list(flexible_items_binding.values()):
        values_in_attribute.append(len(items))

    sum_nodes = 0
    for i in range(len(values_in_attribute)):
        for comb in itertools.combinations(values_in_attribute, i + 1):
            sum_nodes += int(numpy.prod(comb))
    return sum_nodes

cross_validate(data, stable_attributes, flexible_attributes, target, target_undesired_state, target_desired_state, *, n_splits=5, stratify=True, strategies=None, metrics=None, k_fraction=0.2, ci_method='bootstrap', n_bootstrap=500, risk_lambda=1.96, confidence_level=0.95, random_state=None, n_bootstrap_oof=1000, bootstrap_design='cluster_fold', track_stability=True, use_sparse_matrix=False, scale_support_thresholds=True, compute_insample_baseline=False)

Run stratified K-fold cross-validation on the action-rule pipeline.

Each fold receives a fresh :class:ActionRules instance configured with the same hyperparameters and utility tables as self. Per fold, rules are mined on the train split, confidence intervals are computed on the train split, and every discovered rule is re-scored on the held-out test split (test_uplift, test_realistic_gain). Targeting metrics (uplift@k, Qini, AUUC, profit@k) are evaluated under several targeting strategies on the test split.

Parameters:

Name	Type	Description	Default
data	DataFrame	The full dataset, pre-encoding.	required
stable_attributes	list of str		required
flexible_attributes	list of str		required
target	str		required
target_undesired_state	str		required
target_desired_state	str		required
n_splits	int	Number of folds (default 5). Must be ≥ 2.	5
stratify	bool	Whether to stratify folds by target value (default True).	True
strategies	sequence of str	Subset of ('point', 'lower', 'lower_positive', 'risk_adjusted'). Defaults to all four.	None
metrics	sequence of str	Subset of ('uplift_at_k', 'qini', 'auuc', 'profit_at_k'). Defaults to all four.	None
k_fraction	float	Top-k cutoff used by the *_at_k metrics (default 0.2).	0.2
ci_method	forwarded to	:class:~action_rules.evaluation.cv.CrossValidator.	'bootstrap'
n_bootstrap	forwarded to	:class:~action_rules.evaluation.cv.CrossValidator.	'bootstrap'
confidence_level	forwarded to	:class:~action_rules.evaluation.cv.CrossValidator.	'bootstrap'
risk_lambda	forwarded to	:class:~action_rules.evaluation.cv.CrossValidator.	'bootstrap'
random_state	int	Seed for fold splitting and bootstrap CIs.	None
n_bootstrap_oof	int	Bootstrap replicates for the across-fold rule-resampling CI. Set to 0 to disable bootstrap CIs.	1000
bootstrap_design	str	'cluster_fold' (default) resamples rules within each fold, computes the metric per fold, and averages — so the bootstrap CI estimates the same fold-mean quantity as the mean column. 'oof_pool' (legacy) resamples within fold then concatenates into one pool before computing the metric; estimates a pool-level statistic that differs from the fold mean by roughly a factor of K.	'cluster_fold'
track_stability	bool	Compute pairwise Jaccard overlap of discovered rule sets across folds (default True).	True
compute_insample_baseline	bool	When True, additionally mine on the full dataset and score on the full dataset to compute an apparent (in-sample) performance baseline; the result is stored on CrossValidationResult.insample_summary. Default False preserves the existing return shape.	False

Returns:

Type	Description
CrossValidationResult

Notes

• Naive K-fold CIs based on mean ± 1.96·std/√K over folds have below-nominal coverage (Bates, Hastie & Tibshirani, 2021, arXiv:2104.00673). This method therefore reports fold spread (std) as a stability indicator and a stratified bootstrap CI over OOF rule records as the inferential interval.
• Calling :meth:cross_validate does not require the model to be fitted on the full data first. It does not mutate self; each fold operates on a fresh internal instance.

Source code in src/action_rules/action_rules.py

def cross_validate(
    self,
    data,
    stable_attributes: list,
    flexible_attributes: list,
    target: str,
    target_undesired_state: str,
    target_desired_state: str,
    *,
    n_splits: int = 5,
    stratify: bool = True,
    strategies=None,
    metrics=None,
    k_fraction: float = 0.2,
    ci_method: str = 'bootstrap',
    n_bootstrap: int = 500,
    risk_lambda: float = 1.96,
    confidence_level: float = 0.95,
    random_state: Optional[int] = None,
    n_bootstrap_oof: int = 1000,
    bootstrap_design: str = 'cluster_fold',
    track_stability: bool = True,
    use_sparse_matrix: bool = False,
    scale_support_thresholds: bool = True,
    compute_insample_baseline: bool = False,
):
    """Run stratified K-fold cross-validation on the action-rule pipeline.

    Each fold receives a fresh :class:`ActionRules` instance configured
    with the same hyperparameters and utility tables as ``self``.  Per
    fold, rules are mined on the train split, confidence intervals are
    computed on the train split, and every discovered rule is re-scored
    on the held-out test split (``test_uplift``, ``test_realistic_gain``).
    Targeting metrics (``uplift@k``, Qini, AUUC, ``profit@k``) are
    evaluated under several targeting strategies on the test split.

    Parameters
    ----------
    data : pandas.DataFrame
        The full dataset, pre-encoding.
    stable_attributes, flexible_attributes : list of str
    target : str
    target_undesired_state, target_desired_state : str
    n_splits : int, optional
        Number of folds (default ``5``).  Must be ≥ 2.
    stratify : bool, optional
        Whether to stratify folds by ``target`` value (default ``True``).
    strategies : sequence of str, optional
        Subset of ``('point', 'lower', 'lower_positive', 'risk_adjusted')``.
        Defaults to all four.
    metrics : sequence of str, optional
        Subset of ``('uplift_at_k', 'qini', 'auuc', 'profit_at_k')``.
        Defaults to all four.
    k_fraction : float, optional
        Top-k cutoff used by the ``*_at_k`` metrics (default ``0.2``).
    ci_method, n_bootstrap, confidence_level, risk_lambda : forwarded to
        :class:`~action_rules.evaluation.cv.CrossValidator`.
    random_state : int, optional
        Seed for fold splitting and bootstrap CIs.
    n_bootstrap_oof : int, optional
        Bootstrap replicates for the across-fold rule-resampling CI.
        Set to ``0`` to disable bootstrap CIs.
    bootstrap_design : str, optional
        ``'cluster_fold'`` (default) resamples rules within each fold,
        computes the metric per fold, and averages — so the bootstrap CI
        estimates the same fold-mean quantity as the ``mean`` column.
        ``'oof_pool'`` (legacy) resamples within fold then concatenates
        into one pool before computing the metric; estimates a pool-level
        statistic that differs from the fold mean by roughly a factor of
        K.
    track_stability : bool, optional
        Compute pairwise Jaccard overlap of discovered rule sets across
        folds (default ``True``).
    compute_insample_baseline : bool, optional
        When ``True``, additionally mine on the full dataset and score on
        the full dataset to compute an apparent (in-sample) performance
        baseline; the result is stored on
        ``CrossValidationResult.insample_summary``.  Default ``False``
        preserves the existing return shape.

    Returns
    -------
    action_rules.evaluation.cv.CrossValidationResult

    Notes
    -----
    - Naive K-fold CIs based on ``mean ± 1.96·std/√K`` over folds have
      below-nominal coverage (Bates, Hastie & Tibshirani, 2021,
      arXiv:2104.00673).  This method therefore reports fold spread
      (``std``) as a stability indicator and a stratified bootstrap CI
      over OOF rule records as the inferential interval.
    - Calling :meth:`cross_validate` does **not** require the model to
      be fitted on the full data first.  It does not mutate ``self``;
      each fold operates on a fresh internal instance.
    """
    from .evaluation.cv import METRICS, STRATEGIES, CrossValidator

    # Snapshot the hyperparameters needed to build pristine per-fold instances.
    min_stable_attributes = self.min_stable_attributes
    min_flexible_attributes = self.min_flexible_attributes
    # Support thresholds are absolute counts.  When mining on a train fold
    # that is ``(n_splits-1)/n_splits`` of the full data, scale them down
    # proportionally so the same prevalence requirements apply on each fold.
    scale = (n_splits - 1) / n_splits if scale_support_thresholds else 1.0
    min_undesired_support = max(1, int(round(self.min_undesired_support * scale)))
    min_desired_support = max(1, int(round(self.min_desired_support * scale)))
    min_undesired_confidence = self.min_undesired_confidence
    min_desired_confidence = self.min_desired_confidence
    verbose = self.verbose
    intrinsic = self._original_intrinsic_utility_table or dict(self.intrinsic_utility_table)
    transition = self._original_transition_utility_table or dict(self.transition_utility_table)

    def _factory():
        return ActionRules(
            min_stable_attributes=min_stable_attributes,
            min_flexible_attributes=min_flexible_attributes,
            min_undesired_support=min_undesired_support,
            min_undesired_confidence=min_undesired_confidence,
            min_desired_support=min_desired_support,
            min_desired_confidence=min_desired_confidence,
            verbose=verbose,
            intrinsic_utility_table=intrinsic or None,
            transition_utility_table=transition or None,
        )

    validator = CrossValidator(
        _factory,
        stable_attributes=stable_attributes,
        flexible_attributes=flexible_attributes,
        target=target,
        target_undesired_state=target_undesired_state,
        target_desired_state=target_desired_state,
        n_splits=n_splits,
        stratify=stratify,
        intrinsic_utility_table=intrinsic,
        transition_utility_table=transition,
        strategies=STRATEGIES if strategies is None else strategies,
        metrics=METRICS if metrics is None else metrics,
        k_fraction=k_fraction,
        ci_method=ci_method,
        n_bootstrap=n_bootstrap,
        risk_lambda=risk_lambda,
        confidence_level=confidence_level,
        random_state=random_state,
        n_bootstrap_oof=n_bootstrap_oof,
        bootstrap_design=bootstrap_design,
        track_stability=track_stability,
        use_sparse_matrix=use_sparse_matrix,
        compute_insample_baseline=compute_insample_baseline,
    )
    return validator.run(data)

df_to_array(df)

Convert a one-hot DataFrame to a binary array.

Parameters:

Name	Type	Description	Default
df	Union[DataFrame, DataFrame]	The DataFrame to convert.	required

Returns:

Type	Description
tuple	A tuple containing the transposed array and the DataFrame columns.

Notes

The data is converted to an unsigned 8-bit array (np.uint8), backed by NumPy or CuPy depending on the selected cpu/gpu backend.

Source code in src/action_rules/action_rules.py

def df_to_array(self, df: Union['cudf.DataFrame', 'pandas.DataFrame']) -> tuple:
    """
    Convert a one-hot DataFrame to a binary array.

    Parameters
    ----------
    df : Union[cudf.DataFrame, pandas.DataFrame]
        The DataFrame to convert.

    Returns
    -------
    tuple
        A tuple containing the transposed array and the DataFrame columns.

    Notes
    -----
    The data is converted to an unsigned 8-bit array (`np.uint8`), backed by
    NumPy or CuPy depending on the selected cpu/gpu backend.
    """
    columns = list(df.columns)
    if self.is_gpu_np:
        data = self.np.asarray(df.values, dtype=self.np.uint8).T  # type: ignore
    elif self.is_gpu_pd:
        data = df.to_numpy().T  # type: ignore
    else:
        data = df.to_numpy(dtype=self.np.uint8).T  # type: ignore
    return data, columns

fit(data, stable_attributes, flexible_attributes, target, target_undesired_state, target_desired_state, use_sparse_matrix=False, use_gpu=False, **kwargs)

Generate action rules for the provided dataset.

Parameters:

Name	Type	Description	Default
data	Union[DataFrame, DataFrame]	The dataset to generate action rules from.	required
stable_attributes	list	List of stable attributes.	required
flexible_attributes	list	List of flexible attributes.	required
target	str	The target attribute.	required
target_undesired_state	str	The undesired state of the target attribute.	required
target_desired_state	str	The desired state of the target attribute.	required
use_sparse_matrix	bool	Kept for backward compatibility with action-rules <= 1.0.11. The bitset backend supersedes sparse matrices, so this flag is accepted and ignored. Other unrecognized keyword arguments (**kwargs) are likewise accepted and ignored for backward compatibility with older call signatures.	False
use_gpu	bool	Use GPU (cuDF) for data processing if available. Default is False.	False

Raises:

Type	Description
RuntimeError	If the model has already been fitted.

Notes

The method runs one-hot encoding (when needed), packs bit masks, explores candidate branches, prunes classification rules by depth, and finally materializes action rules.

Source code in src/action_rules/action_rules.py

def fit(
    self,
    data: Union['cudf.DataFrame', 'pandas.DataFrame'],
    stable_attributes: list,
    flexible_attributes: list,
    target: str,
    target_undesired_state: str,
    target_desired_state: str,
    use_sparse_matrix: bool = False,
    use_gpu: bool = False,
    **kwargs,
):
    """
    Generate action rules for the provided dataset.

    Parameters
    ----------
    data : Union[cudf.DataFrame, pandas.DataFrame]
        The dataset to generate action rules from.
    stable_attributes : list
        List of stable attributes.
    flexible_attributes : list
        List of flexible attributes.
    target : str
        The target attribute.
    target_undesired_state : str
        The undesired state of the target attribute.
    target_desired_state : str
        The desired state of the target attribute.
    use_sparse_matrix : bool, optional
        Kept for backward compatibility with action-rules <= 1.0.11. The bitset
        backend supersedes sparse matrices, so this flag is accepted and ignored.
        Other unrecognized keyword arguments (``**kwargs``) are likewise accepted
        and ignored for backward compatibility with older call signatures.
    use_gpu : bool, optional
        Use GPU (cuDF) for data processing if available. Default is False.

    Raises
    ------
    RuntimeError
        If the model has already been fitted.

    Notes
    -----
    The method runs one-hot encoding (when needed), packs bit masks, explores
    candidate branches, prunes classification rules by depth, and finally
    materializes action rules.
    """
    if self.output is not None:
        raise RuntimeError("The model is already fit.")
    if use_sparse_matrix:
        warnings.warn(
            "The 'use_sparse_matrix' parameter is obsolete and has no effect: action-rules now "
            "always uses the packed-bitset backend.",
            UserWarning,
            stacklevel=2,
        )
    # Forward tolerance: legacy callers passed use_gpu="auto" for backend
    # autoselection. That harness lives outside the package now, so treat any
    # truthy string as a plain GPU request instead of raising.
    if isinstance(use_gpu, str):
        use_gpu = use_gpu.strip().lower() not in ("", "false", "cpu", "no", "0")

    # reset cached bitset structures before fitting a new model
    self.bit_masks = None
    self.target_state_bit_masks = None
    self.frames_bit_masks = None
    self.set_array_library(use_gpu, data)
    if not self.is_onehot:
        data = self.one_hot_encode(data, stable_attributes, flexible_attributes, target)
    data, columns = self.df_to_array(data)

    stable_items_binding, flexible_items_binding, target_items_binding, column_values = self.get_bindings(
        columns, stable_attributes, flexible_attributes, target
    )

    # Preserve original string-keyed tables before remapping to integer indices.
    # confidence_intervals() needs the originals to pass to inference engines.
    self._original_intrinsic_utility_table = dict(self.intrinsic_utility_table)
    self._original_transition_utility_table = dict(self.transition_utility_table)
    self._column_values = column_values
    self.intrinsic_utility_table, self.transition_utility_table = self.remap_utility_tables(column_values)

    local_bit_masks = self.build_bit_masks(data)
    self._cache_bitset_structures(local_bit_masks, target_items_binding, target)
    self.frames_bit_masks = self.get_split_bit_masks(target_items_binding, target)

    if self.verbose:
        print('Maximum number of nodes to check for support:')
        print('_____________________________________________')
        print(self.count_max_nodes(stable_items_binding, flexible_items_binding))
        print('')
    use_gpu_batching = bool(self.is_gpu_np and self.bit_masks is not None and self.frames_bit_masks)

    # Set membership is hot in candidate pruning; use a set internally for O(1) lookups.
    stop_list = set(self.get_stop_list(stable_items_binding, flexible_items_binding))
    undesired_state = columns.index(target + '_<item_target>_' + str(target_undesired_state))
    desired_state = columns.index(target + '_<item_target>_' + str(target_desired_state))

    stop_list_itemset = set()  # type: set

    initial_candidate = {
        'ar_prefix': tuple(),
        'itemset_prefix': tuple(),
        'stable_items_binding': stable_items_binding,
        'flexible_items_binding': flexible_items_binding,
        'actionable_attributes': 0,
    }
    candidates_pool = deque([initial_candidate])
    pending_depth_counts = {0: 1}
    min_pending_depth: Optional[int] = 0
    max_depth_seen = 0
    next_prune_depth = 1
    self.rules = Rules(
        undesired_state,
        desired_state,
        columns,
        data.shape[1],
        self.intrinsic_utility_table,
        self.transition_utility_table,
    )
    candidate_generator = CandidateGenerator(
        frames_bit_masks=self.frames_bit_masks,
        bit_masks=self.bit_masks,
        min_stable_attributes=self.min_stable_attributes,
        min_flexible_attributes=self.min_flexible_attributes,
        min_undesired_support=self.min_undesired_support,
        min_desired_support=self.min_desired_support,
        min_undesired_confidence=self.min_undesired_confidence,
        min_desired_confidence=self.min_desired_confidence,
        undesired_state=undesired_state,
        desired_state=desired_state,
        rules=self.rules,
        verbose=self.verbose,
    )
    # Default GPU node batch; the adaptive VRAM budgeting in CandidateGenerator
    # shrinks this automatically to fit available device memory.
    effective_gpu_node_batch_size = 32

    def pop_next_candidate() -> dict:
        """Pop one pending candidate and keep pending-depth bookkeeping in sync."""
        nonlocal min_pending_depth
        candidate_to_expand = candidates_pool.popleft()
        depth = len(candidate_to_expand['ar_prefix'])
        pending_depth_counts[depth] -= 1
        if pending_depth_counts[depth] <= 0:
            pending_depth_counts.pop(depth, None)
            if depth == min_pending_depth:
                min_pending_depth = min(pending_depth_counts.keys(), default=None)
        return candidate_to_expand

    while len(candidates_pool) > 0:
        if use_gpu_batching:  # pragma: no cover
            batch: list = []
            while candidates_pool and len(batch) < effective_gpu_node_batch_size:
                batch.append(pop_next_candidate())
            new_candidates = candidate_generator.generate_candidates_batch(
                batch,
                stop_list=stop_list,
                stop_list_itemset=stop_list_itemset,
                batch_size=effective_gpu_node_batch_size,
            )
        else:
            candidate = pop_next_candidate()
            new_candidates = candidate_generator.generate_candidates(
                **candidate,
                stop_list=stop_list,
                stop_list_itemset=stop_list_itemset,
            )
        if new_candidates:
            candidates_pool.extend(new_candidates)
            for new_candidate in new_candidates:
                new_depth = len(new_candidate['ar_prefix'])
                pending_depth_counts[new_depth] = pending_depth_counts.get(new_depth, 0) + 1
                if min_pending_depth is None or new_depth < min_pending_depth:
                    min_pending_depth = new_depth
                if new_depth > max_depth_seen:
                    max_depth_seen = new_depth
        while next_prune_depth <= max_depth_seen and (
            min_pending_depth is None or min_pending_depth >= next_prune_depth
        ):
            self.rules.prune_classification_rules(next_prune_depth, stop_list)
            next_prune_depth += 1
    self.rules.generate_action_rules()
    self.output = Output(
        self.rules.action_rules, target, stable_items_binding, flexible_items_binding, column_values
    )
    del data
    if self.is_gpu_np:  # pragma: no cover
        gpu_pool = self.np.get_default_memory_pool()  # type: ignore[union-attr, attr-defined]
        gpu_pool.free_all_blocks()

fit_onehot(data, stable_attributes, flexible_attributes, target, target_undesired_state, target_desired_state, use_sparse_matrix=False, use_gpu=False, **kwargs)

Fit the model when input data is already one-hot encoded.

The method remaps one-hot columns to the internal naming convention (_<item_stable>_, _<item_flexible>_, _<item_target>_), drops unrelated columns, and forwards execution to fit.

Parameters:

Name	Type	Description	Default
data	Union[DataFrame, DataFrame]	The dataset to be processed and used for fitting the model.	required
stable_attributes	dict	A dictionary mapping stable attribute names to lists of column names corresponding to those attributes.	required
flexible_attributes	dict	A dictionary mapping flexible attribute names to lists of column names corresponding to those attributes.	required
target	dict	A dictionary mapping the target attribute name to a list of column names corresponding to that attribute.	required
target_undesired_state	str	The undesired state of the target attribute, used in action rule generation.	required
target_desired_state	str	The desired state of the target attribute, used in action rule generation.	required
use_sparse_matrix	bool	Kept for backward compatibility with action-rules <= 1.0.11. The bitset backend supersedes sparse matrices, so this flag is accepted and ignored. Other unrecognized keyword arguments (**kwargs) are likewise accepted and ignored for backward compatibility with older call signatures.	False
use_gpu	bool	If True, the GPU (cuDF) is used for data processing if available. Default is False.	False

Notes

This method expects boolean/binary one-hot columns.

Source code in src/action_rules/action_rules.py

def fit_onehot(
    self,
    data: Union['cudf.DataFrame', 'pandas.DataFrame'],
    stable_attributes: dict,
    flexible_attributes: dict,
    target: dict,
    target_undesired_state: str,
    target_desired_state: str,
    use_sparse_matrix: bool = False,
    use_gpu: bool = False,
    **kwargs,
):
    """
    Fit the model when input data is already one-hot encoded.

    The method remaps one-hot columns to the internal naming convention
    (`_<item_stable>_`, `_<item_flexible>_`, `_<item_target>_`), drops
    unrelated columns, and forwards execution to `fit`.

    Parameters
    ----------
    data : Union[cudf.DataFrame, pandas.DataFrame]
        The dataset to be processed and used for fitting the model.
    stable_attributes : dict
        A dictionary mapping stable attribute names to lists of column
        names corresponding to those attributes.
    flexible_attributes : dict
        A dictionary mapping flexible attribute names to lists of column
        names corresponding to those attributes.
    target : dict
        A dictionary mapping the target attribute name to a list of
        column names corresponding to that attribute.
    target_undesired_state : str
        The undesired state of the target attribute, used in action rule generation.
    target_desired_state : str
        The desired state of the target attribute, used in action rule generation.
    use_sparse_matrix : bool, optional
        Kept for backward compatibility with action-rules <= 1.0.11. The bitset
        backend supersedes sparse matrices, so this flag is accepted and ignored.
        Other unrecognized keyword arguments (``**kwargs``) are likewise accepted
        and ignored for backward compatibility with older call signatures.
    use_gpu : bool, optional
        If True, the GPU (cuDF) is used for data processing if available.
        Default is False.

    Notes
    -----
    This method expects boolean/binary one-hot columns.
    """
    self.is_onehot = True
    data = data.copy()
    data = data.astype('bool')
    new_labels = []
    attributes_stable = set([])
    attribtes_flexible = set([])
    attribute_target = ''
    remove_cols = []
    for label in data.columns:
        to_remove = True
        for attribute, columns in stable_attributes.items():
            if label in columns:
                new_labels.append(attribute + '_<item_stable>_' + label)
                attributes_stable.add(attribute)
                to_remove = False
        for attribute, columns in flexible_attributes.items():
            if label in columns:
                new_labels.append(attribute + '_<item_flexible>_' + label)
                attribtes_flexible.add(attribute)
                to_remove = False
        for attribute, columns in target.items():
            if label in columns:
                new_labels.append(attribute + '_<item_target>_' + label)
                attribute_target = attribute
                to_remove = False
        if to_remove:
            new_labels.append(label)
            remove_cols.append(label)
    data.columns = new_labels
    data = data.drop(columns=remove_cols)
    self.fit(
        data,
        list(attributes_stable),
        list(attribtes_flexible),
        attribute_target,
        target_undesired_state,
        target_desired_state,
        use_sparse_matrix=use_sparse_matrix,
        use_gpu=use_gpu,
    )

get_bindings(columns, stable_attributes, flexible_attributes, target)

Bind stable/flexible/target attribute to corresponding column in the dataset.

Parameters:

Name	Type	Description	Default
columns	list	List of column names in the dataset.	required
stable_attributes	list	List of stable attributes.	required
flexible_attributes	list	List of flexible attributes.	required
target	str	The target attribute.	required

Returns:

Type	Description
tuple	A tuple containing the bindings for stable attributes, flexible attributes, and target items.

Notes

The method generates mappings from column indices to attribute values for stable, flexible, and target attributes.

Source code in src/action_rules/action_rules.py

def get_bindings(
    self,
    columns: list,
    stable_attributes: list,
    flexible_attributes: list,
    target: str,
) -> tuple:
    """
    Bind stable/flexible/target attribute to corresponding column in the dataset.

    Parameters
    ----------
    columns : list
        List of column names in the dataset.
    stable_attributes : list
        List of stable attributes.
    flexible_attributes : list
        List of flexible attributes.
    target : str
        The target attribute.

    Returns
    -------
    tuple
        A tuple containing the bindings for stable attributes, flexible attributes, and target items.

    Notes
    -----
    The method generates mappings from column indices to attribute values for stable, flexible, and target
    attributes.
    """
    stable_items_binding = defaultdict(lambda: [])
    flexible_items_binding = defaultdict(lambda: [])
    target_items_binding = defaultdict(lambda: [])
    column_values = {}

    for i, col in enumerate(columns):
        is_continue = False
        # stable
        for attribute in stable_attributes:
            if col.startswith(attribute + '_<item_stable>_'):
                stable_items_binding[attribute].append(i)
                column_values[i] = (attribute, col.split('_<item_stable>_', 1)[1])
                is_continue = True
                break
        if is_continue is True:
            continue
        # flexible
        for attribute in flexible_attributes:
            if col.startswith(attribute + '_<item_flexible>_'):
                flexible_items_binding[attribute].append(i)
                column_values[i] = (attribute, col.split('_<item_flexible>_', 1)[1])
                is_continue = True
                break
        if is_continue is True:
            continue
        # target
        if col.startswith(target + '_<item_target>_'):
            target_items_binding[target].append(i)
            column_values[i] = (target, col.split('_<item_target>_', 1)[1])
    return stable_items_binding, flexible_items_binding, target_items_binding, column_values

get_rules()

Return the generated action rules if available.

Raises:

Type	Description
RuntimeError	If the model has not been fitted.

Returns:

Type	Description
Output	The generated action rules.

Notes

This method returns the Output object containing the generated action rules.

Source code in src/action_rules/action_rules.py

def get_rules(self) -> Output:
    """
    Return the generated action rules if available.

    Raises
    ------
    RuntimeError
        If the model has not been fitted.

    Returns
    -------
    Output
        The generated action rules.

    Notes
    -----
    This method returns the `Output` object containing the generated action rules.
    """
    if self.output is None:
        raise RuntimeError("The model is not fit.")
    return self.output

get_split_bit_masks(target_items_binding, target)

Return packed bit-mask rows for each target state.

Parameters:

Name	Type	Description	Default
target_items_binding	dict	Indexes of target attributes columns in one-hot table.	required
target	str	Name of the target attribute.	required

Returns:

Type	Description
dict	Dictionary mapping target attributes to the corresponding packed mask rows.

Notes

Requires that build_bit_masks has been executed beforehand.

Source code in src/action_rules/action_rules.py

def get_split_bit_masks(self, target_items_binding: dict, target: str) -> dict:
    """
    Return packed bit-mask rows for each target state.

    Parameters
    ----------
    target_items_binding : dict
        Indexes of target attributes columns in one-hot table.
    target : str
        Name of the target attribute.

    Returns
    -------
    dict
        Dictionary mapping target attributes to the corresponding packed mask rows.

    Notes
    -----
    Requires that `build_bit_masks` has been executed beforehand.
    """
    if self.bit_masks is None:
        raise RuntimeError("Bit masks are not available. Ensure fit() was run first.")

    target_state_masks = {}
    for item_index in target_items_binding.get(target, []):
        target_state_masks[item_index] = self.bit_masks[item_index]
    return target_state_masks

get_stop_list(stable_items_binding, flexible_items_binding)

Generate a stop list to prevent certain combinations of attributes.

Parameters:

Name	Type	Description	Default
stable_items_binding	dict	Dictionary containing bindings for stable items.	required
flexible_items_binding	dict	Dictionary containing bindings for flexible items.	required

Returns:

Type	Description
list	A list of stop combinations.

Notes

The stop list is generated by creating pairs of stable item indices and ensuring flexible items do not repeat.

Source code in src/action_rules/action_rules.py

def get_stop_list(self, stable_items_binding: dict, flexible_items_binding: dict) -> list:
    """
    Generate a stop list to prevent certain combinations of attributes.

    Parameters
    ----------
    stable_items_binding : dict
        Dictionary containing bindings for stable items.
    flexible_items_binding : dict
        Dictionary containing bindings for flexible items.

    Returns
    -------
    list
        A list of stop combinations.

    Notes
    -----
    The stop list is generated by creating pairs of stable item indices and ensuring flexible items do not repeat.
    """
    stop_list = []
    for items in stable_items_binding.values():
        for stop_couple in itertools.product(items, repeat=2):
            stop_list.append(tuple(stop_couple))
    for item in flexible_items_binding.keys():
        stop_list.append(tuple([item, item]))
    return stop_list

one_hot_encode(data, stable_attributes, flexible_attributes, target)

Perform one-hot encoding on the attributes of the DataFrame.

Parameters:

Name	Type	Description	Default
data	Union[DataFrame, DataFrame]	The input DataFrame containing the data to be encoded.	required
stable_attributes	list	List of stable attributes to be one-hot encoded.	required
flexible_attributes	list	List of flexible attributes to be one-hot encoded.	required
target	str	The target attribute to be one-hot encoded.	required

Returns:

Type	Description
Union[DataFrame, DataFrame]	A DataFrame with the specified attributes one-hot encoded.

Notes

Stable and flexible (antecedent) columns are cast to strings only for non-missing values; NaN is preserved so that pd.get_dummies skips it instead of creating a phantom <attr>_<item_*>_nan category. This implements the pessimistic interpretation of null values in incomplete information systems --- a missing antecedent does not match any value-specific itemset and therefore cannot appear in a discovered rule --- as defined for action-rule mining by Dardzinska, Action Rules Mining (Springer 2013, Section 2.3.2). The target column is cast to strings in full so that any NaN target value becomes its own explicit category (downstream get_split_bit_masks will exclude it from both the undesired and desired splits, which is the intended behaviour when callers want to ignore unlabelled rows).

Source code in src/action_rules/action_rules.py

def one_hot_encode(
    self,
    data: Union['cudf.DataFrame', 'pandas.DataFrame'],
    stable_attributes: list,
    flexible_attributes: list,
    target: str,
) -> Union['cudf.DataFrame', 'pandas.DataFrame']:
    """
    Perform one-hot encoding on the attributes of the DataFrame.

    Parameters
    ----------
    data : Union[cudf.DataFrame, pandas.DataFrame]
        The input DataFrame containing the data to be encoded.
    stable_attributes : list
        List of stable attributes to be one-hot encoded.
    flexible_attributes : list
        List of flexible attributes to be one-hot encoded.
    target : str
        The target attribute to be one-hot encoded.

    Returns
    -------
    Union[cudf.DataFrame, pandas.DataFrame]
        A DataFrame with the specified attributes one-hot encoded.

    Notes
    -----
    Stable and flexible (antecedent) columns are cast to strings only for non-missing values; ``NaN`` is
    preserved so that ``pd.get_dummies`` skips it instead of creating a phantom ``<attr>_<item_*>_nan``
    category.  This implements the *pessimistic* interpretation of null values in incomplete information
    systems --- a missing antecedent does not match any value-specific itemset and therefore cannot appear
    in a discovered rule --- as defined for action-rule mining by Dardzinska, *Action Rules Mining*
    (Springer 2013, Section 2.3.2).  The target column is cast to strings in full so that any ``NaN``
    target value becomes its own explicit category (downstream ``get_split_bit_masks`` will exclude it
    from both the undesired and desired splits, which is the intended behaviour when callers want to
    ignore unlabelled rows).
    """

    def _prepare_antecedent_frame(frame, attributes):
        """Stringify non-missing antecedent cells while keeping ``NaN`` as ``NaN``.

        Letting ``get_dummies`` see a real ``NaN`` is the documented way to make it skip the value;
        calling ``astype(str)`` first would convert ``np.nan`` into the literal string ``'nan'`` and
        spawn a spurious one-hot column.
        """
        antecedent = frame[attributes].copy()
        return antecedent.where(antecedent.isna(), antecedent.astype(str))

    to_concat = []
    if len(stable_attributes) > 0:
        stable_frame = _prepare_antecedent_frame(data, stable_attributes)
        data_stable = self.pd.get_dummies(stable_frame, sparse=False, prefix_sep='_<item_stable>_')  # type: ignore
        to_concat.append(data_stable)
    if len(flexible_attributes) > 0:
        flexible_frame = _prepare_antecedent_frame(data, flexible_attributes)
        data_flexible = self.pd.get_dummies(  # type: ignore
            flexible_frame, sparse=False, prefix_sep='_<item_flexible>_'
        )
        to_concat.append(data_flexible)
    data_target = self.pd.get_dummies(  # type: ignore
        data[[target]].astype(str), sparse=False, prefix_sep='_<item_target>_'
    )
    to_concat.append(data_target)
    data = self.pd.concat(to_concat, axis=1)  # type: ignore
    return data

predict(frame_row)

Predict recommended actions based on the provided row of data.

This method applies the fitted action rules to the given row of data and generates a DataFrame with recommended actions if any of the action rules are triggered.

Parameters:

Name	Type	Description	Default
frame_row	Union[Series, Series]	A row of data in the form of a cuDF or pandas Series. The Series should contain the features required by the action rules.	required

Returns:

Type

Description

Union[DataFrame, DataFrame]

A DataFrame with the recommended actions. The DataFrame includes the following columns: - The original attributes with recommended changes. - 'ActionRules_RuleIndex': Index of the action rule applied. - 'ActionRules_UndesiredSupport': Support of the undesired part of the rule. - 'ActionRules_DesiredSupport': Support of the desired part of the rule. - 'ActionRules_UndesiredConfidence': Confidence of the undesired part of the rule. - 'ActionRules_DesiredConfidence': Confidence of the desired part of the rule. - 'ActionRules_Uplift': Uplift value of the rule.

Raises:

Type	Description
RuntimeError	If the model has not been fitted.

Notes

The method compares the given row of data against the undesired itemsets of the action rules. If a match is found, it applies the desired itemset changes and records the action rule's metadata. The result is a DataFrame with one or more rows representing the recommended actions for the given data.

Source code in src/action_rules/action_rules.py

def predict(self, frame_row: Union['cudf.Series', 'pandas.Series']) -> Union['cudf.DataFrame', 'pandas.DataFrame']:
    """
    Predict recommended actions based on the provided row of data.

    This method applies the fitted action rules to the given row of data and generates
    a DataFrame with recommended actions if any of the action rules are triggered.

    Parameters
    ----------
    frame_row : Union['cudf.Series', 'pandas.Series']
        A row of data in the form of a cuDF or pandas Series. The Series should
        contain the features required by the action rules.

    Returns
    -------
    Union['cudf.DataFrame', 'pandas.DataFrame']
        A DataFrame with the recommended actions. The DataFrame includes the following columns:
        - The original attributes with recommended changes.
        - 'ActionRules_RuleIndex': Index of the action rule applied.
        - 'ActionRules_UndesiredSupport': Support of the undesired part of the rule.
        - 'ActionRules_DesiredSupport': Support of the desired part of the rule.
        - 'ActionRules_UndesiredConfidence': Confidence of the undesired part of the rule.
        - 'ActionRules_DesiredConfidence': Confidence of the desired part of the rule.
        - 'ActionRules_Uplift': Uplift value of the rule.

    Raises
    ------
    RuntimeError
        If the model has not been fitted.

    Notes
    -----
    The method compares the given row of data against the undesired itemsets of the action rules.
    If a match is found, it applies the desired itemset changes and records the action rule's
    metadata. The result is a DataFrame with one or more rows representing the recommended actions
    for the given data.
    """
    if self.output is None:
        raise RuntimeError("The model is not fit.")
    index_value_tuples = list(zip(frame_row.index, frame_row))
    values = []
    column_values = self.output.column_values
    for index_value_tuple in index_value_tuples:
        values.append(list(column_values.keys())[list(column_values.values()).index(index_value_tuple)])
    new_values = tuple(values)
    predicted = []
    for i, action_rule in enumerate(self.output.action_rules):
        if set(action_rule['undesired']['itemset']) <= set(new_values):
            predicted_row = frame_row.copy()
            for recommended in set(action_rule['desired']['itemset']) - set(new_values):
                attribute, value = column_values[recommended]
                predicted_row[attribute + ' (Recommended)'] = value
            predicted_row['ActionRules_RuleIndex'] = i
            predicted_row['ActionRules_UndesiredSupport'] = action_rule['undesired']['support']
            predicted_row['ActionRules_DesiredSupport'] = action_rule['desired']['support']
            predicted_row['ActionRules_UndesiredConfidence'] = action_rule['undesired']['confidence']
            predicted_row['ActionRules_DesiredConfidence'] = action_rule['desired']['confidence']
            predicted_row['ActionRules_Uplift'] = action_rule['uplift']
            predicted.append(predicted_row)
    return self.pd.DataFrame(predicted)  # type: ignore

remap_utility_tables(column_values)

Remap the keys of intrinsic and transition utility tables using the provided column mapping.

The function uses column_values, a dictionary mapping internal column indices to (attribute, value) tuples, to invert the mapping so that utility table keys are replaced with the corresponding integer index (for intrinsic utilities) or a tuple of integer indices (for transition utilities).

Parameters:

Name	Type	Description	Default
column_values	dict	Dictionary mapping integer column indices to (attribute, value) pairs. Example: {0: ('Age', 'O'), 1: ('Age', 'Y'), 2: ('Sex', 'F'), ...}	required

Returns:

Type	Description
tuple	A tuple (remapped_intrinsic, remapped_transition) where: - remapped_intrinsic is a dict mapping integer column index to utility value. - remapped_transition is a dict mapping (from_index, to_index) to utility value.

Notes

• The method performs case-insensitive matching by converting attribute names and values to lowercase.
• If a key in a utility table does not have a corresponding entry in column_values, it is skipped.

Source code in src/action_rules/action_rules.py

def remap_utility_tables(self, column_values):
    """
    Remap the keys of intrinsic and transition utility tables using the provided column mapping.

    The function uses `column_values`, a dictionary mapping internal column indices to
    (attribute, value) tuples, to invert the mapping so that utility table keys are replaced
    with the corresponding integer index (for intrinsic utilities) or a tuple of integer indices
    (for transition utilities).

    Parameters
    ----------
    column_values : dict
        Dictionary mapping integer column indices to (attribute, value) pairs.
        Example: {0: ('Age', 'O'), 1: ('Age', 'Y'), 2: ('Sex', 'F'), ...}

    Returns
    -------
    tuple
        A tuple (remapped_intrinsic, remapped_transition) where:
          - remapped_intrinsic is a dict mapping integer column index to utility value.
          - remapped_transition is a dict mapping (from_index, to_index) to utility value.

    Notes
    -----
    - The method performs case-insensitive matching by converting attribute names and values to lowercase.
    - If a key in a utility table does not have a corresponding entry in column_values, it is skipped.
    """
    # Invert column_values to map (attribute.lower(), value.lower()) -> column index.
    inv_map = {(attr.lower(), val.lower()): idx for idx, (attr, val) in column_values.items()}

    remapped_intrinsic = {}
    # Remap intrinsic utility table keys: ('Attribute', 'Value') -> utility
    for key, utility in self.intrinsic_utility_table.items():
        # Normalize key to lowercase
        attr, val = key
        lookup_key = (attr.lower(), val.lower())
        # Look up the corresponding column index; if not found, skip this key.
        if lookup_key in inv_map:
            col_index = inv_map[lookup_key]
            remapped_intrinsic[col_index] = utility
        # Else: optionally, one could log or warn about a missing mapping.

    remapped_transition = {}
    # Remap transition utility table keys: ('Attribute', from_value, to_value) -> utility
    for key, utility in self.transition_utility_table.items():
        attr, from_val, to_val = key
        lookup_from = (attr.lower(), from_val.lower())
        lookup_to = (attr.lower(), to_val.lower())
        # Only remap if both the from and to values exist in inv_map.
        if lookup_from in inv_map and lookup_to in inv_map:
            from_index = inv_map[lookup_from]
            to_index = inv_map[lookup_to]
            remapped_transition[(from_index, to_index)] = utility
        # Else: skip or log missing mapping.

    return remapped_intrinsic, remapped_transition

set_array_library(use_gpu, df)

Set the appropriate array and DataFrame libraries (cuDF or pandas) based on the user's preference.

Parameters:

Name	Type	Description	Default
use_gpu	bool	Indicates whether to use GPU (cuDF) for data processing if available.	required
df	Union[DataFrame, DataFrame]	The DataFrame to convert.	required

Raises:

Type	Description
ImportError	If use_gpu is True but cuDF is not available and pandas cannot be imported as fallback.

Warnings

UserWarning If use_gpu is True but cuDF is not available, a warning is issued indicating fallback to pandas.

Notes

This method determines whether to use GPU-accelerated libraries for processing data, falling back to CPU-based libraries if necessary.

Source code in src/action_rules/action_rules.py

def set_array_library(self, use_gpu: bool, df: Union['cudf.DataFrame', 'pandas.DataFrame']):
    """
    Set the appropriate array and DataFrame libraries (cuDF or pandas) based on the user's preference.

    Parameters
    ----------
    use_gpu : bool
        Indicates whether to use GPU (cuDF) for data processing if available.
    df : Union[cudf.DataFrame, pandas.DataFrame]
        The DataFrame to convert.

    Raises
    ------
    ImportError
        If `use_gpu` is True but cuDF is not available and pandas cannot be imported as fallback.

    Warnings
    --------
    UserWarning
        If `use_gpu` is True but cuDF is not available, a warning is issued indicating fallback to pandas.

    Notes
    -----
    This method determines whether to use GPU-accelerated libraries for processing data, falling back to CPU-based
    libraries if necessary.
    """
    if use_gpu:
        try:
            import cupy as np

            is_gpu_np = True
        except ImportError:
            warnings.warn("CuPy is not available. Falling back to Numpy.")
            import numpy as np

            is_gpu_np = False
    else:
        import numpy as np

        is_gpu_np = False

    df_library_imported = False
    try:
        import pandas as pd

        if isinstance(df, pd.DataFrame):
            is_gpu_pd = False
            df_library_imported = True
    except ImportError:
        df_library_imported = False

    if not df_library_imported:
        try:
            import cudf as pd

            if isinstance(df, pd.DataFrame):
                is_gpu_pd = True
                df_library_imported = True
        except ImportError:
            df_library_imported = False

    if not df_library_imported:
        raise ImportError('Just Pandas or cuDF dataframes are supported.')

    self.np = np
    self.pd = pd
    self.is_gpu_np = is_gpu_np
    self.is_gpu_pd = is_gpu_pd

ConfidenceIntervalResult dataclass

Confidence interval result for a single action rule.

Parameters:

Name	Type	Description	Default
rule_index	int	Zero-based index of the action rule in the source Output.action_rules.	required
method	str	Name of the CI method used: 'bootstrap', 'analytic', or 'bayesian'.	required
confidence_level	float	Nominal coverage probability, e.g. 0.95.	required
uplift_point	float	Point estimate of the uplift measure.	required
uplift_ci_lower	float	Lower bound of the uplift confidence interval.	required
uplift_ci_upper	float	Upper bound of the uplift confidence interval.	required
uplift_se	float	Standard error of the uplift estimate.	required
realistic_rule_gain_point	float	Point estimate of the realistic rule gain (only when utility tables are given).	None
realistic_rule_gain_ci_lower	float	Lower bound of the realistic rule gain CI.	None
realistic_rule_gain_ci_upper	float	Upper bound of the realistic rule gain CI.	None
realistic_rule_gain_se	float	Standard error of the realistic rule gain.	None
support	int	Transaction support of the action rule.	0
confidence	float	Confidence of the action rule.	0.0
category	RuleCategory	Qualitative verdict derived from the uplift CI vs. a threshold.	None
undefined_bootstrap_fraction	float	For the bootstrap engine: fraction of resamples that hit a degenerate configuration (n_u_ante == 0 or n_d_ante == 0) and were therefore dropped from the resulting percentile interval. 0.0 is the well-behaved case. Set to None for analytic / Bayesian engines.	None
samples_uplift	ndarray	Raw bootstrap/posterior samples for uplift (excluded from repr to keep it brief).	None
samples_gain	ndarray	Raw bootstrap/posterior samples for gain (excluded from repr to keep it brief).	None

Source code in src/action_rules/inference/base.py

@dataclass
class ConfidenceIntervalResult:
    """Confidence interval result for a single action rule.

    Parameters
    ----------
    rule_index : int
        Zero-based index of the action rule in the source ``Output.action_rules``.
    method : str
        Name of the CI method used: ``'bootstrap'``, ``'analytic'``, or ``'bayesian'``.
    confidence_level : float
        Nominal coverage probability, e.g. ``0.95``.
    uplift_point : float
        Point estimate of the uplift measure.
    uplift_ci_lower : float
        Lower bound of the uplift confidence interval.
    uplift_ci_upper : float
        Upper bound of the uplift confidence interval.
    uplift_se : float
        Standard error of the uplift estimate.
    realistic_rule_gain_point : float, optional
        Point estimate of the realistic rule gain (only when utility tables are given).
    realistic_rule_gain_ci_lower : float, optional
        Lower bound of the realistic rule gain CI.
    realistic_rule_gain_ci_upper : float, optional
        Upper bound of the realistic rule gain CI.
    realistic_rule_gain_se : float, optional
        Standard error of the realistic rule gain.
    support : int
        Transaction support of the action rule.
    confidence : float
        Confidence of the action rule.
    category : RuleCategory, optional
        Qualitative verdict derived from the uplift CI vs. a threshold.
    undefined_bootstrap_fraction : float, optional
        For the bootstrap engine: fraction of resamples that hit a degenerate
        configuration (``n_u_ante == 0`` or ``n_d_ante == 0``) and were therefore
        dropped from the resulting percentile interval.  ``0.0`` is the
        well-behaved case.  Set to ``None`` for analytic / Bayesian engines.
    samples_uplift : np.ndarray, optional
        Raw bootstrap/posterior samples for uplift (excluded from repr to keep it brief).
    samples_gain : np.ndarray, optional
        Raw bootstrap/posterior samples for gain (excluded from repr to keep it brief).
    """

    rule_index: int
    method: str
    confidence_level: float
    uplift_point: float
    uplift_ci_lower: float
    uplift_ci_upper: float
    uplift_se: float
    realistic_rule_gain_point: Optional[float] = None
    realistic_rule_gain_ci_lower: Optional[float] = None
    realistic_rule_gain_ci_upper: Optional[float] = None
    realistic_rule_gain_se: Optional[float] = None
    support: int = 0
    confidence: float = 0.0
    category: Optional[RuleCategory] = None
    undefined_bootstrap_fraction: Optional[float] = None
    samples_uplift: Optional[np.ndarray] = field(default=None, repr=False)
    samples_gain: Optional[np.ndarray] = field(default=None, repr=False)

    def to_dict(self, include_samples: bool = False) -> dict:
        """Return a plain dict representation of this result.

        Parameters
        ----------
        include_samples : bool
            When ``False`` (default) the ``samples_uplift`` and ``samples_gain``
            arrays are omitted to keep the dict compact and JSON-friendly.
            When ``True`` the arrays are converted to lists.

        Returns
        -------
        dict
            Keys mirror the dataclass fields.  ``category`` (if present) is
            converted to its ``RuleCategory.value`` string.
        """
        out: dict = {
            'rule_index': self.rule_index,
            'method': self.method,
            'confidence_level': self.confidence_level,
            'uplift_point': self.uplift_point,
            'uplift_ci_lower': self.uplift_ci_lower,
            'uplift_ci_upper': self.uplift_ci_upper,
            'uplift_se': self.uplift_se,
            'realistic_rule_gain_point': self.realistic_rule_gain_point,
            'realistic_rule_gain_ci_lower': self.realistic_rule_gain_ci_lower,
            'realistic_rule_gain_ci_upper': self.realistic_rule_gain_ci_upper,
            'realistic_rule_gain_se': self.realistic_rule_gain_se,
            'support': self.support,
            'confidence': self.confidence,
            'category': self.category.value if self.category is not None else None,
            'undefined_bootstrap_fraction': self.undefined_bootstrap_fraction,
        }
        if include_samples:
            out['samples_uplift'] = None if self.samples_uplift is None else self.samples_uplift.tolist()
            out['samples_gain'] = None if self.samples_gain is None else self.samples_gain.tolist()
        return out

to_dict(include_samples=False)

Return a plain dict representation of this result.

Parameters:

Name	Type	Description	Default
include_samples	bool	When False (default) the samples_uplift and samples_gain arrays are omitted to keep the dict compact and JSON-friendly. When True the arrays are converted to lists.	False

Returns:

Type	Description
dict	Keys mirror the dataclass fields. category (if present) is converted to its RuleCategory.value string.

Source code in src/action_rules/inference/base.py

def to_dict(self, include_samples: bool = False) -> dict:
    """Return a plain dict representation of this result.

    Parameters
    ----------
    include_samples : bool
        When ``False`` (default) the ``samples_uplift`` and ``samples_gain``
        arrays are omitted to keep the dict compact and JSON-friendly.
        When ``True`` the arrays are converted to lists.

    Returns
    -------
    dict
        Keys mirror the dataclass fields.  ``category`` (if present) is
        converted to its ``RuleCategory.value`` string.
    """
    out: dict = {
        'rule_index': self.rule_index,
        'method': self.method,
        'confidence_level': self.confidence_level,
        'uplift_point': self.uplift_point,
        'uplift_ci_lower': self.uplift_ci_lower,
        'uplift_ci_upper': self.uplift_ci_upper,
        'uplift_se': self.uplift_se,
        'realistic_rule_gain_point': self.realistic_rule_gain_point,
        'realistic_rule_gain_ci_lower': self.realistic_rule_gain_ci_lower,
        'realistic_rule_gain_ci_upper': self.realistic_rule_gain_ci_upper,
        'realistic_rule_gain_se': self.realistic_rule_gain_se,
        'support': self.support,
        'confidence': self.confidence,
        'category': self.category.value if self.category is not None else None,
        'undefined_bootstrap_fraction': self.undefined_bootstrap_fraction,
    }
    if include_samples:
        out['samples_uplift'] = None if self.samples_uplift is None else self.samples_uplift.tolist()
        out['samples_gain'] = None if self.samples_gain is None else self.samples_gain.tolist()
    return out

Input

A class used to import action rules.

Methods:

Name	Description
import_action_rules	Import action rules from a JSON string and set the action_rules attribute.

Source code in src/action_rules/input/input.py

class Input:
    """
    A class used to import action rules.

    Methods
    -------
    import_action_rules(json_data)
        Import action rules from a JSON string and set the action_rules attribute.
    """

    def __init__(self):
        """
        Initialize the Input class.

        Notes
        -----
        This class is used to import action rules from a JSON string and convert them into an Output object.
        """

    def import_action_rules(self, json_data: str) -> Output:
        """
        Import action rules from a JSON string and set the action_rules attribute.

        Parameters
        ----------
        json_data : str
            JSON string representing the action rules.

        Returns
        -------
        Output
            Output object representing the action rules.

        Notes
        -----
        This method parses a JSON string containing action rules, extracts relevant information,
        and constructs an Output object. The method initializes the target attribute, stable items,
        flexible items, and column values. It processes both stable and flexible items for each rule
        and updates the corresponding dictionaries.

        The JSON structure is expected to have the following format:
        [
            {
                "target": {
                    "attribute": "target_attribute",
                    "undesired": "undesired_value",
                    "desired": "desired_value"
                },
                "support of undesired part": int,
                "confidence of undesired part": float,
                "support of desired part": int,
                "confidence of desired part": float,
                "uplift": float,
                "max_rule_gain": float,                   (optional)
                "realistic_rule_gain": float,             (optional)
                "realistic_dataset_gain": float,          (optional)
                "stable": [
                    {"attribute": "attribute_name", "value": "attribute_value"},
                    ...
                ],
                "flexible": [
                    {"attribute": "attribute_name", "undesired": "undesired_value", "desired": "desired_value"},
                    ...
                ]
            },
            ...
        ]

        The method ensures that each attribute-value pair is assigned a unique index and maintains
        the mappings in the column_values dictionary. The stable_items_binding and flexible_items_binding
        dictionaries are updated accordingly.

        Example
        -------
        json_data = '''
        [
            {
                "target": {
                    "attribute": "target",
                    "undesired": "no",
                    "desired": "yes"
                },
                "support of undesired part": 10,
                "confidence of undesired part": 0.5,
                "support of desired part": 20,
                "confidence of desired part": 0.8,
                "uplift": 0.3,
                "stable": [
                    {"attribute": "age", "value": "young"},
                    {"attribute": "income", "value": "high"}
                ],
                "flexible": [
                    {"attribute": "education", "undesired": "low", "desired": "high"}
                ]
            }
        ]
        '''
        input_obj = Input()
        output = input_obj.import_action_rules(json_data)
        """
        rules = json.loads(json_data)
        action_rules = []
        target = rules[0]['target']['attribute']
        stable_items_binding = {}  # type: dict
        flexible_items_binding = {}  # type: dict
        column_values = {}
        highest_index = 0
        for rule in rules:
            if highest_index == 0:
                column_values[highest_index] = (rule['target']['attribute'], rule['target']['undesired'])
                highest_index += 1
                column_values[highest_index] = (rule['target']['attribute'], rule['target']['desired'])
                highest_index += 1
            ar_dict = {
                'undesired': {
                    'itemset': [],
                    'support': rule['support of undesired part'],
                    'confidence': rule['confidence of undesired part'],
                    'target': 0,
                },
                'desired': {
                    'itemset': [],
                    'support': rule['support of desired part'],
                    'confidence': rule['confidence of desired part'],
                    'target': 1,
                },
                'uplift': rule['uplift'],
                'support': (
                    rule['support']
                    if 'support' in rule
                    else min(rule['support of undesired part'], rule['support of desired part'])
                ),
                'confidence': (
                    rule['confidence']
                    if 'confidence' in rule
                    else rule['confidence of undesired part'] * rule['confidence of desired part']
                ),
            }
            for item in rule['stable']:
                if (item['attribute'], item['value']) not in column_values.values():
                    column_values[highest_index] = (item['attribute'], item['value'])
                    if 'flexible_as_stable' in item:
                        if item['attribute'] not in flexible_items_binding.keys():
                            flexible_items_binding.update({item['attribute']: []})
                        flexible_items_binding[item['attribute']].append(highest_index)
                    else:
                        if item['attribute'] not in stable_items_binding.keys():
                            stable_items_binding.update({item['attribute']: []})
                        stable_items_binding[item['attribute']].append(highest_index)
                    highest_index += 1
                value = [
                    key
                    for key, (attr, value) in column_values.items()
                    if value == item['value'] and attr == item['attribute']
                ][0]
                ar_dict['undesired']['itemset'].append(value)
                ar_dict['desired']['itemset'].append(value)

            for item in rule['flexible']:
                if item['attribute'] not in flexible_items_binding.keys():
                    flexible_items_binding.update({item['attribute']: []})
                if (item['attribute'], item['undesired']) not in column_values.values():
                    column_values[highest_index] = (item['attribute'], item['undesired'])
                    flexible_items_binding[item['attribute']].append(highest_index)
                    highest_index += 1
                if (item['attribute'], item['desired']) not in column_values.values():
                    column_values[highest_index] = (item['attribute'], item['desired'])
                    flexible_items_binding[item['attribute']].append(highest_index)
                    highest_index += 1
                value_0 = [
                    key
                    for key, (attr, value) in column_values.items()
                    if value == item['undesired'] and attr == item['attribute']
                ][0]
                value_1 = [
                    key
                    for key, (attr, value) in column_values.items()
                    if value == item['desired'] and attr == item['attribute']
                ][0]
                ar_dict['undesired']['itemset'].append(value_0)
                ar_dict['desired']['itemset'].append(value_1)

            # If the JSON rule includes utility parameters, add them to the rule dictionary.
            for utility_key in [
                'max_rule_gain',
                'realistic_rule_gain',
                'realistic_dataset_gain',
            ]:
                if utility_key in rule:
                    ar_dict[utility_key] = rule[utility_key]

            action_rules.append(ar_dict)

        return Output(action_rules, target, stable_items_binding, flexible_items_binding, column_values)

__init__()

Initialize the Input class.

Notes

This class is used to import action rules from a JSON string and convert them into an Output object.

Source code in src/action_rules/input/input.py

def __init__(self):
    """
    Initialize the Input class.

    Notes
    -----
    This class is used to import action rules from a JSON string and convert them into an Output object.
    """

import_action_rules(json_data)

Import action rules from a JSON string and set the action_rules attribute.

Parameters:

Name	Type	Description	Default
json_data	str	JSON string representing the action rules.	required

Returns:

Type	Description
Output	Output object representing the action rules.

Notes

This method parses a JSON string containing action rules, extracts relevant information, and constructs an Output object. The method initializes the target attribute, stable items, flexible items, and column values. It processes both stable and flexible items for each rule and updates the corresponding dictionaries.

The JSON structure is expected to have the following format: [ { "target": { "attribute": "target_attribute", "undesired": "undesired_value", "desired": "desired_value" }, "support of undesired part": int, "confidence of undesired part": float, "support of desired part": int, "confidence of desired part": float, "uplift": float, "max_rule_gain": float, (optional) "realistic_rule_gain": float, (optional) "realistic_dataset_gain": float, (optional) "stable": [ {"attribute": "attribute_name", "value": "attribute_value"}, ... ], "flexible": [ {"attribute": "attribute_name", "undesired": "undesired_value", "desired": "desired_value"}, ... ] }, ... ]

The method ensures that each attribute-value pair is assigned a unique index and maintains the mappings in the column_values dictionary. The stable_items_binding and flexible_items_binding dictionaries are updated accordingly.

Example

json_data = ''' [ { "target": { "attribute": "target", "undesired": "no", "desired": "yes" }, "support of undesired part": 10, "confidence of undesired part": 0.5, "support of desired part": 20, "confidence of desired part": 0.8, "uplift": 0.3, "stable": [ {"attribute": "age", "value": "young"}, {"attribute": "income", "value": "high"} ], "flexible": [ {"attribute": "education", "undesired": "low", "desired": "high"} ] } ] ''' input_obj = Input() output = input_obj.import_action_rules(json_data)

Source code in src/action_rules/input/input.py

def import_action_rules(self, json_data: str) -> Output:
    """
    Import action rules from a JSON string and set the action_rules attribute.

    Parameters
    ----------
    json_data : str
        JSON string representing the action rules.

    Returns
    -------
    Output
        Output object representing the action rules.

    Notes
    -----
    This method parses a JSON string containing action rules, extracts relevant information,
    and constructs an Output object. The method initializes the target attribute, stable items,
    flexible items, and column values. It processes both stable and flexible items for each rule
    and updates the corresponding dictionaries.

    The JSON structure is expected to have the following format:
    [
        {
            "target": {
                "attribute": "target_attribute",
                "undesired": "undesired_value",
                "desired": "desired_value"
            },
            "support of undesired part": int,
            "confidence of undesired part": float,
            "support of desired part": int,
            "confidence of desired part": float,
            "uplift": float,
            "max_rule_gain": float,                   (optional)
            "realistic_rule_gain": float,             (optional)
            "realistic_dataset_gain": float,          (optional)
            "stable": [
                {"attribute": "attribute_name", "value": "attribute_value"},
                ...
            ],
            "flexible": [
                {"attribute": "attribute_name", "undesired": "undesired_value", "desired": "desired_value"},
                ...
            ]
        },
        ...
    ]

    The method ensures that each attribute-value pair is assigned a unique index and maintains
    the mappings in the column_values dictionary. The stable_items_binding and flexible_items_binding
    dictionaries are updated accordingly.

    Example
    -------
    json_data = '''
    [
        {
            "target": {
                "attribute": "target",
                "undesired": "no",
                "desired": "yes"
            },
            "support of undesired part": 10,
            "confidence of undesired part": 0.5,
            "support of desired part": 20,
            "confidence of desired part": 0.8,
            "uplift": 0.3,
            "stable": [
                {"attribute": "age", "value": "young"},
                {"attribute": "income", "value": "high"}
            ],
            "flexible": [
                {"attribute": "education", "undesired": "low", "desired": "high"}
            ]
        }
    ]
    '''
    input_obj = Input()
    output = input_obj.import_action_rules(json_data)
    """
    rules = json.loads(json_data)
    action_rules = []
    target = rules[0]['target']['attribute']
    stable_items_binding = {}  # type: dict
    flexible_items_binding = {}  # type: dict
    column_values = {}
    highest_index = 0
    for rule in rules:
        if highest_index == 0:
            column_values[highest_index] = (rule['target']['attribute'], rule['target']['undesired'])
            highest_index += 1
            column_values[highest_index] = (rule['target']['attribute'], rule['target']['desired'])
            highest_index += 1
        ar_dict = {
            'undesired': {
                'itemset': [],
                'support': rule['support of undesired part'],
                'confidence': rule['confidence of undesired part'],
                'target': 0,
            },
            'desired': {
                'itemset': [],
                'support': rule['support of desired part'],
                'confidence': rule['confidence of desired part'],
                'target': 1,
            },
            'uplift': rule['uplift'],
            'support': (
                rule['support']
                if 'support' in rule
                else min(rule['support of undesired part'], rule['support of desired part'])
            ),
            'confidence': (
                rule['confidence']
                if 'confidence' in rule
                else rule['confidence of undesired part'] * rule['confidence of desired part']
            ),
        }
        for item in rule['stable']:
            if (item['attribute'], item['value']) not in column_values.values():
                column_values[highest_index] = (item['attribute'], item['value'])
                if 'flexible_as_stable' in item:
                    if item['attribute'] not in flexible_items_binding.keys():
                        flexible_items_binding.update({item['attribute']: []})
                    flexible_items_binding[item['attribute']].append(highest_index)
                else:
                    if item['attribute'] not in stable_items_binding.keys():
                        stable_items_binding.update({item['attribute']: []})
                    stable_items_binding[item['attribute']].append(highest_index)
                highest_index += 1
            value = [
                key
                for key, (attr, value) in column_values.items()
                if value == item['value'] and attr == item['attribute']
            ][0]
            ar_dict['undesired']['itemset'].append(value)
            ar_dict['desired']['itemset'].append(value)

        for item in rule['flexible']:
            if item['attribute'] not in flexible_items_binding.keys():
                flexible_items_binding.update({item['attribute']: []})
            if (item['attribute'], item['undesired']) not in column_values.values():
                column_values[highest_index] = (item['attribute'], item['undesired'])
                flexible_items_binding[item['attribute']].append(highest_index)
                highest_index += 1
            if (item['attribute'], item['desired']) not in column_values.values():
                column_values[highest_index] = (item['attribute'], item['desired'])
                flexible_items_binding[item['attribute']].append(highest_index)
                highest_index += 1
            value_0 = [
                key
                for key, (attr, value) in column_values.items()
                if value == item['undesired'] and attr == item['attribute']
            ][0]
            value_1 = [
                key
                for key, (attr, value) in column_values.items()
                if value == item['desired'] and attr == item['attribute']
            ][0]
            ar_dict['undesired']['itemset'].append(value_0)
            ar_dict['desired']['itemset'].append(value_1)

        # If the JSON rule includes utility parameters, add them to the rule dictionary.
        for utility_key in [
            'max_rule_gain',
            'realistic_rule_gain',
            'realistic_dataset_gain',
        ]:
            if utility_key in rule:
                ar_dict[utility_key] = rule[utility_key]

        action_rules.append(ar_dict)

    return Output(action_rules, target, stable_items_binding, flexible_items_binding, column_values)

Output

A class used to format and export action rules.

Attributes:

Name	Type	Description
action_rules	list	List containing the action rules.
target	str	The target attribute for the action rules.
stable_cols	list	List of indices for stable columns.
flexible_cols	list	List of indices for flexible columns.
column_values	dict	Dictionary containing the values of the columns.

Methods:

Name	Description
get_ar_notation	Generate a string representation of the action rules in a human-readable format.
get_export_notation	Generate a JSON string of dictionaries representing the action rules for export.
get_pretty_ar_notation	Generate a list of text strings representing the action rules.

Source code in src/action_rules/output/output.py

class Output:
    """
    A class used to format and export action rules.

    Attributes
    ----------
    action_rules : list
        List containing the action rules.
    target : str
        The target attribute for the action rules.
    stable_cols : list
        List of indices for stable columns.
    flexible_cols : list
        List of indices for flexible columns.
    column_values : dict
        Dictionary containing the values of the columns.

    Methods
    -------
    get_ar_notation()
        Generate a string representation of the action rules in a human-readable format.
    get_export_notation()
        Generate a JSON string of dictionaries representing the action rules for export.
    get_pretty_ar_notation()
        Generate a list of text strings representing the action rules.
    """

    def __init__(
        self,
        action_rules: list,
        target: str,
        stable_items_binding: dict,
        flexible_items_binding: dict,
        column_values: dict,
    ):
        """
        Initialize the Output class with the specified action rules and target attribute.

        Parameters
        ----------
        action_rules : list
            List containing the action rules.
        target : str
            The target attribute for the action rules.
        stable_items_binding : dict
            Dictionary containing bindings for stable items.
        flexible_items_binding : dict
            Dictionary containing bindings for flexible items.
        column_values : dict
            Dictionary containing the values of the columns.

        Notes
        -----
        The constructor initializes the Output object by setting the provided action rules, target attribute,
        stable items, flexible items, and column values. It flattens the stable and flexible items bindings to
        create lists of indices for stable and flexible columns.
        """
        self.action_rules = action_rules
        self.target = target
        self.stable_cols = [item for sublist in stable_items_binding.values() for item in sublist]
        self.flexible_cols = [item for sublist in flexible_items_binding.values() for item in sublist]
        self.column_values = column_values
        self.ci_results = None

    def set_confidence_intervals(self, results):
        """Store confidence interval results.

        Parameters
        ----------
        results : list
            List of ConfidenceIntervalResult objects.
        """
        self.ci_results = results

    def get_ar_notation(self):
        """
        Generate a string representation of the action rules in a human-readable format.

        Returns
        -------
        str
            String representation of the action rules.

        Notes
        -----
        This method constructs a human-readable string representation of the action rules. Each rule is
        formatted to show the attribute-value conditions and transitions. The representation includes
        the support and confidence values for both the undesired and desired parts, as well as the uplift.
        """
        ar_notation = []
        for rule_idx, action_rule in enumerate(self.action_rules):
            rule = '['
            for i, item in enumerate(action_rule['undesired']['itemset']):
                if i > 0:
                    rule += ' ∧ '
                rule += '('
                if item == action_rule['desired']['itemset'][i]:
                    if item in self.stable_cols:
                        val = self.column_values[item]
                        rule += str(val[0]) + ': ' + str(val[1])
                    else:
                        val = self.column_values[item]
                        rule += str(val[0]) + '*: ' + str(val[1])
                else:
                    val = self.column_values[item]
                    val_desired = self.column_values[action_rule['desired']['itemset'][i]]
                    rule += str(val[0]) + ': ' + str(val[1]) + ' → ' + str(val_desired[1])
                rule += ')'
            rule += (
                '] ⇒ ['
                + str(self.target)
                + ': '
                + str(self.column_values[action_rule['undesired']['target']][1])
                + ' → '
                + str(self.column_values[action_rule['desired']['target']][1])
                + ']'
            )
            rule += (
                ', support of undesired part: '
                + str(action_rule['undesired']['support'])
                + ', confidence of undesired part: '
                + str(action_rule['undesired']['confidence'])
            )
            rule += (
                ', support of desired part: '
                + str(action_rule['desired']['support'])
                + ', confidence of desired part: '
                + str(action_rule['desired']['confidence'])
            )
            rule += ', support: ' + str(action_rule['support']) + ', confidence: ' + str(action_rule['confidence'])
            rule += ', uplift: ' + str(action_rule['uplift'])
            # If utility measures exist, include them in the output.
            if 'realistic_rule_gain' in action_rule:
                rule += ", max_rule_gain: " + str(action_rule['max_rule_gain'])
                rule += ", realistic_rule_gain: " + str(action_rule['realistic_rule_gain'])
                rule += ", realistic_dataset_gain: " + str(action_rule['realistic_dataset_gain'])
            # Append CI information when available.
            if self.ci_results is not None and rule_idx < len(self.ci_results):
                ci = self.ci_results[rule_idx]
                rule += ', uplift CI [{:.4f}, {:.4f}]'.format(ci.uplift_ci_lower, ci.uplift_ci_upper)
                if ci.category is not None:
                    rule += ', category: ' + ci.category.value
            ar_notation.append(rule)
        return ar_notation

    def get_export_notation(self):
        """
        Generate a JSON string of dictionaries representing the action rules for export.

        Returns
        -------
        str
            JSON string of dictionaries representing the action rules.

        Notes
        -----
        This method constructs a list of dictionaries where each dictionary represents an action rule.
        The dictionaries include attributes for stable and flexible items, as well as the target attribute,
        support, confidence, and uplift values. The list is then converted to a JSON string for export.
        """
        rules = []
        for rule_idx, ar_dict in enumerate(self.action_rules):
            rule = {'stable': [], 'flexible': []}
            for i, item in enumerate(ar_dict['undesired']['itemset']):
                if item == ar_dict['desired']['itemset'][i]:
                    if item in self.stable_cols:
                        val = self.column_values[item]
                        rule['stable'].append({'attribute': val[0], 'value': val[1]})
                    else:
                        val = self.column_values[item]
                        rule['stable'].append({'attribute': val[0], 'value': val[1], 'flexible_as_stable': True})
                else:
                    val = self.column_values[item]
                    val_desired = self.column_values[ar_dict['desired']['itemset'][i]]
                    rule['flexible'].append({'attribute': val[0], 'undesired': val[1], 'desired': val_desired[1]})
            rule['target'] = {
                'attribute': self.target,
                'undesired': str(self.column_values[ar_dict['undesired']['target']][1]),
                'desired': str(self.column_values[ar_dict['desired']['target']][1]),
            }
            rule['support of undesired part'] = int(ar_dict['undesired']['support'])
            rule['confidence of undesired part'] = float(ar_dict['undesired']['confidence'])
            rule['support of desired part'] = int(ar_dict['desired']['support'])
            rule['confidence of desired part'] = float(ar_dict['desired']['confidence'])
            rule['uplift'] = float(ar_dict['uplift'])
            rule['support'] = int(ar_dict['support'])
            rule['confidence'] = float(ar_dict['confidence'])
            # Include utility measures if available.
            if 'realistic_rule_gain' in ar_dict:
                rule['max_rule_gain'] = float(ar_dict['max_rule_gain'])
                rule['realistic_rule_gain'] = float(ar_dict['realistic_rule_gain'])
                rule['realistic_dataset_gain'] = float(ar_dict['realistic_dataset_gain'])
            # Append CI information when available.
            if self.ci_results is not None and rule_idx < len(self.ci_results):
                ci = self.ci_results[rule_idx]
                rule['ci'] = {
                    'method': ci.method,
                    'confidence_level': ci.confidence_level,
                    'uplift_ci_lower': _safe_float(ci.uplift_ci_lower),
                    'uplift_ci_upper': _safe_float(ci.uplift_ci_upper),
                    'uplift_se': _safe_float(ci.uplift_se),
                }
                if ci.realistic_rule_gain_point is not None:
                    rule['ci']['realistic_rule_gain_ci_lower'] = _safe_float(ci.realistic_rule_gain_ci_lower)
                    rule['ci']['realistic_rule_gain_ci_upper'] = _safe_float(ci.realistic_rule_gain_ci_upper)
                    rule['ci']['realistic_rule_gain_se'] = _safe_float(ci.realistic_rule_gain_se)
                if ci.category is not None:
                    rule['ci']['category'] = ci.category.value
            rules.append(rule)
        return json.dumps(rules)

    def get_pretty_ar_notation(self):
        """
        Generate a list of text strings representing the action rules.

        Returns
        -------
        list
            List of text strings representing the action rules.

        Notes
        -----
        This method constructs a list of text strings where each string represents an action rule in a
        readable format. The format includes conditions and transitions for each attribute, along with
        the target attribute change, support, confidence, and uplift values.
        """
        rules = []
        for rule_idx, ar_dict in enumerate(self.action_rules):
            text = "If "
            for i, item in enumerate(ar_dict['undesired']['itemset']):
                if item == ar_dict['desired']['itemset'][i]:
                    if item in self.stable_cols:
                        val = self.column_values[item]
                        text += "attribute '" + val[0] + "' is '" + val[1] + "', "
                    else:
                        val = self.column_values[item]
                        text += "attribute (flexible is used as stable) '" + val[0] + "' is '" + val[1] + "', "
                else:
                    val = self.column_values[item]
                    val_desired = self.column_values[ar_dict['desired']['itemset'][i]]
                    text += "attribute '" + val[0] + "' value '" + val[1] + "' is changed to '" + val_desired[1] + "', "
            text += (
                "then '"
                + self.target
                + "' value '"
                + self.column_values[ar_dict['undesired']['target']][1]
                + "' is changed to '"
                + self.column_values[ar_dict['desired']['target']][1]
                + " with support: "
                + str(ar_dict['support'])
                + ", confidence: "
                + str(ar_dict['confidence'])
                + ", uplift: "
                + str(ar_dict['uplift'])
                + ", support of undesired part: "
                + str(ar_dict['undesired']['support'])
                + ", confidence of undesired part: "
                + str(ar_dict['undesired']['confidence'])
                + ", support of desired part: "
                + str(ar_dict['desired']['support'])
                + ", confidence of desired part: "
                + str(ar_dict['desired']['confidence'])
            )
            if 'realistic_rule_gain' in ar_dict:
                text += ", max_rule_gain: " + str(ar_dict['max_rule_gain'])
                text += ", realistic_rule_gain: " + str(ar_dict['realistic_rule_gain'])
                text += ", realistic_dataset_gain: " + str(ar_dict['realistic_dataset_gain'])
            # Append CI information when available.
            if self.ci_results is not None and rule_idx < len(self.ci_results):
                ci = self.ci_results[rule_idx]
                text += ', uplift CI [{:.4f}, {:.4f}]'.format(ci.uplift_ci_lower, ci.uplift_ci_upper)
                if ci.category is not None:
                    text += ', category: ' + ci.category.value
            text += "."
            rules.append(text)
        return rules

    def get_dominant_rules(self):
        """
        Identify and select the dominant (Pareto-optimal) action rules.

        This method compares action rules based on the union of their 'undesired'
        and 'desired' itemsets, as well as their 'uplift' values. It applies a
        Pareto-dominance approach:

        - If the new candidate rule is a superset of a current dominant rule
          with smaller or equal uplift, the candidate is dominated and not added.
        - If the new candidate rule is a subset of a current dominant rule
          with larger or equal uplift, the current dominant rule is dominated
          and removed.
        - Otherwise, the new candidate is added to the set of dominant rules.

        After processing all rules, the remaining dominant rules are sorted
        by 'uplift' in descending order, and the method returns their indices.

        Returns
        -------
        list
            A list of indices representing the dominant (Pareto-optimal)
            action rules, sorted by uplift in descending order.
        """
        dominant_rules = []

        # Initialize the first candidate rule
        first_rule = self.action_rules[0]
        dominant_rules.append(
            {
                'candidate_set': set(first_rule['undesired']['itemset']) | set(first_rule['desired']['itemset']),
                'rule_index': 0,
                'uplift': first_rule['uplift'],
                'to_delete': False,
            }
        )

        # Iterate through remaining rules
        for idx, new_candidate in enumerate(self.action_rules[1:], start=1):
            new_candidate_set = set(new_candidate['undesired']['itemset']) | set(new_candidate['desired']['itemset'])
            is_add_rule = True
            # Compare the new dominant rule candidate with all current dominant rule candidates
            for dominant_rule in dominant_rules:
                # If the new candidate is superset of the dominant rule candidate and its uplift is smaller or the same,
                # the rule is not added to dominant rule candidates
                if (
                    dominant_rule['candidate_set'] < new_candidate_set
                    and dominant_rule['uplift'] >= new_candidate['uplift']
                ):
                    is_add_rule = False
                    break
                # If the new candidate is subset of the dominant rule candidate and its uplift is higher or the same,
                # the dominant rule candidate is removed from the dominant rule candidates
                elif (
                    dominant_rule['candidate_set'] > new_candidate_set
                    and dominant_rule['uplift'] <= new_candidate['uplift']
                ):
                    dominant_rule['to_delete'] = True
            # If the candidate rule did not find any rule that would be dominant to its, add the candidate to dominant
            # rule candidates
            if is_add_rule:
                dominant_rules.append(
                    {
                        'candidate_set': new_candidate_set,
                        'rule_index': idx,
                        'uplift': new_candidate['uplift'],
                        'to_delete': False,
                    }
                )
            # Remove rules that are not anymore dominant
            dominant_rules = [rule for rule in dominant_rules if not rule['to_delete']]
        # Sort the action rules from the highest uplift
        sorted_indices = sorted(dominant_rules, key=lambda x: x["uplift"], reverse=True)
        important_rules_indices = [rule['rule_index'] for rule in sorted_indices]
        return important_rules_indices

__init__(action_rules, target, stable_items_binding, flexible_items_binding, column_values)

Initialize the Output class with the specified action rules and target attribute.

Parameters:

Name	Type	Description	Default
action_rules	list	List containing the action rules.	required
target	str	The target attribute for the action rules.	required
stable_items_binding	dict	Dictionary containing bindings for stable items.	required
flexible_items_binding	dict	Dictionary containing bindings for flexible items.	required
column_values	dict	Dictionary containing the values of the columns.	required

Notes

The constructor initializes the Output object by setting the provided action rules, target attribute, stable items, flexible items, and column values. It flattens the stable and flexible items bindings to create lists of indices for stable and flexible columns.

Source code in src/action_rules/output/output.py

def __init__(
    self,
    action_rules: list,
    target: str,
    stable_items_binding: dict,
    flexible_items_binding: dict,
    column_values: dict,
):
    """
    Initialize the Output class with the specified action rules and target attribute.

    Parameters
    ----------
    action_rules : list
        List containing the action rules.
    target : str
        The target attribute for the action rules.
    stable_items_binding : dict
        Dictionary containing bindings for stable items.
    flexible_items_binding : dict
        Dictionary containing bindings for flexible items.
    column_values : dict
        Dictionary containing the values of the columns.

    Notes
    -----
    The constructor initializes the Output object by setting the provided action rules, target attribute,
    stable items, flexible items, and column values. It flattens the stable and flexible items bindings to
    create lists of indices for stable and flexible columns.
    """
    self.action_rules = action_rules
    self.target = target
    self.stable_cols = [item for sublist in stable_items_binding.values() for item in sublist]
    self.flexible_cols = [item for sublist in flexible_items_binding.values() for item in sublist]
    self.column_values = column_values
    self.ci_results = None

get_ar_notation()

Generate a string representation of the action rules in a human-readable format.

Returns:

Type	Description
str	String representation of the action rules.

Notes

This method constructs a human-readable string representation of the action rules. Each rule is formatted to show the attribute-value conditions and transitions. The representation includes the support and confidence values for both the undesired and desired parts, as well as the uplift.

Source code in src/action_rules/output/output.py

def get_ar_notation(self):
    """
    Generate a string representation of the action rules in a human-readable format.

    Returns
    -------
    str
        String representation of the action rules.

    Notes
    -----
    This method constructs a human-readable string representation of the action rules. Each rule is
    formatted to show the attribute-value conditions and transitions. The representation includes
    the support and confidence values for both the undesired and desired parts, as well as the uplift.
    """
    ar_notation = []
    for rule_idx, action_rule in enumerate(self.action_rules):
        rule = '['
        for i, item in enumerate(action_rule['undesired']['itemset']):
            if i > 0:
                rule += ' ∧ '
            rule += '('
            if item == action_rule['desired']['itemset'][i]:
                if item in self.stable_cols:
                    val = self.column_values[item]
                    rule += str(val[0]) + ': ' + str(val[1])
                else:
                    val = self.column_values[item]
                    rule += str(val[0]) + '*: ' + str(val[1])
            else:
                val = self.column_values[item]
                val_desired = self.column_values[action_rule['desired']['itemset'][i]]
                rule += str(val[0]) + ': ' + str(val[1]) + ' → ' + str(val_desired[1])
            rule += ')'
        rule += (
            '] ⇒ ['
            + str(self.target)
            + ': '
            + str(self.column_values[action_rule['undesired']['target']][1])
            + ' → '
            + str(self.column_values[action_rule['desired']['target']][1])
            + ']'
        )
        rule += (
            ', support of undesired part: '
            + str(action_rule['undesired']['support'])
            + ', confidence of undesired part: '
            + str(action_rule['undesired']['confidence'])
        )
        rule += (
            ', support of desired part: '
            + str(action_rule['desired']['support'])
            + ', confidence of desired part: '
            + str(action_rule['desired']['confidence'])
        )
        rule += ', support: ' + str(action_rule['support']) + ', confidence: ' + str(action_rule['confidence'])
        rule += ', uplift: ' + str(action_rule['uplift'])
        # If utility measures exist, include them in the output.
        if 'realistic_rule_gain' in action_rule:
            rule += ", max_rule_gain: " + str(action_rule['max_rule_gain'])
            rule += ", realistic_rule_gain: " + str(action_rule['realistic_rule_gain'])
            rule += ", realistic_dataset_gain: " + str(action_rule['realistic_dataset_gain'])
        # Append CI information when available.
        if self.ci_results is not None and rule_idx < len(self.ci_results):
            ci = self.ci_results[rule_idx]
            rule += ', uplift CI [{:.4f}, {:.4f}]'.format(ci.uplift_ci_lower, ci.uplift_ci_upper)
            if ci.category is not None:
                rule += ', category: ' + ci.category.value
        ar_notation.append(rule)
    return ar_notation

get_dominant_rules()

Identify and select the dominant (Pareto-optimal) action rules.

This method compares action rules based on the union of their 'undesired' and 'desired' itemsets, as well as their 'uplift' values. It applies a Pareto-dominance approach:

• If the new candidate rule is a superset of a current dominant rule with smaller or equal uplift, the candidate is dominated and not added.
• If the new candidate rule is a subset of a current dominant rule with larger or equal uplift, the current dominant rule is dominated and removed.
• Otherwise, the new candidate is added to the set of dominant rules.

After processing all rules, the remaining dominant rules are sorted by 'uplift' in descending order, and the method returns their indices.

Returns:

Type	Description
list	A list of indices representing the dominant (Pareto-optimal) action rules, sorted by uplift in descending order.

Source code in src/action_rules/output/output.py

def get_dominant_rules(self):
    """
    Identify and select the dominant (Pareto-optimal) action rules.

    This method compares action rules based on the union of their 'undesired'
    and 'desired' itemsets, as well as their 'uplift' values. It applies a
    Pareto-dominance approach:

    - If the new candidate rule is a superset of a current dominant rule
      with smaller or equal uplift, the candidate is dominated and not added.
    - If the new candidate rule is a subset of a current dominant rule
      with larger or equal uplift, the current dominant rule is dominated
      and removed.
    - Otherwise, the new candidate is added to the set of dominant rules.

    After processing all rules, the remaining dominant rules are sorted
    by 'uplift' in descending order, and the method returns their indices.

    Returns
    -------
    list
        A list of indices representing the dominant (Pareto-optimal)
        action rules, sorted by uplift in descending order.
    """
    dominant_rules = []

    # Initialize the first candidate rule
    first_rule = self.action_rules[0]
    dominant_rules.append(
        {
            'candidate_set': set(first_rule['undesired']['itemset']) | set(first_rule['desired']['itemset']),
            'rule_index': 0,
            'uplift': first_rule['uplift'],
            'to_delete': False,
        }
    )

    # Iterate through remaining rules
    for idx, new_candidate in enumerate(self.action_rules[1:], start=1):
        new_candidate_set = set(new_candidate['undesired']['itemset']) | set(new_candidate['desired']['itemset'])
        is_add_rule = True
        # Compare the new dominant rule candidate with all current dominant rule candidates
        for dominant_rule in dominant_rules:
            # If the new candidate is superset of the dominant rule candidate and its uplift is smaller or the same,
            # the rule is not added to dominant rule candidates
            if (
                dominant_rule['candidate_set'] < new_candidate_set
                and dominant_rule['uplift'] >= new_candidate['uplift']
            ):
                is_add_rule = False
                break
            # If the new candidate is subset of the dominant rule candidate and its uplift is higher or the same,
            # the dominant rule candidate is removed from the dominant rule candidates
            elif (
                dominant_rule['candidate_set'] > new_candidate_set
                and dominant_rule['uplift'] <= new_candidate['uplift']
            ):
                dominant_rule['to_delete'] = True
        # If the candidate rule did not find any rule that would be dominant to its, add the candidate to dominant
        # rule candidates
        if is_add_rule:
            dominant_rules.append(
                {
                    'candidate_set': new_candidate_set,
                    'rule_index': idx,
                    'uplift': new_candidate['uplift'],
                    'to_delete': False,
                }
            )
        # Remove rules that are not anymore dominant
        dominant_rules = [rule for rule in dominant_rules if not rule['to_delete']]
    # Sort the action rules from the highest uplift
    sorted_indices = sorted(dominant_rules, key=lambda x: x["uplift"], reverse=True)
    important_rules_indices = [rule['rule_index'] for rule in sorted_indices]
    return important_rules_indices

get_export_notation()

Generate a JSON string of dictionaries representing the action rules for export.

Returns:

Type	Description
str	JSON string of dictionaries representing the action rules.

Notes

This method constructs a list of dictionaries where each dictionary represents an action rule. The dictionaries include attributes for stable and flexible items, as well as the target attribute, support, confidence, and uplift values. The list is then converted to a JSON string for export.

Source code in src/action_rules/output/output.py

def get_export_notation(self):
    """
    Generate a JSON string of dictionaries representing the action rules for export.

    Returns
    -------
    str
        JSON string of dictionaries representing the action rules.

    Notes
    -----
    This method constructs a list of dictionaries where each dictionary represents an action rule.
    The dictionaries include attributes for stable and flexible items, as well as the target attribute,
    support, confidence, and uplift values. The list is then converted to a JSON string for export.
    """
    rules = []
    for rule_idx, ar_dict in enumerate(self.action_rules):
        rule = {'stable': [], 'flexible': []}
        for i, item in enumerate(ar_dict['undesired']['itemset']):
            if item == ar_dict['desired']['itemset'][i]:
                if item in self.stable_cols:
                    val = self.column_values[item]
                    rule['stable'].append({'attribute': val[0], 'value': val[1]})
                else:
                    val = self.column_values[item]
                    rule['stable'].append({'attribute': val[0], 'value': val[1], 'flexible_as_stable': True})
            else:
                val = self.column_values[item]
                val_desired = self.column_values[ar_dict['desired']['itemset'][i]]
                rule['flexible'].append({'attribute': val[0], 'undesired': val[1], 'desired': val_desired[1]})
        rule['target'] = {
            'attribute': self.target,
            'undesired': str(self.column_values[ar_dict['undesired']['target']][1]),
            'desired': str(self.column_values[ar_dict['desired']['target']][1]),
        }
        rule['support of undesired part'] = int(ar_dict['undesired']['support'])
        rule['confidence of undesired part'] = float(ar_dict['undesired']['confidence'])
        rule['support of desired part'] = int(ar_dict['desired']['support'])
        rule['confidence of desired part'] = float(ar_dict['desired']['confidence'])
        rule['uplift'] = float(ar_dict['uplift'])
        rule['support'] = int(ar_dict['support'])
        rule['confidence'] = float(ar_dict['confidence'])
        # Include utility measures if available.
        if 'realistic_rule_gain' in ar_dict:
            rule['max_rule_gain'] = float(ar_dict['max_rule_gain'])
            rule['realistic_rule_gain'] = float(ar_dict['realistic_rule_gain'])
            rule['realistic_dataset_gain'] = float(ar_dict['realistic_dataset_gain'])
        # Append CI information when available.
        if self.ci_results is not None and rule_idx < len(self.ci_results):
            ci = self.ci_results[rule_idx]
            rule['ci'] = {
                'method': ci.method,
                'confidence_level': ci.confidence_level,
                'uplift_ci_lower': _safe_float(ci.uplift_ci_lower),
                'uplift_ci_upper': _safe_float(ci.uplift_ci_upper),
                'uplift_se': _safe_float(ci.uplift_se),
            }
            if ci.realistic_rule_gain_point is not None:
                rule['ci']['realistic_rule_gain_ci_lower'] = _safe_float(ci.realistic_rule_gain_ci_lower)
                rule['ci']['realistic_rule_gain_ci_upper'] = _safe_float(ci.realistic_rule_gain_ci_upper)
                rule['ci']['realistic_rule_gain_se'] = _safe_float(ci.realistic_rule_gain_se)
            if ci.category is not None:
                rule['ci']['category'] = ci.category.value
        rules.append(rule)
    return json.dumps(rules)

get_pretty_ar_notation()

Generate a list of text strings representing the action rules.

Returns:

Type	Description
list	List of text strings representing the action rules.

Notes

This method constructs a list of text strings where each string represents an action rule in a readable format. The format includes conditions and transitions for each attribute, along with the target attribute change, support, confidence, and uplift values.

Source code in src/action_rules/output/output.py

def get_pretty_ar_notation(self):
    """
    Generate a list of text strings representing the action rules.

    Returns
    -------
    list
        List of text strings representing the action rules.

    Notes
    -----
    This method constructs a list of text strings where each string represents an action rule in a
    readable format. The format includes conditions and transitions for each attribute, along with
    the target attribute change, support, confidence, and uplift values.
    """
    rules = []
    for rule_idx, ar_dict in enumerate(self.action_rules):
        text = "If "
        for i, item in enumerate(ar_dict['undesired']['itemset']):
            if item == ar_dict['desired']['itemset'][i]:
                if item in self.stable_cols:
                    val = self.column_values[item]
                    text += "attribute '" + val[0] + "' is '" + val[1] + "', "
                else:
                    val = self.column_values[item]
                    text += "attribute (flexible is used as stable) '" + val[0] + "' is '" + val[1] + "', "
            else:
                val = self.column_values[item]
                val_desired = self.column_values[ar_dict['desired']['itemset'][i]]
                text += "attribute '" + val[0] + "' value '" + val[1] + "' is changed to '" + val_desired[1] + "', "
        text += (
            "then '"
            + self.target
            + "' value '"
            + self.column_values[ar_dict['undesired']['target']][1]
            + "' is changed to '"
            + self.column_values[ar_dict['desired']['target']][1]
            + " with support: "
            + str(ar_dict['support'])
            + ", confidence: "
            + str(ar_dict['confidence'])
            + ", uplift: "
            + str(ar_dict['uplift'])
            + ", support of undesired part: "
            + str(ar_dict['undesired']['support'])
            + ", confidence of undesired part: "
            + str(ar_dict['undesired']['confidence'])
            + ", support of desired part: "
            + str(ar_dict['desired']['support'])
            + ", confidence of desired part: "
            + str(ar_dict['desired']['confidence'])
        )
        if 'realistic_rule_gain' in ar_dict:
            text += ", max_rule_gain: " + str(ar_dict['max_rule_gain'])
            text += ", realistic_rule_gain: " + str(ar_dict['realistic_rule_gain'])
            text += ", realistic_dataset_gain: " + str(ar_dict['realistic_dataset_gain'])
        # Append CI information when available.
        if self.ci_results is not None and rule_idx < len(self.ci_results):
            ci = self.ci_results[rule_idx]
            text += ', uplift CI [{:.4f}, {:.4f}]'.format(ci.uplift_ci_lower, ci.uplift_ci_upper)
            if ci.category is not None:
                text += ', category: ' + ci.category.value
        text += "."
        rules.append(text)
    return rules

set_confidence_intervals(results)

Store confidence interval results.

Parameters:

Name	Type	Description	Default
results	list	List of ConfidenceIntervalResult objects.	required

Source code in src/action_rules/output/output.py

def set_confidence_intervals(self, results):
    """Store confidence interval results.

    Parameters
    ----------
    results : list
        List of ConfidenceIntervalResult objects.
    """
    self.ci_results = results

RuleCategory

Bases: Enum

Classification of an action rule based on its confidence interval.

Attributes:

Name	Type	Description
ACCEPT	str	The entire CI is at or above the threshold — rule is reliable.
REJECT	str	The entire CI is below the threshold — rule should be discarded.
UNCERTAIN	str	The CI straddles the threshold — insufficient evidence to decide.

Source code in src/action_rules/inference/base.py

class RuleCategory(Enum):
    """Classification of an action rule based on its confidence interval.

    Attributes
    ----------
    ACCEPT : str
        The entire CI is at or above the threshold — rule is reliable.
    REJECT : str
        The entire CI is below the threshold — rule should be discarded.
    UNCERTAIN : str
        The CI straddles the threshold — insufficient evidence to decide.
    """

    ACCEPT = 'accept'
    REJECT = 'reject'
    UNCERTAIN = 'uncertain'

RuleMasks dataclass

Structured representation of a single action rule for data-driven inference.

Parameters:

Name	Type	Description	Default
mask_undesired	dict	Mapping of {attribute: value} for conditions in the undesired classification rule (stable attributes at their fixed value plus flexible attributes at their undesired value).	required
mask_desired	dict	Mapping of {attribute: value} for conditions in the desired classification rule (stable attributes at their fixed value plus flexible attributes at their desired value).	required
target_attribute	str	Name of the target column.	required
target_undesired	str	Target value representing the undesired outcome.	required
target_desired	str	Target value representing the desired outcome.	required
rule_index	int	Zero-based position of this rule in the original Output.action_rules list.	required
undesired_itemset	tuple	Original integer column indices that form the undesired itemset. Preserved so that utility tables (keyed by integer index) can be looked up without re-encoding.	required
desired_itemset	tuple	Original integer column indices that form the desired itemset.	required

Source code in src/action_rules/inference/base.py

@dataclass
class RuleMasks:
    """Structured representation of a single action rule for data-driven inference.

    Parameters
    ----------
    mask_undesired : dict
        Mapping of {attribute: value} for conditions in the undesired classification
        rule (stable attributes at their fixed value plus flexible attributes at their
        undesired value).
    mask_desired : dict
        Mapping of {attribute: value} for conditions in the desired classification
        rule (stable attributes at their fixed value plus flexible attributes at their
        desired value).
    target_attribute : str
        Name of the target column.
    target_undesired : str
        Target value representing the undesired outcome.
    target_desired : str
        Target value representing the desired outcome.
    rule_index : int
        Zero-based position of this rule in the original ``Output.action_rules`` list.
    undesired_itemset : tuple
        Original integer column indices that form the undesired itemset.
        Preserved so that utility tables (keyed by integer index) can be looked up
        without re-encoding.
    desired_itemset : tuple
        Original integer column indices that form the desired itemset.
    """

    mask_undesired: dict
    mask_desired: dict
    target_attribute: str
    target_undesired: str
    target_desired: str
    rule_index: int
    undesired_itemset: tuple
    desired_itemset: tuple

Rules

A class used to manage and generate classification and action rules.

Attributes:

Name	Type	Description
classification_rules	defaultdict	Default dictionary to store classification rules for undesired and desired states.
undesired_state	str	The undesired state of the target attribute.
desired_state	str	The desired state of the target attribute.
columns	list	List of columns in the dataset.
action_rules	list	List to store generated action rules.
undesired_prefixes_without_conf	set	Set to store prefixes of undesired states without conflicts.
desired_prefixes_without_conf	set	Set to store prefixes of desired states without conflicts.
count_transactions	int	The number of transactions in the data.
intrinsic_utility_table	(dict, optional)	(attribute, value) -> float A lookup table for the intrinsic utility of each attribute-value pair. If None, no intrinsic utility is considered.
transition_utility_table	(dict, optional)	(attribute, from_value, to_value) -> float A lookup table for cost/gain of transitions between values. If None, no transition utility is considered.

Methods:

Name	Description
add_prefix_without_conf	Add a prefix to the set of prefixes without conflicts.
add_classification_rules	Add classification rules for undesired and desired states.
generate_action_rules	Generate action rules from classification rules.
prune_classification_rules	Prune classification rules based on their length and update the stop list.
calculate_confidence	Calculate the confidence of a rule.
calculate_uplift	Calculate the uplift of an action rule.

Source code in src/action_rules/rules/rules.py

class Rules:
    """
    A class used to manage and generate classification and action rules.

    Attributes
    ----------
    classification_rules : defaultdict
        Default dictionary to store classification rules for undesired and desired states.
    undesired_state : str
        The undesired state of the target attribute.
    desired_state : str
        The desired state of the target attribute.
    columns : list
        List of columns in the dataset.
    action_rules : list
        List to store generated action rules.
    undesired_prefixes_without_conf : set
        Set to store prefixes of undesired states without conflicts.
    desired_prefixes_without_conf : set
        Set to store prefixes of desired states without conflicts.
    count_transactions : int
        The number of transactions in the data.
    intrinsic_utility_table : dict, optional
        (attribute, value) -> float
        A lookup table for the intrinsic utility of each attribute-value pair.
        If None, no intrinsic utility is considered.
    transition_utility_table : dict, optional
        (attribute, from_value, to_value) -> float
        A lookup table for cost/gain of transitions between values.
        If None, no transition utility is considered.

    Methods
    -------
    add_prefix_without_conf(prefix, is_desired)
        Add a prefix to the set of prefixes without conflicts.
    add_classification_rules(new_ar_prefix, itemset_prefix, undesired_states, desired_states)
        Add classification rules for undesired and desired states.
    generate_action_rules()
        Generate action rules from classification rules.
    prune_classification_rules(k, stop_list)
        Prune classification rules based on their length and update the stop list.
    calculate_confidence(support, opposite_support)
        Calculate the confidence of a rule.
    calculate_uplift(undesired_support, undesired_confidence, desired_confidence)
        Calculate the uplift of an action rule.
    """

    def __init__(
        self,
        undesired_state: str,
        desired_state: str,
        columns: list,
        count_transactions: int,
        intrinsic_utility_table: Optional[dict] = None,
        transition_utility_table: Optional[dict] = None,
    ):
        """
        Initialize the Rules class with the specified undesired and desired states, columns, and transaction count.

        Parameters
        ----------
        undesired_state : str
            The undesired state of the target attribute.
        desired_state : str
            The desired state of the target attribute.
        columns : list
            List of columns in the dataset.
        count_transactions : int
            The number of transactions in the data.
        intrinsic_utility_table : dict, optional
            (attribute, value) -> float
            A lookup table for the intrinsic utility of each attribute-value pair.
            If None, no intrinsic utility is considered.
        transition_utility_table : dict, optional
            (attribute, from_value, to_value) -> float
            A lookup table for cost/gain of transitions between values.
            If None, no transition utility is considered.

        Notes
        -----
        The classification_rules attribute is initialized as a defaultdict with a lambda function that creates
        dictionaries for 'desired' and 'undesired' states.
        """
        self.classification_rules = defaultdict(lambda: {'desired': [], 'undesired': []})  # type: defaultdict
        self.undesired_state = undesired_state
        self.columns = columns
        self.desired_state = desired_state
        self.action_rules = []  # type: list
        self.undesired_prefixes_without_conf = set()  # type: set
        self.desired_prefixes_without_conf = set()  # type: set
        self.count_transactions = count_transactions
        self.intrinsic_utility_table = intrinsic_utility_table or {}
        self.transition_utility_table = transition_utility_table or {}
        self.has_utility_tables = bool(self.intrinsic_utility_table or self.transition_utility_table)

    def add_prefix_without_conf(self, prefix: tuple, is_desired: bool):
        """
        Add a prefix to the set of prefixes without conflicts.

        Parameters
        ----------
        prefix : tuple
            The prefix to be added.
        is_desired : bool
            If True, add the prefix to the desired prefixes set; otherwise, add it to the undesired prefixes set.

        Notes
        -----
        This method is useful for keeping track of prefixes that have no conflicting rules and can be
        used directly in rule generation.
        """
        if is_desired:
            self.desired_prefixes_without_conf.add(prefix)
        else:
            self.undesired_prefixes_without_conf.add(prefix)

    @staticmethod
    def _add_stop_entry(stop_collection, value: tuple) -> None:
        """Add a stop entry to either a list or a set."""
        if hasattr(stop_collection, "add"):
            stop_collection.add(value)
        else:
            stop_collection.append(value)

    def add_classification_rules(self, new_ar_prefix, itemset_prefix, undesired_states, desired_states):
        """
        Add classification rules for undesired and desired states.

        Parameters
        ----------
        new_ar_prefix : tuple
            Prefix of the action rule.
        itemset_prefix : tuple
            Prefix of the itemset.
        undesired_states : list
            List of dictionaries containing undesired state information.
        desired_states : list
            List of dictionaries containing desired state information.

        Notes
        -----
        This method updates the classification_rules attribute with new rules based on the provided
        undesired and desired states. Each state is represented as a dictionary containing item, support,
        confidence, and target information.
        """
        for undesired_item in undesired_states:
            new_itemset_prefix = itemset_prefix + (undesired_item['item'],)
            self.classification_rules[new_ar_prefix]['undesired'].append(
                {
                    'itemset': new_itemset_prefix,
                    'support': undesired_item['support'],
                    'confidence': undesired_item['confidence'],
                    'target': self.undesired_state,
                }
            )
        for desired_item in desired_states:
            new_itemset_prefix = itemset_prefix + (desired_item['item'],)
            self.classification_rules[new_ar_prefix]['desired'].append(
                {
                    'itemset': new_itemset_prefix,
                    'support': desired_item['support'],
                    'confidence': desired_item['confidence'],
                    'target': self.desired_state,
                }
            )

    def generate_action_rules(self):
        """
        Generate action rules from classification rules.

        Notes
        -----
        This method creates action rules by combining classification rules for undesired and desired states.
        The uplift for each action rule is calculated using the `calculate_uplift` method and the result is
        stored in the action_rules attribute.
        """
        for attribute_prefix, rules in self.classification_rules.items():
            for desired_rule in rules['desired']:
                for undesired_rule in rules['undesired']:
                    # Uplift
                    uplift = self.calculate_uplift(
                        undesired_rule['support'],
                        undesired_rule['confidence'],
                        desired_rule['confidence'],
                    )
                    # Utility
                    utility = None
                    if self.intrinsic_utility_table is not None or self.transition_utility_table is not None:
                        max_rule_gain, realistic_rule_gain, realistic_rule_gain_dataset = self.compute_rule_utility(
                            undesired_rule, desired_rule
                        )
                        utility = {
                            'max_rule_gain': max_rule_gain,
                            'realistic_rule_gain': realistic_rule_gain,
                            'realistic_dataset_gain': realistic_rule_gain_dataset,
                        }
                    # Action rule measures
                    ar_support, ar_confidence = self.compute_action_rule_measures(
                        undesired_rule.get('support', 0.0),
                        undesired_rule.get('confidence', 0.0),
                        desired_rule.get('support', 0.0),
                        desired_rule.get('confidence', 0.0),
                    )
                    self.action_rules.append(
                        {
                            'undesired': undesired_rule,
                            'desired': desired_rule,
                            'uplift': uplift,
                            'support': ar_support,
                            'confidence': ar_confidence,
                            **utility,
                        }
                    )

    def prune_classification_rules(self, k: int, stop_list: Union[list, set]):
        """
        Prune classification rules based on their length and update the stop list.

        Parameters
        ----------
        k : int
            Length of the attribute prefix.
        stop_list : list
            List of prefixes to stop generating rules for.

        Notes
        -----
        This method removes classification rules whose prefix length equals k and either desired or undesired
        states are empty. The corresponding prefixes are also added to the stop_list to avoid further rule generation.
        """
        del_prefixes = []
        for attribute_prefix, rules in self.classification_rules.items():
            if k == len(attribute_prefix):
                len_desired = len(rules['desired'])
                len_undesired = len(rules['undesired'])
                if len_desired == 0 or len_undesired == 0:
                    if (len_desired == 0 and attribute_prefix not in self.desired_prefixes_without_conf) or (
                        len_undesired == 0 and attribute_prefix not in self.undesired_prefixes_without_conf
                    ):
                        self._add_stop_entry(stop_list, attribute_prefix)
                    del_prefixes.append(attribute_prefix)
        for attribute_prefix in del_prefixes:
            del self.classification_rules[attribute_prefix]

    def calculate_confidence(self, support, opposite_support):
        """
        Calculate the confidence of a rule.

        Parameters
        ----------
        support : int
            The support value for the desired or undesired state.
        opposite_support : int
            The support value for the opposite state.

        Returns
        -------
        float
            The confidence value calculated as support / (support + opposite_support).
            Returns 0 if the sum of support and opposite_support is 0.

        Notes
        -----
        Confidence is a measure of the reliability of a rule. A higher confidence indicates a stronger
        association between the conditions of the rule and the target state.
        """
        if support + opposite_support == 0:
            return 0
        return support / (support + opposite_support)

    def calculate_uplift(self, undesired_support: int, undesired_confidence: float, desired_confidence: float) -> float:
        """
        Calculate the uplift of an action rule.

        Parameters
        ----------
        undesired_support : int
            The support value for the undesired state.
        undesired_confidence : float
            The confidence value for the undesired state.
        desired_confidence : float
            The confidence value for the desired state.

        Returns
        -------
        float
            The uplift value calculated as:
            ((desired_confidence - (1 - undesired_confidence)) * (undesired_support / undesired_confidence))
            / self.count_transactions.

        Notes
        -----
        Uplift measures the increase in the probability of achieving the desired state when applying the action rule
        compared to not applying it. It is used to assess the effectiveness of the rule.
        """
        return (
            (desired_confidence - (1 - undesired_confidence)) * (undesired_support / undesired_confidence)
        ) / self.count_transactions

    def compute_rule_utility(self, undesired_rule: dict, desired_rule: dict) -> tuple:
        """
        Compute various utility gains for a rule transition from undesired to desired.

        The function computes intrinsic utilities for items in both the undesired and desired rule itemsets,
        calculates a transition gain for changes in flexible attributes, and adjusts these gains using target
        state utilities and rule confidences to derive realistic gain metrics at both the rule and dataset levels.

        Parameters
        ----------
        undesired_rule : dict
            Dictionary representing the undesired rule. Expected keys:
                - 'itemset': list of item indices in the undesired rule.
                - 'confidence': (optional) confidence level of the undesired rule.
                - 'support': (optional) support count of the undesired rule.
        desired_rule : dict
            Dictionary representing the desired rule. Expected keys:
                - 'itemset': list of item indices in the desired rule.
                - 'confidence': (optional) confidence level of the desired rule.

        Returns
        -------
        tuple of (float, float, float)
            - max_rule_gain : float
                The maximum rule gain computed as the sum of rule gain and target gain.
            - realistic_rule_gain : float
                The rule gain adjusted with a realistic target gain, incorporating rule confidences.
            - realistic_rule_gain_dataset : float
                The dataset-level realistic rule gain, computed by scaling realistic rule gain with the estimated
                number of transactions.
        """
        u_undesired = 0.0
        # Sum intrinsic utilities for each item index in the undesired rule's itemset.
        for idx in undesired_rule.get('itemset', []):
            intrinsic_value = self.intrinsic_utility_table.get(idx, 0.0)
            u_undesired += intrinsic_value

        # Initialize the desired rule utility.
        u_desired = 0.0
        # Sum intrinsic utilities for each item index in the desired rule's itemset.
        for idx in desired_rule.get('itemset', []):
            intrinsic_value = self.intrinsic_utility_table.get(idx, 0.0)
            u_desired += intrinsic_value

        # Initialize additional transition gain.
        transition_gain = 0.0
        # Iterate over corresponding item indices from undesired and desired itemsets.
        for u_idx, d_idx in zip(undesired_rule.get('itemset', []), desired_rule.get('itemset', [])):
            # Only add transition gain if the indices differ (indicating a change in a flexible attribute).
            if u_idx != d_idx:
                trans_value = self.transition_utility_table.get((u_idx, d_idx), 0.0)
                transition_gain += trans_value

        rule_gain = u_desired - u_undesired + transition_gain

        # Target utility
        u_undesired_target = self.intrinsic_utility_table.get(self.undesired_state, 0.0)
        u_desired_target = self.intrinsic_utility_table.get(self.desired_state, 0.0)
        transition_gain_target = self.transition_utility_table.get((self.undesired_state, self.desired_state), 0.0)

        target_gain = u_desired_target - u_undesired_target + transition_gain_target

        # Realistic target gain
        undesired_rule_confidence = undesired_rule.get('confidence', 0.0)
        desired_rule_confidence = desired_rule.get('confidence', 0.0)
        target_gain_realistic = (desired_rule_confidence - (1 - undesired_rule_confidence)) * target_gain

        # Rule gain
        max_rule_gain = rule_gain + target_gain
        realistic_rule_gain = rule_gain + target_gain_realistic

        # Compute dataset-level realistic gain.
        support = undesired_rule.get('support', 0)
        if undesired_rule_confidence > 0:
            transactions = support / undesired_rule_confidence
        else:
            transactions = 0.0
        realistic_rule_gain_dataset = transactions * realistic_rule_gain

        return max_rule_gain, realistic_rule_gain, realistic_rule_gain_dataset

    def compute_action_rule_measures(
        self, support_undesired, confidence_undesired, support_desired, confidence_desired
    ):
        """
        Compute the support and confidence for an action rule formed from an undesired rule and a desired rule.

        The action rule is derived by pairing a classification rule that leads to an undesired outcome
        with a classification rule that leads to a desired outcome. In this formulation, the support
        of the action rule is defined as the minimum of the supports of the two component rules, and
        the confidence of the action rule is defined as the product of their confidences.

        Parameters
        ----------
        support_undesired : float
            The support of the undesired rule (e.g., count or relative frequency).
        confidence_undesired : float
            The confidence of the undesired rule (a value between 0 and 1).
        support_desired : float
            The support of the desired rule.
        confidence_desired : float
            The confidence of the desired rule (a value between 0 and 1).

        Returns
        -------
        tuple of (float, float)
            A tuple containing:
                - action_support : float
                    The support of the action rule, computed as min(support_undesired, support_desired).
                - action_confidence : float
                    The confidence of the action rule, computed as confidence_undesired * confidence_desired.

        """
        action_support = min(support_undesired, support_desired)
        action_confidence = confidence_undesired * confidence_desired
        return action_support, action_confidence

__init__(undesired_state, desired_state, columns, count_transactions, intrinsic_utility_table=None, transition_utility_table=None)

Initialize the Rules class with the specified undesired and desired states, columns, and transaction count.

Parameters:

Name	Type	Description	Default
undesired_state	str	The undesired state of the target attribute.	required
desired_state	str	The desired state of the target attribute.	required
columns	list	List of columns in the dataset.	required
count_transactions	int	The number of transactions in the data.	required
intrinsic_utility_table	dict	(attribute, value) -> float A lookup table for the intrinsic utility of each attribute-value pair. If None, no intrinsic utility is considered.	None
transition_utility_table	dict	(attribute, from_value, to_value) -> float A lookup table for cost/gain of transitions between values. If None, no transition utility is considered.	None

Notes

The classification_rules attribute is initialized as a defaultdict with a lambda function that creates dictionaries for 'desired' and 'undesired' states.

Source code in src/action_rules/rules/rules.py

def __init__(
    self,
    undesired_state: str,
    desired_state: str,
    columns: list,
    count_transactions: int,
    intrinsic_utility_table: Optional[dict] = None,
    transition_utility_table: Optional[dict] = None,
):
    """
    Initialize the Rules class with the specified undesired and desired states, columns, and transaction count.

    Parameters
    ----------
    undesired_state : str
        The undesired state of the target attribute.
    desired_state : str
        The desired state of the target attribute.
    columns : list
        List of columns in the dataset.
    count_transactions : int
        The number of transactions in the data.
    intrinsic_utility_table : dict, optional
        (attribute, value) -> float
        A lookup table for the intrinsic utility of each attribute-value pair.
        If None, no intrinsic utility is considered.
    transition_utility_table : dict, optional
        (attribute, from_value, to_value) -> float
        A lookup table for cost/gain of transitions between values.
        If None, no transition utility is considered.

    Notes
    -----
    The classification_rules attribute is initialized as a defaultdict with a lambda function that creates
    dictionaries for 'desired' and 'undesired' states.
    """
    self.classification_rules = defaultdict(lambda: {'desired': [], 'undesired': []})  # type: defaultdict
    self.undesired_state = undesired_state
    self.columns = columns
    self.desired_state = desired_state
    self.action_rules = []  # type: list
    self.undesired_prefixes_without_conf = set()  # type: set
    self.desired_prefixes_without_conf = set()  # type: set
    self.count_transactions = count_transactions
    self.intrinsic_utility_table = intrinsic_utility_table or {}
    self.transition_utility_table = transition_utility_table or {}
    self.has_utility_tables = bool(self.intrinsic_utility_table or self.transition_utility_table)

add_classification_rules(new_ar_prefix, itemset_prefix, undesired_states, desired_states)

Add classification rules for undesired and desired states.

Parameters:

Name	Type	Description	Default
new_ar_prefix	tuple	Prefix of the action rule.	required
itemset_prefix	tuple	Prefix of the itemset.	required
undesired_states	list	List of dictionaries containing undesired state information.	required
desired_states	list	List of dictionaries containing desired state information.	required

Notes

This method updates the classification_rules attribute with new rules based on the provided undesired and desired states. Each state is represented as a dictionary containing item, support, confidence, and target information.

Source code in src/action_rules/rules/rules.py

def add_classification_rules(self, new_ar_prefix, itemset_prefix, undesired_states, desired_states):
    """
    Add classification rules for undesired and desired states.

    Parameters
    ----------
    new_ar_prefix : tuple
        Prefix of the action rule.
    itemset_prefix : tuple
        Prefix of the itemset.
    undesired_states : list
        List of dictionaries containing undesired state information.
    desired_states : list
        List of dictionaries containing desired state information.

    Notes
    -----
    This method updates the classification_rules attribute with new rules based on the provided
    undesired and desired states. Each state is represented as a dictionary containing item, support,
    confidence, and target information.
    """
    for undesired_item in undesired_states:
        new_itemset_prefix = itemset_prefix + (undesired_item['item'],)
        self.classification_rules[new_ar_prefix]['undesired'].append(
            {
                'itemset': new_itemset_prefix,
                'support': undesired_item['support'],
                'confidence': undesired_item['confidence'],
                'target': self.undesired_state,
            }
        )
    for desired_item in desired_states:
        new_itemset_prefix = itemset_prefix + (desired_item['item'],)
        self.classification_rules[new_ar_prefix]['desired'].append(
            {
                'itemset': new_itemset_prefix,
                'support': desired_item['support'],
                'confidence': desired_item['confidence'],
                'target': self.desired_state,
            }
        )

add_prefix_without_conf(prefix, is_desired)

Add a prefix to the set of prefixes without conflicts.

Parameters:

Name	Type	Description	Default
prefix	tuple	The prefix to be added.	required
is_desired	bool	If True, add the prefix to the desired prefixes set; otherwise, add it to the undesired prefixes set.	required

Notes

This method is useful for keeping track of prefixes that have no conflicting rules and can be used directly in rule generation.

Source code in src/action_rules/rules/rules.py

def add_prefix_without_conf(self, prefix: tuple, is_desired: bool):
    """
    Add a prefix to the set of prefixes without conflicts.

    Parameters
    ----------
    prefix : tuple
        The prefix to be added.
    is_desired : bool
        If True, add the prefix to the desired prefixes set; otherwise, add it to the undesired prefixes set.

    Notes
    -----
    This method is useful for keeping track of prefixes that have no conflicting rules and can be
    used directly in rule generation.
    """
    if is_desired:
        self.desired_prefixes_without_conf.add(prefix)
    else:
        self.undesired_prefixes_without_conf.add(prefix)

calculate_confidence(support, opposite_support)

Calculate the confidence of a rule.

Parameters:

Name	Type	Description	Default
support	int	The support value for the desired or undesired state.	required
opposite_support	int	The support value for the opposite state.	required

Returns:

Type	Description
float	The confidence value calculated as support / (support + opposite_support). Returns 0 if the sum of support and opposite_support is 0.

Notes

Confidence is a measure of the reliability of a rule. A higher confidence indicates a stronger association between the conditions of the rule and the target state.

Source code in src/action_rules/rules/rules.py

def calculate_confidence(self, support, opposite_support):
    """
    Calculate the confidence of a rule.

    Parameters
    ----------
    support : int
        The support value for the desired or undesired state.
    opposite_support : int
        The support value for the opposite state.

    Returns
    -------
    float
        The confidence value calculated as support / (support + opposite_support).
        Returns 0 if the sum of support and opposite_support is 0.

    Notes
    -----
    Confidence is a measure of the reliability of a rule. A higher confidence indicates a stronger
    association between the conditions of the rule and the target state.
    """
    if support + opposite_support == 0:
        return 0
    return support / (support + opposite_support)

calculate_uplift(undesired_support, undesired_confidence, desired_confidence)

Calculate the uplift of an action rule.

Parameters:

Name	Type	Description	Default
undesired_support	int	The support value for the undesired state.	required
undesired_confidence	float	The confidence value for the undesired state.	required
desired_confidence	float	The confidence value for the desired state.	required

Returns:

Type	Description
float	The uplift value calculated as: ((desired_confidence - (1 - undesired_confidence)) * (undesired_support / undesired_confidence)) / self.count_transactions.

Notes

Uplift measures the increase in the probability of achieving the desired state when applying the action rule compared to not applying it. It is used to assess the effectiveness of the rule.

Source code in src/action_rules/rules/rules.py

def calculate_uplift(self, undesired_support: int, undesired_confidence: float, desired_confidence: float) -> float:
    """
    Calculate the uplift of an action rule.

    Parameters
    ----------
    undesired_support : int
        The support value for the undesired state.
    undesired_confidence : float
        The confidence value for the undesired state.
    desired_confidence : float
        The confidence value for the desired state.

    Returns
    -------
    float
        The uplift value calculated as:
        ((desired_confidence - (1 - undesired_confidence)) * (undesired_support / undesired_confidence))
        / self.count_transactions.

    Notes
    -----
    Uplift measures the increase in the probability of achieving the desired state when applying the action rule
    compared to not applying it. It is used to assess the effectiveness of the rule.
    """
    return (
        (desired_confidence - (1 - undesired_confidence)) * (undesired_support / undesired_confidence)
    ) / self.count_transactions

compute_action_rule_measures(support_undesired, confidence_undesired, support_desired, confidence_desired)

Compute the support and confidence for an action rule formed from an undesired rule and a desired rule.

The action rule is derived by pairing a classification rule that leads to an undesired outcome with a classification rule that leads to a desired outcome. In this formulation, the support of the action rule is defined as the minimum of the supports of the two component rules, and the confidence of the action rule is defined as the product of their confidences.

Parameters:

Name	Type	Description	Default
support_undesired	float	The support of the undesired rule (e.g., count or relative frequency).	required
confidence_undesired	float	The confidence of the undesired rule (a value between 0 and 1).	required
support_desired	float	The support of the desired rule.	required
confidence_desired	float	The confidence of the desired rule (a value between 0 and 1).	required

Returns:

Type	Description
tuple of (float, float)	A tuple containing: - action_support : float The support of the action rule, computed as min(support_undesired, support_desired). - action_confidence : float The confidence of the action rule, computed as confidence_undesired * confidence_desired.

Source code in src/action_rules/rules/rules.py

def compute_action_rule_measures(
    self, support_undesired, confidence_undesired, support_desired, confidence_desired
):
    """
    Compute the support and confidence for an action rule formed from an undesired rule and a desired rule.

    The action rule is derived by pairing a classification rule that leads to an undesired outcome
    with a classification rule that leads to a desired outcome. In this formulation, the support
    of the action rule is defined as the minimum of the supports of the two component rules, and
    the confidence of the action rule is defined as the product of their confidences.

    Parameters
    ----------
    support_undesired : float
        The support of the undesired rule (e.g., count or relative frequency).
    confidence_undesired : float
        The confidence of the undesired rule (a value between 0 and 1).
    support_desired : float
        The support of the desired rule.
    confidence_desired : float
        The confidence of the desired rule (a value between 0 and 1).

    Returns
    -------
    tuple of (float, float)
        A tuple containing:
            - action_support : float
                The support of the action rule, computed as min(support_undesired, support_desired).
            - action_confidence : float
                The confidence of the action rule, computed as confidence_undesired * confidence_desired.

    """
    action_support = min(support_undesired, support_desired)
    action_confidence = confidence_undesired * confidence_desired
    return action_support, action_confidence

compute_rule_utility(undesired_rule, desired_rule)

Compute various utility gains for a rule transition from undesired to desired.

The function computes intrinsic utilities for items in both the undesired and desired rule itemsets, calculates a transition gain for changes in flexible attributes, and adjusts these gains using target state utilities and rule confidences to derive realistic gain metrics at both the rule and dataset levels.

Parameters:

Name	Type	Description	Default
undesired_rule	dict	Dictionary representing the undesired rule. Expected keys: - 'itemset': list of item indices in the undesired rule. - 'confidence': (optional) confidence level of the undesired rule. - 'support': (optional) support count of the undesired rule.	required
desired_rule	dict	Dictionary representing the desired rule. Expected keys: - 'itemset': list of item indices in the desired rule. - 'confidence': (optional) confidence level of the desired rule.	required

Returns:

Type	Description
tuple of (float, float, float)	max_rule_gain : float The maximum rule gain computed as the sum of rule gain and target gain. realistic_rule_gain : float The rule gain adjusted with a realistic target gain, incorporating rule confidences. realistic_rule_gain_dataset : float The dataset-level realistic rule gain, computed by scaling realistic rule gain with the estimated number of transactions.

Source code in src/action_rules/rules/rules.py

def compute_rule_utility(self, undesired_rule: dict, desired_rule: dict) -> tuple:
    """
    Compute various utility gains for a rule transition from undesired to desired.

    The function computes intrinsic utilities for items in both the undesired and desired rule itemsets,
    calculates a transition gain for changes in flexible attributes, and adjusts these gains using target
    state utilities and rule confidences to derive realistic gain metrics at both the rule and dataset levels.

    Parameters
    ----------
    undesired_rule : dict
        Dictionary representing the undesired rule. Expected keys:
            - 'itemset': list of item indices in the undesired rule.
            - 'confidence': (optional) confidence level of the undesired rule.
            - 'support': (optional) support count of the undesired rule.
    desired_rule : dict
        Dictionary representing the desired rule. Expected keys:
            - 'itemset': list of item indices in the desired rule.
            - 'confidence': (optional) confidence level of the desired rule.

    Returns
    -------
    tuple of (float, float, float)
        - max_rule_gain : float
            The maximum rule gain computed as the sum of rule gain and target gain.
        - realistic_rule_gain : float
            The rule gain adjusted with a realistic target gain, incorporating rule confidences.
        - realistic_rule_gain_dataset : float
            The dataset-level realistic rule gain, computed by scaling realistic rule gain with the estimated
            number of transactions.
    """
    u_undesired = 0.0
    # Sum intrinsic utilities for each item index in the undesired rule's itemset.
    for idx in undesired_rule.get('itemset', []):
        intrinsic_value = self.intrinsic_utility_table.get(idx, 0.0)
        u_undesired += intrinsic_value

    # Initialize the desired rule utility.
    u_desired = 0.0
    # Sum intrinsic utilities for each item index in the desired rule's itemset.
    for idx in desired_rule.get('itemset', []):
        intrinsic_value = self.intrinsic_utility_table.get(idx, 0.0)
        u_desired += intrinsic_value

    # Initialize additional transition gain.
    transition_gain = 0.0
    # Iterate over corresponding item indices from undesired and desired itemsets.
    for u_idx, d_idx in zip(undesired_rule.get('itemset', []), desired_rule.get('itemset', [])):
        # Only add transition gain if the indices differ (indicating a change in a flexible attribute).
        if u_idx != d_idx:
            trans_value = self.transition_utility_table.get((u_idx, d_idx), 0.0)
            transition_gain += trans_value

    rule_gain = u_desired - u_undesired + transition_gain

    # Target utility
    u_undesired_target = self.intrinsic_utility_table.get(self.undesired_state, 0.0)
    u_desired_target = self.intrinsic_utility_table.get(self.desired_state, 0.0)
    transition_gain_target = self.transition_utility_table.get((self.undesired_state, self.desired_state), 0.0)

    target_gain = u_desired_target - u_undesired_target + transition_gain_target

    # Realistic target gain
    undesired_rule_confidence = undesired_rule.get('confidence', 0.0)
    desired_rule_confidence = desired_rule.get('confidence', 0.0)
    target_gain_realistic = (desired_rule_confidence - (1 - undesired_rule_confidence)) * target_gain

    # Rule gain
    max_rule_gain = rule_gain + target_gain
    realistic_rule_gain = rule_gain + target_gain_realistic

    # Compute dataset-level realistic gain.
    support = undesired_rule.get('support', 0)
    if undesired_rule_confidence > 0:
        transactions = support / undesired_rule_confidence
    else:
        transactions = 0.0
    realistic_rule_gain_dataset = transactions * realistic_rule_gain

    return max_rule_gain, realistic_rule_gain, realistic_rule_gain_dataset

generate_action_rules()

Generate action rules from classification rules.

Notes

This method creates action rules by combining classification rules for undesired and desired states. The uplift for each action rule is calculated using the calculate_uplift method and the result is stored in the action_rules attribute.

Source code in src/action_rules/rules/rules.py

def generate_action_rules(self):
    """
    Generate action rules from classification rules.

    Notes
    -----
    This method creates action rules by combining classification rules for undesired and desired states.
    The uplift for each action rule is calculated using the `calculate_uplift` method and the result is
    stored in the action_rules attribute.
    """
    for attribute_prefix, rules in self.classification_rules.items():
        for desired_rule in rules['desired']:
            for undesired_rule in rules['undesired']:
                # Uplift
                uplift = self.calculate_uplift(
                    undesired_rule['support'],
                    undesired_rule['confidence'],
                    desired_rule['confidence'],
                )
                # Utility
                utility = None
                if self.intrinsic_utility_table is not None or self.transition_utility_table is not None:
                    max_rule_gain, realistic_rule_gain, realistic_rule_gain_dataset = self.compute_rule_utility(
                        undesired_rule, desired_rule
                    )
                    utility = {
                        'max_rule_gain': max_rule_gain,
                        'realistic_rule_gain': realistic_rule_gain,
                        'realistic_dataset_gain': realistic_rule_gain_dataset,
                    }
                # Action rule measures
                ar_support, ar_confidence = self.compute_action_rule_measures(
                    undesired_rule.get('support', 0.0),
                    undesired_rule.get('confidence', 0.0),
                    desired_rule.get('support', 0.0),
                    desired_rule.get('confidence', 0.0),
                )
                self.action_rules.append(
                    {
                        'undesired': undesired_rule,
                        'desired': desired_rule,
                        'uplift': uplift,
                        'support': ar_support,
                        'confidence': ar_confidence,
                        **utility,
                    }
                )

prune_classification_rules(k, stop_list)

Prune classification rules based on their length and update the stop list.

Parameters:

Name	Type	Description	Default
k	int	Length of the attribute prefix.	required
stop_list	list	List of prefixes to stop generating rules for.	required

Notes

This method removes classification rules whose prefix length equals k and either desired or undesired states are empty. The corresponding prefixes are also added to the stop_list to avoid further rule generation.

Source code in src/action_rules/rules/rules.py

def prune_classification_rules(self, k: int, stop_list: Union[list, set]):
    """
    Prune classification rules based on their length and update the stop list.

    Parameters
    ----------
    k : int
        Length of the attribute prefix.
    stop_list : list
        List of prefixes to stop generating rules for.

    Notes
    -----
    This method removes classification rules whose prefix length equals k and either desired or undesired
    states are empty. The corresponding prefixes are also added to the stop_list to avoid further rule generation.
    """
    del_prefixes = []
    for attribute_prefix, rules in self.classification_rules.items():
        if k == len(attribute_prefix):
            len_desired = len(rules['desired'])
            len_undesired = len(rules['undesired'])
            if len_desired == 0 or len_undesired == 0:
                if (len_desired == 0 and attribute_prefix not in self.desired_prefixes_without_conf) or (
                    len_undesired == 0 and attribute_prefix not in self.undesired_prefixes_without_conf
                ):
                    self._add_stop_entry(stop_list, attribute_prefix)
                del_prefixes.append(attribute_prefix)
    for attribute_prefix in del_prefixes:
        del self.classification_rules[attribute_prefix]