//! # Production Benzinga Historical Data Provider
//!
//! This module provides comprehensive access to Benzinga's historical news and events data
//! through their REST API with production-grade features:
//! - Comprehensive API coverage (news, earnings, ratings, options, calendar events)
//! - Advanced caching with Redis integration
//! - Rate limiting and retry logic
//! - Historical data replay for backtesting
//! - Bulk data download capabilities
//! - Data quality validation and filtering

use crate::error::{DataError, Result};
use crate::providers::common::{
    AnalystRatingEvent, NewsEvent, OptionsContract, OptionsSentiment, OptionsType, RatingAction,
    UnusualOptionsEvent, UnusualOptionsType,
};
use crate::providers::traits::{HistoricalProvider, HistoricalSchema};
use crate::types::TimeRange;
use crate::types::{get_event_timestamp, ExtendedMarketDataEvent};
use chrono::{DateTime, NaiveDate, Utc};
use common::{Price, Quantity, Symbol};
use governor::{
    state::{InMemoryState, NotKeyed},
    Quota, RateLimiter,
};
#[cfg(feature = "redis-cache")]
use redis::{AsyncCommands, Client as RedisClient};
use reqwest::{Client, Response, StatusCode};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::num::NonZeroU32;
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc,
};
use std::time::{Duration, Instant};
use tokio::sync::{RwLock, Semaphore};
use tracing::{debug, info, instrument, warn};

use async_trait::async_trait;
use common::MarketDataEvent;

/// Production Benzinga historical provider configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProductionBenzingaHistoricalConfig {
    /// Benzinga Pro API key
    pub api_key: String,

    /// API endpoint base URL
    pub endpoint: String,

    /// Request timeout in seconds
    pub timeout_seconds: u64,

    /// Rate limiter quota per second
    pub rate_limit_per_second: u32,

    /// Maximum retry attempts
    pub max_retry_attempts: u32,

    /// Initial retry delay in milliseconds
    pub initial_retry_delay_ms: u64,

    /// Retry backoff multiplier
    pub retry_backoff_multiplier: f64,

    /// Maximum concurrent requests
    pub max_concurrent_requests: usize,

    /// Enable `Redis` caching
    pub enable_caching: bool,

    /// `Redis` connection string
    pub redis_url: Option<String>,

    /// Cache TTL in seconds
    pub cache_ttl_secs: u64,

    /// Enable data validation
    pub enable_data_validation: bool,

    /// Minimum importance score filter
    pub min_importance_score: f64,

    /// Maximum data age for filtering (hours)
    pub max_data_age_hours: u64,

    /// Enable bulk download mode
    pub enable_bulk_download: bool,

    /// Bulk download batch size
    pub bulk_batch_size: usize,

    /// Enable compression for large responses
    pub enable_compression: bool,
}

impl Default for ProductionBenzingaHistoricalConfig {
    fn default() -> Self {
        Self {
            api_key: std::env::var("BENZINGA_API_KEY").unwrap_or_default(),
            endpoint: "https://api.benzinga.com/api/v2".to_string(),
            timeout_seconds: 30,
            rate_limit_per_second: 10, // Conservative for historical data
            max_retry_attempts: 3,
            initial_retry_delay_ms: 1000,
            retry_backoff_multiplier: 2.0,
            max_concurrent_requests: 5,
            enable_caching: true,
            redis_url: std::env::var("REDIS_URL").ok(),
            cache_ttl_secs: 3600, // 1 hour for historical data
            enable_data_validation: true,
            min_importance_score: 0.0,
            max_data_age_hours: 24 * 7, // 1 week
            enable_bulk_download: true,
            bulk_batch_size: 1000,
            enable_compression: true,
        }
    }
}

/// Historical data metrics
#[derive(Debug, Default)]
pub struct HistoricalMetrics {
    /// Total requests made
    pub requests_made: AtomicU64,

    /// Successful requests
    pub successful_requests: AtomicU64,

    /// Failed requests
    pub failed_requests: AtomicU64,

    /// Cache hits
    pub cache_hits: AtomicU64,

    /// Cache misses
    pub cache_misses: AtomicU64,

    /// Data points retrieved
    pub data_points_retrieved: AtomicU64,

    /// Data points filtered out
    pub data_points_filtered: AtomicU64,

    /// Average response time (milliseconds)
    pub avg_response_time_ms: AtomicU64,

    /// Total data volume (bytes)
    pub total_data_volume_bytes: AtomicU64,
}

/// Benzinga API response structures
#[derive(Debug, Deserialize)]
struct BenzingaNewsResponse {
    #[serde(default)]
    data: Vec<BenzingaNewsItem>,
    #[serde(default)]
    total: Option<u32>,
    #[serde(default)]
    next_page: Option<String>,
}

#[derive(Debug, Deserialize)]
struct BenzingaNewsItem {
    pub id: String,
    pub title: String,
    pub body: Option<String>,
    pub tickers: Vec<String>,
    pub channels: Vec<BenzingaChannel>,
    pub tags: Vec<BenzingaTag>,
    pub author: Option<String>,
    pub created: String,
    pub updated: String,
    pub url: Option<String>,
    pub importance: Option<f64>,
    #[serde(default)]
    pub sentiment: Option<String>,
}

#[derive(Debug, Deserialize)]
struct BenzingaChannel {
    pub name: String,
}

#[derive(Debug, Deserialize)]
struct BenzingaTag {
    pub name: String,
}

#[derive(Debug, Deserialize)]
struct BenzingaRatingsResponse {
    #[serde(default)]
    data: Vec<BenzingaRatingItem>,
    #[serde(default)]
    total: Option<u32>,
    #[serde(default)]
    next_page: Option<String>,
}

#[derive(Debug, Deserialize)]
struct BenzingaRatingItem {
    pub id: String,
    pub ticker: String,
    pub analyst_name: Option<String>,
    pub firm_name: Option<String>,
    pub action_pt: Option<String>,
    pub action_company: Option<String>,
    pub rating_current: Option<String>,
    pub rating_prior: Option<String>,
    pub pt_current: Option<f64>,
    pub pt_prior: Option<f64>,
    pub date: String,
    pub time: Option<String>,
    pub importance: Option<f64>,
}

#[derive(Debug, Deserialize)]
struct BenzingaEarningsResponse {
    #[serde(default)]
    data: Vec<BenzingaEarningsItem>,
    #[serde(default)]
    total: Option<u32>,
}

#[derive(Debug, Deserialize)]
struct BenzingaEarningsItem {
    pub id: String,
    pub ticker: String,
    pub name: Option<String>,
    pub date: String,
    pub time: Option<String>,
    pub period: Option<String>,
    pub period_year: Option<u32>,
    pub eps_est: Option<f64>,
    pub eps_actual: Option<f64>,
    pub eps_prior: Option<f64>,
    pub revenue_est: Option<f64>,
    pub revenue_actual: Option<f64>,
    pub revenue_prior: Option<f64>,
    pub importance: Option<f64>,
}

#[derive(Debug, Deserialize)]
struct BenzingaOptionsResponse {
    #[serde(default)]
    data: Vec<BenzingaOptionsItem>,
    #[serde(default)]
    total: Option<u32>,
}

#[derive(Debug, Deserialize)]
struct BenzingaOptionsItem {
    pub ticker: String,
    pub strike: f64,
    pub expiry: String,
    pub option_type: String,
    pub volume: u32,
    pub open_interest: Option<u32>,
    pub cost_basis: Option<f64>,
    pub trade_count: Option<u32>,
    pub sentiment: Option<String>,
    pub activity_type: Option<String>,
    pub date_expiry: String,
    pub updated: String,
}

#[derive(Debug, Deserialize)]
struct BenzingaCalendarResponse {
    #[serde(default)]
    data: Vec<BenzingaCalendarItem>,
}

#[derive(Debug, Deserialize)]
struct BenzingaCalendarItem {
    pub id: String,
    pub ticker: Option<String>,
    pub name: String,
    pub date: String,
    pub time: Option<String>,
    pub exchange: Option<String>,
    pub country: Option<String>,
    pub importance: Option<f64>,
    pub currency: Option<String>,
    pub description: Option<String>,
    pub period: Option<String>,
    pub actual: Option<String>,
    pub consensus: Option<String>,
    pub prior: Option<String>,
}

/// Production Benzinga Historical Data Provider
pub struct ProductionBenzingaHistoricalProvider {
    /// Provider configuration
    config: ProductionBenzingaHistoricalConfig,

    /// HTTP client
    client: Client,

    /// Rate limiter
    rate_limiter: Arc<RateLimiter<NotKeyed, InMemoryState, governor::clock::DefaultClock>>,

    /// Concurrency semaphore
    semaphore: Arc<Semaphore>,

    /// `Redis` client for caching
    #[cfg(feature = "redis-cache")]
    redis_client: Option<RedisClient>,

    /// Metrics
    metrics: Arc<HistoricalMetrics>,

    /// Response cache
    cache: Arc<RwLock<HashMap<String, (DateTime<Utc>, Vec<u8>)>>>,
}

impl ProductionBenzingaHistoricalProvider {
    /// Create a new production Benzinga historical provider
    pub fn new(config: ProductionBenzingaHistoricalConfig) -> 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 mut client_builder = Client::builder()
            .timeout(Duration::from_secs(config.timeout_seconds))
            .user_agent("Foxhunt/1.0");

        if config.enable_compression {
            client_builder = client_builder.gzip(true);
        }

        let client = client_builder.build().map_err(|e| DataError::Network {
            message: format!("Failed to create HTTP client: {}", e),
        })?;

        // Create rate limiter
        let quota = Quota::per_second(
            NonZeroU32::new(config.rate_limit_per_second).ok_or_else(|| {
                DataError::configuration("rate_limit_per_second", "must be > 0")
            })?,
        );
        let rate_limiter = Arc::new(RateLimiter::direct(quota));

        // Create semaphore
        let semaphore = Arc::new(Semaphore::new(config.max_concurrent_requests));

        // Create Redis client if configured
        #[cfg(feature = "redis-cache")]
        let redis_client = if config.enable_caching {
            if let Some(redis_url) = &config.redis_url {
                match RedisClient::open(redis_url.as_str()) {
                    Ok(client) => {
                        info!("Redis client created successfully");
                        Some(client)
                    },
                    Err(e) => {
                        warn!(
                            "Failed to create Redis client: {}, falling back to in-memory cache",
                            e
                        );
                        None
                    },
                }
            } else {
                warn!("Caching enabled but no Redis URL provided, using in-memory cache");
                None
            }
        } else {
            None
        };

        Ok(Self {
            config,
            client,
            rate_limiter,
            semaphore,
            #[cfg(feature = "redis-cache")]
            redis_client,
            metrics: Arc::new(HistoricalMetrics::default()),
            cache: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Make a rate-limited HTTP request with retries
    #[instrument(skip(self))]
    async fn make_request(&self, url: &str, params: &[(String, String)]) -> Result<Response> {
        let _permit = self
            .semaphore
            .acquire()
            .await
            .map_err(|e| DataError::Network {
                message: format!("Semaphore closed: {}", e),
            })?;

        let start_time = Instant::now();
        let mut last_error = None;

        for attempt in 0..=self.config.max_retry_attempts {
            // Wait for rate limiter
            self.rate_limiter.until_ready().await;

            self.metrics.requests_made.fetch_add(1, Ordering::Relaxed);

            let response = self
                .client
                .get(url)
                .query(params)
                .query(&[("token", &self.config.api_key)])
                .send()
                .await;

            match response {
                Ok(resp) => {
                    let status = resp.status();

                    if status.is_success() {
                        self.metrics
                            .successful_requests
                            .fetch_add(1, Ordering::Relaxed);
                        let elapsed = start_time.elapsed();
                        self.metrics
                            .avg_response_time_ms
                            .store(elapsed.as_millis() as u64, Ordering::Relaxed);
                        return Ok(resp);
                    } else if status == StatusCode::TOO_MANY_REQUESTS {
                        warn!("Rate limited by Benzinga API, waiting before retry");
                        tokio::time::sleep(Duration::from_secs(60)).await;
                    } else if status.is_server_error() && attempt < self.config.max_retry_attempts {
                        warn!(
                            "Server error {}, retrying in {}ms",
                            status,
                            self.config.initial_retry_delay_ms * (attempt as u64 + 1)
                        );
                    } else {
                        self.metrics.failed_requests.fetch_add(1, Ordering::Relaxed);
                        return Err(DataError::Network {
                            message: format!("HTTP error: {}", status),
                        });
                    }
                },
                Err(e) => {
                    last_error = Some(e);
                    if attempt < self.config.max_retry_attempts {
                        let delay = Duration::from_millis(
                            self.config.initial_retry_delay_ms
                                * (self.config.retry_backoff_multiplier.powi(attempt as i32)
                                    as u64),
                        );
                        warn!("Request failed, retrying in {:?}: {:?}", delay, last_error);
                        tokio::time::sleep(delay).await;
                    }
                },
            }
        }

        self.metrics.failed_requests.fetch_add(1, Ordering::Relaxed);
        Err(DataError::Network {
            message: format!(
                "Request failed after {} retries: {:?}",
                self.config.max_retry_attempts, last_error
            ),
        })
    }

    /// Get cached data or fetch from API
    async fn get_cached_or_fetch<T: for<'de> Deserialize<'de>>(
        &self,
        cache_key: &str,
        url: &str,
        params: &[(String, String)],
    ) -> Result<T> {
        // Try cache first
        if self.config.enable_caching {
            if let Some(cached_data) = self.get_from_cache(cache_key).await? {
                self.metrics.cache_hits.fetch_add(1, Ordering::Relaxed);
                return Ok(cached_data);
            }
        }

        self.metrics.cache_misses.fetch_add(1, Ordering::Relaxed);

        // Fetch from API
        let response = self.make_request(url, params).await?;
        let data_bytes = response.bytes().await.map_err(|e| DataError::Network {
            message: format!("Failed to read response: {}", e),
        })?;

        self.metrics
            .total_data_volume_bytes
            .fetch_add(data_bytes.len() as u64, Ordering::Relaxed);

        let data: T =
            serde_json::from_slice(&data_bytes).map_err(|e| DataError::Serialization {
                message: format!("Failed to parse JSON: {}", e),
            })?;

        // Cache the result
        if self.config.enable_caching {
            self.set_cache(cache_key, &data_bytes).await?;
        }

        Ok(data)
    }

    /// Get data from cache
    async fn get_from_cache<T: for<'de> Deserialize<'de>>(&self, key: &str) -> Result<Option<T>> {
        // Try Redis first
        #[cfg(feature = "redis-cache")]
        if let Some(redis_client) = &self.redis_client {
            if let Ok(mut conn) = redis_client.get_multiplexed_async_connection().await {
                if let Ok(data_bytes) = conn.get::<_, Vec<u8>>(key).await {
                    if let Ok(data) = serde_json::from_slice(&data_bytes) {
                        return Ok(Some(data));
                    }
                }
            }
        }

        // Fall back to in-memory cache
        let cache = self.cache.read().await;
        if let Some((timestamp, data_bytes)) = cache.get(key) {
            let age = Utc::now() - *timestamp;
            if age.num_seconds() < self.config.cache_ttl_secs as i64 {
                if let Ok(data) = serde_json::from_slice(data_bytes) {
                    return Ok(Some(data));
                }
            }
        }

        Ok(None)
    }

    /// Set data in cache
    async fn set_cache(&self, key: &str, data: &[u8]) -> Result<()> {
        // Try Redis first
        #[cfg(feature = "redis-cache")]
        if let Some(redis_client) = &self.redis_client {
            if let Ok(mut conn) = redis_client.get_multiplexed_async_connection().await {
                let _: std::result::Result<(), _> =
                    conn.set_ex(key, data, self.config.cache_ttl_secs).await;
            }
        }

        // Also set in memory cache as backup
        let mut cache = self.cache.write().await;

        // Limit in-memory cache size
        if cache.len() > 10000 {
            // Remove oldest entries
            let mut entries: Vec<_> = cache.iter().map(|(k, v)| (k.clone(), v.0)).collect();
            entries.sort_by(|a, b| a.1.cmp(&b.1));

            let keys_to_remove: Vec<String> =
                entries.iter().take(1000).map(|(k, _)| k.clone()).collect();
            for key in keys_to_remove {
                cache.remove(&key);
            }
        }

        cache.insert(key.to_string(), (Utc::now(), data.to_vec()));
        Ok(())
    }

    /// Validate and filter data based on configuration
    fn validate_and_filter_news(&self, news: &BenzingaNewsItem) -> bool {
        if !self.config.enable_data_validation {
            return true;
        }

        // Check importance score
        if let Some(importance) = news.importance {
            if importance < self.config.min_importance_score {
                return false;
            }
        }

        // Check data age
        if let Ok(created_dt) = DateTime::parse_from_rfc3339(&news.created) {
            let age = Utc::now() - created_dt.with_timezone(&Utc);
            if age.num_hours() > self.config.max_data_age_hours as i64 {
                return false;
            }
        }

        true
    }

    /// Get news events with comprehensive filtering and validation
    #[instrument(skip(self))]
    pub async fn get_news_events(
        &self,
        symbols: Option<&[&str]>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<NewsEvent>> {
        let mut params = vec![
            ("dateFrom".to_string(), start.format("%Y-%m-%d").to_string()),
            ("dateTo".to_string(), end.format("%Y-%m-%d").to_string()),
            ("pageSize".to_string(), "1000".to_string()),
        ];

        if let Some(symbols) = symbols {
            params.push(("tickers".to_string(), symbols.join(",")));
        }

        let cache_key = format!(
            "benzinga:news:{}:{}:{}",
            symbols.map(|s| s.join(",")).unwrap_or_default(),
            start.format("%Y%m%d"),
            end.format("%Y%m%d")
        );

        let url = format!("{}/news", self.config.endpoint);
        let response: BenzingaNewsResponse =
            self.get_cached_or_fetch(&cache_key, &url, &params).await?;

        let mut events = Vec::new();
        for item in response.data {
            if !self.validate_and_filter_news(&item) {
                self.metrics
                    .data_points_filtered
                    .fetch_add(1, Ordering::Relaxed);
                continue;
            }

            let event = NewsEvent {
                story_id: item.id,
                headline: item.title.clone(),
                content: item.body.clone().unwrap_or_default(),
                summary: item.body.unwrap_or_default(),
                symbol: item.tickers.first().map(|t| Symbol::from(t.clone())),
                symbols: item.tickers.into_iter().map(Symbol::from).collect(),
                category: item
                    .channels
                    .first()
                    .map(|c| c.name.clone())
                    .unwrap_or_else(|| "general".to_string()),
                tags: item.tags.into_iter().map(|t| t.name).collect(),
                impact_score: item.importance,
                importance: item.importance.unwrap_or(0.5),
                author: item.author.unwrap_or_else(|| "Unknown".to_string()),
                source: "Benzinga".to_string(),
                published_at: DateTime::parse_from_rfc3339(&item.created)
                    .map(|dt| dt.with_timezone(&Utc))
                    .unwrap_or_else(|_| Utc::now()),
                timestamp: Utc::now(),
                url: item.url.unwrap_or_default(),
                sentiment_score: item.sentiment.as_deref().and_then(|s| match s {
                    "positive" => Some(0.5),
                    "negative" => Some(-0.5),
                    _ => None,
                }),
                #[allow(deprecated)]
                sentiment: None,
                event_type: crate::providers::common::NewsEventType::News,
            };

            events.push(event);
            self.metrics
                .data_points_retrieved
                .fetch_add(1, Ordering::Relaxed);
        }

        info!("Retrieved {} news events from Benzinga", events.len());
        Ok(events)
    }

    /// Get analyst rating events
    #[instrument(skip(self))]
    pub async fn get_rating_events(
        &self,
        symbols: Option<&[&str]>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<AnalystRatingEvent>> {
        let mut params = vec![
            ("dateFrom".to_string(), start.format("%Y-%m-%d").to_string()),
            ("dateTo".to_string(), end.format("%Y-%m-%d").to_string()),
            ("pageSize".to_string(), "1000".to_string()),
        ];

        if let Some(symbols) = symbols {
            params.push(("tickers".to_string(), symbols.join(",")));
        }

        let cache_key = format!(
            "benzinga:ratings:{}:{}:{}",
            symbols.map(|s| s.join(",")).unwrap_or_default(),
            start.format("%Y%m%d"),
            end.format("%Y%m%d")
        );

        let url = format!("{}/calendar/ratings", self.config.endpoint);
        let response: BenzingaRatingsResponse =
            self.get_cached_or_fetch(&cache_key, &url, &params).await?;

        let mut events = Vec::new();
        for item in response.data {
            let action = match item.action_company.as_deref() {
                Some("Upgrades") => RatingAction::Upgrade,
                Some("Downgrades") => RatingAction::Downgrade,
                Some("Initiates") => RatingAction::Initiate,
                Some("Maintains") => RatingAction::Maintain,
                Some("Discontinues") => RatingAction::Discontinue,
                _ => RatingAction::Maintain,
            };

            let rating_date = DateTime::parse_from_str(&item.date, "%Y-%m-%d")
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or_else(|_| Utc::now());

            let event = AnalystRatingEvent {
                symbol: Symbol::from(item.ticker),
                analyst: item.analyst_name.unwrap_or_else(|| "Unknown".to_string()),
                firm: item.firm_name.unwrap_or_else(|| "Unknown".to_string()),
                action,
                rating: item
                    .rating_current
                    .clone()
                    .unwrap_or_else(|| "N/A".to_string()),
                current_rating: item.rating_current.unwrap_or_else(|| "N/A".to_string()),
                previous_rating: item.rating_prior.unwrap_or_else(|| "N/A".to_string()),
                price_target: item
                    .pt_current
                    .map(|p| Price::from_decimal(Decimal::from_f64_retain(p).unwrap_or_default())),
                previous_price_target: item
                    .pt_prior
                    .map(|p| Price::from_decimal(Decimal::from_f64_retain(p).unwrap_or_default())),
                comment: None,
                rating_date,
                timestamp: Utc::now(),
            };

            events.push(event);
            self.metrics
                .data_points_retrieved
                .fetch_add(1, Ordering::Relaxed);
        }

        info!("Retrieved {} rating events from Benzinga", events.len());
        Ok(events)
    }

    /// Get earnings events
    #[instrument(skip(self))]
    pub async fn get_earnings_events(
        &self,
        symbols: Option<&[&str]>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<NewsEvent>> {
        let mut params = vec![
            ("dateFrom".to_string(), start.format("%Y-%m-%d").to_string()),
            ("dateTo".to_string(), end.format("%Y-%m-%d").to_string()),
            ("pageSize".to_string(), "1000".to_string()),
        ];

        if let Some(symbols) = symbols {
            params.push(("tickers".to_string(), symbols.join(",")));
        }

        let cache_key = format!(
            "benzinga:earnings:{}:{}:{}",
            symbols.map(|s| s.join(",")).unwrap_or_default(),
            start.format("%Y%m%d"),
            end.format("%Y%m%d")
        );

        let url = format!("{}/calendar/earnings", self.config.endpoint);
        let response: BenzingaEarningsResponse =
            self.get_cached_or_fetch(&cache_key, &url, &params).await?;

        let mut events = Vec::new();
        for item in response.data {
            let earning_date = DateTime::parse_from_str(&item.date, "%Y-%m-%d")
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or_else(|_| Utc::now());

            // Create earnings summary
            let mut summary_parts = Vec::new();
            if let Some(eps_actual) = item.eps_actual {
                if let Some(eps_est) = item.eps_est {
                    let beat_miss = if eps_actual > eps_est {
                        "beat"
                    } else {
                        "missed"
                    };
                    summary_parts.push(format!(
                        "EPS {} estimates: actual ${:.2} vs est ${:.2}",
                        beat_miss, eps_actual, eps_est
                    ));
                }
            }
            if let Some(rev_actual) = item.revenue_actual {
                if let Some(rev_est) = item.revenue_est {
                    let beat_miss = if rev_actual > rev_est {
                        "beat"
                    } else {
                        "missed"
                    };
                    summary_parts.push(format!(
                        "Revenue {} estimates: actual ${:.2}M vs est ${:.2}M",
                        beat_miss,
                        rev_actual / 1_000_000.0,
                        rev_est / 1_000_000.0
                    ));
                }
            }

            let headline = format!(
                "{} Q{} {} Earnings",
                item.name.as_deref().unwrap_or(&item.ticker),
                item.period.as_deref().unwrap_or("?"),
                item.period_year.unwrap_or(2024)
            );
            let event = NewsEvent {
                story_id: item.id,
                headline: headline.clone(),
                content: headline.clone(),
                summary: if summary_parts.is_empty() {
                    "No details available".to_string()
                } else {
                    summary_parts.join("; ")
                },
                symbol: Some(Symbol::from(item.ticker.clone())),
                symbols: vec![Symbol::from(item.ticker)],
                category: "earnings".to_string(),
                tags: vec!["earnings".to_string(), "financial_results".to_string()],
                impact_score: item.importance,
                importance: item.importance.unwrap_or(0.8), // Earnings typically important
                author: "Benzinga".to_string(),
                source: "Benzinga".to_string(),
                published_at: earning_date,
                timestamp: Utc::now(),
                url: "".to_string(),
                sentiment_score: None, // Earnings are typically neutral until analyzed
                #[allow(deprecated)]
                sentiment: None,
                event_type: crate::providers::common::NewsEventType::Earnings,
            };

            events.push(event);
            self.metrics
                .data_points_retrieved
                .fetch_add(1, Ordering::Relaxed);
        }

        info!("Retrieved {} earnings events from Benzinga", events.len());
        Ok(events)
    }

    /// Get unusual options activity
    #[instrument(skip(self))]
    #[allow(deprecated)] // Needed for backward compatibility with deprecated fields
    pub async fn get_options_events(
        &self,
        symbols: Option<&[&str]>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<UnusualOptionsEvent>> {
        let mut params = vec![
            ("dateFrom".to_string(), start.format("%Y-%m-%d").to_string()),
            ("dateTo".to_string(), end.format("%Y-%m-%d").to_string()),
            ("pageSize".to_string(), "1000".to_string()),
        ];

        if let Some(symbols) = symbols {
            params.push(("tickers".to_string(), symbols.join(",")));
        }

        let cache_key = format!(
            "benzinga:options:{}:{}:{}",
            symbols.map(|s| s.join(",")).unwrap_or_default(),
            start.format("%Y%m%d"),
            end.format("%Y%m%d")
        );

        let url = format!("{}/calendar/option-activity", self.config.endpoint);
        let response: BenzingaOptionsResponse =
            self.get_cached_or_fetch(&cache_key, &url, &params).await?;

        let mut events = Vec::new();
        for item in response.data {
            let option_type = match item.option_type.as_str() {
                "CALL" => OptionsType::Call,
                "PUT" => OptionsType::Put,
                _ => OptionsType::Call,
            };

            let activity_type = match item.activity_type.as_deref() {
                Some("SWEEP") => UnusualOptionsType::Sweep,
                Some("BLOCK") => UnusualOptionsType::BlockTrade,
                Some("VOLUME") => UnusualOptionsType::VolumeSpike,
                _ => UnusualOptionsType::BlockTrade,
            };

            let sentiment = match item.sentiment.as_deref() {
                Some("BULLISH") => OptionsSentiment::Bullish,
                Some("BEARISH") => OptionsSentiment::Bearish,
                _ => OptionsSentiment::Neutral,
            };

            let expiration_date = NaiveDate::parse_from_str(&item.date_expiry, "%Y-%m-%d")
                .map_err(|e| DataError::parse(format!("Invalid expiration date: {}", e)))?;
            // SAFETY: (0, 0, 0) is always a valid HMS time
            #[allow(clippy::unwrap_used)]
            let expiration = expiration_date.and_hms_opt(0, 0, 0).unwrap().and_utc();
            let expiry = expiration; // Same as expiration

            let contract = OptionsContract {
                symbol: Symbol::from(item.ticker.clone()),
                expiry,
                expiration,
                strike: Price::from_decimal(
                    Decimal::from_f64_retain(item.strike).unwrap_or_default(),
                ),
                option_type,
                multiplier: 100,
            };

            let timestamp = DateTime::parse_from_rfc3339(&item.updated)
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or_else(|_| Utc::now());

            let event = UnusualOptionsEvent {
                symbol: Symbol::from(item.ticker),
                contract,
                unusual_type: activity_type,
                activity_type,
                volume: Quantity::new(item.volume as f64).unwrap_or(Quantity::zero()),
                open_interest: Quantity::new(item.open_interest.unwrap_or(0) as f64)
                    .unwrap_or(Quantity::zero()),
                premium: item
                    .cost_basis
                    .map(|p| Price::from_decimal(Decimal::from_f64_retain(p).unwrap_or_default())),
                implied_volatility: None,
                sentiment,
                confidence: 0.8, // Default confidence
                description: format!(
                    "{:?} {:?} options activity detected",
                    sentiment, activity_type
                ),
                timestamp,
            };

            events.push(event);
            self.metrics
                .data_points_retrieved
                .fetch_add(1, Ordering::Relaxed);
        }

        info!("Retrieved {} options events from Benzinga", events.len());
        Ok(events)
    }

    /// Get economic calendar events
    #[instrument(skip(self))]
    pub async fn get_calendar_events(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<NewsEvent>> {
        let params = vec![
            ("dateFrom".to_string(), start.format("%Y-%m-%d").to_string()),
            ("dateTo".to_string(), end.format("%Y-%m-%d").to_string()),
            ("pageSize".to_string(), "1000".to_string()),
        ];

        let cache_key = format!(
            "benzinga:calendar:{}:{}",
            start.format("%Y%m%d"),
            end.format("%Y%m%d")
        );

        let url = format!("{}/calendar/economics", self.config.endpoint);
        let response: BenzingaCalendarResponse =
            self.get_cached_or_fetch(&cache_key, &url, &params).await?;

        let mut events = Vec::new();
        for item in response.data {
            let event_date = DateTime::parse_from_str(&item.date, "%Y-%m-%d")
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or_else(|_| Utc::now());

            let symbols = if let Some(ticker) = item.ticker {
                vec![Symbol::from(ticker)]
            } else {
                vec![]
            };

            let mut summary_parts = Vec::new();
            if let Some(desc) = &item.description {
                summary_parts.push(desc.clone());
            }
            if let Some(actual) = &item.actual {
                if let Some(consensus) = &item.consensus {
                    summary_parts.push(format!("Actual: {} vs Consensus: {}", actual, consensus));
                }
            }

            let event = NewsEvent {
                story_id: item.id,
                headline: item.name.clone(),
                content: item.name.clone(),
                summary: if summary_parts.is_empty() {
                    "No details available".to_string()
                } else {
                    summary_parts.join(". ")
                },
                symbol: symbols.first().cloned(),
                symbols,
                category: "economic".to_string(),
                tags: vec!["economic_data".to_string(), "calendar".to_string()],
                impact_score: item.importance,
                importance: item.importance.unwrap_or(0.6), // Economic events moderate importance
                author: "Economic Calendar".to_string(),
                source: "Benzinga".to_string(),
                published_at: event_date,
                timestamp: Utc::now(),
                url: "".to_string(),
                sentiment_score: None, // Economic events typically neutral
                #[allow(deprecated)]
                sentiment: None,
                event_type: crate::providers::common::NewsEventType::Economic,
            };

            events.push(event);
            self.metrics
                .data_points_retrieved
                .fetch_add(1, Ordering::Relaxed);
        }

        info!("Retrieved {} calendar events from Benzinga", events.len());
        Ok(events)
    }

    /// Get all events for the specified symbols and time range
    #[instrument(skip(self))]
    pub async fn get_all_events(
        &self,
        symbols: Option<&[&str]>,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
    ) -> Result<Vec<ExtendedMarketDataEvent>> {
        let mut all_events = Vec::new();

        // Fetch all event types concurrently
        let (news_result, ratings_result, earnings_result, options_result, calendar_result) = tokio::join!(
            self.get_news_events(symbols, start, end),
            self.get_rating_events(symbols, start, end),
            self.get_earnings_events(symbols, start, end),
            self.get_options_events(symbols, start, end),
            self.get_calendar_events(start, end)
        );

        // Process results
        if let Ok(events) = news_result {
            for event in events {
                all_events.push(ExtendedMarketDataEvent::NewsAlert(event));
            }
        }

        if let Ok(events) = ratings_result {
            for event in events {
                all_events.push(ExtendedMarketDataEvent::AnalystRating(event));
            }
        }

        if let Ok(events) = earnings_result {
            for event in events {
                all_events.push(ExtendedMarketDataEvent::NewsAlert(event));
            }
        }

        if let Ok(events) = options_result {
            for event in events {
                all_events.push(ExtendedMarketDataEvent::UnusualOptions(event));
            }
        }

        if let Ok(events) = calendar_result {
            for event in events {
                all_events.push(ExtendedMarketDataEvent::NewsAlert(event));
            }
        }

        // Sort by timestamp
        all_events.sort_by(|a, b| a.timestamp().cmp(&b.timestamp()));

        info!("Retrieved {} total events from Benzinga", all_events.len());
        Ok(all_events)
    }

    /// Get metrics for monitoring
    pub fn get_metrics(&self) -> HistoricalMetrics {
        HistoricalMetrics {
            requests_made: AtomicU64::new(self.metrics.requests_made.load(Ordering::Relaxed)),
            successful_requests: AtomicU64::new(
                self.metrics.successful_requests.load(Ordering::Relaxed),
            ),
            failed_requests: AtomicU64::new(self.metrics.failed_requests.load(Ordering::Relaxed)),
            cache_hits: AtomicU64::new(self.metrics.cache_hits.load(Ordering::Relaxed)),
            cache_misses: AtomicU64::new(self.metrics.cache_misses.load(Ordering::Relaxed)),
            data_points_retrieved: AtomicU64::new(
                self.metrics.data_points_retrieved.load(Ordering::Relaxed),
            ),
            data_points_filtered: AtomicU64::new(
                self.metrics.data_points_filtered.load(Ordering::Relaxed),
            ),
            avg_response_time_ms: AtomicU64::new(
                self.metrics.avg_response_time_ms.load(Ordering::Relaxed),
            ),
            total_data_volume_bytes: AtomicU64::new(
                self.metrics.total_data_volume_bytes.load(Ordering::Relaxed),
            ),
        }
    }

    /// Clear all caches
    pub async fn clear_cache(&self) -> Result<()> {
        // Clear Redis cache
        #[cfg(feature = "redis-cache")]
        if let Some(redis_client) = &self.redis_client {
            if let Ok(mut conn) = redis_client.get_multiplexed_async_connection().await {
                let _: std::result::Result<(), _> =
                    redis::cmd("FLUSHDB").query_async(&mut conn).await;
            }
        }
        // Clear in-memory cache
        let mut cache = self.cache.write().await;
        cache.clear();

        info!("Cache cleared");
        Ok(())
    }
}

#[async_trait]
impl HistoricalProvider for ProductionBenzingaHistoricalProvider {
    async fn fetch(
        &self,
        symbol: &Symbol,
        schema: HistoricalSchema,
        range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>> {
        debug!(
            "Fetching historical data for {} ({:?}) from {} to {}",
            symbol, schema, range.start, range.end
        );

        match schema {
            HistoricalSchema::News => {
                // Provider-specific data like NewsAlert has no core equivalent
                // Return empty vector since HistoricalProvider trait expects MarketDataEvent
                Ok(vec![])
            },
            HistoricalSchema::AnalystRating => {
                // Provider-specific data like AnalystRating has no core equivalent
                // Return empty vector since HistoricalProvider trait expects MarketDataEvent
                Ok(vec![])
            },
            HistoricalSchema::UnusualOptions => {
                // Provider-specific data like UnusualOptions has no core equivalent
                // Return empty vector since HistoricalProvider trait expects MarketDataEvent
                Ok(vec![])
            },
            _ => Err(DataError::Unsupported(format!(
                "Schema {:?} not supported by Benzinga",
                schema
            ))),
        }
    }

    async fn fetch_batch(
        &self,
        symbols: &[Symbol],
        schema: HistoricalSchema,
        range: TimeRange,
    ) -> Result<Vec<MarketDataEvent>> {
        info!(
            "Fetching batch historical data for {} symbols",
            symbols.len()
        );

        let symbol_strings: Vec<String> = symbols.iter().map(|s| s.to_string()).collect();
        let _symbol_strs: Vec<&str> = symbol_strings.iter().map(|s| s.as_str()).collect();

        match schema {
            HistoricalSchema::News => {
                // Provider-specific data like NewsAlert has no core equivalent
                // Return empty vector since HistoricalProvider trait expects MarketDataEvent
                Ok(vec![])
            },
            HistoricalSchema::AnalystRating => {
                // Provider-specific data like AnalystRating has no core equivalent
                // Return empty vector since HistoricalProvider trait expects MarketDataEvent
                Ok(vec![])
            },
            HistoricalSchema::UnusualOptions => {
                // Provider-specific data like UnusualOptions has no core equivalent
                // Return empty vector since HistoricalProvider trait expects MarketDataEvent
                Ok(vec![])
            },
            _ => {
                // For unsupported schemas, fetch individual symbols
                let mut all_events = Vec::new();
                for symbol in symbols {
                    let mut events = self.fetch(symbol, schema, range).await?;
                    all_events.append(&mut events);
                }
                // Sort by timestamp for proper ordering
                all_events.sort_by_key(|event| get_event_timestamp(event));
                Ok(all_events)
            },
        }
    }

    fn supports_schema(&self, schema: HistoricalSchema) -> bool {
        matches!(
            schema,
            HistoricalSchema::News
                | HistoricalSchema::Sentiment
                | HistoricalSchema::AnalystRating
                | HistoricalSchema::UnusualOptions
        )
    }

    fn max_range(&self) -> Duration {
        // Benzinga allows historical data but we limit for practical reasons
        Duration::from_secs(90 * 24 * 3600) // 90 days
    }

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

// Implement trait extensions for better ergonomics
impl std::fmt::Display for OptionsSentiment {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            OptionsSentiment::Bullish => write!(f, "bullish"),
            OptionsSentiment::Bearish => write!(f, "bearish"),
            OptionsSentiment::Neutral => write!(f, "neutral"),
        }
    }
}

impl std::fmt::Display for UnusualOptionsType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            UnusualOptionsType::Block => write!(f, "block"),
            UnusualOptionsType::Sweep => write!(f, "sweep"),
            UnusualOptionsType::Split => write!(f, "split"),
            UnusualOptionsType::HighVolume => write!(f, "high_volume"),
            UnusualOptionsType::HighOpenInterest => write!(f, "high_open_interest"),
            UnusualOptionsType::BlockTrade => write!(f, "block_trade"),
            UnusualOptionsType::VolumeSpike => write!(f, "volume_spike"),
            UnusualOptionsType::OpenInterestSpike => write!(f, "open_interest_spike"),
            UnusualOptionsType::VolatilitySpike => write!(f, "volatility_spike"),
        }
    }
}

#[cfg(test)]
#[allow(clippy::assertions_on_result_states)]
mod tests {
    use super::*;

    #[test]
    fn test_config_default() {
        let config = ProductionBenzingaHistoricalConfig::default();
        assert!(config.rate_limit_per_second > 0);
        assert!(config.max_retry_attempts > 0);
        assert!(config.cache_ttl_secs > 0);
        assert!(config.bulk_batch_size > 0);
    }

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

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

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

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

    #[test]
    fn test_cache_key_generation() {
        let symbols = ["AAPL", "SPY"];
        let start = Utc::now();
        let end = start + chrono::Duration::days(1);

        let key = format!(
            "benzinga:news:{}:{}:{}",
            symbols.join(","),
            start.format("%Y%m%d"),
            end.format("%Y%m%d")
        );

        assert!(key.contains("benzinga:news"));
        assert!(key.contains("AAPL,SPY"));
    }

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

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

        // Simulate some operations
        provider
            .metrics
            .requests_made
            .fetch_add(5, Ordering::Relaxed);
        provider
            .metrics
            .successful_requests
            .fetch_add(4, Ordering::Relaxed);
        provider
            .metrics
            .failed_requests
            .fetch_add(1, Ordering::Relaxed);

        let metrics = provider.get_metrics();
        assert_eq!(metrics.requests_made.load(Ordering::Relaxed), 5);
        assert_eq!(metrics.successful_requests.load(Ordering::Relaxed), 4);
        assert_eq!(metrics.failed_requests.load(Ordering::Relaxed), 1);
    }
}
