''' Correlation Trading Strategy Overview: Lead Asset Analysis: - Monitors ETH price movements for breakouts - Uses True Range and support/resistance levels as breakout signals - Detects when ETH moves outside recent trading range Correlation Logic: - When ETH breaks above resistance/ATR: Look for LONG opportunities in lagging altcoins - When ETH breaks below support/ATR: Look for SHORT opportunities in lagging altcoins - Identifies the most "lagging" altcoin (least price movement) to trade Entry Strategy: - Calculates percentage change from last candle for each altcoin - Trades the altcoin with minimal movement (highest correlation lag) - Assumes lagging altcoins will catch up to ETH's direction Risk Management: - Stop loss: 0.2% against position - Take profit: 0.25% in favor of position - Quick execution to capture correlation momentum Altcoin Universe: - ADAUSDT, DOTUSDT, MANAUSDT, XRPUSDT, UNIUSDT, SOLUSDT - All major altcoins that typically correlate with ETH movements Technical Setup: - Timeframe: 15 minutes (900 seconds) - Lookback period: 20 candles - Uses ATR and support/resistance for breakout confirmation WARNING: Do not run without thorough backtesting and understanding! ''' import ccxt, time, schedule from Functions.functions import * import os from dotenv import load_dotenv load_dotenv() api_key = os.getenv('BINANCE_API_KEY') api_secret = os.getenv('BINANCE_SECRET_KEY') exchange = ccxt.binance({ 'apiKey': api_key, 'secret': api_secret, 'enableRateLimit': True, 'options': { 'defaultType': 'future', } }) # Configuration timeframe = '15m' data_range = 20 sl_percent = 0.2 tp_percent = 0.25 size = 1 params = {'timeInForce': 'GTC'} # Lead symbol to monitor lead_symbol = 'ETHUSDT' # Altcoins to trade based on correlation alt_coins = ['ADAUSDT', 'DOTUSDT', 'MANAUSDT', 'XRPUSDT', 'UNIUSDT', 'SOLUSDT'] def get_eth_signal(): """Get ETH breakout signal and current price""" try: # Get ETH candle data candles = get_candle_df(exchange, lead_symbol, timeframe, data_range) # Calculate technical indicators calc_atr(candles) support, resistance = calc_sup_res(candles, data_range) # Get current price ticker = exchange.fetch_ticker(lead_symbol) current_price = ticker['bid'] last_close = candles['close'].iloc[-1] atr = candles['ATR'].iloc[-1] print(f'ETH - Price: {current_price:.2f} | Last Close: {last_close:.2f} | ATR: {atr:.2f}') print(f'Support: {support:.2f} | Resistance: {resistance:.2f}') # Check for breakout signals if current_price > last_close + atr or current_price > resistance: return 'BULLISH', current_price elif current_price < last_close - atr or current_price < support: return 'BEARISH', current_price else: return 'NEUTRAL', current_price except Exception as e: print(f'Error getting ETH signal: {e}') return 'NEUTRAL', 0 def find_most_lagging_altcoin(eth_price): """Find the altcoin with the least movement (most lagging)""" try: coin_data = {} for coin in alt_coins: try: # Get current price and recent candle data ticker = exchange.fetch_ticker(coin) current_price = ticker['bid'] candles = get_candle_df(exchange, coin, timeframe, 5) # Last 5 candles recent_close = candles['close'].iloc[-1] # Calculate percentage change from recent close pct_change = abs((current_price - recent_close) / recent_close) * 100 coin_data[coin] = pct_change print(f'{coin}: Current: {current_price:.4f} | Recent Close: {recent_close:.4f} | Change: {pct_change:.2f}%') except Exception as e: print(f'Error getting data for {coin}: {e}') continue if coin_data: # Find coin with minimum movement (most lagging) most_lagging = min(coin_data, key=coin_data.get) lag_amount = coin_data[most_lagging] print(f'Most lagging coin: {most_lagging} (Change: {lag_amount:.2f}%)') return most_lagging else: return None except Exception as e: print(f'Error finding lagging altcoin: {e}') return None def place_correlation_trade(signal, target_coin): """Place trade on the most lagging altcoin""" try: if not target_coin: print('No target coin identified') return # Check if already in position for this coin _, in_position, _, _ = get_position(exchange, target_coin) if in_position: print(f'Already in position for {target_coin}') return # Get current price for target coin ticker = exchange.fetch_ticker(target_coin) current_price = ticker['bid'] if signal == 'BULLISH' else ticker['ask'] # Calculate stop loss and take profit if signal == 'BULLISH': stop_loss = current_price * (1 - (sl_percent / 100)) take_profit = current_price * (1 + (tp_percent / 100)) print(f'Placing LONG on {target_coin} at {current_price:.4f}') print(f'Stop Loss: {stop_loss:.4f} | Take Profit: {take_profit:.4f}') exchange.create_limit_buy_order(target_coin, size, current_price, params) elif signal == 'BEARISH': stop_loss = current_price * (1 + (sl_percent / 100)) take_profit = current_price * (1 - (tp_percent / 100)) print(f'Placing SHORT on {target_coin} at {current_price:.4f}') print(f'Stop Loss: {stop_loss:.4f} | Take Profit: {take_profit:.4f}') exchange.create_limit_sell_order(target_coin, size, current_price, params) # Note: In a full implementation, you'd want to place actual stop loss and take profit orders # This would require additional order management logic except Exception as e: print(f'Error placing trade: {e}') def manage_existing_positions(): """Check and manage existing positions""" for coin in alt_coins: try: position_info, in_position, _ = get_position(exchange, coin) if in_position: # Here you would implement PnL checking and position management # For now, just log that we have a position print(f'Managing position in {coin}') # You could call a PnL management function here # hit_target() or similar logic from other strategies except Exception as e: print(f'Error managing position for {coin}: {e}') def bot(): try: print('\n---- CORRELATION STRATEGY ACTIVE ----') # Step 1: Manage existing positions manage_existing_positions() # Step 2: Check ETH for breakout signal eth_signal, eth_price = get_eth_signal() if eth_signal == 'NEUTRAL': print('ETH: No breakout signal detected') return print(f'ETH BREAKOUT: {eth_signal} signal at {eth_price:.2f}') # Step 3: Find most lagging altcoin target_coin = find_most_lagging_altcoin(eth_price) if target_coin: # Step 4: Place correlation trade place_correlation_trade(eth_signal, target_coin) else: print('No suitable target coin found') print('=' * 50) except Exception as e: print(f'Bot error: {e}') schedule.every(20).seconds.do(bot) while True: try: schedule.run_pending() except Exception as e: print(f'Schedule error: {e} - Sleeping 30 seconds') time.sleep(30)