Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# ci_test = self.ci_par_corr
a = 0.
c = .3
T = 500
# Each key refers to a variable and the incoming links are supplied as a
# list of format [((driver, lag), coeff), ...]
links_coeffs = {0: [((0, -1), a)],
1: [((1, -1), a), ((0, -1), c)],
}
numpy.random.seed(42)
data, true_parents_neighbors = pp.var_process(
links_coeffs,
use='inv_inno_cov', T=T)
dataframe = pp.DataFrame(data)
ci_test.set_dataframe(dataframe)
# ci_test.set_tau_max(1)
# X=[(1, -1)]
# Y=[(1, 0)]
# Z=[(0, -1)] + [(1, -tau) for tau in range(1, 2)]
# array, xyz, XYZ = ci_test.get_array(X, Y, Z,
# verbosity=0)]
# ci_test.run_test(X, Y, Z,)
def func(x):
return x * (1. - 4. * x**0 * numpy.exp(-x**2 / 2.))
true_residual = numpy.random.randn(3, T)
array = numpy.copy(true_residual)
array[1] += c*func(array[2]) #.sum(axis=0)
xyz = numpy.array([0,1] + [2 for i in range(array.shape[0]-2)])
def gen_data_frame(links_coeffs, time, seed_val):
# Set the random seed
np.random.seed(seed_val)
# Generate the data
data, _ = pp.var_process(links_coeffs, T=time)
# Get the true parents
true_parents = _get_parent_graph(links_coeffs)
return pp.DataFrame(data), true_parents
def a_sample(request):
# Set the parameters
links_coeffs, time, seed_val = request.param
# Set the random seed
np.random.seed(seed_val)
# Generate the data
data, _ = pp.var_process(links_coeffs, T=time)
# Get the true parents
true_parents = _get_parent_graph(links_coeffs)
return pp.DataFrame(data), true_parents
pyplot.savefig(save_name)
else:
pyplot.show()
if __name__ == '__main__':
from tigramite.independence_tests import ParCorr
import tigramite.data_processing as pp
np.random.seed(42)
val_matrix = np.random.rand(3,3,4)
link_matrix = np.abs(val_matrix) > .7
# print link_matrix
data = np.random.randn(100,3)
mask = np.random.randint(0, 2, size=(100,3))
dataframe = pp.DataFrame(data, mask=mask)
# data = np.random.randn(100, 3)
# datatime = np.arange(100)
# mask = np.zeros(data.shape)
# mask[:int(len(data)/2)]=True
# data[:,0] = -99.
# plot_lagfuncs(val_matrix=val_matrix,
# setup_args={'figsize':(10,10),
# 'label_space_top':0.05,
# 'label_space_left':0.1,
# 'x_base':1, 'y_base':5,
# 'var_names':range(3),
# 'lag_array':np.array(['a%d' % i for i in range(4)])},
# (T, N)
numpy.random.seed(42) # Fix random seed
links_coeffs = {0: [((0, -1), 0.7)],
1: [((1, -1), 0.8), ((0, -1), 0.8)],
2: [((2, -1), 0.5), ((1, -2), 0.5)],
}
T = 500 # time series length
data, true_parents_neighbors = pp.var_process(links_coeffs, T=T)
T, N = data.shape
# Optionally specify variable names
var_names = [r'$X^0$', r'$X^1$', r'$X^2$', r'$X^3$']
# Initialize dataframe object
dataframe = pp.DataFrame(data, var_names=var_names)
# Significance level in condition-selection step. If a list of levels is is
# provided or pc_alpha=None, the optimal pc_alpha is automatically chosen via
# model-selection.
pc_alpha = 0.2 # [0.01, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5]
selected_variables = range(N) #[2] # [2] # [2]
# Maximum time lag
tau_max = 3
# Optional minimum time lag in MCI step (in PC-step this is 1)
tau_min = 0
# Maximum cardinality of conditions in PC condition-selection step. The
# recommended default choice is None to leave it unrestricted.
max_conds_dim = None
test_indices,
prediction_model,
cond_ind_test=None,
data_transform=None,
verbosity=0):
# Default value for the mask
mask = dataframe.mask
if mask is None:
mask = np.zeros(dataframe.values.shape, dtype='bool')
# Get the dataframe shape
T = len(dataframe.values)
# Have the default dataframe be the training data frame
train_mask = np.copy(mask)
train_mask[[t for t in range(T) if t not in train_indices]] = True
self.dataframe = DataFrame(dataframe.values,
mask=train_mask,
missing_flag=dataframe.missing_flag)
# Initialize the models baseclass with the training dataframe
Models.__init__(self,
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:
self.all_parents = all_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))