Typesafe Genetic Algorithm in Python

genetic_algorithm.py

from typing import Tuple, Iterable, Callable, TypeVar
import functools

Solution = TypeVar("Solution")
Population = Iterable[Solution]
EvaluatedPopulation = Iterable[Tuple[float, Solution]]
EvolutionState = Tuple[EvaluatedPopulation, bool]

FitnessFunction = Callable[[Solution], float]
Mutator = Callable[[Solution], Solution]
Selector = Callable[[EvaluatedPopulation], EvaluatedPopulation]
Evaluator = Callable[[Population], EvaluatedPopulation]
Proliferator = Callable[[EvaluatedPopulation], Population]
Iterator = Callable[[EvolutionState, int], EvaluatedPopulation]
Stopper = Callable[[EvaluatedPopulation, int], bool]
Logger = Callable[[EvaluatedPopulation, int], None]


def evaluator_constructor(fitness_function: FitnessFunction) -> Evaluator:
    def evaluate(population: Population) -> EvaluatedPopulation:
        fitnesses = [fitness_function(solution) for solution in population]
        return sorted(zip(fitnesses, population), reverse=True, key=lambda f_: f_[0])
    return evaluate


def proliferate_constructor(reproduction_select: Selector, mutate: Mutator) -> Proliferator:
    def proliferate(evaluated_population: EvaluatedPopulation) -> Population:
        selected = reproduction_select(evaluated_population)
        mutated_population = [mutate(solution[1]) for solution in selected]
        return mutated_population
    return proliferate


def iterator_constructor(evaluate: Evaluator, proliferate: Proliferator,
                         survival_selector: Selector, elite_selector: Selector,
                         stopper: Stopper, logger: Logger) -> Iterator:
    def iterator(evolution_state: EvolutionState, g: int):
        evaluated_population = evolution_state[0]
        ended = evolution_state[1]

        if logger is not None and not ended:
            logger(evaluated_population, g)

        if ended or stopper(evaluated_population, g):
            return (evaluated_population, True)

        proliferated_population = proliferate(evaluated_population)
        evaluated_proliferated_population = list(evaluate(proliferated_population))
        elite = list(elite_selector(evaluated_population))

        next_population = survival_selector(elite +
                                            evaluated_proliferated_population)
        return (next_population, False)
    return iterator


def identity(evaluated_population: EvaluatedPopulation) -> EvaluatedPopulation:
    evaluated_population_list = list(evaluated_population)
    return evaluated_population_list


def create_n_best_selector(n: int) -> Selector:
    def n_best(evaluated_population: EvaluatedPopulation) -> EvaluatedPopulation:
        return sorted(evaluated_population, reverse=True, key=lambda x: x[0])[:n]
    return n_best


def run_mu_plus_lambda(initial_population: Population, mutate: Mutator,
                       fitness: FitnessFunction, stopper: Stopper,
                       population_size: int, generations: int,
                       logger: Logger=None) -> Solution:
    evaluate = evaluator_constructor(fitness)
    proliferate = proliferate_constructor(identity, mutate)
    n_best = create_n_best_selector(population_size)
    iterator = iterator_constructor(evaluate=evaluate,
                                    proliferate=proliferate,
                                    survival_selector=n_best,
                                    elite_selector=identity,
                                    stopper=stopper,
                                    logger=logger)

    initial_population_list = list(initial_population)
    initial_population_evaluated = evaluate(initial_population_list)
    initial_state = (initial_population_evaluated, False)
    final_population = functools.reduce(iterator, range(generations),
                                        initial_state)

    return list(final_population)[0][1]


if __name__ == "__main__":
    import random

    def add_sub(x: float) -> float:
        if random.random() < 0.1:
            if random.random() > 0.5:
                return x + 0.1
            else:
                return x - 0.1
        return x

    def fitness(x: float) -> float:
        return -abs(x * x - 2 * x - 5)

    # Run all generations
    def stopper(population, generation):
        return False

    initial_population = [random.random() * 2.0 - 1 for i in range(10)]
    print(run_mu_plus_lambda(initial_population, add_sub, fitness, stopper, 10,
                             100000))
    # Return best tuple(fitness, solution)
    # > (-0.10598746517817759, -1.471029636645052)