From 8508cfe3b55bcc6944f210164939a07475de9231 Mon Sep 17 00:00:00 2001 From: graceyangfan <1825617022.yf@gmail.com> Date: Sat, 14 Sep 2024 20:13:16 +0800 Subject: [PATCH] Add Multi-Instrument Rotation Trading Example with Controller for Binance --- ...tnet_ema_cross_with_instrument_selector.py | 132 ++++++++++++ .../simple_binance_symbols_filter.py | 121 +++++++++++ .../simple_cross_sectional_metrics.py | 199 ++++++++++++++++++ .../simple_insturment_selector_controller.py | 199 ++++++++++++++++++ 4 files changed, 651 insertions(+) create mode 100644 examples/live/binance/binance_futures_testnet_ema_cross_with_instrument_selector.py create mode 100644 nautilus_trader/examples/strategies/simple_binance_symbols_filter.py create mode 100644 nautilus_trader/examples/strategies/simple_cross_sectional_metrics.py create mode 100644 nautilus_trader/examples/strategies/simple_insturment_selector_controller.py diff --git a/examples/live/binance/binance_futures_testnet_ema_cross_with_instrument_selector.py b/examples/live/binance/binance_futures_testnet_ema_cross_with_instrument_selector.py new file mode 100644 index 000000000000..ab9fc5a4998f --- /dev/null +++ b/examples/live/binance/binance_futures_testnet_ema_cross_with_instrument_selector.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +# ------------------------------------------------------------------------------------------------- +# Copyright (C) 2015-2024 Nautech Systems Pty Ltd. All rights reserved. +# https://nautechsystems.io +# +# Licensed under the GNU Lesser General Public License Version 3.0 (the "License"); +# You may not use this file except in compliance with the License. +# You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------------------------------------------- + + +from decimal import Decimal +from datetime import datetime + +from nautilus_trader.adapters.binance.common.enums import BinanceAccountType +from nautilus_trader.adapters.binance.config import ( + BinanceDataClientConfig, + BinanceExecClientConfig, +) +from nautilus_trader.adapters.binance.factories import ( + BinanceLiveDataClientFactory, + BinanceLiveExecClientFactory, +) +from nautilus_trader.cache.config import CacheConfig +from nautilus_trader.config import ( + ImportableControllerConfig, + InstrumentProviderConfig, + LiveExecEngineConfig, + LoggingConfig, + TradingNodeConfig, +) +from nautilus_trader.live.node import TradingNode +from nautilus_trader.model.data import BarType +from nautilus_trader.model.identifiers import InstrumentId, TraderId + + +# *** THIS IS A TEST STRATEGY WITH NO ALPHA ADVANTAGE WHATSOEVER. *** +# *** IT IS NOT INTENDED TO BE USED TO TRADE LIVE WITH REAL MONEY. *** + +# Configure the trading node +config_node = TradingNodeConfig( + trader_id=TraderId("TESTER-001"), + logging=LoggingConfig(log_level="INFO"), + exec_engine=LiveExecEngineConfig( + # debug=True, + reconciliation=True, + reconciliation_lookback_mins=1440, + # snapshot_orders=True, + # snapshot_positions=True, + # snapshot_positions_interval_secs=5.0, + ), + cache=CacheConfig( + # database=DatabaseConfig(timeout=2), + timestamps_as_iso8601=True, + flush_on_start=False, + ), + # message_bus=MessageBusConfig( + # database=DatabaseConfig(timeout=2), + # timestamps_as_iso8601=True, + # use_instance_id=False, + # # types_filter=[QuoteTick], + # stream_per_topic=False, + # external_streams=["bybit"], + # autotrim_mins=30, + # ), + controller = ImportableControllerConfig( + controller_path="nautilus_trader.examples.strategies.simple_insturment_selector_controller:BinanceFutureInstrumentSelectorController", + config_path="nautilus_trader.examples.strategies.simple_insturment_selector_controller:BinanceFutureInstrumentSelectorControllerConfig", + config={ + "interval_secs": 3600, + "min_notional_threshold": 6, + "quote_asset": "USDT", + "onboard_date_filter_type": "range", + "onboard_date_reference_date": datetime(2024, 1, 1), + "onboard_date_end_date": datetime(2024, 6, 1), + }, + ), + data_clients={ + "BINANCE": BinanceDataClientConfig( + api_key=None, # 'BINANCE_API_KEY' env var + api_secret=None, # 'BINANCE_API_SECRET' env var + account_type=BinanceAccountType.USDT_FUTURE, + base_url_http=None, # Override with custom endpoint + base_url_ws=None, # Override with custom endpoint + us=False, # If client is for Binance US + testnet=True, # If client uses the testnet + instrument_provider=InstrumentProviderConfig(load_all=True), + ), + }, + exec_clients={ + "BINANCE": BinanceExecClientConfig( + api_key=None, # 'BINANCE_API_KEY' env var + api_secret=None, # 'BINANCE_API_SECRET' env var + account_type=BinanceAccountType.USDT_FUTURE, + base_url_http=None, # Override with custom endpoint + base_url_ws=None, # Override with custom endpoint + us=False, # If client is for Binance US + testnet=True, # If client uses the testnet + instrument_provider=InstrumentProviderConfig(load_all=True), + use_position_ids=False, + max_retries=3, + retry_delay=1.0, + ), + }, + timeout_connection=30.0, + timeout_reconciliation=10.0, + timeout_portfolio=10.0, + timeout_disconnection=10.0, + timeout_post_stop=5.0, +) + +# Instantiate the node with a configuration +node = TradingNode(config=config_node) + +# Register your client factories with the node (can take user-defined factories) +node.add_data_client_factory("BINANCE", BinanceLiveDataClientFactory) +node.add_exec_client_factory("BINANCE", BinanceLiveExecClientFactory) +node.build() + + +# Stop and dispose of the node with SIGINT/CTRL+C +if __name__ == "__main__": + try: + node.run() + finally: + node.dispose() diff --git a/nautilus_trader/examples/strategies/simple_binance_symbols_filter.py b/nautilus_trader/examples/strategies/simple_binance_symbols_filter.py new file mode 100644 index 000000000000..1dcb67a0f1e8 --- /dev/null +++ b/nautilus_trader/examples/strategies/simple_binance_symbols_filter.py @@ -0,0 +1,121 @@ +# ------------------------------------------------------------------------------------------------- +# Copyright (C) 2015-2024 Nautech Systems Pty Ltd. All rights reserved. +# https://nautechsystems.io +# +# Licensed under the GNU Lesser General Public License Version 3.0 (the "License"); +# You may not use this file except in compliance with the License. +# You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------------------------------------------- + +import json +from datetime import datetime, timedelta +from typing import Optional, Tuple + +import pandas as pd +import requests + + +def extract_symbol_info(api_url: str = "https://fapi.binance.com/fapi/v1/exchangeInfo") -> Tuple[Optional[pd.DataFrame], Optional[str]]: + """Fetch and extract symbol info from Binance Futures API and return it as a DataFrame. + + Args: + - api_url (str): The API URL to fetch data from (default: Binance Futures API). + + Returns: + - Tuple containing the DataFrame and an optional error message. + """ + try: + # Send a GET request to the Binance API + response = requests.get(api_url) + response.raise_for_status() + + # Load the JSON response + json_data = response.json() + + # Extract symbols data from JSON + symbols = json_data['symbols'] + df = pd.DataFrame(symbols) + + # Filter DataFrame for 'PERPETUAL' contract type and 'TRADING' status + df = df[(df['contractType'] == 'PERPETUAL') & (df['status'] == 'TRADING')] + + return df, None + + except requests.exceptions.RequestException as e: + return None, f"Request error: {e}" + except json.JSONDecodeError: + return None, "Error decoding JSON" + except KeyError as e: + return None, f"Unexpected JSON structure, missing key: {e}" + + +def select_with_quoteAsset(df: pd.DataFrame, quoteAsset: str) -> pd.DataFrame: + """Filter DataFrame for products with a specific quote asset. + + Args: + - df (pd.DataFrame): The DataFrame to filter. + - quoteAsset (str): The quote asset to filter by. + + Returns: + - pd.DataFrame: Filtered DataFrame. + """ + return df[df['quoteAsset'] == quoteAsset] + +def select_with_min_notional(df: pd.DataFrame, min_notional_threshold: float) -> pd.DataFrame: + """Add a min_notional column to the DataFrame and filter it based on a threshold. + + Args: + - df (pd.DataFrame): The DataFrame to process. + - min_notional_threshold (float): The threshold to filter min_notional values. + + Returns: + - pd.DataFrame: DataFrame with min_notional columns and filtered values. + """ + # Extract 'MIN_NOTIONAL' from filters and set it in the DataFrame + df = df.copy() + df['min_notional'] = df['filters'].apply(lambda x: next( + (float(filter_item.get('notional', 0)) for filter_item in x if filter_item.get('filterType') == 'MIN_NOTIONAL'), 0)) + # Filter the DataFrame where min_notional is less than the threshold + return df[df['min_notional'] < min_notional_threshold] + + +def filter_with_onboard_date(df: pd.DataFrame, + filter_type: str, + reference_date: datetime, + end_date: datetime = None) -> pd.DataFrame: + """Filter the DataFrame based on onboardDate according to the specified criteria. + + Args: + - df (pd.DataFrame): The DataFrame containing the onboardDate column. + - filter_type (str): The type of filter to apply. Can be 'before', 'range', or 'after'. + - reference_date (datetime): The date to use for filtering. + - end_date (datetime, optional): The end date for 'range' filter type. Required if filter_type is 'range'. + + Returns: + - pd.DataFrame: Filtered DataFrame based on the specified filter type and dates. + """ + # Convert onboardDate from milliseconds timestamp to datetime + df = df.copy() + df['onboardDate_convert'] = pd.to_datetime(df['onboardDate'], unit='ms') + + if filter_type == 'before': + # Filter the DataFrame for onboardDate earlier than the reference date + filtered_df = df[df['onboardDate_convert'] < reference_date] + elif filter_type == 'range': + if end_date is None: + raise ValueError("end_date must be provided for 'range' filter type.") + # Filter the DataFrame for onboardDate within the specified date range + filtered_df = df[(df['onboardDate_convert'] >= reference_date) & (df['onboardDate_convert'] <= end_date)] + elif filter_type == 'after': + # Filter the DataFrame for onboardDate later than the reference date + filtered_df = df[df['onboardDate_convert'] > reference_date] + else: + raise ValueError("filter_type must be one of 'before', 'range', or 'after'.") + + return filtered_df diff --git a/nautilus_trader/examples/strategies/simple_cross_sectional_metrics.py b/nautilus_trader/examples/strategies/simple_cross_sectional_metrics.py new file mode 100644 index 000000000000..b336706051f9 --- /dev/null +++ b/nautilus_trader/examples/strategies/simple_cross_sectional_metrics.py @@ -0,0 +1,199 @@ +# ------------------------------------------------------------------------------------------------- +# Copyright (C) 2015-2024 Nautech Systems Pty Ltd. All rights reserved. +# https://nautechsystems.io +# +# Licensed under the GNU Lesser General Public License Version 3.0 (the "License"); +# You may not use this file except in compliance with the License. +# You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------------------------------------------- + +import time +from datetime import datetime + +import numpy as np +import pandas as pd +import requests + +try: + from scipy.stats import linregress +except ImportError: + # If scipy is not installed, raise an error with installation instructions + raise ImportError("scipy is not installed. Please install it using 'pip install scipy'") + +def get_binance_historical_bars( + symbol='BTCUSDT', + start=datetime(2020, 8, 10), + end=datetime(2021, 8, 10), + interval='1h', + base='fapi', + version='v1' +): + """Fetch historical Kline data from the Binance API for a specific symbol and return it as a DataFrame. + + Args: + - symbol (str): The trading pair symbol. + - start (datetime): The start date as a datetime object. + - end (datetime): The end date as a datetime object. + - interval (str): The interval period for Klines (e.g., '1h'). + - base (str): The base endpoint for the Binance API. + - version (str): The API version. + + Returns: + - pd.DataFrame: DataFrame containing Kline data with correctly typed columns. + """ + klines = [] + start_time = int(start.timestamp()) * 1000 + end_time = min(int(end.timestamp()) * 1000, int(time.time() * 1000)) + + interval_map = {'m': 60 * 1000, 'h': 3600 * 1000, 'd': 86400 * 1000} + interval_ms = int(interval[:-1]) * interval_map[interval[-1]] + + while start_time < end_time: + adjusted_end_time = min(start_time + 1000 * interval_ms, end_time) + url = f'https://{base}.binance.com/{base}/{version}/klines' + params = { + 'symbol': symbol, + 'interval': interval, + 'startTime': start_time, + 'endTime': adjusted_end_time, + 'limit': 1000 + } + + try: + response = requests.get(url, params=params) + response.raise_for_status() + data = response.json() + except requests.exceptions.RequestException as e: + print(f"Request error for {symbol}: {e}") + break + + if not data: + break + + # Add new data to klines and update the start_time to avoid overlap + klines.extend(data) + # Update start_time to one millisecond past the last returned kline to avoid duplicates + start_time = data[-1][0] + interval_ms + + # Sleep to prevent hitting rate limits + time.sleep(0.1) + + columns = ['time', 'open', 'high', 'low', 'close', 'volume', 'end_time', + 'quote_asset_volume', 'number_of_trades', 'taker_buy_base_volume', 'taker_buy_quote_volume', 'ignore'] + + # Create DataFrame and convert columns to suitable types + df = pd.DataFrame(klines, columns=columns) + + # Convert columns to appropriate types + df['symbol'] = symbol + df['time'] = pd.to_datetime(df['time'], unit='ms') # Convert time to datetime + df['open'] = pd.to_numeric(df['open'], errors='coerce') + df['high'] = pd.to_numeric(df['high'], errors='coerce') + df['low'] = pd.to_numeric(df['low'], errors='coerce') + df['close'] = pd.to_numeric(df['close'], errors='coerce') + df['volume'] = pd.to_numeric(df['volume'], errors='coerce') + df['quote_asset_volume'] = pd.to_numeric(df['quote_asset_volume'], errors='coerce') + df['number_of_trades'] = pd.to_numeric(df['number_of_trades'], errors='coerce') + df['taker_buy_base_volume'] = pd.to_numeric(df['taker_buy_base_volume'], errors='coerce') + df['taker_buy_quote_volume'] = pd.to_numeric(df['taker_buy_quote_volume'], errors='coerce') + + df.set_index('time', inplace=True) + + return df + +def compute_bollinger_bands(df, period=20, num_std=2): + """Computes Bollinger Bands for a given DataFrame.""" + df['SMA'] = df['close'].rolling(window=period).mean() + df['STD_DEV'] = df['close'].rolling(window=period).std() + df['BB_upper'] = df['SMA'] + (df['STD_DEV'] * num_std) + df['BB_lower'] = df['SMA'] - (df['STD_DEV'] * num_std) + df['BB_width'] = df['BB_upper'] - df['BB_lower'] + return df + +def compute_atr(df, period=14): + """Computes the ATR for a given DataFrame.""" + df['High-Low'] = df['high'] - df['low'] + df['High-Close'] = np.abs(df['high'] - df['close'].shift()) + df['Low-Close'] = np.abs(df['low'] - df['close'].shift()) + df['TrueRange'] = df[['High-Low', 'High-Close', 'Low-Close']].max(axis=1) + df['ATR'] = df['TrueRange'].rolling(window=period).mean() + return df['ATR'] + +def calculate_slope(df, window=8): + """Calculates slope of the close price using linear regression.""" + y = df['close'].tail(window).values + x = np.arange(len(y)) + slope, _, _, _, _ = linregress(x, y) + return slope + +def generate_metrics(df_dict, period_bollinger=20, period_atr=14, slope_window=8): + """ + Computes metrics for multiple symbols and returns them as a DataFrame. + + Args: + - df_dict (dict): Dictionary of DataFrames with symbol as key and DataFrame as value. + - period_bollinger (int): Period for computing Bollinger Bands (default 20). + - period_atr (int): Period for computing ATR (default 14). + - slope_window (int): Window size for calculating slope (default 8). + + Returns: + - pd.DataFrame: DataFrame containing metrics and average score for each symbol. + """ + results = [] + + # Helper function to normalize a series + def normalize(series): + return (series - series.min()) / (series.max() - series.min()) if series.max() != series.min() else series * 0 + + for symbol, df in df_dict.items(): + # Calculate mean volatility (standard deviation of percentage changes in close prices) + mean_volatility = df['close'].pct_change().std() + + # Calculate ATR and mean NATR (normalized ATR) + df['ATR'] = compute_atr(df, period=period_atr) + mean_natr = (df['ATR'] / df['close']).mean() + + # Calculate Bollinger Bands width and mean width + df = compute_bollinger_bands(df, period=period_bollinger) + mean_bb_width = df['BB_width'].mean() + + # Calculate the latest trend using slope of close prices + latest_trend = calculate_slope(df, window=slope_window) + + # Calculate average volume per hour (assuming data is in 15-minute intervals) + avg_volume_per_hour = df['volume'].mean() * 4 + + # Store metrics in a dictionary + metrics = { + 'symbol': symbol, + 'mean_volatility': mean_volatility, + 'mean_natr': mean_natr, + 'mean_bb_width': mean_bb_width, + 'latest_trend': latest_trend, + 'avg_volume_per_hour': avg_volume_per_hour + } + + results.append(metrics) + + # Create DataFrame from the metrics list + df_metrics = pd.DataFrame(results) + # Normalize the metrics across symbols + for column in ['mean_volatility', 'mean_natr', 'mean_bb_width', 'latest_trend', 'avg_volume_per_hour']: + df_metrics[f'normalized_{column}'] = normalize(df_metrics[column]) + + # Compute average score as the mean of all normalized metrics + normalized_columns = [f'normalized_{col}' for col in ['mean_volatility', 'mean_natr', 'mean_bb_width', 'latest_trend', 'avg_volume_per_hour']] + df_metrics['average_score'] = df_metrics[normalized_columns].mean(axis=1) + + # Sort by average score and return + df_metrics = df_metrics.sort_values(by='average_score', ascending=False) + + return df_metrics + + diff --git a/nautilus_trader/examples/strategies/simple_insturment_selector_controller.py b/nautilus_trader/examples/strategies/simple_insturment_selector_controller.py new file mode 100644 index 000000000000..1958afffaead --- /dev/null +++ b/nautilus_trader/examples/strategies/simple_insturment_selector_controller.py @@ -0,0 +1,199 @@ +# ------------------------------------------------------------------------------------------------- +# Copyright (C) 2015-2024 Nautech Systems Pty Ltd. All rights reserved. +# https://nautechsystems.io +# +# Licensed under the GNU Lesser General Public License Version 3.0 (the "License"); +# You may not use this file except in compliance with the License. +# You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ------------------------------------------------------------------------------------------------- + +from datetime import datetime, timedelta +from decimal import Decimal +from typing import Dict, List + +import pandas as pd + +from nautilus_trader.common.actor import Actor +from nautilus_trader.common.component import TimeEvent +from nautilus_trader.common.config import ActorConfig +from nautilus_trader.core.correctness import PyCondition +from nautilus_trader.core.uuid import UUID4 +from nautilus_trader.examples.strategies.ema_cross import EMACross, EMACrossConfig +from nautilus_trader.examples.strategies.simple_binance_symbols_filter import ( + extract_symbol_info, + filter_with_onboard_date, + select_with_min_notional, + select_with_quoteAsset, +) +from nautilus_trader.examples.strategies.simple_cross_sectional_metrics import ( + generate_metrics, + get_binance_historical_bars, +) +from nautilus_trader.model.data import BarType +from nautilus_trader.model.identifiers import InstrumentId +from nautilus_trader.model.instruments import Instrument +from nautilus_trader.trading.controller import Controller +from nautilus_trader.trading.strategy import Strategy +from nautilus_trader.trading.trader import Trader + + + +class BinanceFutureInstrumentSelectorControllerConfig(ActorConfig, frozen=True): + interval_secs: int = 3600 + min_notional_threshold: float = 6 + quote_asset: str = "USDT" + # Filter the symbol df based on the onboard date + onboard_date_filter_type: str = "range" + onboard_date_reference_date: datetime = datetime(2023, 1, 1) + onboard_date_end_date: datetime = datetime(2024, 6, 1) + + +class BinanceFutureInstrumentSelectorController(Controller): + """ + This class is a controller for managing strategies dynamically based on top instruments by score. + It filters symbols based on various criteria, computes scores, and manages strategies accordingly. + """ + def __init__( + self, + trader: Trader, + config: BinanceFutureInstrumentSelectorControllerConfig | None = None, + ) -> None: + if config is None: + config = BinanceFutureInstrumentSelectorControllerConfig() + PyCondition.type(config, BinanceFutureInstrumentSelectorControllerConfig, "config") + super().__init__(config=config, trader=trader) + + self.interval_secs: int = config.interval_secs + self.min_notional_threshold: float = config.min_notional_threshold + self.quote_asset: str = config.quote_asset + self.onboard_date_filter_type: str = config.onboard_date_filter_type + self.onboard_date_reference_date: datetime = config.onboard_date_reference_date + self.onboard_date_end_date: datetime = config.onboard_date_end_date + self._trader: Trader = trader + self.filtered_instrument_id_values: List[str] = [] + self.filtered_symbols: pd.DataFrame = pd.DataFrame() + self.active_strategies: Dict[str, Strategy] = {} + + def on_start(self) -> None: + """Initialize and filter symbols on start.""" + # Get symbol info + symbol_df, error_message = extract_symbol_info() + if error_message: + self.log.error(f"Error extracting symbol info: {error_message}") + return + + # Filter the symbol df based on the quote asset + filtered_by_asset = select_with_quoteAsset(symbol_df, self.quote_asset) + + # Filter the symbol df based on the min notional threshold + filtered_by_min_notional = select_with_min_notional(filtered_by_asset, self.min_notional_threshold) + + # Filter the symbol df based on the onboard date + self.filtered_symbols = filter_with_onboard_date( + filtered_by_min_notional, + self.onboard_date_filter_type, + self.onboard_date_reference_date, + self.onboard_date_end_date, + ) + + self.filtered_instrument_id_values = [f"{item}-PERP.BINANCE" for item in self.filtered_symbols.symbol.values] + self.log.info(f"the length of filtered_instrument_id_values is {len(self.filtered_instrument_id_values)}") + # Set timer for periodic strategy management + self.clock.set_timer( + name="instrument_selector", + interval=timedelta(seconds=self.interval_secs), + callback=self.strategy_dynamic_management, + ) + + def strategy_dynamic_management(self, event: TimeEvent) -> None: + """ + Manage strategies dynamically based on top instruments by score. + + Args: + event (TimeEvent): The time event triggering this method. + """ + # Get the top 2 instruments by volume + top_2_instrument_id_values = self.get_top_2_instruments_by_score() + + # Remove strategies that are not in the top 2 and have no open positions + for instrument_id_value in list(self.active_strategies.keys()): + instrument_id = InstrumentId.from_str(instrument_id_value) + if instrument_id_value not in top_2_instrument_id_values and self.active_strategies[instrument_id_value].portfolio.is_flat(instrument_id): + self.log.info(f"stopping and removing strategy for {instrument_id_value}") + self.stop_strategy(self.active_strategies[instrument_id_value]) + self.remove_strategy(self.active_strategies[instrument_id_value]) + del self.active_strategies[instrument_id_value] + + # Add strategies for top 2 instruments not in active strategies + for instrument_id_value in top_2_instrument_id_values: + if instrument_id_value not in self.active_strategies: + instrument = self.cache.instrument(instrument_id=InstrumentId.from_str(instrument_id_value)) + if instrument: + self.log.info(f"creating strategy for {instrument_id_value}") + strategy = self.create_strategy_instance(instrument) + self.create_strategy(strategy, start=True) + self.active_strategies[instrument_id_value] = strategy + else: + self.log.warning(f"Instrument not found in cache: {instrument_id_value}") + + def get_top_2_instruments_by_score(self,) -> List[str]: + """ + Get the top 2 instruments by score. + + Returns: + List[str]: List of top 2 instrument IDs. + """ + end_time = datetime.now() + start_time = end_time - timedelta(days=7) + + # Fetch the 15-minute interval Kline data for the past 7 days for each symbol + df_dict = {} + for symbol in self.filtered_symbols.symbol.values: + try: + df_dict[symbol] = get_binance_historical_bars( + symbol=symbol, + start=start_time, + end=end_time, + interval='15m' + ) + except Exception as e: + self.log.error(f"Error fetching data for {symbol}: {e}") + + # Compute metrics for each symbol + final_scores_df = generate_metrics(df_dict) + + # Sort the DataFrame by average score and filter the top 2 symbols + top_2_symbols_df = final_scores_df.nlargest(2, 'average_score') + + # Get the top 2 symbols with the highest average score + top_2_symbols = top_2_symbols_df['symbol'].tolist() + self.log.info(f"the top 2 symbols are {top_2_symbols}") + return [f"{symbol}-PERP.BINANCE" for symbol in top_2_symbols] + + def create_strategy_instance(self, instrument: Instrument) -> Strategy: + """ + Create a strategy instance for the given instrument. + + Args: + instrument (Instrument): The instrument to create a strategy for. + + Returns: + Strategy: An instance of EMACross strategy. + """ + strategy_config = EMACrossConfig( + instrument_id=instrument.id, + external_order_claims=[instrument.id], + bar_type=BarType.from_str(f"{instrument.id.value}-1-MINUTE-LAST-EXTERNAL"), + fast_ema_period=10, + slow_ema_period=20, + trade_size=Decimal('100.0') * instrument.size_increment.as_decimal(), + order_id_tag=str(UUID4()), + oms_type="HEDGING", + ) + return EMACross(config=strategy_config) \ No newline at end of file