“该策略通过10年滚动回归估算夏普比率，当夏普比率高于0.2时投资于股票，否则转向无风险资产。每月调整投资组合，以实现最佳的风险收益比。”
资产类别：差价合约（CFDs）、交易所交易基金（ETFs）、基金、期货 | 地区：全球 | 频率：每月 | 市场：股票市场 | 关键词：市场时机、夏普比率
I. 策略概述
该策略通过10年滚动回归估算条件夏普比率，分别使用回归方程预测股票收益的均值和波动率，变量包括股息收益率和股票回购收益率等。每月比较预测的夏普比率与固定阈值（0.2）。
- 如果夏普比率超过阈值，策略投资于股票市场。
- 如果夏普比率低于阈值，策略投资于无风险资产。
该策略灵活支持其他预测变量（可参考论文详细描述），以动态调整投资组合分配，优化条件风险收益权衡，实现系统化的风险管理与回报最大化。
II. 策略合理性
学术研究提供了两种关于夏普比率可预测性的理论：
- 消费者情绪理论：
消费者情绪与商业周期相关联。投资者在周期高峰期往往过于乐观，导致股票被高估，进而未来风险收益比下降；在周期低谷期，悲观情绪导致股票被低估，风险收益比上升。 - 理性风险厌恶理论：
条件收益和波动的时间变化由总体风险厌恶的变化驱动。在经济扩张期间，风险厌恶下降，预期收益和波动率较低；在经济衰退期间，风险厌恶上升，预期收益和波动率上升。
两种理论共同揭示了情绪与理性调整在商业周期中对夏普比率可预测性的影响，为市场时机策略提供了理论支持。
III. 论文来源
Time-Varying Sharpe Ratios and Market Timing [点击浏览原文]
- 作者：Tang, Whitelaw
- 机构：福特汉姆大学Gabelli商学院、纽约大学、美国国家经济研究局（NBER）
<摘要>
本文记录了股票市场夏普比率的时间变化及其可预测性。研究使用预先设定的金融变量估计股票收益的条件均值和波动率，将两者结合估算条件夏普比率，或直接通过这些变量构建线性方程预测夏普比率。研究发现，条件夏普比率的显著时间变化与商业周期阶段一致。一般而言，夏普比率在周期高峰期较低，而在周期低谷期较高。这表明，通过预测夏普比率的变化，投资者能够有效调整资产配置以优化风险收益权衡。
IV. 回测表现
| 年化收益率 | 14.18% |
| 波动率 | 15.38% |
| Beta | 0.376 |
| 夏普比率 | 0.66 |
| 索提诺比率 | N/A |
| 最大回撤 | N/A |
| 胜率 | 64% |
V. 完整python代码
from AlgorithmImports import *
import numpy as np
import statsmodels.api as sm
#endregion
class MarketTimingUsingSharpeRatios(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
self.data = {}
self.period = 21 # One month period
self.regression_data = {}
# self.regression_period = 10 * 12 + 1 # Need 10 years and one month of data, to predict Y
self.regression_period = 5 * 12 + 1
self.rf_asset = self.AddEquity('BIL', Resolution.Daily).Symbol # 5/29/2007.
self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.spy_monthly_return = RollingWindow[float](self.regression_period)
self.spy_volatility = RollingWindow[float](self.regression_period)
self.data[self.symbol] = SymbolData(self.period)
self.dividend_yield = self.AddData(QuandlValue, 'MULTPL/SP500_DIV_YIELD_MONTH', Resolution.Daily).Symbol
self.regression_data[self.dividend_yield] = RollingWindow[float](self.regression_period)
self.buyback_yield = self.AddData(QuantpediaBuyBackYield, 'BUYBACK_YIELD', Resolution.Daily).Symbol
self.regression_data[self.buyback_yield] = RollingWindow[float](self.regression_period)
self.symbols = [self.symbol, self.buyback_yield, self.dividend_yield]
self.threshold = 0.2
self.selection_flag = False
self.temp_buyback_storing = 0
self.UniverseSettings.Resolution = Resolution.Daily
self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(5)
def OnData(self, data):
for symbol in self.symbols:
if symbol in data:
if data[symbol]:
value = data[symbol].Value
if value != 0:
if symbol == self.symbol: # Storing daily SPY close
self.data[symbol].update(value)
elif symbol == self.buyback_yield: # This will secure last value of the month
self.temp_buyback_storing = value
elif symbol == self.dividend_yield: # Annoucment is at the end of month, that's why we use self.temp_buyback_storing for best synchronization
self.regression_data[symbol].Add(value)
if not self.selection_flag:
return
# make sure data is still comming in
if self.Securities[self.buyback_yield].GetLastData() and (self.Time.date() - self.Securities[self.buyback_yield].GetLastData().Time.date()).days > 31:
self.Liquidate()
return
if self.Securities[self.dividend_yield].GetLastData() and (self.Time.date() - self.Securities[self.dividend_yield].GetLastData().Time.date()).days > 5:
self.Liquidate()
return
# Dividend_yield is announced at the end of month, so we take the data of buyback_yield at the end of month, if those two can't be synchronized
self.regression_data[self.buyback_yield].Add(self.temp_buyback_storing)
self.selection_flag = False
# Storing monthly return and volatility of SPY
if self.data[self.symbol].is_ready():
self.spy_monthly_return.Add(self.data[self.symbol].monthly_return())
self.spy_volatility.Add(self.data[self.symbol].volatility())
sharpe_ratio = None
# Check if our Y's are ready
if self.spy_monthly_return.IsReady and self.spy_volatility.IsReady:
# Check if data for our X are ready
if self.regression_data[self.dividend_yield].IsReady and self.regression_data[self.buyback_yield].IsReady:
Y1 = [x for x in self.spy_volatility][:-1] # Everything except the first data
Y2 = [x for x in self.spy_monthly_return][:-1] # Everything except the first data
dividend_yield = [x for x in self.regression_data[self.dividend_yield]]
buyback_yield = [x for x in self.regression_data[self.buyback_yield]]
X = [
dividend_yield[1:], # Everything except the last data
buyback_yield[1:] # Everything except the last data
]
# beta = slope, alpha = intercept
# Get expected volatility
regression_model = self.MultipleLinearRegression(X, Y1) # Volatility regression
expected_volatility = regression_model.predict([1, dividend_yield[0], buyback_yield[0]]) # 1 is needed constant
# Get expected monthly return
regression_model = self.MultipleLinearRegression(X, Y2) # Monthly return regression
expected_return = regression_model.predict([1, dividend_yield[0], buyback_yield[0]]) # 1 is needed constant
sharpe_ratio = (expected_return / expected_volatility)[0]
# Trade execution
if sharpe_ratio:
# Liquidate risk-free asset and invest into SPY
if sharpe_ratio > self.threshold:
if self.Securities[self.rf_asset].Invested:
self.Liquidate(self.rf_asset)
self.SetHoldings(self.symbol, 1)
else: # Liquidate SPY and invest into risk-free asset
if self.Securities[self.symbol].Invested:
self.Liquidate(self.symbol)
self.SetHoldings(self.rf_asset, 1)
def MultipleLinearRegression(self, x, y):
x = np.array(x).T
x = sm.add_constant(x)
result = sm.OLS(endog=y, exog=x).fit()
return result
def Selection(self):
self.selection_flag = True
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 volatility(self):
values = [x for x in self.Closes]
values = np.array(values)
returns = (values[:-1] - values[1:]) / values[1:]
return np.std(returns)
def monthly_return(self):
values = [x for x in self.Closes]
return (values[0] - values[-1]) / values[-1]
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
# Quandl "value" data
class QuandlValue(PythonQuandl):
def __init__(self):
self.ValueColumnName = 'Value'
# Quantpedia data.
# NOTE: IMPORTANT: Data order must be ascending (datewise)
class QuantpediaBuyBackYield(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/buyback_yield/{0}.csv".format(config.Symbol.Value), SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
data = QuantpediaBuyBackYield()
data.Symbol = config.Symbol
if not line[0].isdigit(): return None
split = line.split(';')
data.Time = datetime.strptime(split[0], "%d.%m.%Y") + timedelta(days=1)
buyback = float(split[1])
SPX = float(split[2])
data.Value = float(SPX / buyback)
return data
