3
\$\begingroup\$

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)])
\$\endgroup\$
3
\$\begingroup\$
  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.

| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.