1
Fork 0
crypto_bot_training/Session_05/bots/Consollidation_Pop.py
2025-06-26 15:46:49 +02:00

137 lines
No EOL
4.9 KiB
Python

'''
Consolidation Pop Strategy Overview:
Consolidation Detection:
- Identifies sideways price movement using True Range analysis
- Consolidation confirmed when TR deviation < 0.7% of current price
- Uses recent 20-candle period to establish consolidation range
Range Trading Logic:
- LONG: Buy in lower 1/3 of consolidation range (near support)
- SHORT: Sell in upper 1/3 of consolidation range (near resistance)
- Anticipates breakouts in either direction from established ranges
Entry Strategy:
- Waits for price to enter specific zones within consolidation
- Lower 1/3: Expects bounce from support with breakout potential
- Upper 1/3: Expects rejection from resistance with breakdown potential
- Contrarian approach: Buy weakness, sell strength within range
Risk Management:
- Take profit: 0.3% gain from entry
- Stop loss: 0.25% loss from entry
- Quick scalping approach for range-bound markets
Technical Setup:
- Timeframe: 5 minutes (adjustable for 1m, 3m, 15m, 1h)
- Lookback: 20 candles for range establishment
- Consolidation threshold: 0.7% TR deviation from price
- Targets small, quick moves within established ranges
Market Conditions:
- Best suited for sideways, range-bound markets
- Avoids trending markets where consolidation patterns break down
- Designed for short-term momentum bursts within ranges
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 = '5m'
limit = 20
symbol = 'ETHUSDT'
size = 1
tp_percent = 0.3
sl_percent = 0.25
consolidation_percent = 0.7
params = {'timeInForce': 'GTC'}
def bot():
try:
# Check current position status
position_info, in_position, is_long = get_position(exchange, symbol)
if in_position:
print('Already in position - monitoring...')
return
# Get market data
candles = get_candle_df(exchange, symbol, timeframe, limit)
tr = calc_tr(candles)
current_price = exchange.fetch_ticker(symbol)['bid']
# Calculate TR deviation from current price
tr_deviation = (tr / candles['close'].iloc[-1]) * 100
print(f'Price: {current_price:.2f} | TR: {tr:.2f} | TR Deviation: {tr_deviation:.2f}%')
# Only trade during consolidation periods
if tr_deviation < consolidation_percent:
print('CONSOLIDATION detected - Analyzing range...')
# Get consolidation range extremes
low, high = get_extreme_of_consolidation(candles, consolidation_percent)
range_size = high - low
lower_third = low + (range_size / 3)
upper_third = high - (range_size / 3)
print(f'Range: {low:.2f} - {high:.2f} | Size: {range_size:.2f}')
print(f'Lower 1/3: {lower_third:.2f} | Upper 1/3: {upper_third:.2f}')
# LONG setup: Buy in lower 1/3 of consolidation (near support)
if current_price <= lower_third:
stop_loss = current_price * (1 - (sl_percent / 100))
take_profit = current_price * (1 + (tp_percent / 100))
print(f'LONG setup - Price in lower 1/3 of range')
print(f'Entry: {current_price:.2f} | SL: {stop_loss:.2f} | TP: {take_profit:.2f}')
exchange.create_limit_buy_order(symbol, size, current_price, params)
print('BUY order placed - Expecting consolidation pop upward')
# SHORT setup: Sell in upper 1/3 of consolidation (near resistance)
elif current_price >= upper_third:
stop_loss = current_price * (1 + (sl_percent / 100))
take_profit = current_price * (1 - (tp_percent / 100))
print(f'SHORT setup - Price in upper 1/3 of range')
print(f'Entry: {current_price:.2f} | SL: {stop_loss:.2f} | TP: {take_profit:.2f}')
exchange.create_limit_sell_order(symbol, size, current_price, params)
print('SELL order placed - Expecting consolidation pop downward')
else:
print(f'Price in middle zone ({current_price:.2f}) - No trade')
else:
print(f'TR deviation too high ({tr_deviation:.2f}%) - Not consolidating')
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)