Streaming Polyphase Factor Decomposition / SKAT

Download original file

This is the original source text. Reading it does not run or verify the program.

Read the typeset paper

1 import unittest
2 import numpy as np
3 
4 from streaming_skat import (
5     orthonormalize_covariates,
6     full_factor,
7     stream_sample_gram,
8     stream_kernel_action,
9     stream_A_times_V,
10     stream_AT_times_U,
11     stream_dual_action,
12     tiled_sample_gram,
13 )
14 
15 
16 class StreamingSKATTests(unittest.TestCase):
17     @classmethod
18     def setUpClass(cls):
19         rng=np.random.default_rng(123)
20         cls.n=96; cls.m=311; cls.s=5
21         maf=rng.uniform(.005,.25,cls.m)
22         cls.G=rng.binomial(2,maf[None,:],size=(cls.n,cls.m)).astype(float)
23         cls.w=(1-maf)**24
24         X=np.column_stack([np.ones(cls.n),rng.normal(size=cls.n),rng.normal(size=cls.n)])
25         cls.Q=orthonormalize_covariates(X)
26         cls.A=full_factor(cls.G,cls.w,cls.Q)
27         cls.K=cls.A@cls.A.T
28         cls.rng=rng
29 
30     def test_stream_gram(self):
31         K2=stream_sample_gram(self.G,self.w,self.Q,variant_block=37)
32         self.assertLess(np.linalg.norm(K2-self.K)/np.linalg.norm(self.K),1e-12)
33 
34     def test_kernel_action(self):
35         V=self.rng.normal(size=(self.n,self.s))
36         got=stream_kernel_action(self.G,self.w,self.Q,V,variant_block=41)
37         ref=self.K@V
38         self.assertLess(np.linalg.norm(got-ref)/np.linalg.norm(ref),1e-12)
39 
40     def test_A_action(self):
41         V=self.rng.normal(size=(self.m,self.s))
42         got=stream_A_times_V(self.G,self.w,self.Q,V,variant_block=43)
43         ref=self.A@V
44         self.assertLess(np.linalg.norm(got-ref)/np.linalg.norm(ref),1e-12)
45 
46     def test_AT_action(self):
47         U=self.rng.normal(size=(self.n,self.s))
48         got=stream_AT_times_U(self.G,self.w,self.Q,U,variant_block=47)
49         ref=self.A.T@U
50         self.assertLess(np.linalg.norm(got-ref)/np.linalg.norm(ref),1e-12)
51 
52     def test_dual_action(self):
53         V=self.rng.normal(size=(self.m,self.s))
54         got=stream_dual_action(self.G,self.w,self.Q,V,variant_block=53)
55         ref=self.A.T@(self.A@V)
56         self.assertLess(np.linalg.norm(got-ref)/np.linalg.norm(ref),1e-12)
57 
58     def test_tiled_gram(self):
59         got=tiled_sample_gram(
60             self.G,self.w,self.Q,
61             variant_block=61,
62             sample_tile=23,
63         )
64         self.assertLess(np.linalg.norm(got-self.K)/np.linalg.norm(self.K),1e-12)
65 
66 
67 if __name__=="__main__":
68     unittest.main(verbosity=2)

Back to source controls

Back to Streaming Polyphase Factor Decomposition / SKAT