s2gpp 1.0.2

Algorithm for Highly Efficient Detection of Correlation Anomalies in Multivariate Time Series
Documentation 
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

use crate::training::node_estimation::multi_kde::actors::messages::MultiKDEMessage;
use crate::training::node_estimation::multi_kde::actors::MultiKDEActor;
use actix::{Actor, Context, Handler};
use meanshift_rs::ClusteringResponse;
use ndarray::{arr2, Array1, Array2, Axis};
use ndarray_stats::CorrelationExt;
use std::ops::{Deref, DerefMut};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::sleep;

struct TestReceiver {
    labels: Arc<Mutex<Vec<usize>>>,
    coords: Option<Arc<Mutex<Option<Array2<f32>>>>>,
}

impl Actor for TestReceiver {
    type Context = Context<Self>;
}

impl Handler<ClusteringResponse<f32>> for TestReceiver {
    type Result = ();

    fn handle(&mut self, msg: ClusteringResponse<f32>, _ctx: &mut Self::Context) -> Self::Result {
        *(self.labels.lock().unwrap().deref_mut()) = msg.labels;
        *(self.coords.as_mut().unwrap().lock().unwrap().deref_mut()) = Some(msg.cluster_centers);
    }
}

fn setup_data(size: usize) -> (Array2<f32>, Vec<usize>) {
    let mut points = vec![1.0; size];
    points.extend(vec![2.0; size]);
    let data = Array1::from(points).insert_axis(Axis(1));
    let mut expected = vec![0; size];
    expected.extend(vec![1; size]);
    (data, expected)
}

#[actix_rt::test]
async fn correct_result() {
    let (data, expected) = setup_data(50);

    let labels = Arc::new(Mutex::new(vec![]));
    let receiver = (TestReceiver {
        labels: labels.clone(),
        coords: None,
    })
    .start();
    let mkde = MultiKDEActor::new(receiver.recipient(), 1).start();
    mkde.do_send(MultiKDEMessage { data });
    sleep(Duration::from_millis(2000)).await;
    assert_eq!(labels.lock().unwrap().deref().clone(), expected);
}

#[actix_rt::test]
async fn correct_result_multithreading() {
    let (data, expected) = setup_data(50);

    let labels = Arc::new(Mutex::new(vec![]));
    let receiver = (TestReceiver {
        labels: labels.clone(),
        coords: None,
    })
    .start();
    let mkde = MultiKDEActor::new(receiver.recipient(), 4).start();
    mkde.do_send(MultiKDEMessage { data });
    sleep(Duration::from_millis(1000)).await;
    assert_eq!(labels.lock().unwrap().deref().clone(), expected);
}

#[ignore]
#[actix_rt::test]
async fn test_cov() {
    let x = arr2(&[
        [-1.1044426],
        [3.3079557],
        [-2.6531453],
        [-0.7709985],
        [-1.1930218],
        [11.356522],
        [-1.1818404],
        [-2.8623846],
        [-1.6397605],
        [-1.9247131],
        [-0.33105826],
        [-0.57120085],
        [6.5464177],
        [0.91874397],
        [-3.3256016],
        [1.6511793],
        [-4.301383],
        [0.3333578],
        [-14.434255],
        [-1.6871758],
        [2.098528],
        [3.397635],
        [-1.4510679],
        [0.039141178],
        [-0.9916377],
        [1.8591633],
        [3.3673396],
        [-1.5271537],
        [-2.0864549],
        [0.05202365],
        [-6.286045],
        [6.923059],
        [-5.992464],
        [2.4069607],
        [-0.14090195],
        [-5.526738],
        [1.3183851],
        [-0.04570675],
        [-1.3094206],
        [1.5622854],
        [2.3346844],
        [-0.519022],
        [-0.01082468],
        [2.4892955],
        [1.676794],
        [-3.2251143],
        [-4.219629],
        [-1.8621855],
        [-2.6605353],
    ]);
    println!("{:?}", x.t().cov(1.).unwrap());
}