CandidateArchive for model-assisted multi-fidelity global search algorithm

For my research I'm working on global search methods where a candidate solution can have it's fitness (=score) evaluated in multiple fidelities (=accuracy levels). The goal of the CandidateArchive is to keep a clear overview of which candidate solutions have been evaluated under which fidelities. A candidate is an ndim-dimensional vector of floating point values, with ndim known in advance.

During its usage, new candidates will be added to the archive together with the fitness score for some fidelity-level. High(er) fidelity fitness scores may be added later. The archive can give back all candidates that have a fitness score for a given list of fidelities.

A trivial example of the stored data at some point might look like this:

candidate  | high-fidelity | low-fidelity
-----------+---------------+---------------
(0.1, 0.2) | NaN           |  1.57
(0.2, 0.7) | 2.5           |  1.25
(1.5, 2.1) | NaN           |  2.78
(0.2, 1.1) | 3.8           |  4.31
(1.5, 0.5) | NaN           |  1.57


I feel like my interface design for this CandidateArchive is not what I want it to be, but don't quite know how I can improve it. Suggestions for tests (pytest/hypothesis) are also more than welcome.

import numpy as np
from warnings import warn
from collections import namedtuple

CandidateSet = namedtuple('CandidateSet', ['candidates', 'fitnesses'])

class CandidateArchive:

def __init__(self, ndim, fidelities=None):
"""An archive of candidate: fitnesses pairs, for one or multiple fidelities"""
self.ndim = ndim

if not fidelities:
fidelities = ['fitness']
self.fidelities = fidelities

self.data = {}
self.max = {fid: -np.inf for fid in self.fidelities}
self.min = {fid: np.inf for fid in self.fidelities}

def __len__(self):
return len(self.data)

def addcandidates(self, candidates, fitnesses, fidelity=None, *, verbose=False):
"""Add multiple candidates to the archive"""
for cand, fit in zip(candidates, fitnesses):

def addcandidate(self, candidate, fitness, fidelity=None, *, verbose=False):
"""Add a candidate to the archive. Will overwrite fitness value if candidate is already present"""

if len(self.fidelities) == 1 and fidelity is not None and verbose:
warn(f"fidelity specification {fidelity} ignored in single-fidelity case", RuntimeWarning)
elif len(self.fidelities) > 1 and fidelity is None:
raise ValueError('must specify fidelity level in multi-fidelity case')

if fidelity is None:
fidelity = self.fidelities

# Checking types to make sure they are iterable in the right way
if isinstance(fitness, (np.float64, float)):
fitness = [fitness]

if isinstance(fidelity, str):
fidelity = [fidelity]

for fid, fit in zip(fidelity, list(fitness)):
if tuple(candidate) not in self.data:
else:
self._updatecandidate(candidate, fit, fid, verbose=verbose)

def _addnewcandidate(self, candidate, fitness, fidelity=None, *, verbose=False):
if len(self.fidelities) == 1:
fit_values = [fitness]
else:
fit_values = np.array([np.nan] * len(self.fidelities))
idx = self.fidelities.index(fidelity)
fit_values[idx] = fitness

self._updateminmax(fidelity, fitness)
self.data[tuple(candidate)] = fit_values

def _updatecandidate(self, candidate, fitness, fidelity=None, *, verbose=False):
fit_values = self.data[tuple(candidate)]

if fidelity is None:
fidelity = 'fitness'

fid_idx = self.fidelities.index(fidelity)

if verbose and not np.isnan(fit_values[fid_idx]):
warn(f"overwriting existing value '{self.data[tuple(candidate), fid_idx]}' with '{fitness}'", RuntimeWarning)

fit_values[fid_idx] = fitness
self._updateminmax(fidelity, fitness)

def getcandidates(self, num_recent_candidates=None, fidelity=None):
"""Retrieve candidates and fitnesses from the archive.

:param num_recent_candidates:   (optional) Only return the last n candidates added to the archive
:param fidelity:                (optional) Only return candidate and fitness information for the specified fidelities
:return:                        Candidates, Fitnesses (tuple of numpy arrays)
"""

if type(fidelity) in [tuple, list]:
pass
elif fidelity:
fidelity = [fidelity]
else:
fidelity = ['fitness']

indices = [self.fidelities.index(fid) for fid in fidelity]

candidates = []
fitnesses = []
for candidate, fits in self.data.items():
for idx in indices:
if np.isnan(fits[idx]):
break
else:
candidates.append(list(candidate))
fitnesses.append([fits[idx] for idx in indices])

candidates = np.array(candidates)
fitnesses = np.array(fitnesses)

if num_recent_candidates is not None:
candidates = candidates[-num_recent_candidates:]
fitnesses = fitnesses[-num_recent_candidates:]

return CandidateSet(candidates, fitnesses)

def _updateminmax(self, fidelity, value):
if value > self.max[fidelity]:
self.max[fidelity] = value
elif value < self.min[fidelity]:
self.min[fidelity] = value


Type hints

PEP484 type hints, such as ndim: int, will help better-define your interface.

Mutability

Reading your code, the other members of CandidateArchive only make sense if fidelities are immutable. As such, don't make them a list - make them a tuple. One advantage is that you can safely give a default argument of ('fitness',), whereas you can't safely give a default argument that is a mutable list.

lower_camel_case

addcandidates should be add_candidates.

Logic inversion

        if tuple(candidate) not in self.data:
else:
self._updatecandidate(candidate, fit, fid, verbose=verbose)


Since you have both code paths here, put the positive one first so that you don't need a not:

        if tuple(candidate) in self.data:
fun = self._update_candidate
else:
fun(candidate, fit, fid, verbose=verbose)


Default arguments

    if fidelity is None:
fidelity = 'fitness'


should be replaced with a default argument of 'fitness'.

No-op code

    if type(fidelity) in [tuple, list]:
pass


This can be made into an outer condition so that you don't need to pass:

if not isinstance(fidelity, (tuple, list)):
if fidelity: ...
else: ...

• Thanks for the tips! – Energya Sep 12 '19 at 13:14