Worker designs for data science workflows.

amos uses the designs of chrisjen (flow, benchmark, contest, and survey) and adds one of its own.

Contents

Experiment: tries every combination of techniques, keeps the best, and records how every combination did.

Experiment(contents=dict(), name=None, parameters=dict(), criteria=None, results=dict(), winner=None, scores=dict()) dataclass

Bases: Contest

Tries every combination of techniques and records how each one did.

Like a contest, an experiment applies every combination of one technique from each step (every path through its graph) to its own copy of the dataset, and keeps the dataset with the best score from its criteria. It also adds a comparison table to the tables of the winning dataset, named "{name}_comparison", with one row for each combination: its score, its rank, and every metric it computed. And it keeps a Branch for each combination in the winning dataset's branches, so that a scorecard can compare every combination on every metric. This is what a paper reports to show that a result does not depend on one arbitrary choice of preprocessing or model.

If the criteria is a Metric, its score is also stored in the metrics of each combination's dataset, so the table shows the real value of the metric (not the negated value used to rank metrics for which lower is better).

Parameters:

Name	Type	Description	Default
name	str | None	name used to refer to the worker in a workflow.	None
contents	MutableMapping[Hashable, set[Hashable]]	the graph of nodes. Defaults to an empty dict.	dict()
parameters	GenericDict	keyword arguments for the worker to use. Defaults to an empty dict.	dict()
criteria	Criteria | None	the criteria used to score each result. Higher scores are better. Defaults to None.	None

Attributes:

Name	Type	Description
results	dict[str, Any]	the result of each path in the most recent call, by the names of the nodes in the path (joined by " > ").
scores	dict[str, Any]	the score of each path in the most recent call.
winner	str | None	label of the path with the best result in the most recent call.

scores = dataclasses.field(default_factory=dict) class-attribute instance-attribute

Public Methods

implement(item, **kwargs)

Returns the result of the path with the best score.

Parameters:

Name	Type	Description	Default
item	Any	data or object to change.	required
**kwargs	Any	keyword arguments passed to the nodes.	{}

Raises:

Type	Description
ValueError	if there are no paths or no criteria.

Returns:

Type	Description
Any	The best result. Ties go to the first path.

Source code in src/amos/workers.py

def implement(self, item: Any, **kwargs: Any) -> Any:
    """Returns the result of the path with the best score.

    Args:
        item: data or object to change.
        **kwargs: keyword arguments passed to the nodes.

    Raises:
        ValueError: if there are no paths or no `criteria`.

    Returns:
        The best result. Ties go to the first path.

    """
    results = self.try_paths(item, **kwargs)
    self.scores = dict(zip(
        results, self.compare(list(results.values())), strict = True))
    criteria = self.criteria
    if isinstance(criteria, metrics.Metric):
        # Stores the real value of the metric, which is negated in the
        # score if lower values are better.
        sign = 1 if criteria.greater_is_better else -1
        for label, result in results.items():
            if isinstance(result, base.Dataset):
                result.metrics[criteria.name] = sign * self.scores[label]
    self.winner = max(self.scores, key = self.scores.__getitem__)
    best = results[self.winner]
    if isinstance(best, base.Dataset):
        best.branches = self._branches(results)
        best.tables[f'{self.name}_comparison'] = self.tabulate()
        best.record(
            str(self.name),
            winner = self.winner,
            criterion = getattr(self.criteria, 'name', None),
            paths = len(results))
    return best

tabulate()

Returns a table comparing every path of the most recent call.

Returns:

Type	Description
DataFrame	One row for each path (labeled with the names of its nodes), sorted from best to worst, with "rank" and "score" columns and a column for each metric that the paths computed.

Source code in src/amos/workers.py

def tabulate(self) -> pd.DataFrame:
    """Returns a table comparing every path of the most recent call.

    Returns:
        One row for each path (labeled with the names of its nodes),
            sorted from best to worst, with "rank" and "score" columns and
            a column for each metric that the paths computed.

    """
    rows = {}
    for label, score in self.scores.items():
        result = self.results[label]
        values = dict(getattr(result, 'metrics', {}))
        rows[label] = {'score': score, **values}
    table = pd.DataFrame.from_dict(rows, orient = 'index')
    table.index.name = 'path'
    table = table.sort_values('score', ascending = False, kind = 'stable')
    table.insert(0, 'rank', range(1, len(table) + 1))
    return table