Overall Statistics
Total Orders
1026
Average Win
0.11%
Average Loss
-0.10%
Compounding Annual Return
20.861%
Drawdown
18.800%
Expectancy
0.198
Start Equity
100000
End Equity
120923.95
Net Profit
20.924%
Sharpe Ratio
0.794
Sortino Ratio
0.837
Probabilistic Sharpe Ratio
38.979%
Loss Rate
41%
Win Rate
59%
Profit-Loss Ratio
1.04
Alpha
0.053
Beta
0.636
Annual Standard Deviation
0.196
Annual Variance
0.039
Information Ratio
-0.041
Tracking Error
0.132
Treynor Ratio
0.245
Total Fees
$1028.11
Estimated Strategy Capacity
$5900000.00
Lowest Capacity Asset
BRKB R735QTJ8XC9X
Portfolio Turnover
3.18%
Drawdown Recovery
117
 
from AlgorithmImports import *
from hmmlearn import hmm


class HMMRegimeDetection(QCAlgorithm):
    """
    Hidden Markov Model (HMM) based regime detection.
    
    Uses QuantConnect Algorithm Framework structure:
    https://www.quantconnect.com/docs/v2/writing-algorithms/algorithm-framework/overview

    Based on:
    https://www.quantconnect.com/research/17900/intraday-application-of-hidden-markov-models/p1
    """
    def initialize(self):
        # keep in mind data snooping bias:
        # always leave some (typically more recent) time period for testing!
        #self.set_start_date(2017, 1, 1)
        #self.set_end_date(2022, 12, 31)
        self.set_start_date(2020, 1, 1)
        self.set_end_date(2020, 12, 31)

        # key settings
        self.minute_bars = 30
        self.roc_window_size = 25
        n_top_stocks = 10
        
        # dynamic universe selection
        self.add_universe(lambda fundamental: [f.symbol for f in sorted(fundamental, key=lambda x: x.market_cap, reverse=True)[:n_top_stocks]])
        
        # alternative: static universe (uncomment below instead)
        #tickers = ["QQQ", "IWM", "GLD", "TLT"]
        #symbols = [self.add_equity(ticker, Resolution.MINUTE).Symbol for ticker in tickers]
        #self.add_universe_selection(ManualUniverseSelectionModel(symbols))

        self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel(timedelta(minutes=self.minute_bars)))
        self.settings.rebalance_portfolio_on_insight_changes = False
        self.set_warm_up(timedelta(self.minute_bars))
    

    def on_securities_changed(self, changes):
        for added in changes.added_securities:
            added.roc = RateOfChange(1)
            added.roc.window.size = self.roc_window_size
            
            # HMM detection assuming 3 regimes
            added.model = hmm.GaussianHMM(n_components = 3, n_iter = 100, random_state = 100)
            
            added.model_month = -1
            added.consolidator = TradeBarConsolidator(timedelta(minutes = self.minute_bars))
            added.consolidator.data_consolidated += self.on_consolidated
            self.subscription_manager.add_consolidator(added.symbol, added.consolidator)
    

    def on_consolidated(self, _, bar):
        security = self.securities[bar.symbol]
        security.roc.update(bar.end_time, bar.price)
        if security.roc.window.is_ready:
            if security.model_month != bar.end_time.month:
                security.model.fit(np.array([point.value for point in security.roc.window])[::-1].reshape(-1, 1))
                security.model_month = bar.end_time.month
            post_prob = security.model.predict_proba(np.array([security.roc.current.value]).reshape(1, -1)).flatten()
            
            if post_prob[2] > post_prob[0]:
                direction = InsightDirection.UP
            else:
                direction = InsightDirection.DOWN
            
            self.emit_insights(Insight.price(bar.symbol, timedelta(minutes = self.minute_bars), direction))