Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
stock_GA_deap.py
Created by Huaizheng ZHANG on 7.6.
Copyright (c) 2015 zhzHNN. All rights reserved.
"""
import random
from deap import base
from deap import creator
from deap import tools
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)
toolbox = base.Toolbox()
# Attribute generator
toolbox.register("attr_bool", random.randint, 0, 1)
# Structure initializers
toolbox.register("individual", tools.initRepeat, creator.Individual,
toolbox.attr_bool, 100)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
def evalOneMax(individual):
return sum(individual),
# Operator registering
toolbox.register("evaluate", evalOneMax)
toolbox.register("mate", tools.cxTwoPoint)
toolbox.register("mutate", tools.mutFlipBit, indpb=0.05)
toolbox.register("select", tools.selTournament, tournsize=3)
self.y_ = y
self._X_mean = X.mean()
self._X_std = X.std()
X = (X - self._X_mean)/self._X_std
# There is no test created in this. Private oob is used.
df = pd.concat([X, y], axis = 1)
### Setting toolbox
creator.create("FitnessMax", base.Fitness, weights=(-1.0,))
creator.create("Individual", list , fitness=creator.FitnessMax)
toolbox = base.Toolbox()
## In this case we will also need to pass the base estimator.
toolbox.register("mdl_sample", get_mdl_sample, self.init_sample_percentage, df, self.base_estimator)
# n = 10, defines an ensemble of ten
toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.mdl_sample, self.n)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
toolbox.register("evaluate", eval_each_model_oob_KNN_EG, df = df, base_estimator = self.base_estimator, n_votes = self.n_votes)
toolbox.register("mate", self.crossover_func)
toolbox.register("mutate", segment_mutator_EG, pool_data = df, indpb = self.indpb)
toolbox.register("select", tools.selTournament, tournsize= self.tournsize)
## 3.7 Variations
import random
from deap import base
from deap import creator
from deap import tools
## Data structure and initializer creation
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)
toolbox = base.Toolbox()
toolbox.register("attr_float", random.random)
toolbox.register("individual", tools.initRepeat, creator.Individual, toolbox.attr_float, 10)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
def onemax(individual):
return sum(individual),
toolbox.register("mate", tools.cxTwoPoint)
toolbox.register("mutate", tools.mutGaussian, mu=0, sigma=1, indpb=0.2)
toolbox.register("select", tools.selTournament, tournsize=3)
toolbox.register("evaluate", onemax)
pop = toolbox.population(n=100)
CXPB, MUTPB, NGEN= 0.7, 0.3, 25
fitnesses = toolbox.map(toolbox.evaluate, pop)
## 3.1 A First Individual
import random
from deap import base
from deap import creator
from deap import tools
IND_SIZE = 5
creator.create("FitnessMin", base.Fitness, weights=(-1.0, -1.0))
creator.create("Individual", list, fitness=creator.FitnessMin)
toolbox = base.Toolbox()
toolbox.register("attr_float", random.random)
toolbox.register("individual", tools.initRepeat, creator.Individual,
toolbox.attr_float, n=IND_SIZE)
ind1 = toolbox.individual()
print ind1 # [0.86..., 0.27..., 0.70..., 0.03..., 0.87...]
print ind1.fitness.valid # False
## 3.2 Evaluation
def evaluate(individual):
# Do some hard computing on the individual
a = sum(individual)
b = len(individual)
return a, 1. / b
Returns:
toolbox (deap.base.Toolbox): Configured DEAP Toolbox for the algorithm.
"""
logger.log("Initializing toolbox...")
# Set seed
seed = int(time.time())
random.seed(seed)
logger.log('TOOLBOX.PY random seed is {}'.format(seed))
logger.start_timer()
# Initialize Toolbox
toolbox = base.Toolbox()
logger.stop_timer('TOOLBOX.PY Initializing toolbox')
logger.start_timer()
# Defining tools specific to model
if model_type == "nn":
# ATTRIBUTE
toolbox.register("attribute", random.random)
logger.stop_timer('TOOLBOX.PY register("attribute")')
logger.start_timer()
# INDIVIDUAL
toolbox.register("individual",
getattr(tools, 'initRepeat'),
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))
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, list, toolbox.particle)
toolbox.register("update", updateParticle, chi=0.729843788, c=2.05)
toolbox.register("convert", convert_quantum)
toolbox.register("evaluate", mpb)
def main(verbose=True):
NPARTICLES = 100
RS = (BOUNDS[1] - BOUNDS[0]) / (50**(1.0/NDIM)) # between 1/20 and 1/10 of the domain's range
PMAX = 10
RCLOUD = 1.0 # 0.5 times the move severity
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", numpy.mean)
import random
import numpy
from collections import deque
from multiprocessing import Event, Pipe, Process
from deap import algorithms
from deap import base
from deap import creator
from deap import tools
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", array.array, typecode='b', fitness=creator.FitnessMax)
toolbox = base.Toolbox()
# Attribute generator
toolbox.register("attr_bool", random.randint, 0, 1)
# Structure initializers
toolbox.register("individual", tools.initRepeat, creator.Individual,
toolbox.attr_bool, 100)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
def evalOneMax(individual):
return sum(individual),
def migPipe(deme, k, pipein, pipeout, selection, replacement=None):
"""Migration using pipes between initialized processes. It first selects
*k* individuals from the *deme* and writes them in *pipeout*. Then it
reads the individuals from *pipein* and replace some individuals in the
return sum((f - o)**2 for f, o in zip(feasible_ind, original_ind))
def closest_feasible(individual):
"""A function returning a valid individual from an invalid one."""
feasible_ind = numpy.array(individual)
feasible_ind = numpy.maximum(MIN_BOUND, feasible_ind)
feasible_ind = numpy.minimum(MAX_BOUND, feasible_ind)
return feasible_ind
def valid(individual):
"""Determines if the individual is valid or not."""
if any(individual < MIN_BOUND) or any(individual > MAX_BOUND):
return False
return True
toolbox = base.Toolbox()
toolbox.register("evaluate", benchmarks.zdt2)
toolbox.decorate("evaluate", ClosestValidPenalty(valid, closest_feasible, 1.0e-6, distance))
ind1 = creator.Individual((-5.6468535666e-01,2.2483050478e+00,-1.1087909644e+00,-1.2710112861e-01,1.1682438733e+00,-1.3642007438e+00,-2.1916417835e-01,-5.9137308999e-01,-1.0870160336e+00,6.0515070232e-01,2.1532075914e+00,-2.6164718271e-01,1.5244071578e+00,-1.0324305612e+00,1.2858152343e+00,-1.2584683962e+00,1.2054392372e+00,-1.7429571973e+00,-1.3517256013e-01,-2.6493429355e+00,-1.3051320798e-01,2.2641961090e+00,-2.5027232340e+00,-1.2844874148e+00,1.9955852925e+00,-1.2942218834e+00,3.1340109155e+00,1.6440111097e+00,-1.7750105857e+00,7.7610242710e-01))
print(toolbox.evaluate(ind1))
print("Individuals is valid: %s" % ("True" if valid(ind1) else "False"))
pset = gp.PrimitiveSet("MAIN", 1)
pset.addPrimitive(operator.add, 2)
pset.addPrimitive(operator.sub, 2)
pset.addPrimitive(operator.mul, 2)
pset.addPrimitive(safeDiv, 2)
pset.addPrimitive(operator.neg, 1)
pset.addPrimitive(math.cos, 1)
pset.addPrimitive(math.sin, 1)
pset.addEphemeralConstant("rand101", lambda: random.randint(-1,1))
pset.renameArguments(ARG0='x')
creator.create("FitnessMin", base.Fitness, weights=(-1.0,))
creator.create("Individual", gp.PrimitiveTree, fitness=creator.FitnessMin)
toolbox = base.Toolbox()
toolbox.register("expr", gp.genRamped, pset=pset, min_=1, max_=2)
toolbox.register("individual", tools.initIterate, creator.Individual, toolbox.expr)
toolbox.register("population", tools.initRepeat, list, toolbox.individual)
toolbox.register("lambdify", gp.lambdify, pset=pset)
def evalSymbReg(individual):
# Transform the tree expression in a callable function
func = toolbox.lambdify(expr=individual)
# Evaluate the sum of squared difference between the expression
# and the real function : x**4 + x**3 + x**2 + x
values = (x/10. for x in range(-10,10))
diff_func = lambda x: (func(x)-(x**4 + x**3 + x**2 + x))**2
diff = sum(map(diff_func, values))
return diff,
toolbox.register("evaluate", evalSymbReg)
def geneticAlgorithm(X, y, n_population, n_generation):
"""
Deap global variables
Initialize variables to use eaSimple
"""
# create individual
creator.create("FitnessMax", base.Fitness, weights=(1.0,))
creator.create("Individual", list, fitness=creator.FitnessMax)
# create toolbox
toolbox = base.Toolbox()
toolbox.register("attr_bool", random.randint, 0, 1)
toolbox.register("individual", tools.initRepeat,
creator.Individual, toolbox.attr_bool, len(X.columns))
toolbox.register("population", tools.initRepeat, list,
toolbox.individual)
toolbox.register("evaluate", getFitness, X=X, y=y)
toolbox.register("mate", tools.cxOnePoint)
toolbox.register("mutate", tools.mutFlipBit, indpb=0.05)
toolbox.register("select", tools.selTournament, tournsize=3)
# initialize parameters
pop = toolbox.population(n=n_population)
hof = tools.HallOfFame(n_population * n_generation)
stats = tools.Statistics(lambda ind: ind.fitness.values)
stats.register("avg", np.mean)
stats.register("min", np.min)