Skip to content

DILS: Dual Iterative Local Search para Clustering con Restricciones

DILS (Dual Iterative Local Search) es un algoritmo metaheurístico diseñado específicamente para abordar el problema de clustering con restricciones de tipo instancia-instancia (must-link y cannot-link). Se basa en un esquema de búsqueda local iterada que busca mantener un equilibrio entre exploración global y explotación local del espacio de soluciones.

Motivación

El problema de clustering con restricciones es NP-completo incluso para instancias simples. Las soluciones exactas no escalan, y los métodos clásicos pueden quedar atrapados en óptimos locales. DILS fue propuesto como una alternativa robusta que combina técnicas de búsqueda local con mecanismos de escape de óptimos locales.

Características Principales

  • Capaz de manejar grandes volúmenes de restricciones (miles o millones).
  • Emplea un esquema dual: mantiene dos soluciones para mejorar la diversidad.
  • Balance entre exploración (diversificación) y explotación (intensificación).
  • Usa una función de penalización para restricciones violadas (soft-constraints).

Estructura del Algoritmo

DILS se basa en la metaheurística de Búsqueda Local Iterada (ILS), mejorándola con elementos duales y adaptativos:

1. Representación

  • Cada solución es un vector de asignaciones de clusters: $$ \mathbf{l} = [l_1, l_2, ..., l_n] \quad \text{con } l_i \in {1, ..., K} $$

2. Función Objetivo

DILS minimiza una función compuesta por dos términos: $$ f(\mathbf{l}) = \text{Varianza intra-cluster} + \alpha \cdot \text{Infeasibility} $$ Donde: - La varianza mide la cohesión interna. - La infeasibility penaliza violaciones a restricciones must-link y cannot-link.

3. Búsqueda Local

Se aplica una búsqueda local sobre la solución actual, explorando vecinos generados por pequeños cambios (por ejemplo, reasignación de puntos a otros clusters).

4. Perturbación

Para escapar de óptimos locales, se perturba la solución parcialmente (p. ej., se reasignan aleatoriamente algunos puntos), iniciando una nueva búsqueda local.

5. Dualidad

DILS mantiene dos soluciones en paralelo. En cada iteración se perturba y mejora una de ellas, y se explora el espacio entre ambas mediante recombinación y aceptación adaptativa.

6. Criterio de Parada

Número máximo de iteraciones sin mejora o tiempo límite.

Ventajas de DILS

  • Altamente escalable y robusto ante ruido.
  • No requiere número fijo de restricciones.
  • Admite restricciones inconsistentes o redundantes.

Referencia

González-Almagro et al. (2020). "DILS: Constrained clustering through dual iterative local search", Computers and Operations Research, 121, 104979.

API

Bases: BaseEstimator

Dual Intra-cluster Local Search (DILS) clustering algorithm.

This algorithm do an iterative local search using a dual approach of bes and worst chromosomes. It generate a random mutation in the worst chromosome to explore

Attributes:

Name Type Description
n_clusters int

The number of clusters to form.
init str

Initialization method ('random' or 'custom').
distance callable

Distance metric function.
tol float

Convergence tolerance.
custom_initial_centroids ndarray

Custom initial centroids.
constraints Sequence[Sequence]

Constraints for clustering.
probability float

Probability for crossover.
similarity_threshold float

Threshold for similarity in fitness.
mutation_size int

Size of the mutation segment.
local_search_max_iter int

Maximum iterations for local search.
best ndarray

Best chromosome found.
worst ndarray

Worst chromosome found.
_fitness ndarray

Fitness values of the chromosomes.
Source code in clustlib/sin/dils.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
class DILS(BaseEstimator):
    """Dual Intra-cluster Local Search (DILS) clustering algorithm.

    This algorithm do an iterative local search using a dual approach of bes and worst
    chromosomes. It generate a random mutation in the worst chromosome to explore

    Attributes:
        n_clusters (int): The number of clusters to form.
        init (str): Initialization method ('random' or 'custom').
        distance (callable): Distance metric function.
        tol (float): Convergence tolerance.
        custom_initial_centroids (numpy.ndarray, optional): Custom initial centroids.
        constraints (Sequence[Sequence], optional): Constraints for clustering.
        probability (float): Probability for crossover.
        similarity_threshold (float): Threshold for similarity in fitness.
        mutation_size (int): Size of the mutation segment.
        local_search_max_iter (int): Maximum iterations for local search.
        best (numpy.ndarray): Best chromosome found.
        worst (numpy.ndarray): Worst chromosome found.
        _fitness (numpy.ndarray): Fitness values of the chromosomes.

    """

    def __init__(
        self,
        n_clusters=8,
        init="random",
        distance="euclidean",
        max_iter=300,
        tol=1e-4,
        custom_init_centroids=None,
        constraints: Sequence[Sequence] | None = None,
        probability=0.2,
        similarity_threshold=0.5,
        mutation_size=10,
        local_search_max_iter=10,
    ):
        self.init = init
        self.distance = match_distance(distance)
        self.tol = tol
        self.custom_initial_centroids = custom_init_centroids
        self.constraints = constraints

        self.n_clusters = n_clusters
        self._evals_done = 0
        self._probability = probability
        self._threshold = similarity_threshold
        self._mutation_size = mutation_size
        self.max_iter = max_iter
        self.local_search_max_iter = local_search_max_iter

        self.best = None
        self.worst = None
        self._fitness = None

    def initialize(self):
        """Initialize the chromosomes and their fitness values.

        This method initializes two chromosomes with random cluster assignments and
        calculates their fitness values.
        """
        cromosomes = np.random.randint(0, self.n_clusters, (2, self.X.shape[0]))
        self._fitness = np.empty(2)
        self._fitness[0] = self.get_single_fitness(cromosomes[0, :])
        self._fitness[1] = self.get_single_fitness(cromosomes[1, :])

        self.best = cromosomes[np.argmin(self._fitness)]
        self.worst = cromosomes[np.argmax(self._fitness)]

    def _intra_cluster_distance(self, labels):
        """Calculate the intra-cluster distance.

        This method calculates the average distance between all points in the same
        cluster.

        Args:
            labels (numpy.ndarray): The labels of the clusters.

        Returns:
            float: The average intra-cluster distance.

        """
        result = 0

        if self.n_clusters == 1:
            result = pdist(self.X, metric=lambda u, v: self.distance(u - v))
            return result.mean()

        if len(labels) != self.X.shape[0]:
            raise ValueError("Labels must match the number of data points.")

        for j in np.unique(labels):
            if self.X[labels == j, :].shape[0] < 2:
                # If there is only one point in the cluster, distance is zero
                result += 0
                continue

            pdist_matrix = pdist(
                self.X[labels == j, :], metric=lambda u, v: self.distance(v - u)
            )
            result += pdist_matrix.mean()

        return result / self.n_clusters if self.n_clusters > 0 else 0.0

    def ml_infeasability(self, cromosome):
        """Must-link infeasibility.

        Calculate the infeasibility of the current clustering based on must-link
        constraints.

        Args:
            cromosome (numpy.ndarray): The current clustering labels.

        Returns:
            int: The number of must-link constraints that are not satisfied.

        """
        infeasability = 0

        for x in range(self.X.shape[0]):
            ml_constraints = np.argwhere(self.constraints[x] > 0).flatten()

            infeasability += np.sum(cromosome[ml_constraints] != cromosome[x])

        return infeasability // 2  # Each must-link constraint is counted twice

    def cl_infeasability(self, cromosome):
        """Cannot-link infeasibility.

        Calculate the infeasibility of the current clustering based on cannot-link
        constraints.

        Args:
            cromosome (numpy.ndarray): The current clustering labels.

        Returns:
            int: The number of cannot-link constraints that are not satisfied.

        """
        infeasability = 0

        for x in range(self.X.shape[0]):
            cl_constraints = np.argwhere(self.constraints[x] < 0).flatten()

            infeasability += np.sum(cromosome[cl_constraints] == cromosome[x])

        return infeasability // 2

    def get_single_fitness(self, cromosome):
        """Calculate the fitness of a single chromosome.

        Args:
            cromosome (numpy.ndarray): The chromosome to evaluate.

        Returns:
            float: The fitness value of the chromosome.

        """
        distance = self._intra_cluster_distance(cromosome)
        ml_infeasability = self.ml_infeasability(cromosome)
        cl_infeasability = self.cl_infeasability(cromosome)

        penalty = distance * (ml_infeasability + cl_infeasability)
        fitness = distance + penalty

        return fitness

    def mutation(self, chromosome):
        """Perform mutation on a chromosome.

        Args:
            chromosome (numpy.ndarray): The chromosome to mutate.

        Returns:
            numpy.ndarray: The mutated chromosome.

        """
        n = self.X.shape[0]
        segment_start = np.random.randint(n)
        segment_end = (segment_start + self._mutation_size) % n
        new_segment = np.random.randint(0, self.n_clusters, self._mutation_size)

        if segment_start < segment_end:
            chromosome[segment_start:segment_end] = new_segment
        else:
            chromosome[segment_start:] = new_segment[: n - segment_start]
            chromosome[:segment_end] = new_segment[n - segment_start :]
        return chromosome

    def crossover(self, parent1, parent2):
        """Perform crossover between two parents.

        Args:
            parent1 (numpy.ndarray): The first parent chromosome.
            parent2 (numpy.ndarray): The second parent chromosome.

        Returns:
            numpy.ndarray: The new chromosome created by crossover.

        """
        if parent1.shape != parent2.shape:
            raise ValueError("Parent chromosomes must have the same shape.")

        v = np.argwhere(np.random.rand(self.X.shape[0]) > self._probability)
        new_cromosome = np.copy(parent1)
        new_cromosome[v] = parent2[v]
        return new_cromosome

    def local_search(self, chromosome):
        """Perform local search on a chromosome.

        Args:
            chromosome (numpy.ndarray): The chromosome to improve.

        Returns:
            numpy.ndarray: The improved chromosome.

        """
        index_list = np.arange(len(chromosome))
        fitness = self.get_single_fitness(chromosome)
        iterations = 0

        random.shuffle(index_list)

        for index in index_list[: self.local_search_max_iter]:
            original_label = chromosome[index]
            labels = np.arange(self.n_clusters)

            for label in labels:
                if label == original_label:
                    continue

                iterations += 1

                chromosome[index] = label
                new_fitness = self.get_single_fitness(chromosome)

                if new_fitness < fitness:
                    fitness = new_fitness
                    break
                else:
                    chromosome[index] = original_label

            if iterations == self.local_search_max_iter:
                break

        return chromosome

    def _update(self):
        new_chromosome = self.crossover(self.best, self.worst)
        mutant = self.mutation(new_chromosome)
        improved_mutant = self.local_search(mutant)
        improved_mutant_fitness = self.get_single_fitness(improved_mutant)

        if improved_mutant_fitness < np.max(self._fitness):
            if improved_mutant_fitness < np.min(self._fitness):
                self.worst = self.best
                self.best = improved_mutant
            else:
                self.worst = improved_mutant
            self._fitness[np.argmax(self._fitness)] = improved_mutant_fitness

        threshold = np.min(self._fitness) * self._threshold
        if (np.max(self._fitness) - np.min(self._fitness)) > threshold:
            worst = np.argmax(self._fitness)
            self.worst = np.random.randint(0, self.n_clusters, self.X.shape[0])
            self._fitness[worst] = self.get_single_fitness(self.worst)

        self._labels = self.best
        self.calculate_centroids()

    def calculate_centroids(self):
        """Calculate the centroids of the clusters based on the current labels.

        Returns:
            numpy.ndarray: The centroids of the clusters.

        """
        centroids = np.zeros((self.n_clusters, self.X.shape[1]))

        for i in range(self.n_clusters):
            centroids[i] = np.mean(self.X[self._labels == i, :], axis=0)

        self.centroids = centroids

    def _fit(self):
        """Fit the model to the data.

        Returns:
            numpy.ndarray: The best chromosome found.

        """
        self.initialize()
        iteration = 0

        while not self.stop_criteria(iteration):
            self.update()
            iteration += 1

        return self._labels

calculate_centroids()

Calculate the centroids of the clusters based on the current labels.

Returns:

Type Description

numpy.ndarray: The centroids of the clusters.
Source code in clustlib/sin/dils.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
def calculate_centroids(self):
    """Calculate the centroids of the clusters based on the current labels.

    Returns:
        numpy.ndarray: The centroids of the clusters.

    """
    centroids = np.zeros((self.n_clusters, self.X.shape[1]))

    for i in range(self.n_clusters):
        centroids[i] = np.mean(self.X[self._labels == i, :], axis=0)

    self.centroids = centroids

cl_infeasability(cromosome)

Cannot-link infeasibility.

Calculate the infeasibility of the current clustering based on cannot-link constraints.

Parameters:

Name Type Description Default
cromosome ndarray

The current clustering labels.

 required

Returns:

Name Type Description
int

The number of cannot-link constraints that are not satisfied.
Source code in clustlib/sin/dils.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def cl_infeasability(self, cromosome):
    """Cannot-link infeasibility.

    Calculate the infeasibility of the current clustering based on cannot-link
    constraints.

    Args:
        cromosome (numpy.ndarray): The current clustering labels.

    Returns:
        int: The number of cannot-link constraints that are not satisfied.

    """
    infeasability = 0

    for x in range(self.X.shape[0]):
        cl_constraints = np.argwhere(self.constraints[x] < 0).flatten()

        infeasability += np.sum(cromosome[cl_constraints] == cromosome[x])

    return infeasability // 2

crossover(parent1, parent2)

Perform crossover between two parents.

Parameters:

Name Type Description Default
parent1 ndarray

The first parent chromosome.

 required
parent2 ndarray

The second parent chromosome.

 required

Returns:

Type Description

numpy.ndarray: The new chromosome created by crossover.
Source code in clustlib/sin/dils.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def crossover(self, parent1, parent2):
    """Perform crossover between two parents.

    Args:
        parent1 (numpy.ndarray): The first parent chromosome.
        parent2 (numpy.ndarray): The second parent chromosome.

    Returns:
        numpy.ndarray: The new chromosome created by crossover.

    """
    if parent1.shape != parent2.shape:
        raise ValueError("Parent chromosomes must have the same shape.")

    v = np.argwhere(np.random.rand(self.X.shape[0]) > self._probability)
    new_cromosome = np.copy(parent1)
    new_cromosome[v] = parent2[v]
    return new_cromosome

get_single_fitness(cromosome)

Calculate the fitness of a single chromosome.

Parameters:

Name Type Description Default
cromosome ndarray

The chromosome to evaluate.

 required

Returns:

Name Type Description
float

The fitness value of the chromosome.
Source code in clustlib/sin/dils.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def get_single_fitness(self, cromosome):
    """Calculate the fitness of a single chromosome.

    Args:
        cromosome (numpy.ndarray): The chromosome to evaluate.

    Returns:
        float: The fitness value of the chromosome.

    """
    distance = self._intra_cluster_distance(cromosome)
    ml_infeasability = self.ml_infeasability(cromosome)
    cl_infeasability = self.cl_infeasability(cromosome)

    penalty = distance * (ml_infeasability + cl_infeasability)
    fitness = distance + penalty

    return fitness

initialize()

Initialize the chromosomes and their fitness values.

This method initializes two chromosomes with random cluster assignments and calculates their fitness values.

Source code in clustlib/sin/dils.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
def initialize(self):
    """Initialize the chromosomes and their fitness values.

    This method initializes two chromosomes with random cluster assignments and
    calculates their fitness values.
    """
    cromosomes = np.random.randint(0, self.n_clusters, (2, self.X.shape[0]))
    self._fitness = np.empty(2)
    self._fitness[0] = self.get_single_fitness(cromosomes[0, :])
    self._fitness[1] = self.get_single_fitness(cromosomes[1, :])

    self.best = cromosomes[np.argmin(self._fitness)]
    self.worst = cromosomes[np.argmax(self._fitness)]

Perform local search on a chromosome.

Parameters:

Name Type Description Default
chromosome ndarray

The chromosome to improve.

 required

Returns:

Type Description

numpy.ndarray: The improved chromosome.
Source code in clustlib/sin/dils.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def local_search(self, chromosome):
    """Perform local search on a chromosome.

    Args:
        chromosome (numpy.ndarray): The chromosome to improve.

    Returns:
        numpy.ndarray: The improved chromosome.

    """
    index_list = np.arange(len(chromosome))
    fitness = self.get_single_fitness(chromosome)
    iterations = 0

    random.shuffle(index_list)

    for index in index_list[: self.local_search_max_iter]:
        original_label = chromosome[index]
        labels = np.arange(self.n_clusters)

        for label in labels:
            if label == original_label:
                continue

            iterations += 1

            chromosome[index] = label
            new_fitness = self.get_single_fitness(chromosome)

            if new_fitness < fitness:
                fitness = new_fitness
                break
            else:
                chromosome[index] = original_label

        if iterations == self.local_search_max_iter:
            break

    return chromosome

ml_infeasability(cromosome)

Must-link infeasibility.

Calculate the infeasibility of the current clustering based on must-link constraints.

Parameters:

Name Type Description Default
cromosome ndarray

The current clustering labels.

 required

Returns:

Name Type Description
int

The number of must-link constraints that are not satisfied.
Source code in clustlib/sin/dils.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def ml_infeasability(self, cromosome):
    """Must-link infeasibility.

    Calculate the infeasibility of the current clustering based on must-link
    constraints.

    Args:
        cromosome (numpy.ndarray): The current clustering labels.

    Returns:
        int: The number of must-link constraints that are not satisfied.

    """
    infeasability = 0

    for x in range(self.X.shape[0]):
        ml_constraints = np.argwhere(self.constraints[x] > 0).flatten()

        infeasability += np.sum(cromosome[ml_constraints] != cromosome[x])

    return infeasability // 2  # Each must-link constraint is counted twice

mutation(chromosome)

Perform mutation on a chromosome.

Parameters:

Name Type Description Default
chromosome ndarray

The chromosome to mutate.

 required

Returns:

Type Description

numpy.ndarray: The mutated chromosome.
Source code in clustlib/sin/dils.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def mutation(self, chromosome):
    """Perform mutation on a chromosome.

    Args:
        chromosome (numpy.ndarray): The chromosome to mutate.

    Returns:
        numpy.ndarray: The mutated chromosome.

    """
    n = self.X.shape[0]
    segment_start = np.random.randint(n)
    segment_end = (segment_start + self._mutation_size) % n
    new_segment = np.random.randint(0, self.n_clusters, self._mutation_size)

    if segment_start < segment_end:
        chromosome[segment_start:segment_end] = new_segment
    else:
        chromosome[segment_start:] = new_segment[: n - segment_start]
        chromosome[:segment_end] = new_segment[n - segment_start :]
    return chromosome