Liquidation Heatmap Analysis

# Long position liquidation
long_liquidation_price = entry_price * (1 - 1/leverage + maintenance_margin)

# Short position liquidation
short_liquidation_price = entry_price * (1 + 1/leverage - maintenance_margin)

# Example: BTC long at $65,000, 10x leverage, 0.5% maintenance margin
# Liquidation: $65,000 * (1 - 1/10 + 0.005) = $58,825
# A 9.5% move against the position triggers liquidation

Price Level    Long Liquidations    Short Liquidations    Interpretation
$70,000        ░░░░░░░░░░          ████████████████      Heavy short liquidation zone
$68,000        ░░░░                ██████████            Moderate short liquidation
$66,000        ███████             ███████               Balanced (current price area)
$64,000        ██████████          ░░░░                  Moderate long liquidation
$62,000        ████████████████    ░░░░░░░░░░            Heavy long liquidation zone

def identify_liquidation_clusters(open_interest_by_price, leverage_distribution):
    """
    Estimate where liquidation clusters exist based on
    open interest and leverage distribution.
    """
    clusters = []

    for price_level in price_range:
        # Long liquidations: positions opened above this level with high leverage
        long_liq_volume = estimate_long_liq_at_price(
            open_interest_by_price, leverage_distribution, price_level
        )

        # Short liquidations: positions opened below this level with high leverage
        short_liq_volume = estimate_short_liq_at_price(
            open_interest_by_price, leverage_distribution, price_level
        )

        total = long_liq_volume + short_liq_volume

        if total > significance_threshold:
            clusters.append({
                "price": price_level,
                "long_liq": long_liq_volume,
                "short_liq": short_liq_volume,
                "type": "long" if long_liq_volume > short_liq_volume else "short",
                "magnitude": total,
            })

    return sorted(clusters, key=lambda x: x["magnitude"], reverse=True)

def liquidation_magnet_signal(current_price, clusters):
    """
    Price is likely to move toward the nearest large liquidation cluster.
    """
    # Find nearest cluster above and below
    above = [c for c in clusters if c["price"] > current_price]
    below = [c for c in clusters if c["price"] < current_price]

    nearest_above = min(above, key=lambda c: c["price"] - current_price) if above else None
    nearest_below = min(below, key=lambda c: current_price - c["price"]) if below else None

    if nearest_above and nearest_below:
        above_magnitude = nearest_above["magnitude"]
        below_magnitude = nearest_below["magnitude"]

        if above_magnitude > below_magnitude * 2:
            return "upward_magnet"      # Larger cluster above → price likely moves up
        elif below_magnitude > above_magnitude * 2:
            return "downward_magnet"    # Larger cluster below → price likely moves down
        else:
            return "balanced"           # Both sides have similar clusters

def cascade_risk(current_price, clusters, direction="down"):
    """
    Assess risk of liquidation cascade — multiple clusters stacked close together.
    """
    if direction == "down":
        relevant = sorted([c for c in clusters if c["price"] < current_price and c["type"] == "long"],
                         key=lambda c: c["price"], reverse=True)
    else:
        relevant = sorted([c for c in clusters if c["price"] > current_price and c["type"] == "short"],
                         key=lambda c: c["price"])

    if len(relevant) < 2:
        return "low_cascade_risk"

    # Check if clusters are stacked within 5% of each other
    gaps = []
    for i in range(len(relevant) - 1):
        gap = abs(relevant[i]["price"] - relevant[i+1]["price"]) / current_price * 100
        gaps.append(gap)

    if min(gaps) < 2:
        return "high_cascade_risk"      # Clusters stacked tightly → cascade likely
    elif min(gaps) < 5:
        return "moderate_cascade_risk"
    else:
        return "low_cascade_risk"

def post_liquidation_sr(price_history, liquidation_events):
    """
    After a large liquidation event, that price level often becomes S/R.
    """
    sr_levels = []
    for event in liquidation_events:
        if event.total_liquidated > 100_000_000:  # >$100M liquidated
            sr_levels.append({
                "price": event.price_level,
                "type": "support" if event.liquidation_type == "long" else "resistance",
                "strength": event.total_liquidated,
                "date": event.date,
            })
    return sr_levels