| Overall Statistics |
|
Total Orders 1624 Average Win 0.17% Average Loss -0.23% Compounding Annual Return 2.956% Drawdown 15.300% Expectancy 0.170 Start Equity 100000000 End Equity 154202574.08 Net Profit 54.203% Sharpe Ratio 0.088 Sortino Ratio 0.058 Probabilistic Sharpe Ratio 0.100% Loss Rate 33% Win Rate 67% Profit-Loss Ratio 0.75 Alpha -0.02 Beta 0.296 Annual Standard Deviation 0.069 Annual Variance 0.005 Information Ratio -0.722 Tracking Error 0.114 Treynor Ratio 0.02 Total Fees $832244.39 Estimated Strategy Capacity $3700000000.00 Lowest Capacity Asset QQQ RIWIV7K5Z9LX Portfolio Turnover 3.72% |
# region imports
from AlgorithmImports import *
# endregion
class SwimmingLightBrownGalago(QCAlgorithm):
def initialize(self):
self.set_cash(100_000_000)
self.settings.daily_precise_end_time = False
self.settings.minimum_order_margin_portfolio_percentage = 0
lookback_years = 12 # 12 years includes 3 non-leap years & 3 election cycles
self._lookback = lookback_years * 252
self._period = 21 # Trading days
warm_up_period = timedelta(lookback_years * 365)
self.set_start_date(datetime(1998, 1, 1) + warm_up_period)
self.set_warm_up(warm_up_period)
self._vix = self.add_data(CBOE, "VIX", Resolution.DAILY)
self._vix.threshold = 30
tickers = ['SPY', 'QQQ']
self._equities = [self.add_equity(ticker, Resolution.DAILY, fill_forward=False) for ticker in tickers]
def on_securities_changed(self, changes):
for security in changes.added_securities:
security.history = pd.Series()
security.rebalance = False
security.signal = 0
security.previous_weight = 0
def on_data(self, data: Slice):
self.plot('VIX', 'Value', self._vix.price)
for equity in self._equities:
# Get the current day's trade bar.
bar = data.bars.get(equity.symbol)
if bar:
# Update the historical data.
equity.history.loc[bar.time] = bar.open # `time` and `open` since we trade at next market open.
equity.history = equity.history.iloc[-self._lookback:]
# Wait until there is enough history.
if len(equity.history) < self._lookback or self.is_warming_up or self._vix.price >= self._vix.threshold:
continue
# Calculate expected return of entering at next market open.
period_returns = equity.history.pct_change(self._period).shift(-self._period).dropna()
expected_return_by_day = period_returns.groupby(period_returns.index.strftime('%m-%d')).mean()
next_market_open = equity.exchange.hours.get_next_market_open(self.time, False)
expected_return = expected_return_by_day[next_market_open.strftime('%m-%d')] # The expected return of buying next market open and holdings n days.
# Calculate signal for this trade.
std = expected_return_by_day.std()
mean = expected_return_by_day.mean()
if expected_return >= mean + std:
signal = 1
elif expected_return <= mean - std:
signal = -1
else:
signal = 0
if signal:
equity.signal += signal
equity.rebalance = True
# Create a Scheduled Event to remove the signal later.
exit_day = next_market_open
for _ in range(self._period):
exit_day = equity.exchange.hours.get_next_market_open(exit_day, False)
self.schedule.on(
self.date_rules.on(exit_day.replace(hour=0, minute=0)),
self.time_rules.midnight,
lambda equity=equity, signal=signal: self._remove_signal(equity, signal)
)
equities_with_enough_history = sum([len(equity.history) >= self._lookback for equity in self._equities])
for equity in self._equities:
if equity.rebalance:
equity.rebalance = False
weight = max(0, equity.signal / self._period) # Make the strategy long-only.
# Adjust weight so that it imediately goes to 100% when scaling in.
adjusted_weight = weight
if weight and weight > equity.previous_weight:
adjusted_weight = 1
equity.previous_weight = weight
self.set_holdings(equity.symbol, adjusted_weight / equities_with_enough_history)
def _remove_signal(self, equity, signal):
equity.signal -= signal
equity.rebalance = True