The strategy trades Chinese stocks based on 12-month intraday momentum, going long on the top decile and short on the bottom, with value-weighted, monthly rebalanced intraday positions.

I. STRATEGY IN A NUTSHELL

The strategy trades Chinese stocks using 12-month intraday momentum, going long the top-decile and short the bottom-decile stocks, holding intraday positions and rebalancing monthly.

II. ECONOMIC RATIONALE

Unique market structure and investor heterogeneity in China create intraday return patterns, where small, high-turnover, growth stocks exhibit persistent momentum exploitable through intraday trading strategies.

III. SOURCE PAPER

Investor Heterogeneity and Momentum-based Trading Strategies in China [Click to Open PDF]

Ya Gao, Tianjin University – College of Management and Economics; Xing Han, University of Auckland Business School; Youwei Li, Hull University Business School; Xiong Xiong, College of Management and Economics and China Center for Social Computing and Analytics

<Abstract>

The conventional momentum strategy performs poorly overall in China, because stock prices behave very differently when markets are open for trading versus when they are closed. Stocks that are past intraday (overnight) winners persistently outperform those that are past intraday (overnight) losers in the subsequent intraday (overnight) periods. However, the same intraday- (overnight-) momentum strategy suffers dramatically in the subsequent overnight (intraday) periods. Further analysis shows that past intraday (overnight) winners tend to be more (less) speculative stocks which are highly demanded during the day (night). Overall, our results are consistent with investor heterogeneity, and this persistent tug of war virtually eliminate the effectiveness of investors pursuing the momentum-based trading strategy in China.

IV. BACKTEST PERFORMANCE

Annualised Return36.87%
Volatility23.5%
Beta0.01
Sharpe Ratio1.57
Sortino Ratio-0.018
Maximum DrawdownN/A
Win Rate47%

V. FULL PYTHON CODE

from AlgorithmImports import *
from numpy import floor
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class TheIntradayMomentumInChina(QCAlgorithm):
    def Initialize(self) -> None:
        self.SetStartDate(2015, 1, 1)
        self.SetCash(100_000)
        
        self.tickers_to_ignore: List[str] = ['EVK']
        self.period: int = 12 * 21
        self.quantile: int = 10
        self.sort_chunk: float = 0.3
        self.leverage: int = 10
        self.traded_portion: float = 0.2
        self.data: Dict[Symbol, SymbolData] = {}
        self.weight: Dict[Symbol, float] = {}
        
        symbol: Symbol = self.AddEquity("SPY", Resolution.Minute).Symbol
        time_offset: int = 16
        
        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(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
        self.Schedule.On(self.DateRules.EveryDay(symbol), self.TimeRules.AfterMarketOpen(symbol, -time_offset), self.BeforeOpen)
        self.Schedule.On(self.DateRules.EveryDay(symbol), self.TimeRules.BeforeMarketClose(symbol, time_offset), self.BeforeClose)
    def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
        for security in changes.AddedSecurities:
            security.SetFeeModel(CustomFeeModel())
            security.SetLeverage(self.leverage)
            
    def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
        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.CompanyReference.BusinessCountryID == 'CHN'
            and x.Symbol.Value not in self.tickers_to_ignore
        ]
        selected_stocks = [x for x in sorted(selected, key=lambda x:x.MarketCap, reverse=True)][:int(len(selected) * self.sort_chunk)]
        market_cap: Dict[Symbol, float] = {}
        cum_intraday_returns: Dict[Symbol, float] = {}
        
        for stock in selected_stocks:
            symbol: Symbol = stock.Symbol
            
            if symbol not in self.data:
                # Collect opens and closes of stock
                self.data[symbol] = SymbolData(self.period)
                history: dataframe = self.History(symbol, self.period, Resolution.Daily)
                
                if history.empty:
                    self.Log(f"Not enough data for {symbol} yet")
                    continue
                
                closes: Series = history.loc[symbol].close
                opens: Series = history.loc[symbol].open
                
                for (_, close_price), (_, open_price) in zip(closes.items(), opens.items()):
                    self.data[symbol].update(close_price, open_price)
                
            if self.data[symbol].is_ready():
                cum_intraday_returns[symbol] = self.data[symbol].cumulative_intraday_returns()
                market_cap[symbol] = stock.MarketCap
        
        if len(cum_intraday_returns) < self.quantile:
            return Universe.Unchanged
        quantile: int = int(len(cum_intraday_returns) / self.quantile)
        sorted_by_cum_intraday_returns: List[Symbol] = [x[0] for x in sorted(cum_intraday_returns.items(), key=lambda item: item[1])]
    
        long: List[Symbol] = sorted_by_cum_intraday_returns[-quantile:]
        short: List[Symbol] = sorted_by_cum_intraday_returns[:quantile]
    
        # Need to clear last long and short portfolio before creating the new one.
        self.weight.clear()
        for i, portfolio in enumerate([long, short]):
            mc_sum: float = sum(list(map(lambda symbol: market_cap[symbol], portfolio)))
            for symbol in portfolio:
                self.weight[symbol] = ((-1)**i) * market_cap[symbol] / mc_sum
        # Long and short portfolio is selected only once in a month.    
        self.selection_flag = False
        
        return long + short
    def OnData(self, data: Slice) -> None:
        if (self.Time.hour == 0 and self.Time.minute == 0):
            # Updating RollingWindow each day with new open and close price.
            for symbol in self.data:
                history: History = self.History(symbol, 1, Resolution.Daily)
                
                if history.empty:
                    self.Log(f"Not enough data for {symbol} yet")
                    continue
                
                close: Series = history.loc[symbol].close
                open: Series = history.loc[symbol].open
                
                for (_, close_price), (_, open_price) in zip(close.items(), open.items()):
                    self.data[symbol].update(close_price, open_price)
        
    def BeforeOpen(self) -> None:
        # open new positions
        for symbol, w in self.weight.items():
            if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable: 
                quantity = floor((1 * self.Portfolio.TotalPortfolioValue * w) * self.traded_portion / self.data[symbol].LastPrice)
                self.MarketOnOpenOrder(symbol, quantity)
                    
    def BeforeClose(self) -> None:
        # liquidate
        for symbol, w in self.weight.items():
            if self.Portfolio[symbol].Invested:
                quantity: int = self.Portfolio[symbol].Quantity
                self.MarketOnCloseOrder(symbol, -quantity)
    def Selection(self):
        self.selection_flag = True
        
class SymbolData():
    def __init__(self, period: int) -> None:
        self.Closes: RollingWindow = RollingWindow[float](period)
        self.Opens: RollingWindow = RollingWindow[float](period)
        self.LastPrice: float = 0.
        
    def update(self, close_price: float, open_price: float) -> None:
        self.Closes.Add(close_price)
        self.Opens.Add(open_price)
        self.LastPrice = close_price
        
    def update_opens(self, open_price: float) -> None:
        self.Opens.Add(open_price)
    
    def update_closes(self, close_price: float) -> None:
        self.Closes.Add(close_price)
        
    def is_ready(self) -> bool:
        return self.Closes.IsReady and self.Opens.IsReady
        
    def cumulative_intraday_returns(self) -> float:
        closes: List[float] = [x for x in self.Closes]
        opens: List[float] = [x for x in self.Opens]
        intraday_returns: List[float] = [(close_price - open_price) / open_price for close_price, open_price in zip(closes, opens)]
        return sum(intraday_returns)
        
# 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"))

Leave a Reply

Discover more from Quant Buffet

Subscribe now to keep reading and get access to the full archive.

Continue reading