233 lines
No EOL
7.9 KiB
Python
233 lines
No EOL
7.9 KiB
Python
'''
|
|
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) |