//! BinanceProtocol — WsProtocol implementation for Binance.
//!
//! Declarative shim: supplies endpoint URLs, ping frame (None = native WS ping),
//! subscribe/unsubscribe frames, topic extraction, and topic registry to
//! UniversalWsTransport.
//!
//! ## Combined-stream format
//! All subscriptions use the `/stream` endpoint (combined-stream mode).
//! Frames arrive as `{"stream":"btcusdt@trade","data":{...}}`.
//! The `stream` field IS the topic key.
//!
//! ## Silent-stream fix (spec §3.3)
//! Old code had `_ => Ok(None)` catch-all in `parse_event_by_type`.
//! The framework now emits `tracing::warn!` for every unmatched topic,
//! making silent drops visible.  All known event types are covered here.

use std::sync::OnceLock;
use std::time::Duration;

use serde_json::{json, Value};
use url::Url;

use crate::core::rt::WsFrame;
use crate::core::traits::Credentials;
use crate::core::types::{
    AccountType, OrderBookLevel, StreamEvent, WebSocketError, WebSocketResult,
    OrderbookDelta as OrderbookDeltaData,
};
use crate::core::websocket::{
    KlineInterval, StreamKind, StreamSpec, TopicKey, TopicRegistry, WsProtocol,
};

use super::endpoints::BinanceUrls;
use super::parser::BinanceParser;

// ─────────────────────────────────────────────────────────────────────────────
// Registry caches
// ─────────────────────────────────────────────────────────────────────────────

static SPOT_REGISTRY: OnceLock<TopicRegistry> = OnceLock::new();
static FUTURES_REGISTRY: OnceLock<TopicRegistry> = OnceLock::new();

// ─────────────────────────────────────────────────────────────────────────────
// BinanceProtocol
// ─────────────────────────────────────────────────────────────────────────────

/// Declarative Binance WS protocol shim.
pub struct BinanceProtocol {
    _account_type: AccountType,
    _testnet: bool,
    urls: BinanceUrls,
}

impl BinanceProtocol {
    pub fn new(account_type: AccountType, testnet: bool) -> Self {
        let urls = if testnet {
            BinanceUrls::TESTNET
        } else {
            BinanceUrls::MAINNET
        };
        Self { _account_type: account_type, _testnet: testnet, urls }
    }

    fn spot_registry() -> &'static TopicRegistry {
        SPOT_REGISTRY.get_or_init(|| build_registry(AccountType::Spot))
    }

    fn futures_registry() -> &'static TopicRegistry {
        FUTURES_REGISTRY.get_or_init(|| build_registry(AccountType::FuturesCross))
    }

    /// Build the wire stream name for a StreamSpec (Binance combined-stream format).
    ///
    /// Symbol is always lowercase, e.g. "btcusdt".
    fn stream_name(spec: &StreamSpec) -> Result<String, WebSocketError> {
        // spec.symbol is now a raw exchange-native string (e.g. "BTCUSDT").
        // Binance combined-stream format requires it lowercase.
        let symbol = spec.symbol.to_lowercase();

        let name = match &spec.kind {
            StreamKind::Ticker => format!("{}@ticker", symbol),
            StreamKind::Trade => format!("{}@trade", symbol),
            StreamKind::AggTrade => format!("{}@aggTrade", symbol),

            StreamKind::Orderbook => {
                let depth = spec.depth.unwrap_or(20);
                let speed = spec.speed_ms.unwrap_or(100);
                format!("{}@depth{}@{}ms", symbol, depth, speed)
            }
            StreamKind::OrderbookDelta => {
                let speed = spec.speed_ms.unwrap_or(100);
                format!("{}@depth@{}ms", symbol, speed)
            }

            StreamKind::Kline { interval } => format!("{}@kline_{}", symbol, interval.as_str()),
            StreamKind::MarkPriceKline { interval } => {
                format!("{}@markPriceKline_{}", symbol, interval.as_str())
            }
            StreamKind::IndexPriceKline { interval } => {
                format!("{}@indexPriceKline_{}", symbol, interval.as_str())
            }
            StreamKind::PremiumIndexKline { interval } => {
                format!("{}@premiumIndexKline_{}", symbol, interval.as_str())
            }

            StreamKind::MarkPrice => {
                if spec.symbol.is_empty() {
                    "!markPrice@arr@1s".to_string()
                } else {
                    // @1s suffix is the post-2026-04-23 required wire name on /market path.
                    format!("{}@markPrice@1s", symbol)
                }
            }
            // FundingRate piggybacks the markPrice@1s frame (field "r" = funding rate).
            StreamKind::FundingRate => {
                if spec.symbol.is_empty() {
                    "!markPrice@arr@1s".to_string()
                } else {
                    format!("{}@markPrice@1s", symbol)
                }
            }

            StreamKind::Liquidation => {
                if spec.symbol.is_empty() {
                    "!forceOrder@arr".to_string()
                } else {
                    format!("{}@forceOrder", symbol)
                }
            }

            StreamKind::CompositeIndex => format!("{}@compositeIndex", symbol),
            StreamKind::IndexPrice => format!("{}@indexPrice@1s", symbol),

            StreamKind::OpenInterest => {
                return Err(WebSocketError::WireAbsent(
                    "Binance does not expose a realtime WS open interest feed — \
                     use REST GET /fapi/v1/openInterest with polling".to_string(),
                ));
            }

            // Private streams — no wire name needed (listenKey URL handles routing).
            StreamKind::OrderUpdate | StreamKind::BalanceUpdate | StreamKind::PositionUpdate => {
                return Err(WebSocketError::NotImplemented(
                    "binance: private streams use listenKey, not subscribe frames".into(),
                ));
            }

            other => {
                return Err(WebSocketError::NotImplemented(format!(
                    "binance: unsupported stream kind {:?}",
                    other
                )));
            }
        };

        Ok(name)
    }

    fn build_sub_frame(op: &str, spec: &StreamSpec) -> Result<WsFrame, WebSocketError> {
        let stream = Self::stream_name(spec)?;
        // Use a simple incrementing id via thread-local or constant — framework doesn't
        // inspect the id value, only the server uses it for correlation.
        let frame = json!({
            "method": op,
            "params": [stream],
            "id": 1u64,
        });
        Ok(WsFrame::Text(frame.to_string()))
    }
}

impl WsProtocol for BinanceProtocol {
    fn name(&self) -> &'static str {
        "binance"
    }

    fn endpoint(&self, account_type: AccountType, _testnet: bool) -> Url {
        // Use combined-stream endpoint for multiplexing.
        let base = self.urls.ws_url(account_type);
        let url = format!("{}/stream", base);
        Url::parse(&url).expect("binance ws url is valid")
    }

    /// Binance uses native WS Ping frames; server sends them, tokio-tungstenite
    /// auto-responds with Pong.  No application-level ping frame needed.
    fn ping_frame(&self) -> Option<WsFrame> {
        None
    }

    fn ping_interval(&self) -> Duration {
        // Binance closes after 24h of inactivity; 20s interval keeps connection warm.
        Duration::from_secs(20)
    }

    fn subscribe_frame(&self, spec: &StreamSpec) -> Result<WsFrame, WebSocketError> {
        Self::build_sub_frame("SUBSCRIBE", spec)
    }

    fn unsubscribe_frame(&self, spec: &StreamSpec) -> Result<WsFrame, WebSocketError> {
        Self::build_sub_frame("UNSUBSCRIBE", spec)
    }

    fn auth_frame(&self, _credentials: &Credentials) -> Option<Result<WsFrame, WebSocketError>> {
        // Binance public WS — no auth frame.
        // Private streams use listenKey URL rather than an auth frame.
        None
    }

    fn is_auth_ack(&self, _raw: &Value) -> bool {
        false
    }

    fn is_pong(&self, raw: &Value) -> bool {
        // Binance uses native WS pong, not a JSON pong frame.
        // This method is only called for text/binary JSON frames.
        // Native pong frames never reach here (transport handles them).
        // Return false — no JSON pong to recognize.
        let _ = raw;
        false
    }

    fn is_subscribe_ack(&self, raw: &Value) -> bool {
        // {"result":null,"id":N} or {"result":[...],"id":N}
        raw.get("id").is_some() && raw.get("result").is_some()
    }

    fn extract_topic(&self, raw: &Value) -> Option<TopicKey> {
        // Subscribe/unsubscribe ack: {"result":null,"id":N}
        if raw.get("id").is_some() && raw.get("result").is_some() {
            return None;
        }

        // Error frame: {"error":{"code":...,"msg":"..."},"id":N}
        if raw.get("error").is_some() {
            return None;
        }

        // Combined-stream frame: {"stream":"btcusdt@trade","data":{...}}
        if let Some(stream) = raw.get("stream").and_then(|s| s.as_str()) {
            return Some(TopicKey::new(stream));
        }

        // Single-stream frame (raw mode): look at "e" event type.
        // In raw mode the stream name isn't in the envelope; we reconstruct
        // a pseudo-topic from the event type so registry dispatch works.
        if let Some(event_type) = raw.get("e").and_then(|e| e.as_str()) {
            return Some(TopicKey::new(event_type));
        }

        // Partial depth snapshot (no "e" field, no "stream", has "lastUpdateId"):
        // These arrive in raw mode; map to a synthetic "partialDepth" topic.
        if raw.get("lastUpdateId").is_some() && raw.get("bids").is_some() {
            return Some(TopicKey::new("partialDepth"));
        }

        None
    }

    fn topic_registry(&self, account_type: AccountType) -> &TopicRegistry {
        match account_type {
            AccountType::Spot | AccountType::Margin | AccountType::Earn
            | AccountType::Lending | AccountType::Convert => Self::spot_registry(),
            _ => Self::futures_registry(),
        }
    }

    fn unsupported_by_exchange(&self, account_type: AccountType) -> &'static [StreamKind] {
        match account_type {
            AccountType::Spot | AccountType::Margin => SPOT_UNSUPPORTED,
            _ => &[],
        }
    }

    fn requires_auth_kinds(&self, _account_type: AccountType) -> &'static [StreamKind] {
        &[StreamKind::OrderUpdate, StreamKind::BalanceUpdate, StreamKind::PositionUpdate]
    }
}

static SPOT_UNSUPPORTED: &[StreamKind] = &[
    // Spot has no mark price, funding, or liquidation streams.
    StreamKind::MarkPrice,
    StreamKind::FundingRate,
    StreamKind::Liquidation,
];

// ─────────────────────────────────────────────────────────────────────────────
// Registry builder
// ─────────────────────────────────────────────────────────────────────────────

fn build_registry(account_type: AccountType) -> TopicRegistry {
    let mut b = TopicRegistry::builder();

    // ── Public market streams (spot + futures) ────────────────────────────
    b = b
        .register(StreamKind::Ticker, account_type, "*@ticker", parse_ticker)
        .register(StreamKind::Trade, account_type, "*@trade", parse_trade)
        .register(StreamKind::AggTrade, account_type, "*@aggTrade", parse_agg_trade)
        .register(StreamKind::OrderbookDelta, account_type, "*@depth@*", parse_depth_update)
        // Partial depth snapshots via combined-stream: "btcusdt@depth5@100ms"
        .register(StreamKind::Orderbook, account_type, "*@depth5@*", parse_partial_depth)
        .register(StreamKind::Orderbook, account_type, "*@depth10@*", parse_partial_depth)
        .register(StreamKind::Orderbook, account_type, "*@depth20@*", parse_partial_depth)
        // Raw-mode partial depth (no event type field, synthetic "partialDepth" topic)
        .register(StreamKind::Orderbook, account_type, "partialDepth", parse_partial_depth_raw)
        // miniTicker / bookTicker
        .register(StreamKind::Ticker, account_type, "*@miniTicker", parse_mini_ticker)
        .register(StreamKind::Ticker, account_type, "*@bookTicker", parse_book_ticker);

    // ── Kline streams (all intervals, same parser) ────────────────────────
    for (wire, internal) in BINANCE_KLINE_INTERVALS {
        let kind = StreamKind::Kline {
            interval: KlineInterval::new(*internal),
        };
        let pattern = format!("*@kline_{}", wire);
        b = b.register(kind, account_type, pattern, parse_kline);
    }

    // ── Futures-only streams ──────────────────────────────────────────────
    if !matches!(account_type, AccountType::Spot | AccountType::Margin) {
        b = b
            .register(StreamKind::MarkPrice, account_type, "*@markPrice", parse_mark_price)
            .register(StreamKind::MarkPrice, account_type, "*@markPrice@1s", parse_mark_price)
            .register(StreamKind::MarkPrice, account_type, "!markPrice@arr", parse_mark_price_arr)
            // FundingRate: same markPrice@1s frame, field "r" carries the rate.
            // Registered on both patterns so the topic key always matches.
            .register(StreamKind::FundingRate, account_type, "*@markPrice", parse_funding_rate)
            .register(StreamKind::FundingRate, account_type, "*@markPrice@1s", parse_funding_rate)
            .register(StreamKind::FundingRate, account_type, "!markPrice@arr", parse_funding_rate_arr)
            .register(StreamKind::Liquidation, account_type, "*@forceOrder", parse_force_order)
            .register(StreamKind::Liquidation, account_type, "!forceOrder@arr", parse_force_order_arr)
            .register(StreamKind::CompositeIndex, account_type, "*@compositeIndex", parse_composite_index)
            .register(StreamKind::IndexPrice, account_type, "*@indexPrice@1s", parse_index_price);

        // Futures kline variants
        for (wire, internal) in BINANCE_KLINE_INTERVALS {
            let mk_kind = StreamKind::MarkPriceKline {
                interval: KlineInterval::new(*internal),
            };
            let ix_kind = StreamKind::IndexPriceKline {
                interval: KlineInterval::new(*internal),
            };
            let pm_kind = StreamKind::PremiumIndexKline {
                interval: KlineInterval::new(*internal),
            };
            b = b
                .register(mk_kind, account_type, format!("*@markPriceKline_{}", wire), parse_mark_price_kline)
                .register(ix_kind, account_type, format!("*@indexPriceKline_{}", wire), parse_index_price_kline)
                .register(pm_kind, account_type, format!("*@premiumIndexKline_{}", wire), parse_premium_index_kline);
        }

        // Private stream event types (dispatched via raw event type key)
        b = b
            .register(StreamKind::OrderUpdate, account_type, "executionReport", parse_execution_report)
            .register(StreamKind::OrderUpdate, account_type, "ORDER_TRADE_UPDATE", parse_futures_order_update)
            .register(StreamKind::BalanceUpdate, account_type, "outboundAccountPosition", parse_account_position)
            .register(StreamKind::BalanceUpdate, account_type, "balanceUpdate", parse_balance_update)
            .register(StreamKind::BalanceUpdate, account_type, "ACCOUNT_UPDATE", parse_futures_account_update);
    } else {
        // Spot private streams
        b = b
            .register(StreamKind::OrderUpdate, account_type, "executionReport", parse_execution_report)
            .register(StreamKind::BalanceUpdate, account_type, "outboundAccountPosition", parse_account_position)
            .register(StreamKind::BalanceUpdate, account_type, "balanceUpdate", parse_balance_update);
    }

    b.build()
}

/// Binance wire kline suffixes → internal interval strings.
const BINANCE_KLINE_INTERVALS: &[(&str, &str)] = &[
    // 1s is native on Binance spot (`<sym>@kline_1s`) — register its parser so
    // the topic is matched instead of dropped as unmatched. Binance is the only
    // CEX with a 1-second WS kline; every other venue's sub-minute bars are
    // aggregated from trades by the station.
    ("1s", "1s"),
    ("1m", "1m"),
    ("3m", "3m"),
    ("5m", "5m"),
    ("15m", "15m"),
    ("30m", "30m"),
    ("1h", "1h"),
    ("2h", "2h"),
    ("4h", "4h"),
    ("6h", "6h"),
    ("8h", "8h"),
    ("12h", "12h"),
    ("1d", "1d"),
    ("3d", "3d"),
    ("1w", "1w"),
    ("1M", "1M"),
];

// ─────────────────────────────────────────────────────────────────────────────
// Parsers (fn(&Value) -> WebSocketResult<StreamEvent>)
//
// Each parser receives the full combined-stream frame:
//   {"stream":"btcusdt@trade","data":{...}}
// or the raw data object in raw-mode.
//
// Helper: extract "data" field from combined-stream envelope, or use frame
// directly if it looks like a raw-mode frame.
// ─────────────────────────────────────────────────────────────────────────────

/// Extract the inner data object from a combined-stream frame.
/// If "data" field exists, return it; otherwise return the frame itself.
fn frame_data(raw: &Value) -> &Value {
    raw.get("data").unwrap_or(raw)
}

fn parse_ticker(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let ticker = BinanceParser::parse_ticker(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    Ok(StreamEvent::Ticker { symbol, ticker })
}

fn parse_mini_ticker(raw: &Value) -> WebSocketResult<StreamEvent> {
    use crate::core::Ticker;

    let data = frame_data(raw);
    let parse_f64 = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    // @miniTicker short keys: e=eventType, E=eventTime, s=symbol,
    //   c=close, o=open, h=high, l=low, v=baseVolume, q=quoteVolume.
    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    Ok(StreamEvent::Ticker {
        symbol,
        ticker: Ticker {
            last_price: parse_f64("c").unwrap_or(0.0),
            bid_price: None,
            ask_price: None,
            high_24h: parse_f64("h"),
            low_24h: parse_f64("l"),
            volume_24h: parse_f64("v"),
            quote_volume_24h: parse_f64("q"),
            price_change_24h: None,
            price_change_percent_24h: None,
            timestamp: data.get("E").and_then(|t| t.as_i64()).unwrap_or(0),
            open_price: parse_f64("o"),
            ..Default::default()
        },
    })
}

fn parse_book_ticker(raw: &Value) -> WebSocketResult<StreamEvent> {
    use crate::core::Ticker;

    let data = frame_data(raw);
    let parse_f64 = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    // @bookTicker short keys: u=updateId, s=symbol, b=bidPrice, B=bidQty,
    //   a=askPrice, A=askQty, T=transactionTime, E=eventTime.
    // Live curl 2026-06-15: all verified present on both spot and futures.
    let bid = parse_f64("b");
    let ask = parse_f64("a");
    let last_price = bid.unwrap_or(0.0);

    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    Ok(StreamEvent::Ticker {
        symbol,
        ticker: Ticker {
            last_price,
            bid_price: bid,
            ask_price: ask,
            high_24h: None,
            low_24h: None,
            volume_24h: None,
            quote_volume_24h: None,
            price_change_24h: None,
            price_change_percent_24h: None,
            timestamp: data.get("T").and_then(|t| t.as_i64()).unwrap_or(0),
            bid_qty: parse_f64("B"),
            ask_qty: parse_f64("A"),
            update_id: data.get("u").and_then(|v| v.as_i64()),
            ..Default::default()
        },
    })
}

fn parse_trade(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let trade = BinanceParser::parse_ws_trade(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    Ok(StreamEvent::Trade { symbol, trade })
}

fn parse_agg_trade(raw: &Value) -> WebSocketResult<StreamEvent> {
    use crate::core::types::TradeSide;

    let data = frame_data(raw);
    let parse_f64 = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    // @aggTrade short keys: a=aggId, p=price, q=qty, f=firstId, l=lastId,
    //   T=timestamp, m=isBuyerMaker, M=isBestMatch (spot only), nq=nonRpiQty (futures only).
    // Live curl 2026-06-15: M present on spot, absent on futures; nq present on futures.
    let is_buyer_maker = data.get("m").and_then(|m| m.as_bool()).unwrap_or(false);
    let side = if is_buyer_maker { TradeSide::Sell } else { TradeSide::Buy };

    let _ = side;
    Ok(StreamEvent::AggTrade {
        symbol: data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string(),
        agg: crate::core::types::AggTrade {
            aggregate_id: data.get("a").and_then(|a| a.as_i64()).unwrap_or(0),
            price: parse_f64("p").unwrap_or(0.0),
            quantity: parse_f64("q").unwrap_or(0.0),
            first_trade_id: data.get("f").and_then(|f| f.as_i64()).unwrap_or(0),
            last_trade_id: data.get("l").and_then(|l| l.as_i64()).unwrap_or(0),
            is_buy: !is_buyer_maker,
            timestamp: data.get("T").and_then(|t| t.as_i64()).unwrap_or(0),
            is_best_match: data.get("M").and_then(|v| v.as_bool()),
            non_rpi_qty: parse_f64("nq"),
            ..Default::default()
        },
    })
}

fn parse_levels(data: &Value, key: &str) -> Vec<OrderBookLevel> {
    data.get(key)
        .and_then(|v| v.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|pair| {
                    let p = pair.get(0)?.as_str()?.parse().ok()?;
                    let s = pair.get(1)?.as_str()?.parse().ok()?;
                    Some(OrderBookLevel::new(p, s))
                })
                .collect()
        })
        .unwrap_or_default()
}

fn parse_depth_update(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event_time = data.get("E").and_then(|e| e.as_i64());
    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();

    Ok(StreamEvent::OrderbookDelta {
        symbol,
        delta: OrderbookDeltaData {
            bids: parse_levels(data, "b"),
            asks: parse_levels(data, "a"),
            timestamp: event_time.unwrap_or(0),
            first_update_id: data.get("U").and_then(|v| v.as_u64()),
            last_update_id: data.get("u").and_then(|v| v.as_u64()),
            prev_update_id: data.get("pu").and_then(|v| v.as_u64()),
            event_time,
            checksum: None,
        },
    })
}

fn parse_partial_depth(raw: &Value) -> WebSocketResult<StreamEvent> {
    // Combined-stream partial depth: {"stream":"btcusdt@depth20@100ms","data":{...}}
    // data has "lastUpdateId", "bids", "asks" — no "e" event type.
    // Symbol comes from the stream name, not from data.
    let symbol = raw
        .get("stream")
        .and_then(|s| s.as_str())
        .and_then(|s| s.split('@').next())
        .map(|s| s.to_ascii_uppercase())
        .unwrap_or_default();
    let data = frame_data(raw);
    parse_partial_depth_inner(data, symbol)
}

fn parse_partial_depth_raw(raw: &Value) -> WebSocketResult<StreamEvent> {
    // Raw-mode partial depth (single-stream URL): the frame IS the data.
    // No symbol embedded; StreamSpec carries it at dispatch level.
    parse_partial_depth_inner(raw, String::new())
}

fn parse_partial_depth_inner(data: &Value, symbol: String) -> WebSocketResult<StreamEvent> {
    let event_time = data.get("E").and_then(|e| e.as_i64());

    Ok(StreamEvent::OrderbookSnapshot {
        symbol,
        book: crate::core::OrderBook {
            bids: parse_levels(data, "bids"),
            asks: parse_levels(data, "asks"),
            timestamp: event_time.unwrap_or(0),
            sequence: None,
            last_update_id: data.get("lastUpdateId").and_then(|v| v.as_u64()),
            first_update_id: None,
            prev_update_id: None,
            event_time,
            transaction_time: None,
            checksum: None,
            ..Default::default()
        },
    })
}

fn parse_kline(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let kline = BinanceParser::parse_ws_kline(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    let interval = KlineInterval::new(
        data.get("k")
            .and_then(|k| k.get("i"))
            .and_then(|i| i.as_str())
            .unwrap_or(""),
    );
    Ok(StreamEvent::Kline { symbol, interval, kline })
}

fn parse_mark_price(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let parse_f64 = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    // @markPrice@1s short keys (futures only):
    //   s=symbol, p=markPrice, i=indexPrice, P=estimatedSettlePrice,
    //   r=fundingRate, T=nextFundingTime, R=interestRate, E=eventTime.
    // Source: Binance USDⓈ-M WS mark price stream spec.
    Ok(StreamEvent::MarkPrice {
        symbol: data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string(),
        mark: crate::core::types::MarkPrice {
            mark_price: parse_f64("p").unwrap_or(0.0),
            index_price: parse_f64("i"),
            estimated_settle_price: parse_f64("P"),
            funding_rate: parse_f64("r"),
            next_funding_time: data.get("T").and_then(|t| t.as_i64()),
            interest_rate: parse_f64("R"),
            timestamp: data.get("E").and_then(|e| e.as_i64()).unwrap_or(0),
            ..Default::default()
        },
    })
}

fn parse_mark_price_arr(raw: &Value) -> WebSocketResult<StreamEvent> {
    // !markPrice@arr arrives as {"stream":"!markPrice@arr","data":[{...},{...}]}
    // Emit first element; the transport's multi-emit logic handles arrays.
    let data = frame_data(raw);
    let item = if let Some(arr) = data.as_array() {
        arr.first().unwrap_or(data)
    } else {
        data
    };
    let parse_f64 = |key: &str| -> Option<f64> {
        item.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| item.get(key).and_then(|v| v.as_f64()))
    };
    Ok(StreamEvent::MarkPrice {
        symbol: item.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string(),
        mark: crate::core::types::MarkPrice {
            mark_price: parse_f64("p").unwrap_or(0.0),
            index_price: parse_f64("i"),
            estimated_settle_price: parse_f64("P"),
            funding_rate: parse_f64("r"),
            next_funding_time: item.get("T").and_then(|t| t.as_i64()),
            interest_rate: parse_f64("R"),
            timestamp: item.get("E").and_then(|e| e.as_i64()).unwrap_or(0),
            ..Default::default()
        },
    })
}

/// Parse a markPrice@1s frame and emit `StreamEvent::FundingRate`.
/// The "r" field carries the funding rate; "T" is next funding time ms.
/// Returns `WebSocketError::Parse` if "r" is absent (stream has no funding for this symbol).
fn parse_funding_rate(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let parse_f64 = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    let rate = parse_f64("r").ok_or_else(|| {
        WebSocketError::Parse("markPrice frame: 'r' (funding rate) field absent".into())
    })?;

    Ok(StreamEvent::FundingRate {
        symbol: data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string(),
        funding: crate::core::types::FundingRate {
            rate,
            next_funding_time: data.get("T").and_then(|t| t.as_i64()),
            timestamp: data.get("E").and_then(|e| e.as_i64()).unwrap_or(0),
            ..Default::default()
        },
    })
}

/// Parse !markPrice@arr frame and emit FundingRate from first element.
fn parse_funding_rate_arr(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let item = if let Some(arr) = data.as_array() {
        arr.first().unwrap_or(data)
    } else {
        data
    };
    // Wrap as combined frame so parse_funding_rate's frame_data call works correctly.
    let wrapped = serde_json::json!({"data": item});
    parse_funding_rate(&wrapped)
}

fn parse_force_order(raw: &Value) -> WebSocketResult<StreamEvent> {
    use crate::core::types::TradeSide;

    let data = frame_data(raw);
    let o = data.get("o").ok_or_else(|| {
        WebSocketError::Parse("forceOrder: missing 'o' field".into())
    })?;

    let parse_f64 = |key: &str| -> Option<f64> {
        o.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| o.get(key).and_then(|v| v.as_f64()))
    };

    let side = match o.get("S").and_then(|s| s.as_str()).unwrap_or("") {
        "BUY" => TradeSide::Buy,
        _ => TradeSide::Sell,
    };

    // ap = average filled price (better than original p for executed liquidations).
    // z = accumulated filled qty (actual size liquidated, better than original q).
    let price = parse_f64("ap").unwrap_or_else(|| parse_f64("p").unwrap_or(0.0));
    let quantity = parse_f64("z").unwrap_or_else(|| parse_f64("q").unwrap_or(0.0));

    Ok(StreamEvent::Liquidation {
        symbol: o.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string(),
        liquidation: crate::core::types::Liquidation {
            symbol: o.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string(),
            side,
            price,
            quantity,
            timestamp: o.get("T").and_then(|t| t.as_i64()).unwrap_or(0),
            value: Some(price * quantity),
            order_type: o.get("o").and_then(|v| v.as_str()).map(String::from),
            status: o.get("X").and_then(|v| v.as_str()).map(String::from),
            avg_price: parse_f64("ap"),
            executed_qty: parse_f64("z"),
            order_qty: parse_f64("q"),
            order_price: parse_f64("p"),
            ..Default::default()
        },
    })
}

fn parse_force_order_arr(raw: &Value) -> WebSocketResult<StreamEvent> {
    // !forceOrder@arr combined-stream frame:
    //   {"stream":"!forceOrder@arr","data":{"e":"forceOrder","E":...,"o":{...}}}
    // The "data" field is a single event object identical in shape to the
    // per-symbol forceOrder event — NOT an array despite the "@arr" suffix.
    // Delegate directly: parse_force_order calls frame_data which extracts "data".
    parse_force_order(raw)
}

fn parse_composite_index(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let parse_f64_field = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    let price = parse_f64_field("p").unwrap_or(0.0);
    let timestamp = data.get("E").and_then(|e| e.as_i64()).unwrap_or(0);

    let components: Vec<(String, f64)> = data
        .get("c")
        .and_then(|c| c.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|item| {
                    let base = item.get("b").and_then(|v| v.as_str()).unwrap_or("");
                    let quote = item.get("q").and_then(|v| v.as_str()).unwrap_or("");
                    let comp_symbol = format!("{}{}", base, quote);
                    let weight = item
                        .get("W")
                        .and_then(|v| v.as_str())
                        .and_then(|s| s.parse::<f64>().ok())
                        .or_else(|| item.get("W").and_then(|v| v.as_f64()))
                        .or_else(|| {
                            item.get("w")
                                .and_then(|v| v.as_str())
                                .and_then(|s| s.parse().ok())
                        })
                        .or_else(|| item.get("w").and_then(|v| v.as_f64()))
                        .unwrap_or(0.0);
                    if comp_symbol.is_empty() {
                        None
                    } else {
                        Some((comp_symbol, weight))
                    }
                })
                .collect()
        })
        .unwrap_or_default();

    Ok(StreamEvent::CompositeIndex {
        symbol,
        index: crate::core::types::CompositeIndex { price, components, timestamp },
    })
}

fn parse_index_price(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let parse_f64_field = |key: &str| -> Option<f64> {
        data.get(key)
            .and_then(|v| v.as_str())
            .and_then(|s| s.parse().ok())
            .or_else(|| data.get(key).and_then(|v| v.as_f64()))
    };

    let symbol = data
        .get("i")
        .and_then(|s| s.as_str())
        .or_else(|| data.get("s").and_then(|s| s.as_str()))
        .unwrap_or("")
        .to_string();

    Ok(StreamEvent::IndexPrice {
        symbol,
        index_price: crate::core::types::IndexPrice {
            price: parse_f64_field("p").unwrap_or(0.0),
            timestamp: data.get("E").and_then(|e| e.as_i64()).unwrap_or(0),
            ..Default::default()
        },
    })
}

fn parse_mark_price_kline(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event = BinanceParser::parse_ws_mark_price_kline(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    Ok(event)
}

fn parse_index_price_kline(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event = BinanceParser::parse_ws_index_price_kline(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    Ok(event)
}

fn parse_premium_index_kline(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event = BinanceParser::parse_ws_premium_index_kline(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    Ok(event)
}

fn parse_execution_report(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let symbol = data.get("s").and_then(|s| s.as_str()).unwrap_or("").to_string();
    let event = BinanceParser::parse_ws_execution_report(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    Ok(StreamEvent::OrderUpdate { symbol, event })
}

fn parse_futures_order_update(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let symbol = data.get("o")
        .and_then(|o| o.get("s"))
        .and_then(|s| s.as_str())
        .unwrap_or("")
        .to_string();
    let event = BinanceParser::parse_ws_futures_order_update(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    Ok(StreamEvent::OrderUpdate { symbol, event })
}

fn parse_account_position(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event = BinanceParser::parse_ws_account_position(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    match event {
        Some(ev) => Ok(StreamEvent::BalanceUpdate(ev)),
        None => Err(WebSocketError::Parse(
            "outboundAccountPosition: no non-zero balance found".into(),
        )),
    }
}

fn parse_balance_update(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event = BinanceParser::parse_ws_balance_update(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    Ok(StreamEvent::BalanceUpdate(event))
}

fn parse_futures_account_update(raw: &Value) -> WebSocketResult<StreamEvent> {
    let data = frame_data(raw);
    let event = BinanceParser::parse_ws_futures_account_update(data)
        .map_err(|e| WebSocketError::Parse(e.to_string()))?;
    match event {
        Some(ev) => Ok(StreamEvent::BalanceUpdate(ev)),
        None => Err(WebSocketError::Parse(
            "ACCOUNT_UPDATE: no balance entry found".into(),
        )),
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::websocket::StreamSpec;

    fn spot_spec(kind: StreamKind) -> StreamSpec {
        StreamSpec {
            kind,
            symbol: crate::core::types::OwnedSymbolInput::Raw("BTCUSDT".to_string()),
            account_type: AccountType::Spot,
            depth: None,
            speed_ms: None,
        }
    }

    #[test]
    fn test_topic_registry_non_empty() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let reg = proto.topic_registry(AccountType::Spot);
        let keys: Vec<_> = reg.native_pairs().collect();
        assert!(!keys.is_empty(), "spot registry must have entries");
        assert!(reg.supports(&StreamKind::Ticker, AccountType::Spot));
        assert!(reg.supports(&StreamKind::Trade, AccountType::Spot));
        assert!(reg.supports(
            &StreamKind::Kline { interval: KlineInterval::new("1m") },
            AccountType::Spot
        ));

        let futures_proto = BinanceProtocol::new(AccountType::FuturesCross, false);
        let freg = futures_proto.topic_registry(AccountType::FuturesCross);
        assert!(freg.supports(&StreamKind::MarkPrice, AccountType::FuturesCross));
        assert!(freg.supports(&StreamKind::Liquidation, AccountType::FuturesCross));
    }

    #[test]
    fn test_subscribe_frame_spot_trade() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let spec = spot_spec(StreamKind::Trade);
        let msg = proto.subscribe_frame(&spec).expect("subscribe_frame must succeed");
        let text = match msg {
            WsFrame::Text(t) => t,
            _ => panic!("expected text frame"),
        };
        let v: serde_json::Value = serde_json::from_str(&text).expect("valid JSON");
        assert_eq!(v["method"], "SUBSCRIBE");
        let params = v["params"].as_array().expect("params array");
        assert_eq!(params.len(), 1);
        // Symbol must be lowercase
        assert_eq!(params[0], "btcusdt@trade");
    }

    #[test]
    fn test_extract_topic_combined_stream() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let frame = serde_json::json!({
            "stream": "btcusdt@trade",
            "data": {"e": "trade", "s": "BTCUSDT", "p": "50000", "q": "0.1"}
        });
        let topic = proto.extract_topic(&frame).expect("should extract topic");
        assert_eq!(topic.as_str(), "btcusdt@trade");
    }

    #[test]
    fn test_extract_topic_subscribe_ack() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let ack = serde_json::json!({"result": null, "id": 1});
        assert!(proto.extract_topic(&ack).is_none());
    }

    #[test]
    fn test_is_subscribe_ack() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let ack = serde_json::json!({"result": null, "id": 1});
        assert!(proto.is_subscribe_ack(&ack));
        let not_ack = serde_json::json!({"stream": "btcusdt@trade", "data": {}});
        assert!(!proto.is_subscribe_ack(&not_ack));
    }

    #[test]
    fn test_ping_frame_is_none() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        assert!(
            proto.ping_frame().is_none(),
            "Binance uses native WS ping, not application-level"
        );
    }

    #[test]
    fn test_kline_registry_all_intervals() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let reg = proto.topic_registry(AccountType::Spot);
        for (_, internal) in BINANCE_KLINE_INTERVALS {
            let kind = StreamKind::Kline {
                interval: KlineInterval::new(*internal),
            };
            assert!(
                reg.supports(&kind, AccountType::Spot),
                "spot registry missing kline interval {}",
                internal
            );
        }
    }

    #[test]
    fn test_subscribe_kline_frame() {
        let proto = BinanceProtocol::new(AccountType::Spot, false);
        let spec = spot_spec(StreamKind::Kline {
            interval: KlineInterval::new("1h"),
        });
        let msg = proto.subscribe_frame(&spec).expect("subscribe_frame");
        let text = match msg {
            WsFrame::Text(t) => t,
            _ => panic!("expected text"),
        };
        let v: serde_json::Value = serde_json::from_str(&text).expect("valid JSON");
        assert_eq!(v["params"][0], "btcusdt@kline_1h");
    }
}
