投资者情绪与货币市场的动量效应
登录后收藏学术论文
Investor Sentiment, Attention and Profitability of Currency Momentum Strategies
Pawee Maryniak
- Wrocław University of Science and Technology
- Wroclaw University of Economics and Business
- ?Uniwersytet Ekonomiczny we Wrocławiu
- ?Wroclaw University of Technology
策略概要
投资范围包括71种与美元交易的货币。情绪通过贝克-伍格勒(BW)指数衡量,该指数将情绪分为高、中、低水平。货币也按其过去表现进行排名,前六分之一和后六分之一分别归类为“赢家”和“输家”。根据上个月的情绪水平,投资者采取不同的仓位:在低情绪期间做多赢家,在高情绪期间做空输家,在中等情绪期间同时做多和做空。投资组合等权重,并每月重新平衡。
II. 策略合理性
贝克-伍格勒(BW)指数用于衡量情绪,并解释市场中的动量效应。根据该理论,在情绪低落时期,投资者通常寻求安全资产,由于资本转移到更安全的市场,这些资产往往被高估。这种避险行为通常导致外币多头的高回报和空头的低回报,从而形成动量。相反,在高情绪时期,资本流入风险较高的资产导致美元升值,与多头相比,空头获得高回报。
回测表现
波动率10.36%
夏普比率0.56
胜率37%
完整 Python 代码
from AlgorithmImports import *
import numpy as np
from dateutil.relativedelta import relativedelta
#endregion
class InvestorSentiment(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2011, 1, 1)
self.SetCash(100000)
self.period:int = 21
self.SetWarmUp(self.period, Resolution.Daily)
self.symbols = [
"USDAUD", "USDCAD", "USDCHF", "USDCZK", "USDDKK", "USDEUR",
"USDGBP", "USDHKD", "USDHUF", "USDJPY", "USDMXN", "USDPLN",
"USDNOK", "USDSAR", "USDSGD", "USDTHB", "USDTRY", "USDTWD",
"USDZAR", "USDSEK"
]
self.data:dict[Symbol, SymbolData] = {}
self.sentiment_warmup_period:int = 12
self.sentiment_history = RollingWindow[float](self.sentiment_warmup_period)
self.max_missing_days:int = 31
self.quantile:int = 6
for symbol in self.symbols:
data = self.AddForex(symbol, Resolution.Daily, Market.Oanda)
data.SetLeverage(5)
self.data[symbol] = SymbolData(symbol, self.period)
# Import custom data
self.sentimet_index:Symbol = self.AddData(SentimentData, "sentiment", Resolution.Daily).Symbol
self.recent_month:int = -1
def OnData(self, data:Slice) -> None:
perf:dict[str, float] = {}
# store daily prices
for symbol in self.symbols:
if symbol in data and data[symbol]:
self.data[symbol].Update(data[symbol].Value)
if self.recent_month != self.Time.month and not self.IsWarmingUp:
if self.data[symbol].IsReady():
perf[symbol] = self.data[symbol].Return()
if self.IsWarmingUp: return
# monthly rebalance
if self.recent_month == self.Time.month:
return
self.recent_month = self.Time.month
# check sentiment index data arrival
if self.Securities[self.sentimet_index].GetLastData() and (self.Time.date() - self.Securities[self.sentimet_index].GetLastData().Time.date()).days > self.max_missing_days:
if self.Portfolio.Invested:
self.Liquidate()
self.sentiment_history.Reset()
else:
sentiment_index:float = self.Securities[self.sentimet_index].Price
self.sentiment_history.Add(sentiment_index)
if not self.sentiment_history.IsReady: return
sorted_by_ret:List = sorted([x for x in self.data.items() if x[1].IsReady()], key=lambda x: x[1].Return(), reverse = True)
if len(sorted_by_ret) < self.quantile:
self.Liquidate()
return
quantile:int = int(len(self.symbols) / self.quantile)
winners:List[str] = [x[0] for x in sorted_by_ret[:quantile]]
losers:List[str] = [x[0] for x in sorted_by_ret[-quantile:]]
percentile_33:float = np.percentile(list(self.sentiment_history), 0.33)
percentile_66:float = np.percentile(list(self.sentiment_history), 0.66)
long:List[str] = []
short:List[str] = []
# When the sentiment level in the previous month is low then only long position in winners is taken. When the sentiment level is high then only short position is taken. When sentiment level is medium then both short and long positions are taken.
if sentiment_index < percentile_33:
long = winners
if sentiment_index > percentile_66:
short = losers
else:
long = winners
short = losers
# liquidate
invested = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long + short:
self.Liquidate(symbol)
# market execution
for symbol in long:
self.SetHoldings(symbol, 1 / len(long))
for symbol in short:
self.SetHoldings(symbol, -1 / len(short))
class SymbolData:
def __init__(self, symbol:str, lookback:int) -> None:
self.Symbol:str = symbol
self.History:RollingWindow = RollingWindow[float](lookback)
self.Lookback = lookback
def Update(self, value:float) -> None:
self.History.Add(value)
def IsReady(self) -> bool:
return self.History.IsReady
# Monthly return
def Return(self) -> float:
prices:List[float] = list(self.History)
return (prices[0] / prices[self.Lookback - 1]) - 1
class SentimentData(PythonData):
def GetSource(self, config, date, isLiveMode):
return SubscriptionDataSource("data.quantpedia.com/backtesting_data/index/baker_wurgler_sentiment_index.csv", SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
def Reader(self, config, line, date, isLiveMode):
index = SentimentData()
index.Symbol = config.Symbol
try:
data = line.split(';')
index.Time = datetime.strptime(data[0], "%Y%m") + relativedelta(months=1)
index.Value = data[1]
except:
return None
return index