Passed
Pull Request — master (#233)
by
unknown
01:19
created

EvolutionStrategy1p1.mutate()   A

Complexity

Conditions 1

Size

Total Lines 11
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 2
nop 3
dl 0
loc 11
rs 10
c 0
b 0
f 0
1
# encoding=utf8
2
# pylint: disable=mixed-indentation, trailing-whitespace, multiple-statements, attribute-defined-outside-init, logging-not-lazy, unused-argument, singleton-comparison, no-else-return, line-too-long, arguments-differ, no-self-use, superfluous-parens, redefined-builtin, bad-continuation, unused-variable, assignment-from-no-return
3
import logging
4
from math import ceil
5
6
from numpy import argmin, argsort, log, sum, fmax, sqrt, full, exp, eye, diag, apply_along_axis, round, any, asarray, dot, random as rand, tile, inf, where, append
7
from numpy.linalg import norm, cholesky as chol, eig, solve, lstsq
8
9
from NiaPy.algorithms.algorithm import Algorithm, Individual, defaultIndividualInit
10
from NiaPy.util.utility import objects2array
11
12
logging.basicConfig()
13
logger = logging.getLogger('NiaPy.algorithms.basic')
14
logger.setLevel('INFO')
15
16
__all__ = ['EvolutionStrategy1p1', 'EvolutionStrategyMp1', 'EvolutionStrategyMpL', 'EvolutionStrategyML', 'CovarianceMatrixAdaptionEvolutionStrategy']
17
18
class IndividualES(Individual):
19
	r"""Individual for Evolution Strategies.
20
21
	See Also:
22
		* :class:`NiaPy.algorithms.Individual`
23
	"""
24
	def __init__(self, **kwargs):
25
		r"""Initialize individual.
26
27
		Args:
28
			kwargs (Dict[str, Any]): Additional arguments.
29
30
		See Also:
31
			* :func:`NiaPy.algorithms.Individual.__init__`
32
		"""
33
		Individual.__init__(self, **kwargs)
34
		task, x, rho = kwargs.get('task', None), kwargs.get('x', None), kwargs.get('rho', 1)
35
		self.rho = rho
36
37
class EvolutionStrategy1p1(Algorithm):
38
	r"""Implementation of (1 + 1) evolution strategy algorithm. Uses just one individual.
39
40
	Algorithm:
41
		(1 + 1) Evolution Strategy Algorithm
42
43
	Date:
44
		2018
45
46
	Authors:
47
		Klemen Berkovič
48
49
	License:
50
		MIT
51
52
	Reference URL:
53
54
	Reference paper:
55
56
	Attributes:
57
		Name (List[str]): List of strings representing algorithm names.
58
		mu (int): Number of parents.
59
		k (int): Number of iterations before checking and fixing rho.
60
		c_a (float): Search range amplification factor.
61
		c_r (float): Search range reduction factor.
62
63
	See Also:
64
		* :class:`NiaPy.algorithms.Algorithm`
65
	"""
66
	Name = ['EvolutionStrategy1p1', 'EvolutionStrategy(1+1)', 'ES(1+1)']
67
68 View Code Duplication
	@staticmethod
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
69
	def typeParameters():
70
		r"""Get dictionary with functions for checking values of parameters.
71
72
		Returns:
73
			Dict[str, Callable]:
74
				* mu (Callable[[int], bool])
75
				* k (Callable[[int], bool])
76
				* c_a (Callable[[Union[float, int]], bool])
77
				* c_r (Callable[[Union[float, int]], bool])
78
				* epsilon (Callable[[float], bool])
79
		"""
80
		return {
81
			'mu': lambda x: isinstance(x, int) and x > 0,
82
			'k': lambda x: isinstance(x, int) and x > 0,
83
			'c_a': lambda x: isinstance(x, (float, int)) and x > 1,
84
			'c_r': lambda x: isinstance(x, (float, int)) and 0 < x < 1,
85
			'epsilon': lambda x: isinstance(x, float) and 0 < x < 1
86
		}
87
88
	def setParameters(self, mu=1, k=10, c_a=1.1, c_r=0.5, epsilon=1e-20, **ukwargs):
89
		r"""Set the arguments of an algorithm.
90
91
		Arguments:
92
			mu (Optional[int]): Number of parents
93
			k (Optional[int]): Number of iterations before checking and fixing rho
94
			c_a (Optional[float]): Search range amplification factor
95
			c_r (Optional[float]): Search range reduction factor
96
97
		See Also:
98
			* :func:`NiaPy.algorithms.Algorithm.setParameters`
99
		"""
100
		Algorithm.setParameters(self, NP=mu, itype=ukwargs.pop('itype', IndividualES), **ukwargs)
101
		self.mu, self.k, self.c_a, self.c_r, self.epsilon = mu, k, c_a, c_r, epsilon
102
103
	def mutate(self, x, rho):
104
		r"""Mutate individual.
105
106
		Args:
107
			x (Individual): Current individual.
108
			rho (float): Current standard deviation.
109
110
		Returns:
111
			Individual: Mutated individual.
112
		"""
113
		return x + self.normal(0, rho, len(x))
114
115
	def updateRho(self, rho, k):
116
		r"""Update standard deviation.
117
118
		Args:
119
			rho (float): Current standard deviation.
120
			k (int): Number of succesfull mutations.
121
122
		Returns:
123
			float: New standard deviation.
124
		"""
125
		phi = k / self.k
126
		if phi < 0.2: return self.c_r * rho if rho > self.epsilon else 1
127
		elif phi > 0.2: return self.c_a * rho if rho > self.epsilon else 1
128
		else: return rho
129
130
	def initPopulation(self, task):
131
		r"""Initialize starting individual.
132
133
		Args:
134
			task (Task): Optimization task.
135
136
		Returns:
137
			Tuple[Individual, float, Dict[str, Any]]:
138
				1, Initialized individual.
139
				2, Initialized individual fitness/function value.
140
				3. Additional arguments:
141
					* ki (int): Number of successful rho update.
142
		"""
143
		c, ki = IndividualES(task=task, rnd=self.Rand), 0
144
		return c, c.f, {'ki': ki}
145
146
	def runIteration(self, task, c, fpop, xb, fxb, ki, **dparams):
147
		r"""Core function of EvolutionStrategy(1+1) algorithm.
148
149
		Args:
150
			task (Task): Optimization task.
151
			pop (Individual): Current position.
152
			fpop (float): Current position function/fitness value.
153
			xb (Individual): Global best position.
154
			fxb (float): Global best function/fitness value.
155
			ki (int): Number of successful updates before rho update.
156
			**dparams (Dict[str, Any]): Additional arguments.
157
158
		Returns:
159
			Tuple[Individual, float, Individual, float, Dict[str, Any]]:
160
				1, Initialized individual.
161
				2, Initialized individual fitness/function value.
162
				3. New global best solution.
163
				4. New global best soluitons fitness/objective value.
164
				5. Additional arguments:
165
					* ki (int): Number of successful rho update.
166
		"""
167
		if task.Iters % self.k == 0: c.rho, ki = self.updateRho(c.rho, ki), 0
168
		cn = objects2array([task.repair(self.mutate(c.x, c.rho), self.Rand) for _i in range(self.mu)])
169
		cn_f = asarray([task.eval(cn[i]) for i in range(len(cn))])
170
		ib = argmin(cn_f)
171
		if cn_f[ib] < c.f:
172
			c.x, c.f, ki = cn[ib], cn_f[ib], ki + 1
173
			if cn_f[ib] < fxb: xb, fxb = self.getBest(cn[ib], cn_f[ib], xb, fxb)
174
		return c, c.f, xb, fxb, {'ki': ki}
175
176
class EvolutionStrategyMp1(EvolutionStrategy1p1):
177
	r"""Implementation of (mu + 1) evolution strategy algorithm. Algorithm creates mu mutants but into new generation goes only one individual.
178
179
	Algorithm:
180
		(:math:`\mu + 1`) Evolution Strategy Algorithm
181
182
	Date:
183
		2018
184
185
	Authors:
186
		Klemen Berkovič
187
188
	License:
189
		MIT
190
191
	Reference URL:
192
193
	Reference paper:
194
195
	Attributes:
196
		Name (List[str]): List of strings representing algorithm names.
197
198
	See Also:
199
		* :class:`NiaPy.algorithms.basic.EvolutionStrategy1p1`
200
	"""
201
	Name = ['EvolutionStrategyMp1', 'EvolutionStrategy(mu+1)', 'ES(m+1)']
202
203
	def setParameters(self, **kwargs):
204
		r"""Set core parameters of EvolutionStrategy(mu+1) algorithm.
205
206
		Args:
207
			**kwargs (Dict[str, Any]):
208
209
		See Also:
210
			* :func:`NiaPy.algorithms.basic.EvolutionStrategy1p1.setParameters`
211
		"""
212
		mu = kwargs.pop('mu', 40)
213
		EvolutionStrategy1p1.setParameters(self, mu=mu, **kwargs)
214
215
class EvolutionStrategyMpL(EvolutionStrategy1p1):
216
	r"""Implementation of (mu + lambda) evolution strategy algorithm. Mulation creates lambda individual. Lambda individual compete with mu individuals for survival, so only mu individual go to new generation.
217
218
	Algorithm:
219
		(:math:`\mu + \lambda`) Evolution Strategy Algorithm
220
221
	Date:
222
		2018
223
224
	Authors:
225
		Klemen Berkovič
226
227
	License:
228
		MIT
229
230
	Reference URL:
231
232
	Reference paper:
233
234
	Attributes:
235
		Name (List[str]): List of strings representing algorithm names
236
		lam (int): TODO
237
238
	See Also:
239
		* :class:`NiaPy.algorithms.basic.EvolutionStrategy1p1`
240
	"""
241
	Name = ['EvolutionStrategyMpL', 'EvolutionStrategy(mu+lambda)', 'ES(m+l)']
242
243
	@staticmethod
244
	def typeParameters():
245
		r"""TODO.
246
247
		Returns:
248
			Dict[str, Any]:
249
				* lam (Callable[[int], bool]): TODO.
250
251
		See Also:
252
			* :func:`NiaPy.algorithms.basic.EvolutionStrategy1p1`
253
		"""
254
		d = EvolutionStrategy1p1.typeParameters()
255
		d['lam'] = lambda x: isinstance(x, int) and x > 0
256
		return d
257
258
	def setParameters(self, lam=45, **ukwargs):
259
		r"""Set the arguments of an algorithm.
260
261
		Arguments:
262
			lam (int): Number of new individual generated by mutation.
263
264
		See Also:
265
			* :func:`NiaPy.algorithms.basic.es.EvolutionStrategy1p1.setParameters`
266
		"""
267
		EvolutionStrategy1p1.setParameters(self, InitPopFunc=defaultIndividualInit, **ukwargs)
268
		self.lam = lam
269
270
	def updateRho(self, pop, k):
271
		r"""Update standard deviation for population.
272
273
		Args:
274
			pop (numpy.ndarray[Individual]): Current population.
275
			k (int): Number of successful mutations.
276
		"""
277
		phi = k / self.k
278
		if phi < 0.2:
279
			for i in pop: i.rho = self.c_r * i.rho
280
		elif phi > 0.2:
281
			for i in pop: i.rho = self.c_a * i.rho
282
283
	def changeCount(self, c, cn):
284
		r"""Update number of successful mutations for population.
285
286
		Args:
287
			c (numpy.ndarray[Individual]): Current population.
288
			cn (numpy.ndarray[Individual]): New population.
289
290
		Returns:
291
			int: Number of successful mutations.
292
		"""
293
		k = 0
294
		for e in cn:
295
			if e not in c: k += 1
296
		return k
297
298
	def mutateRand(self, pop, task):
299
		r"""Mutate random individual form population.
300
301
		Args:
302
			pop (numpy.ndarray[Individual]): Current population.
303
			task (Task): Optimization task.
304
305
		Returns:
306
			numpy.ndarray: Random individual from population that was mutated.
307
		"""
308
		i = self.randint(self.mu)
309
		return task.repair(self.mutate(pop[i].x, pop[i].rho), rnd=self.Rand)
310
311
	def initPopulation(self, task):
312
		r"""Initialize starting population.
313
314
		Args:
315
			task (Task): Optimization task.
316
317
		Returns:
318
			Tuple[numpy.ndarray[Individual], numpy.ndarray[float], Dict[str, Any]]:
319
				1. Initialized populaiton.
320
				2. Initialized populations function/fitness values.
321
				3. Additional arguments:
322
					* ki (int): Number of successful mutations.
323
324
		See Also:
325
			* :func:`NiaPy.algorithms.algorithm.Algorithm.initPopulation`
326
		"""
327
		c, fc, d = Algorithm.initPopulation(self, task)
328
		d.update({'ki': 0})
329
		return c, fc, d
330
331
	def runIteration(self, task, c, fpop, xb, fxb, ki, **dparams):
332
		r"""Core function of EvolutionStrategyMpL algorithm.
333
334
		Args:
335
			task (Task): Optimization task.
336
			c (numpy.ndarray): Current population.
337
			fpop (numpy.ndarray): Current populations fitness/function values.
338
			xb (numpy.ndarray): Global best individual.
339
			fxb (float): Global best individuals fitness/function value.
340
			ki (int): Number of successful mutations.
341
			**dparams (Dict[str, Any]): Additional arguments.
342
343
		Returns:
344
			Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, float, Dict[str, Any]]:
345
				1. New population.
346
				2. New populations function/fitness values.
347
				3. New global best solution.
348
				4. New global best solutions fitness/objective value.
349
				5. Additional arguments:
350
					* ki (int): Number of successful mutations.
351
		"""
352
		if task.Iters % self.k == 0: _, ki = self.updateRho(c, ki), 0
353
		cn = objects2array([IndividualES(x=self.mutateRand(c, task), task=task, rnd=self.Rand) for _ in range(self.lam)])
354
		cn = append(cn, c)
355
		cn = objects2array([cn[i] for i in argsort([i.f for i in cn])[:self.mu]])
356
		ki += self.changeCount(c, cn)
357
		fcn = asarray([x.f for x in cn])
358
		xb, fxb = self.getBest(cn, fcn, xb, fxb)
359
		return cn, fcn, xb, fxb, {'ki': ki}
360
361
class EvolutionStrategyML(EvolutionStrategyMpL):
362
	r"""Implementation of (mu, lambda) evolution strategy algorithm. Algorithm is good for dynamic environments. Mu individual create lambda chields. Only best mu chields go to new generation. Mu parents are discarded.
363
364
	Algorithm:
365
		(:math:`\mu + \lambda`) Evolution Strategy Algorithm
366
367
	Date:
368
		2018
369
370
	Authors:
371
		Klemen Berkovič
372
373
	License:
374
		MIT
375
376
	Reference URL:
377
378
	Reference paper:
379
380
	Attributes:
381
		Name (List[str]): List of strings representing algorithm names
382
383
	See Also:
384
		* :class:`NiaPy.algorithm.basic.es.EvolutionStrategyMpL`
385
	"""
386
	Name = ['EvolutionStrategyML', 'EvolutionStrategy(mu,lambda)', 'ES(m,l)']
387
388
	def newPop(self, pop):
389
		r"""Return new population.
390
391
		Args:
392
			pop (numpy.ndarray): Current population.
393
394
		Returns:
395
			numpy.ndarray: New population.
396
		"""
397
		pop_s = argsort([i.f for i in pop])
398
		if self.mu < self.lam: return objects2array([pop[i] for i in pop_s[:self.mu]])
399
		npop = list()
400
		for i in range(int(ceil(float(self.mu) / self.lam))): npop.extend(pop[:self.lam if (self.mu - i * self.lam) >= self.lam else self.mu - i * self.lam])
401
		return objects2array(npop)
402
403
	def initPopulation(self, task):
404
		r"""Initialize starting population.
405
406
		Args:
407
			task (Task): Optimization task.
408
409
		Returns:
410
			Tuple[numpy.ndarray[Individual], numpy.ndarray[float], Dict[str, Any]]:
411
				1. Initialized population.
412
				2. Initialized populations fitness/function values.
413
				2. Additional arguments.
414
415
		See Also:
416
			* :func:`NiaPy.algorithm.basic.es.EvolutionStrategyMpL.initPopulation`
417
		"""
418
		c, fc, _ = EvolutionStrategyMpL.initPopulation(self, task)
419
		return c, fc, {}
420
421
	def runIteration(self, task, c, fpop, xb, fxb, **dparams):
422
		r"""Core function of EvolutionStrategyML algorithm.
423
424
		Args:
425
			task (Task): Optimization task.
426
			c (numpy.ndarray): Current population.
427
			fpop (numpy.ndarray): Current population fitness/function values.
428
			xb (numpy.ndarray): Global best individual.
429
			fxb (float): Global best individuals fitness/function value.
430
			**dparams Dict[str, Any]: Additional arguments.
431
432
		Returns:
433
			Tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, float, Dict[str, Any]]:
434
				1. New population.
435
				2. New populations fitness/function values.
436
				3. New global best solution.
437
				4. New global best solutions fitness/objective value.
438
				5. Additional arguments.
439
		"""
440
		cn = objects2array([IndividualES(x=self.mutateRand(c, task), task=task, rand=self.Rand) for _ in range(self.lam)])
441
		c = self.newPop(cn)
442
		fc = asarray([x.f for x in c])
443
		xb, fxb = self.getBest(c, fc, xb, fxb)
444
		return c, fc, xb, fxb, {}
445
446
def CovarianceMaatrixAdaptionEvolutionStrategyF(task, epsilon=1e-20, rnd=rand):
447
	lam, alpha_mu, hs, sigma0 = (4 + round(3 * log(task.D))) * 10, 2, 0, 0.3 * task.bcRange()
448
	mu = int(round(lam / 2))
449
	w = log(mu + 0.5) - log(range(1, mu + 1))
450
	w = w / sum(w)
451
	mueff = 1 / sum(w ** 2)
452
	cs = (mueff + 2) / (task.D + mueff + 5)
453
	ds = 1 + cs + 2 * max(sqrt((mueff - 1) / (task.D + 1)) - 1, 0)
454
	ENN = sqrt(task.D) * (1 - 1 / (4 * task.D) + 1 / (21 * task.D ** 2))
455
	cc, c1 = (4 + mueff / task.D) / (4 + task.D + 2 * mueff / task.D), 2 / ((task.D + 1.3) ** 2 + mueff)
456
	cmu, hth = min(1 - c1, alpha_mu * (mueff - 2 + 1 / mueff) / ((task.D + 2) ** 2 + alpha_mu * mueff / 2)), (1.4 + 2 / (task.D + 1)) * ENN
457
	ps, pc, C, sigma, M = full(task.D, 0.0), full(task.D, 0.0), eye(task.D), sigma0, full(task.D, 0.0)
458
	x = rnd.uniform(task.bcLower(), task.bcUpper())
459
	x_f = task.eval(x)
460
	while not task.stopCondI():
461
		pop_step = asarray([rnd.multivariate_normal(full(task.D, 0.0), C) for _ in range(int(lam))])
462
		pop = asarray([task.repair(x + sigma * ps, rnd) for ps in pop_step])
463
		pop_f = apply_along_axis(task.eval, 1, pop)
464
		isort = argsort(pop_f)
465
		pop, pop_f, pop_step = pop[isort[:mu]], pop_f[isort[:mu]], pop_step[isort[:mu]]
466
		if pop_f[0] < x_f: x, x_f = pop[0], pop_f[0]
467
		M = sum(w * pop_step.T, axis=1)
468
		ps = solve(chol(C).conj() + epsilon, ((1 - cs) * ps + sqrt(cs * (2 - cs) * mueff) * M + epsilon).T)[0].T
469
		sigma *= exp(cs / ds * (norm(ps) / ENN - 1)) ** 0.3
470
		ifix = where(sigma == inf)
471
		if any(ifix): sigma[ifix] = sigma0
472
		if norm(ps) / sqrt(1 - (1 - cs) ** (2 * (task.Iters + 1))) < hth: hs = 1
473
		else: hs = 0
474
		delta = (1 - hs) * cc * (2 - cc)
475
		pc = (1 - cc) * pc + hs * sqrt(cc * (2 - cc) * mueff) * M
476
		C = (1 - c1 - cmu) * C + c1 * (tile(pc, [len(pc), 1]) * tile(pc.reshape([len(pc), 1]), [1, len(pc)]) + delta * C)
477
		for i in range(mu): C += cmu * w[i] * tile(pop_step[i], [len(pop_step[i]), 1]) * tile(pop_step[i].reshape([len(pop_step[i]), 1]), [1, len(pop_step[i])])
478
		E, V = eig(C)
479
		if any(E < epsilon):
480
			E = fmax(E, 0)
481
			C = lstsq(V.T, dot(V, diag(E)).T, rcond=None)[0].T
482
	return x, x_f
483
484
class CovarianceMatrixAdaptionEvolutionStrategy(Algorithm):
485
	r"""Implementation of (mu, lambda) evolution strategy algorithm. Algorithm is good for dynamic environments. Mu individual create lambda chields. Only best mu chields go to new generation. Mu parents are discarded.
486
487
	Algorithm:
488
		(:math:`\mu + \lambda`) Evolution Strategy Algorithm
489
490
	Date:
491
		2018
492
493
	Authors:
494
		Klemen Berkovič
495
496
	License:
497
		MIT
498
499
	Reference URL:
500
		https://arxiv.org/abs/1604.00772
501
502
	Reference paper:
503
		Hansen, Nikolaus. "The CMA evolution strategy: A tutorial." arXiv preprint arXiv:1604.00772 (2016).
504
505
	Attributes:
506
		Name (List[str]): List of names representing algorithm names
507
		epsilon (float): TODO
508
	"""
509
	Name = ['CovarianceMatrixAdaptionEvolutionStrategy', 'CMA-ES', 'CMAES']
510
	epsilon = 1e-20
511
512
	@staticmethod
513
	def typeParameters(): return {
514
			'epsilon': lambda x: isinstance(x, (float, int)) and 0 < x < 1
515
	}
516
517
	def setParameters(self, epsilon=1e-20, **ukwargs):
518
		r"""Set core parameters of CovarianceMatrixAdaptionEvolutionStrategy algorithm.
519
520
		Args:
521
			epsilon (float): Small number.
522
			**ukwargs (Dict[str, Any]): Additional arguments.
523
		"""
524
		Algorithm.setParameters(self, **ukwargs)
525
		self.epsilon = epsilon
526
527
	def runTask(self, task):
528
		r"""TODO.
529
530
		Args:
531
			task (Task): Optimization task.
532
533
		Returns:
534
			TODO.
535
		"""
536
		return CovarianceMaatrixAdaptionEvolutionStrategyF(task, self.epsilon, rnd=self.Rand)
537
538
# vim: tabstop=3 noexpandtab shiftwidth=3 softtabstop=3
539