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à.