02Data Lab

The Lab

A selection of technical implementations and applied engineering.

24/24snippets

ES|QL

FROM "index-prod-*"
| WHERE event.action IN ("account_created", "account_deleted")
| EVAL user_id = user.identifier.keyword 
| SORT @timestamp ASC
| STATS

Identifier les anomalies de cycle de vie (Création vs Suppression)

Utilisation de MAX(CASE(...)) pour extraire et comparer chronologiquement les états d'un compte utilisateur.

ELKAnomalieAudit

MQL5

void TrailingStopATR(int ticket, int atrPeriod, double multiplier)
{
   double atrBuffer[];
   int atrHandle = iATR(_Symbol, PERIOD_CURRENT, atrPeriod);
   CopyBuffer(atrHandle, 0, 0, 1, atrBuffer);

Trailing Stop Dynamique basé sur ATR

Implémentation d'un trailing stop auto-adaptatif qui ajuste la distance de protection selon la volatilité du marché mesurée par l'indicateur ATR.

TradingExpert AdvisorATR

Machine Learning

import numpy as np
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType
import onnxruntime as rt

Inférence ONNX depuis un modèle Scikit-learn

Pipeline complète : entraînement d'un RandomForest en Scikit-learn, export au format ONNX, puis chargement pour inférence.

PythonONNXScikit-learn

Python

import pandas as pd
import numpy as np

def compute_features(df: pd.DataFrame) -> pd.DataFrame:
    df["log_return"] = np.log(df["close"] / df["close"].shift(1))

Feature Engineering : Signaux Techniques pour ML

Calcul automatisé de features techniques (RSI, EMA, Volatilité) à partir de données OHLCV.

PandasFeature EngineeringRSI

Spark

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

val windowSpec = Window
  .partitionBy("user_id")

Déduplication par fenêtre glissante (Spark SQL)

Utilisation des Window Functions en Spark/Scala pour dédoublonner des enregistrements.

ScalaBig DataWindow Functions

Vega

{
  "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
  "description": "En production Kibana : remplacer data.values par data.url { index: logs-*, body: { aggs terms log.level } }",
  "data": { "values": [
    { "key": "info", "doc_count": 48210 },

Donut Chart : Répartition des niveaux de logs

Visualisation Vega-Lite pour Kibana affichant la proportion des niveaux de logs avec une esthétique néon.

Vega-LiteKibanaMonitoring

SQL

SELECT
    customer_id,
    COUNT(*)             AS orders,
    SUM(amount)          AS total_spent,
    ROUND(AVG(amount),2) AS avg_order

Agrégation filtrée : GROUP BY + HAVING

Regrouper, agréger (COUNT/SUM/AVG) puis filtrer sur le résultat agrégé avec HAVING — ce que WHERE ne peut pas faire.

SQLGROUP BYHAVING

ES|QL

FROM "logs-transactions-*"
| WHERE event.action IN ("process_start", "process_end")
| STATS start_time = MAX(@timestamp) WHERE event.action == "process_start",
        end_time = MIN(@timestamp) WHERE event.action == "process_end"
  BY process_id = COALESCE(req.videoId.keyword, resp.videoId.keyword)

Corrélation d'événements et calcul de durée (Timeout)

Utilisation de COALESCE pour joindre des logs asynchrones et DATE_DIFF pour identifier les processus dépassant un seuil de temps (ex: 20 mins).

ELKCorrélationDATE_DIFF

MQL5

bool IsWithinSession(string session)
{
   MqlDateTime dt;
   TimeCurrent(dt);
   int hour = dt.hour;

Filtre de Session de Trading (London / NY)

Fonction utilitaire pour restreindre l'exécution d'un EA aux heures d'ouverture d'une session de marché spécifique.

TradingSessionFiltre

Python

import time
from functools import wraps

def retry(max_attempts=3, delay=1):
    def decorator(func):

Décorateur de Retry robuste pour appels API

Gestion des erreurs transitoires avec backoff exponentiel pour les requêtes HTTP.

APIRequestsDecorators

Spark

df.repartition(col("year"), col("month"))
  .write
  .partitionBy("year", "month")
  .mode("overwrite")
  .parquet("s3a://bucket/data/")

Optimisation de l'écriture Parquet partitionnée

Optimisation du stockage pour accélérer les requêtes via partitionnement intelligent et repartitioning.

ScalaParquetPerformance

Vega

{
  "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
  "data": { "values": [
    { "day": "Lun", "hour": 8, "usage": 42 }, { "day": "Lun", "hour": 10, "usage": 78 }, { "day": "Lun", "hour": 14, "usage": 91 }, { "day": "Lun", "hour": 18, "usage": 55 },
    { "day": "Mar", "hour": 8, "usage": 47 }, { "day": "Mar", "hour": 10, "usage": 82 }, { "day": "Mar", "hour": 14, "usage": 96 }, { "day": "Mar", "hour": 18, "usage": 61 },

Heatmap de trafic horaire interactif

Génère une carte thermique pour identifier les pics d'utilisation par jour et par heure.

VegaKibanaAnalytics

SQL

SELECT
    sale_date,
    category,
    amount,
    SUM(amount) OVER (PARTITION BY category ORDER BY sale_date

Fonctions de fenêtre : cumul & classement

Total cumulé, rang et valeur précédente sans collapser les lignes, via les window functions (OVER ... PARTITION BY ... ORDER BY).

SQLWindow FunctionsCumul

ES|QL

FROM "logs-system-*"
| WHERE event.category == "authentication" AND event.outcome == "failure"
| STATS fail_count = COUNT(*) BY source.ip
| WHERE fail_count > 50
| SORT fail_count DESC

Top N des IPs suspectes par échecs d'authentification

Requête performante pour agréger les échecs de connexion par IP et filtrer celles dépassant un seuil de risque.

SécuritéAuthForensics

MQL5

bool HasOpenPosition(int magic)
{
   for(int i = PositionsTotal()-1; i >= 0; i--)
   {
      ulong ticket = PositionGetTicket(i);

Vérification sécurisée des positions ouvertes

Vérifie si une position appartient à l'Expert Advisor en utilisant le Magic Number.

TradingPositionsMagicNumber

Python

import pandas as pd

agg = (
    df.groupby("category")
      .agg(

Pandas : agrégations multiples par groupe

Agréger plusieurs colonnes avec des fonctions différentes en un seul groupby().agg() nommé — l'équivalent pandas d'un GROUP BY riche.

PythonPandasgroupby

Vega

{
  "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
  "data": { "values": [
    { "req_per_sec": 120, "latency_ms": 38, "error_rate": 0.1 },
    { "req_per_sec": 240, "latency_ms": 42, "error_rate": 0.1 },

Scatter Plot de Corrélation Temps de Réponse vs Volume

Visualisation complexe permettant d'identifier la dégradation des performances sous charge.

VegaPerfCorrelation

SQL

SELECT
    COALESCE(region,  'ALL REGIONS') AS region,
    COALESCE(country, 'Subtotal')    AS country,
    SUM(revenue)                     AS revenue
FROM sales

Sous-totaux automatiques : ROLLUP

Produire des sous-totaux et un total général en une requête avec GROUP BY ROLLUP — idéal pour un rapport hiérarchique (région → pays → total).

SQLROLLUPGROUPING SETS

Vega

{
  "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
  "data": { "values": [
    { "t": "10:00", "cpu": 32, "mem": 58 }, { "t": "10:05", "cpu": 41, "mem": 59 },
    { "t": "10:10", "cpu": 55, "mem": 61 }, { "t": "10:15", "cpu": 73, "mem": 66 },

Graphique multicouche CPU vs Mémoire

Superpose deux métriques hétérogènes sur un même graphique avec double axe Y.

VegaSystemDashboard

SQL

SELECT
    product,
    SUM(CASE WHEN EXTRACT(MONTH FROM sale_date) = 1 THEN amount ELSE 0 END) AS jan,
    SUM(CASE WHEN EXTRACT(MONTH FROM sale_date) = 2 THEN amount ELSE 0 END) AS feb,
    SUM(CASE WHEN EXTRACT(MONTH FROM sale_date) = 3 THEN amount ELSE 0 END) AS mar,

Pivot (table croisée) par agrégation conditionnelle

Transformer des lignes en colonnes avec SUM(CASE WHEN ...) — un pivot portable, sans clause PIVOT propriétaire.

SQLPivotCASE

SQL

WITH ranked AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY user_id
                              ORDER BY updated_at DESC) AS rn
    FROM user_events

Dédoublonnage : garder la ligne la plus récente

Garder 1 ligne par clé (la plus récente) avec ROW_NUMBER() dans une CTE — robuste et lisible.

SQLDéduplicationROW_NUMBER

SQL

WITH RECURSIVE org AS (
    SELECT id, name, manager_id, 1 AS depth          -- ancre : la racine
    FROM employees
    WHERE manager_id IS NULL
    UNION ALL

CTE récursive : parcourir une hiérarchie

Descendre une arborescence (organigramme, catégories, nomenclature) avec WITH RECURSIVE.

SQLRécursifCTE

SQL

SELECT
    endpoint,
    COUNT(*)                                                 AS calls,
    ROUND(AVG(latency_ms))                                   AS avg_ms,
    PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY latency_ms) AS p50_ms,

Médiane & percentiles (P50 / P95)

Médiane et percentiles robustes avec PERCENTILE_CONT — bien plus représentatif que la moyenne sur des données asymétriques.

SQLPercentileMédiane

SQL

SELECT
    DATE_TRUNC('hour', created_at) AS bucket,
    COUNT(*)                       AS events,
    COUNT(DISTINCT user_id)        AS unique_users,
    SUM(amount)                    AS revenue

Agrégation par tranche de temps (time bucketing)

Regrouper des événements par fenêtre temporelle (heure/jour/semaine) avec DATE_TRUNC pour des séries temporelles propres.

SQLTime SeriesDATE_TRUNC