diff --git a/customhys/collections/default_initializers.txt b/customhys/collections/default_initializers.txt new file mode 100644 index 0000000..6c1c525 --- /dev/null +++ b/customhys/collections/default_initializers.txt @@ -0,0 +1,12 @@ +('initialize_random', {}, 'all') +('initialize_sobol', {'scramble': True}, 'all') +('initialize_sobol', {'scramble': False}, 'all') +('initialize_halton', {'scramble': True}, 'all') +('initialize_halton', {'scramble': False}, 'all') +('initialize_grid', {}, 'all') +('initialize_lhs', {}, 'all') +('initialize_beta', {'a': 0.5, 'b': 0.5}, 'all') +('initialize_normal', {'mean': 0.5, 'std': 0.25}, 'all') +('initialize_lognormal', {'mean': 0.0, 'sigma': 0.5}, 'all') +('initialize_exponential', {'scale': 0.5}, 'all') +('initialize_rayleigh', {'scale': 0.4}, 'all') diff --git a/customhys/hyperheuristic.py b/customhys/hyperheuristic.py index 608bccc..c524e52 100644 --- a/customhys/hyperheuristic.py +++ b/customhys/hyperheuristic.py @@ -172,6 +172,13 @@ def _select_initial_scheme(self, initial_scheme): return scheme + @property + def initialization_operators(self): + """Return a list of (index, op_tuple) for operators with role='initialize' in the heuristic space.""" + return [ + (i, op_tuple) for i, op_tuple in enumerate(self.heuristic_space) if op._get_role(op_tuple) == "initialize" + ] + def toggle_seq_as_meta(self, as_mh=None): if as_mh is None: self.parameters["as_mh"] = not self.parameters["as_mh"] @@ -202,6 +209,9 @@ def _choose_action(self, current_cardinality, previous_action=None, available_op "Roll", "RollMany", ] + # Allow changing the init operator when the space has at least 2 init operators + if len(self.initialization_operators) >= 2: + available_options.append("ChangeInitOperator") # Black list (to avoid repeating the some actions in a row) if previous_action: @@ -265,6 +275,9 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No """ # Create a new MH with min_cardinality from scratch by using a weights array (if so) # if action is given, it is assumed the way of obtaining this intial solution + init_indices = [i for i, _ in self.initialization_operators] + has_init_ops = len(init_indices) > 0 + if sol is None: if action == "max_frequency": # Each search operator per step corresponds to the most frequent one: uMH weight matrix is required @@ -278,26 +291,42 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No else (self.max_cardinality + self.min_cardinality) // 2 ) + # When init operators exist, index 0 is reserved for one of them. + # Keep at least one operator in the tail so a perturbation sequence can run. + if has_init_ops: + initial_cardinality = max(initial_cardinality, 2) + operators_weights = operators_weights if operators_weights else self.weights + tail_size = initial_cardinality - 1 if has_init_ops else initial_cardinality encoded_neighbour = np.random.choice( self.current_space if (operators_weights is None) else self.num_operators, - initial_cardinality, + tail_size, replace=self.parameters["repeat_operators"], p=operators_weights, ) + if has_init_ops: + encoded_neighbour = np.concatenate( + ([np.random.choice(init_indices)], np.asarray(encoded_neighbour)) + ) + # If sol is an integer, assume that it refers to the cardinality elif isinstance(sol, int): operators_weights = self.weights if operators_weights is None else operators_weights + cardinality = max(sol, 2) if has_init_ops else sol + tail_size = cardinality - 1 if has_init_ops else cardinality encoded_neighbour = np.random.choice( self.current_space if (operators_weights is None) else self.num_operators, - sol, + tail_size, replace=self.parameters["repeat_operators"], p=operators_weights, ) + if has_init_ops: + encoded_neighbour = np.concatenate(([np.random.choice(init_indices)], np.asarray(encoded_neighbour))) + elif isinstance(sol, np.ndarray | list): # Bypass the current weights vector to highlight the ``top`` most relevant ones. if (operators_weights is not None) and (top is not None): @@ -326,7 +355,8 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No # # | operator 1 | operator 2 | operator 3 | ... | operator N | # 0 <--------> 1 <--------> 2 <--------> 3 <-- ... --> N-1 <---------> N - operator_location = np.random.randint(current_cardinality + 1) + min_insert = 1 if has_init_ops else 0 + operator_location = np.random.randint(min_insert, current_cardinality + 1) # Add the selected operator encoded_neighbour = np.array((*sol[:operator_location], selected_operator, *sol[operator_location:])) @@ -336,23 +366,43 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No for _ in range(np.random.randint(1, self.max_cardinality - current_cardinality + 1)): encoded_neighbour = self._obtain_candidate_solution(sol=encoded_neighbour, action="Add") - elif (action == "Remove") and (current_cardinality > self.min_cardinality): - # Delete an operator randomly selected - encoded_neighbour = np.delete(sol, np.random.randint(current_cardinality)) + elif action == "Remove": + min_card = max(self.min_cardinality, 2) if has_init_ops else self.min_cardinality + if current_cardinality > min_card: + remove_idx = ( + np.random.randint(1, current_cardinality) + if has_init_ops + else np.random.randint(current_cardinality) + ) + encoded_neighbour = np.delete(sol, remove_idx) + else: + encoded_neighbour = np.copy(sol) - elif (action == "RemoveLast") and (current_cardinality > self.min_cardinality): - # Delete an operator randomly selected - encoded_neighbour = np.delete(sol, -1) + elif action == "RemoveLast": + min_card = max(self.min_cardinality, 2) if has_init_ops else self.min_cardinality + if current_cardinality > min_card: + encoded_neighbour = np.delete(sol, -1) + else: + encoded_neighbour = np.copy(sol) - elif (action == "RemoveMany") and (current_cardinality > self.min_cardinality + 1): - encoded_neighbour = np.copy(sol) - for _ in range(np.random.randint(1, current_cardinality - self.min_cardinality + 1)): - encoded_neighbour = self._obtain_candidate_solution(sol=encoded_neighbour, action="Remove") + elif action == "RemoveMany": + min_card = max(self.min_cardinality, 2) if has_init_ops else self.min_cardinality + if current_cardinality > min_card + 1: + encoded_neighbour = np.copy(sol) + for _ in range(np.random.randint(1, current_cardinality - min_card + 1)): + encoded_neighbour = self._obtain_candidate_solution(sol=encoded_neighbour, action="Remove") + else: + encoded_neighbour = np.copy(sol) elif action == "Shift": # Perturbate an operator randomly selected excluding the existing ones encoded_neighbour = np.copy(sol) - encoded_neighbour[np.random.randint(current_cardinality)] = np.random.choice( + shift_idx = ( + np.random.randint(1, current_cardinality) + if has_init_ops + else np.random.randint(current_cardinality) + ) + encoded_neighbour[shift_idx] = np.random.choice( np.setdiff1d(self.current_space, sol) if not self.parameters["repeat_operators"] else self.num_operators @@ -366,7 +416,11 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No elif action == "LocalShift": # It only works with the full set # Perturbate an operator randomly selected +/- 1 excluding the existing ones encoded_neighbour = np.copy(sol) - operator_location = np.random.randint(current_cardinality) + operator_location = ( + np.random.randint(1, current_cardinality) + if has_init_ops + else np.random.randint(current_cardinality) + ) neighbour_direction = 1 if random.random() < 0.5 else -1 # +/- 1 selected_operator = (encoded_neighbour[operator_location] + neighbour_direction) % self.num_operators @@ -383,7 +437,22 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No elif (action == "Swap") and (current_cardinality > 1): # Swap two elements randomly chosen - if current_cardinality == 2: + start_idx = 1 if has_init_ops else 0 + tail_len = current_cardinality - start_idx + + if tail_len <= 1: + encoded_neighbour = np.copy(sol) + + elif tail_len == 2: + encoded_neighbour = np.copy(sol) + encoded_neighbour[start_idx:] = encoded_neighbour[start_idx:][::-1] + + elif tail_len > 2: + encoded_neighbour = np.copy(sol) + ind1, ind2 = np.random.choice(np.arange(start_idx, current_cardinality), 2, replace=False) + encoded_neighbour[ind1], encoded_neighbour[ind2] = encoded_neighbour[ind2], encoded_neighbour[ind1] + + elif current_cardinality == 2: encoded_neighbour = np.copy(sol)[::-1] elif current_cardinality > 2: @@ -396,22 +465,41 @@ def _obtain_candidate_solution(self, sol=None, action=None, operators_weights=No elif action == "Mirror": # Mirror the sequence of the encoded_neighbour - encoded_neighbour = np.copy(sol)[::-1] + encoded_neighbour = np.copy(sol) + if has_init_ops: + encoded_neighbour[1:] = encoded_neighbour[1:][::-1] + else: + encoded_neighbour = encoded_neighbour[::-1] elif action == "Roll": # Move a step forward or backward, depending on a random variable, all the sequence - encoded_neighbour = np.roll(sol, 1 if random.random() < 0.5 else -1) + encoded_neighbour = np.copy(sol) + roll_step = 1 if random.random() < 0.5 else -1 + if has_init_ops: + encoded_neighbour[1:] = np.roll(encoded_neighbour[1:], roll_step) + else: + encoded_neighbour = np.roll(encoded_neighbour, roll_step) elif action == "RollMany": # Move many (at random) steps forward or backward, depending on a random variable, all the sequence - encoded_neighbour = np.roll( - sol, np.random.randint(current_cardinality) * (1 if random.random() < 0.5 else -1) - ) + encoded_neighbour = np.copy(sol) + roll_step = np.random.randint(current_cardinality) * (1 if random.random() < 0.5 else -1) + if has_init_ops: + encoded_neighbour[1:] = np.roll(encoded_neighbour[1:], roll_step) + else: + encoded_neighbour = np.roll(encoded_neighbour, roll_step) elif action == "Restart": # Restart the entire sequence encoded_neighbour = self._obtain_candidate_solution(current_cardinality) + elif action == "ChangeInitOperator": + # Swap the initialization operator at position 0 for another one + init_candidates = [i for i in init_indices if i != sol[0]] if sol[0] in init_indices else init_indices + encoded_neighbour = np.copy(sol) + if init_candidates: + encoded_neighbour[0] = np.random.choice(init_candidates) + else: raise HyperheuristicError(f"Invalid action = {action} to perform!") else: @@ -1023,7 +1111,8 @@ def evaluate_candidate_solution(self, encoded_sequence, initial_scheme=None): module for further information. :return float, dict: Performance and raw data. """ - initial_scheme = self._select_initial_scheme(self.initial_scheme if initial_scheme is None else initial_scheme) + _raw_scheme = self.initial_scheme if initial_scheme is None else initial_scheme + initial_scheme = _raw_scheme if isinstance(_raw_scheme, tuple) else self._select_initial_scheme(_raw_scheme) # Decode the sequence corresponding to the hyper/meta-heuristic search_operators = encoded_sequence @@ -1032,6 +1121,23 @@ def evaluate_candidate_solution(self, encoded_sequence, initial_scheme=None): if isinstance(encoded_sequence[0], int | np.integer): search_operators = self.get_operators(np.asarray(encoded_sequence, dtype=int).tolist()) + # Extract an initialization operator from the head of the sequence (if present). + # When found, it overrides initial_scheme and is removed from the search operator list. + if search_operators and op._get_role(search_operators[0]) == "initialize": + init_op = search_operators[0] + initial_scheme = (init_op[0], init_op[1]) # ('scheme_name', {params}) + search_operators = search_operators[1:] + if not search_operators: + raise HyperheuristicError( + "encoded_sequence must contain at least one perturbation operator after the initialization operator" + ) + + # Metaheuristic expects operators as 3-tuples (name, params, selector). + # Keep role handling at HH level and strip optional role suffix here. + search_operators = [ + (op_tuple[0], op_tuple[1], op_tuple[2]) if len(op_tuple) > 3 else op_tuple for op_tuple in search_operators + ] + # Initialise the historical registers historical_data = [] fitness_data = [] diff --git a/customhys/operators.py b/customhys/operators.py index 258ae31..9c70710 100644 --- a/customhys/operators.py +++ b/customhys/operators.py @@ -30,6 +30,16 @@ "genetic_mutation", "genetic_crossover", "linear_system", + "initialize_random", + "initialize_sobol", + "initialize_halton", + "initialize_grid", + "initialize_lhs", + "initialize_beta", + "initialize_normal", + "initialize_lognormal", + "initialize_exponential", + "initialize_rayleigh", ] @@ -53,9 +63,33 @@ def get_operator_aliases(): "spiral_dynamic": "SD", "swarm_dynamic": "PS", "linear_system": "LS", + "initialize_random": "IR", + "initialize_sobol": "IS", + "initialize_halton": "IH", + "initialize_grid": "IG", + "initialize_lhs": "IL", + "initialize_beta": "IB", + "initialize_normal": "IN", + "initialize_lognormal": "IW", + "initialize_exponential": "IE", + "initialize_rayleigh": "IY", }, {"greedy": "g", "all": "d", "metropolis": "m", "probabilistic": "p"} +_ROLE_PREFIXES: dict[str, str] = { + "initialize_": "initialize", +} + + +def _get_role(op_tuple) -> str: + """Return the role of an operator tuple derived from its name prefix.""" + name = op_tuple[0] + for prefix, role in _ROLE_PREFIXES.items(): + if name.startswith(prefix): + return role + return "perturb" + + # %% SEARCH OPERATORS @@ -1156,6 +1190,122 @@ def get_rotation_matrix(dimensions, angle=0.3927): return rotation_matrix +# %% INITIALIZER OPERATORS + + +def initialize_random(pop): + """ + Reinitialise the population using uniform random sampling in [-1, 1]. + + This operator resets all agent positions, functioning as a destructive + restart that explores the full search space. + + :param population pop: population object. + :return: None. + """ + pop.initialise_positions("random") + + +def initialize_sobol(pop, scramble=True): + """ + Reinitialise the population using Sobol quasi-random sequences. + + :param population pop: population object. + :param bool scramble: optional. Whether to apply scrambling. Default is True. + :return: None. + """ + pop.initialise_positions("sobol", scramble=scramble) + + +def initialize_halton(pop, scramble=True): + """ + Reinitialise the population using Halton quasi-random sequences. + + :param population pop: population object. + :param bool scramble: optional. Whether to apply scrambling. Default is True. + :return: None. + """ + pop.initialise_positions("halton", scramble=scramble) + + +def initialize_grid(pop): + """ + Reinitialise the population on the vertices of nested hypercubes. + + :param population pop: population object. + :return: None. + """ + pop.initialise_positions("vertex") + + +def initialize_lhs(pop): + """ + Reinitialise the population using Latin Hypercube Sampling. + + :param population pop: population object. + :return: None. + """ + pop.initialise_positions("lhs") + + +def initialize_beta(pop, a=0.5, b=0.5): + """ + Reinitialise the population using a Beta distribution. + + :param population pop: population object. + :param float a: optional. Alpha parameter of Beta distribution. Default is 0.5. + :param float b: optional. Beta parameter of Beta distribution. Default is 0.5. + :return: None. + """ + pop.initialise_positions("beta", a=a, b=b) + + +def initialize_normal(pop, mean=0.5, std=0.25): + """ + Reinitialise the population using a Normal (Gaussian) distribution. + + :param population pop: population object. + :param float mean: optional. Mean of the distribution on [0, 1]. Default is 0.5. + :param float std: optional. Standard deviation on [0, 1]. Default is 0.25. + :return: None. + """ + pop.initialise_positions("normal", mean=mean, std=std) + + +def initialize_lognormal(pop, mean=0.0, sigma=0.5): + """ + Reinitialise the population using a Lognormal distribution. + + :param population pop: population object. + :param float mean: optional. Mean of the underlying normal distribution. Default is 0.0. + :param float sigma: optional. Standard deviation of the underlying normal. Default is 0.5. + :return: None. + """ + pop.initialise_positions("lognormal", mean=mean, sigma=sigma) + + +def initialize_exponential(pop, scale=0.5): + """ + Reinitialise the population using an Exponential distribution. + + :param population pop: population object. + :param float scale: optional. Scale parameter. Default is 0.5. + :return: None. + """ + pop.initialise_positions("exponential", scale=scale) + + +def initialize_rayleigh(pop, scale=0.4): + """ + Reinitialise the population using a Rayleigh distribution. + + :param population pop: population object. + :param float scale: optional. Scale parameter. Default is 0.4. + :return: None. + """ + pop.initialise_positions("rayleigh", scale=scale) + + class OperatorsError(Exception): """ Simple OperatorError to manage exceptions. @@ -1353,7 +1503,8 @@ def build_operators(heuristics: list | None = None, file_name: str = "operators_ file = open("collections/" + file_name + ".txt", "w", encoding="utf-8") # For each simple heuristic, read their parameters and values - for operator, parameters, selectors in heuristics: + for heuristic in heuristics: + operator, parameters, selectors = heuristic[0], heuristic[1], heuristic[2] # Update the total classes counter total_counters[0] += 1 @@ -1386,7 +1537,7 @@ def build_operators(heuristics: list | None = None, file_name: str = "operators_ elif num_parameters == 0: num_combinations = num_selectors for selector in selectors: - file.write("('{}', {}, '{}')\n".format(operator, "{}", selector)) + file.write(f"('{operator}', {{}}, '{selector}')\n") # Update the total combination counter total_counters[1] += num_combinations * num_selectors @@ -1415,28 +1566,18 @@ def process_operators(simple_heuristics): selectors = [] # For each simple heuristic, read their parameters and values - for operator, parameters, selector in simple_heuristics: + for op_tuple in simple_heuristics: + operator, parameters, selector = op_tuple[0], op_tuple[1], op_tuple[2] # Store selectors selectors.append(selector) - if len(parameters) >= 0: - sep = "," - str_parameters = [] - - for parameter, value in parameters.items(): - - # Check if a value is string - if isinstance(value, str): - str_parameters.append(parameter) - str_parameters.append(f"{parameter}='{value}'") - else: - str_parameters.append(f"{parameter}={value}") - - # Create an executable string with given arguments - full_string = f"{operator}({sep.join(str_parameters)})" - else: - # Create an executable string with default arguments - full_string = f"{operator}()" + str_parameters = [] + for parameter, value in parameters.items(): + if isinstance(value, str): + str_parameters.append(f"{parameter}='{value}'") + else: + str_parameters.append(f"{parameter}={value}") + full_string = f"{operator}({','.join(str_parameters)})" # Store the read operator executable_operators.append(full_string) diff --git a/customhys/population.py b/customhys/population.py index aaf5b95..9cb9c61 100644 --- a/customhys/population.py +++ b/customhys/population.py @@ -295,12 +295,14 @@ def evaluate_fitness(self, problem_function): # ============== # TODO Add more initialisation operators like grid, boundary, etc. - def initialise_positions(self, scheme="random"): + def initialise_positions(self, scheme=None, **kwargs): """ Initialise population by an initialisation scheme. - :param str scheme: optional - Initialisation scheme. Available schemes are: + :param str|tuple scheme: optional + Initialisation scheme. Accepts either a string name or a tuple + ``('scheme_name', {'param': value, ...})`` for parameterised schemes. + Available schemes are: - 'random': Random uniform distribution in [-1,1] (default) - 'vertex': Vertices of nested hyper-cubes - 'lhs': Latin Hypercube Sampling @@ -311,29 +313,43 @@ def initialise_positions(self, scheme="random"): - 'lognormal': Lognormal distribution - 'exponential': Exponential distribution - 'rayleigh': Rayleigh distribution + :param kwargs: optional keyword arguments forwarded to the selected scheme. :returns: None. """ + # Handle tuple format: ('scheme_name', {param: value, ...}) + if isinstance(scheme, tuple): + params = scheme[1] if len(scheme) > 1 else {} + kwargs = {**params, **kwargs} + scheme = scheme[0] + scheme = "random" if scheme is None else str(scheme).strip().lower() + # Strip the 'initialize_' prefix used by operators.py function names + # e.g. 'initialize_sobol' → 'sobol', 'initialize_grid' → 'vertex' + if scheme.startswith("initialize_"): + scheme = scheme[len("initialize_") :] + if scheme == "grid": + scheme = "vertex" + if scheme == "vertex": self._positions = self._grid_matrix(self.num_dimensions, self.num_agents) elif scheme == "lhs": self._positions = self._lhs(self.num_dimensions, self.num_agents) elif scheme == "sobol": - self._positions = self._sobol(self.num_dimensions, self.num_agents) + self._positions = self._sobol(self.num_dimensions, self.num_agents, **kwargs) elif scheme == "halton": - self._positions = self._halton(self.num_dimensions, self.num_agents) + self._positions = self._halton(self.num_dimensions, self.num_agents, **kwargs) elif scheme == "beta": - self._positions = self._beta(self.num_dimensions, self.num_agents) + self._positions = self._beta(self.num_dimensions, self.num_agents, **kwargs) elif scheme == "normal": - self._positions = self._normal(self.num_dimensions, self.num_agents) + self._positions = self._normal(self.num_dimensions, self.num_agents, **kwargs) elif scheme == "lognormal": - self._positions = self._lognormal(self.num_dimensions, self.num_agents) + self._positions = self._lognormal(self.num_dimensions, self.num_agents, **kwargs) elif scheme == "exponential": - self._positions = self._exponential(self.num_dimensions, self.num_agents) + self._positions = self._exponential(self.num_dimensions, self.num_agents, **kwargs) elif scheme == "rayleigh": - self._positions = self._rayleigh(self.num_dimensions, self.num_agents) + self._positions = self._rayleigh(self.num_dimensions, self.num_agents, **kwargs) else: # Default to random if scheme is not recognized self._positions = np.random.uniform(-1, 1, (self.num_agents, self.num_dimensions)) @@ -349,7 +365,11 @@ def _grid_matrix(num_dimensions, num_agents): total_vertices = 2**num_dimensions basic_matrix = ( - 2 * np.array([[int(x) for x in list(format(k, f"0{num_dimensions}b"))] for k in range(total_vertices)]) - 1 + 2 + * np.array( + [[int(x) for x in list(format(k, f"0{num_dimensions}b"))] for k in range(total_vertices)], dtype=float + ) + - 1 ) output_matrix = np.copy(basic_matrix) @@ -389,7 +409,7 @@ def _lhs(num_dimensions, num_agents): return sample * 2 - 1 @staticmethod - def _sobol(num_dimensions, num_agents): + def _sobol(num_dimensions, num_agents, scramble=True): """ Initialise population using Sobol quasi-random sequences. @@ -401,11 +421,12 @@ def _sobol(num_dimensions, num_agents): :param int num_dimensions: Number of dimensions :param int num_agents: Number of agents + :param bool scramble: optional. Whether to apply scrambling. Default is True. :returns: numpy.ndarray """ m = int(np.ceil(np.log2(num_agents))) - sobol = Sobol(d=num_dimensions, scramble=True) + sobol = Sobol(d=num_dimensions, scramble=scramble) sample = sobol.random_base2(m) @@ -414,7 +435,7 @@ def _sobol(num_dimensions, num_agents): return seq * 2 - 1 @staticmethod - def _halton(num_dimensions, num_agents): + def _halton(num_dimensions, num_agents, scramble=True): """ Initialise population using Halton quasi-random sequences. @@ -423,9 +444,10 @@ def _halton(num_dimensions, num_agents): :param int num_dimensions: Number of dimensions :param int num_agents: Number of agents + :param bool scramble: optional. Whether to apply scrambling. Default is True. :returns: numpy.ndarray """ - halton = Halton(d=num_dimensions, scramble=True) + halton = Halton(d=num_dimensions, scramble=scramble) sample = halton.random(n=num_agents) diff --git a/tests/test_hyperheuristic_initialization.py b/tests/test_hyperheuristic_initialization.py new file mode 100644 index 0000000..4295e8e --- /dev/null +++ b/tests/test_hyperheuristic_initialization.py @@ -0,0 +1,353 @@ +""" +Tests for role-based operator classification in Hyperheuristic. + +Covers: +- initialization_operators property +- ChangeInitOperator action in _choose_action and _obtain_candidate_solution +- evaluate_candidate_solution with init op at head (including bug regression) +- build_operators output (no role suffix — role is derived from name prefix) +- SA loop integration: HH selects and uses init ops correctly with real benchmark problems +""" + +import ast + +import numpy as np +import pytest + +from customhys import benchmark_func as bf +from customhys import hyperheuristic as hh +from customhys import operators as op +from customhys.hyperheuristic import HyperheuristicError + +# --------------------------------------------------------------------------- +# Shared operator tuples — 3-tuples only (role derived from name prefix) +# --------------------------------------------------------------------------- + +INIT_OP_SOBOL = ("initialize_sobol", {"scramble": True}, "all") +INIT_OP_HALTON = ("initialize_halton", {"scramble": True}, "all") +INIT_OP_GRID = ("initialize_grid", {}, "all") +PERTURB_OP_RS = ("random_search", {"scale": 1.0, "distribution": "uniform"}, "greedy") +PERTURB_OP_LRW = ( + "local_random_walk", + {"probability": 0.75, "scale": 1.0, "distribution": "gaussian"}, + "greedy", +) + +MIXED_SPACE = [INIT_OP_SOBOL, INIT_OP_HALTON, PERTURB_OP_RS, PERTURB_OP_LRW] +# idx 0 = INIT_OP_SOBOL, idx 1 = INIT_OP_HALTON, idx 2 = PERTURB_OP_RS, idx 3 = PERTURB_OP_LRW + +PURE_PERTURB_SPACE = [PERTURB_OP_RS, PERTURB_OP_LRW] + +SINGLE_INIT_SPACE = [INIT_OP_SOBOL, PERTURB_OP_RS, PERTURB_OP_LRW] +# idx 0 = INIT_OP_SOBOL, idx 1 = PERTURB_OP_RS, idx 2 = PERTURB_OP_LRW + +GRID_MIXED_SPACE = [INIT_OP_GRID, INIT_OP_SOBOL, PERTURB_OP_RS, PERTURB_OP_LRW] +# idx 0 = INIT_OP_GRID, idx 1 = INIT_OP_SOBOL, idx 2 = PERTURB_OP_RS, idx 3 = PERTURB_OP_LRW + +# --------------------------------------------------------------------------- +# Minimal HH parameters — fast but functional +# --------------------------------------------------------------------------- + +MINI_PARAMS = { + "cardinality": 3, + "cardinality_min": 1, + "num_replicas": 1, + "num_iterations": 2, + "num_agents": 3, + "as_mh": False, + "num_steps": 3, + "stagnation_percentage": 0.37, + "max_temperature": 1.0, + "min_temperature": 1e-6, + "cooling_rate": 1e-3, + "temperature_scheme": "fast", + "acceptance_scheme": "exponential", + "allow_weight_matrix": True, + "trial_overflow": False, + "learnt_dataset": None, + "repeat_operators": True, + "verbose": False, + "learning_portion": 0.37, + "solver": "static", +} + + +def make_hh(space, problem=None): + """Create a minimal Hyperheuristic instance with the given heuristic space.""" + if problem is None: + problem = bf.Sphere(variable_num=2).get_formatted_problem() + return hh.Hyperheuristic( + heuristic_space=space, + problem=problem, + parameters=MINI_PARAMS.copy(), + ) + + +# =========================================================================== +# 1. initialization_operators property +# =========================================================================== + + +class TestInitializationOperatorsProperty: + def test_empty_when_no_init_ops(self): + hyper = make_hh(PURE_PERTURB_SPACE) + assert hyper.initialization_operators == [] + + def test_returns_only_init_ops_from_mixed_space(self): + hyper = make_hh(MIXED_SPACE) + assert len(hyper.initialization_operators) == 2 + + def test_returns_correct_indices(self): + hyper = make_hh(MIXED_SPACE) + for idx, op_tuple in hyper.initialization_operators: + assert hyper.heuristic_space[idx] is op_tuple + + def test_single_init_op_returns_one_entry(self): + hyper = make_hh(SINGLE_INIT_SPACE) + assert len(hyper.initialization_operators) == 1 + + def test_all_results_have_initialize_role(self): + hyper = make_hh(MIXED_SPACE) + for _, op_tuple in hyper.initialization_operators: + assert op._get_role(op_tuple) == "initialize" + + +# =========================================================================== +# 2. _choose_action — ChangeInitOperator availability +# =========================================================================== + + +class TestChooseActionWithInitOps: + def test_change_init_op_included_when_2_init_ops(self): + hyper = make_hh(MIXED_SPACE) + seen = {hyper._choose_action(2) for _ in range(200)} + assert "ChangeInitOperator" in seen + + def test_change_init_op_excluded_when_1_init_op(self): + hyper = make_hh(SINGLE_INIT_SPACE) + for _ in range(50): + assert hyper._choose_action(2) != "ChangeInitOperator" + + def test_change_init_op_excluded_when_no_init_ops(self): + hyper = make_hh(PURE_PERTURB_SPACE) + for _ in range(50): + assert hyper._choose_action(2) != "ChangeInitOperator" + + +# =========================================================================== +# 3. _obtain_candidate_solution — ChangeInitOperator action +# =========================================================================== + + +class TestObtainCandidateSolutionChangeInitOperator: + def test_preserves_sequence_length(self): + hyper = make_hh(MIXED_SPACE) + sol = np.array([0, 2, 3]) + result = hyper._obtain_candidate_solution(sol=sol, action="ChangeInitOperator") + assert len(result) == len(sol) + + def test_tail_of_sequence_unchanged(self): + hyper = make_hh(MIXED_SPACE) + sol = np.array([0, 2, 3]) + result = hyper._obtain_candidate_solution(sol=sol, action="ChangeInitOperator") + np.testing.assert_array_equal(result[1:], sol[1:]) + + def test_first_element_is_a_valid_init_op_index(self): + hyper = make_hh(MIXED_SPACE) + init_indices = {i for i, _ in hyper.initialization_operators} + sol = np.array([0, 2]) + result = hyper._obtain_candidate_solution(sol=sol, action="ChangeInitOperator") + assert int(result[0]) in init_indices + + def test_new_first_element_can_differ_from_original(self): + hyper = make_hh(MIXED_SPACE) + sol = np.array([0, 2]) + seen_first = {int(hyper._obtain_candidate_solution(sol=sol, action="ChangeInitOperator")[0]) for _ in range(50)} + assert 1 in seen_first + + +class TestHeadInitializerConstraint: + def test_shift_keeps_initializer_at_head(self): + hyper = make_hh(MIXED_SPACE) + init_indices = {i for i, _ in hyper.initialization_operators} + sol = np.array([0, 2, 3]) + result = hyper._obtain_candidate_solution(sol=sol, action="Shift") + assert int(result[0]) in init_indices + + def test_swap_keeps_initializer_at_head(self): + hyper = make_hh(MIXED_SPACE) + init_indices = {i for i, _ in hyper.initialization_operators} + sol = np.array([0, 2, 3]) + result = hyper._obtain_candidate_solution(sol=sol, action="Swap") + assert int(result[0]) in init_indices + + def test_mirror_keeps_initializer_at_head(self): + hyper = make_hh(MIXED_SPACE) + init_indices = {i for i, _ in hyper.initialization_operators} + sol = np.array([0, 2, 3]) + result = hyper._obtain_candidate_solution(sol=sol, action="Mirror") + assert int(result[0]) in init_indices + + def test_roll_keeps_initializer_at_head(self): + hyper = make_hh(MIXED_SPACE) + init_indices = {i for i, _ in hyper.initialization_operators} + sol = np.array([0, 2, 3]) + result = hyper._obtain_candidate_solution(sol=sol, action="Roll") + assert int(result[0]) in init_indices + + def test_initial_solution_starts_with_initializer(self): + hyper = make_hh(MIXED_SPACE) + init_indices = {i for i, _ in hyper.initialization_operators} + result = hyper._obtain_candidate_solution() + assert int(result[0]) in init_indices + + +# =========================================================================== +# 4. evaluate_candidate_solution — init op extraction +# =========================================================================== + + +class TestEvaluateCandidateSolutionWithInitOp: + def test_raises_when_sequence_is_only_init_op(self): + hyper = make_hh(MIXED_SPACE) + with pytest.raises(HyperheuristicError): + hyper.evaluate_candidate_solution(np.array([0])) + + def test_valid_sequence_with_init_op_returns_performance(self): + hyper = make_hh(MIXED_SPACE) + perf, details = hyper.evaluate_candidate_solution(np.array([0, 2])) + assert isinstance(perf, float) + assert "fitness" in details + + def test_pure_perturb_sequence_still_works(self): + hyper = make_hh(MIXED_SPACE) + perf, details = hyper.evaluate_candidate_solution(np.array([2, 3])) + assert isinstance(perf, float) + assert "fitness" in details + + def test_sobol_init_op_produces_finite_performance(self): + hyper = make_hh(MIXED_SPACE) + perf, details = hyper.evaluate_candidate_solution(np.array([0, 2])) + assert np.isfinite(perf) + assert all(np.isfinite(f) for f in details["fitness"]) + + def test_grid_init_op_does_not_raise(self): + hyper = make_hh(GRID_MIXED_SPACE) + perf, _ = hyper.evaluate_candidate_solution(np.array([0, 2])) + assert isinstance(perf, float) + + def test_initializer_in_tail_is_allowed(self): + hyper = make_hh(MIXED_SPACE) + perf, details = hyper.evaluate_candidate_solution(np.array([0, 1, 2])) + assert isinstance(perf, float) + assert all(np.isfinite(f) for f in details["fitness"]) + + +# =========================================================================== +# 5. build_operators — no role suffix in generated file (role from name prefix) +# =========================================================================== + + +class TestBuildOperatorsOutput: + def test_init_operator_writes_no_role_suffix(self, tmp_path, monkeypatch): + """build_operators writes clean 3-tuples; role is derived from name, not stored.""" + monkeypatch.chdir(tmp_path) + heuristics = [("initialize_random", {}, ["all"])] + op.build_operators(heuristics=heuristics, file_name="test_init") + content = (tmp_path / "collections" / "test_init.txt").read_text(encoding="utf-8") + assert "'initialize'" not in content + assert "'perturb'" not in content + + def test_perturb_operator_writes_no_role_suffix(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + heuristics = [("random_search", {"scale": [1.0]}, ["greedy"])] + op.build_operators(heuristics=heuristics, file_name="test_perturb") + content = (tmp_path / "collections" / "test_perturb.txt").read_text(encoding="utf-8") + assert "'initialize'" not in content + assert "'perturb'" not in content + + def test_generated_file_entries_have_correct_role_via_prefix(self, tmp_path, monkeypatch): + """Entries written by build_operators must have the correct role when read back.""" + monkeypatch.chdir(tmp_path) + heuristics = [ + ("initialize_lhs", {}, ["all"]), + ("random_search", {"scale": [1.0]}, ["greedy"]), + ] + op.build_operators(heuristics=heuristics, file_name="test_mixed") + content = (tmp_path / "collections" / "test_mixed.txt").read_text(encoding="utf-8") + entries = [ast.literal_eval(line.strip()) for line in content.splitlines() if line.strip()] + roles = [op._get_role(e) for e in entries] + assert roles[0] == "initialize" + assert roles[1] == "perturb" + + +# =========================================================================== +# 6. SA loop integration — HH selects and uses init ops with real problems +# =========================================================================== + + +@pytest.mark.parametrize( + "problem_cls,dims", + [ + (bf.Sphere, 2), + (bf.Rastrigin, 2), + (bf.Ackley1, 2), + ], +) +class TestHyperheuristicSolveWithInitOps: + """ + Integration tests: run the full SA loop and verify that initializer operators + are selected as part of the best solution. + """ + + def test_solve_completes_without_error(self, problem_cls, dims): + problem = problem_cls(variable_num=dims).get_formatted_problem() + hyper = make_hh(MIXED_SPACE, problem=problem) + encoded, performance, _, _ = hyper.solve(save_steps=False) + assert encoded is not None + assert isinstance(performance, float) + + def test_solve_returns_finite_performance(self, problem_cls, dims): + problem = problem_cls(variable_num=dims).get_formatted_problem() + hyper = make_hh(MIXED_SPACE, problem=problem) + _, performance, _, _ = hyper.solve(save_steps=False) + assert np.isfinite(performance) + + def test_best_encoded_solution_starts_with_init_op(self, problem_cls, dims): + """The encoded solution from the SA loop must have an init op at position 0.""" + problem = problem_cls(variable_num=dims).get_formatted_problem() + hyper = make_hh(MIXED_SPACE, problem=problem) + init_indices = {i for i, _ in hyper.initialization_operators} + encoded, _, _, _ = hyper.solve(save_steps=False) + assert ( + int(encoded[0]) in init_indices + ), f"Expected encoded[0] to be an init op index {init_indices}, got {encoded[0]}" + + def test_best_solution_operator_names_include_init_prefix(self, problem_cls, dims): + """The decoded best solution must contain an initialize_* operator at position 0.""" + problem = problem_cls(variable_num=dims).get_formatted_problem() + hyper = make_hh(MIXED_SPACE, problem=problem) + encoded, _, _, _ = hyper.solve(save_steps=False) + first_op_name = hyper.get_operators(encoded.tolist())[0][0] + assert first_op_name.startswith( + "initialize_" + ), f"Expected first operator to start with 'initialize_', got '{first_op_name}'" + + +class TestHyperheuristicSolveWithoutInitOps: + """Ensure backward compatibility: pure perturb space still works normally.""" + + def test_solve_without_init_ops_completes(self): + problem = bf.Sphere(variable_num=2).get_formatted_problem() + hyper = make_hh(PURE_PERTURB_SPACE, problem=problem) + _, performance, _, _ = hyper.solve(save_steps=False) + assert np.isfinite(performance) + + def test_encoded_solution_has_no_init_op_when_space_is_pure_perturb(self): + problem = bf.Sphere(variable_num=2).get_formatted_problem() + hyper = make_hh(PURE_PERTURB_SPACE, problem=problem) + encoded, _, _, _ = hyper.solve(save_steps=False) + init_indices = {i for i, _ in hyper.initialization_operators} + assert len(init_indices) == 0 + for idx in encoded: + assert op._get_role(hyper.heuristic_space[int(idx)]) == "perturb" diff --git a/tests/test_initialization_operators.py b/tests/test_initialization_operators.py new file mode 100644 index 0000000..7a7581f --- /dev/null +++ b/tests/test_initialization_operators.py @@ -0,0 +1,210 @@ +""" +Tests for the role-based operator classification and initialization operators. + +Covers: +- _get_role() derives role from operator name prefix via _ROLE_PREFIXES +- process_operators handles 3-tuples +- All initialize_* functions exist and reset positions correctly +- population.initialise_positions accepts **kwargs and tuple format +- Hyperheuristic.initialization_operators property +- evaluate_candidate_solution extracts init op from sequence +""" + +import ast + +import numpy as np +import pytest + +from customhys import operators as op +from customhys import population as pp + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_pop(num_dimensions: int = 3, num_agents: int = 10) -> pp.Population: + boundaries = ([-5.0] * num_dimensions, [5.0] * num_dimensions) + return pp.Population(boundaries, num_agents=num_agents) + + +# --------------------------------------------------------------------------- +# _get_role +# --------------------------------------------------------------------------- + + +class TestGetRole: + def test_perturb_operator_returns_perturb(self): + t = ("random_search", {"scale": 0.1}, "greedy") + assert op._get_role(t) == "perturb" + + def test_initialize_prefix_returns_initialize(self): + t = ("initialize_random", {}, "all") + assert op._get_role(t) == "initialize" + + def test_initialize_prefix_applies_to_all_variants(self): + for name in ["initialize_sobol", "initialize_halton", "initialize_lhs", "initialize_beta"]: + assert op._get_role((name, {}, "all")) == "initialize" + + def test_unknown_prefix_defaults_to_perturb(self): + t = ("some_new_operator", {}, "greedy") + assert op._get_role(t) == "perturb" + + def test_role_prefixes_dict_is_extensible(self): + """_ROLE_PREFIXES is the single source of truth for role derivation.""" + assert "initialize_" in op._ROLE_PREFIXES + assert op._ROLE_PREFIXES["initialize_"] == "initialize" + + def test_all_init_ops_in_default_initializers_txt_have_initialize_role(self): + """Every entry in default_initializers.txt must have role='initialize' via name prefix.""" + import os + + path = os.path.join( + os.path.dirname(__file__), + "..", + "customhys", + "collections", + "default_initializers.txt", + ) + with open(path, encoding="utf-8") as f: + entries = [ast.literal_eval(line.strip()) for line in f if line.strip()] + for entry in entries: + assert op._get_role(entry) == "initialize", f"Expected 'initialize' role for {entry}" + + +# --------------------------------------------------------------------------- +# process_operators — backward compat with 3-tuples and 4-tuples +# --------------------------------------------------------------------------- + + +class TestProcessOperators: + def test_three_tuple(self): + seq = [("random_search", {"scale": 0.1}, "greedy")] + perturbators, selectors = op.process_operators(seq) + assert len(perturbators) == 1 + assert len(selectors) == 1 + assert selectors[0] == "greedy" + + def test_initialize_tuple(self): + seq = [("initialize_random", {}, "all")] + perturbators, selectors = op.process_operators(seq) + assert len(perturbators) == 1 + assert selectors[0] == "all" + + def test_mixed_operators(self): + seq = [ + ("random_search", {"scale": 0.1}, "greedy"), + ("initialize_sobol", {"scramble": True}, "all"), + ] + perturbators, selectors = op.process_operators(seq) + assert len(perturbators) == 2 + assert len(selectors) == 2 + + +# --------------------------------------------------------------------------- +# initialize_* functions via operator interface +# --------------------------------------------------------------------------- + + +class TestInitializeOperators: + @pytest.mark.parametrize( + "func_name", + [ + "initialize_random", + "initialize_sobol", + "initialize_halton", + "initialize_grid", + "initialize_lhs", + "initialize_beta", + "initialize_normal", + "initialize_lognormal", + "initialize_exponential", + "initialize_rayleigh", + ], + ) + def test_function_exists_in_all(self, func_name): + assert func_name in op.__all__ + + @pytest.mark.parametrize( + "func_name", + [ + "initialize_random", + "initialize_sobol", + "initialize_halton", + "initialize_grid", + "initialize_lhs", + "initialize_beta", + "initialize_normal", + "initialize_lognormal", + "initialize_exponential", + "initialize_rayleigh", + ], + ) + def test_function_resets_positions_to_finite_values(self, func_name): + pop = _make_pop() + func = getattr(op, func_name) + func(pop) + assert np.all(np.isfinite(pop.positions)) + assert pop.positions.shape == (10, 3) + + def test_initialize_sobol_scramble_false(self): + """initialize_sobol passes scramble kwarg to population.""" + pop = _make_pop() + op.initialize_sobol(pop, scramble=False) + assert np.all(np.isfinite(pop.positions)) + + def test_initialize_halton_scramble_false(self): + pop = _make_pop() + op.initialize_halton(pop, scramble=False) + assert np.all(np.isfinite(pop.positions)) + + def test_initialize_beta_custom_params(self): + pop = _make_pop() + op.initialize_beta(pop, a=0.2, b=0.8) + assert np.all(np.isfinite(pop.positions)) + + def test_initialize_normal_custom_params(self): + pop = _make_pop() + op.initialize_normal(pop, mean=0.3, std=0.1) + assert np.all(np.isfinite(pop.positions)) + + +# --------------------------------------------------------------------------- +# population.initialise_positions — tuple format and **kwargs +# --------------------------------------------------------------------------- + + +class TestInitialisePositionsTupleFormat: + def test_tuple_format_with_params(self): + pop = _make_pop() + pop.initialise_positions(("beta", {"a": 0.3, "b": 0.7})) + assert np.all(np.isfinite(pop.positions)) + + def test_tuple_format_without_params(self): + pop = _make_pop() + pop.initialise_positions(("sobol", {})) + assert np.all(np.isfinite(pop.positions)) + + def test_kwargs_forwarded_to_sobol(self): + pop1 = _make_pop(num_agents=8) + pop2 = _make_pop(num_agents=8) + pop1.initialise_positions("sobol", scramble=True) + pop2.initialise_positions("sobol", scramble=True) + # Both should produce finite arrays (determinism not guaranteed) + assert np.all(np.isfinite(pop1.positions)) + assert np.all(np.isfinite(pop2.positions)) + + def test_kwargs_forwarded_to_halton(self): + pop = _make_pop() + pop.initialise_positions("halton", scramble=False) + assert np.all(np.isfinite(pop.positions)) + + def test_kwargs_forwarded_to_normal(self): + pop = _make_pop() + pop.initialise_positions("normal", mean=0.6, std=0.05) + assert np.all(np.isfinite(pop.positions)) + + def test_none_scheme_uses_random(self): + pop = _make_pop() + pop.initialise_positions(None) + assert np.all(np.isfinite(pop.positions))