Differential Evolution
Basic DE
import array
import numpy
from deap_er import Fitness, Toolbox, creator, tools
tools.rng.seed(1234) # disables randomization
NDIM = 10
CR = 0.25
F = 1
MU = 300
NGEN = 200
def setup():
creator.create_type("FitnessMin", Fitness, weights=(-1.0,))
creator.create_type("Individual", array.array, typecode="d", fitness=creator.FitnessMin)
toolbox = Toolbox()
toolbox.register("attr_float", tools.rng.uniform, -3, 3)
toolbox.register("individual", tools.init_repeat, creator.Individual, toolbox.attr_float, NDIM)
toolbox.register("population", tools.init_repeat, list, toolbox.individual)
toolbox.register("select", tools.sel_random, sel_count=3)
toolbox.register("mutate", tools.mut_de, scale=F, cx_prob=CR)
toolbox.register("evaluate", tools.bm_sphere)
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", numpy.mean)
stats.register("std", numpy.std)
stats.register("min", numpy.min)
stats.register("max", numpy.max)
logbook = tools.Logbook()
logbook.header = "gen", "evals", "std", "min", "avg", "max"
return toolbox, stats, logbook
def print_results(best_ind):
if best_ind.fitness.values >= (1e-3,):
raise RuntimeError("Evolution failed to converge.")
print("\nEvolution converged correctly.")
def main():
toolbox, stats, logbook = setup()
pop = toolbox.population(size=MU)
hof = tools.HallOfFame(1)
def log_stats(ngen=0):
record = stats.compile(pop)
logbook.record(gen=ngen, evals=len(pop), **record)
print(logbook.stream)
fitness = toolbox.map(toolbox.evaluate, pop)
for ind, fit in zip(pop, fitness, strict=False):
ind.fitness.values = fit
log_stats()
for gen in range(1, NGEN):
for k, agent in enumerate(pop):
a, b, c = toolbox.select(pop)
y = toolbox.clone(agent)
(y,) = toolbox.mutate(y, a, b, c)
y.fitness.values = toolbox.evaluate(y)
if y.fitness > agent.fitness:
pop[k] = y
hof.update(pop)
log_stats(gen)
print_results(hof[0])
if __name__ == "__main__":
main()
Dynamic DE
import array
import itertools
import math
import numpy
from deap_er import Fitness, Toolbox, creator, tools
# Disable randomization to guarantee reproducibility
tools.rng.seed(1234)
# Define constants, objects and functions.
REG_POP_SIZE = 4
RAND_POP_SIZE = 2
TOTAL_POP_SIZE = 6
NDIMS = 5
NPOPS = 10
CR = 0.6
F = 0.4
SCENARIO = tools.MPConfigs.ALT1
BOUNDS = [SCENARIO["min_coord"], SCENARIO["max_coord"]]
MPB = tools.MovingPeaks(dimensions=NDIMS, **SCENARIO)
AVG_OE_MEASURE_INTERVAL = 100
AVG_OE_THRESHOLD = 3
VERBOSE = True
def brown_ind(iter_, best, sigma):
return iter_(tools.rng.gauss(x, sigma) for x in best)
def setup():
creator.create_type("FitnessMax", Fitness, weights=(1.0,))
creator.create_type("Individual", array.array, typecode="d", fitness=creator.FitnessMax)
toolbox = Toolbox()
toolbox.register("attr_float", tools.rng.uniform, BOUNDS[0], BOUNDS[1])
toolbox.register("individual", tools.init_repeat, creator.Individual, toolbox.attr_float, NDIMS)
toolbox.register("brownian_individual", brown_ind, creator.Individual, sigma=0.3)
toolbox.register("population", tools.init_repeat, list, toolbox.individual)
toolbox.register("select", tools.rng.sample, k=4)
toolbox.register("best", tools.sel_best, sel_count=1)
toolbox.register("evaluate", MPB)
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", numpy.mean)
stats.register("std", numpy.std)
stats.register("min", numpy.min)
stats.register("max", numpy.max)
logbook = tools.Logbook()
logbook.header = "gen", "evals", "error", "offline_error", "avg", "max"
return toolbox, stats, logbook
def stop_condition(logbook):
interval = AVG_OE_MEASURE_INTERVAL
if len(logbook) >= 5e5:
raise RuntimeError("Evolution failed to converge.")
elif len(logbook) % interval == 0:
err_sum = 0
for i in range(interval, 0, -1):
val = logbook.select("offline_error")[-i]
err_sum += val
avg_err = err_sum / interval
if avg_err <= AVG_OE_THRESHOLD:
print_results(avg_err)
return 1
return 0
def print_results(avg_err):
print(f"\nAverage offline error: {avg_err:.3f} (<={AVG_OE_THRESHOLD}).")
print("\nEvolution converged correctly.")
class Logger:
def __init__(self, logbook, stats):
self.logbook = logbook
self.stats = stats
def log(self, ngen, pops):
chain = itertools.chain(*pops)
record = self.stats.compile(chain)
args = {
"gen": ngen,
"evals": MPB.nevals,
"error": MPB.current_error,
"offline_error": MPB.offline_error,
}
self.logbook.record(**args, **record)
if VERBOSE:
print(self.logbook.stream)
def get_best_and_invalidate(toolbox, populations) -> list:
bests = [toolbox.best(subpop)[0] for subpop in populations]
if any(b.fitness.values != toolbox.evaluate(b) for b in bests):
for individual in itertools.chain(*populations):
del individual.fitness.values
return bests
def exclude_best_inds(toolbox, bests, populations) -> None:
rex_cl = (BOUNDS[1] - BOUNDS[0]) / (2 * NPOPS ** (1.0 / NDIMS))
for i, j in itertools.combinations(range(NPOPS), 2):
if bests[i].fitness.is_valid() and bests[j].fitness.is_valid():
d = sum((bests[i][k] - bests[j][k]) ** 2 for k in range(NDIMS))
d = math.sqrt(d)
if d < rex_cl:
k = i if bests[i].fitness < bests[j].fitness else j
populations[k] = toolbox.population(size=TOTAL_POP_SIZE)
def regular_diff_evo(toolbox, subpop, xbest, new_pop) -> None:
for individual in subpop[:REG_POP_SIZE]:
x1, x2, x3, x4 = toolbox.select(subpop)
offspring = toolbox.clone(individual)
index = tools.rng.randrange(NDIMS)
for i, _value in enumerate(individual):
if i == index or tools.rng.random() < CR:
offspring[i] = xbest[i] + F * (x1[i] + x2[i] - x3[i] - x4[i])
offspring.fitness.values = toolbox.evaluate(offspring)
if offspring.fitness >= individual.fitness:
new_pop.append(offspring)
else:
new_pop.append(individual)
def brownian_diff_evo(toolbox, xbest, new_pop) -> None:
inds = []
for _ in range(RAND_POP_SIZE):
ind = toolbox.brownian_individual(xbest)
inds.append(ind)
new_pop.extend(inds)
def main():
toolbox, stats, logbook = setup()
logger = Logger(logbook, stats)
# Generate the initial populations.
populations = [toolbox.population(size=TOTAL_POP_SIZE) for _ in range(NPOPS)]
# Evaluate the initial populations.
for subpop in populations:
fitness = toolbox.map(toolbox.evaluate, subpop)
for ind, fit in zip(subpop, fitness, strict=False):
ind.fitness.values = fit
logger.log(0, populations)
generation = 1
# Define the main evolution loop.
while not stop_condition(logbook):
# Detect changes and invalidate fitness if necessary.
bests = get_best_and_invalidate(toolbox, populations)
# Apply exclusionary pressure to the best individuals.
exclude_best_inds(toolbox, bests, populations)
# Evaluate the individuals with an invalid fitness.
tools.evaluate_invalid(toolbox, list(itertools.chain(*populations)))
logger.log(generation, populations)
# Evolve the subpopulations.
for idx, subpop in enumerate(populations):
new_pop = []
(xbest,) = toolbox.best(subpop)
# Apply regular DE to the first part of the population.
regular_diff_evo(toolbox, subpop, xbest, new_pop)
# Apply brownian DE to the last part of the population.
brownian_diff_evo(toolbox, xbest, new_pop)
# Evaluate the brownian individuals.
for individual in new_pop[-RAND_POP_SIZE:]:
individual.fitness.value = toolbox.evaluate(individual)
# Replace the population with the new one.
populations[idx] = new_pop
# Update iteration counter.
generation += 1
if __name__ == "__main__":
main()