csle_agents.agents.mcs package

Subpackages

• csle_agents.agents.mcs.mcs_utils package
  • Submodules
  • csle_agents.agents.mcs.mcs_utils.gls_utils module
  • csle_agents.agents.mcs.mcs_utils.ls_utils module
  • csle_agents.agents.mcs.mcs_utils.mcs_fun module
  • Module contents

Submodules

csle_agents.agents.mcs.mcs_agent module

MIT License

Copyright (c) 2019 MCS developers https://github.com/vojha-code/Multilevel-Coordinate-Search

class csle_agents.agents.mcs.mcs_agent.MCSAgent(simulation_env_config: csle_common.dao.simulation_config.simulation_env_config.SimulationEnvConfig, emulation_env_config: Union[None, csle_common.dao.emulation_config.emulation_env_config.EmulationEnvConfig], experiment_config: csle_common.dao.training.experiment_config.ExperimentConfig, env: Optional[csle_common.dao.simulation_config.base_env.BaseEnv] = None, training_job: Optional[csle_common.dao.jobs.training_job_config.TrainingJobConfig] = None, save_to_metastore: bool = True)

Bases: csle_agents.agents.base.base_agent.BaseAgent

Multi-Level Coordinate Search Agent

MCS(exp_result: csle_common.dao.training.experiment_result.ExperimentResult, seed: int, random_seeds: List[int], training_job: csle_common.dao.jobs.training_job_config.TrainingJobConfig, u: List[int], v: List[int], smax: int, nf: int, stop: List[Union[float, int]], iinit: int, local: int, gamma: float, hess: numpy.ndarray[Any, numpy.dtype[numpy.float64]], stopping_actions: int, eps: float, n: int, prt: int = 1) → csle_common.dao.training.experiment_result.ExperimentResult

The Multilevel Coordinate Search algorithm

Parameters

• exp_result – the experiment result

• seed – the seed

• random_seeds – the list of random seeds

• training_job – the configuration of the training job

• u – the initial lower bound (“lower corner” in 3D)

• v – the initial upper bound (“upper corner” in 3D)

• smax – maximum level depth

• nf – maximum number of function calls

• stop – stopping test

• iinit – the initial list

• local – command for lsearch or no lsearch

• gamma – acceptable relative accuracy for local search

• hess –

• stopping_actions – number of stopping actions

• hess – the hessian of the multidimensional function

• eps – parameter value for the golden ratio

• n – the number of iterations

• prt – print option

Returns

the experiment result

basket(x: List[float], f: float, policy: Union[csle_common.dao.training.multi_threshold_stopping_policy.MultiThresholdStoppingPolicy, csle_common.dao.training.linear_threshold_stopping_policy.LinearThresholdStoppingPolicy], avg_metrics: Optional[Dict[str, Union[float, int]]], xmin: List[Union[float, List[float], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]], fmi: List[float], xbest: List[float], fbest: float, stop: List[Union[float, int]], nbasket: int, nsweep: int, nsweepbest: int, stopping_actions: int, loc: int = 1, flag: int = 1, ncall: Union[float, int] = 0)

Function representing the basket functional :param x: starting point :param f: function value :param policy: current policy :param avg_metrics: current average metrics :param xmin: minum evaluation argumen (position) :param fmi: :param xbest: current best position :param fbest: current best function value :param stop: stopping test :param nbasket: counter for boxes in the ‘shopping basket’ :param nsweep: sweep counter :param nsweepbest: number of sweep in which fbest was updated for the last :param stopping_actions: number of stopping actions :return: a collection of parameters and metrics after the basket functional

basket1(x: numpy.ndarray[Any, numpy.dtype[numpy.float64]], f: float, xmin: List[Union[float, List[float], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]], fmi: List[float], xbest: List[float], fbest: float, stop: List[Union[float, int]], nbasket: int, nsweep: int, nsweepbest: int, stopping_actions: int, loc: int = 1, flag: int = 1, ncall: int = 0)

Basket 1 :param x: starting point :param f: function value(s) :param xmin: current minimum evaluation argument (position) :param fmi: :param xbest: current best evaluation argument (position) :param fbest: current best function value :param stop: stopping test :param nbasket: counter for boxes in the ‘shopping basket’ :param nsweep: sweep counter :param nsweepbest: number of sweep in which fbest was updated for the last :param stopping_actions: number of stopping actions :return: the metrics and parameters from basket1

static compute_avg_metrics(metrics: Dict[str, List[Union[float, int]]]) → Dict[str, Union[float, int]]

Computes the average metrics of a dict with aggregated metrics

Parameters

metrics – the dict with the aggregated metrics

Returns

the average metrics

csearch(x: List[Union[float, int]], f: float, u: List[int], v: List[int], hess: numpy.ndarray[Any, numpy.dtype[numpy.float64]], stopping_actions: int, eps: float)

Performs the csearch algorithm :param x: starting point :param f: function value :param u: lower initial guess (“Lower corner” in 3D) :param v: upper initial guess (“upper corner” in 3D) :param hess: the function Hessian :param stopping_actions: the number of stopping actions :return: a collection of parameters and metrics after doing the csearch

eval_theta(policy: Union[csle_common.dao.training.multi_threshold_stopping_policy.MultiThresholdStoppingPolicy, csle_common.dao.training.linear_threshold_stopping_policy.LinearThresholdStoppingPolicy], max_steps: int = 200) → Dict[str, Union[float, int]]

Evaluates a given threshold policy by running monte-carlo simulations

Parameters

policy – the policy to evaluate

Returns

the average metrics of the evaluation

get_policy(theta: Union[List[Union[float, int]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]], L: int) → Union[csle_common.dao.training.multi_threshold_stopping_policy.MultiThresholdStoppingPolicy, csle_common.dao.training.linear_threshold_stopping_policy.LinearThresholdStoppingPolicy]

Gets the policy of a given parameter vector

Parameters

• theta – the parameter vector

• L – the number of parameters

Returns

the policy

gls(xl: List[int], xu: List[int], x: List[Union[float, int]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.int32, numpy.float64]]], alist: List[Union[float, int]], flist: List[Union[float, int]], nloc: int, small: Union[float, int], smax: int, stopping_actions: int, prt: int = 2, short: float = 0.381966, bend: int = 0)

Global line search main function

Parameters

• func – funciton name which is subjected to optimization

• xl – lower bound

• xu – upper bound

• x – starting point

• p – search direction

• alist – list of known steps

• flist – function values of known steps

• nloc – (for local ~= 0) counter of points that have been

• small – tolerance values

• smax – search list size

• prt – print - unsued in this implementation so far

• short –

• bend –

Returns

search list,function values,number of fucntion evaluation

hparam_names() → List[str]

Function that contains the hyperparameter names

Returns

a list with the hyperparameter names

init_list(theta0: numpy.ndarray[Any, numpy.dtype[numpy.int32]], l: numpy.ndarray[Any, numpy.dtype[numpy.int32]], L: numpy.ndarray[Any, numpy.dtype[numpy.int32]], stopping_actions: int, n: int, ncall: int = 0) → Tuple[numpy.ndarray[Any, numpy.dtype[numpy.float32]], numpy.ndarray[Any, numpy.dtype[numpy.float32]], int, Union[csle_common.dao.training.multi_threshold_stopping_policy.MultiThresholdStoppingPolicy, csle_common.dao.training.linear_threshold_stopping_policy.LinearThresholdStoppingPolicy]]

Computes the function values corresponding to the initialization list and the pointer istar to the final best point x^* of the init. list :param theta0: theta0 :param l: Indication of the mid point :param L: Indication of the end point (or total number of partition of the value x in the i’th dimenstion) :param stopping actions: stopping actions for the eval_theta function :param n: dimension (should equal the number of stopping actions) :return : initial conditions

lsdescent(x: Union[List[Union[float, int]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], alist: List[Union[float, int]], flist: List[Union[float, int]], alp: Union[int, float], abest: float, fbest: float, fmed: float, up: List[float], down: List[float], monotone: int, minima: List[int], nmin: int, unitlen: float, s: int, stopping_actions: int)

The lsdescent algorithm :param x: starting point :param p: search direction :param alist: list of known steps :param flist: function values of known steps :param alp: :param abest: best step :param fbest: best function value so far :param fmed: :param up: :param down: :param monotone: :param minima: :param nmin: :param unitlen: :param s: the current depth level :param stopping_actions: number of stopping actions :return: the set pf parameters and metrics obtained from performing lsdescent

lsearch(x: List[Union[float, int]], f: float, f0: numpy.ndarray[Any, numpy.dtype[numpy.float64]], u: List[int], v: List[int], nf: int, stop: List[Union[float, int]], maxstep: int, gamma: float, hess: numpy.ndarray[Any, numpy.dtype[numpy.float64]], nsweep: int, nsweepbest: int, stopping_actions: int, eps: float, ncall: Union[float, int] = 0, flag: int = 1, eps0: float = 0.001, nloc: int = 1, small: float = 0.1, smaxls: int = 15, diag: int = 0, nstep: int = 0)

The local search algorithm :param x: starting point :param f: function value :param f0: function value :param u: lower initial guess (“lower corner” in 3D) :param v: initial upper guess (“upper corner” in 3D) :param nf: :param stop: stopping test :param maxstep: maximum steps in the local search (mainly determined by the local command) :param gamma: acceptable relative accuracy for local search :param hess: the function Hessian :param nsweep: sweep counter :param nsweepbest: number of sweep in which fbest was updated for the last :param stopping_actions: number of stopping actions :param eps: parameter value for the golden ratio :return: a collection of parameters and metrics afdter the local search

lsinit(x, p, alist, flist, amin, amax, scale, stopping_actions)

Line search algorithm

Parameters

• x – starting point

• p – search direction

• alist – list of known steps

• flist – function values of known steps

• amin –

• amax –

• scale –

• stopping_actions – number of stopping actions

Returns

set of parameters obtained from performing the line search

lslocal(nloc: int, small: float, sinit: int, short: float, x: Union[List[Union[float, int]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], alist: List[Union[float, int]], flist: List[float], amin: float, amax: float, alp: float, abest: float, fbest: float, fmed: float, up: List[float], down: List[float], monotone: int, minima: List[Union[int, float, bool]], nmin: int, unitlen: float, s: int, saturated: int, stopping_actions: int)

The lslocal function :param nloc: (for local ~= 0) counter of points that have been :param small: tolerance values :param sinit: the initial depth level: :param short: :param x: starting point :param p: search direction :param alist: list of known steps :param flist: function values of known steps :param amin: :param amax: :param alp: :param abest: best step :param fbest: best function value so far :param fmed: median function value :param up: :param down: :param monotone: if function is monotone or not :param minima: :param nmin: :param unitlen: :param s: current depth level :return: the parameters and metrics obtained from lslocal

lsnew(nloc: int, small: Union[float, int], sinit: int, short: float, x: Union[List[Union[float, int]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], s: int, alist: List[Union[float, int]], flist: List[Union[float, int]], amin: float, amax: float, alp: Union[int, float], abest: float, fmed: float, unitlen: float, stopping_actions: int)

The lsnew function :param nloc: (for local ~= 0) counter of points that have been :param small: tolerance values :param sinit: :param short: :param x: starting point :param p: search direction :param s: current depth level :param alist: list of known steps :param flist: function values of known steps :param amin: :param amax: :param alp: :param abest: best step :param fmed: :param unitlen: :param stopping_actions :return: set of parameters and metrics obtained after performing lsnew

lspar(nloc: int, small: Union[float, int], sinit: int, short: float, x: Union[List[Union[float, int]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], alist: List[Union[float, int]], flist: List[Union[float, int]], amin: float, amax: float, alp: Union[int, float], abest: float, fbest: float, fmed: float, up: List[float], down: List[float], monotone: int, minima: List[int], nmin: int, unitlen: float, s: int, stopping_actions: int)

The lspar function :param nloc: (for local ~= 0) counter of points that have been :param small: tolerance values :param sinit: length of list of known steps :param short: :param x: starting point :param p: search direction :param alist: list of known steps :param flist: function values of known steps :param amin: :param amax: :param alp: :param abest: best step :param fbest: best function value so far :param fmed: :param up: :param down: :param monotone: :param minima: :param nmin: :param unitlen: :param s: :param stopping_actions: number if stopping actions :return: the set of parameters and metrics after performing lspar

lsquart(nloc: int, small: Union[float, int], sinit: int, short: float, x: numpy.ndarray[Any, numpy.dtype[numpy.float64]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], alist: List[Union[float, int]], flist: List[float], amin: float, amax: float, alp: float, abest: float, fbest: float, fmed: float, up: List[float], down: List[float], monotone: int, minima: List[Union[int, float, bool]], nmin: int, unitlen: float, s: int, saturated: int, stopping_actions: int)

The lsaquart function :param nloc: (for local ~= 0) counter of points that have been :param small: tolerance values :param sinit: initial depth level :param short: :param x: starting point :param p: search direction :param alist: list of known steps :param flist: function values of known steps :param amin: :param amax: :param alp: :param up: :param down: :param monotone: :param minima: :param nmin :param unitlen: :param s: the current depth level :param saturated: :param stopping_actions: the number of stopping actions :return: the parameters and metrics obtained from performing lsquart

lssep(nloc: int, small: float, sinit: int, short: float, x: Union[List[Union[float, int]], numpy.ndarray[Any, numpy.dtype[numpy.float64]]], p: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], alist: List[Union[float, int]], flist: List[float], amin: float, amax: float, alp: float, abest: float, fbest: float, fmed: float, up: List[float], down: List[float], monotone: int, minima: List[int], nmin: int, unitlen: float, s: int, stopping_actions: int)

The lssep function :param nloc: (for local ~= 0) counter of points that have been :param small: tolerance values :param sinit: initial depth levekl :param short: :param x: starting point :param p: search direction :param alist: list of known steps :param flist: function values of known steps :param amin: :param amax: :param alp: :param abest: best step :param fbest: best function value so far :param fmed: median function value :param up: :param down: :param monotone: :param minima: :param nmin: :param untilen: :param s: current depth level :param stopping_actions: the number of stopping actions :return: the parameters and metrics obtained from performing lssep

static round_vec(vec) → List[float]

Rounds a vector to 3 decimals

Parameters

vec – the vector to round

Returns

the rounded vector

splinit(i: int, s: int, smax: int, par: int, x0: numpy.ndarray[Any, numpy.dtype[numpy.int32]], n0: int, u: List[int], v: List[int], x: numpy.ndarray[Any, numpy.dtype[numpy.float64]], y: numpy.ndarray[Any, numpy.dtype[numpy.float64]], x1: numpy.ndarray[Any, numpy.dtype[numpy.float64]], x2: numpy.ndarray[Any, numpy.dtype[numpy.float64]], L: numpy.ndarray[Any, numpy.dtype[numpy.int32]], l: numpy.ndarray[Any, numpy.dtype[numpy.int32]], xmin: List[Union[float, List[float], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]], fmi: List[float], ipar: numpy.ndarray[Any, numpy.dtype[numpy.int32]], level: numpy.ndarray[Any, numpy.dtype[numpy.int32]], ichild: numpy.ndarray[Any, numpy.dtype[numpy.int32]], f: numpy.ndarray[Any, numpy.dtype[numpy.float64]], xbest: numpy.ndarray[Any, numpy.dtype[numpy.float64]], fbest: numpy.ndarray[Any, numpy.dtype[numpy.float64]], stop: List[Union[float, int]], prt: int, record: numpy.ndarray[Any, numpy.dtype[Union[numpy.int32, numpy.float64]]], nboxes: int, nbasket: int, nsweepbest: int, nsweep: int, stopping_actions: int, ncall: int = 0, nchild: int = 0)

Splitting box at specified level s according to an initialization list :param i: specified index :param s: current depth level :param smax: maximum depth level :param par: :param x0: initial position :param n0: :param u: initial lower guess (“lower corner” in 3D) :param v: initial upper guess (“upper corner” in 3D) :param x: starting point :param y: :param x1: evaluation argument (position) :param x2: evaluation argument (position) :param L: :param l: :param xmin: evaluation argument (position) :param fmi: :param ipar: :param level: :param ichild: :param f: function value :param xbest: best evaluation argument (position) :param fbest: best function value :param stop: stopping test :param prt: print - unsued in this implementation so far :param record: :param nboxes: counter for boxes not in the ‘shopping bas :param nbasket: counter for boxes in the ‘shopping bas :param nsweepbest: number of sweep in which fbest was updated for the last :param nsweep: sweep counter :stopping_actions: number of stopping actions :return: a collection of parameters and metrics from the initial split

split(i: int, s: int, smax: int, par: int, n0: int, u: List[int], v: List[int], x: numpy.ndarray[Any, numpy.dtype[numpy.float64]], y: numpy.ndarray[Any, numpy.dtype[numpy.float64]], x1: numpy.ndarray[Any, numpy.dtype[numpy.float64]], x2: numpy.ndarray[Any, numpy.dtype[numpy.float64]], z: numpy.ndarray[Any, numpy.dtype[numpy.float64]], xmin: List[Union[float, List[float], numpy.ndarray[Any, numpy.dtype[numpy.float64]]]], fmi: List[float], ipar: numpy.ndarray[Any, numpy.dtype[numpy.int32]], level: numpy.ndarray[Any, numpy.dtype[numpy.int32]], ichild: numpy.ndarray[Any, numpy.dtype[numpy.int32]], f: numpy.ndarray[Any, numpy.dtype[numpy.float64]], xbest: numpy.ndarray[Any, numpy.dtype[numpy.float64]], fbest: numpy.ndarray[Any, numpy.dtype[numpy.float64]], stop: List[Union[float, int]], prt: int, record: numpy.ndarray[Any, numpy.dtype[Union[numpy.float64, numpy.int32]]], nboxes: int, nbasket: int, nsweepbest: int, nsweep: int, stopping_actions: int, ncall: int = 0, flag: int = 1)

Function that performs a box split :param i: :param s: current depth level :param smax: maximum depth level :param par: :param n0: :param u: initial lower guess (“lower corner” in 3D) :param v: initial upper guess (“upper corner” in 3D) :param x: starting point :param y: :param x1: evaluation argument (position) :param x2: evaluation argument (position) :param param z: :param xmin: minimum position :param fmi: :param ipar: :param level: :param ichild: :param f: function value :param xbest: currently best position :param fbest: current best function value’ :param stop: stopping test :param prt: print - unsued in this implementation so far :param record: :param nboxes: counter for boxes not in the ‘shopping bas :param nbasket: counter for boxes in the ‘shopping basket’ :param nsweepbest: number of sweep in which fbest was updated for the last :param nsweep: sweep counter :param stopping_actions: the number of stopping actions :return: a collection of parameters and metrics afdter the arbitrary split

train() → csle_common.dao.training.experiment_execution.ExperimentExecution

Initiating the parameters of performing the MCS algorithm, using external functions

Returns

The experiment execution

triple(x: List[Union[float, int]], f: float, x1: List[Union[float, int]], x2: List[Union[float, int]], u: List[int], v: List[int], hess, G, stopping_actions, setG=False)

The triple function :param x: starting point :param f: function value :param x1: evaluation argument (position) :param x2: evaluation argument (position) :param u: lower initial guess (“Lower corner” in 3D) :param v: lower initial guess (“upper corner” in 3D) :param hess: the hessian of the function :param G: :param stopping_actions: number of stopping actions :param setG: :return: the set of parameters and metrics after performing the triple

static update_metrics(metrics: Dict[str, List[Union[float, int]]], info: Dict[str, Union[float, int]]) → Dict[str, List[Union[float, int]]]

Update a dict with aggregated metrics using new information from the environment

Parameters

• metrics – the dict with the aggregated metrics

• info – the new information

Returns

the updated dict of metrics

Module contents