CEO访谈效应

“该策略在公司CEO接受CNBC等媒体访谈后做空NYSE、AMEX和NASDAQ的相关股票，持有10个交易日，并使用标普500期货对冲，以捕捉市场反应中的异常收益。”

资产类别：期货、股票 | 地区：美国 | 频率：每日 | 市场：股票市场 | 关键词：CEO

I. 策略概述

该策略针对NYSE、AMEX和NASDAQ上市的所有股票，每日筛选前一日接受CNBC采访的公司CEO。具体操作如下：

1. 筛选股票：选择CEO接受CNBC访谈的公司股票。
2. 做空交易：在当天收盘时对这些股票建立空头头寸。
3. 持有周期：空头头寸持有10个交易日。
4. 对冲风险：通过在标普500期货市场建立多头头寸，对冲系统性市场风险。
5. 等权配置：投资组合中的所有股票按等权重分配。

该策略旨在捕捉CEO访谈引发的短期市场反应，剥离市场整体波动的影响，专注于异常回报。

II. 策略合理性

学术研究表明，个体投资者倾向于购买因媒体关注度高而备受瞩目的股票，而内幕或专业投资者则可能利用这一市场行为在价格上涨时卖出这些股票，从中获利。CEO访谈通常具有较高的公众关注度，导致这些股票的价格在短期内偏离基本价值，随后在接下来的交易日内逐步回归。这种价格回归为短期套利策略提供了机会。

通过对访谈后的股票进行做空并对冲市场系统性风险，该策略能够有效隔离CEO访谈带来的异常收益模式。

III. 论文来源

CEO Interviews on CNBC [点击浏览原文]

• 作者：Young Han Kim 和 Felix Meschke
• 机构：成均馆大学、堪萨斯大学金融系

<摘要>

我们研究了媒体关注是否通过个体投资者交易系统性地影响股票价格，利用了6,937次CEO在CNBC访谈中信息内容感知与实际信息内容之间的显著差异。研究发现，股票在[-2, 0]交易日窗口内的平均累计异常回报为1.62%，但在随后的10个交易日内回撤了1.08%。价格反应的幅度与访谈观众人数及CEO语言语调的正面程度正相关。个体投资者和空头交易者是定价模式的关键驱动因素。

这些结果表明，CEO访谈效应不仅是短期定价偏差的来源，还反映了媒体关注对市场行为的显著影响，为套利策略提供了明确的机会。

IV. 回测表现

年化收益率	32.08%
波动率	13.28%
Beta	-0.108
夏普比率	2.11
索提诺比率	-0.331
最大回撤	N/A
胜率	47%

V. 完整python代码

from AlgorithmImports import *
from typing import Dict, List
import json
#endregion
class CEOInterviewsEffect(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2010, 1, 1) # Interviews data starts in December 2006
        self.SetCash(100_000)
        self.UniverseSettings.Leverage = 10
        self.UniverseSettings.Resolution = Resolution.Minute
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Settings.MinimumOrderMarginPortfolioPercentage = 0.0
        self.settings.daily_precise_end_time = False
    
        self.holding_period: int = 10 # Days
        self.selection_flag: bool = False
        self.interviews_data: Dict[date, List[str]] = {}
        self.selected_securities: Dict[str, Symbol] = {}
        self.opened_short_positions_period: Dict[Symbol, int] = {}
        self.universe_tickers: Set(str) = set()
        
        url: str = 'data.quantpedia.com/backtesting_data/index/sp500_ceo_interviews.json'
        response: str = self.Download(url)
        ceo_interviews: List[Dict[str, str]] = json.loads(response)
        
        for interview_data in ceo_interviews:
            date: datetime.date = datetime.strptime(interview_data['date'], '%d.%m.%Y').date()
            self.interviews_data[date] = []
            for ticker in interview_data['tickers']:
                self.universe_tickers.add(ticker)
                self.interviews_data[date].append(ticker)
        self.market: Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
        self.Schedule.On(
            self.DateRules.MonthStart(self.market), 
            self.TimeRules.BeforeMarketClose(self.market), 
            self.Selection)
        self.Schedule.On(
            self.DateRules.EveryDay(self.market), 
            self.TimeRules.BeforeMarketClose(self.market, 1), 
            self.ManageTrade)
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # Rebalance monthly
        if not self.selection_flag:
            return Universe.Unchanged
        self.selection_flag = False
          
        self.selected_securities = {
            f.Symbol.Value: f.Symbol for f in fundamental
            if f.Symbol.Value in self.universe_tickers
        }
        return list(self.selected_securities.values())
        
    def ManageTrade(self) -> None:
        # Liquidate opened symbols
        symbols_to_remove: List[Symbol] = []
        rebalance_flag: bool = False
        
        for symbol in self.opened_short_positions_period:
            holding_period_remaining: int = self.opened_short_positions_period[symbol]
            if holding_period_remaining == 0:
                symbols_to_remove.append(symbol)
            else:
                self.opened_short_positions_period[symbol] -= 1
        
        for symbol in symbols_to_remove:
            rebalance_flag = True
            self.Liquidate(symbol)
            del self.opened_short_positions_period[symbol]
        # Storing symbol of stocks, which CEOs had interview today
        short_symbols: List[Symbol] = [] 
        curr_date: datetime.date = self.Time.date()
        
        if curr_date in self.interviews_data:
            tickers: List[str] = self.interviews_data[curr_date]
            for ticker in tickers:
                if ticker not in self.selected_securities:
                    continue
                rebalance_flag = True
                symbol: Symbol = self.selected_securities[ticker]
                self.opened_short_positions_period[symbol] = self.holding_period
        
        if rebalance_flag:
            opened_shorts: int = len(self.opened_short_positions_period)
            if opened_shorts != 0:                
                # Rebalance whole trade selection alongside the hedge
                for symbol in self.opened_short_positions_period:
                    price: float = self.Securities[symbol].Price
                    if price != 0:
                        self.SetHoldings(symbol, -1 / opened_shorts)
                
                # Hedge with S&P500 future
                self.SetHoldings(self.market, 1)
            else:
                # Liquidate hedge
                self.Liquidate(self.market)
    def Selection(self) -> None:
        self.selection_flag = True
class CustomFeeModel(FeeModel):
    def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
        fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
        return OrderFee(CashAmount(fee, "USD"))