Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def a_pcmci(a_sample, a_test, a_common_params, request):
# Unpack the test data and true parent graph
dataframe, true_parents = a_sample
# Unpack the common parameters
tau_min, tau_max, sel_link = a_common_params
# Get the parameters from this request
select_vars = request.param
# Build the PCMCI instance
pcmci = PCMCI(selected_variables=select_vars,
dataframe=dataframe,
cond_ind_test=a_test,
verbosity=VERBOSITY)
# If there are selected variables, edit the true parents to reflect this
if select_vars is not None:
true_parents = {sel_v : true_parents[sel_v] for sel_v in select_vars}
# Select the correct links if they are given
select_links = _select_links(sel_link, true_parents)
# Ensure we change the true parents to be the same as the selected links
if select_links is not None:
true_parents = select_links
# Return the constructed PCMCI, expected results, and common parameters
return pcmci, true_parents, tau_min, tau_max, select_links
def a_pcmci(a_sample, request):
# Unpack the test data and true parent graph
dataframe, true_parents = a_sample
# Build the PCMCI instance
pcmci = PCMCI(selected_variables=None,
dataframe=dataframe,
cond_ind_test=ParCorr(verbosity=VERBOSITY),
verbosity=VERBOSITY)
# Return the constructed PCMCI, expected results, and common parameters
return pcmci, true_parents
# Cache the resulting values in the return dictionary
return_dict = {'val_matrix': val_matrix,
'p_matrix': p_matrix,
'q_matrix': q_matrix,
'conf_matrix': conf_matrix}
# Print the information
if self.verbosity > 0:
self.print_results(return_dict)
# Return the dictionary
return return_dict
if __name__ == '__main__':
from tigramite.independence_tests import ParCorr
import tigramite.data_processing as pp
dataframe = pp.DataFrame(np.random.randn(100,3),)
pcmci = PCMCI(dataframe, ParCorr())
pcmci.get_corrected_pvalues(np.random.rand(2,2,2))
Parameters
----------
j : int
Variable index.
Returns
-------
j, pcmci_of_j, parents_of_j : tuple
Variable index, PCMCI object, and parents of j
"""
# CondIndTest is initialized globally below
# Further parameters of PCMCI as described in the documentation can be
# supplied here:
pcmci_of_j = PCMCI(
dataframe=dataframe,
cond_ind_test=cond_ind_test,
selected_variables=[j],
verbosity=verbosity)
# Run PC condition-selection algorithm. Also here further parameters can be
# specified:
parents_of_j = pcmci_of_j.run_pc_stable(
selected_links=selected_links,
tau_max=tau_max,
pc_alpha=pc_alpha,
)
# We return also the PCMCI object because it may contain pre-computed
# results can be re-used in the MCI step (such as residuals or null
# distributions)
dataframe=self.dataframe,
model=prediction_model,
data_transform=data_transform,
mask_type='y',
verbosity=verbosity)
# Build the testing dataframe as well
self.test_mask = np.copy(mask)
self.test_mask[[t for t in range(T) if t not in test_indices]] = True
# Setup the PCMCI instance
if cond_ind_test is not None:
# Force the masking
cond_ind_test.set_mask_type('y')
cond_ind_test.verbosity = verbosity
PCMCI.__init__(self,
dataframe=self.dataframe,
cond_ind_test=cond_ind_test,
selected_variables=None,
var_names=None,
verbosity=verbosity)
# Set the member variables
self.cond_ind_test = cond_ind_test
# Initialize member varialbes that are set outside
self.target_predictors = None
self.selected_targets = None
self.fitted_model = None
self.test_array = None