BTC intraday trading systems

Here's a robust QuantConnect QCAlgorithm template for you, Eduardo. This example covers the key aspects you mentioned and is designed for learning and for future expansion:

from AlgorithmImports import *
from enum import Enum, auto
import numpy as np
import pandas as pd

class Regime(Enum):
    momentum = auto()
    mean_reversion = auto()
    neutral = auto()

class RegimeDetector:
    """Detects market regimes using volatility, trend strength, and (optionally) an HMM."""

    def __init__(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
        self._algo = algorithm
        self._symbol = symbol

        # --- indicators
        self._atr = AverageTrueRange(14)
        self._adx = AverageDirectionalIndex(14)
        self._std = StandardDeviation(14)
        self._algo.register_indicator(symbol, self._atr, Resolution.MINUTE)
        self._algo.register_indicator(symbol, self._adx, Resolution.MINUTE)
        self._algo.register_indicator(symbol, self._std, Resolution.MINUTE)

        # manual rolling-window for std‐dev %
        self._window = RollingWindow[float](14)

        # user-tunable thresholds
        self._vol_threshold = 0.002      # 0.20 % of price
        self._trend_threshold = 25        # ADX threshold

        self._current_regime: Regime = Regime.neutral

    @property
    def current_regime(self) -> Regime:
        return self._current_regime

    def update(self, bar: TradeBar) -> Regime:
        # --- keep a rolling std-dev
        self._window.add(bar.close)
        if self._window.is_ready:
            std_value = np.std(list(self._window))
            self._std.update(bar.end_time, std_value)

        # wait until everything is ready
        if not (self._atr.is_ready and self._adx.is_ready and self._std.is_ready):
            return self._current_regime
        if bar.close == 0:
            return self._current_regime

        atr_pct = self._atr.current.value / bar.close
        std_pct = self._std.current.value / bar.close
        vol_high = atr_pct > self._vol_threshold or std_pct > self._vol_threshold
        trend_strong = self._adx.current.value > self._trend_threshold

        self._current_regime = Regime.momentum if (vol_high and trend_strong) else Regime.mean_reversion
        return self._current_regime

    # ---- optional Hidden-Markov-Model placeholder ----
    def hmm_detect(self, features: np.ndarray) -> Regime:
        # sklearn / hmmlearn are unavailable live – train offline & load mapping here if desired
        return self._current_regime

class BaseStrategy:
    def __init__(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
        self._algo = algorithm
        self._symbol = symbol

    def update(self, data: Slice) -> None:
        raise NotImplementedError

    def liquidate(self) -> None:
        if self._algo.portfolio[self._symbol].invested:
            self._algo.liquidate(self._symbol, "Regime change liquidation")

class MomentumEMAStrategy(BaseStrategy):
    """Fast/slow EMA cross momentum strategy."""

    def __init__(self, algorithm: QCAlgorithm, symbol: Symbol,
                 fast: int = 10, slow: int = 30) -> None:
        super().__init__(algorithm, symbol)
        self._fast = ExponentialMovingAverage(fast)
        self._slow = ExponentialMovingAverage(slow)
        algorithm.register_indicator(symbol, self._fast, Resolution.MINUTE)
        algorithm.register_indicator(symbol, self._slow, Resolution.MINUTE)

    def update(self, data: Slice) -> None:
        if not (self._fast.is_ready and self._slow.is_ready):
            return
        holdings = self._algo.portfolio[self._symbol].quantity
        if self._fast.current.value > self._slow.current.value * 1.0005:
            if holdings <= 0:
                self._algo.set_holdings(self._symbol, 1.0)
        elif self._fast.current.value < self._slow.current.value * 0.9995:
            if holdings >= 0:
                self._algo.set_holdings(self._symbol, -1.0)
        elif holdings != 0:
            self._algo.liquidate(self._symbol, "Flat EMA crossover")

class MeanReversionRSIStrategy(BaseStrategy):
    """RSI mean-reversion strategy."""

    def __init__(self, algorithm: QCAlgorithm, symbol: Symbol,
                 period: int = 14, oversold: float = 30, overbought: float = 70) -> None:
        super().__init__(algorithm, symbol)
        self._rsi = RelativeStrengthIndex(period)
        algorithm.register_indicator(symbol, self._rsi, Resolution.MINUTE)
        self._oversold, self._overbought = oversold, overbought

    def update(self, data: Slice) -> None:
        if not self._rsi.is_ready:
            return
        rsi = self._rsi.current.value
        holdings = self._algo.portfolio[self._symbol].quantity
        if rsi < self._oversold and holdings <= 0:
            self._algo.set_holdings(self._symbol, 1.0)
        elif rsi > self._overbought and holdings >= 0:
            self._algo.set_holdings(self._symbol, -1.0)

class RegimeSwitchingTemplate(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100000)

        # execution feed (live): BTC-USDT minute data
        self._symbol = self.add_crypto("BTCUSDT", Resolution.MINUTE).symbol

        # warm-up for indicators
        self.set_warm_up(timedelta(days=2))

        # ---------- research-only (offline) PKL load ----------
        self._research_df = self._load_research_data("research_btc_intraday.pkl")

        # ---------- modules ----------
        self._detector = RegimeDetector(self, self._symbol)
        self._mom_strategy = MomentumEMAStrategy(self, self._symbol)
        self._mean_rev_strategy = MeanReversionRSIStrategy(self, self._symbol)

        # ---------- plotting ----------
        chart = Chart("Regime")
        chart.add_series(Series("Regime", SeriesType.LINE, 0))
        self.add_chart(chart)

    # main event-loop
    def on_data(self, data: Slice) -> None:
        if self.is_warming_up or self._symbol not in data.bars:
            return

        bar = data.bars[self._symbol]
        regime = self._detector.update(bar)

        # plot regime value (1 = momentum, –1 = mean-rev)
        self.plot("Regime", "Regime", float(regime.value))

        if regime == Regime.momentum:
            self._mean_rev_strategy.liquidate()
            self._mom_strategy.update(data)
        elif regime == Regime.mean_reversion:
            self._mom_strategy.liquidate()
            self._mean_rev_strategy.update(data)
        else:
            self._mom_strategy.liquidate()
            self._mean_rev_strategy.liquidate()

    # ---------- helpers ----------
    def _load_research_data(self, filename: str) -> pd.DataFrame:
        """Read a local pickle file for offline analysis (no live dependency)."""
        import os
        path = os.path.join(os.getcwd(), filename)
        if not os.path.exists(path):
            self.debug(f"Research data '{filename}' not found.")
            return pd.DataFrame()
        try:
            return pd.read_pickle(path)
        except Exception as err:
            self.error(f"Failed loading research data: {err}")
            return pd.DataFrame()

Key Features:

• Loads a local research PKL for offline research (does not use it for trading).
• Flexible regime switching logic—tweak thresholds for sensitivity.
• Modular design—easy to expand, swap or stack regimes and strategies.
• Logs and plots regime changes for analysis.
• Well-commented for educational use and experimentation.

Feel free to ask for more specific implementations (such as more advanced regime detectors, custom backtesting analyzers, etc.) or join the discussion with your own improvements and insights!

Sources: Research Guide - QuantConnect.com, Key Concepts - QuantConnect.com, Hidden Markov Models - QuantConnect.com, Overview - QuantConnect.com, docs writing algorithms indicators automatic indicators