De l'API REST au modèle quantitatif : un guide pratique

Une API n’a rien de glamour, jusqu’au jour où elle vous évite de télécharger 162 000 déclarations une par une.

Construire un modèle quant avec notre API REST, tutoriel Python

Les tutoriels API promettent souvent une illumination rapide. En pratique, on commence par récupérer des JSON, on finit par débattre avec des horodatages. C’est normal. Pour un cas d’usage quantitatif sur les transactions d’initiés, il faut accepter une vérité peu romanesque, le rendement du projet vient plus de la plomberie que de la poésie.

Cet article montre comment passer d’une API REST à un prototype de signal en Python. L’angle est volontairement utilitaire, endpoints, authentification, pagination, puis un exemple d’alpha simple. Pas de magie, pas de “secret sauce”, juste une chaîne de traitement propre, reproductible, et suffisamment disciplinée pour ne pas confondre un amendement réglementaire avec une conviction d’achat.

De l’API au signal, la chaîne minimale

Un bon modèle quant ne commence pas par un modèle. Il commence par un contrat de données. Si votre API expose des transactions d’initiés, vous avez en général besoin de quatre familles d’endpoints.

Les endpoints qui comptent vraiment

Le minimum utile ressemble à ceci :

• GET /filings pour lister les déclarations
• GET /filings/{id} pour récupérer le détail d’une déclaration
• GET /issuers pour les métadonnées émetteurs
• GET /insiders ou un champ embarqué pour identifier le déclarant
• éventuellement GET /prices ou une intégration externe pour les rendements post-événement

Dans un monde bien élevé, GET /filings accepte des filtres du type :

• from et to sur la date de publication ou de transaction
• issuer, ticker, isin
• transaction_type
• country
• page et page_size, ou mieux, un curseur

Le point important n’est pas la liste des paramètres. C’est la sémantique. Une API quant doit répondre clairement à trois questions :

1. La date filtrée est-elle la date de transaction ou la date de publication ?
2. Les amendements remplacent-ils l’original, ou coexistent-ils ?
3. L’ordre des résultats est-il stable entre deux appels paginés ?

Si vous n’avez pas une réponse nette aux trois, vous avez déjà une source de biais.

Authentification, simple mais propre

En Python, l’authentification par bearer token suffit dans la plupart des cas. Le piège n’est pas technique, il est opérationnel. Les tokens finissent dans les notebooks, puis dans Git, puis dans les larmes.

Utilisez une variable d’environnement, et centralisez la session HTTP.

import os
import requests

BASE_URL = os.getenv("INSIDER_API_BASE_URL", "https://api.example.com/v1")
API_TOKEN = os.getenv("INSIDER_API_TOKEN")

if not API_TOKEN:
    raise RuntimeError("INSIDER_API_TOKEN manquant")

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {API_TOKEN}",
    "Accept": "application/json",
    "User-Agent": "sigma-journal-tutorial/1.0"
})

Ajoutez ensuite une petite fonction de garde pour les erreurs réseau et les codes HTTP.

def get_json(path, params=None, timeout=30):
    url = f"{BASE_URL}{path}"
    resp = session.get(url, params=params, timeout=timeout)
    resp.raise_for_status()
    return resp.json()

Oui, c’est banal. Oui, c’est exactement ce qui évite les scripts de 300 lignes avec une gestion d’erreur inventée à 23h17.

Pagination, le détail qui décide de la qualité du dataset

La pagination est souvent traitée comme une formalité. C’est une erreur classique. Si votre historique est volumineux, 162 000 déclarations dans votre cas, la pagination devient un sujet de qualité des données.

Deux schémas existent :

• pagination par page, page=1&page_size=100
• pagination par curseur, next_cursor=abc123

Le curseur est généralement plus sûr si les données évoluent pendant l’extraction. La pagination par page est acceptable si l’ordre est stable, par exemple published_at asc, id asc.

Exemple générique avec pagination par page :

import pandas as pd

def fetch_all_filings(start_date, end_date, page_size=500):
    rows = []
    page = 1

    while True:
        payload = get_json(
            "/filings",
            params={
                "from": start_date,
                "to": end_date,
                "page": page,
                "page_size": page_size,
                "sort": "published_at:asc,id:asc"
            }
        )

        items = payload.get("results", [])
        rows.extend(items)

        if not items or page >= payload.get("total_pages", page):
            break

        page += 1

    return pd.DataFrame(rows)

Si l’API renvoie un curseur :

def fetch_all_filings_cursor(start_date, end_date, limit=500):
    rows = []
    cursor = None

    while True:
        params = {
            "from": start_date,
            "to": end_date,
            "limit": limit,
            "sort": "published_at:asc,id:asc"
        }
        if cursor:
            params["cursor"] = cursor

        payload = get_json("/filings", params=params)
        items = payload.get("results", [])
        rows.extend(items)

        cursor = payload.get("next_cursor")
        if not cursor:
            break

    return pd.DataFrame(rows)

Le point non négociable, stockez aussi les métadonnées de collecte, date d’extraction, paramètres, version d’API si disponible. Quand un backtest bouge sans raison apparente, le coupable est souvent là.

Le schéma de données, ou comment éviter un faux signal

Une transaction d’initié n’est pas une ligne de prix. C’est un événement réglementaire. Cela change tout.

Les champs à conserver dès le départ

Même si votre endpoint renvoie cinquante colonnes, quelques champs sont structurants :

• filing_id
• issuer_id, issuer_name, ticker, isin
• insider_id, insider_name, insider_role
• transaction_date
• published_at
• transaction_type
• security_type
• shares
• price
• currency
• value
• ownership_nature si disponible, direct/indirect
• amendment_flag ou status

À ce stade, ne cherchez pas encore à “faire du signal”. Cherchez à rendre les événements comparables.

Exemple de normalisation en pandas :

def normalize_filings(df):
    out = df.copy()

    # Dates
    out["transaction_date"] = pd.to_datetime(out["transaction_date"], utc=True, errors="coerce")
    out["published_at"] = pd.to_datetime(out["published_at"], utc=True, errors="coerce")

    # Numériques
    for col in ["shares", "price", "value"]:
        if col in out.columns:
            out[col] = pd.to_numeric(out[col], errors="coerce")

    # Standardisation des libellés
    if "transaction_type" in out.columns:
        out["transaction_type"] = (
            out["transaction_type"]
            .astype(str)
            .str.strip()
            .str.lower()
        )

    # Valeur calculée si absente
    if "value" not in out.columns or out["value"].isna().all():
        if {"shares", "price"}.issubset(out.columns):
            out["value"] = out["shares"] * out["price"]

    return out

Ce qu’il faut filtrer, sans état d’âme

Toutes les transactions d’initiés ne portent pas de contenu informationnel équivalent. Une vente liée à l’exercice automatique d’options n’a pas le même sens qu’un achat au marché par un dirigeant.

Un filtre de départ raisonnable consiste à :

• retenir les achats et ventes au marché
• exclure les dons, successions, transferts internes, conversions mécaniques
• traiter séparément les plans automatiques ou programmés
• dédupliquer les amendements ou ne garder que la version la plus récente

Exemple simple :

BUY_LABELS = {"buy", "purchase", "acquisition"}
SELL_LABELS = {"sell", "sale", "disposition"}

def filter_informative_transactions(df):
    out = df.copy()

    out = out[out["transaction_type"].isin(BUY_LABELS | SELL_LABELS)]

    if "amendment_flag" in out.columns:
        out = out[out["amendment_flag"] != True]

    out = out.dropna(subset=["issuer_id", "transaction_date", "value"])
    out = out[out["value"] > 0]

    return out

Évidemment, les libellés exacts dépendent de votre taxonomie. Le principe, lui, ne change pas.

Date de transaction contre date de publication

C’est le point où beaucoup de prototypes deviennent involontairement prophétiques. Une transaction a lieu à une date donnée, mais le marché n’en prend connaissance qu’à la publication. Pour un backtest ou un signal exploitable, la date pertinente est généralement published_at, pas transaction_date.

La réglementation européenne, via le règlement abus de marché, impose aux personnes exerçant des responsabilités dirigeantes et aux personnes qui leur sont étroitement liées de notifier les transactions dans un délai défini, souvent résumé en T+3 jours ouvrés pour les notifications visées à l’article 19 du MAR. Cela crée un décalage structurel entre événement économique et disponibilité de l’information. Si vous scorez au jour de transaction sans tenir compte de ce délai, vous ajoutez un léger parfum de voyage temporel à votre stratégie. Les investisseurs appellent cela de l’alpha. Les relecteurs sérieux appellent cela un bug.

Python, extraction robuste et table d’événements

Une fois l’authentification et la pagination en place, l’objectif est de construire une table d’événements propre, une ligne par événement exploitable.

Extraire par fenêtres temporelles

Même avec une pagination correcte, il est souvent plus prudent d’extraire par mois ou par trimestre, puis de concaténer. Vous réduisez le risque de timeout, et vous facilitez les reprises partielles.

from datetime import datetime
import pandas as pd

def month_starts(start, end):
    dates = pd.date_range(start=start, end=end, freq="MS", tz="UTC")
    return list(dates)

def fetch_range_by_month(start="2020-01-01", end="2024-12-31"):
    starts = month_starts(start, end)
    chunks = []

    for i, dt in enumerate(starts):
        chunk_start = dt.strftime("%Y-%m-%d")
        if i + 1 < len(starts):
            chunk_end = (starts[i + 1] - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
        else:
            chunk_end = pd.Timestamp(end, tz="UTC").strftime("%Y-%m-%d")

        df = fetch_all_filings(chunk_start, chunk_end)
        df["extract_window_start"] = chunk_start
        df["extract_window_end"] = chunk_end
        chunks.append(df)

    return pd.concat(chunks, ignore_index=True) if chunks else pd.DataFrame()

Dédupliquer proprement

Le cas classique, une déclaration initiale, puis un amendement. Ou bien la même transaction apparaissant dans deux extractions successives. Il faut définir une clé de déduplication.

Si filing_id est stable, parfait. Sinon, utilisez une clé composite prudente :

• issuer_id
• insider_id
• transaction_date
• transaction_type
• shares
• price

Puis gardez la ligne à published_at la plus récente.

def deduplicate_filings(df):
    out = df.copy()

    if "filing_id" in out.columns and out["filing_id"].notna().any():
        out = out.sort_values("published_at").drop_duplicates("filing_id", keep="last")
        return out

    key_cols = ["issuer_id", "insider_id", "transaction_date", "transaction_type", "shares", "price"]
    key_cols = [c for c in key_cols if c in out.columns]

    out = out.sort_values("published_at").drop_duplicates(key_cols, keep="last")
    return out

Enrichir avec les prix de marché

Pour calculer un alpha, même rudimentaire, il faut relier les événements à des rendements futurs. Si votre API ne fournit pas les prix, utilisez une source externe. L’important est d’aligner proprement :

• la date de signal, généralement le jour de publication ou le prochain jour de bourse
• le prix d’entrée, à la clôture du jour de signal ou à l’ouverture suivante
• l’horizon de sortie, par exemple 20 jours de bourse

Exemple conceptuel :

def build_event_table(filings_df, prices_df):
    events = filings_df.copy()

    # Date de signal = date de publication normalisée au jour
    events["signal_date"] = events["published_at"].dt.floor("D")

    # Agrégation minimale
    cols = ["issuer_id", "ticker", "signal_date", "transaction_type", "value", "insider_role"]
    events = events[cols].copy()

    # Score brut signé
    events["signed_value"] = events["value"].where(
        events["transaction_type"].isin(BUY_LABELS), -events["value"]
    )

    daily = (
        events.groupby(["issuer_id", "ticker", "signal_date"], as_index=False)
        .agg(net_value=("signed_value", "sum"),
             gross_value=("value", "sum"),
             n_events=("value", "count"))
    )

    # prices_df attendu avec ticker, date, close
    prices = prices_df.copy()
    prices["date"] = pd.to_datetime(prices["date"], utc=True).dt.floor("D")

    daily = daily.merge(
        prices.rename(columns={"date": "signal_date", "close": "entry_close"}),
        on=["ticker", "signal_date"],
        how="left"
    )

    return daily

Le tutoriel s’arrête ici sur la partie marché, car les conventions de prix dépendent de votre stack. Mais la logique est stable.

Un alpha simple, et volontairement modeste

Le premier signal ne doit pas être sophistiqué. Il doit être falsifiable.

Idée de base, achats nets agrégés

Le signal le plus simple consiste à agréger les achats et ventes d’initiés par émetteur et par date de publication, puis à classer les titres selon un score.

Une version minimale :

[ score_{i,t} = \frac{achats_nets_en_valeur_{i,t}}{ADV_{i,t}} ]

où ADV est le volume monétaire moyen quotidien, ou à défaut une autre mesure d’échelle. Si vous n’avez pas l’ADV, vous pouvez commencer par un score brut ou normaliser par capitalisation boursière si elle est disponible. Si vous n’avez ni l’un ni l’autre, écrivez n/a dans votre documentation et assumez que le prototype est incomplet. C’est plus élégant qu’un ratio improvisé.

Exemple sans ADV, avec une transformation logarithmique prudente :

import numpy as np

def compute_simple_score(events_daily):
    df = events_daily.copy()

    df["net_value_clipped"] = df["net_value"].clip(lower=-1e9, upper=1e9)
    df["score"] = np.sign(df["net_value_clipped"]) * np.log1p(np.abs(df["net_value_clipped"]))

    return df

Ajouter un peu de structure, sans suringénierie

Vous pouvez améliorer ce score avec trois raffinements simples :

1. Poids par rôle
   Un achat de CEO n’est pas nécessairement équivalent à celui d’un administrateur non exécutif.

2. Consensus d’insiders
   Plusieurs achats indépendants le même jour sont souvent plus informatifs qu’un seul.

3. Récence
   Un cluster d’achats sur quelques jours peut être plus intéressant qu’un événement isolé.

Exemple :

ROLE_WEIGHTS = {
    "ceo": 1.5,
    "cfo": 1.3,
    "chair": 1.2,
    "director": 1.0
}

def add_role_weight(events):
    df = events.copy()
    df["role_weight"] = (
        df["insider_role"]
        .astype(str)
        .str.lower()
        .map(ROLE_WEIGHTS)
        .fillna(1.0)
    )
    df["weighted_signed_value"] = df["signed_value"] * df["role_weight"]
    return df

Puis agrégation :

def aggregate_weighted(events):
    daily = (
        events.groupby(["issuer_id", "ticker", "signal_date"], as_index=False)
        .agg(
            weighted_net_value=("weighted_signed_value", "sum"),
            n_events=("weighted_signed_value", "count")
        )
    )
    daily["score"] = (
        np.sign(daily["weighted_net_value"])
        * np.log1p(np.abs(daily["weighted_net_value"]))
        * np.log1p(daily["n_events"])
    )
    return daily

Ce n’est pas un saint Graal. C’est un point de départ testable. Ce qui est déjà plus utile que la moitié des pitch decks quant.

Transformer le score en portefeuille

Une règle simple :

• chaque jour, classer les titres par score
• prendre le décile supérieur en long
• éventuellement le décile inférieur en short, si le marché et la liquidité le permettent
• rebalancer chaque jour ou chaque semaine
• mesurer les rendements à 5, 20 ou 60 jours

Pseudo-code :

def select_portfolio(scored_df, top_q=0.1):
    df = scored_df.copy()

    def pick(group):
        n = max(1, int(len(group) * top_q))
        return group.nlargest(n, "score")

    longs = (
        df.groupby("signal_date", group_keys=False)
        .apply(pick)
        .assign(weight=lambda x: 1 / x.groupby("signal_date")["ticker"].transform("count"))
    )

    return longs

Sans backtest chiffré fourni par vos données live, il serait malhonnête de prétendre à une performance. Le bon réflexe est donc de s’arrêter avant l’exagération. Le modèle est défini, pas validé.

Les pièges méthodologiques qui coûtent cher

Les modèles sur transactions d’initiés ont une réputation flatteuse. Elle est parfois méritée, parfois obtenue au prix d’un peu trop d’optimisme statistique.

Le look-ahead bias, encore lui

Nous l’avons déjà mentionné, mais il mérite sa propre sous-section. Si vous utilisez transaction_date comme date de signal, vous supposez que le marché connaissait l’information avant sa publication. C’est faux. Le bon alignement est :

• transaction à T
• publication réglementaire à T+k
• entrée en position à T+k ou T+k+1 selon votre convention d’exécution

Cette distinction est triviale à écrire, moins triviale à respecter quand on joint des tables à la va-vite.

Le survivorship bias

Si votre univers d’émetteurs est construit à partir des sociétés encore cotées aujourd’hui, vous éliminez mécaniquement les disparues, souvent pour de mauvaises raisons, acquisition, faillite, radiation. Votre backtest devient plus propre, et moins vrai.

Il faut donc idéalement :

• un historique complet des émetteurs au moment de l’événement
• des identifiants stables
• des prix incluant les titres sortis de cote

Les fusions de tickers et identifiants

Les tickers changent. Les ISIN peuvent changer. Les noms changent plus souvent qu’on ne le souhaiterait. Pour les jointures, privilégiez un identifiant interne stable d’émetteur, puis mappez vers les symboles de marché pour les prix.

Le bruit des petites transactions

Une transaction de faible montant peut être sincère, mais elle peut aussi être statistiquement inutile. Beaucoup de praticiens imposent un seuil minimal en valeur, ou rapportent la transaction à la rémunération estimée du dirigeant, à sa détention existante, ou à la liquidité du titre.

Sans donnée de position préalable, un filtre simple en valeur absolue est souvent préférable à rien. Il est imparfait, mais explicite.

Un squelette de pipeline de production en Python

Le prototype devient utile quand il peut tourner de façon répétable. Pas forcément “en production” au sens noble du terme, mais au moins sans surveillance continue et sans drame.

Organisation minimale du projet

Une structure simple suffit :

project/
  config.py
  api.py
  extract.py
  normalize.py
  signals.py
  backtest.py
  data/
  notebooks/

Avec :

• api.py pour la session HTTP et les appels
• extract.py pour les extractions incrémentales
• normalize.py pour les schémas et filtres
• signals.py pour les scores
• backtest.py pour l’évaluation

Extraction incrémentale

Une API REST se prête bien à une logique de “watermark”, on mémorise la dernière date de publication ingérée, puis on ne récupère que ce qui est nouveau.

from pathlib import Path
import json

STATE_FILE = Path("data/state.json")

def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"last_published_at": "2020-01-01T00:00:00Z"}

def save_state(state):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(json.dumps(state))

def incremental_update():
    state = load_state()
    df = fetch_all_filings(state["last_published_at"], "2099-12-31")
    if df.empty:
        return df

    df = normalize_filings(df)
    df = deduplicate_filings(df)

    max_ts = df["published_at"].max()
    state["last_published_at"] = max_ts.isoformat().replace("+00:00", "Z")
    save_state(state)

    return df

Il faudra ensuite écrire vers un stockage local ou une base analytique. Parquet est souvent un bon compromis pour commencer.

Validation automatique

Ajoutez quelques tests simples :

• pas de published_at manquant
• pas de value négative si la convention ne le permet pas
• unicité de filing_id
• proportion d’événements filtrés raisonnable
• distribution des types de transaction stable dans le temps

Quand ces tests cassent, ce n’est pas forcément votre code. C’est parfois l’API qui a changé. Ce qui est une autre façon de dire que votre code a cassé, mais avec un responsable externe.

Journalisation et reprise

Logguez :

• nombre de pages récupérées
• nombre de lignes brutes
• nombre de lignes après déduplication
• nombre de lignes après filtrage
• fenêtre temporelle traitée
• version du schéma si disponible

Ce journal est souvent plus utile qu’un dashboard sophistiqué. Surtout le lendemain.

Ce que disent les sources sérieuses, et ce qu’elles ne disent pas

La littérature académique et réglementaire est utile, à condition de ne pas lui faire dire plus qu’elle ne dit.

Le cadre réglementaire est une contrainte de données

L’ESMA et les autorités nationales, comme l’AMF en France, encadrent la notification et la publication des transactions des dirigeants sous MAR. Aux États-Unis, la SEC impose les déclarations pertinentes via les formulaires de la Section 16, notamment le Form 4. Pour le quant, cela signifie surtout deux choses :

• les délais de publication sont codifiés, donc modélisables
• les catégories de transactions sont réglementées, donc filtrables

Autrement dit, la réglementation n’est pas seulement un arrière-plan juridique. C’est une partie du schéma de données.

La littérature académique suggère une asymétrie achats versus ventes

Les travaux classiques sur l’insider trading légal montrent souvent que les achats d’initiés sont plus informatifs que les ventes, ces dernières pouvant répondre à des besoins de diversification, de fiscalité ou de liquidité personnelle. C’est un résultat robuste dans l’esprit, avec des nuances selon les marchés, les périodes et les filtres.

Pour un premier modèle, cela justifie une décision simple, commencer par les achats. C’est moins ambitieux, et souvent plus propre.

Ce que la littérature ne fait pas à votre place

Elle ne nettoie pas vos amendements. Elle ne résout pas vos identifiants. Elle ne vous dit pas si votre endpoint renvoie l’heure locale ou UTC. Et elle ne vous pardonnera pas si vous confondez date de transaction et date de publication.

Autrement dit, les papiers donnent des hypothèses. Le pipeline décide si elles survivent au contact des données.

Le lecteur attentif notera que cette “citation” est une synthèse, pas une phrase canonique tirée mot à mot d’un papier. C’est précisément pour cela qu’elle est étiquetée comme telle. Les nuances existent, et elles méritent mieux qu’un slogan.

Le plus important n’est pas de construire un score brillant en une après-midi. C’est de bâtir une chaîne de données qui ne ment pas. Avec votre API REST, la prochaine étape concrète est simple, implémenter une extraction incrémentale sur published_at, figer un schéma d’événements, puis backtester un signal achat-only sur un horizon de 20 jours avec filtres de liquidité explicites. La vraie question ouverte n’est pas “l’insider buying marche-t-il ?”, elle est plus exigeante, “quelle part du signal survit après nettoyage réglementaire, délais réels de publication, et coûts d’exécution ?”

Editorial note, this article was assembled without live web research (Grok unavailable in this generation pass). Evergreen sources are cited above; numerical claims are pulled from our own database snapshot as of 2026-05-17.

Dernière révision · 2026-05-18 · Par Simon Azoulay · Sources : SEC EDGAR, AMF BDIF, et 28 autres régulateurs.