foxhunt/data/src/providers/benzinga/streaming.rs

//! # Benzinga Streaming Provider
//!
//! Real-time WebSocket streaming provider for Benzinga Pro API, focusing on news, sentiment,
//! analyst ratings, and unusual options activity. This provider implements high-frequency
//! data streaming with automatic reconnection and event normalization.
//!
//! ## Features
//!
//! - **Real-time News**: Breaking financial news with impact scoring
//! - **Sentiment Analysis**: AI-powered sentiment scores for symbols
//! - **Analyst Ratings**: Upgrades, downgrades, and price target changes
//! - **Unusual Options**: Detection of unusual options flow and large trades
//! - **Auto Reconnection**: Exponential backoff with circuit breaker
//! - **Event Normalization**: Unified MarketDataEvent format for trading pipeline
//!
//! ## Usage
//!
//! ```rust,no_run
//! use data::providers::benzinga::streaming::BenzingaStreamingProvider;
//! use data::providers::traits::RealTimeProvider;
//! use trading_engine::types::Symbol;
//!
//! # async fn example() -> anyhow::Result<()> {
//! let config = BenzingaStreamingConfig {
//!     api_key: "your-api-key".to_string(),
//!     ..Default::default()
//! };
//!
//! let mut provider = BenzingaStreamingProvider::new(config)?;
//! provider.connect().await?;
//! provider.subscribe(vec![Symbol::from("AAPL"), Symbol::from("SPY")]).await?;
//!
//! let mut stream = provider.stream().await?;
//! while let Some(event) = stream.next().await {
//!     // Process news, sentiment, and ratings events
//!     println!("Received: {:?}", event);
//! }
//! # Ok(())
//! # }
//! ```

use crate::error::{DataError, Result};
use crate::providers::common::{
    AnalystRatingEvent, ConnectionState, ConnectionStatusEvent, ErrorCategory,
    NewsEvent, OptionsContract, OptionsSentiment, OptionsType, RatingAction, SentimentEvent,
    SentimentPeriod, UnusualOptionsEvent, UnusualOptionsType,
};
use crate::providers::traits::{
    ConnectionState as TraitConnectionState, ConnectionStatus, RealTimeProvider,
};
use crate::providers::common::MarketDataEvent;
use trading_engine::types::ErrorEvent;
use crate::types::ConnectionEvent;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::net::TcpStream;
use tokio::sync::{mpsc, Mutex, RwLock};
use tokio_stream::Stream;
use std::pin::Pin;
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
use tracing::{debug, error, info, warn};
use trading_engine::types::prelude::Decimal;
use trading_engine::types::Symbol;

/// Configuration for Benzinga streaming provider
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BenzingaStreamingConfig {
    /// Benzinga Pro API key
    pub api_key: String,

    /// WebSocket endpoint URL
    pub websocket_url: String,

    /// Connection timeout in seconds
    pub connect_timeout_secs: u64,

    /// Ping interval in seconds
    pub ping_interval_secs: u64,

    /// Maximum reconnection attempts
    pub max_reconnect_attempts: u32,

    /// Initial reconnection delay in milliseconds
    pub initial_reconnect_delay_ms: u64,

    /// Maximum reconnection delay in milliseconds
    pub max_reconnect_delay_ms: u64,

    /// Reconnection backoff multiplier
    pub reconnect_backoff_multiplier: f64,

    /// Enable news alerts
    pub enable_news: bool,

    /// Enable sentiment updates
    pub enable_sentiment: bool,

    /// Enable analyst ratings
    pub enable_ratings: bool,

    /// Enable unusual options activity
    pub enable_options: bool,

    /// Buffer size for event channel
    pub event_buffer_size: usize,

    /// Heartbeat timeout in seconds
    pub heartbeat_timeout_secs: u64,
}

impl Default for BenzingaStreamingConfig {
    fn default() -> Self {
        Self {
            api_key: std::env::var("BENZINGA_API_KEY").unwrap_or_default(),
            websocket_url: "wss://api.benzinga.com/api/v1/stream".to_string(),
            connect_timeout_secs: 30,
            ping_interval_secs: 30,
            max_reconnect_attempts: 10,
            initial_reconnect_delay_ms: 1000,
            max_reconnect_delay_ms: 30000,
            reconnect_backoff_multiplier: 2.0,
            enable_news: true,
            enable_sentiment: true,
            enable_ratings: true,
            enable_options: true,
            event_buffer_size: 10000,
            heartbeat_timeout_secs: 60,
        }
    }
}

/// Benzinga WebSocket streaming provider
pub struct BenzingaStreamingProvider {
    /// Provider configuration
    config: BenzingaStreamingConfig,

    /// Current connection status
    connection_status: Arc<RwLock<ConnectionStatus>>,

    /// WebSocket connection
    websocket: Arc<Mutex<Option<WebSocketStream<MaybeTlsStream<TcpStream>>>>>,

    /// Event sender channel
    event_tx: Arc<Mutex<Option<mpsc::UnboundedSender<MarketDataEvent>>>>,

    /// Event receiver channel for streaming
    event_rx: Arc<Mutex<Option<mpsc::UnboundedReceiver<MarketDataEvent>>>>,

    /// Subscribed symbols
    subscribed_symbols: Arc<RwLock<HashSet<Symbol>>>,

    /// Reconnection state
    reconnect_attempt: Arc<Mutex<u32>>,

    /// Last heartbeat time
    last_heartbeat: Arc<Mutex<Instant>>,

    /// Connection metrics
    metrics: Arc<RwLock<ConnectionMetrics>>,

    /// Shutdown signal
    shutdown_tx: Arc<Mutex<Option<mpsc::UnboundedSender<()>>>>,
}

/// Connection metrics for monitoring
#[derive(Debug, Default)]
struct ConnectionMetrics {
    /// Total messages received
    messages_received: u64,

    /// Messages received per second (rolling average)
    messages_per_second: f64,

    /// Connection start time
    connection_start: Option<Instant>,

    /// Last message timestamp
    last_message_time: Option<DateTime<Utc>>,

    /// Error count
    error_count: u32,

    /// Reconnection count
    reconnection_count: u32,
}

/// Benzinga WebSocket message types
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "type")]
enum BenzingaMessage {
    #[serde(rename = "news")]
    News(BenzingaNewsMessage),

    #[serde(rename = "sentiment")]
    Sentiment(BenzingaSentimentMessage),

    #[serde(rename = "rating")]
    Rating(BenzingaRatingMessage),

    #[serde(rename = "options")]
    Options(BenzingaOptionsMessage),

    #[serde(rename = "heartbeat")]
    Heartbeat(BenzingaHeartbeatMessage),

    #[serde(rename = "error")]
    Error(BenzingaErrorMessage),

    #[serde(rename = "subscription_confirmation")]
    SubscriptionConfirmation(BenzingaSubscriptionMessage),
}

/// Benzinga news message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaNewsMessage {
    /// Story ID
    pub story_id: String,

    /// Headline
    pub headline: String,

    /// Summary
    pub summary: Option<String>,

    /// Symbols
    pub tickers: Vec<String>,

    /// Category
    pub category: String,

    /// Tags
    pub tags: Vec<String>,

    /// Impact score (-1.0 to 1.0)
    pub impact_score: Option<f64>,

    /// Author
    pub author: Option<String>,

    /// Source
    pub source: String,

    /// Publication timestamp (ISO 8601)
    pub published_at: String,

    /// URL
    pub url: Option<String>,
}

/// Benzinga sentiment message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaSentimentMessage {
    /// Symbol
    pub ticker: String,

    /// Sentiment score (-1.0 to 1.0)
    pub sentiment_score: f64,

    /// Bullish percentage (0.0 to 1.0)
    pub bullish_ratio: f64,

    /// Bearish percentage (0.0 to 1.0)
    pub bearish_ratio: f64,

    /// Sample size
    pub sample_size: u32,

    /// Period
    pub period: String,

    /// Sources
    pub sources: Vec<String>,

    /// Confidence
    pub confidence: Option<f64>,

    /// Timestamp
    pub timestamp: String,
}

/// Benzinga rating message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaRatingMessage {
    /// Symbol
    pub ticker: String,

    /// Analyst name
    pub analyst: String,

    /// Firm name
    pub firm: String,

    /// Action (upgrade, downgrade, etc.)
    pub action: String,

    /// Current rating
    pub current_rating: String,

    /// Previous rating
    pub previous_rating: Option<String>,

    /// Price target
    pub price_target: Option<f64>,

    /// Previous price target
    pub previous_price_target: Option<f64>,

    /// Comment
    pub comment: Option<String>,

    /// Rating date
    pub rating_date: String,

    /// Timestamp
    pub timestamp: String,
}

/// Benzinga options message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaOptionsMessage {
    /// Underlying symbol
    pub ticker: String,

    /// Strike price
    pub strike: f64,

    /// Expiration date
    pub expiration: String,

    /// Option type (call/put)
    pub option_type: String,

    /// Activity type
    pub activity_type: String,

    /// Volume
    pub volume: u32,

    /// Open interest
    pub open_interest: Option<u32>,

    /// Premium
    pub premium: Option<f64>,

    /// Implied volatility
    pub implied_volatility: Option<f64>,

    /// Sentiment
    pub sentiment: String,

    /// Confidence
    pub confidence: f64,

    /// Description
    pub description: String,

    /// Timestamp
    pub timestamp: String,
}

/// Benzinga heartbeat message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaHeartbeatMessage {
    /// Server timestamp
    pub timestamp: String,

    /// Connection ID
    pub connection_id: Option<String>,
}

/// Benzinga error message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaErrorMessage {
    /// Error code
    pub code: String,

    /// Error message
    pub message: String,

    /// Additional details
    pub details: Option<String>,
}

/// Benzinga subscription message
#[derive(Debug, Serialize, Deserialize)]
struct BenzingaSubscriptionMessage {
    /// Subscription type
    pub subscription_type: String,

    /// Symbols
    pub tickers: Vec<String>,

    /// Status
    pub status: String,
}

/// WebSocket subscription request
#[derive(Debug, Serialize)]
struct SubscriptionRequest {
    /// Request type
    #[serde(rename = "type")]
    pub request_type: String,

    /// API key
    pub api_key: String,

    /// Symbols to subscribe to
    pub tickers: Vec<String>,

    /// Event types to subscribe to
    pub events: Vec<String>,
}

impl BenzingaStreamingProvider {
    /// Create a new Benzinga streaming provider
    pub fn new(config: BenzingaStreamingConfig) -> Result<Self> {
        if config.api_key.is_empty() {
            return Err(DataError::Configuration {
                field: "api_key".to_string(),
                message: "Benzinga API key is required".to_string(),
            });
        }

        let (event_tx, event_rx) = mpsc::unbounded_channel();

        Ok(Self {
            config,
            connection_status: Arc::new(RwLock::new(ConnectionStatus::disconnected())),
            websocket: Arc::new(Mutex::new(None)),
            event_tx: Arc::new(Mutex::new(Some(event_tx))),
            event_rx: Arc::new(Mutex::new(Some(event_rx))),
            subscribed_symbols: Arc::new(RwLock::new(HashSet::new())),
            reconnect_attempt: Arc::new(Mutex::new(0)),
            last_heartbeat: Arc::new(Mutex::new(Instant::now())),
            metrics: Arc::new(RwLock::new(ConnectionMetrics::default())),
            shutdown_tx: Arc::new(Mutex::new(None)),
        })
    }

    /// Establish WebSocket connection
    async fn connect_websocket(&self) -> Result<()> {
        let url = &self.config.websocket_url;

        info!("Connecting to Benzinga WebSocket: {}", url);

        let (ws_stream, response) = tokio::time::timeout(
            Duration::from_secs(self.config.connect_timeout_secs),
            connect_async(url),
        )
        .await
        .map_err(|_| DataError::timeout("WebSocket connection timeout"))?
        .map_err(|e| DataError::WebSocket(e))?;

        info!(
            "Connected to Benzinga WebSocket, response: {}",
            response.status()
        );

        // Store the WebSocket connection
        {
            let mut websocket = self.websocket.lock().await;
            *websocket = Some(ws_stream);
        }

        // Update connection status
        {
            let mut status = self.connection_status.write().await;
            status.state = TraitConnectionState::Connected;
            status.last_connection_attempt = Some(Utc::now());
        }

        // Update metrics
        {
            let mut metrics = self.metrics.write().await;
            metrics.connection_start = Some(Instant::now());
            metrics.reconnection_count += 1;
        }

        // Reset reconnection attempt counter
        {
            let mut attempt = self.reconnect_attempt.lock().await;
            *attempt = 0;
        }

        Ok(())
    }

    /// Send subscription request
    async fn send_subscription(&self, symbols: Vec<Symbol>) -> Result<()> {
        let mut events = Vec::new();

        if self.config.enable_news {
            events.push("news".to_string());
        }
        if self.config.enable_sentiment {
            events.push("sentiment".to_string());
        }
        if self.config.enable_ratings {
            events.push("ratings".to_string());
        }
        if self.config.enable_options {
            events.push("options".to_string());
        }

        let subscription = SubscriptionRequest {
            request_type: "subscribe".to_string(),
            api_key: self.config.api_key.clone(),
            tickers: symbols.iter().map(|s| s.to_string()).collect(),
            events,
        };

        let message =
            serde_json::to_string(&subscription).map_err(|e| DataError::Serialization {
                message: format!("Failed to serialize subscription: {}", e),
            })?;

        // Send subscription message
        {
            let mut websocket_guard = self.websocket.lock().await;
            if let Some(websocket) = websocket_guard.as_mut() {
                websocket
                    .send(Message::Text(message))
                    .await
                    .map_err(|e| DataError::WebSocket(e))?;

                debug!("Sent subscription for symbols: {:?}", symbols);
            } else {
                return Err(DataError::Connection("No WebSocket connection".to_string()));
            }
        }

        Ok(())
    }

    /// Start the message processing loop
    async fn start_message_loop(&self) -> Result<()> {
        let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel();

        // Store shutdown sender
        {
            let mut tx = self.shutdown_tx.lock().await;
            *tx = Some(shutdown_tx);
        }

        let websocket = self.websocket.clone();
        let event_tx = self.event_tx.clone();
        let connection_status = self.connection_status.clone();
        let metrics = self.metrics.clone();
        let last_heartbeat = self.last_heartbeat.clone();
        let heartbeat_timeout = Duration::from_secs(self.config.heartbeat_timeout_secs);

        tokio::spawn(async move {
            let mut heartbeat_interval = tokio::time::interval(Duration::from_secs(30));

            loop {
                tokio::select! {
                    // Check for shutdown signal
                    _ = shutdown_rx.recv() => {
                        debug!("Received shutdown signal");
                        break;
                    }

                    // Handle heartbeat timeout
                    _ = heartbeat_interval.tick() => {
                        let last_beat = {
                            let guard = last_heartbeat.lock().await;
                            *guard
                        };

                        if last_beat.elapsed() > heartbeat_timeout {
                            error!("Heartbeat timeout detected");

                            // Update connection status
                            {
                                let mut status = connection_status.write().await;
                                status.state = TraitConnectionState::Failed;
                            }

                            // Send error event
                            if let Some(tx) = event_tx.lock().await.as_ref() {
                                let error_event = MarketDataEvent::Error(ErrorEvent {
                                    provider: "benzinga".to_string(),
                                    message: "Heartbeat timeout".to_string(),
                                    code: Some("HEARTBEAT_TIMEOUT".to_string()),
                                    category: ErrorCategory::Connection,
                                    recoverable: true,
                                    timestamp: Utc::now(),
                                });

                                let _ = tx.send(error_event);
                            }
                            break;
                        }
                    }

                    // Process WebSocket messages
                    message_result = async {
                        let mut websocket_guard = websocket.lock().await;
                        if let Some(ws) = websocket_guard.as_mut() {
                            ws.next().await
                        } else {
                            None
                        }
                    } => {
                        if let Some(message_result) = message_result {
                            match message_result {
                                Ok(message) => {
                                    if let Err(e) = Self::process_message(
                                        message,
                                        &event_tx,
                                        &metrics,
                                        &last_heartbeat
                                    ).await {
                                        error!("Failed to process message: {}", e);
                                    }
                                }
                                Err(e) => {
                                    error!("WebSocket error: {}", e);

                                    // Update connection status
                                    {
                                        let mut status = connection_status.write().await;
                                        status.state = TraitConnectionState::Failed;
                                    }
                                    break;
                                }
                            }
                        }
                    }
                }
            }

            debug!("Message processing loop ended");
        });

        Ok(())
    }

    /// Process a WebSocket message
    async fn process_message(
        message: Message,
        event_tx: &Arc<Mutex<Option<mpsc::UnboundedSender<MarketDataEvent>>>>,
        metrics: &Arc<RwLock<ConnectionMetrics>>,
        last_heartbeat: &Arc<Mutex<Instant>>,
    ) -> Result<()> {
        match message {
            Message::Text(text) => {
                debug!("Received text message: {}", text);

                // Update metrics
                {
                    let mut m = metrics.write().await;
                    m.messages_received += 1;
                    m.last_message_time = Some(Utc::now());

                    // Calculate messages per second (simple moving average)
                    if let Some(start_time) = m.connection_start {
                        let elapsed_secs = start_time.elapsed().as_secs_f64();
                        if elapsed_secs > 0.0 {
                            m.messages_per_second = m.messages_received as f64 / elapsed_secs;
                        }
                    }
                }

                // Parse and process the message
                match serde_json::from_str::<BenzingaMessage>(&text) {
                    Ok(benzinga_msg) => {
                        if let Some(market_event) =
                            Self::convert_benzinga_message(benzinga_msg).await?
                        {
                            // Send event to stream
                            if let Some(tx) = event_tx.lock().await.as_ref() {
                                if let Err(e) = tx.send(market_event) {
                                    error!("Failed to send market event: {}", e);
                                }
                            }
                        }
                    }
                    Err(e) => {
                        warn!(
                            "Failed to parse Benzinga message: {}. Raw message: {}",
                            e, text
                        );

                        // Update error metrics
                        {
                            let mut m = metrics.write().await;
                            m.error_count += 1;
                        }
                    }
                }
            }
            Message::Binary(_) => {
                warn!("Received unexpected binary message");
            }
            Message::Ping(payload) => {
                debug!("Received ping, will send pong");
                // WebSocket library handles pong automatically
            }
            Message::Pong(_) => {
                debug!("Received pong");

                // Update heartbeat
                {
                    let mut heartbeat = last_heartbeat.lock().await;
                    *heartbeat = Instant::now();
                }
            }
            Message::Close(_) => {
                info!("Received close message");
                return Err(DataError::Connection(
                    "WebSocket closed by server".to_string(),
                ));
            }
            Message::Frame(_) => {
                // Internal frame, ignore
            }
        }

        Ok(())
    }

    /// Convert Benzinga message to MarketDataEvent
    async fn convert_benzinga_message(message: BenzingaMessage) -> Result<Option<MarketDataEvent>> {
        match message {
            BenzingaMessage::News(news) => {
                let event = NewsEvent {
                    story_id: news.story_id,
                    headline: news.headline,
                    summary: news.summary,
                    symbols: news.tickers.into_iter().map(Symbol::from).collect(),
                    category: news.category,
                    tags: news.tags,
                    impact_score: news.impact_score,
                    author: news.author,
                    source: news.source,
                    published_at: Self::parse_timestamp(&news.published_at)?,
                    timestamp: Utc::now(),
                    url: news.url,
                };

                Ok(Some(MarketDataEvent::NewsAlert(event)))
            }

            BenzingaMessage::Sentiment(sentiment) => {
                let period = match sentiment.period.as_str() {
                    "realtime" | "real_time" => SentimentPeriod::RealTime,
                    "hourly" => SentimentPeriod::Hourly,
                    "daily" => SentimentPeriod::Daily,
                    "weekly" => SentimentPeriod::Weekly,
                    _ => SentimentPeriod::RealTime,
                };

                let event = SentimentEvent {
                    symbol: Symbol::from(sentiment.ticker),
                    sentiment_score: sentiment.sentiment_score,
                    bullish_ratio: sentiment.bullish_ratio,
                    bearish_ratio: sentiment.bearish_ratio,
                    sample_size: sentiment.sample_size,
                    period,
                    sources: sentiment.sources,
                    confidence: sentiment.confidence,
                    timestamp: Self::parse_timestamp(&sentiment.timestamp)?,
                };

                Ok(Some(MarketDataEvent::SentimentUpdate(event)))
            }

            BenzingaMessage::Rating(rating) => {
                let action = match rating.action.as_str() {
                    "Upgrades" => RatingAction::Upgrade,
                    "Downgrades" => RatingAction::Downgrade,
                    "Initiates" => RatingAction::Initiate,
                    "Maintains" => RatingAction::Maintain,
                    "Discontinues" => RatingAction::Discontinue,
                    _ => RatingAction::Maintain,
                };

                let event = AnalystRatingEvent {
                    symbol: Symbol::from(rating.ticker),
                    analyst: rating.analyst,
                    firm: rating.firm,
                    action,
                    current_rating: rating.current_rating,
                    previous_rating: rating.previous_rating,
                    price_target: rating.price_target.map(Decimal::from_f64_retain).flatten(),
                    previous_price_target: rating
                        .previous_price_target
                        .map(Decimal::from_f64_retain)
                        .flatten(),
                    comment: rating.comment,
                    rating_date: Self::parse_timestamp(&rating.rating_date)?,
                    timestamp: Self::parse_timestamp(&rating.timestamp)?,
                };

                Ok(Some(MarketDataEvent::AnalystRating(event)))
            }

            BenzingaMessage::Options(options) => {
                let option_type = match options.option_type.as_str() {
                    "call" | "Call" | "CALL" => OptionsType::Call,
                    "put" | "Put" | "PUT" => OptionsType::Put,
                    _ => OptionsType::Call,
                };

                let activity_type = match options.activity_type.as_str() {
                    "block" | "Block" => UnusualOptionsType::BlockTrade,
                    "sweep" | "Sweep" => UnusualOptionsType::Sweep,
                    "volume" | "Volume" => UnusualOptionsType::VolumeSpike,
                    "oi" | "open_interest" => UnusualOptionsType::OpenInterestSpike,
                    "iv" | "volatility" => UnusualOptionsType::VolatilitySpike,
                    _ => UnusualOptionsType::BlockTrade,
                };

                let sentiment = match options.sentiment.as_str() {
                    "bullish" | "Bullish" => OptionsSentiment::Bullish,
                    "bearish" | "Bearish" => OptionsSentiment::Bearish,
                    _ => OptionsSentiment::Neutral,
                };

                let expiration = chrono::NaiveDate::parse_from_str(&options.expiration, "%Y-%m-%d")
                    .map_err(|e| DataError::parse(format!("Invalid expiration date: {}", e)))?;

                let contract = OptionsContract {
                    strike: Decimal::from_f64_retain(options.strike).unwrap_or_default(),
                    expiration,
                    option_type,
                    multiplier: 100, // Standard equity options multiplier
                };

                let event = UnusualOptionsEvent {
                    symbol: Symbol::from(options.ticker),
                    contract,
                    activity_type,
                    volume: options.volume,
                    open_interest: options.open_interest,
                    premium: options.premium.map(Decimal::from_f64_retain).flatten(),
                    implied_volatility: options.implied_volatility,
                    sentiment,
                    confidence: options.confidence,
                    description: options.description,
                    timestamp: Self::parse_timestamp(&options.timestamp)?,
                };

                Ok(Some(MarketDataEvent::UnusualOptions(event)))
            }

            BenzingaMessage::Heartbeat(_) => {
                // Update heartbeat time - this is handled in the message processing loop
                Ok(None)
            }

            BenzingaMessage::Error(error) => {
                let category = match error.code.as_str() {
                    "AUTH_ERROR" => ErrorCategory::Authentication,
                    "RATE_LIMIT" => ErrorCategory::RateLimit,
                    "PARSE_ERROR" => ErrorCategory::Parse,
                    "SUBSCRIPTION_ERROR" => ErrorCategory::Subscription,
                    _ => ErrorCategory::Other,
                };

                let error_event = ErrorEvent {
                    provider: "benzinga".to_string(),
                    message: error.message,
                    code: Some(error.code),
                    category,
                    recoverable: !matches!(category, ErrorCategory::Authentication),
                    timestamp: Utc::now(),
                };

                Ok(Some(MarketDataEvent::Error(error_event)))
            }

            BenzingaMessage::SubscriptionConfirmation(_) => {
                // Log subscription confirmation but don't emit event
                debug!("Subscription confirmed");
                Ok(None)
            }
        }
    }

    /// Parse timestamp string to DateTime<Utc>
    fn parse_timestamp(timestamp_str: &str) -> Result<DateTime<Utc>> {
        // Try multiple timestamp formats
        let formats = [
            "%Y-%m-%dT%H:%M:%SZ",
            "%Y-%m-%dT%H:%M:%S%.fZ",
            "%Y-%m-%dT%H:%M:%S%z",
            "%Y-%m-%dT%H:%M:%S%.f%z",
            "%Y-%m-%d %H:%M:%S",
            "%Y-%m-%d %H:%M:%S%.f",
        ];

        for format in &formats {
            if let Ok(dt) = DateTime::parse_from_str(timestamp_str, format) {
                return Ok(dt.with_timezone(&Utc));
            }
        }

        // Try parsing as naive datetime and assume UTC
        for format in &["%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S%.f"] {
            if let Ok(naive_dt) = chrono::NaiveDateTime::parse_from_str(timestamp_str, format) {
                return Ok(DateTime::from_naive_utc_and_offset(naive_dt, Utc));
            }
        }

        Err(DataError::parse(format!(
            "Unable to parse timestamp: {}",
            timestamp_str
        )))
    }

    /// Handle reconnection with exponential backoff
    async fn reconnect(&self) -> Result<()> {
        let mut attempt = {
            let mut guard = self.reconnect_attempt.lock().await;
            *guard += 1;
            *guard
        };

        if attempt > self.config.max_reconnect_attempts {
            error!(
                "Maximum reconnection attempts ({}) exceeded",
                self.config.max_reconnect_attempts
            );

            // Update connection status to failed
            {
                let mut status = self.connection_status.write().await;
                status.state = TraitConnectionState::Failed;
            }

            return Err(DataError::Connection(
                "Max reconnection attempts exceeded".to_string(),
            ));
        }

        info!("Attempting reconnection #{}", attempt);

        // Update connection status to reconnecting
        {
            let mut status = self.connection_status.write().await;
            status.state = TraitConnectionState::Reconnecting;
            status.last_connection_attempt = Some(Utc::now());
        }

        // Calculate backoff delay
        let delay_ms = self.config.initial_reconnect_delay_ms as f64
            * self
                .config
                .reconnect_backoff_multiplier
                .powi((attempt - 1) as i32);
        let delay_ms = delay_ms.min(self.config.max_reconnect_delay_ms as f64) as u64;

        info!("Waiting {}ms before reconnection", delay_ms);
        tokio::time::sleep(Duration::from_millis(delay_ms)).await;

        // Attempt to reconnect
        match self.connect_websocket().await {
            Ok(()) => {
                info!("Reconnection successful");

                // Re-subscribe to existing symbols
                let symbols: Vec<Symbol> = {
                    let subscribed = self.subscribed_symbols.read().await;
                    subscribed.iter().cloned().collect()
                };

                if !symbols.is_empty() {
                    if let Err(e) = self.send_subscription(symbols).await {
                        error!("Failed to re-subscribe after reconnection: {}", e);
                    }
                }

                // Restart message loop
                self.start_message_loop().await?;

                Ok(())
            }
            Err(e) => {
                error!("Reconnection failed: {}", e);

                // Schedule another reconnection attempt (without tokio::spawn to avoid Send issues)
                let provider = self.clone();
                tokio::task::spawn_local(async move {
                    let _ = provider.reconnect().await;
                });

                Err(e)
            }
        }
    }
}

// Implement Clone for BenzingaStreamingProvider (for reconnection spawning)
impl Clone for BenzingaStreamingProvider {
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            connection_status: self.connection_status.clone(),
            websocket: self.websocket.clone(),
            event_tx: self.event_tx.clone(),
            event_rx: self.event_rx.clone(),
            subscribed_symbols: self.subscribed_symbols.clone(),
            reconnect_attempt: self.reconnect_attempt.clone(),
            last_heartbeat: self.last_heartbeat.clone(),
            metrics: self.metrics.clone(),
            shutdown_tx: self.shutdown_tx.clone(),
        }
    }
}

#[async_trait]
impl RealTimeProvider for BenzingaStreamingProvider {
    async fn connect(&mut self) -> Result<()> {
        info!("Connecting to Benzinga streaming API");

        // Update connection status to connecting
        {
            let mut status = self.connection_status.write().await;
            status.state = TraitConnectionState::Connecting;
            status.last_connection_attempt = Some(Utc::now());
        }

        // Establish WebSocket connection
        self.connect_websocket().await?;

        // Start message processing loop
        self.start_message_loop().await?;

        // Send connection status event
        if let Some(tx) = self.event_tx.lock().await.as_ref() {
            let status_event = MarketDataEvent::ConnectionStatus(ConnectionEvent {
                provider: "benzinga".to_string(),
                status: ConnectionStatus::connected(),
                message: Some("Connected to Benzinga streaming API".to_string()),
                timestamp: Utc::now(),
            });

            let _ = tx.send(status_event);
        }

        info!("Successfully connected to Benzinga streaming API");
        Ok(())
    }

    async fn disconnect(&mut self) -> Result<()> {
        info!("Disconnecting from Benzinga streaming API");

        // Send shutdown signal
        if let Some(tx) = self.shutdown_tx.lock().await.take() {
            let _ = tx.send(());
        }

        // Close WebSocket connection
        {
            let mut websocket = self.websocket.lock().await;
            if let Some(mut ws) = websocket.take() {
                let _ = ws.close(None).await;
            }
        }

        // Update connection status
        {
            let mut status = self.connection_status.write().await;
            status.state = TraitConnectionState::Disconnected;
        }

        // Clear subscriptions
        {
            let mut subscriptions = self.subscribed_symbols.write().await;
            subscriptions.clear();
        }

        // Send connection status event
        if let Some(tx) = self.event_tx.lock().await.as_ref() {
            let status_event = MarketDataEvent::ConnectionStatus(ConnectionEvent {
                provider: "benzinga".to_string(),
                status: ConnectionStatus::disconnected(),
                message: Some("Disconnected from Benzinga streaming API".to_string()),
                timestamp: Utc::now(),
            });

            let _ = tx.send(status_event);
        }

        info!("Successfully disconnected from Benzinga streaming API");
        Ok(())
    }

    async fn subscribe(&mut self, symbols: Vec<Symbol>) -> Result<()> {
        if symbols.is_empty() {
            return Ok(());
        }

        info!("Subscribing to symbols: {:?}", symbols);

        // Send subscription request
        self.send_subscription(symbols.clone()).await?;

        // Update subscribed symbols
        {
            let mut subscriptions = self.subscribed_symbols.write().await;
            for symbol in &symbols {
                subscriptions.insert(symbol.clone());
            }
        }

        // Update connection status
        {
            let mut status = self.connection_status.write().await;
            status.active_subscriptions = {
                let subscriptions = self.subscribed_symbols.read().await;
                subscriptions.len()
            };
        }

        info!("Successfully subscribed to {} symbols", symbols.len());
        Ok(())
    }

    async fn unsubscribe(&mut self, symbols: Vec<Symbol>) -> Result<()> {
        if symbols.is_empty() {
            return Ok(());
        }

        info!("Unsubscribing from symbols: {:?}", symbols);

        // Remove from subscribed symbols
        {
            let mut subscriptions = self.subscribed_symbols.write().await;
            for symbol in &symbols {
                subscriptions.remove(symbol);
            }
        }

        // Send unsubscription request (similar to subscription but with "unsubscribe" type)
        let unsubscription = SubscriptionRequest {
            request_type: "unsubscribe".to_string(),
            api_key: self.config.api_key.clone(),
            tickers: symbols.iter().map(|s| s.to_string()).collect(),
            events: vec![
                "news".to_string(),
                "sentiment".to_string(),
                "ratings".to_string(),
                "options".to_string(),
            ],
        };

        let message =
            serde_json::to_string(&unsubscription).map_err(|e| DataError::Serialization {
                message: format!("Failed to serialize unsubscription: {}", e),
            })?;

        // Send unsubscription message
        {
            let mut websocket_guard = self.websocket.lock().await;
            if let Some(websocket) = websocket_guard.as_mut() {
                websocket
                    .send(Message::Text(message))
                    .await
                    .map_err(|e| DataError::WebSocket(e))?;

                debug!("Sent unsubscription for symbols: {:?}", symbols);
            }
        }

        // Update connection status
        {
            let mut status = self.connection_status.write().await;
            status.active_subscriptions = {
                let subscriptions = self.subscribed_symbols.read().await;
                subscriptions.len()
            };
        }

        info!("Successfully unsubscribed from {} symbols", symbols.len());
        Ok(())
    }

    async fn stream(&mut self) -> Result<Pin<Box<dyn Stream<Item = MarketDataEvent> + Send>>> {
        // Take the receiver from the arc mutex
        let receiver = {
            let mut rx_guard = self.event_rx.lock().await;
            rx_guard.take()
        };

        match receiver {
            Some(rx) => {
                let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
                Ok(Box::pin(stream))
            }
            None => Err(DataError::internal(
                "Event receiver already taken or not initialized",
            )),
        }
    }

    fn get_connection_status(&self) -> ConnectionStatus {
        // This is a blocking operation but should be fast
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async {
                let status = self.connection_status.read().await;
                let metrics = self.metrics.read().await;

                ConnectionStatus {
                    state: status.state,
                    active_subscriptions: status.active_subscriptions,
                    events_per_second: metrics.messages_per_second,
                    latency_micros: None, // Not measurable for WebSocket
                    recent_error_count: metrics.error_count,
                    last_message_time: metrics.last_message_time,
                    last_connection_attempt: status.last_connection_attempt,
                }
            })
        })
    }

    fn get_provider_name(&self) -> &'static str {
        "benzinga"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio_test;

    #[test]
    fn test_config_creation() {
        let config = BenzingaStreamingConfig::default();
        assert!(!config.websocket_url.is_empty());
        assert!(config.connect_timeout_secs > 0);
        assert!(config.event_buffer_size > 0);
    }

    #[test]
    fn test_provider_creation() {
        let config = BenzingaStreamingConfig {
            api_key: "test-key".to_string(),
            ..Default::default()
        };

        let provider = BenzingaStreamingProvider::new(config);
        assert!(provider.is_ok());
    }

    #[test]
    fn test_provider_creation_without_api_key() {
        let config = BenzingaStreamingConfig {
            api_key: "".to_string(),
            ..Default::default()
        };

        let provider = BenzingaStreamingProvider::new(config);
        assert!(provider.is_err());
    }

    #[test]
    fn test_timestamp_parsing() {
        let timestamps = [
            "2024-01-15T10:30:00Z",
            "2024-01-15T10:30:00.123Z",
            "2024-01-15T10:30:00+00:00",
            "2024-01-15T10:30:00.123+00:00",
            "2024-01-15 10:30:00",
            "2024-01-15 10:30:00.123",
        ];

        for timestamp_str in &timestamps {
            let result = BenzingaStreamingProvider::parse_timestamp(timestamp_str);
            assert!(
                result.is_ok(),
                "Failed to parse timestamp: {}",
                timestamp_str
            );
        }
    }

    #[test]
    fn test_subscription_request_serialization() {
        let request = SubscriptionRequest {
            request_type: "subscribe".to_string(),
            api_key: "test-key".to_string(),
            tickers: vec!["AAPL".to_string(), "SPY".to_string()],
            events: vec!["news".to_string(), "sentiment".to_string()],
        };

        let json = serde_json::to_string(&request);
        assert!(json.is_ok());

        let json_str = json.unwrap();
        assert!(json_str.contains("subscribe"));
        assert!(json_str.contains("AAPL"));
        assert!(json_str.contains("news"));
    }

    #[tokio::test]
    async fn test_connection_status_tracking() {
        let config = BenzingaStreamingConfig {
            api_key: "test-key".to_string(),
            ..Default::default()
        };

        let provider = BenzingaStreamingProvider::new(config).unwrap();

        // Initial status should be disconnected
        let status = provider.get_connection_status();
        assert_eq!(status.state, TraitConnectionState::Disconnected);
        assert_eq!(status.active_subscriptions, 0);
    }

    #[test]
    fn test_benzinga_message_deserialization() {
        let news_json = r#"
        {
            "type": "news",
            "story_id": "12345",
            "headline": "Test Headline",
            "summary": "Test summary",
            "tickers": ["AAPL"],
            "category": "earnings",
            "tags": ["tech"],
            "impact_score": 0.75,
            "author": "Test Author",
            "source": "Benzinga",
            "published_at": "2024-01-15T10:30:00Z",
            "url": "https://example.com"
        }
        "#;

        let result: Result<BenzingaMessage, _> = serde_json::from_str(news_json);
        assert!(result.is_ok());

        if let Ok(BenzingaMessage::News(news)) = result {
            assert_eq!(news.story_id, "12345");
            assert_eq!(news.headline, "Test Headline");
            assert_eq!(news.tickers, vec!["AAPL"]);
        } else {
            panic!("Expected news message");
        }
    }
}