#
# Sub-module containing several optimisation routines
#
# This file is part of PINTS (https://github.com/pints-team/pints/) which is
# released under the BSD 3-clause license. See accompanying LICENSE.md for
# copyright notice and full license details.
#
import pints
import numpy as np
[docs]
class Optimiser(pints.Loggable, pints.TunableMethod):
"""
Base class for optimisers implementing an ask-and-tell interface.
This interface provides fine-grained control. Users seeking to simply run
an optimisation may wish to use the :class:`OptimisationController`
instead.
Optimisation using "ask-and-tell" proceed by the user repeatedly "asking"
the optimiser for points, and then "telling" it the function evaluations at
those points. This allows a user to have fine-grained control over an
optimisation, and implement custom parallelisation, logging, stopping
criteria etc. Users who don't need this functionality can use optimisers
via the :class:`OptimisationController` class instead.
All PINTS optimisers are _minimisers_. To maximise a function simply pass
in the negative of its evaluations to :meth:`tell()` (this is handled
automatically by the :class:`OptimisationController`).
All optimisers implement the :class:`pints.Loggable` and
:class:`pints.TunableMethod` interfaces.
Parameters
----------
x0
A starting point for searches in the parameter space. This must be a
1-dimensional vector, and its length will determine the dimensionality
of the search space. The initial position ``x0`` may may be used
directly (for example as the initial position of a particle in
:class:`PSO`) or indirectly (for example as the center of a
distribution in :class:`XNES`).
sigma0
An optional initial standard deviation around ``x0``. Can be specified
either as a scalar value (one standard deviation for all coordinates)
or as an array with one entry per dimension. Not all methods will use
this information, and some methods will only use part of the provided
information (e.g. ``CMA-ES`` will only use the smallest value in
``sigma0``). If no value for ``sigma0`` is provided, a guess will be
made. If a :meth:`range<Boundaries.range>` can be obtained from
provided boundaries, then 1/6th of this range will be used. If not, the
value will be set as ``abs(x0[i]) / 3`` for any ``i`` where
``x0[i] != 0`` or as ``1`` where ``x0[i] == 0``.
boundaries
An optional set of boundaries on the parameter space.
Example
-------
An optimisation with ask-and-tell, proceeds roughly as follows::
optimiser = MyOptimiser()
running = True
while running:
# Ask for points to evaluate
xs = optimiser.ask()
# Evaluate the score function or pdf at these points
# At this point, code to parallelise evaluation can be added in
fs = [f(x) for x in xs]
# Tell the optimiser the evaluations; allowing it to update its
# internal state.
optimiser.tell(fs)
# Check stopping criteria
# At this point, custom stopping criteria can be added in
if optimiser.f_best() < threshold:
running = False
# Check for optimiser issues
if optimiser.stop():
running = False
# At this point, code to visualise or benchmark optimiser behaviour
# could be added in, for example by plotting `xs` in the parameter
# space.
"""
def __init__(self, x0, sigma0=None, boundaries=None):
# Convert and store initial position
self._x0 = pints.vector(x0)
# Get dimension
self._n_parameters = len(self._x0)
if self._n_parameters < 1:
raise ValueError('Problem dimension must be greater than zero.')
# Store boundaries
self._boundaries = boundaries
if self._boundaries:
if self._boundaries.n_parameters() != self._n_parameters:
raise ValueError(
'Boundaries must have same dimension as starting point.')
# Check initial position is within boundaries
if self._boundaries:
if not self._boundaries.check(self._x0):
raise ValueError(
'Initial position must lie within given boundaries.')
# Check initial standard deviation
if sigma0 is None:
# Set a standard deviation
# Try and use boundaries to guess
try:
self._sigma0 = (1 / 6) * self._boundaries.range()
except AttributeError:
# No boundaries set, or boundaries don't support range()
# Use initial position to guess at parameter scaling
self._sigma0 = (1 / 3) * np.abs(self._x0)
# But add 1 for any initial value that's zero
self._sigma0 += (self._sigma0 == 0)
self._sigma0.setflags(write=False)
elif np.isscalar(sigma0):
# Single number given, convert to vector
sigma0 = float(sigma0)
if sigma0 <= 0:
raise ValueError(
'Initial standard deviation must be greater than zero.')
self._sigma0 = np.ones(self._n_parameters) * sigma0
self._sigma0.setflags(write=False)
else:
# Vector given
self._sigma0 = pints.vector(sigma0)
if len(self._sigma0) != self._n_parameters:
raise ValueError(
'Initial standard deviation must be None, scalar, or have'
' dimension ' + str(self._n_parameters) + '.')
if np.any(self._sigma0 <= 0):
raise ValueError(
'Initial standard deviations must be greater than zero.')
[docs]
def ask(self):
"""
Returns a list of positions in the search space to evaluate.
"""
raise NotImplementedError
[docs]
def fbest(self):
"""Deprecated alias of :meth:`f_best()`."""
# Deprecated on 2021-11-29
import warnings
warnings.warn(
'The method `pints.Optimiser.fbest` is deprecated.'
' Please use `pints.Optimiser.f_best` instead.')
return self.f_best()
[docs]
def f_best(self):
"""
Returns the best objective function evaluation seen by this optimiser,
such that ``f_best = f(x_best)``.
"""
raise NotImplementedError
[docs]
def f_guessed(self):
"""
For optimisers in which the best guess of the optimum (see
:meth:`x_guessed`) differs from the best-seen point (see
:meth:`x_best`), this method returns an estimate of the objective
function value at ``x_guessed``.
Notes:
1. For many optimisers the best guess is simply the best point seen
during the optimisation, so that this method is equivalent to
:meth:`f_best()`.
2. Because ``x_guessed`` is not required to be a point that the
optimiser has visited, the value ``f(x_guessed)`` may be unkown. In
these cases, an approximation of ``f(x_guessed)`` may be returned.
"""
return self.f_best()
[docs]
@classmethod
def name(self):
""" Returns this method's full name. """
raise NotImplementedError
[docs]
def needs_sensitivities(self):
"""
Returns ``True`` if this methods needs sensitivities to be passed in to
``tell`` along with the evaluated error.
"""
return False
[docs]
def running(self):
"""
Returns ``True`` if this an optimisation is in progress.
"""
raise NotImplementedError
[docs]
def stop(self):
"""
Checks if this method has run into trouble and should terminate.
Returns ``False`` if everything's fine, or a short message (e.g.
"Ill-conditioned matrix.") if the method should terminate.
"""
return False
[docs]
def tell(self, fx):
"""
Performs an iteration of the optimiser algorithm, using the evaluations
``fx`` of the points ``x`` previously specified by ``ask``.
For methods that require sensitivities (see
:meth:`needs_sensitivities`), ``fx`` should be a tuple
``(objective, sensitivities)``, containing the values returned by
:meth:`pints.ErrorMeasure.evaluateS1()`.
"""
raise NotImplementedError
[docs]
def xbest(self):
"""Deprecated alias of :meth:`x_best()`."""
# Deprecated on 2021-11-29
import warnings
warnings.warn(
'The method `pints.Optimiser.xbest` is deprecated.'
' Please use `pints.Optimiser.x_best` instead.')
return self.x_best()
[docs]
def x_best(self):
"""
Returns the best position seen during an optimisation, i.e. the point
for which the minimal error or maximum probability density was
observed.
"""
raise NotImplementedError
[docs]
def x_guessed(self):
"""
Returns the optimiser's current best estimate of where the optimum is.
For many optimisers, this will simply be the point for which the
minimal error or maximum probability density was observed, so that
``x_guessed = x_best``. However, optimisers like :class:`pints.CMAES`
and its derivatives, maintain a separate "best guess" value that does
not necessarily correspond to any of the points evaluated during the
optimisation.
"""
return self.x_best()
[docs]
class PopulationBasedOptimiser(Optimiser):
"""
Base class for optimisers that work by moving multiple points through the
search space.
Extends :class:`Optimiser`.
"""
def __init__(self, x0, sigma0=None, boundaries=None):
super().__init__(x0, sigma0, boundaries)
# Set initial population size using heuristic
self._population_size = self._suggested_population_size()
[docs]
def population_size(self):
"""
Returns this optimiser's population size.
If no explicit population size has been set, ``None`` may be returned.
Once running, the correct value will always be returned.
"""
return self._population_size
[docs]
def set_population_size(self, population_size=None):
"""
Sets a population size to use in this optimisation.
If `population_size` is set to ``None``, the population size will be
set using the heuristic :meth:`suggested_population_size()`.
"""
if self.running():
raise Exception('Cannot change population size during run.')
# Check population size or set using heuristic
if population_size is not None:
population_size = int(population_size)
if population_size < 1:
raise ValueError('Population size must be at least 1.')
# Store
self._population_size = population_size
[docs]
def suggested_population_size(self, round_up_to_multiple_of=None):
"""
Returns a suggested population size for this method, based on the
dimension of the search space (e.g. the parameter space).
If the optional argument ``round_up_to_multiple_of`` is set to an
integer greater than 1, the method will round up the estimate to a
multiple of that number. This can be useful to obtain a population size
based on e.g. the number of worker processes used to perform objective
function evaluations.
"""
population_size = self._suggested_population_size()
if round_up_to_multiple_of is not None:
n = int(round_up_to_multiple_of)
if n > 1:
population_size = n * (((population_size - 1) // n) + 1)
return population_size
def _suggested_population_size(self):
"""
Returns a suggested population size for use by
:meth:`suggested_population_size`.
"""
raise NotImplementedError
[docs]
def n_hyper_parameters(self):
""" See :meth:`TunableMethod.n_hyper_parameters()`. """
return 1
[docs]
def set_hyper_parameters(self, x):
"""
The hyper-parameter vector is ``[population_size]``.
See :meth:`TunableMethod.set_hyper_parameters()`.
"""
self.set_population_size(x[0])
[docs]
class OptimisationController():
"""
Finds the parameter values that minimise an :class:`ErrorMeasure` or
maximise a :class:`LogPDF`.
Parameters
----------
function
An :class:`pints.ErrorMeasure` or a :class:`pints.LogPDF` that
evaluates points in the parameter space.
x0
The starting point for searches in the parameter space. For details,
see :class:`Optimiser`.
sigma0
An optional initial standard deviation around ``x0``. For details, see
:class:`Optimiser`.
boundaries
An optional set of boundaries on the parameter space. For details, see
:class:`Optimiser`.
transformation
An optional :class:`pints.Transformation` to allow the optimiser to
search in a transformed parameter space. If used, points shown or
returned to the user will first be detransformed back to the original
space.
method
The class of :class:`pints.Optimiser` to use for the optimisation.
If no method is specified, :class:`CMAES` is used.
"""
def __init__(
self, function, x0, sigma0=None, boundaries=None,
transformation=None, method=None):
# Convert x0 to vector
# This converts e.g. (1, 7) shapes to (7, ), giving users a bit more
# freedom with the exact shape passed in. For example, to allow the
# output of LogPrior.sample(1) to be passed in.
x0 = pints.vector(x0)
# Check dimension of x0 against function
if function.n_parameters() != len(x0):
raise ValueError(
'Starting point must have same dimension as function to'
' optimise.')
# Check if minimising or maximising
self._minimising = not isinstance(function, pints.LogPDF)
# Apply a transformation (if given). From this point onward the
# optimiser will see only the transformed search space and will know
# nothing about the model parameter space.
if transformation is not None:
# Convert error measure or log pdf
if self._minimising:
function = transformation.convert_error_measure(function)
else:
function = transformation.convert_log_pdf(function)
# Convert initial position
x0 = transformation.to_search(x0)
# Convert sigma0, if provided
if sigma0 is not None:
sigma0 = transformation.convert_standard_deviation(sigma0, x0)
if boundaries:
boundaries = transformation.convert_boundaries(boundaries)
# Store transformation for later detransformation: if using a
# transformation, any parameters logged to the filesystem or printed to
# screen should be detransformed first!
self._transformation = transformation
# Store function
if self._minimising:
self._function = function
else:
self._function = pints.ProbabilityBasedError(function)
del function
# Create optimiser
if method is None:
method = pints.CMAES
elif not issubclass(method, pints.Optimiser):
raise ValueError('Method must be subclass of pints.Optimiser.')
self._optimiser = method(x0, sigma0, boundaries)
# Check if sensitivities are required
self._needs_sensitivities = self._optimiser.needs_sensitivities()
# Track optimiser's f_best or f_guessed
self._use_f_guessed = None
self.set_f_guessed_tracking()
# Logging
self._log_to_screen = True
self._log_filename = None
self._log_csv = False
self.set_log_interval()
# Parallelisation
self._parallel = False
self._n_workers = 1
self.set_parallel()
# User callback
self._callback = None
# :meth:`run` can only be called once
self._has_run = False
# Post-run statistics
self._evaluations = None
self._iterations = None
self._time = None
#
# Stopping criteria
# Note that we always minimise: PDFs are wrapped in an Error class that
# multiplies by -1
#
# Maximum iterations
self._max_iterations = None
# Maximum number of iterations without significant change in f(x)
self._ftol_max = None # max number of iterations without change
self._ftol_threshold = None # smallest significant change
# Maximum number of iterations without significant change in x
self._xtol_max = None # max number of iterations without change
self._xtol_threshold = None # smallest significant change per param
# Maximum evaluations
self._max_evaluations = None
# Function threshold: stop if f(x) < threshold
self._function_threshold = None
# Default stopping critera
self.set_max_iterations()
self.set_function_tolerance()
def _check_stopping_criteria(self, iterations, unchanged_f_iterations,
unchanged_x_iterations, evaluations, f_new):
"""
Checks the stopping criteria, returns either ``None`` or a string
explaining why to stop.
Note: The 'error in optimiser' criterion is not checked here.
Parameters
----------
iterations
The current number of iterations.
unchanged_f_iterations
The current number of iterations without a change in f (best or
guessed).
unchanged_x_iterations
The current number of iterations without a change in x (best or
guessed).
evaluations
The current number of function evaluations.
f_new
The current function value (best or guessed).
"""
# Maximum number of iterations
if (self._max_iterations is not None and
iterations >= self._max_iterations):
return f'Maximum number of iterations reached ({iterations}).'
# Maximum number of evaluations
if (self._max_evaluations is not None and
evaluations >= self._max_evaluations):
return ('Maximum number of evaluations reached'
f' ({self._max_evaluations}).')
# Maximum number of iterations without significant change in f
if (self._ftol_max is not None and
unchanged_f_iterations >= self._ftol_max):
return ('No significant change in best function evaluation for'
f' {unchanged_f_iterations} iterations.')
# Maximum number of iterations without significant change in x
if (self._xtol_max is not None and
unchanged_x_iterations >= self._xtol_max):
return ('No significant change in best parameters for'
f' {unchanged_x_iterations} iterations.')
# Threshold function value
if (self._function_threshold is not None and
f_new < self._function_threshold):
return ('Objective function crossed threshold ('
f'{self._function_threshold}).')
# All ok
return None
[docs]
def evaluations(self):
"""
Returns the number of evaluations performed during the last run, or
``None`` if the controller hasn't ran yet.
"""
return self._evaluations
[docs]
def f_guessed_tracking(self):
"""
Returns ``True`` if the controller is set to track the optimiser
progress using :meth:`pints.Optimiser.f_guessed()` rather than
:meth:`pints.Optimiser.f_best()`.
See also :meth:`set_f_guessed_tracking`.
"""
return self._use_f_guessed
[docs]
def function_tolerance(self):
"""
Returns a tuple ``(iterations, threshold)`` specifying the maximum
iterations without a significant change in best function evaluation,
if this stopping criterion is set, else ``(None, None)``.
The entries in the tuple correspond directly to the arguments to
:meth:`set_function_tolerance()`.
"""
if self._ftol_max is None:
return (None, None)
return (self._ftol_max, self._ftol_threshold)
def _has_stopping_criterion(self):
""" Returns whether a stopping criterion has been set. """
return any((
self._max_iterations is not None,
self._max_evaluations is not None,
self._ftol_max is not None,
self._xtol_max is not None,
self._function_threshold is not None,
))
[docs]
def iterations(self):
"""
Returns the number of iterations performed during the last run, or
``None`` if the controller hasn't ran yet.
"""
return self._iterations
[docs]
def max_evaluations(self):
"""
Returns the maximum number of evaluations if this stopping criterion is
set, or ``None`` if it is not.
See :meth:`set_max_evaluations`.
"""
return self._max_evaluations
[docs]
def max_iterations(self):
"""
Returns the maximum iterations if this stopping criterion is set, or
``None`` if it is not.
See :meth:`set_max_iterations()`.
"""
return self._max_iterations
[docs]
def max_unchanged_iterations(self):
"""
Deprecated alias of :meth:`function_tolerance()`.
"""
# Deprecated on 2026-02-05
import warnings
warnings.warn(
'The method `max_unchanged_iterations` is deprecated.'
' Please use `function_tolerance` instead.')
return self.function_tolerance()
[docs]
def optimiser(self):
"""
Returns the underlying optimiser object, allowing detailed
configuration.
"""
return self._optimiser
[docs]
def parallel(self):
"""
Returns the number of parallel worker processes this routine will be
run on, or ``False`` if parallelisation is disabled.
"""
return self._n_workers if self._parallel else False
[docs]
def parameter_tolerance(self):
"""
Returns a tuple ``(iterations, threshold)`` specifying the maximum
iterations without a significant change in best parameters, if this
stopping criterion is set, else ``(None, None)``.
The entries in the tuple correspond directly to the arguments to
:meth:`set_parameter_tolerance()`.
"""
if self._xtol_max is None:
return (None, None)
return (self._xtol_max, self._xtol_threshold)
[docs]
def run(self):
"""
Runs the optimisation, returns a tuple ``(x, f)``.
The returned ``x`` and ``f`` correspond to either the best ``f`` seen
during the optimisation, or to the best guessed ``f``, depending on the
setting for :meth:`set_f_guessed_tracking()`. See
:meth:Optimiser.f_guessed()` for details.
An optional ``callback`` function can be passed in that will be called
at the end of every iteration. The callback should take the arguments
``(iteration, optimiser)``, where ``iteration`` is the iteration count
(an integer) and ``optimiser`` is the optimiser object.
"""
# Can only run once for each controller instance
if self._has_run:
raise RuntimeError("Controller is valid for single use only")
self._has_run = True
# Check if any stopping criteria have been set
if not self._has_stopping_criterion():
raise ValueError('At least one stopping criterion must be set.')
# Iterations and function evaluations
iteration = 0
evaluations = 0
# Number of iterations without a change in f(x) or x
unchanged_f_iterations = 0
unchanged_x_iterations = 0
# Choose method to evaluate
f = self._function
if self._needs_sensitivities:
f = f.evaluateS1
# Create evaluator object
if self._parallel:
# Get number of workers
n_workers = self._n_workers
# For population based optimisers, don't use more workers than
# particles!
if isinstance(self._optimiser, PopulationBasedOptimiser):
n_workers = min(n_workers, self._optimiser.population_size())
evaluator = pints.ParallelEvaluator(f, n_workers=n_workers)
else:
evaluator = pints.SequentialEvaluator(f)
# Keep track of current best and best-guess scores.
fb = fg = np.inf
# Internally we always minimise! Keep a 2nd value to show the user.
fb_user, fg_user = (fb, fg) if self._minimising else (-fb, -fg)
# Keep track of the last significant change in f and x
f_sig = np.inf
x_sig = np.ones(self._function.n_parameters()) * np.inf
# Set up progress reporting
next_message = 0
# Start logging
logging = self._log_to_screen or self._log_filename
if logging:
if self._log_to_screen:
# Show direction
if self._minimising:
print('Minimising error measure')
else:
print('Maximising LogPDF')
# Show method
print('Using ' + str(self._optimiser.name()))
# Show parallelisation
if self._parallel:
print('Running in parallel with ' + str(n_workers) +
' worker processes.')
else:
print('Running in sequential mode.')
# Show population size
pop_size = 1
if isinstance(self._optimiser, PopulationBasedOptimiser):
pop_size = self._optimiser.population_size()
if pop_size is None:
pop_size = self._optimiser.suggested_population_size(
n_workers if self._parallel else None)
self._optimiser.set_population_size(pop_size)
if self._log_to_screen:
print('Population size: ' + str(pop_size))
# Set up logger
logger = pints.Logger()
if not self._log_to_screen:
logger.set_stream(None)
if self._log_filename:
logger.set_filename(self._log_filename, csv=self._log_csv)
# Add fields to log
max_iter_guess = max(self._max_iterations or 0, 10000)
max_eval_guess = max(
self._max_evaluations or 0, max_iter_guess * pop_size)
logger.add_counter('Iter.', max_value=max_iter_guess)
logger.add_counter('Eval.', max_value=max_eval_guess)
logger.add_float('Best')
logger.add_float('Current')
self._optimiser._log_init(logger)
# Note: No units shown in time field, for the reason why see
# https://github.com/pints-team/pints/issues/1467
logger.add_time('Time')
# Start searching
timer = pints.Timer()
running = True
try:
while running:
# Get points
xs = self._optimiser.ask()
# Calculate scores
fs = evaluator.evaluate(xs)
# Perform iteration
self._optimiser.tell(fs)
# Update current scores
fb = self._optimiser.f_best()
fg = self._optimiser.f_guessed()
fb_user, fg_user = (fb, fg) if self._minimising else (-fb, -fg)
f_new = fg if self._use_f_guessed else fb
# Check for significant changes in f or in x
if self._ftol_max:
if np.abs(f_new - f_sig) >= self._ftol_threshold:
unchanged_f_iterations = 0
# Note: f_sig is only updated after a change, so that a
# slow drift that becomes significant over multiple
# iterations is still detected.
f_sig = f_new
else:
unchanged_f_iterations += 1
if self._xtol_max:
x_new = (self._optimiser.x_guessed() if self._use_f_guessed
else self._optimiser.x_best())
if np.any(np.abs(x_new - x_sig) >= self._xtol_threshold):
unchanged_x_iterations = 0
# Note: Only update here (see above)
x_sig = x_new
else:
unchanged_x_iterations += 1
# Update evaluation count
evaluations += len(fs)
# Show progress
if logging and iteration >= next_message:
# Log state
logger.log(iteration, evaluations, fb_user, fg_user)
self._optimiser._log_write(logger)
logger.log(timer.time())
# Choose next logging point
if iteration < self._message_warm_up:
next_message = iteration + 1
else:
next_message = self._message_interval * (
1 + iteration // self._message_interval)
# Update iteration count
iteration += 1
# Check stopping criteria, set message if stopping
halt_message = self._check_stopping_criteria(
iteration, unchanged_f_iterations, unchanged_x_iterations,
evaluations, f_new)
running = halt_message is None
# Error in optimiser
error = self._optimiser.stop()
if error: # pragma: no cover
running = False
halt_message = str(error)
elif self._callback is not None:
self._callback(iteration - 1, self._optimiser)
except (Exception, SystemExit, KeyboardInterrupt): # pragma: no cover
# Unexpected end!
# Show last result and exit
print('\n' + '-' * 40)
print('Unexpected termination.')
print('Current score: ' + str(fg_user))
print('Current position:')
# Show current parameters
x_user = self._optimiser.x_guessed()
if self._transformation is not None:
x_user = self._transformation.to_model(x_user)
for p in x_user:
print(pints.strfloat(p))
print('-' * 40)
raise
# Stop timer
self._time = timer.time()
# Log final values and show halt message
if logging:
if iteration - 1 < next_message:
logger.log(iteration, evaluations, fb_user, fg_user)
self._optimiser._log_write(logger)
logger.log(self._time)
if self._log_to_screen:
print('Halting: ' + halt_message)
# Save post-run statistics
self._evaluations = evaluations
self._iterations = iteration
# Get best parameters
if self._use_f_guessed:
x = self._optimiser.x_guessed()
f = self._optimiser.f_guessed()
else:
x = self._optimiser.x_best()
f = self._optimiser.f_best()
# Inverse transform search parameters
if self._transformation is not None:
x = self._transformation.to_model(x)
# Return best position and score
return x, f if self._minimising else -f
[docs]
def set_callback(self, cb=None):
"""
Allows a "callback" function to be passed in that will be called at the
end of every iteration.
This can be used for e.g. visualising optimiser progress.
Example::
def cb(opt):
plot(opt.xbest())
opt.set_callback(cb)
"""
if cb is not None and not callable(cb):
raise ValueError('The argument cb must be None or a callable.')
self._callback = cb
[docs]
def set_f_guessed_tracking(self, use_f_guessed=False):
"""
Sets the method used to track the optimiser progress to
:meth:`pints.Optimiser.f_guessed()` or
:meth:`pints.Optimiser.f_best()` (default).
The tracked ``f`` (and/or ``x``) value is used to evaluate stopping
criteria, and is the one returned from :meth:`run`.
"""
self._use_f_guessed = bool(use_f_guessed)
[docs]
def set_log_interval(self, iters=20, warm_up=3):
"""
Changes the frequency with which messages are logged.
Parameters
----------
interval
A log message will be shown every ``iters`` iterations.
warm_up
A log message will be shown every iteration, for the first
``warm_up`` iterations.
"""
iters = int(iters)
if iters < 1:
raise ValueError('Interval must be greater than zero.')
warm_up = max(0, int(warm_up))
self._message_interval = iters
self._message_warm_up = warm_up
[docs]
def set_log_to_file(self, filename=None, csv=False):
"""
Enables logging to file when a filename is passed in, disables it if
``filename`` is ``False`` or ``None``.
The argument ``csv`` can be set to ``True`` to write the file in comma
separated value (CSV) format. By default, the file contents will be
similar to the output on screen.
"""
if filename:
self._log_filename = str(filename)
self._log_csv = True if csv else False
else:
self._log_filename = None
self._log_csv = False
[docs]
def set_log_to_screen(self, enabled):
"""
Enables or disables logging to screen.
"""
self._log_to_screen = True if enabled else False
[docs]
def set_max_evaluations(self, evaluations=None):
"""
Adds a stopping criterion so that the routine halts after the given
number of function ``evaluations``.
This criterion is disabled by default. To enable, pass in any positive
integer. To disable again, use ``set_max_evaluations(None)``.
"""
if evaluations is not None:
evaluations = int(evaluations)
if evaluations < 0:
raise ValueError(
'Maximum number of evaluations cannot be negative.')
self._max_evaluations = evaluations
[docs]
def set_max_iterations(self, iterations=10000):
"""
Adds a stopping criterion so that the routine halts after the given
number of ``iterations``.
This criterion is enabled by default. To disable it, use
``set_max_iterations(None)``.
"""
if iterations is not None:
iterations = int(iterations)
if iterations < 0:
raise ValueError(
'Maximum number of iterations cannot be negative.')
self._max_iterations = iterations
[docs]
def set_max_unchanged_iterations(self, iterations=200, threshold=1e-11):
"""
Deprecated alias of :meth:`function_tolerance()`.
"""
# Deprecated on 2026-02-05
import warnings
warnings.warn(
'The method `set_max_unchanged_iterations` is deprecated.'
' Please use `set_function_tolerance` instead.')
self.set_function_tolerance(iterations, threshold)
[docs]
def set_function_tolerance(self, iterations=200, threshold=1e-11):
"""
Adds a stopping criterion so that the routine halts if the objective
function does not change by more than ``threshold`` for the given
number of ``iterations``.
This criterion is enabled by default. To disable it, use
``set_function_tolerance(None)``.
Note that this can be used to implement an absolute "ftol" stopping
criteria, by calling
``set_function_tolerance(1, ftol)``.
"""
if iterations is not None:
iterations = int(iterations)
if iterations < 0:
raise ValueError(
'Maximum number of iterations cannot be negative.')
threshold = float(threshold)
if threshold < 0:
raise ValueError(
'Minimum significant function change cannot be negative.')
else:
threshold = None
self._ftol_max = iterations
self._ftol_threshold = threshold
[docs]
def set_parameter_tolerance(self, iterations=200, threshold=1e-11):
"""
Adds a stopping criterion so that the routine halts if the position in
parameter space does not change by more ``threshold`` for the given
number of ``iterations``.
Thresholds can be defined per parameter, or a single scalar value can
be passed in. The position is deemed to have changed if
``np.any(np.abs(x_new - x_sig) >= threshold)``, where ``x_sig`` is
either the starting position, or the last position for which the
criterion was met.
This criterion is disabled by default. Once enabled, it can be disabled
again by calling ``set_parameter_tolerance(None)``.
Note that this can be used to implement an absolute "xtol" stopping
criteria, by calling
``set_parameter_tolerance(1, xtol)``.
"""
if iterations is not None:
iterations = int(iterations)
if iterations < 0:
raise ValueError(
'Maximum number of iterations cannot be negative.')
# Test threshold size, convert scalar if needed, check sign
n_parameters = self._function.n_parameters()
if np.isscalar(threshold):
threshold = np.ones(n_parameters) * float(threshold)
elif len(threshold) == n_parameters:
threshold = pints.vector(threshold)
else:
raise ValueError(
'Minimum significant parameter change must be a scalar or'
f' have length {n_parameters}, got {len(threshold)}.')
if np.any(threshold < 0):
raise ValueError(
'Minimum significant parameter change cannot be negative.')
else:
threshold = None
self._xtol_max = iterations
self._xtol_threshold = threshold
[docs]
def set_parallel(self, parallel=False):
"""
Enables/disables parallel evaluation.
If ``parallel=True``, the method will run using a number of worker
processes equal to the detected cpu core count. The number of workers
can be set explicitly by setting ``parallel`` to an integer greater
than 0.
Parallelisation can be disabled by setting ``parallel`` to ``0`` or
``False``.
"""
if parallel is True:
self._parallel = True
self._n_workers = pints.ParallelEvaluator.cpu_count()
elif parallel >= 1:
self._parallel = True
self._n_workers = int(parallel)
else:
self._parallel = False
self._n_workers = 1
[docs]
def set_threshold(self, threshold):
"""
Adds a stopping criterion causing the routine to stop once the
objective function is less than the given ``threshold``
(when minimising, or more when maximising).
This criterion is disabled by default, but can be enabled by calling
this method with a valid ``threshold``. To disable it, use
``set_treshold(None)``.
"""
if threshold is None:
self._function_threshold = None
else:
self._function_threshold = float(threshold)
[docs]
def threshold(self):
"""
Returns the threshold stopping criterion, or ``None`` if no threshold
stopping criterion is set. See :meth:`set_threshold()`.
"""
return self._function_threshold
[docs]
def time(self):
"""
Returns the time needed for the last run, in seconds, or ``None`` if
the controller hasn't run yet.
"""
return self._time
[docs]
class Optimisation(OptimisationController):
""" Deprecated alias for :class:`OptimisationController`. """
def __init__(
self, function, x0, sigma0=None, boundaries=None,
transformation=None, method=None):
# Deprecated on 2019-02-12
import warnings
warnings.warn(
'The class `pints.Optimisation` is deprecated.'
' Please use `pints.OptimisationController` instead.')
super().__init__(
function, x0, sigma0, boundaries, transformation, method=method)
[docs]
def optimise(
function, x0, sigma0=None, boundaries=None, transformation=None,
method=None):
"""
Finds the parameter values that minimise an :class:`ErrorMeasure` or
maximise a :class:`LogPDF`.
Parameters
----------
function
An :class:`pints.ErrorMeasure` or a :class:`pints.LogPDF` that
evaluates points in the parameter space.
x0
The starting point for searches in the parameter space. This value may
be used directly (for example as the initial position of a particle in
:class:`PSO`) or indirectly (for example as the center of a
distribution in :class:`XNES`).
sigma0
An optional initial standard deviation around ``x0``, or a parameter
representing the "scale" of the parameters. Can be specified either as
a scalar value (same value for all dimensions) or as an array with one
entry per dimension. Not all methods will use this information.
boundaries
An optional set of boundaries on the parameter space.
transformation
An optional :class:`pints.Transformation` to allow the optimiser to
search in a transformed parameter space. If used, points shown or
returned to the user will first be detransformed back to the original
space.
method
The class of :class:`pints.Optimiser` to use for the optimisation.
If no method is specified, :class:`CMAES` is used.
Returns
-------
x_best : numpy array
The best parameter set obtained
f_best : float
The corresponding score.
"""
return OptimisationController(
function, x0, sigma0, boundaries, transformation, method).run()
[docs]
def curve_fit(f, x, y, p0, boundaries=None, threshold=None, max_iter=None,
max_unchanged=200, verbose=False, parallel=False, method=None):
"""
Fits a function ``f(x, *p)`` to a dataset ``(x, y)`` by finding the value
of ``p`` for which ``sum((y - f(x, *p))**2) / n`` is minimised (where ``n``
is the number of entries in ``y``).
Returns a tuple ``(x_best, f_best)`` with the best position found, and the
corresponding value ``f_best = f(x_best)``.
Parameters
----------
f : callable
A function or callable class to be minimised.
x
The values of an independent variable, at which ``y`` was recorded.
y
Measured values ``y = f(x, p) + noise``.
p0
An initial guess for the optimal parameters ``p``.
boundaries
An optional :class:`pints.Boundaries` object or a tuple
``(lower, upper)`` specifying lower and upper boundaries for the
search. If no boundaries are provided an unbounded search is run.
threshold
An optional absolute threshold stopping criterium.
max_iter
An optional maximum number of iterations stopping criterium.
max_unchanged
A stopping criterion based on the maximum number of successive
iterations without a signficant change in ``f`` (see
:meth:`pints.OptimisationController`).
verbose
Set to ``True`` to print progress messages to the screen.
parallel
Allows parallelisation to be enabled.
If set to ``True``, the evaluations will happen in parallel using a
number of worker processes equal to the detected cpu core count. The
number of workers can be set explicitly by setting ``parallel`` to an
integer greater than 0.
method
The :class:`pints.Optimiser` to use. If no method is specified,
``pints.CMAES`` is used.
Returns
-------
x_best : numpy array
The best parameter set obtained.
f_best : float
The corresponding score.
Example
-------
::
import numpy as np
import pints
def f(x, a, b, c):
return a + b * x + c * x ** 2
x = np.linspace(-5, 5, 100)
y = f(x, 1, 2, 3) + np.random.normal(0, 1)
p0 = [0, 0, 0]
popt = pints.curve_fit(f, x, y, p0)
"""
# Test function
if not callable(f):
raise ValueError('The argument `f` must be callable.')
# Get problem dimension from p0
d = len(p0)
# First dimension of x and y must agree
x = np.asarray(x)
y = np.asarray(y)
if x.shape[0] != y.shape[0]:
raise ValueError(
'The first dimension of `x` and `y` must be the same.')
# Check boundaries
if not (boundaries is None or isinstance(boundaries, pints.Boundaries)):
lower, upper = boundaries
boundaries = pints.RectangularBoundaries(lower, upper)
# Create an error measure
e = _CurveFitError(f, d, x, y)
# Set up optimisation
opt = pints.OptimisationController(
e, p0, boundaries=boundaries, method=method)
# Set stopping criteria
opt.set_threshold(threshold)
opt.set_max_iterations(max_iter)
opt.set_function_tolerance(max_unchanged)
# Set parallelisation
opt.set_parallel(parallel)
# Set output
opt.set_log_to_screen(True if verbose else False)
# Run and return
return opt.run()
class _CurveFitError(pints.ErrorMeasure):
""" Error measure for :meth:`curve_fit()`. """
def __init__(self, function, dimension, x, y):
self.f = function
self.d = dimension
self.x = x
self.y = y
self.n = 1 / np.prod(y.shape) # Total number of points in data
def n_parameters(self):
return self.d
def __call__(self, p):
return np.sum((self.y - self.f(self.x, *p))**2) * self.n
[docs]
def fmin(f, x0, args=None, boundaries=None, threshold=None, max_iter=None,
max_unchanged=200, verbose=False, parallel=False, method=None):
"""
Minimises a callable function ``f``, starting from position ``x0``, using a
:class:`pints.Optimiser`.
Returns a tuple ``(x_best, f_best)`` with the best position found, and the
corresponding value ``f_best = f(x_best)``.
Parameters
----------
f
A function or callable class to be minimised.
x0
The initial point to search at. Must be a 1-dimensional sequence (e.g.
a list or a numpy array).
args
An optional tuple of extra arguments for ``f``.
boundaries
An optional :class:`pints.Boundaries` object or a tuple
``(lower, upper)`` specifying lower and upper boundaries for the
search. If no boundaries are provided an unbounded search is run.
threshold
An optional absolute threshold stopping criterium.
max_iter
An optional maximum number of iterations stopping criterium.
max_unchanged
A stopping criterion based on the maximum number of successive
iterations without a signficant change in ``f`` (see
:meth:`pints.OptimisationController`).
verbose
Set to ``True`` to print progress messages to the screen.
parallel
Allows parallelisation to be enabled.
If set to ``True``, the evaluations will happen in parallel using a
number of worker processes equal to the detected cpu core count. The
number of workers can be set explicitly by setting ``parallel`` to an
integer greater than 0.
method
The :class:`pints.Optimiser` to use. If no method is specified,
``pints.CMAES`` is used.
Example
-------
::
import pints
def f(x):
return (x[0] - 3) ** 2 + (x[1] + 5) ** 2
xopt, fopt = pints.fmin(f, [1, 1])
"""
# Test function
if not callable(f):
raise ValueError('The argument `f` must be callable.')
# Get problem dimension from x0
d = len(x0)
# Test extra arguments
if args is not None:
args = tuple(args)
# Check boundaries
if not (boundaries is None or isinstance(boundaries, pints.Boundaries)):
lower, upper = boundaries
boundaries = pints.RectangularBoundaries(lower, upper)
# Create an error measure
e = _FminError(f, d) if args is None else _FminErrorWithArgs(f, d, args)
# Set up optimisation
opt = pints.OptimisationController(
e, x0, boundaries=boundaries, method=method)
# Set stopping criteria
opt.set_threshold(threshold)
opt.set_max_iterations(max_iter)
opt.set_function_tolerance(max_unchanged)
# Set parallelisation
opt.set_parallel(parallel)
# Set output
opt.set_log_to_screen(True if verbose else False)
# Run and return
return opt.run()
class _FminError(pints.ErrorMeasure):
""" Error measure for :meth:`fmin()`. """
def __init__(self, f, d):
self.f = f
self.d = d
def n_parameters(self):
return self.d
def __call__(self, x):
return self.f(x)
class _FminErrorWithArgs(pints.ErrorMeasure):
""" Error measure for :meth:`fmin()` for functions with args. """
def __init__(self, f, d, args):
self.f = f
self.d = d
self.args = args
def n_parameters(self):
return self.d
def __call__(self, x):
return self.f(x, *self.args)