File size: 5,159 Bytes
31d1d63 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | from itertools import product
import numpy as np
import pytest
from scoobdoob import _conditioning
from scoobdoob.validation import bridge
def test_conditioned_probabilities_equal_enumerated_weighted_path_law():
P = np.array([[0.6, 0.4, 0], [0.2, 0.4, 0.4], [0.3, 0.2, 0.5]])
weights = np.array(
[[1.0, 2.0, 3.0], [2.0, 0.0, 1.0], [0.5, 2.0, 1.0], [1.0, 3.0, 0.8]]
)
kernels, _ = _conditioning.build(P, weights, [1, 2], 4, starts=[0, 2])
for start in [0, 2]:
raw, actual = [], []
for tail in product(range(3), repeat=4):
path = (start,) + tail
reference = float(tail[-1] in [1, 2])
probability = 1.0
for n in range(4):
reference *= P[path[n], path[n + 1]] * weights[n, path[n + 1]]
probability *= kernels[n][path[n], path[n + 1]]
raw.append(reference)
actual.append(probability)
np.testing.assert_allclose(
actual, np.array(raw) / sum(raw), atol=1e-14, rtol=1e-12
)
def test_infeasible_start_rejected_without_filtering_initial_distribution():
P = np.array([[1.0, 0.0, 0.0], [0.0, 0.0, 1.0], [0.0, 0.0, 1.0]])
kernels = bridge.doob_kernels(P, [2], 2)
assert not kernels[0][0].any()
with pytest.raises(bridge.InfeasibleBridgeError, match="h_0=0"):
bridge.sample_bridge_paths(kernels, [0, 1], 20)
with pytest.raises(_conditioning.InfeasibleBridgeError):
_conditioning.build(P, np.ones((2, 3)), [2], 2, starts=[0])
def test_exact_horizon_feasibility_not_just_reachability_by_horizon():
P = np.array([[0.0, 1.0], [1.0, 0.0]])
with pytest.raises(bridge.InfeasibleBridgeError):
bridge.sample_bridge_paths(bridge.doob_kernels(P, [1], 2), [0])
paths = bridge.sample_bridge_paths(bridge.doob_kernels(P, [1], 3), [0], 3)
np.testing.assert_array_equal(paths, [[0, 1, 0, 1]] * 3)
def test_terminal_and_zero_weight_support_remain_exactly_zero():
P = np.array(
[
[0.0, 0.5, 0.5, 0.0],
[0.0, 0.0, 0.0, 1.0],
[0.0, 0.0, 0.0, 1.0],
[0.0, 0.0, 0.0, 1.0],
]
)
weights = np.array([1.0, 0.0, 1.0, 1.0])
kernels, h = _conditioning.build(P, [weights] * 2, [3], 2, starts=[0])
assert np.array_equal(h[-1], [0.0, 0.0, 0.0, 1.0])
for matrix in kernels:
assert not matrix[:, 1].any()
assert not matrix[P == 0].any()
assert not kernels[-1][:, :3].any()
paths = bridge.sample_bridge_paths(kernels, [0], 10)
np.testing.assert_array_equal(paths, [[0, 2, 3]] * 10)
def test_zero_weights_can_make_every_start_infeasible():
P = np.ones((2, 2)) / 2
kernels = bridge.doob_kernels(P, [1], 4, w=np.zeros(2))
assert all(not k.any() for k in kernels)
with pytest.raises(bridge.InfeasibleBridgeError):
bridge.sample_bridge_paths(kernels, [0])
def test_small_positive_products_do_not_trigger_fallback_or_underflow():
P = np.array([[0.7, 0.3], [0.4, 0.6]])
w = np.full(2, 1e-250)
h = np.array([2e-250, 1e-250])
expected = P * np.array([2.0, 1.0])[None, :]
expected /= expected.sum(axis=1, keepdims=True)
np.testing.assert_allclose(_conditioning.kernel(P, w, h), expected, rtol=1e-12)
tiny = bridge.doob_kernels(P, [1], 1000, w=w)
unit = bridge.doob_kernels(P, [1], 1000)
np.testing.assert_allclose(tiny, unit, atol=1e-12, rtol=1e-12)
def test_backwards_and_kernels_agree_across_public_interfaces():
P = np.array([[0.8, 0.2], [0.3, 0.7]])
weights = np.array([[1.0, 2.0], [0.2, 1.0], [3.0, 0.4]])
terminal = np.array([0.0, 1.0])
scaled = _conditioning.backward(P, weights, 3, terminal)
unscaled = _conditioning.backward(P, weights, 3, terminal, rescale=False)
np.testing.assert_allclose(scaled, unscaled / unscaled.sum(axis=1, keepdims=True))
reference = bridge.doob_kernels(P, [1], 3, weights)
rebuilt = [_conditioning.kernel(P, weights[n], scaled[n + 1]) for n in range(3)]
np.testing.assert_allclose(rebuilt, reference)
kernels, backward = _conditioning.build(P, weights, [1], 3)
np.testing.assert_allclose(kernels, reference)
np.testing.assert_allclose(backward, scaled)
def test_early_target_departures_and_residence_are_still_allowed():
P = np.array([[0.8, 0.2], [0.5, 0.5]])
kernels = bridge.doob_kernels(P, [1], 4)
assert kernels[0][0, 0] > 0
assert kernels[1][1, 0] > 0
paths = bridge.sample_bridge_paths(kernels, [0], 100, seed=7)
assert all(len(p) == 5 and p[-1] == 1 for p in paths)
@pytest.mark.parametrize(
"targets,horizon,weights",
[
([], 2, None),
([2], 2, None),
([1], 0, None),
([1], 2, [1.0, -1.0]),
([1], 2, [1.0, np.nan]),
],
)
def test_invalid_inputs_are_not_silently_repaired(targets, horizon, weights):
with pytest.raises(ValueError):
bridge.doob_kernels(np.ones((2, 2)) / 2, targets, horizon, weights)
def test_invalid_start_indices_are_not_replaced_with_arbitrary_states():
with pytest.raises(ValueError, match="out-of-range"):
_conditioning.build(np.eye(2), np.ones((2, 2)), [1], 2, starts=[8])
|