205 lines
No EOL
6.9 KiB
Python
205 lines
No EOL
6.9 KiB
Python
'''
|
|
Bollinger Bands Volatility Breakout Strategy Overview:
|
|
|
|
Volatility Detection:
|
|
- Uses BTC Bollinger Bands on 1-minute timeframe to gauge market volatility
|
|
- "Tight bands" indicate low volatility periods (compression before expansion)
|
|
- Strategy activates only during low volatility conditions
|
|
|
|
Breakout Anticipation:
|
|
- Places BOTH buy and sell orders simultaneously when bands are tight
|
|
- Anticipates volatility expansion (breakout) in either direction
|
|
- Uses 11th level bid/ask for better execution probability
|
|
|
|
Dual Order Logic:
|
|
- BUY order: Positioned to catch upward breakout
|
|
- SELL order: Positioned to catch downward breakout
|
|
- One order gets filled, the other gets canceled when volatility returns
|
|
- Position size split in half to accommodate dual orders
|
|
|
|
Risk Management:
|
|
- Profit target: 5% gain
|
|
- Stop loss: 10% loss
|
|
- Automatic position closure when volatility returns (bands widen)
|
|
- Maximum 1 position with controlled exposure
|
|
|
|
Technical Setup:
|
|
- Reference: BTC Bollinger Bands (market volatility proxy)
|
|
- Trading Symbol: WIF (or configurable)
|
|
- Timeframe: 1 minute (500 periods for BB calculation)
|
|
- Position sizing: 50% of normal size per order (dual orders)
|
|
- Leverage: 3x (adjustable)
|
|
|
|
Strategy Logic:
|
|
- Low volatility (tight bands): Place dual orders expecting breakout
|
|
- High volatility (wide bands): Close all positions and orders
|
|
- Uses BTC as volatility gauge but trades altcoins for better moves
|
|
|
|
WARNING: Do not run without thorough backtesting and understanding!
|
|
'''
|
|
|
|
import Functions.funcs as func
|
|
import eth_account
|
|
import time
|
|
import schedule
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
# Configuration
|
|
symbol = 'WIF'
|
|
timeframe = '15m'
|
|
sma_window = 20
|
|
lookback_days = 1
|
|
size = 1
|
|
target = 5
|
|
max_loss = -10
|
|
leverage = 3
|
|
max_positions = 1
|
|
|
|
def get_account():
|
|
"""Initialize trading account from private key"""
|
|
secret = os.getenv('PRIVATE_KEY')
|
|
return eth_account.Account.from_key(secret)
|
|
|
|
def get_position_info(symbol, account):
|
|
"""Get current position and setup information"""
|
|
try:
|
|
positions, in_position, position_size, pos_symbol, entry_price, pnl_percent, is_long, num_positions = func.get_position_andmaxpos(
|
|
symbol, account, max_positions
|
|
)
|
|
|
|
print(f'Position Status - In position: {in_position} | Size: {position_size} | PnL: {pnl_percent}%')
|
|
|
|
# Setup leverage and get adjusted position size
|
|
leverage_set, pos_size = func.adjust_leverage_size_signal(symbol, leverage, account)
|
|
|
|
# Split position size for dual orders
|
|
dual_order_size = pos_size / 2
|
|
|
|
return in_position, dual_order_size, positions
|
|
|
|
except Exception as e:
|
|
print(f'Error getting position info: {e}')
|
|
return False, size / 2, []
|
|
|
|
def check_bollinger_volatility():
|
|
"""Check if Bollinger Bands are tight (low volatility condition)"""
|
|
try:
|
|
# Get BTC data for volatility analysis
|
|
snapshot_data = func.get_ohlcv2('BTC', '1m', 500)
|
|
df = func.process_data_to_df(snapshot_data)
|
|
|
|
# Calculate Bollinger Bands
|
|
bb_data = func.calculate_bollinger_bands(df)
|
|
bands_tight = bb_data[1] # Boolean indicating if bands are tight
|
|
|
|
print(f'BTC Bollinger Bands Analysis - Tight (Low Volatility): {bands_tight}')
|
|
|
|
return bands_tight
|
|
|
|
except Exception as e:
|
|
print(f'Error checking Bollinger volatility: {e}')
|
|
return False
|
|
|
|
def get_market_levels(symbol):
|
|
"""Get current market levels for order placement"""
|
|
try:
|
|
ask, bid, l2_data = func.ask_bid(symbol)
|
|
|
|
# Get 11th level for better execution
|
|
bid_11 = float(l2_data[0][10]['px'])
|
|
ask_11 = float(l2_data[1][10]['px'])
|
|
|
|
print(f'Market Levels - Bid: {bid:.4f} | Ask: {ask:.4f}')
|
|
print(f'11th Level - Bid: {bid_11:.4f} | Ask: {ask_11:.4f}')
|
|
|
|
return bid_11, ask_11
|
|
|
|
except Exception as e:
|
|
print(f'Error getting market levels: {e}')
|
|
return 0, 0
|
|
|
|
def place_dual_orders(symbol, bid_11, ask_11, position_size, account):
|
|
"""Place both buy and sell orders for breakout capture"""
|
|
try:
|
|
func.cancel_all_orders(account)
|
|
print('Canceled all existing orders')
|
|
|
|
# Place buy order (upward breakout)
|
|
func.limit_order(symbol, True, position_size, bid_11, False, account)
|
|
print(f'BUY order placed: {position_size} at {bid_11:.4f}')
|
|
|
|
# Place sell order (downward breakout)
|
|
func.limit_order(symbol, False, position_size, ask_11, False, account)
|
|
print(f'SELL order placed: {position_size} at {ask_11:.4f}')
|
|
|
|
print('Dual orders active - Ready for volatility breakout')
|
|
|
|
except Exception as e:
|
|
print(f'Error placing dual orders: {e}')
|
|
|
|
def manage_volatility_exit(account):
|
|
"""Close all positions and orders when volatility returns"""
|
|
try:
|
|
print('High volatility detected - Closing all positions and orders')
|
|
func.cancel_all_orders(account)
|
|
func.close_all_positions(account)
|
|
print('All positions and orders closed')
|
|
|
|
except Exception as e:
|
|
print(f'Error managing volatility exit: {e}')
|
|
|
|
def bot():
|
|
try:
|
|
print(f'\n---- Bollinger Bands Volatility Strategy for {symbol} ----')
|
|
|
|
# Initialize account
|
|
account = get_account()
|
|
|
|
# Get position information
|
|
in_position, dual_order_size, positions = get_position_info(symbol, account)
|
|
|
|
# Check existing positions first
|
|
if in_position:
|
|
func.cancel_all_orders(account)
|
|
print('Managing existing position - checking PnL')
|
|
func.pnl_close(symbol, target, max_loss, account)
|
|
return
|
|
|
|
# Check Bollinger Bands volatility condition
|
|
bands_tight = check_bollinger_volatility()
|
|
|
|
if not in_position and bands_tight:
|
|
print('LOW VOLATILITY detected - Placing dual breakout orders')
|
|
|
|
# Get market levels
|
|
bid_11, ask_11 = get_market_levels(symbol)
|
|
|
|
if bid_11 > 0 and ask_11 > 0:
|
|
place_dual_orders(symbol, bid_11, ask_11, dual_order_size, account)
|
|
else:
|
|
print('Invalid market data - skipping dual orders')
|
|
|
|
elif not bands_tight:
|
|
print('HIGH VOLATILITY detected - Managing exits')
|
|
manage_volatility_exit(account)
|
|
|
|
else:
|
|
print(f'Position status: {in_position} | Volatility condition: {"Low" if bands_tight else "High"}')
|
|
|
|
print('=' * 60)
|
|
|
|
except Exception as e:
|
|
print(f'Bot error: {e}')
|
|
|
|
schedule.every(30).seconds.do(bot)
|
|
|
|
while True:
|
|
try:
|
|
schedule.run_pending()
|
|
time.sleep(10)
|
|
except Exception as e:
|
|
print(f'Schedule error: {e} - Sleeping 30 seconds')
|
|
time.sleep(30) |