ftdt-quant-lab/strategies/nt/iceberg_nt.py

"""
Iceberg Detection NautilusTrader strategy.

Detects whale TWAP/iceberg accumulation by tracking volume spikes and
consecutive same-direction large orders.

Dual-mode:
  - Backtest: volume spike proxy from candles (volume > avg * multiplier
              for >= min_consecutive bars in same direction)
  - Live/Paper: real L2 orderbook wall detection (single level > avg * 3
                persisting for >= 3 updates)

Entry: consecutive same-direction spikes/walls → follow smart money
Exit:  spike count drops below 2 OR trend reverses OR 2% stop-loss
"""
from __future__ import annotations

import logging
from collections import deque
from typing import Any

import numpy as np

from nautilus_trader.model.data import Bar
from nautilus_trader.model.enums import OrderSide

from framework.base_strategy import BaseHlStrategy
from framework.config import StrategyConfig

logger = logging.getLogger(__name__)


class IcebergNT(BaseHlStrategy):
    """Iceberg/whale accumulation detection — follows smart money flow."""

    def __init__(self, config: StrategyConfig):
        super().__init__(config)

        self._vol_lookback = config.params.get("vol_lookback", 40)
        self._vol_spike_mult = config.params.get("vol_spike_mult", 2.5)
        self._min_consecutive = config.params.get("min_consecutive", 3)
        self._max_bars_held = config.params.get("max_bars_held", 8)
        self._stop_loss_pct = config.params.get("stop_loss_pct", 0.02)
        self._cooldown_bars = config.params.get("cooldown_bars", 4)

        # Volume tracking
        self._volumes: deque[float] = deque(maxlen=self._vol_lookback)
        self._spike_count: int = 0
        self._prev_spike_dir: str | None = None

        # State
        self._bars_since_trade = self._cooldown_bars
        self._bars_held: int = 0
        self._in_trade = False
        self._trade_direction: str | None = None
        self._entry_price: float = 0.0

    # ── Candle mode (backtest) ──────────────────────────────────

    def on_bar(self, bar: Bar):
        price = float(bar.close)
        volume = float(bar.volume) if hasattr(bar, 'volume') else 1.0
        self._prices.append(price)
        self._volumes.append(volume)

        self._bars_since_trade += 1

        # Exit check
        if self._in_trade:
            self._bars_held += 1
            if self._check_exit(price):
                return
            return

        if self._bars_since_trade < self._cooldown_bars:
            return

        signal = self._detect_iceberg(price, volume)
        if signal:
            self._last_signal = signal
            self.handle_signal(signal)

    def _detect_iceberg(self, price: float, volume: float) -> dict | None:
        if len(self._volumes) < self._vol_lookback:
            return None

        avg_vol = np.mean(self._volumes)
        if avg_vol <= 0:
            return None

        is_spike = volume > avg_vol * self._vol_spike_mult
        if not is_spike:
            self._spike_count = 0
            self._prev_spike_dir = None
            return None

        # Determine direction: buy if close > previous close (price going up)
        if len(self._prices) < 2:
            return None
        is_buy = self._prices[-1] > self._prices[-2]
        spike_dir = "buy" if is_buy else "sell"

        # Track consecutive same-direction spikes
        if spike_dir == self._prev_spike_dir:
            self._spike_count += 1
        else:
            self._spike_count = 1
            self._prev_spike_dir = spike_dir

        if self._spike_count >= self._min_consecutive:
            self._in_trade = True
            self._trade_direction = "long" if spike_dir == "buy" else "short"
            self._entry_price = price
            self._bars_since_trade = 0
            self._bars_held = 0
            self._spike_count = 0

            return {
                "signal": "BUY" if spike_dir == "buy" else "SELL",
                "strength": min(1.0, self._spike_count / self._min_consecutive),
                "vol_ratio": round(volume / avg_vol, 1),
                "spikes": self._spike_count,
                "reason": f"iceberg_{spike_dir}",
            }

        return None

    # ── L2 mode (live/paper) ────────────────────────────────────

    def compute_signal(self, price: float | None = None,
                       orderbook: dict | None = None) -> dict | None:
        """Entry point for paper trader / deploy orchestrator.

        If orderbook provided, use L2 wall detection.
        Otherwise fall back to candle proxy.
        """
        if orderbook is not None and price is not None:
            return self._detect_l2_walls(orderbook, price)

        if price is None:
            return None
        return self._detect_iceberg(price, 1.0)

    def _detect_l2_walls(self, orderbook: dict, price: float) -> dict | None:
        """Detect walls in real L2 orderbook."""
        bids = orderbook.get("bids", [])
        asks = orderbook.get("asks", [])

        # Find largest single level size
        all_sizes = [b[1] for b in bids] + [a[1] for a in asks]
        if not all_sizes:
            return None
        avg_size = np.mean(all_sizes)

        # Check for bid wall (single level > avg * 3)
        bid_wall = False
        ask_wall = False
        for px, sz in bids:
            if sz > avg_size * 3:
                bid_wall = True
                break
        for px, sz in asks:
            if sz > avg_size * 3:
                ask_wall = True
                break

        if bid_wall and not ask_wall:
            return {"signal": "BUY", "strength": 0.8, "reason": "l2_bid_wall"}
        elif ask_wall and not bid_wall:
            return {"signal": "SELL", "strength": 0.8, "reason": "l2_ask_wall"}
        return None

    # ── Exit logic ──────────────────────────────────────────────

    def _check_exit(self, current_price: float) -> bool:
        if not self._in_trade or self._entry_price <= 0:
            return False

        change_pct = (current_price - self._entry_price) / self._entry_price
        pnl_pct = change_pct if self._trade_direction == "long" else -change_pct

        exit_reason = None

        if pnl_pct <= -self._stop_loss_pct:
            exit_reason = "stop_loss"
        elif self._bars_held >= self._max_bars_held:
            exit_reason = "time_exit"
        elif self._spike_count < 2:
            exit_reason = "spikes_faded"

        if exit_reason is None:
            return False

        exit_side = "SELL" if self._trade_direction == "long" else "BUY"
        self._last_signal = {
            "signal": exit_side,
            "strength": abs(pnl_pct) / self._stop_loss_pct,
            "pnl_pct": round(pnl_pct * 100, 2),
            "bars_held": self._bars_held,
            "reason": exit_reason,
        }
        self._in_trade = False
        self._trade_direction = None
        self.handle_signal(self._last_signal)
        return True

    def handle_signal(self, signal: dict):
        side_str = signal.get("signal", "")
        if "BUY" in side_str:
            self._submit_order(OrderSide.BUY)
        elif "SELL" in side_str:
            self._submit_order(OrderSide.SELL)