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

use crate::messages::PoisonPill;
use crate::training::node_estimation::multi_kde::actors::gaussian_kde::GaussianKDEActor;
use crate::training::node_estimation::multi_kde::actors::messages::{
    GaussianKDEMessage, GaussianKDEResponse, MultiKDEMessage,
};
use crate::training::node_estimation::multi_kde::MultiKDEBase;
use crate::utils::pop_clear::PopClear;
use crate::utils::stack::Stack;
use actix::{Actor, ActorContext, Addr, AsyncContext, Context, Handler, Recipient};
use meanshift_rs::ClusteringResponse;
use ndarray::{s, ArcArray2, Array1, Array2, Axis};
use ndarray_stats::QuantileExt;

mod gaussian_kde;
mod helper;
pub(crate) mod messages;
#[cfg(test)]
mod tests;

pub(in crate::training::node_estimation) struct MultiKDEActor {
    receiver: Recipient<ClusteringResponse<f32>>,
    n_threads: usize,
    multi_kde_base: MultiKDEBase,
    data: Option<Array2<f32>>,
    next_dim: usize,
    current_column: Option<ArcArray2<f32>>,
    cluster_centers: Vec<Array1<f32>>,
    gaussian_kde: Option<Addr<GaussianKDEActor>>,
}

impl MultiKDEActor {
    pub fn new(receiver: Recipient<ClusteringResponse<f32>>, n_threads: usize) -> Self {
        Self {
            receiver,
            n_threads,
            multi_kde_base: MultiKDEBase::default(),
            data: None,
            next_dim: 0,
            current_column: None,
            cluster_centers: vec![],
            gaussian_kde: None,
        }
    }

    fn cluster_next_dim(&mut self, ctx: &mut Context<Self>) {
        let data = self.data.as_ref().expect("Data is not yet received");
        let gkde = self.gaussian_kde.as_ref().unwrap();
        if self.next_dim < data.shape()[1] {
            let column = data
                .slice(s![.., self.next_dim..self.next_dim + 1])
                .to_shared();
            gkde.do_send(GaussianKDEMessage {
                column: column.clone(),
            });
            self.current_column = Some(column);
            self.next_dim += 1;
        } else {
            let cluster_centers_vec = self.cluster_centers.pop_clear();
            let cluster_centers = cluster_centers_vec.stack(Axis(1)).unwrap();
            let (labels, cluster_centers) = self
                .multi_kde_base
                .extract_labels_from_centers(cluster_centers);
            self.receiver
                .do_send(ClusteringResponse {
                    cluster_centers,
                    labels,
                })
                .unwrap();
            self.gaussian_kde.as_ref().unwrap().do_send(PoisonPill);
            ctx.stop();
        }
    }

    fn find_cluster_centers(&mut self, kernel_estimate: Array1<f32>, ctx: &mut Context<Self>) {
        let column = self
            .current_column
            .as_ref()
            .expect("You are not working on any column at the moment");
        let grid_min = *column.min().unwrap();
        let grid_max = *column.max().unwrap();
        let peaks =
            self.multi_kde_base
                .find_peak_values(kernel_estimate.view(), grid_min, grid_max);
        let assigned_peak_values = if !peaks.is_empty() {
            self.multi_kde_base
                .assign_closest_peak_values(column.view(), peaks)
        } else {
            Array1::from(vec![0.0; column.len()])
        };
        self.cluster_centers.push(assigned_peak_values);

        self.cluster_next_dim(ctx);
    }
}

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

impl Handler<MultiKDEMessage> for MultiKDEActor {
    type Result = ();

    fn handle(&mut self, msg: MultiKDEMessage, ctx: &mut Self::Context) -> Self::Result {
        self.data = Some(msg.data);
        self.gaussian_kde =
            Some(GaussianKDEActor::new(self.n_threads, 250, ctx.address().recipient()).start());
        self.cluster_next_dim(ctx);
    }
}

impl Handler<GaussianKDEResponse> for MultiKDEActor {
    type Result = ();

    fn handle(&mut self, msg: GaussianKDEResponse, ctx: &mut Self::Context) -> Self::Result {
        self.find_cluster_centers(msg.kernel_estimate, ctx);
    }
}

impl Handler<PoisonPill> for MultiKDEActor {
    type Result = ();

    fn handle(&mut self, _msg: PoisonPill, ctx: &mut Self::Context) -> Self::Result {
        ctx.stop();
    }
}