盘中全市场上涨/下跌与收益

“投资范围包括在纽约证券交易所、美国证券交易所或纳斯达克上市的股票。盘中涨跌信号（UDS）衡量的是上午9:45前上涨或下跌股票之间的差异。”

资产类别: 差价合约、ETF、期货 | 地区: 美国 | 周期: 日内 | 市场: 股票 | 关键词: 涨跌

I. 策略概要

投资范围包括CRSP中股票代码为10或11，并在纽约证券交易所、美国证券交易所或纳斯达克上市的股票。对于这些股票，盘中涨跌信号（UDS）通过衡量在t日早上9:45之前上涨或下跌的股票数量来计算。UDS的计算方法是：自前一天收盘以来上涨股票数量与下跌股票数量之差，除以样本中股票总数。该信号用于衡量当天整体市场动量。

II. 策略合理性

该研究表明，盘中策略的回报不能用系统性风险或贝塔来解释，而是由个体投资者活动和盘中情绪驱动的。散户投资者受近期市场表现和短期情绪影响，在整个交易日内推动市场向同一方向发展。这两个盘中信号反映了这些情绪来源，并对回报产生显著的积极影响。研究结果表明，这种模式不仅存在于中国等新兴市场，也存在于美国等成熟市场，尽管交易限制和投资者行为存在差异。结果表明，盘中情绪信号可以预测这两个市场随后的盘中回报。

III. 来源论文

Intraday Market-Wide Ups/Downs and Returns [点击查看论文]

• 张伟、林深、张永杰，天津大学管理与经济学部，天津大学管理与经济学部；清华大学五道口金融学院，天津大学

<摘要>

本研究利用16年的中国股市数据和3年的美国股市数据，探讨了盘中早期全市场涨跌幅对同一交易日后续盘中回报的解释力。与前一交易日收盘价相比，我们引入了两个盘中全市场涨跌指标，即指数回报和每分钟上涨与下跌股票数量的比例差异。时间序列分析表明，盘中指标与市场指数的后续盘中回报之间存在显著的（无论是经济上还是统计上）正相关关系。利用这种盘中关系的盘中交易策略在中国市场产生了4.1%的月回报，在美国市场产生了2.8%的月回报。此外，对于散户投资者活跃度高（即交易价值高、每笔交易量低、小盘股、高市净率、低机构持股比例、低股价和股东数量多）的市场，这些策略的盈利能力更强。结果表明，早期交易中简单的盘中全市场涨跌幅会影响散户投资者的情绪，导致市场在交易日内朝同一方向波动。

IV. 回测表现

年化回报	23.75%
波动率	31.56%
β值	0.009
夏普比率	0.63
索提诺比率	-0.176
最大回撤	N/A
胜率	50%

V. 完整的 Python 代码

from AlgorithmImports import *
from typing import List, Dict
from dataclasses import dataclass
# endregion
class IntradayMarketWideUpsDowns(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2010, 1, 1)
        self.SetCash(100000)
        
        self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']	
        data: Equity = self.AddEquity('SPY', Resolution.Minute)
        data.SetFeeModel(CustomFeeModel())
        self.symbol: Symbol = data.Symbol
        
        self.data: Dict[Symbol, SymbolData] = {} # Storing SymbolData for each stocks
        self.selected_stocks: List[Symbol] = []
        
        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.Minute
        self.AddUniverse(self.FundamentalSelectionFunction)
        self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
            
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # Rebalance monthly
        if not self.selection_flag:
            return Universe.Unchanged
        self.selection_flag = False
        
        # Filter universe
        selected: List[Fundamental] = [
            x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' 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]]    
            
        # Add filtered stocks in self.data dictionary
        for stock in selected:
            symbol: Symbol = stock.Symbol
            self.data[symbol] = SymbolData()
        
        # Store selected stocks symbols in self.selected_stocks parameter
        self.selected_stocks = list(map(lambda x:x.Symbol, selected))
        
        return self.selected_stocks
        
    def OnData(self, data: Slice) -> None:
        # Trade stocks at this time
        if self.Time.hour == 9 and self.Time.minute == 45 and len(self.selected_stocks) > 0:
            # Diffrence between upwards and downwards
            difference: Union[None, float] = None
            
            # Make calculation for each stock in self.selected_stocks
            for symbol in self.selected_stocks:
                # Check if symbol is in data slices and stock with this symbol has close price
                if symbol in data and data[symbol] and self.data[symbol].close:
                    price: float = data[symbol].Value
                    
                    # Make sure, that trade is make based on difference
                    if difference == None:
                        difference = 0
                        
                    # If stock's current price is greater than it's close,
                    # then it is upwards stock, otherwise it is downwards stock
                    if price > self.data[symbol].close:
                        difference += 1
                    else:
                        difference -= 1
            
            # Trade only if it calculated difference
            if difference != None:
                # Calculate uds based od difference between upwards and downwards divided by total count of selected stocks            
                uds: float = difference / len(self.selected_stocks)
                trade_direction: int = 1 if uds >= 0 else -1
                
                self.SetHoldings(self.symbol, trade_direction)
            
        # Store close prices and liquidate portfolio
        if self.Time.hour == 15 and self.Time.minute == 59 and len(self.selected_stocks) != 0:
            
            # Update closes of stocks
            for symbol in self.selected_stocks:
                # Check if symbol is in data slices
                if symbol in data and data[symbol]:
                    self.data[symbol].close = data[symbol].Value
                
            # Liquidate portfolio
            self.Liquidate()
        
    def Selection(self) -> None:
        self.selection_flag = True
        
@dataclass
class SymbolData():
    close: Union[None, float] = None
    open: Union[None, float] = None
# 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"))