khalilbenaz/claude-skills-collection/agent-data-analyst-agent

# SQL — SQLAlchemy (PostgreSQL, MySQL, SQLite, BigQuery, MSSQL)
from sqlalchemy import create_engine, inspect, text
engine = create_engine("postgresql://user:pass@host/db")
inspector = inspect(engine)
# Générer le metadata store à la connexion
schema = {
    table: {
        "columns": inspector.get_columns(table),
        "pk": inspector.get_pk_constraint(table),
        "fk": inspector.get_foreign_keys(table),
    }
    for table in inspector.get_table_names()
}
# Read-only : toujours ouvrir en mode lecture
with engine.connect() as conn:
    conn.execute(text("SET TRANSACTION READ ONLY"))

# Fichiers
import pandas as pd
df = pd.read_csv("data.csv", parse_dates=True, low_memory=False)
df = pd.read_excel("data.xlsx", sheet_name=0)
# API REST
import httpx
resp = httpx.get("https://api.example.com/data", headers={"Authorization": f"Bearer {token}"})
df = pd.DataFrame(resp.json()["data"])

def profile_dataframe(df: pd.DataFrame) -> dict:
    return {
        "shape": df.shape,
        "dtypes": df.dtypes.astype(str).to_dict(),
        "missing_pct": (df.isnull().mean() * 100).round(2).to_dict(),
        "nunique": df.nunique().to_dict(),
        "stats": df.describe(include="all").round(3).to_dict(),
        "sample": df.head(5).to_dict(orient="records"),
    }

system_prompt = f"""
Tu es un analyste SQL expert. Schéma de la base :
{json.dumps(schema, ensure_ascii=False, indent=2)}
Règles :
- Génère UNIQUEMENT des SELECT (pas de DML/DDL)
- Limite toujours à LIMIT 10000
- Explique la requête en une phrase simple en français
- Si la question est ambiguë, demande une clarification avant de générer
"""

pandas_prompt = f"""
DataFrame disponible (df) :
Colonnes : {df.dtypes.to_dict()}
Aperçu :
{df.head(3).to_string()}
Génère le code pandas pour : {user_question}
Utilise uniquement les colonnes existantes. Termine par print(result).
"""

def select_chart_type(df: pd.DataFrame, x: str, y: str) -> str:
    if pd.api.types.is_datetime64_any_dtype(df[x]):
        return "lineplot"      # série temporelle
    if df[x].nunique() <= 15:
        return "barplot"       # catégorielle basse cardinalité
    if pd.api.types.is_numeric_dtype(df[x]) and pd.api.types.is_numeric_dtype(df[y]):
        return "scatter"       # corrélation numérique
    return "histogram"         # distribution par défaut

import plotly.express as px
fig = px.bar(df, x="categorie", y="ventes", title="Ventes par catégorie",
             labels={"ventes": "Ventes (€)", "categorie": "Catégorie"})
fig.write_html("chart.html")

import numpy as np
from scipy import stats
# Détection d'anomalies — z-score
def detect_anomalies_zscore(df, col, threshold=3.0):
    z = np.abs(stats.zscore(df[col].dropna()))
    return df[z > threshold]
# Tendance sur série temporelle — régression linéaire
from sklearn.linear_model import LinearRegression
def compute_trend(df, date_col, value_col):
    df = df.sort_values(date_col)
    X = np.arange(len(df)).reshape(-1, 1)
    y = df[value_col].values
    model = LinearRegression().fit(X, y)
    return {"slope": model.coef_[0], "r2": model.score(X, y)}
# Corrélation avec significativité
def correlate(df, col_a, col_b):
    r, p = stats.pearsonr(df[col_a].dropna(), df[col_b].dropna())
    return {"r": round(r, 3), "p_value": round(p, 4), "significant": p < 0.05}

session = {
    "history": [],       # [(question, sql_or_code, result_summary)]
    "last_df": None,     # dernier DataFrame résultat
    "filters_active": {},# filtres appliqués
}
# Après chaque réponse, proposer 2-3 questions de suivi pertinentes
followup_prompt = f"""
L'utilisateur vient de demander : {question}
Résultat : {summary}
Propose 3 questions de suivi pertinentes, courtes, en français.
"""

1. Executive Summary (3-5 bullets, chiffres clés)
2. Méthodologie (source, période, transformations appliquées, exclusions)
3. Findings (insights classés par impact estimé, graphiques intégrés)
4. Recommandations actionnables (formulées en verbes d'action)
5. Limites et caveats (qualité des données, biais, lacunes)

# Markdown
report.to_markdown("report.md")
# HTML standalone avec graphiques Plotly intégrés
report.to_html("report.html", include_plotlyjs="cdn")

import re
def mask_pii(text: str) -> str:
    text = re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '[EMAIL]', text)
    text = re.sub(r'\b(\+?216|0)[0-9]{8}\b', '[PHONE]', text)
    return text

# Normalisation UTC obligatoire
df["created_at"] = pd.to_datetime(df["created_at"], utc=True)