“通过252天风险调整回报交易Nifty 100股票，使用风险预算，总风险为15%，目标风险为10%，允许负权重，每10天重新平衡一次。”
资产类别: 期货、股票 | 地区: 印度 | 周期: 每周 | 市场: 股票 | 关键词: 时间序列、动量、印度
I. 策略概要
投资范围包括Nifty 100成分股。对于每只股票，计算252天风险调整回报：做多回报为正的股票，做空回报为负的股票。应用风险预算确定投资组合权重，总风险分配设置为15%，目标风险为10%。修改后的风险预算方法允许负权重，优化过程详见第5页。投资组合每10天重新平衡一次，利用风险调整回报和优化配置来指导投资决策。
II. 策略合理性
动量异常在学术文献中得到了充分支持，涵盖了各种资产类别和市场。传统的解释包括投资者羊群效应、过度反应、反应不足和确认偏差等行为因素，一些基于风险的理论则认为高表现资产本身风险更高。本文不提出新的解释，但强调了重要的实践发现。它表明，动量异常，无论是长期还是短期，以及横截面还是时间序列，都存在于印度市场。该策略采用风险调整回报，重申了基于动量策略的广泛适用性和有效性，同时强调了它们在印度的相关性。动量仍然是经过充分研究、广泛接受且有效的。
III. 来源论文
Momentum in the Indian Equity Markets: Positive Convexity and Positive Alpha [点击查看论文]
- Srivastava, Sonam 和 Chakravorty, Gaurav 和 Singhal, Mansi。Wright Research。Qplum。Qplum
<摘要>
我们展示了在印度流动性股票期货市场上有效的动量策略。我们评估并确定了从季度、每周到更细粒度回溯期的回报持久性。我们考察了印度市场交易的流动性股票工具的范围，以评估这种异常现象。我们根据频率——日数据和盘中K线数据——评估了两个数据集的动量。在日尺度上，我们将动量与其他风格因子进行比较。在盘中尺度上，我们评估了时间序列动量或绝对动量以及横截面动量或相对动量。我们证明，在最优时间范围内，印度证券的动量策略可以成为不相关Alpha的来源。我们使用给定目标风险下的主动风险预算进行投资组合构建。我们将在另一份出版物中展示它如何优于均值-方差优化。
IV. 回测表现
| 年化回报 | 16.02% |
| 波动率 | 15.1% |
| β值 | -0.011 |
| 夏普比率 | 1.03 |
| 索提诺比率 | N/A |
| 最大回撤 | -16.24% |
| 胜率 | 52% |
V. 完整的 Python 代码
from AlgorithmImports import *
import numpy as np
import pandas as pd
#endregion
class LongTermTimeSeriesMomentumInIndia(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.data = {}
self.symbols = []
self.period = 252 # Storing 252-days risk-adjusted return
self.vol_target_period = 60
self.target_volatility = 0.10 # Target risk of the allocation is 10%
self.leverage_cap = 4
self.max_missing_days = 5
self.days_counter = 10
self.portfolio_part = 1 / 2 # Trading 1 / 2 of portfolio, because strategy is too volatile
self.vol_targeting_flag = True
csv_string_file = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/india_nifty_500_tickers.csv')
lines = csv_string_file.split('\r\n')
for line in lines:
line_split = line.split(';')
for ticker in line_split:
security = self.AddData(QuantpediaIndiaStocks, ticker, Resolution.Daily)
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(5)
symbol = security.Symbol
self.symbols.append(symbol)
self.data[symbol] = SymbolData(self.period)
def OnData(self, data):
# Update stock prices
for symbol in self.symbols:
if symbol in data and data[symbol]:
self.data[symbol].update(data[symbol].Value)
# Return from function if holding period didn't end
if self.days_counter != 10:
self.days_counter += 1
return
self.days_counter = 1
long = []
short = []
for symbol in self.symbols:
if self.Securities[symbol].GetLastData() and (self.Time.date() - self.Securities[symbol].GetLastData().Time.date()).days <= self.max_missing_days:
# Check if data for stock are ready
if not self.data[symbol].is_ready():
continue
risk_adjusted_return = self.data[symbol].risk_adjusted_return()
# If the return is positive, long the stock.
if risk_adjusted_return >= 0:
long.append(symbol)
# If the return is negative, short the stock
else:
short.append(symbol)
long_length = len(long)
short_length = len(short)
if self.vol_targeting_flag:
# Portfolio volatility calc.
df = pd.dataframe()
weights = []
if long_length > 0:
self.AppendWeights(df, weights, long, long_length, True)
if short_length > 0:
self.AppendWeights(df, weights, short, short_length, False)
weights = np.array(weights)
if len(weights) == 0:
self.Liquidate()
return
daily_returns = df.pct_change()
portfolio_vol = np.sqrt(np.dot(weights.T, np.dot(daily_returns.cov() * 252, weights.T)))
leverage = self.target_volatility / portfolio_vol
leverage = min(self.leverage_cap, leverage) # cap max leverage
# Trade Execution
stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in stocks_invested:
if symbol not in long or symbol not in short:
self.Liquidate(symbol)
for symbol in long:
self.SetHoldings(symbol, (1 / long_length) * leverage * self.portfolio_part)
for symbol in short:
self.SetHoldings(symbol, (-1 / short_length) * leverage * self.portfolio_part)
else:
# Equally weighted trade execution
stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in stocks_invested:
if symbol not in long or symbol not in short:
self.Liquidate(symbol)
for symbol in long:
self.SetHoldings(symbol, 1 / long_length * self.portfolio_part)
for symbol in short:
self.SetHoldings(symbol, -1 / short_length * self.portfolio_part)
def AppendWeights(self, df, weights, symbols_list, total_symbols, long_flag):
for symbol in symbols_list:
df[str(symbol)] = [x for x in self.data[symbol].closes][:self.vol_target_period]
if long_flag:
weights.append(1 / total_symbols)
else:
weights.append(-1 / total_symbols)
class SymbolData():
def __init__(self, period):
self.closes = RollingWindow[float](period)
def update(self, close):
self.closes.Add(close)
def is_ready(self):
return self.closes.IsReady
def risk_adjusted_return(self):
closes = [x for x in self.closes]
return (closes[0] - closes[-1]) / closes[-1]
# 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
try:
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])
except:
return None
return data
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))