import pandas as pd
from moonshot import Moonshot
from moonshot.commission import PerShareCommission
class USStockCommission(PerShareCommission):
BROKER_COMMISSION_PER_SHARE = 0.005
class QuantitativeMomentum(Moonshot):
"""
Momentum strategy modeled on Alpha Architect's QMOM ETF.
Strategy rules:
1. Universe selection
a. Starting universe: all NYSE stocks
b. Exclude financials, ADRs, REITs
c. Liquidity screen: select top N percent of stocks by dollar
volume (N=60)
2. Apply momentum screen: calculate 12-month returns, excluding
most recent month, and select N percent of stocks with best
return (N=10)
3. Filter by smoothness of momentum: of the momentum stocks, select
the N percent with the smoothest momentum, as measured by the number
of positive days in the last 12 months (N=50)
4. Apply equal weights
5. Rebalance portfolio before quarter-end to capture window-dressing seasonality effect
"""
CODE = "qmom"
DB = "sharadar-us-stk-1d"
DB_FIELDS = ["Close", "Volume"]
DOLLAR_VOLUME_TOP_N_PCT = 60
DOLLAR_VOLUME_WINDOW = 90
UNIVERSES = "nyse-stk"
EXCLUDE_UNIVERSES = ["nyse-financials", "nyse-adrs", "nyse-reits"]
MOMENTUM_WINDOW = 252
MOMENTUM_EXCLUDE_MOST_RECENT_WINDOW = 22
TOP_N_PCT = 10
SMOOTHEST_TOP_N_PCT = 50
REBALANCE_INTERVAL = "Q-NOV"
COMMISSION_CLASS = USStockCommission
def prices_to_signals(self, prices: pd.DataFrame):
closes = prices.loc["Close"]
volumes = prices.loc["Volume"]
avg_dollar_volumes = (closes * volumes).rolling(self.DOLLAR_VOLUME_WINDOW).mean()
dollar_volume_ranks = avg_dollar_volumes.rank(axis=1, ascending=False, pct=True)
have_adequate_dollar_volumes = dollar_volume_ranks <= (self.DOLLAR_VOLUME_TOP_N_PCT/100)
year_ago_closes = closes.shift(self.MOMENTUM_WINDOW)
month_ago_closes = closes.shift(self.MOMENTUM_EXCLUDE_MOST_RECENT_WINDOW)
returns = (month_ago_closes - year_ago_closes) / year_ago_closes.where(year_ago_closes != 0)
returns_ranks = returns.where(have_adequate_dollar_volumes).rank(axis=1, ascending=False, pct=True)
have_momentum = returns_ranks <= (self.TOP_N_PCT / 100)
are_positive_days = closes.pct_change() > 0
positive_days_last_twelve_months = are_positive_days.astype(int).rolling(self.MOMENTUM_WINDOW).sum()
positive_days_last_twelve_months_ranks = positive_days_last_twelve_months.where(have_momentum).rank(axis=1, ascending=False, pct=True)
have_smooth_momentum = positive_days_last_twelve_months_ranks <= (self.SMOOTHEST_TOP_N_PCT/100)
signals = have_smooth_momentum.astype(int)
return signals
def signals_to_target_weights(self, signals: pd.DataFrame, prices: pd.DataFrame):
daily_signal_counts = signals.abs().sum(axis=1)
weights = signals.div(daily_signal_counts, axis=0).fillna(0)
weights = weights.resample(self.REBALANCE_INTERVAL).last()
weights = weights.reindex(prices.loc["Close"].index, method="ffill")
return weights
def target_weights_to_positions(self, weights: pd.DataFrame, prices: pd.DataFrame):
return weights.shift()
def positions_to_gross_returns(self, positions: pd.DataFrame, prices: pd.DataFrame):
closes = prices.loc["Close"]
position_ends = positions.shift()
gross_returns = closes.pct_change() * position_ends
return gross_returns
def order_stubs_to_orders(self, orders: pd.DataFrame, prices: pd.DataFrame):
orders["Exchange"] = "SMART"
orders["OrderType"] = "MOC"
orders["Tif"] = "DAY"
return orders