Upload 17 files

collectors/explorers.py

@@ -472,6 +472,71 @@ async def collect_explorer_data() -> List[Dict[str, Any]]:
     return processed_results
 
 
+class ExplorerDataCollector:
+    """
+    Explorer Data Collector class for WebSocket streaming interface
+    Wraps the standalone explorer data collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the explorer data collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect blockchain explorer data from all sources
+
+        Returns:
+            Dict with aggregated explorer data
+        """
+        results = await collect_explorer_data()
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "latest_block": None,
+            "network_hashrate": None,
+            "difficulty": None,
+            "mempool_size": None,
+            "transactions_count": None,
+            "gas_prices": {},
+            "sources": [],
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        for result in results:
+            if result.get("success") and result.get("data"):
+                provider = result.get("provider", "unknown")
+                aggregated["sources"].append(provider)
+
+                data = result["data"]
+
+                # Parse gas price data
+                if "result" in data and isinstance(data["result"], dict):
+                    gas_data = data["result"]
+                    if provider == "Etherscan":
+                        aggregated["gas_prices"]["ethereum"] = {
+                            "safe": gas_data.get("SafeGasPrice"),
+                            "propose": gas_data.get("ProposeGasPrice"),
+                            "fast": gas_data.get("FastGasPrice")
+                        }
+                    elif provider == "BscScan":
+                        aggregated["gas_prices"]["bsc"] = gas_data.get("result")
+
+                # Parse network stats
+                if provider == "TronScan" and "data" in data:
+                    stats = data["data"]
+                    aggregated["latest_block"] = stats.get("latestBlock")
+                    aggregated["transactions_count"] = stats.get("totalTransaction")
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():

collectors/market_data.py

@@ -436,6 +436,92 @@ async def collect_market_data() -> List[Dict[str, Any]]:
     return processed_results
 
 
+class MarketDataCollector:
+    """
+    Market Data Collector class for WebSocket streaming interface
+    Wraps the standalone market data collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the market data collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect market data from all sources
+
+        Returns:
+            Dict with aggregated market data
+        """
+        results = await collect_market_data()
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "prices": {},
+            "volumes": {},
+            "market_caps": {},
+            "price_changes": {},
+            "sources": [],
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        for result in results:
+            if result.get("success") and result.get("data"):
+                provider = result.get("provider", "unknown")
+                aggregated["sources"].append(provider)
+
+                data = result["data"]
+
+                # Parse CoinGecko data
+                if provider == "CoinGecko" and isinstance(data, dict):
+                    for coin_id, coin_data in data.items():
+                        if isinstance(coin_data, dict):
+                            symbol = coin_id.upper()
+                            if "usd" in coin_data:
+                                aggregated["prices"][symbol] = coin_data["usd"]
+                            if "usd_market_cap" in coin_data:
+                                aggregated["market_caps"][symbol] = coin_data["usd_market_cap"]
+                            if "usd_24h_vol" in coin_data:
+                                aggregated["volumes"][symbol] = coin_data["usd_24h_vol"]
+                            if "usd_24h_change" in coin_data:
+                                aggregated["price_changes"][symbol] = coin_data["usd_24h_change"]
+
+                # Parse CoinMarketCap data
+                elif provider == "CoinMarketCap" and isinstance(data, dict):
+                    if "data" in data:
+                        for symbol, coin_data in data["data"].items():
+                            if isinstance(coin_data, dict) and "quote" in coin_data:
+                                quote = coin_data.get("quote", {}).get("USD", {})
+                                if "price" in quote:
+                                    aggregated["prices"][symbol] = quote["price"]
+                                if "market_cap" in quote:
+                                    aggregated["market_caps"][symbol] = quote["market_cap"]
+                                if "volume_24h" in quote:
+                                    aggregated["volumes"][symbol] = quote["volume_24h"]
+                                if "percent_change_24h" in quote:
+                                    aggregated["price_changes"][symbol] = quote["percent_change_24h"]
+
+                # Parse Binance data
+                elif provider == "Binance" and isinstance(data, list):
+                    for ticker in data:
+                        if isinstance(ticker, dict):
+                            symbol = ticker.get("symbol", "").replace("USDT", "")
+                            if "lastPrice" in ticker:
+                                aggregated["prices"][symbol] = float(ticker["lastPrice"])
+                            if "volume" in ticker:
+                                aggregated["volumes"][symbol] = float(ticker["volume"])
+                            if "priceChangePercent" in ticker:
+                                aggregated["price_changes"][symbol] = float(ticker["priceChangePercent"])
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():

collectors/news.py

@@ -358,6 +358,77 @@ async def collect_news_data() -> List[Dict[str, Any]]:
     return processed_results
 
 
+# Alias for backward compatibility
+collect_news = collect_news_data
+
+
+class NewsCollector:
+    """
+    News Collector class for WebSocket streaming interface
+    Wraps the standalone news collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the news collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect news data from all sources
+
+        Returns:
+            Dict with aggregated news data
+        """
+        results = await collect_news_data()
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "articles": [],
+            "sources": [],
+            "categories": [],
+            "breaking": [],
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        for result in results:
+            if result.get("success") and result.get("data"):
+                provider = result.get("provider", "unknown")
+                aggregated["sources"].append(provider)
+
+                data = result["data"]
+
+                # Parse CryptoPanic posts
+                if provider == "CryptoPanic" and "results" in data:
+                    for post in data["results"][:10]:  # Take top 10
+                        aggregated["articles"].append({
+                            "title": post.get("title"),
+                            "url": post.get("url"),
+                            "source": post.get("source", {}).get("title"),
+                            "published_at": post.get("published_at"),
+                            "kind": post.get("kind"),
+                            "votes": post.get("votes", {})
+                        })
+
+                # Parse NewsAPI articles
+                elif provider == "NewsAPI" and "articles" in data:
+                    for article in data["articles"][:10]:  # Take top 10
+                        aggregated["articles"].append({
+                            "title": article.get("title"),
+                            "url": article.get("url"),
+                            "source": article.get("source", {}).get("name"),
+                            "published_at": article.get("publishedAt"),
+                            "description": article.get("description")
+                        })
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():

collectors/onchain.py

@@ -423,6 +423,58 @@ async def collect_onchain_data() -> List[Dict[str, Any]]:
     return processed_results
 
 
+class OnChainCollector:
+    """
+    On-Chain Analytics Collector class for WebSocket streaming interface
+    Wraps the standalone on-chain data collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the on-chain collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect on-chain analytics data from all sources
+
+        Returns:
+            Dict with aggregated on-chain data
+        """
+        results = await collect_onchain_data()
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "active_addresses": None,
+            "transaction_count": None,
+            "total_fees": None,
+            "gas_price": None,
+            "network_utilization": None,
+            "contract_events": [],
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        for result in results:
+            if result.get("success") and result.get("data"):
+                provider = result.get("provider", "unknown")
+                data = result["data"]
+
+                # Skip placeholders but still return basic structure
+                if isinstance(data, dict) and data.get("status") == "placeholder":
+                    continue
+
+                # Parse data from various providers (when implemented)
+                # Currently all are placeholders, so this will be empty
+                pass
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():

collectors/rpc_nodes.py

@@ -527,6 +527,89 @@ async def collect_rpc_data(
     return all_results
 
 
+class RPCNodeCollector:
+    """
+    RPC Node Collector class for WebSocket streaming interface
+    Wraps the standalone RPC node collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the RPC node collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect RPC node data from all sources
+
+        Returns:
+            Dict with aggregated RPC node data
+        """
+        import os
+        infura_key = os.getenv("INFURA_API_KEY")
+        alchemy_key = os.getenv("ALCHEMY_API_KEY")
+        results = await collect_rpc_data(infura_key, alchemy_key)
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "nodes": [],
+            "active_nodes": 0,
+            "total_nodes": 0,
+            "average_latency": 0,
+            "events": [],
+            "block_number": None,
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        total_latency = 0
+        latency_count = 0
+
+        for result in results:
+            aggregated["total_nodes"] += 1
+
+            if result.get("success"):
+                aggregated["active_nodes"] += 1
+                provider = result.get("provider", "unknown")
+                response_time = result.get("response_time_ms", 0)
+                data = result.get("data", {})
+
+                # Track latency
+                if response_time:
+                    total_latency += response_time
+                    latency_count += 1
+
+                # Add node info
+                node_info = {
+                    "provider": provider,
+                    "response_time_ms": response_time,
+                    "status": "active",
+                    "data": data
+                }
+
+                # Extract block number
+                if "result" in data and isinstance(data["result"], str):
+                    try:
+                        block_number = int(data["result"], 16)
+                        node_info["block_number"] = block_number
+                        if aggregated["block_number"] is None or block_number > aggregated["block_number"]:
+                            aggregated["block_number"] = block_number
+                    except:
+                        pass
+
+                aggregated["nodes"].append(node_info)
+
+        # Calculate average latency
+        if latency_count > 0:
+            aggregated["average_latency"] = total_latency / latency_count
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():

collectors/sentiment.py

@@ -213,6 +213,63 @@ async def collect_sentiment_data() -> List[Dict[str, Any]]:
     return processed_results
 
 
+# Alias for backward compatibility
+collect_sentiment = collect_sentiment_data
+
+
+class SentimentCollector:
+    """
+    Sentiment Collector class for WebSocket streaming interface
+    Wraps the standalone sentiment collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the sentiment collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect sentiment data from all sources
+
+        Returns:
+            Dict with aggregated sentiment data
+        """
+        results = await collect_sentiment_data()
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "overall_sentiment": None,
+            "sentiment_score": None,
+            "social_volume": None,
+            "trending_topics": [],
+            "by_source": {},
+            "social_trends": [],
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        for result in results:
+            if result.get("success") and result.get("data"):
+                provider = result.get("provider", "unknown")
+
+                # Parse Fear & Greed Index
+                if provider == "Alternative.me" and "data" in result["data"]:
+                    index_data = result["data"]["data"][0] if result["data"]["data"] else {}
+                    aggregated["sentiment_score"] = int(index_data.get("value", 0))
+                    aggregated["overall_sentiment"] = index_data.get("value_classification", "neutral")
+                    aggregated["by_source"][provider] = {
+                        "value": aggregated["sentiment_score"],
+                        "classification": aggregated["overall_sentiment"]
+                    }
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():

collectors/whale_tracking.py

@@ -465,6 +465,76 @@ async def collect_whale_tracking_data(whalealert_key: Optional[str] = None) -> L
     return processed_results
 
 
+class WhaleTrackingCollector:
+    """
+    Whale Tracking Collector class for WebSocket streaming interface
+    Wraps the standalone whale tracking collection functions
+    """
+
+    def __init__(self, config: Any = None):
+        """
+        Initialize the whale tracking collector
+
+        Args:
+            config: Configuration object (optional, for compatibility)
+        """
+        self.config = config
+        self.logger = logger
+
+    async def collect(self) -> Dict[str, Any]:
+        """
+        Collect whale tracking data from all sources
+
+        Returns:
+            Dict with aggregated whale tracking data
+        """
+        import os
+        whalealert_key = os.getenv("WHALEALERT_API_KEY")
+        results = await collect_whale_tracking_data(whalealert_key)
+
+        # Aggregate data for WebSocket streaming
+        aggregated = {
+            "large_transactions": [],
+            "whale_wallets": [],
+            "total_volume": 0,
+            "alert_threshold": 1000000,  # $1M default threshold
+            "alerts": [],
+            "timestamp": datetime.now(timezone.utc).isoformat()
+        }
+
+        for result in results:
+            if result.get("success") and result.get("data"):
+                provider = result.get("provider", "unknown")
+                data = result["data"]
+
+                # Skip placeholders
+                if isinstance(data, dict) and data.get("status") == "placeholder":
+                    continue
+
+                # Parse WhaleAlert transactions
+                if provider == "WhaleAlert" and isinstance(data, dict):
+                    transactions = data.get("transactions", [])
+                    for tx in transactions:
+                        aggregated["large_transactions"].append({
+                            "amount": tx.get("amount", 0),
+                            "amount_usd": tx.get("amount_usd", 0),
+                            "symbol": tx.get("symbol", "unknown"),
+                            "from": tx.get("from", {}).get("owner", "unknown"),
+                            "to": tx.get("to", {}).get("owner", "unknown"),
+                            "timestamp": tx.get("timestamp"),
+                            "source": provider
+                        })
+                    aggregated["total_volume"] += data.get("total_value_usd", 0)
+
+                # Parse other sources
+                elif isinstance(data, dict):
+                    tx_count = data.get("transaction_count", 0)
+                    total_value = data.get("total_value_usd", data.get("total_value", 0))
+                    aggregated["total_volume"] += total_value
+
+        return aggregated
+
+
 # Example usage
 if __name__ == "__main__":
     async def main():