Genetic algorithm to find the minimum of a three-variable function

Question

I've implemented a simple genetic algorithm for continuous floating point parameter spaces and without recombination. However, the way I'm passing in parameters and acquiring results from multiprocessing feels wrong. Should I be using concurrent.futures?

The code below uses the genetic algorithm to find the minimum of the equation x^2+y^2+z/10 over the parameter space -2 < x < 0, 0 < y < 2 and 10 < z < 11, but I'd like to keep the code easy to modify for various parameter spaces and evaluation functions.

import numpy as np

import multiprocessing
from collections import OrderedDict
import os
import time


def eval_iter(arg_lst, l_lst):
    for c_i, args in enumerate(arg_lst):
        yield c_i, args, l_lst


def eval_func(c_i, args, l_lst):
    assert len(args) == 3
    x = args[0]
    y = args[1]
    z = args[2]
    res = x**2 + y**2 + z/10
    print(f"Eval {x}, {y}, {z}: {res}")
    l_lst[c_i] = res


if __name__ == '__main__':

    generation_num = 10
    child_num = 5

    space = OrderedDict((
        ('x', (-2., 0.)),
        ('y', (0., 2.)),
        ('z', (10., 11.))
    ))

    params = OrderedDict([(nm, []) for nm in space.keys()])
    for nm, v_range in space.items():
        params[nm] = np.random.uniform(v_range[0], v_range[1], size=child_num)

    arg_list = []
    for c_n in range(child_num):
        arg_list.append([val[c_n] for val in params.values()])

    manager = multiprocessing.Manager()
    loss_lst = manager.list([np.inf for i in range(child_num)])

    for r_n in range(generation_num):
        with multiprocessing.Pool(os.cpu_count()) as pool:
            pool.starmap(eval_func, eval_iter(arg_list, loss_lst))

        fittest_idx = int(np.argmin(loss_lst))
        base_args = arg_list[fittest_idx]
        print(f"Best {base_args}\n")

        # mutate offspring from fittest individual
        params = OrderedDict([(nm, []) for nm in space.keys()])
        for s_i, (nm, v_range) in enumerate(space.items()):
            std = (v_range[1] - v_range[0]) / 2
            noise = np.random.normal(0, std, size=child_num)
            new_param = base_args[s_i] + noise
            params[nm] = np.clip(new_param, v_range[0], v_range[1])

        arg_list = []
        for c_n in range(child_num):
            arg_list.append([val[c_n] for val in params.values()])

        loss_lst = manager.list([np.inf for i in range(child_num)])

Answer 1

1. Better organisation

   You program in its current state is one huge chunk of snippet with all logic inside it. You should consider splitting it into separate smaller functions. The limits on \$ x, y, z \$ are preset. Consider putting them as a GLOBAL_CONSTANT.

2. Ideal import order (according to PEP8) is

   • standard library imports
   • related third party imports
   • local application/library specific imports

   os/collections etc. are standard library, and should be imported first, followed by numpy.

3. pythonify your code

   x = args[0]
   y = args[1]
   z = args[2]
   

   can be expressed as:

   x, y, z = args
   

   similarly

   manager.list([np.inf for i in range(child_num)])
   

   can become

   manager.list([np.inf] * child_num)
   

I am not sure about how numpy.argmin works, but I think using None instead of np.inf could be better in the sense that you might extend your program to also find local/global maxima along with minima points.