Moving Peaks Benchmark with Multiswarm PSO

In this example we show how to use the MovingPeaks benchmark. A popular algorithm on this benchmark is the Multiswarm PSO (MPSO) [Blackwell2004] which achieve a great offline error and is able to follow multiple peaks at the same time.

Choosing the Scenario

The moving peak benchmark allows to choose from the 3 original scenarios proposed in the original studies. This is done by retreiving one of the constants defined in the movingpeaks module. Here we will use Scenario 2.

from deap.benchmarks import movingpeaks
scenario = movingpeaks.SCENARIO_2

Once the scenario is retrieved, we need to set a few more constants and instantiate the benchmark, here the number of dimensions and the bounds of the problem.

For a list of all the variables defined in the SENARIO_X dictionaries see MovingPeaks class documentation.

Initialization

As in every DEAP example we are required to create the objects. The moving peak benchmark is a max problem, thus we need a maximizing fitness. And, we associate that fitness to a particle as in the Particle Swarm Optimization Basics example.

creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Particle", list, fitness=creator.FitnessMax, speed=list, 
    best=None, bestfit=creator.FitnessMax)
creator.create("Swarm", list, best=None, bestfit=creator.FitnessMax)

Then, the particle generator is defined. It takes the particle class object pclass into which to put the data. Remeber that creator.Particle, which is gonna be give to this argument in the toolbox, inherits from list and can be initialized with an iterable. The position (elements of the list) and the speed (attribute) of the particle is set to randomly generated numbers between the given bounds.

def generate(pclass, dim, pmin, pmax, smin, smax):
    part = pclass(random.uniform(pmin, pmax) for _ in range(dim)) 
    part.speed = [random.uniform(smin, smax) for _ in range(dim)]
    return part

The next function update the particle position and speed.

def updateParticle(part, best, chi, c):
    ce1 = (c * random.uniform(0, 1) for _ in range(len(part)))
    ce2 = (c * random.uniform(0, 1) for _ in range(len(part)))
    ce1_p = map(operator.mul, ce1, map(operator.sub, best, part))
    ce2_g = map(operator.mul, ce2, map(operator.sub, part.best, part))
    a = map(operator.sub,
                      map(operator.mul,
                                    itertools.repeat(chi),
                                    map(operator.add, ce1_p, ce2_g)),
                      map(operator.mul,
                                     itertools.repeat(1 - chi),
                                     part.speed))
    part.speed = list(map(operator.add, part.speed, a))
    part[:] = list(map(operator.add, part, part.speed))

Thereafter, a function “converting” a particle to a quantum particle with different possible distributions is defined.

def convertQuantum(swarm, rcloud, centre, dist):
    dim = len(swarm[0])
    for part in swarm:
        position = [random.gauss(0, 1) for _ in range(dim)]
        dist = math.sqrt(sum(x**2 for x in position))
        
        if dist == "gaussian":
            u = abs(random.gauss(0, 1.0/3.0))
            part[:] = [(rcloud * x * u**(1.0/dim) / dist) + c for x, c in zip(position, centre)]
        
        elif dist == "uvd":
            u = random.random()
            part[:] = [(rcloud * x * u**(1.0/dim) / dist) + c for x, c in zip(position, centre)]
        
        elif dist == "nuvd":
            u = abs(random.gauss(0, 1.0/3.0))
            part[:] = [(rcloud * x * u / dist) + c for x, c in zip(position, centre)]
        
        del part.fitness.values
        del part.bestfit.values
        part.best = None
    
    return swarm

Finally, all the functions are registered in the toolbox for further use in the algorithm.

toolbox = base.Toolbox()
toolbox.register("particle", generate, creator.Particle, dim=NDIM,
    pmin=BOUNDS[0], pmax=BOUNDS[1], smin=-(BOUNDS[1] - BOUNDS[0])/2.0,
    smax=(BOUNDS[1] - BOUNDS[0])/2.0)
toolbox.register("swarm", tools.initRepeat, creator.Swarm, toolbox.particle)
toolbox.register("update", updateParticle, chi=0.729843788, c=2.05)
toolbox.register("convert", convertQuantum, dist="nuvd")
toolbox.register("evaluate", mpb)

Moving Peaks

The registered evaluation function in the toolbox refers directly to the instance of the MovingPeaks benchmark object mpb. The call to mpb() evaluates the given individuals as any other evaluation function.

Algorithm

The algorithm is fully detailed in the file examples/pso/multiswarm, it reflects what is described in [Blackwell2004].

[Blackwell2004]

(1, 2) Blackwell, T., & Branke, J. (2004). Multi-swarm optimization in dynamic environments. In Applications of Evolutionary Computing (pp. 489-500). Springer Berlin Heidelberg.