Spaces:
Running
Running
File size: 3,321 Bytes
c9f4f90 523fdd5 c9f4f90 523fdd5 c9f4f90 523fdd5 c9f4f90 523fdd5 c9f4f90 523fdd5 c9f4f90 523fdd5 c9f4f90 523fdd5 c9f4f90 | 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 | import math
import sys
import unittest
from pathlib import Path
import numpy as np
try:
from repro.src.markov_core import (
ThetaFamily,
boundary,
parametric_kernel,
poisson_bound,
poisson_solution,
pseudo_spectral_gap,
simulate_parallel_thresholds,
simulate_test,
simulate_thresholds,
stationary,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parent))
from markov_core import (
ThetaFamily,
boundary,
parametric_kernel,
poisson_bound,
poisson_solution,
pseudo_spectral_gap,
simulate_parallel_thresholds,
simulate_test,
simulate_thresholds,
stationary,
)
class TestMarkovCore(unittest.TestCase):
def test_two_state_pseudo_gap_and_poisson_bound(self):
kernel = np.array([[0.8, 0.2], [0.3, 0.7]])
gap, best_k, _ = pseudo_spectral_gap(kernel)
self.assertAlmostEqual(gap, 1 - (1 - 0.2 - 0.3) ** 2, places=10)
self.assertEqual(best_k, 1)
constant, actual, _, _ = poisson_bound(kernel)
self.assertLessEqual(actual, constant)
function = np.array([1.0, -0.5])
omega = poisson_solution(kernel, function)
pi = stationary(kernel)
residual = (np.eye(2) - kernel) @ omega - (function - pi @ function)
self.assertLess(np.max(np.abs(residual)), 1e-10)
def test_algorithm_boundary(self):
visits = np.array([3, 0, 2, 1, 0])
psi = np.log(math.e * (1 + visits / 4)).sum()
self.assertAlmostEqual(boundary(visits, math.log(20)), math.log(20) + 4 * psi)
def test_shared_path_thresholds_match_separate_runs(self):
base = np.array(
[[0.7, 0.2, 0.1], [0.1, 0.7, 0.2], [0.2, 0.1, 0.7]]
)
feature = np.array([1.0, 0.0, -1.0])
family = ThetaFamily(base, feature, 0.4, 0.8, 101)
kernel = parametric_kernel(base, feature, -0.6)
levels = (10.0, 20.0)
shared = simulate_thresholds(kernel, family, levels, 1234, 5000)
for level in levels:
separate = simulate_test(kernel, family, level, 1234, 5000)
self.assertEqual(shared[level], separate["stopping_time"])
def test_parallel_thresholds_match_both_separate_runs(self):
base = np.array(
[[0.7, 0.2, 0.1], [0.1, 0.7, 0.2], [0.2, 0.1, 0.7]]
)
feature = np.array([1.0, 0.0, -1.0])
p_family = ThetaFamily(base, feature, 0.4, 0.8, 101)
q_family = ThetaFamily(base, feature, -0.8, -0.4, 101)
kernel = parametric_kernel(base, feature, -0.6)
level = 10.0
shared = simulate_parallel_thresholds(
kernel,
p_family,
(level,),
q_family,
(level,),
4321,
5000,
"p",
)
p_run = simulate_test(kernel, p_family, level, 4321, 5000)
q_run = simulate_test(kernel, q_family, level, 4321, 5000)
self.assertEqual(shared["p"][level], p_run["stopping_time"])
if q_run["stopping_time"] is not None and q_run["stopping_time"] <= p_run["stopping_time"]:
self.assertEqual(shared["q"][level], q_run["stopping_time"])
if __name__ == "__main__":
unittest.main()
|