Controllers

class Controller

The HCCD-controller. Coordinates repeated CD-algorithm runs with state-space construction.

See also

Specialized versions for PCMCI-family time-series algorithms ControllerTimeseriesMCI and for LPCMCI ControllerTimeseriesLPCMCI are available.

__init__(universal_cd: abstract_cd_t, testing_backend: IProvideIndependenceAtoms, state_space_construction: IConstructStateSpace)

Construct from components.

Parameters:
get_marked_independence(ci)

provide an extension hook to attach an observer to marked-independence lookups by overriding this method, see e.g. ControllerTimeseriesLPCMCI.

run_cd(state: IRepresentState) → tuple[graph_t, set[CI_Identifier]]

Implements part of the “core-algorithm” Algo. 1 in [RR25].

Run underlying CD with a state-specific pseudo-cit.

Parameters:

state (IRepresentState) – state considered active

Returns:

tuple consisting of state-graph and set of (newly) marked tests.

Return type:

tuple[graph_t, set[CI_Identifier]]

run_hccd(max_iterations: int = 10) → IPresentResult

Implements part of the “core-algorithm” Algo. 1 from [RR25].

Run HCCD.

Parameters:

max_iterations (int, optional) – limit for iterations, defaults to 10

Raises:

RuntimeError – Throws an exception if maximum iterations are reached.

Returns:

hccd result

Return type:

IPresentResult

check_for_convergence(previously_marked: set[CI_Identifier], newly_marked: set[CI_Identifier], graphs) → tuple[set[CI_Identifier], bool]

Check if the core-algorithm has converged. May be overridden by more derived controllers.

class ControllerTimeseriesMCI

Timer-series specific controller, for PCMCI [RNK+19] family algorithms. See §B.6 in [RR25].

See also

Details are descibed at Controller.

__init__(universal_cd: abstract_cd_t, testing_backend: IProvideIndependenceAtoms, state_space_construction: IConstructStateSpace)

Construct from components.

Parameters:
class ControllerTimeseriesLPCMCI

LPCMCI [GR20] specific controller.

See also

Details are descibed at Controller.

__init__(universal_cd: abstract_cd_t, testing_backend: IProvideIndependenceAtoms, state_space_construction: IConstructStateSpace)

Construct from components.

Parameters:
check_for_convergence(previously_marked: set[CI_Identifier], newly_marked: set[CI_Identifier], graphs: dict) → tuple[set[CI_Identifier], bool]

Implements functionality of interface Controller.

Overriden from Controller to reset after an initial iteration (union-graph discovery), cf get_marked_independence().

lagged_parents(node: int, lag_shift: int = 0) → set[tuple[int, int]]
contains_all_lagged_parents(ci: CI_Identifier_TimeSeries) → bool
should_consider_mci(ci: CI_Identifier_TimeSeries) → bool
get_marked_independence(ci)

Implements functionality of interface Controller.

Overriden from Controller to attach logic for deciding if this test should be considerd MCI based on the union-graph.