# source: https://raw.githubusercontent.com/djienne/YOUTUBE_STRATEGIES_FREQTRADE/5c660dd6211bab4ea894f6078b8ac99ed8014144/Experiences/4Mousquetaires/user_data/strategies/QuatreMousquetaires.py """ Bitcoin Intermarket Strategy with NASDAQ, S&P 500, M2, and DXY - Improved Version This strategy is based on the intramarket differencing technique presented by Neurotrader in his video. I would like to thank Neurotrader for sharing this algorithm and his insights into intermarket analysis. The original strategy compared ETH to BTC, while this extended version compares BTC to four different markets: NASDAQ, S&P 500, M2 Money Supply, and the Dollar Index (DXY). This allows for a more comprehensive analysis of BTC's behavior relative to traditional financial markets. This strategy is implemented for use with the Freqtrade trading bot framework. """ # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- from warnings import simplefilter import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame from functools import reduce from typing import Optional from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter) from freqtrade.exchange import timeframe_to_seconds from freqtrade.persistence import Trade # -------------------------------- # Add your lib to import here import pandas_ta as ta from tvDatafeed import TvDatafeed, Interval from datetime import datetime import time import warnings import logging import os warnings.filterwarnings( 'ignore', message='The objective has been evaluated at this point before.') simplefilter(action="ignore", category=pd.errors.PerformanceWarning) pd.set_option('display.max_rows', None) pd.set_option('display.max_columns', None) pd.set_option('display.width', None) pd.set_option('display.max_colwidth', None) pd.options.mode.chained_assignment = None # logging.disable(logging.CRITICAL) logger = logging.getLogger(__name__) # Constants for data caching DATA_CACHE_HOURS = 24 # How many hours to cache the external data DEFAULT_DATA_DIR = "./user_data/tv_data" # Default directory for data storage def ensure_directory_exists(directory_path): """ Ensure the directory exists, create it if it doesn't Parameters: ----------- directory_path : str Path to directory """ if not os.path.exists(directory_path): try: os.makedirs(directory_path) logger.info(f"Directory '{directory_path}' created successfully") except Exception as e: logger.error(f"Error creating directory '{directory_path}': {e}") raise def threshold_revert_signal(ind_in, threshold): """ Generate signal based on threshold crossings with reversion to zero Parameters: ----------- ind_in : Series Input indicator values threshold : float Threshold for signal generation Returns: -------- numpy.ndarray Signal values (-1, 0, or 1) """ ind = ind_in.copy() ind = ind.reset_index(drop=True) # Initialize the signal array and current position. signal = np.zeros(len(ind)) position = 0 # Iterate over the indicator values to assign signals. for i in range(len(ind)): if ind[i] > threshold: position = 1 if ind[i] < -threshold: position = -1 # Revert to 0 when the indicator crosses back through zero. if position == 1 and ind[i] <= 0: position = 0 if position == -1 and ind[i] >= 0: position = 0 signal[i] = position return signal def cmma(ohlc, lookback, atr_lookback): """ Compute a normalized indicator: (close - moving average) / (ATR * sqrt(lookback)) Parameters: ----------- ohlc : DataFrame OHLC data with 'high', 'low', 'close' columns lookback : int Lookback period for the moving average atr_lookback : int Lookback period for the ATR calculation Returns: -------- Series Normalized indicator values """ # Compute ATR using pandas_ta with the given lookback length. atr = ta.atr(ohlc['high'], ohlc['low'], ohlc['close'], length=atr_lookback) # Compute the simple moving average of the close prices. ma = ohlc['close'].rolling(lookback, center=False, min_periods=1).mean() # Calculate the indicator: (close - moving average) normalized by (ATR * sqrt(lookback)) ind = (ohlc['close'] - ma) / (atr * np.sqrt(lookback)) return ind def download_process_from_tradingview(symbol, exchange, interval, tf, n_bars, output_file, lookback, atr_lookback, data_dir=DEFAULT_DATA_DIR): """ Download and process data from TradingView Parameters: ----------- symbol : str Symbol to download exchange : str Exchange for the symbol interval : Interval TradingView interval tf : str Timeframe for the resulting data (timeframe used by the Freqtrade strategy) n_bars : int Number of bars to download output_file : str Path to save the downloaded data lookback : int Lookback period for the CMMA calculation atr_lookback : int Lookback period for the ATR calculation data_dir : str Directory to store data Returns: -------- DataFrame Processed data with CMMA indicator """ # Construct full output path full_output_path = os.path.join(data_dir, output_file) folder_path = os.path.dirname(full_output_path) # Check if the file exists and when it was last modified use_existing_file = False if os.path.exists(full_output_path): file_mod_time = os.path.getmtime(full_output_path) current_time = time.time() time_diff = current_time - file_mod_time # If file was modified less than DATA_CACHE_HOURS ago, use the existing file if time_diff < (DATA_CACHE_HOURS * 3600): # Convert hours to seconds use_existing_file = True # Download fresh data if needed if not use_existing_file: # Initialize the tvDatafeed instance tv = TvDatafeed() # Get historical data data = tv.get_hist(symbol=symbol, exchange=exchange, interval=interval, n_bars=n_bars, extended_session=True) #print(data.head()) data.reset_index(drop=False, inplace=True) data['datetime'] = pd.to_datetime(data['datetime'] ) data['datetime'] = data['datetime'].dt.tz_localize('UTC') data = data.rename(columns={'datetime': 'date'}) # Save the data to CSV data.to_csv(full_output_path) # Process data regardless of source out_data = pd.read_csv(full_output_path) # gets loaded with a date column that is datetime UTC because format is like 2025-03-14 01:00:00+00:00 out_data['cmma'] = cmma(out_data, lookback, atr_lookback) out_data.set_index('date', inplace=True) out_data.index = pd.to_datetime(out_data.index) # Keep only the cmma column out_data = out_data[['cmma']] # Create full index for the period and forward fill missing values full_index = pd.date_range(start=out_data.index.min(), end=out_data.index.max(), freq=tf) out_data = out_data.reindex(full_index).ffill() return out_data def download_process_from_m2_weekly(tf, n_bars, output_file, lookback, atr_lookback, data_dir=DEFAULT_DATA_DIR): """ Download and process M2 Money Supply data from FRED Parameters: ----------- tf : str Timeframe for the resulting data n_bars : int Number of bars to download output_file : str Path to save the downloaded data lookback : int Lookback period for the CMMA calculation atr_lookback : int Lookback period for the ATR calculation data_dir : str Directory to store data Returns: -------- DataFrame Processed M2 Money Supply data with CMMA indicator """ # Calculate adjusted lookback periods for weekly data # Use timeframe_to_seconds to get seconds in the timeframe seconds_in_tf = timeframe_to_seconds(tf) seconds_in_week = 7 * 24 * 60 * 60 # Calculate the ratio of timeframe to weekly tf_to_weekly_ratio = seconds_in_tf / seconds_in_week # Adjust lookback periods adjusted_lookback = max(4, int(lookback * tf_to_weekly_ratio)) adjusted_atr_lookback = max(4, int(atr_lookback * tf_to_weekly_ratio)) return download_process_from_tradingview( "WM2NS", "FRED", Interval.in_weekly, tf, n_bars, output_file, adjusted_lookback, adjusted_atr_lookback, data_dir ) def download_process_nasdaq(tf, n_bars, output_file, lookback, atr_lookback, data_dir=DEFAULT_DATA_DIR): """ Download and process NASDAQ data from TradingView Parameters: ----------- tf : str Timeframe for the resulting data n_bars : int Number of bars to download output_file : str Path to save the downloaded data lookback : int Lookback period for the CMMA calculation atr_lookback : int Lookback period for the ATR calculation data_dir : str Directory to store data Returns: -------- DataFrame Processed NASDAQ data with CMMA indicator """ # Calculate adjusted lookback periods for daily data # Use timeframe_to_seconds to get seconds in the timeframe seconds_in_tf = timeframe_to_seconds(tf) seconds_in_day = 24 * 60 * 60 # Calculate the ratio of timeframe to daily tf_to_daily_ratio = seconds_in_tf / seconds_in_day # Adjust lookback periods adjusted_lookback = max(4, int(lookback * tf_to_daily_ratio)) adjusted_atr_lookback = max(4, int(atr_lookback * tf_to_daily_ratio)) return download_process_from_tradingview( "NQ1!", "CME_MINI", Interval.in_daily, tf, n_bars, output_file, adjusted_lookback, adjusted_atr_lookback, data_dir ) def download_process_sp500(tf, n_bars, output_file, lookback, atr_lookback, data_dir=DEFAULT_DATA_DIR): """ Download and process S&P 500 data from TradingView Parameters: ----------- tf : str Timeframe for the resulting data n_bars : int Number of bars to download output_file : str Path to save the downloaded data lookback : int Lookback period for the CMMA calculation atr_lookback : int Lookback period for the ATR calculation data_dir : str Directory to store data Returns: -------- DataFrame Processed S&P 500 data with CMMA indicator """ # Calculate adjusted lookback periods for daily data # Use timeframe_to_seconds to get seconds in the timeframe seconds_in_tf = timeframe_to_seconds(tf) seconds_in_day = 24 * 60 * 60 # Calculate the ratio of timeframe to daily tf_to_daily_ratio = seconds_in_tf / seconds_in_day # Adjust lookback periods adjusted_lookback = max(4, int(lookback * tf_to_daily_ratio)) adjusted_atr_lookback = max(4, int(atr_lookback * tf_to_daily_ratio)) return download_process_from_tradingview( "ES1!", "CME_MINI", Interval.in_daily, tf, n_bars, output_file, adjusted_lookback, adjusted_atr_lookback, data_dir ) def download_process_dxy(tf, n_bars, output_file, lookback, atr_lookback, data_dir=DEFAULT_DATA_DIR): """ Download and process Dollar Index (DXY) data from TradingView Parameters: ----------- tf : str Timeframe for the resulting data n_bars : int Number of bars to download output_file : str Path to save the downloaded data lookback : int Lookback period for the CMMA calculation atr_lookback : int Lookback period for the ATR calculation data_dir : str Directory to store data Returns: -------- DataFrame Processed Dollar Index data with CMMA indicator """ # Calculate adjusted lookback periods for daily data # Use timeframe_to_seconds to get seconds in the timeframe seconds_in_tf = timeframe_to_seconds(tf) seconds_in_day = 24 * 60 * 60 # Calculate the ratio of timeframe to daily tf_to_daily_ratio = seconds_in_tf / seconds_in_day # Adjust lookback periods adjusted_lookback = max(4, int(lookback * tf_to_daily_ratio)) adjusted_atr_lookback = max(4, int(atr_lookback * tf_to_daily_ratio)) return download_process_from_tradingview( "DX1!", "ICEUS", Interval.in_daily, tf, n_bars, output_file, adjusted_lookback, adjusted_atr_lookback, data_dir ) def reindex_asof(df, new_index): """ Aligns df to new_index by performing an asof merge. For each timestamp in new_index, the last available row in df that is less than or equal to that timestamp is used. """ df = df.sort_index() new_index_df = pd.DataFrame({'date': new_index}) df.index.name = 'date' df = df.reset_index(drop=False) aligned = pd.merge_asof(new_index_df, df, on='date', direction='backward') aligned.set_index('date', inplace=True) return aligned class Github_djienne_YOUTUBE_STRATEGIES_FREQTRADE__QuatreMousquetaires__20250608_082133(IStrategy): """ Bitcoin Intermarket Strategy with NASDAQ, S&P 500, M2, and DXY ("Quatre Mousquetaires") This strategy compares BTC to four different markets (NASDAQ, S&P 500, M2 Money Supply, DXY) and generates trading signals based on the differences between them. WARNING: This strategy is primarily designed for daily timeframes. While efforts have been made to adapt it to other timeframes, optimal performance is expected when using the 1d timeframe. """ INTERFACE_VERSION = 3 max_entry_position_adjustment = 100 # Allow adjustments (for 25%, 50%, 75%, 100%) position_adjustment_enable: bool = True # Enable position adjustment can_short: bool = False # Disable shorting - long only use_custom_stoploss: bool = False process_only_new_candles = True debug_mode = False # Ignore ROI parameter minimal_roi = { "0": 500.00 # 50000% i.e., useless } # Stoploss designed for the strategy stoploss = -0.95 # Trailing stoploss trailing_stop = False # Optimal timeframe for the strategy timeframe = '1d' # Number of candles the strategy requires before producing valid signals # This is increased to accommodate the indicators calculation requirements startup_candle_count: int = 205 # Order type mapping order_types = { 'entry': 'market', 'exit': 'market', 'stoploss': 'market', 'stoploss_on_exchange': False } # Order time in force order_time_in_force = { 'entry': 'gtc', 'exit': 'gtc' } # Strategy parameters lookback_btc = IntParameter(14, 100, default=53, space='buy', optimize=True) lookback_nasdaq = IntParameter(14, 100, default=53, space='buy', optimize=True) lookback_sp500 = IntParameter(14, 100, default=53, space='buy', optimize=True) lookback_m2 = IntParameter(14, 100, default=53, space='buy', optimize=True) lookback_dxy = IntParameter(14, 100, default=53, space='buy', optimize=True) atr_lookback = IntParameter(14, 200, default=168, space='buy', optimize=True) threshold_nasdaq = DecimalParameter(0.02, 1.00, decimals=2, default=0.52, space='buy', optimize=True) threshold_sp500 = DecimalParameter(0.02, 1.00, decimals=2, default=0.52, space='buy', optimize=True) threshold_m2 = DecimalParameter(0.02, 1.00, decimals=2, default=0.52, space='buy', optimize=True) threshold_dxy = DecimalParameter(0.02, 1.00, decimals=2, default=0.52, space='buy', optimize=True) flip = CategoricalParameter([-1, 1], default=1, space='buy', optimize=True) fraction_from_number_signals = CategoricalParameter([0.25, 0.5], default=0.25, space='buy', optimize=True) # Market enable/disable parameters use_nasdaq = BooleanParameter(default=True, space='sell', optimize=False) use_sp500 = BooleanParameter(default=True, space='sell', optimize=False) use_m2 = BooleanParameter(default=True, space='sell', optimize=False) use_dxy = BooleanParameter(default=True, space='sell', optimize=False) def bot_start(self, **kwargs) -> None: """ Called at bot initialization and startup. Used for one-time tasks. Parameters: ----------- **kwargs : dict Additional parameters """ # Override data directory if provided in the config self.data_directory = self.config.get('user_data_dir', DEFAULT_DATA_DIR) self.data_directory = os.path.join(self.data_directory, 'tv_data') # Create data directory if it doesn't exist ensure_directory_exists(self.data_directory) # Log warning if not using daily timeframe if self.timeframe != '1d': raise ValueError( f"strategy is for daily timeframe (1d). " f"Current timeframe is set to {self.timeframe}." ) def debug_write_dataframe(self, dataframe, pair, stage): """ Write dataframe to CSV file for debugging Parameters: ----------- dataframe : DataFrame Dataframe to write pair : str Pair name stage : str Stage name for the file (e.g., "initial", "processed", "final") """ if not self.debug_mode: return if not self.config['runmode'].value == 'backtest': return try: # Create a safe filename from the pair safe_pair = pair.replace('/', '_') # Create the filename with timestamp timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = os.path.join(self.data_directory, f"{safe_pair}_{stage}_{timestamp}.csv") # Save dataframe to CSV dataframe.to_csv(filename) logger.info(f"Debug data written to {filename}") except Exception as e: logger.error(f"Error writing debug data: {e}") def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame: """ Populate indicators for the strategy Parameters: ----------- df : DataFrame OHLC data for the current pair metadata : dict Pair metadata Returns: -------- DataFrame OHLC data with added indicators """ # Create a copy of the dataframe to avoid modifying the original data = df.copy() # Create a working copy with DatetimeIndex for calculations working_df = data.copy() # Calculate CMMA for BTC btc_cmma = cmma(working_df, self.lookback_btc.value, self.atr_lookback.value) working_df['cmma'] = btc_cmma working_df.set_index('date', inplace=True) # Download and process data for the four markets n_bars = 10000 # Large enough to be sure to cover the backtest period # NASDAQ data nasdaq_data = download_process_nasdaq( self.timeframe, n_bars, 'NASDAQ_daily_data.csv', self.lookback_nasdaq.value, self.atr_lookback.value, self.data_directory ) # S&P 500 data sp500_data = download_process_sp500( self.timeframe, n_bars, 'SP500_daily_data.csv', self.lookback_sp500.value, self.atr_lookback.value, self.data_directory ) # M2 Money Supply data m2_data = download_process_from_m2_weekly( self.timeframe, int(n_bars/7), 'M2_weekly_data.csv', self.lookback_m2.value, self.atr_lookback.value, self.data_directory ) # Dollar Index data dxy_data = download_process_dxy( self.timeframe, n_bars, 'DXY_daily_data.csv', self.lookback_dxy.value, self.atr_lookback.value, self.data_directory ) working_df.fillna(0, inplace=True) # Align data (reindex) with the Freqtrade dataframe with date as index (working_df, daily index) # shift 1 day to protect against potential forward looking and it should not affect the performance nasdaq_data_aligned = reindex_asof(nasdaq_data, working_df.index) nasdaq_data_aligned = nasdaq_data_aligned.shift(1) sp500_data_aligned = reindex_asof(sp500_data, working_df.index) sp500_data_aligned = sp500_data_aligned.shift(1) m2_data_aligned = reindex_asof(m2_data, working_df.index) m2_data_aligned = m2_data_aligned.shift(7) dxy_data_aligned = reindex_asof(dxy_data, working_df.index) dxy_data_aligned = dxy_data_aligned.shift(1) # Calculate intermarket differences and signals if self.use_nasdaq.value: nasdaq_diff = nasdaq_data_aligned['cmma'] - working_df['cmma'] working_df['nasdaq_signal'] = self.flip.value * threshold_revert_signal(nasdaq_diff, self.threshold_nasdaq.value) else: working_df['nasdaq_signal'] = 0 if self.use_sp500.value: sp500_diff = sp500_data_aligned['cmma'] - working_df['cmma'] working_df['sp500_signal'] = self.flip.value * threshold_revert_signal(sp500_diff, self.threshold_sp500.value) else: working_df['sp500_signal'] = 0 if self.use_m2.value: m2_diff = m2_data_aligned['cmma'] - working_df['cmma'] working_df['m2_signal'] = self.flip.value * threshold_revert_signal(m2_diff, self.threshold_m2.value) else: working_df['m2_signal'] = 0 if self.use_dxy.value: dxy_diff = dxy_data_aligned['cmma'] - working_df['cmma'] working_df['dxy_signal'] = -1.0 * self.flip.value * threshold_revert_signal(dxy_diff, self.threshold_dxy.value) else: working_df['dxy_signal'] = 0 # Count active long signals only working_df['active_long_signals'] = ((working_df['nasdaq_signal'] > 0).astype(int) + (working_df['sp500_signal'] > 0).astype(int) + (working_df['m2_signal'] > 0).astype(int) + (working_df['dxy_signal'] > 0).astype(int)) # Signal is active if we have any active long signals working_df['signal'] = np.zeros(len(working_df)) working_df.loc[working_df['active_long_signals'] > 0, 'signal'] = 1 data.set_index('date', inplace=True) # Copy all new columns from working dataframe to the original dataframe for col in working_df.columns: if col not in data.columns or col in ['cmma', 'nasdaq_signal', 'sp500_signal', 'm2_signal', 'dxy_signal', 'active_long_signals', 'signal']: data[col] = working_df[col] # Restore the original index type data.reset_index(drop=False, inplace=True) # Debug output self.debug_write_dataframe(data, metadata['pair'], "final_output") return data def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Populate entry signals Parameters: ----------- dataframe : DataFrame OHLC data with indicators metadata : dict Pair metadata Returns: -------- DataFrame OHLC data with entry signals """ dataframe.loc[ (dataframe['signal'] == 1), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Populate exit signals Parameters: ----------- dataframe : DataFrame OHLC data with indicators metadata : dict Pair metadata Returns: -------- DataFrame OHLC data with exit signals """ dataframe.loc[ (dataframe['signal'] == 0), 'exit_long' ] = 1 return dataframe def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ Determine initial position size based on number of active signals. Parameters: ----------- pair : str Current pair current_time : datetime Current date and time current_rate : float Current rate for the pair proposed_stake : float Stake amount proposed by the bot min_stake : float or None Minimum stake amount max_stake : float Maximum stake amount leverage : float Leverage being used entry_tag : str or None Entry tag for the trade side : str Side of the trade ('long' or 'short') **kwargs : Other arguments Returns: -------- float Stake amount to use """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: logger.warning(f"Empty dataframe for {pair}, using proposed stake") return proposed_stake if self.config['runmode'].value in ('live', 'dry_run'): last_candle = dataframe.iloc[-2].squeeze() else: last_candle = dataframe.iloc[-1].squeeze() active_signals = last_candle['active_long_signals'] position_pct = min(active_signals * self.fraction_from_number_signals.value, 1.0) # Calculate stake amount stake_amount = max_stake * position_pct return stake_amount def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs) -> Optional[float]: """ Adjust position size based on number of active signals. Parameters: ----------- trade : Trade Current trade object current_time : datetime Current date and time current_rate : float Current rate for the pair current_profit : float Current profit/loss ratio min_stake : float or None Minimum stake amount max_stake : float Maximum stake amount current_entry_rate : float Current entry rate current_exit_rate : float Current exit rate current_entry_profit : float Current entry profit current_exit_profit : float Current exit profit **kwargs : Other arguments Returns: -------- float or None Adjustment to stake amount (positive to increase, negative to decrease, None for no adjustment) """ if not trade.is_open: return None dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if dataframe.empty: logger.warning(f"Empty dataframe for {trade.pair}, no position adjustment") return None if self.config['runmode'].value in ('live', 'dry_run'): last_candle = dataframe.iloc[-2].squeeze() else: last_candle = dataframe.iloc[-1].squeeze() active_signals = last_candle['active_long_signals'] # Calculate target stake amount based on number of active signals position_pct = min(active_signals * self.fraction_from_number_signals.value, 1.0) target_stake = max_stake * position_pct # Calculate adjustment needed current_stake = trade.stake_amount adjustment = target_stake - current_stake if adjustment < min_stake*1.05: return None return adjustment