当一家公司的股价大幅下跌后，首席执行官却选择购买该公司的股票，人们很容易将这一行为解读为一种对该公司前景的信任。毕竟，负责经营企业的人比普通投资者了解得更多，因此这种交易似乎是一个值得关注的信号。
但这里存在一个明显的问题：即使没有内部人士进行任何买卖操作，那些股价下跌了20%或更多的股票往往也会出现反弹。如果我们只关注首席执行官购买股票之后的情况，就可能会将这种反弹归因于“内部人士发出的买入信号”，而实际上这种反弹在许多被市场看淡的股票中本来就是常见的现象。
在这个教程中，我们将构建一个Python工作流程来彻底验证这一观点。我们会提取Form 4文件中的交易记录，筛选出首席执行官的购买行为，将重复出现的交易记录合并为有意义的数据事件，添加历史股价数据，计算股价跌幅和后续收益，然后将这些购买数据与同一只股票在类似日期没有发生任何买卖行为的记录进行对比分析。
有趣的地方不仅仅在于最终的收益统计结果，更重要的是整个分析过程——我们将把那些杂乱无章的监管文件整理成一套可以用来进行公平比较的数据集。在这个过程中，我们需要处理重复的交易记录、同一位首席执行官多次进行的购买行为、交易日期的对齐问题、不完整的价格历史数据，以及如何确保各种数据之间的匹配关系。
最终，我们将得到一个完整的事件分析工作流程，而得到的结论也会比“首席执行官买入股票后股价上涨”这种简单的描述要有意义得多。
目录
先决条件
学习本教程并不需要具备高级的金融或量化分析背景，掌握Python以及pandas的基本知识就足够了。
在开始之前，请确保您已经满足以下要求：
-
已在本地安装了Python，或者能够使用Jupyter Notebook、Google Colab等笔记本环境。
-
对数据框、函数、循环以及API请求有基本的了解。
-
拥有EODHD API密钥，以便访问筛选工具、Form 4文件以及历史数据端点。
-
拥有足够的API调用额度，以便处理您想要分析的股票数量。
完整的案例分析会使用500只证券的Form 4数据。如果您想先了解代码逻辑而不必进行大量的API调用，也可以先在较小的样本集上运行这个流程。
学习本教程不需要预先掌握事件研究或控制组匹配的相关知识。我们会根据分析过程中的实际需要，逐步讲解这些内容。
导入所需包
完成整个分析流程只需要一小部分包。requests用于处理API调用，pandas和NumPy负责大部分数据处理工作，而SciPy则提供了后续分析中所需的匹配算法。
import json
import re
import numpy as np
import pandas as pd
import requests
from scipy.optimize import linear_sum_assignment
以上就是所有的设置内容。我们只导入了分析过程中真正需要的包，没有添加任何多余的库或工具。在将这些包导入环境之前，请务必使用pip进行安装。
构建股票样本库
在获取内部文件信息之前，我们首先需要确定要搜索的公司列表。
我们不会从某个特定的市值区间开始分析，而是会涵盖微型股、小型股、中型股和大型股，这样可以让分析结果更具多样性，避免某个市场板块占据样本的主导地位。
市值划分标准如下：
-
micro_cap：5000万美元至3亿美元 -
small_cap：3亿美元至20亿美元 -
mid_cap：20亿美元至100亿美元 -
large_cap：100亿美元及以上
对于每个市值区间，我们会获取500条筛选结果，随机选取250条，最终组成一个包含1000只股票的样本库。
def fetch_stocksfilters, 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
注意：请将YOUR EODHD API KEY替换为你的实际EODHD API密钥。如果你还没有这个密钥，可以通过打开EODHD开发者账户来获取它。
该筛选工具每次请求最多返回100条数据，因此循环会分五批处理前500条结果。
接下来我们会从这些候选股票中随机选取250只。由于使用了固定的随机种子，因此这种选取方式是可重复的，重新运行程序也会得到相同的样本结果。之后，我们会根据四个市值筛选条件分别运行相应的函数。
最终得到的数据框包含1000只股票的信息，其中每个类别有250只股票。
现在需要说明一点：微型股筛选条件使用的是较宽范围的us交易市场设置，而其他类别则使用NYSE交易市场设置。虽然这个样本被用于案例分析，但它并不能完全代表整个美国股市的情况。
获取CEO购买记录并应用日期筛选条件
一旦股票池准备就绪，我们就可以使用EODHD的内部交易信息API来搜索Form 4文件中关于CEO购买股票的记录。
Form 4文件包含的信息远不止简单的内部购买行为。一份文件中可能包括出售、奖励、期权行权、衍生品交易等内容，因此我们不能简单地下载所有文件并将每条记录都视为购买信号。
对于这次分析来说，一条交易记录必须满足以下所有条件：
-
必须被归类为非衍生品交易
-
必须由公司高管进行报告
-
相关高管的职位名称必须明确表示其为CEO
-
交易代码必须是
P -
交易的对象必须是股票
-
股票数量和交易价格都必须为正数
-
涉及的必须是普通股
我们还会保留交易日期和文件提交日期。交易日期告诉我们CEO是在什么时候购买股票的，而文件提交日期则说明外部投资者何时能够了解到这一交易信息。在后续的分析中，文件提交日期将被用作关键参考日期。
接下来的代码块负责完成整个数据提取过程。它定义了筛选函数，对前500只股票应用该函数，并将所有符合条件的记录合并到一个数据框中。
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
linked_footnotes = ' '.join(
footnote_map.get(footnote_id, '')
for footnote_id in transaction.get('footnote_ids', [])
)
all_purchases.append({
'ticker': ticker,
'accession_number': filing['accession_number'],
'filed_at': filing['filed_at'],
'transaction_date': transaction['transaction_date'],
'reporting_owner_cik': transaction['reporting_owner_cik'],
'reporting_owner_name': transaction['reporting_owner_name'],
'officer_title': officer_title,
'security_title': security_title,
'shares_amount': shares_amount,
'price_per_share': price_per_share,
'total_value': transaction.get('total_value'),
'sharesOwned_after': transaction.get('shares Owned_after'),
'footnotes': linked_footnotes
})
return all_purchases
except:
return None
all_ceo_purchases = []
for ticker in stocks_1000.ticker[:500]:
ticker = ticker + '.US'
ceo_purchases = fetch_ceo_purchases(ticker)
if ceo_purchases:
all_ceo_purchases.extend(ceo_purchases)
print(f'{len(ceo_purchases)}条CEO购买记录在股票{ticker}中被找到')
else:
print(f'在股票{ticker}中未找到任何交易记录')
cp_df = pd.DataFrame(all_ceo_purchases)
cp_df.to_csv('ceo_purchases.csv')
该函数以100条为批次请求相关文件，并将返回的页面内容合并成一条列表。随后，它会逐一检查每一笔非衍生交易，看是否符合CEO购买的相关规定。
在检测CEO的头衔时，我们使用了正则表达式，因为这些文件中使用的CEO头衔并不完全一致。例如，CEO可能会被标记为CEO、Chief Executive Officer或Co-CEO，因此如果只匹配某个特定的字符串，就会遗漏一些有效的记录。
我们还会保留那些带有链接的脚注。虽然交易代码P具有一定的参考价值，但它本身并不能揭示全部信息。有时候，脚注会指出某笔购买交易实际上涉及某种交易计划、募股活动或其他需要进一步关注的安排。
由于Form 4接口可能会很快消耗API信用额度，所以我只对数据库中前500只股票进行了这项处理。当然，这个流程也可以扩展到全部1,000只股票，从而获得更全面的样本数据。
在收集到所有相关数据后，我们会将数据集限制在2022年初至2025年底期间提交的文件范围内。
cp_df = cp_df[cp_df["filed_at"].between("2022-01-01","2025-12-31")]
cp_df.tail()
这些数据目前还只是原始的记录，并不能直接被视为独立的CEO购买信号。例如，当以不同的价格购买了不同数量的股票时，同一笔交易可能会被记录在多条记录中。下一步的任务就是将这些分散的记录整合成每日的购买事件。
将Form 4记录中的数据转换为每日购买事件
CEO可能在同一天内以不同的价格购买股票。相关文件会将每个交易价格段分别记录为一条记录，尽管这些记录实际上属于同一笔交易。如果我们将每一条记录都视为一个独立的信号，那么在一个交易量较大的日子里，由于记录条目较多，相应的分析结果可能会被赋予过高的权重。
因此，下一步就是将那些具有相同股票代码、CEO姓名、提交时间和交易日期的记录合并起来。
对于每一组记录，我们会计算总共购买了多少股股票以及总购买金额，然后计算出加权平均价格：
加权平均价格 = 总购买金额 / 总购买股数
以下代码段完成了全部的汇总工作，最终生成了每笔每日CEO购买事件对应的记录。
cp_df['purchase_value'] = (
cp_df['shares_amount'] * cp_df['price_per_share']
)
group_columns = [
'ticker',
'reporting_owner_cik',
'reporting_owner_name',
'officer_title',
'accession_number',
'filed_at',
'transaction_date'
]
daily_events = (
cp_df.groupby(
group_columns,
as_index=False,
dropna=False
)
.agg(
shares_purchased=('shares_amount', 'sum'),
total_purchase_value>('purchase_value', 'sum'),
transaction_rows=('shares_amount', 'size'),
sharesOwned_after=('shares Owned_after', 'max')
)
)
daily_events['weighted_average_price'] = (
daily_events['total_purchase_value']
/ daily_events['shares_purchased']
)
daily_events.filed_at = pd.to_datetime(daily_events.filed_at)
daily_events.to_csv('ceo_purchases_grouped.csv')
daily_events.tail()
purchase_value这一列在数据汇总之前，为每一条原始记录赋予了一个美元数值。一旦这些记录被分组，对这些数值进行求和运算时，不同的购买价格所带来的影响就不会丢失。
transaction_rows这一列有助于我们了解数据在合并过程中发生了哪些变化。当值为1时，表示该日的交易记录原本就只显示为一条记录；而当值为5时，则说明有五条独立的交易记录被合并成了一条购买记录。
经过汇总处理后，这个数据集中的原始交易记录数量从625条减少到了535条每日购买记录。
这种变化并非仅仅是形式上的调整。它实际上将分析的单位从“一个报告的价格区间”转变为“CEO的一次购买行为”，这样就更接近我们想要研究的经济现象本身了。
不过，我们现在还没有准备好计算投资回报。虽然购买记录能够告诉我们CEO进行了哪些交易，但它们并不能说明当时股票的价格是处于高位、略有下跌，还是已经处于持续下跌的状态。接下来，我们需要添加每条交易记录公开时对应的股价信息。
添加历史价格与跌幅数据
只有当我们知道CEO购买股票时的市场价格时，这些购买记录才会变得有研究价值。
我们会使用调整后的收盘价而不是原始的收盘价，因为调整后的价格已经考虑到了股票分割、分红等因素的影响。这样一来，我们就可以得到一个更加一致的价格序列，以便进行跨时间段的比较分析。
tickers = list(cp_df.ticker.unique())
historical_eod_entries = []
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}
historicalFiltered.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} 已获取完成')
except:
print(f'{ticker} 数据获取失败')这段代码用于获取历史数据，为每个股票代码在每个交易日生成一条价格记录。接下来，我们会计算这些数据的滚动最高价以及回撤幅度。
计算滚动最高价与回撤幅度
对于每一个交易日，我们都需要了解在前252个交易日内该股票所达到的最高调整后收盘价。这大约相当于一年的交易时间。
回撤幅度的计算公式如下：
回撤幅度 = 调整后收盘价 / 过去252天的最高价 – 1
如果回撤幅度的值为-0.20，说明该股票的当前价格比其过去252天的最高价低了20%；如果值为-0.35，则意味着当前价格比这一最高价低了35%。
接下来的代码段会针对每只股票整理其价格历史数据，计算滚动最高价，并将结果分别以小数形式和百分比形式呈现出来。
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)
其中min_periods=200这个参数需要稍作解释。
一个完整的滚动窗口包含252个交易日，但如果严格要求必须使用这252个数据点来进行计算，就会丢失很多早期的交易数据。因此允许在200个交易日后再进行计算，可以在保证仍然拥有足够历史数据的同时，提高计算的灵活性。
那些缺乏足够历史数据的股票记录会被忽略，而不会得到不准确的回撤幅度估计值。
将每次购买行为与最新的可用价格对应起来
现在我们需要为每一笔CEO进行的股票购买交易添加价格信息。
虽然文件提交日期可以作为参考日期，但我们不能使用同一天内的收盘价。因为Form 4表格可以在交易时段之前、期间或之后提交，所以在文件提交时可能还不知道当天的收盘价是多少。
因此，我们应该使用文件提交日期之前的最后一个完整交易日的价格作为参考值。
接下来的代码段会使用merge_asof()函数来实现这一匹配操作。与普通的合并操作不同，这个函数可以将每一笔交易记录与最近的一个有效价格对应起来，而不要求两个日期完全相同。
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'})
关键的设置是 allow_exact_matches=False。这个设置可以防止日期为3月10日的交易记录使用当天收盘价；合并操作会使用3月10日之前的最后一个交易日的数据。
合并后的数据框现在看起来如下：
合并后的数据框现在包含了以下信息：
-
price_date：用于匹配的交易日 -
adjusted_close：当天的股票价格 -
rolling_high_252：过去一年中的最高价 -
drawdown：以小数形式表示的跌幅 -
drawdown pct：以百分比形式表示的跌幅
例如，drawdown_pct 的值为 -32.4，意味着在提交报告之前的最后一个交易日，该股票的股价比过去252天中的最高价低了32.4%。
现在我们既知道这位CEO进行了购买操作，也知道了在这笔交易公开时，该股票的股价已经下跌了多少。
下一个需要解决的问题是重复购买行为。如果一位CEO在几周内多次进行购买操作，这些操作不应该被视作多个独立的交易信号。
将购买事件合并为同一“购买阶段”
目前，数据集中每位CEO的每次购买操作都对应一行记录。虽然这比直接使用原始的Form 4表格要好，但仍然有可能重复计算同一项决策所导致的交易行为。
假设一位CEO在周一购买了股票，接下来的一周又买了同样数量的股票，再过两周又再次进行购买——从技术上讲，这些属于三次独立的购买事件；但从经济角度来看，它们可能实际上是一次连续的购买行动。
如果将这三次购买都视为独立的交易信号，那么那些频繁购股的CEO就会比那些一次性完成相同购买的CEO获得更高的权重。因此，在计算投资回报之前，我们需要将相邻的几次购买行为合并为同一个“购买阶段”。
规则很简单：只要同一位CEO在连续不超过28个自然日的时间内对同一只股票进行了多次购买操作，这些交易就应被视作同一个“购买阶段”。
首先，我们会剔除那些没有可供计算的跌幅数据的购买记录，然后按照股票代码、CEO名称和提交报告的日期对剩余的数据进行排序，并计算从上一次购买至今已经过去了多少天。
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()
对于每一组“股票代码-CEO姓名”的组合来说，第一次购买行为会自动开启一个新的记录阶段，因为此前并没有任何记录可供对比。
此后，只有当两次购买的间隔超过28天时，才会开始新的记录阶段；间隔在28天或更短时间内的购买行为仍会被归入同一个记录阶段中。
new_episode这个数值的累积值为每一组购买行为分配了一个唯一的标识符。
既然每一笔购买行为都属于某个特定的记录阶段，我们就可以将这些事件汇总成每组购买行为对应的一行数据。
对于每一个记录阶段来说，我们会保留该阶段的首次和最后一次交易记录日期，统计购买的股票数量及总金额，计算购买行为的频率，并记录从首次交易日期开始的资产净值下降幅度。
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"))
这里有两种用于统计购买行为频率的指标：
-
purchase_days用于记录该记录阶段中不同的交易记录日期的数量。 -
transaction_events用于统计被归为一组的每日购买事件的数量。
“购买的股票数量”及“总购买金额”这两个指标涵盖了整个记录阶段；而“资产净值下降幅度”则仅从首次交易日期开始计算，因为信号正是从这一时刻开始的。
这个细节对于应用20%的筛选标准来说非常重要。
假设某个记录阶段的开始时，股票的价格已经比其最高价低了18%，而后CEO在资产净值下降到24%时再次进行了购买。如果将这种情况称为“在股价下跌20%之后开始的记录阶段”，那就会产生误导。
因此，我们首先对数据进行处理分组，然后再使用initial_drawdown这个指标来应用筛选条件。
episodes_20 = episodes[episodes["initial_drawdown"] <= -0.20].copy()
print("所有购买记录阶段的数量:", len(episodes))
print("资产净值下降幅度达到20%的记录阶段数量:", len(episodes_20))
print("这些记录阶段所涉及的股票代码数量:", episodes_20["ticker"].nunique())
episodes_20
我们现在将这些重复发生的购买交易记录归纳为137次独立的“CEO购买交易行为”；这些交易都是在股票价格至少低于其最近252天最高价的20%时发生的。
这些交易记录就是我们将长期关注的真实数据。接下来，我们会在这些初始文件提交后的第一个交易日开始观察，并测量在随后的一个月、三个月、六个月以及十二个月内发生了什么变化。
计算CEO购买行为后的收益
我们现在共有137次“CEO购买交易行为”，而这些交易都是在股票价格至少低于其最近252天最高价的20%时发生的。
下一个问题很直接：在这些文件公开之后，发生了什么变化呢？
我们将把这些交易行为发生后的第一个交易日作为观察起始日期。这样的设定使得测试更加符合实际情况——因为外部投资者在Form 4文件公布之前是无法进行任何操作的，而选择下一个交易日作为起始日期也可以涵盖那些在周末或市场假期提交的文件。
收益计算的时间跨度如下：
-
1个月：21个交易日
-
3个月：63个交易日
-
6个月：126个交易日
-
12个月：252个交易日
按股票代码整理价格历史数据
在计算收益之前，我们需要为每一只股票创建一个按时间顺序排列的价格序列。
这样，收益计算函数就可以直接找到正确的起始日期，然后依次计算后续一段时间内的收益情况，而无需反复遍历整个历史数据表格。
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'])
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')]
price_map是一个字典，其中每个股票代码都对应着一个包含该股票价格及调整后收盘价的数据表格。
例如，price_map['AAT.US']中只包含了AAT.US的股票价格数据，并且这些数据已经按照时间顺序进行了排序。
确定起始日期并计算后续收益
现在我们可以编写一个函数，用来逐一分析每一次购买交易行为。
这个函数将会完成以下操作：
-
查找该股票的价格历史数据
-
找出文件提交日期之后的第一个交易日
-
将这一天的调整后收盘价作为起始计算价格
-
依次向前推进21个、63个、126个和252个交易日
-
计算每个时间点上的收益情况
def calculate_forward_returns(row):
ticker_prices = price_map.get(row['ticker'])
if ticker_prices is None:
return pd.Series dtype='object')
dates = ticker_prices['date'].to_numpy(dtype='datetime64[ns')]
entry_index = np.searchsorted(dates, np.datetime64(row['first_filing_date]), side='right')
if entry_index >= len(ticker_prices):
return pd.Series dtype='object')
entry_date = ticker_prices.loc[entry_index, 'date']
entry_price = ticker_prices.loc[entry_index, 'adjusted_close']
result = {'entry_date': entry_date, 'entry_price': entry_price}
horizons = {'1m': 21, '3m': 63, '6m': 126, '12m': 252}
for label, days in horizons.items():
target_index = entry_index + days
if target_index < len(ticker_prices): target_price = ticker_prices.loc[target_index, 'adjusted_close'] result[f'date_{label}'] = ticker_prices.loc[target_index, 'date'] result[f'return_{label}'] = target_price / entry_price - 1 else: result[f'date_{label}'] = pd.NaT result[f'return_{label}'] = np.nan return pd.Series(result) forward_returns = episodes_20.apply(calculate_forward_returns, axis=1) episodereturns = pd.concat([episodes_20.reset_index(drop=True), forward_returns], axis=1) episode_returns
最终得到的数据框中包含了各阶段的详细信息，包括入场日期、入场价格以及各时间点的未来收益情况：
汇总原始收益数据
虽然分析单个阶段的数据很有意义，但我们也需要一个能够整体反映整个样本情况的视图。
对于每个时间阶段，我们将计算以下指标：
-
可用的观测数据数量
-
平均收益
-
中位数收益
-
收益为正的观测值所占的比例
summary = []
for horizon in ['1m', '3m', '6m', '12m']:
returns = episode_returns[f'return_{horizon}'].dropna()
summary.append({
'horizon': horizon,
'observations': len(returns),
'mean_return': returns.mean(),
'median_return': returns.median(),
'positive_rate': (returns > 0).mean()
})
summary_df = pd.DataFrame(summary)
summary_df
乍一看，这些结果似乎相当不错。三个月后的平均收益达到了11.9%，而十二个月后的平均收益则达到了35.4%；此外，大多数在十二个月的观测数据中也显示出了正收益。
<但恰恰在这里，人们很容易得出错误的结论。
这些数据告诉我们，在CEO们进行股票回购之后发生了什么，但它们并不能说明其中有多少业绩提升是直接由这些回购行为带来的。
这些股票的股价原本就已经下跌了至少20%；其中一些股票后来可能有所反弹，因为被严重压低的股价有时确实会恢复元气。
为了区分这两种效应，我们需要建立一个对照组——这个对照组的形成日期应该与那些发生CEO股票回购的日期相近，但在这些日期并没有进行任何回购操作。
建立无回购行为的对照组
初步分析的结果看起来令人满意，但实际上这些数据仍将所有业绩提升都归功于CEO们的回购行为。
这样的测试并不公平。样本中的每一只股票股价原本就已经下跌了至少20%，而被压低的股价有时也会在没有内部人士干预的情况下出现反弹。因此，我们需要将每一次CEO股票回购行为与另一个日期进行对比——在那个日期，同一只股票也面临着类似的压力，但却没有发生任何回购操作。
一个有效的对照组必须满足以下六项要求：
-
相同的股票代码
-
同一个日历年度
-
股价跌幅在5个百分点以内
-
时间间隔不超过180个日历日
-
在回购发生前28天内及发生后28天内都没有进行任何回购操作
-
每个对照组只对应一次回购行为
每一次CEO股票回购行为也只应与一个对应的对照组进行对比。
创建对照组候选对象
我们首先会找出2022年至2025年期间，所有那些股价跌幅至少达到20%的交易日。
不过这里有一个问题：某只股票的股价可能会连续数月处于这一下跌区间内。如果我们将所有这样的交易日都纳入统计范围，那么一次长时间的股价下跌就可能会导致出现数百个几乎完全相同的对照组候选对象。
为避免这种情况，我们需要将每一个连续的股价下跌周期划分为28天的小段，然后从每个小段中选取一个作为对照组候选对象。
hist = historical_df[['ticker', 'date', 'adjusted_close', 'rolling_high_252', 'drawdown', 'drawdown_pct')).dropna(subset=['drawdown'])
hist['date'] = pd.to_datetime(hist['date'])
hist = hist[hist['date'].between('2022-01-01', '2025-12-31')].sort_values(['ticker', 'date'])
hist['below_20'] = hist['drawdown'] <= -0.20
hist['previous_below_20'] = hist.groupby('ticker')['below_20'].shift().fillna(False)
hist['new_state'] = hist['below_20'].ne(hist['previous_below_20'])
hist['drawdown_segment'] = hist.groupby('ticker')[『new_state』].cumsum()
controlcandidates = hist[hist['below_20')).copy()
control_candidates['segment_start'] = controlcandidates.groupby(['ticker', 'drawdown_segment'])['date'].transform('min')
controlcandidates['anchor_block'] = ((controlcandidates['date'] - controlandidates['segment_start']).dt.days // 28)
controlcandidates = controlcandidates.sort_values(['ticker', 'date']).drop_duplicates(['ticker', 'drawdown_segment', 'anchor_block'])
control_candidates = controlcandidates.reset_index(drop = True)
controlcandidates
below_20用于标记那些超过了回撤阈值的日期。
drawdown_segment的作用是将连续的下跌区间区分开来。如果某只股票的股价先回升到阈值以上，随后又再次跌回该阈值以下，那么这一段下跌就会被视为一个新的区间。
在每个区间内，anchor_block会从回撤开始的那天起计算28天的时间窗口。通过为每个时间段保留一行数据，我们可以方便地管理这些日期信息，而不会将同一下跌过程中的所有交易都视为独立的事件。
不过，这些日期仅仅只是潜在的参考依据。我们仍然需要剔除那些发生在CEO进行股票购买附近的日期。
剔除CEO购买附近的日期
作为对照用的日期，应该与内部人士的交易信号真正区分开来。
例如，在CEO提交相关交易文件前三天发生的回撤事件，并不能作为一个有效的对照数据。因为那时该交易可能已经完成，只是相关的文件尚未公布而已。
因此，我们在数据集中收集了所有CEO的购买记录日期，而不仅仅是那些回撤幅度超过了20%的137个案例。
purchase_dates = analysis_df[['ticker', 'filed_at]].dropna().drop_duplicates()
purchase_dates['filed_at'] = pd.to_datetime(purchase_dates['filed_at'])
purchase_dates = purchase_dates.rename(columns={'filed_at': 'purchase_date'})
purchase_dates
对于每一个候选日期，我们现在需要找到它之前最近的购买记录日期，以及它之后最近的购买记录日期。
通过两次merge_asof()操作就可以实现这一目标。第一次操作是向后查找，第二次操作则是向前查找。
controlcandidates = pd.merge_asof(
control_candidates.sort_values('date'),
purchase_dates.sort_values('purchase_date'),
by='ticker',
left_on='date',
right_on='purchase_date',
direction='backward'
)
controlcandidates = control_characters.rename(columns={'purchase_date': 'previous_purchase_date'}
controlcandidates = pd.merge_asof(
control_candidates.sort_values('date'),
purchase_dates.sort_values('purchase_date'),
by='ticker',
left_on='date',
right_on='purchase_date',
direction='forward'
)
controlcandidates = control_characters.rename(columns={'purchase_date': 'next_purchase_date'}
现在，每个候选日期都明确了它之前和之后最近的CEO购买记录日期。
我们可以计算这些购买记录日期与候选日期之间的距离，并且只保留那些距离超过28个日历日的日期。
days_from_previous = (controlcandidates['date'] - control_candidates['previous_purchase_date']).dt.days
days_to_next = (control_characters['next.purchase_date'] - control_characters['date']).dt.days
far_from.previous = control_characters['previous_purchase_date'].isna() | (days_from_previous > 28)
far_from_next = control_characters['next_purchase_date'].isna() | (days_to_next > 28)
controlcandidates = control_characters[far_from_previous & far_from_next]
control_candidates
如果缺少之前的或之后的记录，那也没关系。这仅仅意味着在现有的数据集中，该候选人在相关时间段内没有进行过CEO购买操作。
此时，每一行数据都代表这样一个日期：
-
在这一天，该股票的股价跌幅至少为20%；
-
并且这个日期并不属于CEO购买操作发生的相邻时间段内；
-
同时，这个日期也有足够的历史价格数据，可以用于计算股价的下跌幅度。
通过控制条件匹配购买事件
现在开始进行实际的匹配操作了。
首先，我们会为每一个CEO购买事件以及每一个对照候选对象分配一个唯一的标识符。同时，我们还会提取出这些事件的年份信息，因为只有同一股票、同一年份内的数据才能进行匹配。
purchase_pool = episodes_20.reset_index(drop = True)
purchase_pool['first_filing_date'] = pd.to_datetime(purchase_pool['first_filing_date'])
purchase_pool['year'] = purchase_pool['first_filing_date'].dt.year
purchase_pool['purchase_id'] = np.arange(len(purchase_pool))
controlcandidates['year'] = control_candidates['date'].dt.year
controlandidates['control_id'] = np.arange(len(controlcandidates))
匹配操作是在每个股票代码-年份组合内部分别进行的。
假设某位CEO在股价下跌了32%的时候进行了购买操作，那么我们就会在同一股票、同一年份内寻找那些股价跌幅也接近32%的日期，并且确保这些日期之间的时间间隔不超过180天。
只有满足以下条件时，一对数据才被认为是有效的：
-
股价下跌幅度之差 ≤ 0.05
-
时间间隔 ≤ 180天
接下来，我们会构建所有可能的匹配对，并使用linear_sum_assignment()函数来选出那些一一对应的有效匹配结果。
matches = []
max_drawdown_gap = 0.05
max_calendar_gap = 180
for (ticker, year), purchases in purchase_pool.groupby(['ticker', 'year]):
controls = controlcandidates[(control_candidates['ticker'] == ticker) & (controlcandidates['year'] == year)].copy()
if controls.empty:
continue
purchase_drawdowns = purchases['initial_drawdown'].to_numpy()[:, None]
control_drawdowns = controls['drawdown'].to_numpy()[None, :]
drawdown_cost = np.abs(purchase_drawdowns - control_drawdowns)
purchase_dates = purchases['first_filing_date'].to_numpy(dtype='datetime64[D')]
control_dates = controls['date'].to_numpy dtype='datetime64[D']
calendar_gap = np.abs((purchase_dates[:, None] - control_dates[None, :]).astype('timedelta64[D]').astype(int))
valid = (drawdown_cost &le max_drawdown_gap) & (calendar_gap <= max_calendar_gap)
if not valid.any():
continue
cost = drawdown_cost + calendar_gap / 1000000
cost[~valid] = 1000000
row_indices, columnIndices = linear_sum_assignment(cost)
keep = cost[row_indices, column Indices] < 1000000
selected = pd.DataFrame({
'purchase_id': purchases.iloc[row_indices[keep]]['purchase_id'].to_numpy(),
'control_id': controls.iloc[column_indices[keep]]['control_id'].to_numpy(),
'drawdown_gap': drawdown_cost[row indices[keep], column indices[keep]],
'calendar_gap_days': calendar_gap[row indices[keep], column indices[keep]]
})
matches.append(selected)
matched_pairs = pd.concat(matches, ignore_index=True)
核心思想其实比代码看起来要简单得多。
drawdown_cost这个指标用于衡量两次资产回撤之间的差距。如果在-0.32时进行买入操作，而在-0.34时进行对照操作，那么两者之间的差异就是0.02，也就是两个百分点。
calendar_distance这个因素被用作一个非常次要的判断依据。虽然资产回撤程度的相似性仍是首要考虑的因素，但当两个对照组之间的差异很小时，就会优先选择日期更接近的那个对照组。
linear_sum_assignment()这个函数可以确保同一个对照组不会被分配到多个买入操作中。它会寻找一组一对一的匹配关系，使得整个组合的成本降到最低。
构建最终的匹配数据集
目前的匹配结果中只包含了买入操作的ID、对照组的ID以及距离数值。
<最后一步就是将原始的买入记录和对照组信息重新合并到这些匹配对中，这样我们就可以计算出双方的收益情况了。
selected-controls = controlcandidates[['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',
'drawdownpct': '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
现在，每一行数据都包含了一次CEO的买入操作以及与之匹配的、没有发生买入操作的资产回撤日期：
现在我们终于得到了最初想要得到的两组数据：一是CEO在资产大幅回撤后进行的买入操作；二是同一只股票在类似回撤幅度发生时，但没有CEO进行买入操作的记录。
将CEO的买入操作与类似的、没有买入操作的资产回撤情况进行对比
这就是这个工作流程真正发挥作用的地方了。
matched_sample中的每一行数据都包含了同一只股票的两个日期：
-
CEO进行买入操作的第一个申报日期
-
发生类似资产回撤但此时没有CEO进行买入操作的日期
从这一点开始，双方必须得到完全相同的对待。首席执行官一方应在提交申请后的第一个交易日开始参与交易；而控制权一方则应在匹配后的削减日期之后的第一个交易日开始参与交易。双方都应使用相同的价格调整标准，并且采用相同的回报计算周期。
首先，我们需要准备那两列包含信号日期的数据，然后重新构建之前使用过的股票代码级别的价格走势图。
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')]
这个价格走势图为每个股票代码都提供了按时间顺序排列的价格历史数据。这样一来，我们就可以使用同一个函数来计算购买日期和对照日期对应的收益，而无需为每一组数据分别编写相应的逻辑代码。
从任意信号日期开始计算未来收益
下一个函数只需要两个输入参数：一个股票代码和一个信号日期。
该函数会找到该日期之后的第一个交易日，以当天的收盘价作为买入价格，然后计算21天、63天、126天以及252天后的收益情况。
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)
其中关键的参数是side='right'。这个设置可以确保任何一组数据都不会在自身的信号日期进行交易操作：CEO购买股票后的收益计算会从文件提交日期之后开始，而对照组的收益计算则会从对应的回撤日期之后开始。
如果某个股票代码的数据缺失，或者某个时间范围内的未来价格数据不足，那么该指标的值就会显示为NaN。
对两组数据应用相同的收益计算逻辑
现在，我们需要对每一组匹配的数据对分别运行这个函数两次。
第一次运行时使用CEO购买股票时的代码和日期；第二次运行时则使用对照组的代码和日期。返回的结果列前都会加上相应的前缀，这样就能轻松区分这两组数据了。
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: getforward_returns(row['control_ticker'], row['control_date}], axis=1).add_prefix('control_')
matched_returns = pd.concat([
matched_sample.reset_index(drop=True),
purchasereturns.reset_index.drop=True),
control_returns.reset_index.drop=True), axis=1)
最终得到的数据框将这两种结果并排展示：
-
CEO购买行为的成交日期及价格
-
对照组的成交日期及价格
-
CEO购买行为带来的收益
-
对照组的收益
并不是每一对数据在所有时间范围内都适用。只有当双方都有足够的未来价格数据时，这些数据才能被使用。因此，当我们观察的时间范围延长到12个月时，可用的观测数据数量就会减少。
构建最终对比结果
最后一步是在每个时间点上比较这两组收益数据。
我们将计算以下指标：
-
每组的平均收益
-
每组的中位数收益
-
匹配对内的平均收益差异
-
匹配对内的中位数收益差异
-
正收益率
-
CEO购买行为表现优于对照组的数据对所占的比例
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
在这里，这些成对的统计数据非常重要。
median_difference表示的是：
CEO购买行为带来的收益与对照组收益之间的中位数差异
对于每一对数据来说，这个数值并不简单等于CEO购买组的收益中位数减去对照组的收益中位数。
而“胜率”则直接反映了这样一个问题：在所有匹配的数据对中，有多少对的实际表现更好？
案例研究得出的结论
-
一个月的时间里，这一策略并没有显示出任何优势。
在平均收益、中位数收益差、正回报率以及胜率这些指标上，CEO购买股票策略的表现均低于对照组。
-
三个月的时间跨度是这一策略表现最好的时期。
实施CEO购买策略的组别平均回报率比对照组高出6.2个百分点，并且在59.6%的情况下取得了胜利。这是唯一一个所有指标都呈现出一致方向的观察周期。
-
六个月和十二个月的时间跨度则不太可靠。
虽然实施CEO购买策略的组别的平均回报率更高，但中位数收益差为负值，而且大多数个别案例中，这一策略的表现仍然不如对照组。
-
股价的下跌幅度本身就可以解释很多问题。
那些股价已经大幅下跌的股票，往往即使没有CEO购买股票这一策略，其价格也会出现反弹。因此，原始数据所显示的购买后的回报率其实高估了这一策略的实际效果。
这项测试能说明什么，不能说明什么
<这可能看起来没有“证明首席执行官的购股行为预示着市场会复苏”那么令人兴奋，但实际上这是一个更好的答案。这种工作流程迫使我们将自己想要相信的观点与一个基准数据进行对比进行测试，而这个基准数据的变化最终改变了我们的结论。>