InsidersTradesSigma

Construire un modèle quant avec une API REST en Python, Sigma Journal

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

Vous devez traiter 162 000 déclarations ? Comment passer d'une API REST à un signal quantitatif efficace en Python ? Cet article vous guide à travers les étapes clés.

Par Sigma Research·17 mai 2026·15 min · 3,220 mots

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 :

La date filtrée est-elle la date de transaction ou la date de publication ?
Les amendements remplacent-ils l’original, ou coexistent-ils ?
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à.

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.

Market	Regulator	Rule	Deadline	Notes
UE	ESMA / autorités nationales	MAR Art 19	T+3 jours ouvrés	Notification des dirigeants et personnes étroitement liées, publication ensuite par l’émetteur ou via les canaux requis.
FR	AMF	MAR Art 19	T+3 jours ouvrés	Application française dans le cadre MAR, avec diffusion réglementée et supervision AMF.
US	SEC	Section 16 / Form 4	T+2 jours ouvrés	Déclaration électronique des transactions des insiders concernés sur EDGAR.

Délais de déclaration, simplifiés. Toujours vérifier le texte applicable, les exceptions et les modalités de publication.

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 :

Poids par rôle
Un achat de CEO n’est pas nécessairement équivalent à celui d’un administrateur non exécutif.
Consensus d’insiders
Plusieurs achats indépendants le même jour sont souvent plus informatifs qu’un seul.
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é.

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.