使用 Python 创建实用的内幕交易申报模型

大多数人真正需要的 API 教程

编写量化教程有两种方式。一种是承诺用 40 行代码构建一个跑赢市场的模型。另一种是展示最先出现故障的部分。我们将选择第二种方式，部分原因在于它更诚实，部分原因在于分页故障虽然不那么光鲜，但却更常见。

本文将介绍一个实用的 Python 工作流程，用于使用 REST API 构建第一个内幕交易申报模型。重点关注以下四点：

1. 干净利落地进行身份验证。
2. 逐页拉取数据，不遗漏任何记录。
3. 将申报数据规范化为可供研究的表格。
4. 计算一个简单的 Alpha 分数，以便进行测试或导出。

由于没有提供实时的文章特定数据提取，我不会假装有。如果某个数字不可用，则标记为 n/a。我们确实拥有的唯一硬性内部数据是申报档案的规模，即 162,000 份申报。这足以立即说明一点：一旦数据集足够大，模糊不清的脚本就不再有吸引力了。

本教程的假设

您了解基本的 Python、requests 和 pandas。您不需要一个完整的因子研究平台。一台笔记本电脑、一个令牌和对 JSON 的轻微容忍度就足够了。

我们将使 API 示例保持通用，因为文章要求的是端点、身份验证、分页和一个简单的 Alpha 示例，而不是对任何一个实现细节进行逆向工程的膜拜。请用您实际的 API 模式替换占位符路径和字段。

最小可行架构

对于研究客户端，请将三个层级分开：

• 传输层，处理 HTTP 请求、重试、身份验证头和分页。
• 规范化层，将原始 JSON 转换为类型化的表格字段。
• 信号层，计算排名、筛选和投资组合候选。

如果您将这些都合并到一个 Jupyter Notebook 单元格中，您会很快得到结果。您也会收到一份由过去的自己编写的未来错误报告。

身份验证和端点设计

用于申报的 REST API 通常会公开这些资源的某种组合：

• /filings
• /issuers
• /insiders
• /transactions
• /symbols 或 /instruments

具体命名有所不同。但模式不变。

Python 中的 API 密钥认证

此类别中的大多数 API 都使用不记名令牌认证或 API 密钥标头。一种简洁的模式是封装 requests.Session，以便每个请求共享标头、超时规则和重试逻辑。

客户端骨架示例

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()

这足以开始。但不足以用于生产。在实践中，您还需要：

• 对 429 和瞬时 5xx 进行重试，
• 记录路径、参数、状态和延迟，
• 可选的退避，
• 一种保存原始有效载荷的一致方法。

一种更稳健的请求方法

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()

枯燥、乏味、有用。一个好的开始。

常见端点参数

申报 API 通常支持以下筛选器：

• from_date
• to_date
• country
• issuer_id
• insider_id
• transaction_type
• page
• page_size
• cursor

如果您的API同时提供偏移分页和游标分页，对于大量拉取数据，请优先选择游标分页。偏移分页很简单，但它也是在下载过程中出现新记录时发现竞态条件的好方法。

不丢失记录的分页

分页是许多研究数据集悄然出错的地方。在一个包含162,000份文件的档案中，遗漏一页不会触发剧烈的异常，它只会让您的回测变得更加虚构。

偏移分页

常见的响应形状如下：

{
  "results": [...],
  "page": 3,
  "page_size": 100,
  "total_pages": 57,
  "total_count": 5678
}

一个直接的循环：

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)

这可行，但有一个注意事项。如果您在分页浏览结果集时添加了新的文件，后续页面可能会发生偏移。原则上，解决方案很简单，但在实践中却常常被忽视：按稳定键排序，并在可能的情况下查询固定的历史窗口。

游标分页

更稳健的响应形状：

{
  "results": [...],
  "next_cursor": "eyJpZCI6..."
}

循环变为：

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)

游标分页通常是批量提取的正确答案。如果您的API支持它，请使用它。

增量同步优于大规模重新下载

对于日常研究，不要每天早上重新拉取完整存档，除非您喜欢浪费带宽和引入故障点。请保持本地存储并增量获取：

• 拉取 filed_at > last_seen_timestamp 的文件，
• 通过唯一文件 ID 进行 upsert，
• 重新查询一个小的重叠窗口，例如最近 2 到 7 天，以捕获更正或延迟更新。

这个重叠窗口很重要。监管机构和发行人并不总是像您的 cron 作业那样准时。

从 JSON 到研究表

原始文件负载尚不是信号。它是一个具有嵌套字段、可选值和足够多边缘情况的行政对象，足以让解析器一直忙碌。

值得首先提取的字段

对于第一个模型，您通常需要：

• filing_id
• issuer_id
• issuer_name
• ticker
• insider_id
• insider_name
• insider_role
• transaction_date
• filing_date
• transaction_type
• shares
• price
• currency
• value
• ownership_nature 如果可用，直接或间接
• security_type

如果 API 未提供 value，请在合理的情况下将其计算为 shares * price，并明确标记缺失的组成部分。

规范化示例

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

这有意保持简洁。重点是建立类型化列和基本诊断，而不是赢得装饰性抽象奖。

最重要的清理规则

一个初步的内幕交易策略通常通过合理的排除而非复杂的加权获得更大的改进。考虑过滤掉：

• 如果您的论点是关于股权信念，则排除非普通股工具，
• 低于最低名义阈值的小额交易，
• 标记为自动或基于计划的交易，
• 如果修订文件与之前的记录重复且您的方案尚未整合它们，则排除修订文件，
• 排除日期缺失或价格不可能的交易。

具体规则取决于司法管辖区和数据源设计。原则是普遍的：如果您无法解释为什么某笔交易属于信号，那么它可能就不应该被纳入。

将多个明细项目分组为一个事件

一份申报文件可能包含多行交易。如果一位高管在一天内分三批买入，将其视为三个独立的信号可能会夸大广度。一个常见的解决方案是聚合到事件级别：

• 按发行人、内部人、交易日期和交易方向分组，
• 汇总股份和价值，
• 保留最早的申报时间戳供策略可见。

这为您提供了一个更清晰的排名单位。

一个简单的内幕交易阿尔法原型

现在是人们跳过来看的部分了。这很公平。

关于内幕交易的文献长期以来一直记载，一些内幕购买，特别是高级管理人员和董事在公开市场上的购买，包含有关未来回报的信息。这种效应并非统一，并非免费，也并非不受拥挤影响。但它是一个简单模型的合理起点。

最简单的可行论点

第一个原型可以是：

根据“信心得分”对近期发行人层面的内部人买入活动进行排名，然后选择排名靠前的公司进行进一步测试。

这并非一个完整的策略，而是一个信号候选。

实用信心得分

对于过去 30 个日历日内的每家发行人，计算：

• 总买入价值，
• 购买的独立内部人数量，
• 是否至少有一名买方是董事或C级高管，
• 最新买入的近期程度。

然后定义一个得分，例如：

score =
    log1p(total_buy_value)
    + 0.75 * distinct_buyers
    + 1.0 * senior_buyer_flag
    - 0.05 * days_since_last_buy

你可以对这些系数提出异议。你应该这样做。关键是先从一些清晰易懂的东西开始。

为什么是买入，而不是卖出

第一步，重点关注买入。卖出噪音较大，因为人们卖出的原因有很多，例如税务、多元化、学费、游艇。买入则需要资金流向困难的方向。市场往往会注意到这一点。

信号代码示例

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)

这会生成一个排名表。它尚未回答信号是否有效的问题。但它回答了你的管道是否能在不即兴发挥的情况下生成每日候选列表。

每日工作流程示例

1. 从API中提取新的申报文件。
2. 规范化并附加到本地存储。
3. 重建最新的滚动30天事件集。
4. 计算发行人层面的分数。
5. 将排名靠前的名称导出到CSV、数据库或下游回测引擎。

这足以创建一个可重复的研究循环。

内部人数据建模中可避免的错误

内部人数据有一种特殊的能力，使其看起来比实际更清晰。申报文件是正式文件，因此人们会感到放心。这种放心应该有所节制。

通过错误的日期造成的未来数据偏差

这是经典的错误。如果您使用 transaction_date 对证券进行排名，但只能在 filing_date 或发布时间戳上观察到申报文件，那么您的回测就是作弊。许多已发布的无稽之谈都是由此开始的。

信号可用性的正确日期是您的系统能够看到申报文件的最早时间戳。根据您的数据源，这可能是：

• 监管机构发布时间、
• 供应商摄取时间、
• 您自己的API的 created_at 或 published_at。

如果您只有申报日期而没有盘中时间，则假定信号最早在下一个交易时段才可交易，除非您的市场惯例和数据源时间支持更精确的判断。

重复和修订的申报文件

申报文件可能会被更正、重新申报或以多行表示。如果您的API公开了修订标志或版本控制，请使用它们。如果没有，请根据发行人、内部人、交易日期、股票数量、价格和方向等保守键进行去重，同时保留原始记录存档。

角色分类混乱

内幕人士角色字段，一个案例中显示为“Officer”，另一个案例中显示为“Chief Financial Officer”，这并非分类，而是一种建议。构建一个映射层。对其进行版本控制。预期会进行修订。

货币与价值可比性

如果您的投资范围涵盖多个市场，则以当地货币计价的名义价值不具直接可比性。如果您的评分使用价值，请使用一致的外汇来源将其转换为基础货币。如果您尚未有可靠的外汇映射，请先在市场内部进行排名。一个适度范围的模型总比一个靠一厢情愿的算术维系的全球模型要好。

紧凑的端到端 Python 示例

下面是一个将各部分联系起来的紧凑脚本。它假定有一个 /filings 端点和一个 bearer token。

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))

这足以生成一个排名文件。它也足以揭示您的模式与您的假设之间的差异，这是任何第一个脚本的真正教育价值所在。

将脚本转化为研究资产

接下来的升级是可预测的：

• 将原始 API 响应保存到磁盘或对象存储中，
• 将规范化表持久化到 DuckDB、Postgres 或 Parquet 中，
• 为日期解析和角色映射添加单元测试，
• 参数化回溯窗口和最小交易规模，
• 添加基准回测层。

至此，您已从教程转向工具。

在信任信号之前需要测试什么

一个简单的分数是好的。一个未经测试的分数只是装饰。

健全性检查

在进行任何回溯测试之前, 请运行以下检查:

覆盖率漂移

申报数量随时间推移是否稳定, 或者您的拉取是否遗漏了一周的数据?

空值率

有多少记录缺少 ticker、price、shares 或 transaction_date?

重复率

存在多少重复的申报 ID 或事件键?

申报滞后

filing_date - transaction_date 的分布是怎样的?

如果其中任何一项产生意外, 请推迟您的 Alpha 目标并修复数据。

首次回溯测试设计

即使是简单的事件研究也能提供信息:

• 根据申报可用日期定义信号形成,
• 每天或每周对发行人进行排名,
• 组建前十百分位或前 N 个投资组合,
• 持有 20、60 或 120 个交易日,
• 与等权重基准或市场指数进行比较。

由于文章简介不包含回报数据或实时回溯测试提取, 我不会为了您的娱乐而编造夏普比率。这里的正确数字是 不适用。

首次原型后的有用改进

一旦基本买入信号运行，接下来合理的扩展是：

角色加权

区分首席执行官、首席财务官、董事长和非执行董事。

聚类买入

奖励在短时间内多名内部人士购买同一发行人的行为。

相对规模

如果可用，根据内部人士薪酬代理、市值或日均交易量来衡量交易价值。

市场特定规则

不同司法管辖区对内部人士角色、截止日期和交易类别的编码方式不同。协调是一项工作，而不是一个脚注。

流动性过滤器

在一个无法交易的股票中发现一个绝妙的信号，那只是一种爱好。

为什么这对API用户尤其重要

如果您是使用REST API而不是直接抓取监管机构网站，您的研究质量取决于两个层面的及时性：

1. 监管备案制度，
2. 供应商或平台摄取管道。

这意味着您的模型应尽可能记录监管事件日期和API观察到的可用时间戳。如果您的端点暴露了created_at、updated_at或published_at，请全部保留。存储成本低廉。回测中的虚假精度代价高昂。

实际收益

一个好的API教程应该让您获得可操作的东西，而不仅仅是受到启发。到此工作流程结束时，您的目标交付物很简单：一个可以进行身份验证、增量获取文件、将其规范化为稳定表格、计算透明的买入信心分数并导出每日信号文件的脚本或笔记本。

这不是最终模型。这是第一个真实的模型。

具体的下一步是将此信号接入一个小型事件研究回测，使用申报可用时间戳而非交易日期，然后测试高级内部人集群购买在基本流动性筛选后是否增加了增量预测能力。如果增加了，你就有了研究计划。如果没有，至少你的分页仍然是正确的，这比许多更宏大的努力要好得多。

编辑注：本文是在没有实时网络研究（Grok 在此生成阶段不可用）的情况下编写的。上述引用了常青来源；数值声明来自我们截至 2026-05-17 的数据库快照。

最后审阅 · 2026-05-18 · 作者 Simon Azoulay · 来源：SEC EDGAR、AMF BDIF 和其他 28 家监管机构。