How to Analyze Insider Transactions with Python: A CEO Buying Case Study

import pandas as pd
import numpy as np
import requests
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import seaborn as sns

import json
import re
import numpy as np
import pandas as pd
import requests
from scipy.optimize import linear_sum_assignment

def fetch_stocks(filters, cap):
    api_key = 'YOUR EODHD API KEY'
    base_url = 'https://eodhd.com/api/screener'
    all_stocks = []
    for i in range(0,500,100):
        params = {
            "api_token": api_key,
            "filters": json.dumps(filters),
            "sort": "market_capitalization.desc",
            "limit": 100,
            "offset": i}
        resp = requests.get(base_url, params = params).json()
        stocks = list(pd.DataFrame(resp['data'])['code'])
        all_stocks.append(stocks)
    all_stocks = [item for sublist in all_stocks for item in sublist]
    df = pd.DataFrame(columns = ['ticker', f'cap'])
    df.ticker, df.cap = all_stocks, cap
    df = df.sample(n = 250, random_state = 42)
    return df

micro_filters = [
    ["exchange", "=", "us"],
    ["market_capitalization", ">=", 50_000_000],
    ["market_capitalization", "<", 300_000_000]
]

small_filters = [
    ["exchange", "=", "NYSE"],
    ["market_capitalization", ">=", 300_000_000],
    ["market_capitalization", "<", 2_000_000_000]
]

mid_filters = [
    ["exchange", "=", "NYSE"],
    ["market_capitalization", ">=", 2_000_000_000],
    ["market_capitalization", "<", 10_000_000_000]
]

large_filters = [
    ["exchange", "=", "NYSE"],
    ["market_capitalization", ">=", 10_000_000_000]
]

micro_stocks = fetch_stocks(micro_filters, 'micro_cap')
small_stocks = fetch_stocks(small_filters, 'small_cap')
mid_stocks = fetch_stocks(mid_filters, 'mid_cap')
large_stocks = fetch_stocks(large_filters, 'large_cap')

frames = [micro_stocks, small_stocks, mid_stocks, large_stocks]
stocks_1000 = pd.concat(frames, ignore_index = True)
stocks_1000 = stocks_1000.sample(frac = 1, random_state = 42).reset_index(drop = True)
stocks_1000

def fetch_ceo_purchases(ticker):
    try:
        api_key = 'YOUR EODHD API KEY'

        all_form4 = []

        for i in range(0,1000,100):
            form4_url = f'https://eodhd.com/api/sec-filings/{ticker}/form4?api_token={api_key}&page[limit]=100&page[offset]={i}'
            resp = requests.get(form4_url).json()['data']
            all_form4.append(resp)

        all_form4 = [item for sublist in all_form4 for item in sublist]

        all_purchases = []

        ceo_pattern = re.compile(
            r'\bceo\b|chief executive officer|co-chief executive officer|co-ceo|chief exec officer',
            re.IGNORECASE
        )

        for filing in all_form4:
            footnote_map = {
                footnote['footnote_id']: footnote['text']
                for footnote in filing.get('footnotes', [])
            }

            for transaction in filing.get('non_derivative', []):
                officer_title = transaction.get('officer_title') or ''
                security_title = transaction.get('security_title') or ''
                shares_amount = transaction.get('shares_amount')
                price_per_share = transaction.get('price_per_share')

                is_ceo = bool(ceo_pattern.search(officer_title))

                is_purchase = (
                    transaction.get('is_officer') is True
                    and is_ceo
                    and transaction.get('transaction_code') == 'P'
                    and transaction.get('acquired_or_disposed') == 'A'
                    and shares_amount is not None
                    and shares_amount > 0
                    and price_per_share is not None
                    and price_per_share > 0
                    and 'common stock' in security_title.lower()
                )

                if not is_purchase:
                    continue

该函数以每批100条的形式请求文件，并将返回的页面数据合并为一个列表。随后会检查每条非衍生品交易是否符合CEO购买规则。

CEO头衔的验证使用了正则表达式，因为文件中并未采用完全一致的头衔格式。CEO可能以"CEO"、"Chief Executive Officer"或"Co-CEO"等形式出现，因此仅匹配单一字符串会导致有效记录被遗漏。

我们还保留了关联脚注信息。交易代码P具有参考价值，但单独使用无法完整说明情况。脚注可能揭示该交易涉及交易计划、发行活动或其他需要特别关注的安排。

我仅对宇宙中前500只证券执行了这一步骤，因为Form 4接口会快速消耗API配额。相同的循环也可以扩展到全部1000只股票以获取更大样本量。

收集完数据后，我们将数据集限定在2022年1月1日至2025年12月31日之间提交的文件。

这些仍然是原始申报记录，而非独立的CEO购买信号。当不同批次的股票以不同价格买入时，一次购买可能被拆分到多行记录中。下一步需要将这些碎片合并为每日购买事件。

## 将Form 4记录转换为每日购买事件

CEO可能在同一天以多个价格购买股票。申报文件可能将每个价格区间记录为独立行，尽管这些行属于同一次更大规模的购买。如果将每行都视为独立信号，那么交易频繁的购买日可能会因为被拆分为更多价格层级而获得不成比例的权重。

因此下一步需要将具有相同股票代码、CEO、申报文件和交易日期的行进行分组。

对每个分组，我们将汇总股票数量和总购买金额，并计算加权平均价格：

加权平均价格 = 总购买价值 / 总购买股数

以下代码块执行完整聚合操作，为每个每日CEO购买事件生成一行数据。

purchase_value 列为每个原始数据行赋予聚合前的美元价值。当数据行完成分组后，这些值可以相加求和，而不会丢失不同购买价格的影响。

transaction_rows 列有助于检查数据压缩的程度。数值为1表示该日事件原本就只有一行数据。数值为5表示五个独立的申报行被合并为一个购买事件。

聚合操作将数据集从625条原始交易记录减少到535个每日购买事件。

这种差异不仅仅是数据整理工作。它将分析单位从"一个申报价格区块"转变为"一个CEO购买日"，这更接近我们试图研究的经济事件。

尽管如此，我们仍无法计算回报率。购买事件只告诉我们CEO进行了购买，但无法说明股票是否接近高位、略有下跌，或已深度回调。接下来，我们将添加每个申报公开时可获取的价格背景信息。

## 添加历史价格和回撤数据

在本测试中，只有当我们知道股票当时的交易价格时，CEO的购买行为才变得有意义。

在年高5%下方进行的购买与股票下跌40%后的购买有着显著差异。因此，我们现在需要将历史价格数据引入工作流程，并衡量每只股票在申报公开时距离近期高点的跌幅。

我们将使用调整后收盘价而非原始收盘价，因为调整后价格考虑了股票拆分和股息等事件。这能为我们提供更一致的价格序列，便于跨时间比较。

我们从2021年至2025年获取价格数据，尽管购买分析从2022年开始。额外增加一年数据是因为2022年的首批观测值仍需要足够早的数据来计算一年期高点。

以下代码块从EODHD的历史EOD接口获取CEO购买数据集中所有股票的每日历史价格，并将它们合并到一个数据框中。

def fetch_historical_eod(ticker):
    api_key = 'YOUR EODHD API KEY'
    historical_url = f'https://eodhd.com/api/eod/{ticker}?from=2021-01-01&to=2025-12-31&period=d&api_token={api_key}&fmt=json'
    historical_resp = requests.get(historical_url).json()
    historical_filtered = []
    for item in historical_resp:
        item['ticker'] = ticker
        keys = ['ticker', 'date', 'adjusted_close']
        item = {key: item.get(key) for key in keys}
        historical_filtered.append(item)
    return historical_filtered

for ticker in tickers:
    try:
        historical_eod = fetch_historical_eod(ticker)
        historical_eod_entries.extend(historical_eod)
        print(f'{ticker} done')
    except:
        print(f'{ticker} error')

historical_df = pd.DataFrame(historical_eod_entries)
historical_df['date'] = pd.to_datetime(historical_df.date)

historical_df['rolling_high_252'] = (historical_df.groupby('ticker')['adjusted_close'].transform
                                     (lambda prices: prices.rolling(window=252, min_periods=200).max()))

historical_df['drawdown'] = (historical_df['adjusted_close']/ historical_df['rolling_high_252']- 1)
historical_df['drawdown_pct'] = (historical_df['drawdown'] * 100)

cp_df.filed_at = pd.to_datetime(cp_df.filed_at)

price_columns = ['ticker', 'date', 'adjusted_close', 'rolling_high_252', 'drawdown', 'drawdown_pct']

analysis_df = pd.merge_asof(
    cp_df.sort_values(['filed_at', 'ticker']),
    historical_df[price_columns].sort_values(['date', 'ticker']),
    by='ticker',
    left_on='filed_at',
    right_on='date',
    direction='backward',
    allow_exact_matches=False
)

analysis_df = analysis_df.rename(columns={'date': 'price_date'})

purchase_events = analysis_df.dropna(subset=["drawdown"])
purchase_events.filed_at = pd.to_datetime(purchase_events.filed_at)
purchase_events = purchase_events.sort_values(["ticker", "reporting_owner_cik", "filed_at"])
purchase_events["days_since_previous"] = purchase_events.groupby(["ticker", "reporting_owner_cik"])["filed_at"].diff().dt.days
purchase_events["new_episode"] = purchase_events["days_since_previous"].isna() | (purchase_events["days_since_previous"] > 28)
purchase_events["episode_id"] = purchase_events.groupby(["ticker", "reporting_owner_cik"])["new_episode"].cumsum()

episodes = purchase_events.groupby(["ticker", "reporting_owner_cik", "reporting_owner_name", "episode_id"], 
                               as_index=False).agg(first_filing_date=("filed_at", "min"), 
                                                   last_filing_date=("filed_at", "max"),
                                                   first_transaction_date=("transaction_date", "min"),
                                                   total_shares=("shares_purchased", "sum"),
                                                   total_purchase_value=("total_purchase_value", "sum"),
                                                   purchase_days=("filed_at", "nunique"),
                                                   transaction_events=("filed_at", "size"),
                                                   initial_drawdown=("drawdown", "first"), 
                                                   initial_drawdown_pct=("drawdown_pct", "first"))

episodes_20 = episodes[episodes["initial_drawdown"] <= -0.20].copy()

print("所有购买事件:", len(episodes))
print("20%回撤事件:", len(episodes_20))
print("涉及的股票数量:", episodes_20["ticker"].nunique())

episodes_20

episodes_20 = episodes_20.reset_index(drop = True)
episodes_20['first_filing_date'] = pd.to_datetime(episodes_20['first_filing_date'])
historical_df['date'] = pd.to_datetime(historical_df['date'])

price_map 是一个字典，每个股票代码对应其自身的日期和调整后收盘价数据框。

例如，price_map['AAT.US'] 仅包含 AAT.US 的历史价格数据，且已按从最早到最新的顺序排序。

### 确定买入日期并计算未来收益

现在我们可以编写一个函数来处理单次购买事件。

该函数将执行以下操作：

- 定位股票的价格历史数据
- 找到首次提交日期之后的第一个交易日
- 将该日的调整后收盘价作为买入价格
- 向前推算 21、63、126 和 252 个交易日
- 在每个时间窗口计算收益

生成的数据框包含事件详情、买入日期、买入价格以及每个时间窗口的未来收益：

### 汇总原始收益数据

查看单个事件很有用，但我们也需要一个完整的样本概览。

对于每个时间窗口，我们将计算：

- 可用观测值的数量
- 平均收益
- 中位数收益
- 收益为正的百分比

初步来看，结果似乎很有希望。三个月后的平均收益达到 11.9%，十二个月后达到 35.4%。大多数十二个月的观测值也都是正收益。

但这也正是容易得出错误结论的地方。

这些数字告诉我们的是 CEO 购买之后发生了什么。它们没有告诉我们这些表现中有多少是来自于 CEO 的购买行为本身。

这些股票在购买时已经下跌了至少 20%。其中一些可能只是因为被低估的股票有时会反弹。

为了区分这两种效应，我们需要构建一个对照组，该对照组由相似回撤日期组成，且这些日期附近没有发生CEO购买行为。

## 构建无购买对照组

原始回报表看似令人鼓舞，但仍然将所有信用归于CEO的购买行为。

这并不是一个公平的测试。样本中的每只股票已经下跌了至少20%，而被重创的股票即使没有内幕交易也可能反弹。我们需要将每一起CEO购买事件与另一个日期进行比较，该日期同一只股票面临相似压力，但附近没有发生CEO购买。

有效的对照组必须满足六个规则：

- 相同股票代码
- 相同日历年
- 回撤幅度在5个百分点以内
- 不超过180个日历日
- 购买事件前后28天内无CEO购买
- 每个样本仅使用一次

每个CEO购买事件也仅匹配一次。

### 创建对照候选组

我们将从2022年至2025年期间，找出每只股票价格至少低于其过去252天最高价20%的每个交易日。

不过这里存在一个问题。股票可能在数月内持续低于该阈值。如果我们保留所有交易日，一次长期下跌可能生成数百个几乎相同的对照候选。

为避免这种情况，我们将每个连续回撤期划分为28天的区块，并从每个区块中保留一个候选。

below_20标记了通过回撤阈值的日期。

drawdown_segment则将连续下跌区间与其他区间区分开来。如果股票价格回升至阈值以上，之后再次跌破阈值，这将形成新的区间。

在每个区间内，anchor_block从回撤开始的日期计算28天窗口。每个窗口保留一行数据，这样可以得到一组可管理的日期，而不会将同一轮下跌中的每个交易日都视为独立事件。

这些日期只是潜在的对照组候选。我们仍需剔除那些接近CEO购买的日期。

### 剔除靠近CEO购买的日期

无购买对照组应真正与内幕信号分离。

例如，距离CEO申报日三天的回撤日期将是一个糟糕的对照组。交易可能已经发生，而申报可能尚未公开。

因此，我们收集事件数据集中所有CEO购买申报日期，而不仅仅是通过20%阈值的137起事件。

现在，我们需要为每个候选日期找到其之前最近的购买申报和之后最近的购买申报。

两个 merge_asof() 操作可以完成此任务。第一个操作向后搜索，第二个操作向前搜索。

现在每个候选日期都知晓其两侧最近的CEO购买日期。

我们可以计算这些申报之间的距离，并仅保留与两侧最近的CEO购买日期相隔超过28个日历日的日期。

缺少之前的或之后的申报是可以接受的。这仅仅意味着在候选日期的该侧数据集中没有对应的CEO购买记录。

此时，每个剩余行代表一个日期，满足以下条件：

- 股票价格下跌至少20%
- 该日期不属于任何CEO购买事件的邻近日期范围
- 股票具有足够的历史价格数据用于计算回撤

### 将购买事件与对照组匹配

现在进入实际的匹配过程。

我们首先为每个CEO购买事件和每个对照组候选对象分配唯一的标识符。同时提取日历年份，因为匹配必须来自同一股票和年份。

匹配过程在每个股票-年份组内独立进行。

假设CEO在股票下跌32%时进行了购买，我们将在同一股票和年份中寻找一个无购买记录的日期，该日期的回撤接近32%，同时日期间隔不超过180个日历日。

只有满足以下条件时，配对才是有效的：

- 回撤差异 <= 0.05
- 日历间隔 <= 180 天

接下来的代码块构建可能的配对，并使用 linear_sum_assignment() 选择一对一的匹配集。

selected_controls = control_candidates[['control_id', 'ticker', 'date', 'adjusted_close', 'drawdown', 'drawdown_pct']].rename(columns={'ticker': 'control_ticker',
                                                                                                                                   'date': 'control_date',
                                                                                                                                   'adjusted_close': 'control_signal_price',
                                                                                                                                   'drawdown': 'control_drawdown',
                                                                                                                                   'drawdown_pct': 'control_drawdown_pct'})

matched_sample = matched_pairs.merge(purchase_pool,on='purchase_id',how='left').merge(selected_controls,on='control_id',how='left')
matched_sample

matched_sample['first_filing_date'] = pd.to_datetime(matched_sample['first_filing_date'])
matched_sample['control_date'] = pd.to_datetime(matched_sample['control_date'])
historical_df['date'] = pd.to_datetime(historical_df['date'])

prices = historical_df[['ticker', 'date', 'adjusted_close']].dropna().sort_values(['ticker', 'date'])
price_map = {ticker: group.reset_index(drop=True) for ticker, group in prices.groupby('ticker')}

def get_forward_returns(ticker, signal_date):
    result = {
        'entry_date': pd.NaT,
        'entry_price': np.nan,
        'return_1m': np.nan,
        'return_3m': np.nan,
        'return_6m': np.nan,
        'return_12m': np.nan
    }

    ticker_prices = price_map.get(ticker)

    if ticker_prices is None or pd.isna(signal_date):
        return pd.Series(result)

    dates = ticker_prices['date'].to_numpy(dtype='datetime64[ns]')
    entry_index = np.searchsorted(dates, np.datetime64(signal_date), side='right')

    if entry_index >= len(ticker_prices):
        return pd.Series(result)

    entry_price = ticker_prices.loc[entry_index, 'adjusted_close']
    result['entry_date'] = ticker_prices.loc[entry_index, 'date']
    result['entry_price'] = entry_price

    for label, days in {'1m': 21, '3m': 63, '6m': 126, '12m': 252}.items():
        target_index = entry_index + days

        if target_index < len(ticker_prices):
            target_price = ticker_prices.loc[target_index, 'adjusted_close']
            result[f'return_{label}'] = target_price / entry_price - 1

    return pd.Series(result)

purchase_returns = matched_sample.apply(lambda row: get_forward_returns(row['ticker'], row['first_filing_date']), axis=1).add_prefix('purchase_')
control_returns = matched_sample.apply(lambda row: get_forward_returns(row['control_ticker'], row['control_date']), axis=1).add_prefix('control_')
matched_returns = pd.concat([
    matched_sample.reset_index(drop=True),
    purchase_returns.reset_index(drop=True),
    control_returns.reset_index(drop=True)], axis=1)

comparison = []

for horizon in ['1m', '3m', '6m', '12m']:
    purchase_col = f'purchase_return_{horizon}'
    control_col = f'control_return_{horizon}'
    valid = matched_returns[[purchase_col, control_col]].dropna()
    differences = valid[purchase_col] - valid[control_col]

    comparison.append({
        'horizon': horizon,
        'matched_pairs': len(valid),
        'purchase_mean': valid[purchase_col].mean(),
        'control_mean': valid[control_col].mean(),
        'mean_difference': differences.mean(),
        'purchase_median': valid[purchase_col].median(),
        'control_median': valid[control_col].median(),
        'median_difference': differences.median(),
        'purchase_positive_rate': (valid[purchase_col] > 0).mean(),
        'control_positive_rate': (valid[control_col] > 0).mean(),
        'purchase_win_rate': (valid[purchase_col] > valid[control_col]).mean()
    })

comparison_df = pd.DataFrame(comparison)
comparison_df