“该策略针对米兰证券交易所的股票，筛选出具有异常交易量且当日涨幅超过1%的标的，在收盘时买入，持有一天，并每日重新平衡，旨在捕捉短期动量收益。”
资产类别：股票 | 地区：欧洲 | 频率：每日 | 市场：股票市场 | 关键词：异常、交易量、效应、股票市场
I. 策略概述
该策略分析米兰证券交易所（Milan Stock Exchange）的股票，但也可适用于其他市场。每日筛选符合以下条件的股票：
- 异常交易量：当日交易量超过过去66日平均值的2.33倍标准差。
- 无近期异常交易量：过去30天内未出现异常交易量。
- 价格涨幅：当日收盘价上涨至少1%。
符合条件的股票在当日收盘时买入，并持有一天。投资组合采用等权重配置，并每日重新平衡。该策略利用短期交易量激增和价格动量的结合，寻求潜在的短期收益。
II. 策略合理性
学术研究认为，异常交易量可能是由内幕交易引发的，表明市场参与者之间信息分布的不均衡。交易量的剧烈变化（尤其是在无公开消息的情况下）可能反映非公开信息，从而预示未来的超额收益。这种效应不仅与信息流动有关，还可能受到公司特征（如所有权和治理结构）的影响，从而为投资者提供潜在交易信号。
通过筛选异常交易量和价格动量的股票，该策略旨在捕捉信息流动和短期市场行为带来的套利机会，同时避免因流动性波动导致的价格反转风险。
III. 论文来源
THE INFORMATION CONTENT OF ABNORMAL TRADING VOLUME [点击浏览原文]
- 作者：Bajo
- 机构：博洛尼亚大学管理系
<摘要>
本文实证研究了异常交易量如何向市场参与者揭示新信息。交易量通常被视为信息流的良好代理变量，理论认为它能增强投资者的信息集。然而，尚无研究将异常交易量的存在与公司特征（如所有权和治理结构）联系起来，而这些特征在理论上也与信息质量有关。研究发现，极端交易量水平周围出现显著的超额回报，这仅部分归因于信息披露。此外，这些回报并非由流动性波动引起，因为价格在随后期间并未出现反转。这一现象违反了半强式市场有效性假说，表明异常交易量可能是未公开信息的有力信号。
IV. 回测表现
| 年化收益率 | 33.91% |
| 波动率 | N/A |
| Beta | 0.74 |
| 夏普比率 | N/A |
| 索提诺比率 | 0.394 |
| 最大回撤 | N/A |
| 胜率 | 49% |
V. 完整python代码
import numpy as np
from AlgorithmImports import *
from typing import List, Dict
from pandas.core.frame import DataFrame
class AbnormalVolumeEffectStockMarket(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2000, 1, 1)
self.SetCash(100000)
market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.data:Dict[Symbol, SymbolData] = {}
self.period:int = 66
self.leverage:int = 5
self.std_treshold:float = 2.33
self.performance_treshold:float = 0.01
self.long:List[Symbol] = []
self.selection_flag:bool = False
self.last_selection:List[Fundamental] = []
self.fundamental_sorting_key = lambda x: x.MarketCap
self.fundamental_count:int = 1000
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.settings.daily_precise_end_time = False
self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
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]:
# Update the rolling window every day.
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol in self.data:
# Store daily price and volume.
self.data[symbol].update(stock.AdjustedPrice, stock.Volume)
# return already selected universe during the month
if self.selection_flag:
self.selection_flag = False
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0
]
if len(selected) > self.fundamental_count:
selected = sorted(selected, key = self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]
self.last_selection = selected
# Warmup price rolling windows.
for stock in self.last_selection:
symbol:Symbol = stock.Symbol
if symbol in self.data:
continue
self.data[symbol] = SymbolData(symbol, self.period, -1)
history:DataFrame = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet")
continue
if 'close' in history and 'volume' in history:
closes:Series = history.loc[symbol]['close']
volumes:Series = history.loc[symbol]['volume']
for (time1, close), (time2, volume) in zip(closes.items(), volumes.items()):
self.data[symbol].update(close, volume)
# fundamental returned ready data
for stock in self.last_selection:
symbol:Symbol = stock.Symbol
if symbol not in self.data:
continue
if not self.data[symbol].is_ready():
continue
volumes:List[float] = [x for x in self.data[symbol]._volume]
volume_mean:float = np.mean(volumes)
volume_std:float = np.std(volumes)
volume:float = volumes[0] # Takes todays volume
closes:List[float] = [x for x in self.data[symbol]._price][:2] # First two are newest
todays_return:float = (closes[0] - closes[1]) / closes[1]
if volume > volume_mean + (self.std_treshold * volume_std):
# selects only firms with no abnormal volume over the preceding 30 trading days
if (self.data[symbol]._abnormal_date == -1) or (self.data[symbol]._abnormal_date < (self.Time - timedelta(days=30))):
# if the stocks finished the day with at least a 1% gain
if todays_return >= self.performance_treshold:
self.long.append(symbol)
self.data[symbol]._abnormal_date = self.Time
return self.long
def OnData(self, data: Slice) -> None:
# Trade execution
targets:List[PortfolioTarget] = [PortfolioTarget(symbol, 1. / len(self.long)) for symbol in self.long if symbol in data and data[symbol]]
self.SetHoldings(targets, True)
self.long.clear()
def Selection(self) -> None:
self.selection_flag = True
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
class SymbolData():
def __init__(self, symbol:Symbol, period:int, abnormal_date:datetime):
self._symbol:Symbol = symbol
self._price:RollingWindow = RollingWindow[float](period)
self._volume:RollingWindow = RollingWindow[float](period)
self._abnormal_date:datetime = abnormal_date
def update(self, price:float, volume:float):
self._price.Add(price)
self._volume.Add(volume)
def is_ready(self) -> bool:
return self._price.IsReady and self._volume.IsReady
