序列化内幕交易

“该策略跟踪纽约证券交易所、纳斯达克、美国证券交易所高管的内幕交易，重点关注非常规序列，在确认后添加股票，持有一个月，并每月重新平衡，以获取市场洞察。”

资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 序列化，内幕交易

I. 策略概要

该策略针对纽约证券交易所、纳斯达克和美国证券交易所的股票，分析内幕交易。每月，内幕交易会被汇总并分类为卖出或买入，不包括“常规”交易者（连续三年在同一月份交易的交易者）。该策略重点关注高管（例如，首席执行官）的交易。内幕交易序列在最后一笔交易后两个月结束，并在添加股票到投资组合之前等待额外一个月进行确认。股票持有一个月，投资组合每月根据新完成的内幕交易序列进行重新平衡，利用内幕信号获取潜在的市场洞察。

II. 策略合理性

研究表明，内部人士掌握有关公司未来前景的非公开信息，从而影响他们的交易时机。内部人士的优势可能会持续较长时间，交易序列反映了逐渐影响股价的私人信息。此类信息需要更长时间才能纳入市场价格，这表明异常回报有利于内部人士。然而，这些回报通常仅在交易序列结束后才能实现，这突显了市场对内部人士驱动的、在较长时间范围内传播的私人信号的延迟反应。

III. 来源论文

内幕交易模式 [点击查看论文]

• 西塞罗，温托基，奥本大学哈伯特商学院，堪萨斯大学商学院

<摘要>

我们重新审视了公司内部人士股票交易的信息内容，期望机会主义的内部人士在拥有长期信息优势时，会将其交易分散在更长的时间内，而在优势短暂的情况下，则会在较短的时间窗口内进行交易。通过控制内部人士交易策略的持续时间，我们发现了强有力的新证据，表明内部人士的卖出和买入都能预测异常股票回报。此外，我们提供的证据表明，内部人士试图通过在收盘后披露其交易来保持其信息优势并增加其交易利润。当内部人士在营业时间后报告其交易时，他们更有可能进行更长时间的交易序列，他们的总体交易份额更大，并且他们的交易与更大的异常回报相关。最后，我们展示了如何通过考虑这些交易模式来优化筛选基于信息进行交易的公司内部人士。

IV. 回测表现

年化回报	22.13%
波动率	14.26%
β值	0.087
夏普比率	1.55
索提诺比率	-0.011
最大回撤	N/A
胜率	59%

V. 完整的 Python 代码

from AlgorithmImports import *
from pandas.core.frame import dataframe
from typing import List, Dict
import pandas as pd
from dateutil.relativedelta import relativedelta
#endregion
class SequencedInsiderTrading(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2014, 1, 1)
        self.SetCash(100000)
        self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
        market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
        self.insider_data: Dict[Symbol, QuiverInsiderTrading] = {}
        self.routine_traders_stocks: List[str] = []
        self.leverage: int = 3
        self.min_consecutive_month_count: int = 2
        self.fundamental_count: int = 500
        self.fundamental_sorting_key = lambda x: x.DollarVolume
        self.selection_flag: bool = False
        self.settings.daily_precise_end_time = False
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
        self.UniverseSettings.Resolution = Resolution.Daily
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(self.leverage)
            symbol: Symbol = security.Symbol
            dataset_symbol: Symbol = self.AddData(QuiverInsiderTrading, symbol).Symbol
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # monthly selection
        if not self.selection_flag:
            return Universe.Unchanged
        selected: List[Fundamental] = [
            x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and \
            x.SecurityReference.ExchangeId in self.exchange_codes
        ]
        
        if len(selected) > self.fundamental_count:
            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
        
        selected: Dict[str, Symbol] = {x.Symbol.Value: x.Symbol for x in selected}
        aggregated_data: Dict[Tuple[datetime.date, str], float] = {}
        curr_month: int = self.Time
        insiders: Dict[str, List[Tuple[datetime.date, str, float]]] = {}
        # aggregate data on monthly format
        for name, data in self.insider_data.items():
            if name not in insiders:
                insiders[name] = []
            for data_point in data:
                date: str = data_point[0].strftime('%m-%Y')
                key = (date, data_point[1])
                if key in aggregated_data:
                    if data_point[2] is not None:
                        aggregated_data[key] += data_point[2]
                else:
                    aggregated_data[key] = data_point[2]
            insiders[name].append([(key[0], key[1], shares) for key, shares in aggregated_data.items()])
            aggregated_data.clear()
        # check for routine insider trades
        months_to_check: List[str] = [(curr_month - relativedelta(months=i)).strftime('%m-%Y') for i in [1, 2, 13, 14, 25, 26, 37, 38]]
        for name, data in insiders.items():
            ticker: str = data[0][0][1]
            if len(data[0]) > self.min_consecutive_month_count:
                if (data[0][-1][0] == months_to_check[0] and data[0][-2][0] == months_to_check[1] and \
                    all(i[0] != month for month in months_to_check[2:] for i in data[0])) and \
                    ticker in selected:
                    self.routine_traders_stocks.append(selected[ticker])
        
        return [x for x in selected.values()]
    def OnData(self, data: Slice) -> None:
        for insider_trades in data.Get(QuiverInsiderTrading).values():
            for insider_trade in insider_trades:
                if insider_trade.Name not in self.insider_data:
                    self.insider_data[insider_trade.Name] = []           
                self.insider_data[insider_trade.Name].append((insider_trade.Time, insider_trade.Symbol.Value, insider_trade.Shares))
    
        # monthly rebalance
        if not self.selection_flag:
            return
        self.selection_flag = False
        
        targets: List[PortfolioTarget] = []
        for symbol in self.routine_traders_stocks:
            if symbol in data and data[symbol]:
                targets.append(PortfolioTarget(symbol, 1 / len(self.routine_traders_stocks)))
    
        self.SetHoldings(targets, True)
        self.routine_traders_stocks.clear()
    def Selection(self) -> None:
        self.selection_flag = True
# Custom fee model.
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
        fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))