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_gpdc
# 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)
# 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 a_test(request):
return ParCorr(verbosity=VERBOSITY)
def test_stability_parameters(unstable_parameter_sets):
"""
Test that the correct exceptions are raised for unstable lagged connectivity
matricies
"""
# Unpack the parameter set fixture
stab_matrix, unst_matrix, message = unstable_parameter_sets
error_message = message + " should trigger an assertion error for "+\
"_var_network graph."
# Test the good parameter set
try:
pp._check_stability(stab_matrix)
# Ensure no exception is raised
except:
pytest.fail("Stable matrix set triggers exception incorrectly!")
# Ensure an exception is raised for a bad parameter set
with pytest.raises(AssertionError):
pp._check_stability(unst_matrix)
pytest.fail(error_message)
def _get_parent_graph(parents_neighbors_coeffs, exclude=None):
"""
Iterates through the input parent-neighghbour coefficient dictionary to
return only parent relations (i.e. where tau != 0)
"""
graph = defaultdict(list)
for j, i, tau, _ in pp._iter_coeffs(parents_neighbors_coeffs):
if tau != 0 and (i, tau) != exclude:
graph[j].append((i, tau))
return dict(graph)
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
T = 100000
numpy.random.seed(None)
array = numpy.random.randn(3, T)
cov = numpy.array([[1., val_ana],[val_ana, 1.]])
array[:2, :] = numpy.random.multivariate_normal(
mean=numpy.zeros(2),
cov=cov, size=T).T
# Generate some confounding
if len(array) > 2:
array[0] += 0.5* array[2:].sum(axis=0)
array[1] += 0.7* array[2:].sum(axis=0)
# Transform to symbolic data
array = pp.quantile_bin_array(array.T, bins=16).T
dim, T = array.shape
xyz = numpy.array([0,1,2,2,2])
val_est = ci_cmi_symb.get_dependence_measure(array, xyz)
print(val_est)
print(_par_corr_to_cmi(val_ana))
numpy.testing.assert_allclose(numpy.array(_par_corr_to_cmi(val_ana)),
numpy.array(val_est),
atol=0.02)
def _get_parent_graph(parents_neighbors_coeffs, exclude=None):
"""
Iterates through the input parent-neighghbour coefficient dictionary to
return only parent relations (i.e. where tau != 0)
"""
graph = defaultdict(list)
for j, i, tau, _ in pp._iter_coeffs(parents_neighbors_coeffs):
if tau != 0 and (i, tau) != exclude:
graph[j].append((i, tau))
return dict(graph)
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 gen_process(a_process):
"""
Calls var_process for the process fixtures
"""
# Get the initial values and setup for the decay process
_, init_vals, coefs, expect = a_process
# Deducte the max time from the expected answer shape
max_time = expect.shape[0]
# Generate the data
data, true_parents_neighbors = pp.var_process(coefs,
T=max_time,
initial_values=init_vals,
use="no_noise")
return data, true_parents_neighbors