Overall Statistics |
Total Trades
11314
Average Win
0.63%
Average Loss
-0.62%
Compounding Annual Return
1.772%
Drawdown
43.900%
Expectancy
0.013
Net Profit
27.161%
Sharpe Ratio
0.172
Probabilistic Sharpe Ratio
0.033%
Loss Rate
50%
Win Rate
50%
Profit-Loss Ratio
1.02
Alpha
0.031
Beta
-0.008
Annual Standard Deviation
0.176
Annual Variance
0.031
Information Ratio
-0.298
Tracking Error
0.255
Treynor Ratio
-3.89
Total Fees
$5409.52
Estimated Strategy Capacity
$130000.00
Lowest Capacity Asset
CSPI R735QTJ8XC9X
|
import numpy as np from scipy.optimize import minimize # Custom fee model class CustomFeeModel(FeeModel): def GetOrderFee(self, parameters): fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005 return OrderFee(CashAmount(fee, "USD")) # NOTE: Manager for new trades. It's represented by certain count of equally weighted brackets for long and short positions. # If there's a place for new trade, it will be managed for time of holding period. class TradeManager(): def __init__(self, algorithm, long_size, short_size, holding_period): self.algorithm = algorithm # algorithm to execute orders in. self.long_size = long_size self.short_size = short_size self.long_len = 0 self.short_len = 0 # Arrays of ManagedSymbols self.symbols = [] self.holding_period = holding_period # Days of holding. # Add stock symbol object def Add(self, symbol, long_flag): # Open new long trade. managed_symbol = ManagedSymbol(symbol, self.holding_period, long_flag) if long_flag: # If there's a place for it. if self.long_len < self.long_size: self.symbols.append(managed_symbol) self.algorithm.SetHoldings(symbol, 1 / self.long_size) self.long_len += 1 else: self.algorithm.Log("There's not place for additional trade.") # Open new short trade. else: # If there's a place for it. if self.short_len < self.short_size: self.symbols.append(managed_symbol) self.algorithm.SetHoldings(symbol, - 1 / self.short_size) self.short_len += 1 else: self.algorithm.Log("There's not place for additional trade.") # Decrement holding period and liquidate symbols. def TryLiquidate(self): symbols_to_delete = [] for managed_symbol in self.symbols: managed_symbol.days_to_liquidate -= 1 # Liquidate. if managed_symbol.days_to_liquidate == 0: symbols_to_delete.append(managed_symbol) self.algorithm.Liquidate(managed_symbol.symbol) if managed_symbol.long_flag: self.long_len -= 1 else: self.short_len -= 1 # Remove symbols from management. for managed_symbol in symbols_to_delete: self.symbols.remove(managed_symbol) def LiquidateTicker(self, ticker): symbol_to_delete = None for managed_symbol in self.symbols: if managed_symbol.symbol.Value == ticker: self.algorithm.Liquidate(managed_symbol.symbol) symbol_to_delete = managed_symbol if managed_symbol.long_flag: self.long_len -= 1 else: self.short_len -= 1 break if symbol_to_delete: self.symbols.remove(symbol_to_delete) else: self.algorithm.Debug("Ticker is not held in portfolio!") class ManagedSymbol(): def __init__(self, symbol, days_to_liquidate, long_flag): self.symbol = symbol self.days_to_liquidate = days_to_liquidate self.long_flag = long_flag
# https://quantpedia.com/strategies/reversal-in-post-earnings-announcement-drift/ # # The investment universe consists of all stocks from NYSE, AMEX, and NASDAQ with active options market (so mostly large-cap stocks). # Each day investor selects stocks which would have earnings announcement during the next working day. He then checks the abnormal # performance of these stocks during the previous earnings announcement. Investor goes long decile of stocks with the lowest abnormal # past earnings announcement performance and goes short stocks with the highest abnormal past performance. Stocks are held for two # days, and the portfolio is weighted equally. # # QC Implementation: # - Universe consist of stock, which have earnings dates in Quantpedia data. import data_tools import numpy as np from collections import deque from pandas.tseries.offsets import BDay class ReversalPostEarningsAnnouncementDrift(QCAlgorithm): def Initialize(self): self.SetStartDate(2008, 1, 1) # Earnings dates starts at 2008 self.SetCash(100000) # EAR last quarter data. self.ear_data = {} self.ear_period = 30 self.coarse_count = 500 self.symbol = self.AddEquity('SPY', Resolution.Daily).Symbol # Daily price data. self.data = {} # Custom stock universe from earnings data symbols. self.symbols = set() # Import earnigns data. self.earnings_data = {} self.first_date = None csv_string_file = self.Download('data.quantpedia.com/backtesting_data/economic/earning_dates.csv') lines = csv_string_file.split('\r\n') for line in lines: if line == '': continue line_split = line.split(';') date = datetime.strptime(line_split[0], "%Y-%m-%d").date() if not self.first_date: self.first_date = date self.earnings_data[date] = [] for i in range(1, len(line_split)): symbol = line_split[i] self.earnings_data[date].append(symbol) self.symbols.add(symbol) # Ear quarters history. self.current_quarter_ears = [] self.previous_quarter_ears = [] # Equally weighted brackets for traded symbols. - 10 symbols long and short, 2 days of holding. self.trade_manager = data_tools.TradeManager(self, 10, 10, 2) self.selection_flag = False self.store_sales_data_flag = True self.sales_growth_sort_flag = False self.UniverseSettings.Resolution = Resolution.Daily self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection) def OnSecuritiesChanged(self, changes): for security in changes.AddedSecurities: security.SetFeeModel(data_tools.CustomFeeModel(self)) security.SetLeverage(5) def CoarseSelectionFunction(self, coarse): # update daily prices for stock in coarse: symbol = stock.Symbol if symbol in self.data: self.data[symbol].update(self.Time, stock.AdjustedPrice) # monthly selection if not self.selection_flag: return Universe.Unchanged self.selection_flag = False # Select every available stock. selected = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5 and x.Symbol.Value in self.symbols] for symbol in selected + [self.symbol]: if symbol in self.data: continue # warm up stock prices self.data[symbol] = SymbolData(self.ear_period) history = self.History(symbol, self.ear_period, Resolution.Daily) if history.empty: self.Log(f"Not enough data for {symbol} yet") continue closes = history.loc[symbol].close for time, close in closes.iteritems(): self.data[symbol].update(self.Time, close) return [x for x in selected if self.data[x].is_ready()] def FineSelectionFunction(self, fine): fine = [x for x in fine if x.EarningReports.FileDate] # Stocks with last month's earnings. last_month_date = self.Time - timedelta(self.Time.day) filered_fine = [x for x in fine if (x.EarningReports.FileDate.year == last_month_date.year and x.EarningReports.FileDate.month == last_month_date.month)] for stock in filered_fine: symbol = stock.Symbol # EAR calc. # Get 4 days around earnings. date_from = stock.EarningReports.FileDate.date() - BDay(2) date_to = stock.EarningReports.FileDate.date() + BDay(1) # Month of data is ready. if self.data[self.symbol].is_ready(): market_return = self.data[self.symbol].get_prices([date_from, date_to]) stock_return = self.data[symbol].get_prices([date_from, date_to]) # check if returns are ready if market_return and stock_return: ear = stock_return - market_return ear_data = (stock.EarningReports.FileDate, ear) self.ear_data[symbol] = ear_data # Store ear in this month's history. self.current_quarter_ears.append(ear) # check if there are any symbols, which can be traded if len(self.ear_data) == 0: return Universe.Unchanged # return symbols from self.ear_data, because they will be traded return [x for x in self.ear_data] def OnData(self, data): # Open trades on earnings day. date_to_lookup = self.Time.date() # If there is no earnings data yet. if date_to_lookup < self.first_date: return # Liquidate opened symbols after holding period. self.trade_manager.TryLiquidate() # Wait until we have history data for previous three months. if len(self.previous_quarter_ears) == 0: return ear_values = [x for x in self.previous_quarter_ears] top_ear_decile = np.percentile(ear_values, 90) bottom_ear_decile = np.percentile(ear_values, 10) # Open new trades. if date_to_lookup in self.earnings_data: symbols_to_trade = [symbol for symbol in self.ear_data if symbol.Value in self.earnings_data[date_to_lookup]] symbols_to_delete = [] for symbol in symbols_to_trade: # Last earnings was less than three months ago. last_earings_date = self.ear_data[symbol][0] if last_earings_date >= self.Time - timedelta(days = 90): security = self.Securities[symbol] if security.Price != 0 and security.IsTradable: if self.ear_data[symbol][1] >= top_ear_decile: self.trade_manager.Add(symbol, True) symbols_to_delete.append(symbol) elif self.ear_data[symbol][1] <= bottom_ear_decile: self.trade_manager.Add(symbol, False) symbols_to_delete.append(symbol) # Delete already traded symbols from symbol to trade. for symbol in symbols_to_delete: del self.ear_data[symbol] def Selection(self): self.selection_flag = True if self.Time.month % 3 == 0: # Store previous quarter's history. self.previous_quarter_ears = [x for x in self.current_quarter_ears] self.current_quarter_ears.clear() class SymbolData(): def __init__(self, period): self.closes = deque(maxlen = period) self.times = deque(maxlen = period) def update(self, time, close): self.times.append(time) self.closes.append(close) def is_ready(self): return len(self.closes) == self.closes.maxlen and len(self.times) == self.times.maxlen def get_prices(self, list_period): return_prices = [] for time, close in zip(self.times, self.closes): if time in list_period: return_prices.append(close) # check if there are enough data for performance calculation if len(return_prices) < 2: return None return (return_prices[-1] - return_prices[0]) / return_prices[0]