import asyncio
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Tuple
import websocket
import json
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class LiquidationMonitor:
    """Monitor and detect liquidation cascades across exchanges"""
def __init__(self):
        self.liquidation_events = []
        self.open_positions = {}
        self.liquidation_threshold = 10_000_000  # $10M in liquidations triggers alert
        self.liquidation_density = {}  # Track liquidations per price level
async def subscribe_to_liquidations(self, exchange: str, symbol: str):
        """Subscribe to liquidation stream from exchange"""
if exchange.lower() == 'binance':
            await self._subscribe_binance_liquidations(symbol)
        elif exchange.lower() == 'bybit':
            await self._subscribe_bybit_liquidations(symbol)
        elif exchange.lower() == 'ftx':
            await self._subscribe_ftx_liquidations(symbol)
async def _subscribe_binance_liquidations(self, symbol: str):
        """Subscribe to Binance liquidation stream"""
def on_message(ws, msg):
            try:
                data = json.loads(msg)
                event = data.get('e')
if event == 'forceOrder':
                    liquidation = {
                        'exchange': 'binance',
                        'symbol': symbol,
                        'timestamp': datetime.fromtimestamp(data['E'] / 1000),
                        'price': float(data['o']['ap']),  # average price
                        'quantity': float(data['o']['q']),
                        'side': data['o']['S'],  # BUY or SELL
                        'value': float(data['o']['ap']) * float(data['o']['q'])
                    }
self.liquidation_events.append(liquidation)
                    self._process_liquidation(liquidation)
except Exception as e:
                logger.error(f"Error processing Binance liquidation: {e}")
def on_error(ws, error):
            logger.error(f"WebSocket error: {error}")
def on_close(ws, close_status_code, close_msg):
            logger.info("WebSocket closed")
ws = websocket.WebSocketApp(
            f"wss://fstream.binance.com/ws/{symbol.lower()}@forceOrder",
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )
ws.run_forever()
def _process_liquidation(self, liquidation: Dict):
        """Process incoming liquidation event"""
# Track density at price level
        price_level = round(liquidation['price'] / 100) * 100
if price_level not in self.liquidation_density:
            self.liquidation_density[price_level] = {
                'total': 0,
                'count': 0,
                'last_update': datetime.now()
            }
self.liquidation_density[price_level]['total'] += liquidation['value']
        self.liquidation_density[price_level]['count'] += 1
logger.info(f"Liquidation: {liquidation['symbol']} {liquidation['side']} "
                   f"{liquidation['quantity']:.4f} @ ${liquidation['price']:.2f} "
                   f"(${liquidation['value']:,.0f})")
# Check for cascade threshold
        recent_sum = sum(l['value'] for l in self.liquidation_events[-50:]
                        if (datetime.now() - l['timestamp']).total_seconds() < 300)
if recent_sum > self.liquidation_threshold:
            logger.warning(f"LIQUIDATION CASCADE ALERT: ${recent_sum:,.0f} in liquidations in 5 min!")
            self.on_cascade_detected()
def on_cascade_detected(self):
        """Handle cascade detection - alert trading systems"""
        pass
def identify_liquidation_levels(self, symbol: str) -> List[Dict]:
        """Identify price levels with high liquidation concentration"""
# Filter recent liquidations (last hour)
        recent_liq = [l for l in self.liquidation_events
                      if (datetime.now() - l['timestamp']).total_seconds() < 3600
                      and l['symbol'] == symbol]
if not recent_liq:
            return []
# Group by price level
        price_levels = {}
        for liq in recent_liq:
            level = round(liq['price'] / 50) * 50  # Round to nearest $50
if level not in price_levels:
                price_levels[level] = {
                    'price': level,
                    'longs': 0,
                    'shorts': 0,
                    'long_value': 0,
                    'short_value': 0
                }
if liq['side'] == 'BUY':  # Long liquidations
                price_levels[level]['shorts'] += 1
                price_levels[level]['short_value'] += liq['value']
            else:  # Short liquidations
                price_levels[level]['longs'] += 1
                price_levels[level]['long_value'] += liq['value']
# Sort by concentration
        levels = sorted(price_levels.values(),
                       key=lambda x: max(x['long_value'], x['short_value']),
                       reverse=True)
return levels[:10]  # Top 10 levels
def calculate_liquidation_risk(self, current_price: float,
                                  direction: str) -> Dict:
        """Calculate liquidation risk at current price levels"""
# Find liquidation levels ahead
        levels = self.identify_liquidation_levels('BTC/USDT')
if direction == 'UP':
            levels_ahead = [l for l in levels if l['price'] > current_price]
            resistance_density = sum(l['short_value'] for l in levels_ahead[:5])
return {
                'direction': 'UP',
                'levels_ahead': 5,
                'expected_resistance': resistance_density,
                'risk_level': 'HIGH' if resistance_density > 500_000_000 else 'MEDIUM' if resistance_density > 100_000_000 else 'LOW'
            }
else:  # DOWN
            levels_ahead = [l for l in levels if l['price'] < current_price]
            support_density = sum(l['long_value'] for l in levels_ahead[:5])
return {
                'direction': 'DOWN',
                'levels_ahead': 5,
                'expected_support': support_density,
                'risk_level': 'HIGH' if support_density > 500_000_000 else 'MEDIUM' if support_density > 100_000_000 else 'LOW'
            }

class CascadePredictionModel:
    """Machine learning model to predict liquidation cascades"""
def __init__(self, lookback_period: int = 20):
        self.lookback_period = lookback_period
        self.cascade_history = []
        self.features = []
def extract_cascade_features(self, candles: pd.DataFrame,
                                liquidations: List[Dict]) -> np.ndarray:
        """Extract features predictive of cascades"""
recent_liq = [l for l in liquidations[-100:]]
if len(candles) < self.lookback_period:
            return None
close = candles['close'].values
        volume = candles['volume'].values
features = []
# Volatility spike
        returns = np.diff(close) / close[:-1]
        volatility = np.std(returns[-20:])
        features.append(volatility)
# Volume surge
        avg_volume = np.mean(volume[-20:])
        volume_spike = volume[-1] / avg_volume
        features.append(volume_spike)
# Liquidation density increase
        recent_5m_liq = len([l for l in recent_liq
                            if (datetime.now() - l['timestamp']).total_seconds() < 300])
        features.append(recent_5m_liq)
# Price momentum
        momentum = (close[-1] - close[-20]) / close[-20]
        features.append(momentum)
# Open interest trend
        oi_volume = sum(l['value'] for l in recent_liq[-20:])
        features.append(oi_volume / (avg_volume * 100))
# Liquidation side imbalance
        short_liq = len([l for l in recent_liq if l['side'] == 'BUY'])
        long_liq = len([l for l in recent_liq if l['side'] == 'SELL'])
        imbalance = (short_liq - long_liq) / (short_liq + long_liq + 1)
        features.append(imbalance)
return np.array(features)
def predict_cascade(self, candles: pd.DataFrame,
                       liquidations: List[Dict]) -> Dict:
        """Predict probability of liquidation cascade"""
features = self.extract_cascade_features(candles, liquidations)
if features is None:
            return {'probability': 0, 'confidence': 'low'}
# Simple rule-based prediction (can be upgraded to ML model)
        volatility, volume_spike, liq_count, momentum, oi_ratio, imbalance = features
# Calculate cascade score
        cascade_score = 0
if volatility > 0.03:  # High volatility
            cascade_score += 2
        if volume_spike > 1.5:  # Volume surge
            cascade_score += 2
        if liq_count > 10:  # Recent liquidations
            cascade_score += 1.5
        if abs(momentum) > 0.02:  # Strong momentum
            cascade_score += 1
        if oi_ratio > 2:  # High liquidation value
            cascade_score += 1.5
        if abs(imbalance) > 0.6:  # Imbalanced liquidations
            cascade_score += 1
probability = min(cascade_score / 10, 1.0)
return {
            'cascade_probability': probability,
            'cascade_imminent': probability > 0.6,
            'direction': 'DOWN' if imbalance > 0.3 else 'UP' if imbalance < -0.3 else 'UNCLEAR',
            'score_breakdown': {
                'volatility': volatility,
                'volume_spike': volume_spike,
                'recent_liq': liq_count,
                'momentum': momentum,
                'imbalance': imbalance
            }
        }

class CascadeTrader:
    """Trade liquidation cascades when they occur"""
def __init__(self, exchange, symbol: str = 'BTC/USDT'):
        self.exchange = exchange
        self.symbol = symbol
        self.cascade_positions = []
        self.entry_conditions = {
            'min_price_move': 0.005,  # 0.5% move triggers
            'min_volume_surge': 1.5,
            'min_cascade_prob': 0.6
        }
async def on_cascade_detected(self, cascade_info: Dict):
        """Execute trade when cascade is detected"""
logger.info(f"Cascade detected: Direction={cascade_info['direction']}, "
                   f"Probability={cascade_info['cascade_probability']:.2%}")
# Get current price
        ticker = await self.exchange.fetch_ticker(self.symbol)
        current_price = ticker['last']
# Determine position
        if cascade_info['direction'] == 'DOWN':
            # Short the cascade (sell into it)
            position = await self.enter_short(current_price, cascade_info)
elif cascade_info['direction'] == 'UP':
            # Buy the cascade (bounce play)
            position = await self.enter_long(current_price, cascade_info)
return position
async def enter_short(self, entry_price: float, cascade_info: Dict) -> Dict:
        """Enter short position during downward cascade"""
try:
            size = 0.1  # Trade small size
# Place market sell order
            order = await self.exchange.create_market_sell_order(self.symbol, size)
position = {
                'timestamp': datetime.now(),
                'direction': 'SHORT',
                'entry_price': order['average'],
                'size': size,
                'stop_loss': order['average'] * 1.02,  # 2% above entry
                'take_profit_1': order['average'] * 0.98,  # Take 50% at -2%
                'take_profit_2': order['average'] * 0.95,  # Take rest at -5%
                'cascade_info': cascade_info
            }
self.cascade_positions.append(position)
            logger.info(f"Entered SHORT at ${entry_price:.2f}, Size: {size}")
return position
except Exception as e:
            logger.error(f"Failed to enter short: {e}")
            return None
async def enter_long(self, entry_price: float, cascade_info: Dict) -> Dict:
        """Enter long position for bounce play"""
try:
            size = 0.1
order = await self.exchange.create_market_buy_order(self.symbol, size)
position = {
                'timestamp': datetime.now(),
                'direction': 'LONG',
                'entry_price': order['average'],
                'size': size,
                'stop_loss': order['average'] * 0.98,  # 2% below entry
                'take_profit_1': order['average'] * 1.02,  # Take 50% at +2%
                'take_profit_2': order['average'] * 1.05,  # Take rest at +5%
                'cascade_info': cascade_info
            }
self.cascade_positions.append(position)
            logger.info(f"Entered LONG at ${entry_price:.2f}, Size: {size}")
return position
except Exception as e:
            logger.error(f"Failed to enter long: {e}")
            return None
async def manage_cascade_position(self, position: Dict, current_price: float):
        """Manage cascade position with exit rules"""
if position['direction'] == 'SHORT':
            # Check stop loss
            if current_price > position['stop_loss']:
                await self.exit_position(position, current_price, 'STOP LOSS')
# Check take profit
            elif current_price < position['take_profit_1']:
                # Sell half
                await self.partial_exit(position, current_price, 0.5, 'TAKE PROFIT 1')
elif current_price < position['take_profit_2']:
                # Sell rest
                await self.exit_position(position, current_price, 'TAKE PROFIT 2')
elif position['direction'] == 'LONG':
            # Check stop loss
            if current_price < position['stop_loss']:
                await self.exit_position(position, current_price, 'STOP LOSS')
# Check take profit
            elif current_price > position['take_profit_1']:
                await self.partial_exit(position, current_price, 0.5, 'TAKE PROFIT 1')
elif current_price > position['take_profit_2']:
                await self.exit_position(position, current_price, 'TAKE PROFIT 2')
async def exit_position(self, position: Dict, exit_price: float, reason: str):
        """Exit cascade position"""
try:
            if position['direction'] == 'SHORT':
                order = await self.exchange.create_market_buy_order(self.symbol, position['size'])
            else:
                order = await self.exchange.create_market_sell_order(self.symbol, position['size'])
pnl = (exit_price - position['entry_price']) * position['size']
            if position['direction'] == 'SHORT':
                pnl = -pnl
logger.info(f"Exited {position['direction']} position: {reason}, P&L: ${pnl:.2f}")
self.cascade_positions.remove(position)
except Exception as e:
            logger.error(f"Failed to exit position: {e}")
async def partial_exit(self, position: Dict, exit_price: float,
                          fraction: float, reason: str):
        """Partially exit position"""
exit_size = position['size'] * fraction
        position['size'] -= exit_size
await self.exit_position(
            {**position, 'size': exit_size},
            exit_price,
            reason
        )

class CascadeRiskManager:
    """Special risk management for cascade trading"""
def __init__(self, max_cascade_exposure: float = 5000):
        self.max_cascade_exposure = max_cascade_exposure
        self.current_exposure = 0
        self.cascade_losses = []
def is_cascade_trade_allowed(self, cascade_info: Dict) -> bool:
        """Check if cascade trade is within risk limits"""
# Don't trade if cascade probability is too low
        if cascade_info['cascade_probability'] < 0.6:
            return False
# Check current exposure
        if self.current_exposure >= self.max_cascade_exposure:
            logger.warning("Max cascade exposure reached, pausing trades")
            return False
return True
def calculate_position_size_for_cascade(self, account_balance: float,
                                           cascade_severity: float) -> float:
        """Size position based on cascade severity"""
# More severe = smaller position
        severity_multiplier = 1 - min(cascade_severity / 10, 0.9)
        max_risk = account_balance * 0.02  # 2% risk per trade
return max_risk * severity_multiplier
def set_aggressive_stops_on_cascade(self, position: Dict,
                                       cascade_severity: float):
        """Use tighter stops during extreme cascades"""
if cascade_severity > 8:
            # Extreme cascade - very tight stops
            position['stop_loss'] = position['entry_price'] * (1.01 if position['direction'] == 'SHORT' else 0.99)
        elif cascade_severity > 5:
            # Moderate cascade
            position['stop_loss'] = position['entry_price'] * (1.015 if position['direction'] == 'SHORT' else 0.985)
return position