1from deap_er import creator
2from deap_er import tools
3from deap_er import base
4import itertools
5import random
6import array
7import numpy
8import math
9
10
11# Disable randomization to guarantee reproducibility
12random.seed(1234)
13
14# Define constants, objects and functions.
15REG_POP_SIZE = 4
16RAND_POP_SIZE = 2
17TOTAL_POP_SIZE = 6
18
19NDIMS = 5
20NPOPS = 10
21CR = 0.6
22F = 0.4
23
24SCENARIO = tools.MPConfigs.ALT1
25BOUNDS = [SCENARIO["min_coord"], SCENARIO["max_coord"]]
26MPB = tools.MovingPeaks(dimensions=NDIMS, **SCENARIO)
27
28AVG_OE_MEASURE_INTERVAL = 100
29AVG_OE_THRESHOLD = 3
30VERBOSE = True
31
32
33def brown_ind(iter_, best, sigma):
34 return iter_(random.gauss(x, sigma) for x in best)
35
36
37def setup():
38 creator.create("FitnessMax", base.Fitness, weights=(1.0,))
39 creator.create("Individual", array.array, typecode='d', fitness=creator.FitnessMax)
40
41 toolbox = base.Toolbox()
42 toolbox.register("attr_float", random.uniform, BOUNDS[0], BOUNDS[1])
43 toolbox.register("individual", tools.init_repeat, creator.Individual, toolbox.attr_float, NDIMS)
44 toolbox.register("brownian_individual", brown_ind, creator.Individual, sigma=0.3)
45 toolbox.register("population", tools.init_repeat, list, toolbox.individual)
46 toolbox.register("select", random.sample, k=4)
47 toolbox.register("best", tools.sel_best, sel_count=1)
48 toolbox.register("evaluate", MPB)
49
50 stats = tools.Statistics(lambda ind: ind.fitness.values)
51 stats.register("avg", numpy.mean)
52 stats.register("std", numpy.std)
53 stats.register("min", numpy.min)
54 stats.register("max", numpy.max)
55
56 logbook = tools.Logbook()
57 logbook.header = "gen", "evals", "error", "offline_error", "avg", "max"
58
59 return toolbox, stats, logbook
60
61
62def stop_condition(logbook):
63 interval = AVG_OE_MEASURE_INTERVAL
64 if len(logbook) >= 5e+5:
65 raise RuntimeError('Evolution failed to converge.')
66 elif len(logbook) % interval == 0:
67 err_sum = 0
68 for i in range(interval, 0, -1):
69 val = logbook.select("offline_error")[-i]
70 err_sum += val
71 avg_err = err_sum / interval
72 if avg_err <= AVG_OE_THRESHOLD:
73 print_results(avg_err)
74 return 1
75 return 0
76
77
78def print_results(avg_err):
79 print(f'\nAverage offline error: {avg_err:.3f} (<={AVG_OE_THRESHOLD}).')
80 print(f'\nEvolution converged correctly.')
81
82
83class Logger:
84 def __init__(self, logbook, stats):
85 self.logbook = logbook
86 self.stats = stats
87
88 def log(self, ngen, pops):
89 chain = itertools.chain(*pops)
90 record = self.stats.compile(chain)
91 args = dict(
92 gen=ngen,
93 evals=MPB.nevals,
94 error=MPB.current_error,
95 offline_error=MPB.offline_error
96 )
97 self.logbook.record(**args, **record)
98 if VERBOSE:
99 print(self.logbook.stream)
100
101
102def get_best_and_invalidate(toolbox, populations) -> list:
103 bests = [toolbox.best(subpop)[0] for subpop in populations]
104 if any(b.fitness.values != toolbox.evaluate(b) for b in bests):
105 for individual in itertools.chain(*populations):
106 del individual.fitness.values
107 return bests
108
109
110def exclude_best_inds(toolbox, bests, populations) -> None:
111 rex_cl = (BOUNDS[1] - BOUNDS[0]) / (2 * NPOPS ** (1.0 / NDIMS))
112 for i, j in itertools.combinations(range(NPOPS), 2):
113 if bests[i].fitness.is_valid() and bests[j].fitness.is_valid():
114 d = sum((bests[i][k] - bests[j][k]) ** 2 for k in range(NDIMS))
115 d = math.sqrt(d)
116 if d < rex_cl:
117 k = i if bests[i].fitness < bests[j].fitness else j
118 populations[k] = toolbox.population(size=TOTAL_POP_SIZE)
119
120
121def regular_diff_evo(toolbox, subpop, xbest, new_pop) -> None:
122 for individual in subpop[:REG_POP_SIZE]:
123 x1, x2, x3, x4 = toolbox.select(subpop)
124 offspring = toolbox.clone(individual)
125 index = random.randrange(NDIMS)
126 for i, value in enumerate(individual):
127 if i == index or random.random() < CR:
128 offspring[i] = xbest[i] + F * (x1[i] + x2[i] - x3[i] - x4[i])
129 offspring.fitness.values = toolbox.evaluate(offspring)
130 if offspring.fitness >= individual.fitness:
131 new_pop.append(offspring)
132 else:
133 new_pop.append(individual)
134
135
136def brownian_diff_evo(toolbox, xbest, new_pop) -> None:
137 inds = []
138 for _ in range(RAND_POP_SIZE):
139 ind = toolbox.brownian_individual(xbest)
140 inds.append(ind)
141 new_pop.extend(inds)
142
143
144def main():
145 toolbox, stats, logbook = setup()
146 logger = Logger(logbook, stats)
147
148 # Generate the initial populations.
149 populations = [toolbox.population(size=TOTAL_POP_SIZE) for _ in range(NPOPS)]
150
151 # Evaluate the initial populations.
152 for idx, subpop in enumerate(populations):
153 fitness = toolbox.map(toolbox.evaluate, subpop)
154 for ind, fit in zip(subpop, fitness):
155 ind.fitness.values = fit
156
157 logger.log(0, populations)
158
159 generation = 1
160
161 # Define the main evolution loop.
162 while not stop_condition(logbook):
163
164 # Detect changes and invalidate fitness if necessary.
165 bests = get_best_and_invalidate(toolbox, populations)
166
167 # Apply exclusionary pressure to the best individuals.
168 exclude_best_inds(toolbox, bests, populations)
169
170 # Evaluate the individuals with an invalid fitness.
171 chain = itertools.chain(*populations)
172 invalid_ind = [ind for ind in chain if not ind.fitness.is_valid()]
173 fitness = toolbox.map(toolbox.evaluate, invalid_ind)
174 for ind, fit in zip(invalid_ind, fitness):
175 ind.fitness.values = fit
176
177 logger.log(generation, populations)
178
179 # Evolve the subpopulations.
180 for idx, subpop in enumerate(populations):
181 new_pop = []
182 xbest, = toolbox.best(subpop)
183
184 # Apply regular DE to the first part of the population.
185 regular_diff_evo(toolbox, subpop, xbest, new_pop)
186
187 # Apply brownian DE to the last part of the population.
188 brownian_diff_evo(toolbox, xbest, new_pop)
189
190 # Evaluate the brownian individuals.
191 for individual in new_pop[-RAND_POP_SIZE:]:
192 individual.fitness.value = toolbox.evaluate(individual)
193
194 # Replace the population with the new one.
195 populations[idx] = new_pop
196
197 # Update iteration counter.
198 generation += 1
199
200
201if __name__ == "__main__":
202 main()