Un 'quant stack' es tan bueno como su infraestructura, y la mayoría del alfa muere en algún lugar entre la Authorization y la página 37.
Un 'quant stack' es tan bueno como su infraestructura, y la mayoría del alfa muere en algún lugar entre la Authorization y la página 37.
From raw filings to a workflow: API, MCP, Excel, alerts, and how to read backtest claims honestly.
Hay dos maneras de escribir un tutorial cuantitativo. Una es prometer un modelo que supere al mercado en 40 líneas. La otra es mostrar las partes que fallan primero. Elegiremos la segunda opción, en parte porque es honesta, y en parte porque la paginación rota es menos glamorosa pero más común.
Este artículo describe un flujo de trabajo práctico en Python para construir un primer modelo de "insider filings" con una API REST. El énfasis está en cuatro aspectos:
Dado que no se proporcionó una extracción de datos específica del artículo en vivo, no pretenderé lo contrario. Cuando un número no está disponible, se marca como n/a. La única cifra interna concreta que tenemos es la escala del archivo de "filings", 162,000 "filings". Eso es suficiente para hacer una observación inmediata: una vez que el conjunto de datos es lo suficientemente grande, los scripts improvisados dejan de ser encantadores.
Usted conoce Python básico, requests y pandas. No necesita una plataforma completa de investigación de factores. Una computadora portátil, un token y una ligera tolerancia al JSON son suficientes.
Mantendremos los ejemplos de API genéricos, porque el resumen del artículo solicita "endpoints", autenticación, paginación y un ejemplo alfa simple, no un santuario de ingeniería inversa para un detalle de implementación específico. Reemplace las rutas y campos de los marcadores de posición con su esquema de API real.
Para un cliente de investigación, mantenga tres capas separadas:
Si colapsa eso en una celda de "notebook", obtendrá resultados rápidamente. También obtendrá un futuro informe de errores escrito por su yo pasado.
Una API REST para "filings" generalmente expone alguna combinación de estos recursos:
/filings/issuers/insiders/transactions/symbols o /instrumentsLa nomenclatura exacta varía. El patrón no.
La mayoría de las API en esta categoría utilizan autenticación con token de portador o una cabecera de clave API. Un patrón limpio es envolver un requests.Session para que cada solicitud comparta cabeceras, reglas de tiempo de espera y lógica de reintento.
import time
from typing import Dict, Iterable, List, Optional
import requests
import pandas as pd
class InsiderAPI:
def __init__(self, base_url: str, api_key: str, timeout: int = 30):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
"User-Agent": "sigma-journal-python-tutorial/1.0"
})
self.timeout = timeout
def _get(self, path: str, params: Optional[Dict] = None) -> Dict:
url = f"{self.base_url}{path}"
resp = self.session.get(url, params=params or {}, timeout=self.timeout)
resp.raise_for_status()
return resp.json()
Eso es suficiente para empezar. No es suficiente para producción. En la práctica, también querrá:
429 y 5xx transitorios,from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class InsiderAPI:
def __init__(self, base_url: str, api_key: str, timeout: int = 30):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
"User-Agent": "sigma-journal-python-tutorial/1.0"
})
retry = Retry(
total=5,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"]
)
adapter = HTTPAdapter(max_retries=retry)
self.session.mount("https://", adapter)
self.session.mount("http://", adapter)
self.timeout = timeout
def _get(self, path: str, params=None):
url = f"{self.base_url}{path}"
r = self.session.get(url, params=params or {}, timeout=self.timeout)
r.raise_for_status()
return r.json()
Seco, aburrido, útil. Un buen comienzo.
Las API de filings a menudo admiten filtros como:
from_dateto_datecountryissuer_idinsider_idtransaction_typepagepage_sizecursorSi su API ofrece paginación por offset y paginación por cursor, prefiera la paginación por cursor para extracciones grandes. La paginación por offset es sencilla. También es una buena manera de descubrir condiciones de carrera cuando llegan nuevos registros a mitad de la descarga.
La paginación es donde muchos conjuntos de datos de investigación se vuelven silenciosamente incorrectos. Perder una página en un archivo de 162,000 registros no activará una excepción dramática. Simplemente hará que su backtest sea un poco más ficticio.
Una forma de respuesta común se ve algo así:
{
"results": [...],
"page": 3,
"page_size": 100,
"total_pages": 57,
"total_count": 5678
}
Un bucle sencillo:
def fetch_all_filings_offset(api: InsiderAPI, start_date: str, end_date: str) -> pd.DataFrame:
page = 1
rows = []
while True:
payload = api._get("/filings", params={
"from_date": start_date,
"to_date": end_date,
"page": page,
"page_size": 500
})
batch = payload.get("results", [])
if not batch:
break
rows.extend(batch)
total_pages = payload.get("total_pages")
if total_pages is not None and page >= total_pages:
break
page += 1
return pd.DataFrame(rows)
Esto funciona, con una salvedad. Si se añaden nuevos registros mientras se pagina el conjunto de resultados, las páginas posteriores pueden desplazarse. La solución es sencilla en principio y a menudo se descuida en la práctica: ordenar por una clave estable y, cuando sea posible, consultar una ventana histórica fija.
Una forma de respuesta más robusta:
{
"results": [...],
"next_cursor": "eyJpZCI6..."
}
El bucle se convierte en:
def fetch_all_filings_cursor(api: InsiderAPI, start_date: str, end_date: str) -> pd.DataFrame:
cursor = None
rows = []
while True:
params = {
"from_date": start_date,
"to_date": end_date,
"page_size": 500
}
if cursor:
params["cursor"] = cursor
payload = api._get("/filings", params=params)
batch = payload.get("results", [])
rows.extend(batch)
cursor = payload.get("next_cursor")
if not cursor:
break
return pd.DataFrame(rows)
La paginación por cursor suele ser la respuesta correcta para la extracción masiva. Si su API la admite, úsela.
Para la investigación diaria, no vuelva a extraer el archivo completo cada mañana a menos que disfrute desperdiciar ancho de banda e introducir puntos de fallo. Mantenga un almacenamiento local y realice extracciones incrementales:
filed_at > last_seen_timestamp,Esa ventana de solapamiento es importante. Los reguladores y los emisores no siempre son tan puntuales como su tarea programada (cron job).
Una carga útil de "filing" en bruto aún no es una señal. Es un objeto administrativo con campos anidados, valores opcionales y suficientes casos extremos para mantener ocupado a un "parser".
Para un primer modelo, generalmente querrá:
filing_idissuer_idissuer_nametickerinsider_idinsider_nameinsider_roletransaction_datefiling_datetransaction_typesharespricecurrencyvalueownership_nature si está disponible, directa o indirectasecurity_typeSi la API no proporciona value, calcúlelo como shares * price cuando sea sensato, y marque explícitamente los componentes faltantes.
def normalize_filings(df: pd.DataFrame) -> pd.DataFrame:
out = pd.DataFrame({
"filing_id": df.get("id"),
"issuer_id": df.get("issuer_id"),
"issuer_name": df.get("issuer_name"),
"ticker": df.get("ticker"),
"insider_id": df.get("insider_id"),
"insider_name": df.get("insider_name"),
"insider_role": df.get("insider_role"),
"transaction_date": pd.to_datetime(df.get("transaction_date"), errors="coerce"),
"filing_date": pd.to_datetime(df.get("filing_date"), errors="coerce"),
"transaction_type": df.get("transaction_type"),
"shares": pd.to_numeric(df.get("shares"), errors="coerce"),
"price": pd.to_numeric(df.get("price"), errors="coerce"),
"currency": df.get("currency"),
"security_type": df.get("security_type"),
})
out["value"] = out["shares"] * out["price"]
out["days_to_file"] = (out["filing_date"] - out["transaction_date"]).dt.days
return out
Esto es intencionadamente sencillo. El objetivo es establecer columnas tipadas y diagnósticos básicos, no ganar un premio por abstracción decorativa.
Una primera estrategia de información privilegiada suele mejorar más con exclusiones sensatas que con ponderaciones sofisticadas. Considere filtrar:
Las reglas exactas dependen de la jurisdicción y el diseño de la fuente de datos. El principio es universal: si no puede explicar por qué una transacción pertenece a la señal, probablemente no debería hacerlo.
Una presentación puede contener varias líneas de transacción. Si un ejecutivo compra en tres tramos en un día, tratarlos como tres señales independientes puede exagerar la amplitud. Una solución común es agregar a un nivel de evento:
Eso le da una unidad más limpia para la clasificación.
Ahora, la parte a la que la gente se salta. Justo.
La literatura sobre el uso de información privilegiada ha documentado durante mucho tiempo que algunas compras de información privilegiada, especialmente las compras en el mercado abierto por parte de altos funcionarios y directores, contienen información sobre rendimientos futuros. El efecto no es uniforme, no es gratuito y no es inmune al hacinamiento. Pero es un punto de partida razonable para un modelo simple.
Un primer prototipo puede ser:
Clasifique la actividad reciente de compra de información privilegiada a nivel de emisor por una puntuación de convicción, luego seleccione los nombres principales para pruebas adicionales.
Eso no es una estrategia terminada. Es un candidato a señal.
Para cada emisor durante los últimos 30 días naturales, calcule:
Luego defina una puntuación como:
score =
log1p(total_buy_value)
+ 0.75 * distinct_buyers
+ 1.0 * senior_buyer_flag
- 0.05 * days_since_last_buy
Puede discutir los coeficientes. Debería hacerlo. El objetivo es empezar con algo legible.
Para una primera pasada, concéntrese en las compras. Las ventas son más ruidosas porque la gente vende por muchas razones: impuestos, diversificación, matrículas escolares, yates. Las compras requieren que el efectivo se mueva en la dirección difícil. Los mercados tienden a notarlo.
import numpy as np
def build_buy_signal(df: pd.DataFrame, as_of_date: str) -> pd.DataFrame:
as_of = pd.Timestamp(as_of_date)
x = df.copy()
x = x[
(x["transaction_type"].str.lower().isin(["buy", "purchase", "acquisition"])) &
(x["transaction_date"].notna()) &
(x["transaction_date"] <= as_of) &
(x["transaction_date"] >= as_of - pd.Timedelta(days=30))
]
x["is_senior"] = x["insider_role"].fillna("").str.lower().str.contains(
"ceo|chief|cfo|director|chair"
)
grouped = (
x.groupby(["issuer_id", "issuer_name", "ticker"], dropna=False)
.agg(
total_buy_value=("value", "sum"),
distinct_buyers=("insider_id", "nunique"),
senior_buyer_flag=("is_senior", "max"),
last_buy_date=("transaction_date", "max")
)
.reset_index()
)
grouped["days_since_last_buy"] = (as_of - grouped["last_buy_date"]).dt.days
grouped["score"] = (
np.log1p(grouped["total_buy_value"].fillna(0))
+ 0.75 * grouped["distinct_buyers"].fillna(0)
+ 1.0 * grouped["senior_buyer_flag"].astype(int)
- 0.05 * grouped["days_since_last_buy"].fillna(30)
)
return grouped.sort_values("score", ascending=False)
Esto produce una tabla clasificada. Todavía no responde si la señal funciona. Sí responde si su pipeline puede generar una lista diaria de candidatos sin improvisación.
Eso es suficiente para crear un ciclo de investigación repetible.
Los datos de insiders tienen un talento especial para parecer más limpios de lo que son. Una presentación es un documento formal, por lo tanto, uno se siente tranquilo. Esa tranquilidad debe ser racionada.
Este es el error clásico. Si clasifica valores utilizando transaction_date pero solo pudo haber observado la presentación en filing_date o la marca de tiempo de publicación, su backtest está haciendo trampa. Gran parte de las tonterías publicadas comienzan de esta manera.
La fecha correcta para la disponibilidad de la señal es la marca de tiempo más temprana en la que su sistema pudo haber visto la presentación. Dependiendo de su fuente, puede ser:
created_at o published_at de su propia API.Si todo lo que tiene es una fecha de presentación sin hora intradiaria, asuma que la señal se vuelve negociable no antes de la siguiente sesión, a menos que su convención de mercado y el momento de la alimentación admitan algo más preciso.
Una presentación puede ser corregida, reenviada o representada en múltiples filas. Si su API expone indicadores de enmienda o versiones, úselos. Si no, elimine duplicados en una clave conservadora como emisor, insider, fecha de transacción, acciones, precio y dirección, mientras mantiene los registros brutos archivados.
Un campo de rol de "insider" que dice Officer en un caso y Chief Financial Officer en otro no es una taxonomía, es una sugerencia. Construya una capa de mapeo. Manténgala versionada. Espere revisarla.
Si su universo abarca múltiples mercados, los valores nocionales en moneda local no son directamente comparables. Convierta a una moneda base utilizando una fuente de tipo de cambio consistente si su puntuación utiliza el valor. Si aún no tiene un mapeo de tipo de cambio fiable, clasifique primero dentro del mercado. Es mejor un modelo con un alcance modesto que uno global sostenido por una aritmética ilusoria.
A continuación, se presenta un script compacto que une las piezas. Asume un endpoint /filings y un token de portador.
import pandas as pd
import numpy as np
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class InsiderAPI:
def __init__(self, base_url: str, api_key: str, timeout: int = 30):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
"User-Agent": "sigma-journal-python-tutorial/1.0"
})
retry = Retry(
total=5,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET"]
)
adapter = HTTPAdapter(max_retries=retry)
self.session.mount("https://", adapter)
self.timeout = timeout
def get(self, path: str, params=None):
r = self.session.get(f"{self.base_url}{path}", params=params or {}, timeout=self.timeout)
r.raise_for_status()
return r.json()
def fetch_filings(self, start_date: str, end_date: str) -> pd.DataFrame:
cursor = None
rows = []
while True:
params = {
"from_date": start_date,
"to_date": end_date,
"page_size": 500
}
if cursor:
params["cursor"] = cursor
payload = self.get("/filings", params=params)
batch = payload.get("results", [])
rows.extend(batch)
cursor = payload.get("next_cursor")
if not cursor:
break
return pd.DataFrame(rows)
def normalize_filings(df: pd.DataFrame) -> pd.DataFrame:
out = pd.DataFrame({
"filing_id": df.get("id"),
"issuer_id": df.get("issuer_id"),
"issuer_name": df.get("issuer_name"),
"ticker": df.get("ticker"),
"insider_id": df.get("insider_id"),
"insider_name": df.get("insider_name"),
"insider_role": df.get("insider_role"),
"transaction_date": pd.to_datetime(df.get("transaction_date"), errors="coerce"),
"filing_date": pd.to_datetime(df.get("filing_date"), errors="coerce"),
"transaction_type": df.get("transaction_type"),
"shares": pd.to_numeric(df.get("shares"), errors="coerce"),
"price": pd.to_numeric(df.get("price"), errors="coerce"),
"currency": df.get("currency"),
"security_type": df.get("security_type"),
})
out["value"] = out["shares"] * out["price"]
return out
def build_signal(df: pd.DataFrame, as_of_date: str) -> pd.DataFrame:
as_of = pd.Timestamp(as_of_date)
x = df.copy()
x["transaction_type"] = x["transaction_type"].fillna("").str.lower()
x["insider_role"] = x["insider_role"].fillna("")
x = x[
x["transaction_type"].isin(["buy", "purchase", "acquisition"]) &
x["transaction_date"].between(as_of - pd.Timedelta(days=30), as_of)
]
x["is_senior"] = x["insider_role"].str.lower().str.contains("ceo|chief|cfo|director|chair")
x = x.dropna(subset=["issuer_id", "transaction_date"])
signal = (
x.groupby(["issuer_id", "issuer_name", "ticker"], dropna=False)
.agg(
total_buy_value=("value", "sum"),
distinct_buyers=("insider_id", "nunique"),
senior_buyer_flag=("is_senior", "max"),
last_buy_date=("transaction_date", "max")
)
.reset_index()
)
signal["days_since_last_buy"] = (as_of - signal["last_buy_date"]).dt.days
signal["score"] = (
np.log1p(signal["total_buy_value"].fillna(0))
+ 0.75 * signal["distinct_buyers"].fillna(0)
+ 1.0 * signal["senior_buyer_flag"].astype(int)
- 0.05 * signal["days_since_last_buy"].fillna(30)
)
return signal.sort_values("score", ascending=False)
if __name__ == "__main__":
api = InsiderAPI(base_url="https://api.example.com", api_key="YOUR_API_KEY")
raw = api.fetch_filings(start_date="2026-01-01", end_date="2026-03-31")
filings = normalize_filings(raw)
signal = build_signal(filings, as_of_date="2026-03-31")
signal.to_csv("daily_insider_signal.csv", index=False)
print(signal.head(20))
Esto es suficiente para producir un archivo clasificado. También es suficiente para revelar dónde su esquema difiere de sus suposiciones, que es el verdadero valor educativo de cualquier primer script.
Las próximas actualizaciones son predecibles:
En ese punto, habrá pasado de un tutorial a una herramienta.
Una puntuación simple está bien. Una puntuación no probada es decoración.
Antes de cualquier backtest, ejecute estas comprobaciones:
¿Los recuentos de presentaciones son estables a lo largo del tiempo, o su extracción omitió una semana?
¿Qué proporción de registros tiene ticker, price, shares o transaction_date faltantes?
¿Cuántos ID de presentación o claves de evento duplicados existen?
¿Cuál es la distribución de filing_date - transaction_date?
Si alguno de estos produce sorpresas, posponga sus ambiciones alfa y arregle los datos.
Incluso un estudio de eventos simple es informativo:
Dado que el resumen del artículo no incluía datos de rendimiento ni una extracción de backtest en vivo, no fabricaré ratios de Sharpe para su entretenimiento. La cifra correcta aquí es n/a.
Una vez que la señal de compra básica está en funcionamiento, las siguientes extensiones sensatas son:
Diferenciar entre CEO, CFO, presidente y director no ejecutivo.
Recompensar a múltiples insiders que compran el mismo emisor dentro de un corto período de tiempo.
Escalar el valor de la transacción por el proxy de compensación del insider, la capitalización de mercado o el volumen diario promedio, si está disponible.
Las diferentes jurisdicciones codifican los roles de los insiders, los plazos y las categorías de transacciones de manera diferente. La armonización es un trabajo, no una nota al pie.
Una señal brillante en nombres que no se pueden negociar es un pasatiempo.
| Market | Regulator | Rule | Deadline | Notes |
|---|---|---|---|---|
| FR | AMF | MAR Art 19 | T+3 | Las personas que ejercen responsabilidades de gestión y las personas estrechamente relacionadas deben notificar las transacciones con prontitud, a más tardar en un plazo de tres días hábiles. |
| EU | ESMA y autoridades nacionales competentes | MAR Art 19 | T+3 | El marco principal está armonizado bajo MAR, pero la difusión y el formato difieren según el lugar y la autoridad. |
| US | SEC | Exchange Act Sección 16, Formulario 4 | T+2 | La mayoría de las transacciones de información privilegiada reportables por parte de funcionarios, directores y propietarios del 10 por ciento deben presentarse en un plazo de dos días hábiles. |
Si está consumiendo una API REST en lugar de extraer datos directamente de los sitios web de los reguladores, la calidad de su investigación depende de dos capas de puntualidad:
Esto significa que su modelo debe registrar tanto la fecha del evento regulatorio como la marca de tiempo de disponibilidad observada por la API, siempre que sea posible. Si su endpoint expone created_at, updated_at o published_at, consérvelos todos. El almacenamiento es barato. La falsa precisión en los backtests es costosa.
Un buen tutorial de API debe dejarle con algo operativo, no meramente inspirado. Al final de este flujo de trabajo, su entregable objetivo es simple: un script o notebook que pueda autenticarse, obtener presentaciones de forma incremental, normalizarlas en una tabla estable, calcular una puntuación de convicción de compra transparente y exportar un archivo de señal diario.
Ese no es el modelo final. Es el primero honesto.
El siguiente paso concreto es conectar esta señal a un pequeño backtest de estudio de eventos utilizando las marcas de tiempo de disponibilidad de la presentación en lugar de las fechas de transacción, y luego probar si la compra en grupo por parte de insiders senior agrega poder predictivo incremental después de los filtros básicos de liquidez. Si lo hace, tienes un programa de investigación. Si no, al menos tu paginación seguirá siendo correcta, lo cual es más de lo que se puede decir de muchos esfuerzos más grandes.
Nota editorial: este artículo se elaboró sin investigación web en vivo (Grok no disponible en esta pasada de generación). Las fuentes perennes se citan arriba; las afirmaciones numéricas se extraen de nuestra propia instantánea de la base de datos a partir del 17-05-2026.
Última revisión · 18-05-2026 · Por Simon Azoulay · Fuentes, SEC EDGAR, AMF BDIF y otros 28 reguladores.
Antoine Flamarion bought EUR 3.85m of Tikehau Capital on July 16. Read the filing against the private markets backdrop a...
LVMH rose 2.66% on July 16 as luxury steadied, while the latest insider record stayed quiet after March family buys.
Nuvalent’s CEO filed a EUR 22.1m merger-related Form 4 as GSK closed its $124 cash deal. The real story is the oncology ...
Whitestone REIT’s CEO filed a 1.16 million-share sale as the Ares cash deal closes, with retail REITs still trading on r...
ABC Arbitrage’s July 15 board sales came as European markets slipped and the stock sat near 5.06 EUR, with a 7.90 EUR ta...
Qualys CEO Sumedh S. Thakar sold 3,100 shares in a July 14 10b5-1 plan. Read it against cyber peers, market strength, an...