“该策略交易纽约证券交易所 (NYSE)、美国证券交易所 (AMEX) 和纳斯达克 (NASDAQ) 的股票，做多具有正内部需求 (NID) 的赢家股票，做空沉默的输家股票，采用等权重配置，持有期为12个月，并进行每月再平衡。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 动量，内部交易
I. 策略概要
该策略针对纽约证券交易所 (NYSE)、美国证券交易所 (AMEX) 和纳斯达克 (NASDAQ) 的股票，但排除股价低于 5 美元、市值处于纽约证券交易所最低十分位、或账面权益缺失/非正值的股票。每月，投资者计算净内部需求 (NID)，即过去六个月的内部购买量减去销售量，并归一化为流通股数。股票根据六个月收益率分为“赢家”（前十分位）和“输家”（后十分位）投资组合，并进一步按内部交易活动分为“买入”（NID 为正）、“卖出”（NID 为非正值）和“沉默”（无交易活动）类别。该策略做多具有正 NID 的赢家股票，并做空处于沉默投资组合中的输家股票。所有头寸采用等权重配置，持有期为 12 个月，并按 1/12 比例进行每月再平衡。
II. 策略合理性
学术研究表明，只有当内部交易与过去的回报一致时，短期动量才会持续——积极的内部活动支持赢家，而消极的活动证实输家。这种动量源于投资者对内部交易信息反应不足。然而，由于潜在的监管和诉讼风险，内部人士很谨慎，尤其是对于销售而言，这造成了他们行为的不对称性。当掌握负面私人信息并预期价格大幅下跌时，内部人士通常会避免出售，以避免法律审查。相反，当他们掌握有关公司的正面信息时，他们更有可能进行交易，从而加强了观察到的内部活动与动量驱动的股票表现之间的联系。
III. 来源论文
动量与内部交易 [点击查看论文]
- 马庆忠，康奈尔大学
<摘要>
短期动量和长期反转都归因于投资者对先前的内部交易信息反应不足。只有当过去的赢家（输家）的内部交易活动表明正面（负面）信息时，它们才能在短期内继续获得显著的正面（负面）回报。因此，短期动量归因于投资者对证实过去回报的内部信息反应不足。在长期内，只有当过去的赢家（输家）的内部交易活动表明负面（正面）信息时，它们才能获得显著的负面（正面）回报。因此，长期反转归因于投资者对否定过去回报的内部信息反应不足。在控制了内部交易信息之后，没有过度反应的证据。此外，促成动量的股票和促成反转的股票之间存在明显的“分工”。
IV. 回测表现
| 年化回报 | 15.45% |
| 波动率 | N/A |
| β值 | -0.534 |
| 夏普比率 | N/A |
| 索提诺比率 | N/A |
| 最大回撤 | N/A |
| 胜率 | 28% |
V. 完整的 Python 代码
from AlgorithmImports import *
import pandas as pd
from io import StringIO
from numpy import floor
#endregion
class MomentumCombinedInsiderTrading(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
# NOTE: We use only s&p 100 stocks so it's possible to fetch short interest data from quandl.
self.symbols = [
'AAPL','MSFT','AMZN','FB','BRKB','GOOGL','GOOG','JPM','JNJ','V','PG','XOM','UNH','BAC','MA','T','DIS','INTC','HD','VZ','MRK',
'PFE','CVX','KO','CMCSA','CSCO','PEP','WFC','C','BA','ADBE','WMT','CRM','MCD','MDT','BMY','ABT','NVDA','NFLX','AMGN','PM','PYPL',
'TMO','COST','ABBV','ACN','HON','NKE','UNP','UTX','NEE','IBM','TXN','AVGO','LLY','ORCL','LIN','SBUX','AMT','LMT','GE','MMM','DHR',
'QCOM','CVS','MO','LOW','FIS','AXP','BKNG','UPS','GILD','CHTR','CAT','MDLZ','GS','USB','CI','ANTM','BDX','TJX','ADP','TFC','CME',
'SPGI','COP','INTU','ISRG','CB','SO','D','FISV','PNC','DUK','SYK','ZTS','MS','RTN','AGN','BLK'
]
self.period = 6 * 21
# Trenching
self.holding_period = 12
self.managed_queue = []
# dataframe with insider trades for every stock.
self.insiders_trading = {}
# Create custom universe.
self.selection_flag = False
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.shares_outstanding = {}
for symbol in self.symbols:
# Import insiders trading data.
csv_string_file = self.Download(f'data.quantpedia.com/backtesting_data/economic/insiders_trading/{symbol}.csv')
if csv_string_file == "": continue
parser = lambda x: pd.datetime.strptime(x, "%Y-%m-%d")
self.insiders_trading[symbol] = pd.read_csv(StringIO(csv_string_file), sep=';', parse_dates=['Tran.Date'], date_parser=parser)
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverseSelection(FineFundamentalUniverseSelectionModel(self.CoarseSelectionFunction, self.FineSelectionFunction))
self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(10)
def CoarseSelectionFunction(self, coarse):
if not self.selection_flag:
return Universe.Unchanged
return [Symbol.Create(x, SecurityType.Equity, Market.USA) for x in self.symbols]
def FineSelectionFunction(self, fine):
fine = [x for x in fine if x.EarningReports.BasicAverageShares.ThreeMonths > 0 and x.Symbol.Value in self.insiders_trading]
symbols = [x.Symbol for x in fine]
history = self.History(symbols, self.period, Resolution.Daily)
if history.empty:
self.Log(f'Empty history request for {len(symbols)} symbols')
return Universe.Unchanged
history = history.close.unstack(0)
last_prices = {}
performance = {}
for symbol in symbols:
if symbol in history:
closes = history[symbol]
if len(closes) == self.period:
performance[symbol] = closes[-1] / closes[0] - 1
last_prices[symbol] = closes[-1]
# Stock which have not been traded last 6 months.
silence = []
# Traded stocks.
nid = {}
for stock in fine:
symbol = stock.Symbol
# Get number of buys and sells during last 6 months.
ticker = symbol.Value
buys = [row['Shares'] for index, row in self.insiders_trading[ticker].iterrows() if row['Symbol'] == ticker and row['Tran.Date'] >= (self.Time - timedelta(days = 6 * 30)) and row['Tran.Date'] <= self.Time and row['Action'] == 'B']
sells = [row['Shares'] for index, row in self.insiders_trading[ticker].iterrows() if row['Symbol'] == ticker and row['Tran.Date'] >= (self.Time - timedelta(days = 6 * 30)) and row['Tran.Date'] <= self.Time and row['Action'] == 'S']
total_buy_shares = sum(buys)
total_sell_shares = sum(sells)
if len(buys) != 0 or len(sells) != 0:
nid[symbol] = (total_buy_shares - total_sell_shares) / stock.EarningReports.BasicAverageShares.ThreeMonths
else:
# Stock was not traded during last 6 months.
silence.append(symbol)
decile = int(len(performance) / 10)
sorted_by_performance = [x[0] for x in sorted(performance.items(), key=lambda item: item[1])]
winners = sorted_by_performance[-decile:]
losers = sorted_by_performance[:decile]
# long = [x[0] for x in performance.items() if x[1] > 0 and x[0] in nid and nid[x[0]] > 0]
# short = [x[0] for x in performance.items() if x[1] < 0 and x[0] in silence]
# Each month investor goes long past winners with a positive NID and goes short past losers from “silence” portfolio.
long = [x for x in winners if x in nid and nid[x] > 0]
short = [x for x in losers if x in silence]
if len(long) != 0:
long_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
# symbol/quantity collection
long_symbol_q = [(x, floor(long_w / last_prices[x])) for x in long]
else:
long_symbol_q = []
if len(short) != 0:
short_w = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
# symbol/quantity collection
short_symbol_q = [(x, -floor(short_w / last_prices[x])) for x in short]
else:
short_symbol_q = []
self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
return long + short
def OnData(self, data):
if not self.selection_flag:
return
self.selection_flag = False
# Trade execution
remove_item = None
# Rebalance portfolio
for item in self.managed_queue:
if item.holding_period == self.holding_period + 1: # Each month investor goes long past winners with a positive NID and goes short past losers from “silence” portfolio.
# Liquidate
for symbol, quantity in item.symbol_q:
self.MarketOrder(symbol, -quantity)
remove_item = item
elif item.holding_period == 1: # Each month investor goes long past winners with a positive NID and goes short past losers from “silence” portfolio.
open_symbol_q = []
for symbol, quantity in item.symbol_q:
if symbol in data and data[symbol]:
self.MarketOrder(symbol, quantity)
open_symbol_q.append((symbol, quantity))
# Only opened orders will be closed
item.symbol_q = open_symbol_q
item.holding_period += 1
# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
if remove_item:
self.managed_queue.remove(remove_item)
def Selection(self):
self.selection_flag = True
class RebalanceQueueItem():
def __init__(self, symbol_q):
# symbol/quantity collections
self.symbol_q = symbol_q
self.holding_period = 0
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))