“该策略交易印度排名前500的股票，通过动量和波动率排序，做多高动量、低波动率股票，做空低动量、高波动率股票，并每月重新平衡。”
资产类别: 股票 | 地区: 印度 | 周期: 每月 | 市场: 股票 | 关键词: 印度、动量效应
I. 策略概要
该策略以印度国家证券交易所市值排名前500的股票为目标，这些股票在过去36个月中至少有12个月的回报数据。动量计算为过去12个月的回报减去上个月的回报，而波动率则来源于过去三年超额回报。股票通过动量和波动率双重排序，分为等权重的风险五分位数。该策略做多动量最高且波动率最低的股票，做空动量最低且波动率最高的股票。投资组合每月使用月末数据进行再平衡。
II. 策略合理性
印度市场的低风险投资组合产生了高额正阿尔法（+5.24%），而高风险投资组合则显示出显著的负阿尔法（-6.96%）。在控制了规模、价值、动量、票面金额、流动性和行业敞口等因素后，这种效应仍然稳健。与发达市场不同，印度的低风险投资组合在行业集中度上低于广义市场指数。低风险投资组合带来了13.28%的回报，夏普比率为0.72，并随着风险五分位数的升高而下降。
添加高动量筛选器可增强低风险策略，实现10.60%的阿尔法和17.79%的回报，波动率为18.06%，夏普比率为0.99。高动量风险五分位数投资组合在绝对和风险调整方面均优于纯风险五分位数投资组合。市场中性方法，即做空高风险、低动量股票，回报降低6.95%，但仍保持正阿尔法，证明了该策略的稳健性和投资潜力。
III. 来源论文
Low-Risk Effect: Evidence, Explanations and Approaches to Enhancing the Performance of Low-Risk Investment Strategies [点击查看论文]
- Mayank Joshipura 和 Nehal Joshipura。NMIMS 大学商学院。Durgadevi Saraf 管理学院
<摘要>
作者提供了印度股票市场低风险效应的证据，使用了2004年1月至2018年12月期间在印度国家证券交易所（NSE）上市的流动性排名前500的股票。金融理论预测风险与回报之间存在正相关关系。然而，实证研究表明，低风险股票在风险调整后的基础上跑赢高风险股票，这被称为低风险异常或低风险效应。这种异常的持续存在是现代金融中最大的谜团之一。作者发现了强有力的证据支持低风险效应，基于简单平均（复合）回报，风险与回报之间存在平坦（负）关系。研究表明，低风险效应独立于规模、价值和动量效应，并且在控制了流动性和股票票面金额等变量后仍然稳健。进一步研究表明，低风险效应是股票和行业层面效应的结合，不能完全通过集中的行业敞口来捕捉。通过将动量效应与低波动率效应相结合，低风险投资策略的表现可以在绝对和风险调整方面得到改善。本文通过提供以下证据为知识体系做出了贡献：a) 低风险效应对于股票的流动性和票面金额以及行业敞口的稳健性，b) 如何通过结合动量和低波动率效应来创建一种长期投资策略，该策略比普通的长期低风险投资策略提供更高的风险调整和绝对回报。
IV. 回测表现
| 年化回报 | 10.84% |
| 波动率 | N/A |
| β值 | -0.173 |
| 夏普比率 | -0.287 |
| 索提诺比率 | N/A |
| 最大回撤 | N/A |
| 胜率 | 47% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
#endregion
class MomentumAndLowRiskEffectsInIndia(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.data:dict = {}
self.symbols:list[Symbol] = []
self.period:int = 37 * 21 # 37 months of daily closes
self.SetWarmUp(self.period, Resolution.Daily)
self.sma:dict = {}
self.sma_period:int = 200
self.quantile:int = 5
self.perf_period:int = 12
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
csv_string_file = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/india_nifty_500_tickers.csv')
line_split = csv_string_file.split(';')
# NOTE: Download method is rate-limited to 100 calls (https://github.com/QuantConnect/Documentation/issues/345)
for ticker in line_split[:99]:
security = self.AddData(QuantpediaIndiaStocks, ticker, Resolution.Daily)
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(5)
symbol:Symbol = security.Symbol
self.symbols.append(symbol)
self.data[symbol] = SymbolData(self.period)
self.sma[symbol] = self.SMA(symbol, self.sma_period, Resolution.Daily)
self.max_missing_days:int = 5
self.recent_month:int = -1
def OnData(self, data):
# store daily prices
for symbol in self.symbols:
if symbol in data and data[symbol]:
price:float = data[symbol].Value
if price != 0 and not np.isnan(price):
self.data[symbol].update(price)
if self.IsWarmingUp: return
if self.Time.month == self.recent_month:
return
self.recent_month = self.Time.month
volatility:[Symbol, float] = {}
performance:[Symbol, float] = {}
# price_over_sma:[Symbol, bool] = {}
for symbol in self.symbols:
# prices are ready and data is still comming in
if self.data[symbol].is_ready() and self.sma[symbol].IsReady:
if self.Securities[symbol].GetLastData() and (self.Time.date() - self.Securities[symbol].GetLastData().Time.date()).days <= self.max_missing_days:
perf:float = self.data[symbol].performance(self.perf_period)
vol:float = self.data[symbol].volatility()
if perf != 0 and vol != 0 and not np.isnan(perf) and not np.isnan(vol):
performance[symbol] = perf
volatility[symbol] = vol
# price_over_sma[symbol] = bool(self.data[symbol].closes[0] > self.sma[symbol].Current.Value)
long:list[Symbol] = []
short:list[Symbol] = []
if len(performance) >= self.quantile and len(volatility) >= self.quantile:
quantile:int = int(len(performance) / self.quantile)
sorted_by_performance = sorted(performance.items(), key=lambda item: item[1], reverse=True)
sorted_by_volatility = sorted(volatility.items(), key=lambda item: item[1], reverse=True)
top_by_perf = [x[0] for x in sorted_by_performance[:quantile]]
bottom_by_perf = [x[0] for x in sorted_by_performance[-quantile:]]
top_by_vol = [x[0] for x in sorted_by_volatility[:quantile]]
bottom_by_vol = [x[0] for x in sorted_by_volatility[-quantile:]]
long = [x for x in top_by_perf if x in bottom_by_vol]
short = [x for x in bottom_by_perf if x in top_by_vol]
# trade execution
long_count:int = len(long)
short_count:int = len(short)
stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in stocks_invested:
if symbol not in long + short:
self.Liquidate(symbol)
if long_count:
self.Log(long_count)
if short_count:
self.Log(short_count)
for symbol in long:
if symbol in data and data[symbol]:
self.SetHoldings(symbol, 1 / long_count)
for symbol in short:
if symbol in data and data[symbol]:
self.SetHoldings(symbol, -1 / short_count)
class SymbolData():
def __init__(self, period:int) -> None:
self.closes = RollingWindow[float](period)
def update(self, close:float) -> None:
self.closes.Add(close)
def is_ready(self) -> bool:
return self.closes.IsReady
def performance(self, months:int) -> float:
return self.closes[21] / self.closes[months * 21-1] - 1
def volatility(self) -> float:
closes:list[float] = [x for x in self.closes]
closes:np.ndarray = np.array(closes)
returns:np.ndarray = (closes[:-1] - closes[1:]) / closes[1:]
vol:float = float(np.std(returns))
# monthly_closes:np.array = np.array(closes[0::21])
# monthly_returns:np.array = (monthly_closes[:-1] - monthly_closes[1:]) / monthly_closes[1:]
# vol:float = np.std(monthly_returns)
return vol
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
# Quantpedia data
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaIndiaStocks(PythonData):
def GetSource(self, config, date, isLiveMode):
source = "data.quantpedia.com/backtesting_data/equity/india_stocks/india_nifty_500/{0}.csv".format(config.Symbol.Value)
return SubscriptionDataSource(source, SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaIndiaStocks()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(',')
data.Time = datetime.strptime(split[0], "%Y-%m-%d") + timedelta(days=1)
data['Price'] = float(split[1])
data.Value = float(split[1])
return data