Fenêtre glissante de 24h dans une requête

Bonjour,

je suis dans le cadre de la modélisation de la réaction à un flux de spam.

Je suis en capacité de récupérer les plaintes des gens en dégageant :

• le sujet du spam
• la date de réception du spam
• la date de plainte de la part du client.

J'essaie de détecter des "vagues d'envoie" pour en déterminer la prédictibilité.

Actuellement, pour détecter une "vague" je regarde les plaintes jour par jour et j'affiche un histogramme.

Le problème, c'est qu'actuellement une partie non négligeable des vagues sont envoyés entre 22h et 4h du matin ce qui a tendance à couper la vague en deux.

Existe-t-il un moyen en SQL qui permettrait de faire une fenêtre glissante directement?

Sinon, je peux toujours jouer avec python, reccueillir l'ensemble des données et faire passer une fenêtre, mais du coup ça a tendance à être très long car non seulement il faut attendre que toutes les données soient rapatriées de SQL (on est sur des ensemble de plusieurs millions de lignes) mais en plus il faut le temps de faire notre fenêtre, détecter qu'elle englobe bien une vague et afficher la vague.

Bonsoir,

Existe-t-il un moyen en SQL qui permettrait de faire une fenêtre glissante directement?

j'ai fait un test en générant 100000 signalements sur toute l'année 2015

1
2
3
4

SELECT date_reception, HOUR(date_reception), count(*) as nb
FROM signalements
WHERE date_reception < NOW() AND date_reception > DATE_SUB(NOW(), INTERVAL 24 HOUR)
GROUP BY DAY(date_reception) DESC, HOUR(date_reception) DESC

 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

date_reception  HOUR(date_reception)    nb
2015-04-08 19:37:37     19  4
2015-04-08 18:01:38     18  6
2015-04-08 17:29:39     17  3
2015-04-08 16:57:39     16  6
2015-04-08 15:53:40     15  5
2015-04-08 14:49:41     14  4
2015-04-08 13:13:42     13  5
2015-04-08 12:41:43     12  4
2015-04-08 11:05:44     11  5
2015-04-08 10:33:45     10  4
2015-04-08 09:29:46     9   3
2015-04-08 08:25:47     8   6
2015-04-08 07:21:48     7   6
2015-04-08 06:49:48     6   6
2015-04-08 05:45:49     5   5
2015-04-08 04:09:50     4   6
2015-04-08 03:37:51     3   4
2015-04-08 02:01:52     2   5
2015-04-08 01:29:53     1   3
2015-04-08 00:57:53     0   6
2015-04-07 23:21:55     23  5
2015-04-07 22:49:55     22  5
2015-04-07 21:45:56     21  5
2015-04-07 20:09:57     20  6

C'est dans la clause WHERE que je spécifie la plage de date, sur laquelle s'effectue le déplacement de la fenètre. Dans l'exemple, c'est sur les dernières 24H. Libre à toi de spécifier une plage de DateTime à ta convenance.

En sortie, on obtient le nombre de signalement groupé par jour, et par heures.

Le principe de la fenêtre flottante serait plutôt d'avoir le cumul, mais là j'avoue que je ne vois pas comment faire coté SQL.

edit: Finalement avec le cumul, à l'aide d'une variable

1
2
3
4
5
6
7

set @rt := 0;
SELECT date_reception, HOUR(date_reception) as heure, nb,  @rt := @rt + nb as total
    FROM
        (SELECT date_reception, count(*) as nb
            FROM signalements
            WHERE date_reception < NOW() AND date_reception > DATE_SUB(NOW(), INTERVAL 24 HOUR)
            GROUP BY DAY(date_reception), HOUR(date_reception)) tmp

Il me semble que ça correspond à ta demande. Seulement, je me demande tout de même si il n'y a pas beaucoup plus propre(et simple) pour réaliser la même tâche.

edit2: suppression d'un ORDER BY inutile

Merci beaucoup pour ton aide.

je mettrai demain ce que moi je fais pour que ça soit plus clair, mais c'est pas tout à fait ce que je demande. En effet, j'ai les spam de toute une semaine et je voudrais détecter une "vague" qui aurait eu lieu n'importe quand. On considère qu'une "vague" ne peut pas durer plus de 24h (sinon on n'est plus dans le cadre d'une vague) donc le but est de trouver une fenêtre dans laquelle il y a des spams et les y compter. Il faut donc que la fenêtre soit "mobile" sur toutes les données, il ne s'agit pas de faire une comparaison avec aujourd'hui mais de regarder ce qu'il s'est passé dans les 24h autour de chaque spam.

si je ne peux pas le faire en SQL, je le ferai côté client, mais comme SQL est toujours plus performant que mes scripts python, je veux juste m'assurer que je ne me trompe pas.

il ne s'agit pas de faire une comparaison avec aujourd'hui mais de regarder ce qu'il s'est passé dans les 24h autour de chaque spam.

C'est bien ce que j'avais compris. Rien ne t'empêche de modifier le NOW(), par la date de ton choix en passant la date dans ton script python lors de la requête, non ? Ou alors, il y a un truc qui m'échappe^^

Avant de faire ma "fenêtre glissante", je crée une table avec des éléments précalculés qui me servent à faire des statistiques plutôt intéressantes dans d'autres cas.

 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

CREATE TABLE vague_elements( 
    id INT UNSIGNED NOT NULL AUTO_INCREMENT,
    int_5_min INT NOT NULL,
    int_5_min_rec INT NOT NULL,
    reception_date int NOT NULL,
    complaint_date int NOT NULL,
    jour INT(2) NOT NULL,
    heure INT(2) NOT NULL,
    subject CHAR(255) NOT NULL,
    PRIMARY KEY(id)
)Engine=MyISAM;

INSERT INTO 
    vague_elements
        (int_5_min,
         int_5_min_rec,
         reception_date,
         complaint_date,
         jour,
         heure,
         subject)
SELECT 
        FLOOR(COMPLAINT_DATE/60/5) AS int_5_min,
        FLOOR(RECEPTION_DATE/60/5) AS int_5_min_rec,
        RECEPTION_DATE AS reception_date,
        COMPLAINT_DATE AS complaint_date,
        DAYOFMONTH(FROM_UNIXTIME(COMPLAINT_DATE)) AS jour,
        HOUR(FROM_UNIXTIME(COMPLAINT_DATE)) AS heure,
        SUBJECT_HEADER AS subject,
    FROM COMPLAINTS
    WHERE
        COMPLAINT_DATE >= RECEPTION_DATE
        AND IP NOT IN (SELECT * FROM BANNED_IP)
;

CREATE INDEX vague_subject ON vague_elements(subject(20));

Grâce à cela, j'ai un accès amélioré aux données, surtout si je dois sélectionner un sujet en particulier. int_5_min et int_5_min_rec représente l'intervalle de 5minutes durant lequel le spam a été reçu. C'est utile dans une autre modélisation que je fais.

J'ai une "presque fenêtre glissante" avec cette requête :

1
2
3
4
5
6
7

SELECT 
    subject,
    MAX(int_5_min_rec) AS max_time,
    MIN(int_5_min_rec) AS min_time,
    COUNT(*) AS nb FROM vague_elements
GROUP BY subject
HAVING max_time - min_time < 24*12 AND nb > 200

Le problème de cette requête, c'est qu'elle est longue (environ 6 minutes).

Grâce à la lib python que tu m'as conseillé, je pense pouvoir mettre en place une requête plus simple :

1
2
3
4
5
6

SELECT 
    subject,
    COUNT(*) AS nb
FROM vague_elements WHERE subject != ""
GROUP BY subject
HAVING nb > 500 LIMIT 5

Elle me permet de sélectionner 5 vagues apparentes puis au sein de ces vagues, je fais la fenêtre de 24h grâce à botleneck. Ce qui aura, je pense, pour effet de générer les bons résultats. Seul problème ; la requête prend quand même 3 minutes à s'exécuter.

Faute de précision, je suppose que c'est MySQL que tu emploies. Il manque aussi une définition plus claire d'une "vague", d'après ton dernier message je comprend qu'un spam est identifié par son sujet et que la vague est mesurée sur l'heure de réception du spam.

Je ne comprend pas bien les deux dernières requêtes :

• ta "presque fenêtre glissante" est ennuyeuse parce que trop stricte : si un spam est arrivé une seule fois en dehors de la fenêtre, tu ne le vois plus ;
• la dernière requête, je ne vois pas trop quoi faire avec, c'est juste le nombre de spam reçu par sujet, non ?

Ce que j'avais en tête, c'est de charger la table vague_element dans un tableau numpy (au moins les champs subject et int_5_min_rec) et de faire les stats dessus.

On peut avoir un bout de table vague_element pour jouer avec ?

ta "presque fenêtre glissante" est ennuyeuse parce que trop stricte : si un spam est arrivé une seule fois en dehors de la fenêtre, tu ne le vois plus ;

Mon but est de détecter une "vague" et on considère qu'une vague ne peut pas durer plus de 24h d'affilé (en fait c'est souvent de 15min à 6-7h dans 90% des cas mais on a décidé de prendre 24h). Ainsi on pourrait dire qu'une vague c'est un envoie massif de mail durant une période de 24h d'affilé maximum. On identifie les vagues de trois manière différentes :

• par le sujet de la vague
• par l'adresse mail From
• par l'ip expéditeur

Chaque problématique a ses propres soucis, mais je fais une chose à la fois.

Voici 30 lignes de ma table :

Mon but est de détecter une "vague" et on considère qu'une vague ne peut pas durer plus de 24h d'affilé (en fait c'est souvent de 15min à 6-7h dans 90% des cas mais on a décidé de prendre 24h). Ainsi on pourrait dire qu'une vague c'est un envoie massif de mail durant une période de 24h d'affilé maximum.

Ma question était : que se passe t'il si tu as une vague bien reconnaissable sur 6h d'affilée lundi, puis 2/3 message supplémentaires isolés portant le même sujet mercredi. Avec ta requête, tu ne vois plus la vague de lundi, est-ce bien ce que tu veux ?

En tout cas, ce que tu peux faire s'il y a trop de données pour les envoyer directement à python, c'est une requête SQL qui compresse tes données par tranches de 5 minutes (ou éventuellement plus si nécessaire), en faisant un GROUP BY sur subject (ou autre) et int_5_min_rec, qui retourne le nombre de rapports pour chaque sujet par tranche d'heure. Ainsi tu as moins de données à faire transiter vers python, et une partie du calcul de fenêtre est déjà réalisé. (ça se trouve c'est ce que tu fais déjà, mais je n'ai pas compris ta dernière requête SQL)

Bref, ça devrait être possible de trouver un compromis entre python et ton serveur SQL, mais évidement plus il y aura de lignes plus ce sera lent de toute façon.

Ma question était : que se passe t'il si tu as une vague bien reconnaissable sur 6h d'affilée lundi, puis 2/3 message supplémentaires isolés portant le même sujet mercredi. Avec ta requête, tu ne vois plus la vague de lundi, est-ce bien ce que tu veux ?

Au mieux il faudrait que cela soit détecté comme deux vagues. Au pire ne détecter que la seconde vague peut suffire.

J'utilise désormais les bibliothèques données au dessus et numpy pour découvrir les vagues.

Désolé, le script n'est pas commenté, mais l'idée est là :

 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

def get_vague_with_short_window(cursor, fig, fbl):
    fig.clear()
    cursor.execute("SELECT COUNT(*) FROM vague_per_subject")
    c = cursor.fetchall()
    print c
    subjects = []
    while len(subjects) < 5:
        rands = [str(random.randint(1, c[0][0])) for _ in range(20)]
        query = "SELECT id, subject FROM vague_per_subject WHERE id IN(" + ','.join(rands) +") AND max_rec_time - min_rec_time > 0 AND max_rec_time - min_rec_time <= 3600 * 25 ORDER BY max_rec_time - min_rec_time"
        cursor.execute(query)
        subjects += [_[1] for _ in cursor.fetchall()]

    query = "SELECT id, reception_date, int_5_min_rec, FROM_UNIXTIME(reception_date), complaint_date FROM vague_elements WHERE subject = %s ORDER BY int_5_min_rec "
    index = 1
    for subject in subjects[:5]:
        plt.hold(False)
        plt.subplot(3,2,index)
        print query % subject
        cursor.execute(query, [subject])

        alls =  cursor.fetchall()
        int_5_min_recs = [i[2] for i in alls]
        int_5_arr = np.unique(np.array(int_5_min_recs))

        print len(int_5_arr), max(int_5_arr) - min(int_5_arr)
        int_max = bn.move_max(int_5_arr, window=min([len(int_5_arr), 12*24]), min_count=1)
        int_min = bn.move_min(int_5_arr, window=min([len(int_5_arr), 12*24]), min_count=1)
        diffs = [int_max[i] - int_min[i] for i in range(len(int_min))]
        for i in range(len(diffs)):
            if diffs[i] > 12*24:
                diffs[i] = -1
        vague_diff = max(diffs)


        min_date = min([i[1] for i in alls])
        index_max = np.argmax(diffs)
        if vague_diff > 100:
            diffs[index_max] = -1
            if vague_diff - max(diffs) > 3*12 and max(diffs) > 5: # if there is a gap of more than 3h
                vague_diff = max(diffs)
                index_max = np.argmax(diffs)
            else:
                diffs[index_max] = vague_diff
        print int_min[index_max], int_max[index_max], diffs[index_max]
        dates = [float((i[1]-min_date)/3600.0) for i in alls if int_min[index_max] <= i[2] <= int_max[index_max]]
        dates_c = [float((i[4]-min_date)/3600.0) for i in alls if int_min[index_max] <= i[2] <= int_max[index_max]]
        min_date_c, max_date_c = min(dates_c), max(dates_c)
        complaints_bins = int((max_date_c - min_date_c) * 12)
        dates_label = [i[3] for i in alls if int_min[index_max] <= i[2] <= int_max[index_max]]
        n, bins, patches = plt.hist(dates, int(diffs[index_max]), label=u"réception", color="blue")

        plt.hold(True)
        plt.hist(dates_c, complaints_bins, alpha=0.2, label="plaintes", color="green")
        plt.legend()
        #~ plt.hold(True)
        #~ print len(bn.move_mean(np.array(n), window=5, min_count=1)), len(bins)
        #~ plt.plot(bins[:-1], bn.move_mean(np.array(n), window=5, min_count=1) , label = "Moyenne mobile 25 minutes")
        plt.title(u"Vague de sujet {} \n par intervalle de 5 minutes".format(subject[:20]))

        plt.xlabel(u"Temps par rapport au début de la vague (h) FBL {}".format(fbl))
        plt.ylabel(u"Nombre de mails reçus")

        index += 1
    print "saving "+"graph_avg/essai_vague_sub3.png"
    fig.set_size_inches(20, 30)
    fig.savefig("graph_avg/essai_vague_sub_with_comp_{}_1.png".format(fbl))