swarms/new_features_examples/microstructure.py

import os
import threading
import time
from collections import deque
from dataclasses import dataclass
from datetime import datetime
from queue import Queue
from typing import Any, Dict, List, Optional, Tuple

import ccxt
import numpy as np
import pandas as pd
from dotenv import load_dotenv
from loguru import logger
from scipy import stats
from swarm_models import OpenAIChat

from swarms import Agent

logger.enable("")


@dataclass
class MarketSignal:
    timestamp: datetime
    signal_type: str
    source: str
    data: Dict[str, Any]
    confidence: float
    metadata: Dict[str, Any]


class MarketDataBuffer:
    def __init__(self, max_size: int = 10000):
        self.max_size = max_size
        self.data = deque(maxlen=max_size)
        self.lock = threading.Lock()

    def add(self, item: Any) -> None:
        with self.lock:
            self.data.append(item)

    def get_latest(self, n: int = None) -> List[Any]:
        with self.lock:
            if n is None:
                return list(self.data)
            return list(self.data)[-n:]


class SignalCSVWriter:
    def __init__(self, output_dir: str = "market_data"):
        self.output_dir = output_dir
        self.ensure_output_dir()
        self.files = {}

    def ensure_output_dir(self):
        if not os.path.exists(self.output_dir):
            os.makedirs(self.output_dir)

    def get_filename(self, signal_type: str, symbol: str) -> str:
        date_str = datetime.now().strftime("%Y%m%d")
        return (
            f"{self.output_dir}/{signal_type}_{symbol}_{date_str}.csv"
        )

    def write_order_book_signal(self, signal: MarketSignal):
        symbol = signal.data["symbol"]
        metrics = signal.data["metrics"]
        filename = self.get_filename("order_book", symbol)

        # Create header if file doesn't exist
        if not os.path.exists(filename):
            header = [
                "timestamp",
                "symbol",
                "bid_volume",
                "ask_volume",
                "mid_price",
                "bid_vwap",
                "ask_vwap",
                "spread",
                "depth_imbalance",
                "confidence",
            ]
            with open(filename, "w") as f:
                f.write(",".join(header) + "\n")

        # Write data
        data = [
            str(signal.timestamp),
            symbol,
            str(metrics["bid_volume"]),
            str(metrics["ask_volume"]),
            str(metrics["mid_price"]),
            str(metrics["bid_vwap"]),
            str(metrics["ask_vwap"]),
            str(metrics["spread"]),
            str(metrics["depth_imbalance"]),
            str(signal.confidence),
        ]

        with open(filename, "a") as f:
            f.write(",".join(data) + "\n")

    def write_tick_signal(self, signal: MarketSignal):
        symbol = signal.data["symbol"]
        metrics = signal.data["metrics"]
        filename = self.get_filename("tick_data", symbol)

        if not os.path.exists(filename):
            header = [
                "timestamp",
                "symbol",
                "vwap",
                "price_momentum",
                "volume_mean",
                "trade_intensity",
                "kyle_lambda",
                "roll_spread",
                "confidence",
            ]
            with open(filename, "w") as f:
                f.write(",".join(header) + "\n")

        data = [
            str(signal.timestamp),
            symbol,
            str(metrics["vwap"]),
            str(metrics["price_momentum"]),
            str(metrics["volume_mean"]),
            str(metrics["trade_intensity"]),
            str(metrics["kyle_lambda"]),
            str(metrics["roll_spread"]),
            str(signal.confidence),
        ]

        with open(filename, "a") as f:
            f.write(",".join(data) + "\n")

    def write_arbitrage_signal(self, signal: MarketSignal):
        if (
            "best_opportunity" not in signal.data
            or not signal.data["best_opportunity"]
        ):
            return

        symbol = signal.data["symbol"]
        opp = signal.data["best_opportunity"]
        filename = self.get_filename("arbitrage", symbol)

        if not os.path.exists(filename):
            header = [
                "timestamp",
                "symbol",
                "buy_venue",
                "sell_venue",
                "spread",
                "return",
                "buy_price",
                "sell_price",
                "confidence",
            ]
            with open(filename, "w") as f:
                f.write(",".join(header) + "\n")

        data = [
            str(signal.timestamp),
            symbol,
            opp["buy_venue"],
            opp["sell_venue"],
            str(opp["spread"]),
            str(opp["return"]),
            str(opp["buy_price"]),
            str(opp["sell_price"]),
            str(signal.confidence),
        ]

        with open(filename, "a") as f:
            f.write(",".join(data) + "\n")


class ExchangeManager:
    def __init__(self):
        self.available_exchanges = {
            "kraken": ccxt.kraken,
            "coinbase": ccxt.coinbase,
            "kucoin": ccxt.kucoin,
            "bitfinex": ccxt.bitfinex,
            "gemini": ccxt.gemini,
        }
        self.active_exchanges = {}
        self.test_exchanges()

    def test_exchanges(self):
        """Test each exchange and keep only the accessible ones"""
        for name, exchange_class in self.available_exchanges.items():
            try:
                exchange = exchange_class()
                exchange.load_markets()
                self.active_exchanges[name] = exchange
                logger.info(f"Successfully connected to {name}")
            except Exception as e:
                logger.warning(f"Could not connect to {name}: {e}")

    def get_primary_exchange(self) -> Optional[ccxt.Exchange]:
        """Get the first available exchange"""
        if not self.active_exchanges:
            raise RuntimeError("No exchanges available")
        return next(iter(self.active_exchanges.values()))

    def get_all_active_exchanges(self) -> Dict[str, ccxt.Exchange]:
        """Get all active exchanges"""
        return self.active_exchanges


class BaseMarketAgent(Agent):
    def __init__(
        self,
        agent_name: str,
        system_prompt: str,
        api_key: str,
        model_name: str = "gpt-4-0125-preview",
        temperature: float = 0.1,
    ):
        model = OpenAIChat(
            openai_api_key=api_key,
            model_name=model_name,
            temperature=temperature,
        )
        super().__init__(
            agent_name=agent_name,
            system_prompt=system_prompt,
            llm=model,
            max_loops=1,
            autosave=True,
            dashboard=False,
            verbose=True,
            dynamic_temperature_enabled=True,
            context_length=200000,
            streaming_on=True,
            output_type="str",
        )
        self.signal_queue = Queue()
        self.is_running = False
        self.last_update = datetime.now()
        self.update_interval = 1.0  # seconds

    def rate_limit_check(self) -> bool:
        current_time = datetime.now()
        if (
            current_time - self.last_update
        ).total_seconds() < self.update_interval:
            return False
        self.last_update = current_time
        return True


class OrderBookAgent(BaseMarketAgent):
    def __init__(self, api_key: str):
        system_prompt = """
        You are an Order Book Analysis Agent specialized in detecting institutional flows.
        Monitor order book depth and changes to identify potential large trades and institutional activity.
        Analyze patterns in order placement and cancellation rates.
        """
        super().__init__("OrderBookAgent", system_prompt, api_key)
        exchange_manager = ExchangeManager()
        self.exchange = exchange_manager.get_primary_exchange()
        self.order_book_buffer = MarketDataBuffer(max_size=100)
        self.vwap_window = 20

    def calculate_order_book_metrics(
        self, order_book: Dict
    ) -> Dict[str, float]:
        bids = np.array(order_book["bids"])
        asks = np.array(order_book["asks"])

        # Calculate key metrics
        bid_volume = np.sum(bids[:, 1])
        ask_volume = np.sum(asks[:, 1])
        mid_price = (bids[0][0] + asks[0][0]) / 2

        # Calculate VWAP
        bid_vwap = (
            np.sum(
                bids[: self.vwap_window, 0]
                * bids[: self.vwap_window, 1]
            )
            / bid_volume
            if bid_volume > 0
            else 0
        )
        ask_vwap = (
            np.sum(
                asks[: self.vwap_window, 0]
                * asks[: self.vwap_window, 1]
            )
            / ask_volume
            if ask_volume > 0
            else 0
        )

        # Calculate order book slope
        bid_slope = np.polyfit(
            range(len(bids[:10])), bids[:10, 0], 1
        )[0]
        ask_slope = np.polyfit(
            range(len(asks[:10])), asks[:10, 0], 1
        )[0]

        return {
            "bid_volume": bid_volume,
            "ask_volume": ask_volume,
            "mid_price": mid_price,
            "bid_vwap": bid_vwap,
            "ask_vwap": ask_vwap,
            "bid_slope": bid_slope,
            "ask_slope": ask_slope,
            "spread": asks[0][0] - bids[0][0],
            "depth_imbalance": (bid_volume - ask_volume)
            / (bid_volume + ask_volume),
        }

    def detect_large_orders(
        self, metrics: Dict[str, float], threshold: float = 2.0
    ) -> bool:
        historical_books = self.order_book_buffer.get_latest(20)
        if not historical_books:
            return False

        # Calculate historical volume statistics
        hist_volumes = [
            book["bid_volume"] + book["ask_volume"]
            for book in historical_books
        ]
        volume_mean = np.mean(hist_volumes)
        volume_std = np.std(hist_volumes)

        current_volume = metrics["bid_volume"] + metrics["ask_volume"]
        z_score = (current_volume - volume_mean) / (
            volume_std if volume_std > 0 else 1
        )

        return abs(z_score) > threshold

    def analyze_order_book(self, symbol: str) -> MarketSignal:
        if not self.rate_limit_check():
            return None

        try:
            order_book = self.exchange.fetch_order_book(
                symbol, limit=100
            )
            metrics = self.calculate_order_book_metrics(order_book)
            self.order_book_buffer.add(metrics)

            # Format data for LLM analysis
            analysis_prompt = f"""
            Analyze this order book for {symbol}:
            Bid Volume: {metrics['bid_volume']}
            Ask Volume: {metrics['ask_volume']}
            Mid Price: {metrics['mid_price']}
            Spread: {metrics['spread']}
            Depth Imbalance: {metrics['depth_imbalance']}
            
            What patterns do you see? Is there evidence of institutional activity?
            Are there any significant imbalances that could lead to price movement?
            """

            # Get LLM analysis
            llm_analysis = self.run(analysis_prompt)

            # Original signal creation with added LLM analysis
            return MarketSignal(
                timestamp=datetime.now(),
                signal_type="order_book_analysis",
                source="OrderBookAgent",
                data={
                    "metrics": metrics,
                    "large_order_detected": self.detect_large_orders(
                        metrics
                    ),
                    "symbol": symbol,
                    "llm_analysis": llm_analysis,  # Add LLM insights
                },
                confidence=min(
                    abs(metrics["depth_imbalance"]) * 0.7
                    + (
                        1.0
                        if self.detect_large_orders(metrics)
                        else 0.0
                    )
                    * 0.3,
                    1.0,
                ),
                metadata={
                    "update_latency": (
                        datetime.now() - self.last_update
                    ).total_seconds(),
                    "buffer_size": len(
                        self.order_book_buffer.get_latest()
                    ),
                },
            )
        except Exception as e:
            logger.error(f"Error in order book analysis: {str(e)}")
            return None


class TickDataAgent(BaseMarketAgent):
    def __init__(self, api_key: str):
        system_prompt = """
        You are a Tick Data Analysis Agent specialized in analyzing high-frequency price movements.
        Monitor tick-by-tick data for patterns indicating short-term price direction.
        Analyze trade size distribution and execution speed.
        """
        super().__init__("TickDataAgent", system_prompt, api_key)
        self.tick_buffer = MarketDataBuffer(max_size=5000)
        exchange_manager = ExchangeManager()
        self.exchange = exchange_manager.get_primary_exchange()

    def calculate_tick_metrics(
        self, ticks: List[Dict]
    ) -> Dict[str, float]:
        df = pd.DataFrame(ticks)
        df["price"] = pd.to_numeric(df["price"])
        df["volume"] = pd.to_numeric(df["amount"])

        # Calculate key metrics
        metrics = {}

        # Volume-weighted average price (VWAP)
        metrics["vwap"] = (df["price"] * df["volume"]).sum() / df[
            "volume"
        ].sum()

        # Price momentum
        metrics["price_momentum"] = df["price"].diff().mean()

        # Volume profile
        metrics["volume_mean"] = df["volume"].mean()
        metrics["volume_std"] = df["volume"].std()

        # Trade intensity
        time_diff = (
            df["timestamp"].max() - df["timestamp"].min()
        ) / 1000  # Convert to seconds
        metrics["trade_intensity"] = (
            len(df) / time_diff if time_diff > 0 else 0
        )

        # Microstructure indicators
        metrics["kyle_lambda"] = self.calculate_kyle_lambda(df)
        metrics["roll_spread"] = self.calculate_roll_spread(df)

        return metrics

    def calculate_kyle_lambda(self, df: pd.DataFrame) -> float:
        """Calculate Kyle's Lambda (price impact coefficient)"""
        try:
            price_changes = df["price"].diff().dropna()
            volume_changes = df["volume"].diff().dropna()

            if len(price_changes) > 1 and len(volume_changes) > 1:
                slope, _, _, _, _ = stats.linregress(
                    volume_changes, price_changes
                )
                return abs(slope)
        except Exception as e:
            logger.warning(f"Error calculating Kyle's Lambda: {e}")
        return 0.0

    def calculate_roll_spread(self, df: pd.DataFrame) -> float:
        """Calculate Roll's implied spread"""
        try:
            price_changes = df["price"].diff().dropna()
            if len(price_changes) > 1:
                autocov = np.cov(
                    price_changes[:-1], price_changes[1:]
                )[0][1]
                return 2 * np.sqrt(-autocov) if autocov < 0 else 0.0
        except Exception as e:
            logger.warning(f"Error calculating Roll spread: {e}")
        return 0.0

    def calculate_tick_metrics(
        self, ticks: List[Dict]
    ) -> Dict[str, float]:
        try:
            # Debug the incoming data structure
            logger.info(
                f"Raw tick data structure: {ticks[0] if ticks else 'No ticks'}"
            )

            # Convert trades to proper format
            formatted_trades = []
            for trade in ticks:
                formatted_trade = {
                    "price": float(
                        trade.get("price", trade.get("last", 0))
                    ),  # Handle different exchange formats
                    "amount": float(
                        trade.get(
                            "amount",
                            trade.get(
                                "size", trade.get("quantity", 0)
                            ),
                        )
                    ),
                    "timestamp": trade.get(
                        "timestamp", int(time.time() * 1000)
                    ),
                }
                formatted_trades.append(formatted_trade)

            df = pd.DataFrame(formatted_trades)

            if df.empty:
                logger.warning("No valid trades to analyze")
                return {
                    "vwap": 0.0,
                    "price_momentum": 0.0,
                    "volume_mean": 0.0,
                    "volume_std": 0.0,
                    "trade_intensity": 0.0,
                    "kyle_lambda": 0.0,
                    "roll_spread": 0.0,
                }

            # Calculate metrics with the properly formatted data
            metrics = {}
            metrics["vwap"] = (
                (df["price"] * df["amount"]).sum()
                / df["amount"].sum()
                if not df.empty
                else 0
            )
            metrics["price_momentum"] = (
                df["price"].diff().mean() if len(df) > 1 else 0
            )
            metrics["volume_mean"] = df["amount"].mean()
            metrics["volume_std"] = df["amount"].std()

            time_diff = (
                (df["timestamp"].max() - df["timestamp"].min()) / 1000
                if len(df) > 1
                else 1
            )
            metrics["trade_intensity"] = (
                len(df) / time_diff if time_diff > 0 else 0
            )

            metrics["kyle_lambda"] = self.calculate_kyle_lambda(df)
            metrics["roll_spread"] = self.calculate_roll_spread(df)

            logger.info(f"Calculated metrics: {metrics}")
            return metrics

        except Exception as e:
            logger.error(
                f"Error in calculate_tick_metrics: {str(e)}",
                exc_info=True,
            )
            # Return default metrics on error
            return {
                "vwap": 0.0,
                "price_momentum": 0.0,
                "volume_mean": 0.0,
                "volume_std": 0.0,
                "trade_intensity": 0.0,
                "kyle_lambda": 0.0,
                "roll_spread": 0.0,
            }

    def analyze_ticks(self, symbol: str) -> MarketSignal:
        if not self.rate_limit_check():
            return None

        try:
            # Fetch recent trades
            trades = self.exchange.fetch_trades(symbol, limit=100)

            # Debug the raw trades data
            logger.info(f"Fetched {len(trades)} trades for {symbol}")
            if trades:
                logger.info(f"Sample trade: {trades[0]}")

            self.tick_buffer.add(trades)
            recent_ticks = self.tick_buffer.get_latest(1000)
            metrics = self.calculate_tick_metrics(recent_ticks)

            # Only proceed with LLM analysis if we have valid metrics
            if metrics["vwap"] > 0:
                analysis_prompt = f"""
                Analyze these trading patterns for {symbol}:
                VWAP: {metrics['vwap']:.2f}
                Price Momentum: {metrics['price_momentum']:.2f}
                Trade Intensity: {metrics['trade_intensity']:.2f}
                Kyle's Lambda: {metrics['kyle_lambda']:.2f}
                
                What does this tell us about:
                1. Current market sentiment
                2. Potential price direction
                3. Trading activity patterns
                """
                llm_analysis = self.run(analysis_prompt)
            else:
                llm_analysis = "Insufficient data for analysis"

            return MarketSignal(
                timestamp=datetime.now(),
                signal_type="tick_analysis",
                source="TickDataAgent",
                data={
                    "metrics": metrics,
                    "symbol": symbol,
                    "prediction": np.sign(metrics["price_momentum"]),
                    "llm_analysis": llm_analysis,
                },
                confidence=min(metrics["trade_intensity"] / 100, 1.0)
                * 0.4
                + min(metrics["kyle_lambda"], 1.0) * 0.6,
                metadata={
                    "update_latency": (
                        datetime.now() - self.last_update
                    ).total_seconds(),
                    "buffer_size": len(self.tick_buffer.get_latest()),
                },
            )

        except Exception as e:
            logger.error(
                f"Error in tick analysis: {str(e)}", exc_info=True
            )
            return None


class LatencyArbitrageAgent(BaseMarketAgent):
    def __init__(self, api_key: str):
        system_prompt = """
        You are a Latency Arbitrage Agent specialized in detecting price discrepancies across venues.
        Monitor multiple exchanges for price differences exceeding transaction costs.
        Calculate optimal trade sizes and routes.
        """
        super().__init__(
            "LatencyArbitrageAgent", system_prompt, api_key
        )
        exchange_manager = ExchangeManager()
        self.exchanges = exchange_manager.get_all_active_exchanges()
        self.fee_structure = {
            "kraken": 0.0026,  # 0.26% taker fee
            "coinbase": 0.006,  # 0.6% taker fee
            "kucoin": 0.001,  # 0.1% taker fee
            "bitfinex": 0.002,  # 0.2% taker fee
            "gemini": 0.003,  # 0.3% taker fee
        }
        self.price_buffer = {
            ex: MarketDataBuffer(max_size=100)
            for ex in self.exchanges
        }

    def calculate_effective_prices(
        self, ticker: Dict, venue: str
    ) -> Tuple[float, float]:
        """Calculate effective prices including fees"""
        fee = self.fee_structure[venue]
        return (
            ticker["bid"] * (1 - fee),  # Effective sell price
            ticker["ask"] * (1 + fee),  # Effective buy price
        )

    def calculate_arbitrage_metrics(
        self, prices: Dict[str, Dict]
    ) -> Dict:
        opportunities = []

        for venue1 in prices:
            for venue2 in prices:
                if venue1 != venue2:
                    sell_price, _ = self.calculate_effective_prices(
                        prices[venue1], venue1
                    )
                    _, buy_price = self.calculate_effective_prices(
                        prices[venue2], venue2
                    )

                    spread = sell_price - buy_price
                    if spread > 0:
                        opportunities.append(
                            {
                                "sell_venue": venue1,
                                "buy_venue": venue2,
                                "spread": spread,
                                "return": spread / buy_price,
                                "buy_price": buy_price,
                                "sell_price": sell_price,
                            }
                        )

        return {
            "opportunities": opportunities,
            "best_opportunity": (
                max(opportunities, key=lambda x: x["return"])
                if opportunities
                else None
            ),
        }

    def find_arbitrage(self, symbol: str) -> MarketSignal:
        """
        Find arbitrage opportunities across exchanges with LLM analysis
        """
        if not self.rate_limit_check():
            return None

        try:
            prices = {}
            timestamps = {}

            for name, exchange in self.exchanges.items():
                try:
                    ticker = exchange.fetch_ticker(symbol)
                    prices[name] = {
                        "bid": ticker["bid"],
                        "ask": ticker["ask"],
                    }
                    timestamps[name] = ticker["timestamp"]
                    self.price_buffer[name].add(prices[name])
                except Exception as e:
                    logger.warning(
                        f"Error fetching {name} price: {e}"
                    )

            if len(prices) < 2:
                return None

            metrics = self.calculate_arbitrage_metrics(prices)

            if not metrics["best_opportunity"]:
                return None

            # Calculate confidence based on spread and timing
            opp = metrics["best_opportunity"]
            timing_factor = 1.0 - min(
                abs(
                    timestamps[opp["sell_venue"]]
                    - timestamps[opp["buy_venue"]]
                )
                / 1000,
                1.0,
            )
            spread_factor = min(
                opp["return"] * 5, 1.0
            )  # Scale return to confidence

            confidence = timing_factor * 0.4 + spread_factor * 0.6

            # Format price data for LLM analysis
            price_summary = "\n".join(
                [
                    f"{venue}: Bid ${prices[venue]['bid']:.2f}, Ask ${prices[venue]['ask']:.2f}"
                    for venue in prices.keys()
                ]
            )

            # Create detailed analysis prompt
            analysis_prompt = f"""
            Analyze this arbitrage opportunity for {symbol}:

            Current Prices:
            {price_summary}

            Best Opportunity Found:
            Buy Venue: {opp['buy_venue']} at ${opp['buy_price']:.2f}
            Sell Venue: {opp['sell_venue']} at ${opp['sell_price']:.2f}
            Spread: ${opp['spread']:.2f}
            Expected Return: {opp['return']*100:.3f}%
            Time Difference: {abs(timestamps[opp['sell_venue']] - timestamps[opp['buy_venue']])}ms

            Consider:
            1. Is this opportunity likely to be profitable after execution costs?
            2. What risks might prevent successful execution?
            3. What market conditions might have created this opportunity?
            4. How does the timing difference affect execution probability?
            """

            # Get LLM analysis
            llm_analysis = self.run(analysis_prompt)

            # Create comprehensive signal
            return MarketSignal(
                timestamp=datetime.now(),
                signal_type="arbitrage_opportunity",
                source="LatencyArbitrageAgent",
                data={
                    "metrics": metrics,
                    "symbol": symbol,
                    "best_opportunity": metrics["best_opportunity"],
                    "all_prices": prices,
                    "llm_analysis": llm_analysis,
                    "timing": {
                        "time_difference_ms": abs(
                            timestamps[opp["sell_venue"]]
                            - timestamps[opp["buy_venue"]]
                        ),
                        "timestamps": timestamps,
                    },
                },
                confidence=confidence,
                metadata={
                    "update_latency": (
                        datetime.now() - self.last_update
                    ).total_seconds(),
                    "timestamp_deltas": timestamps,
                    "venue_count": len(prices),
                    "execution_risk": 1.0
                    - timing_factor,  # Higher time difference = higher risk
                },
            )

        except Exception as e:
            logger.error(f"Error in arbitrage analysis: {str(e)}")
            return None


class SwarmCoordinator:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.agents = {
            "order_book": OrderBookAgent(api_key),
            "tick_data": TickDataAgent(api_key),
            "latency_arb": LatencyArbitrageAgent(api_key),
        }
        self.signal_processors = []
        self.signal_history = MarketDataBuffer(max_size=1000)
        self.running = False
        self.lock = threading.Lock()
        self.csv_writer = SignalCSVWriter()

    def register_signal_processor(self, processor):
        """Register a new signal processor function"""
        with self.lock:
            self.signal_processors.append(processor)

    def process_signals(self, signals: List[MarketSignal]):
        """Process signals through all registered processors"""
        if not signals:
            return

        self.signal_history.add(signals)

        try:
            for processor in self.signal_processors:
                processor(signals)
        except Exception as e:
            logger.error(f"Error in signal processing: {e}")

    def aggregate_signals(
        self, signals: List[MarketSignal]
    ) -> Dict[str, Any]:
        """Aggregate multiple signals into a combined market view"""
        if not signals:
            return {}

        self.signal_history.add(signals)

        aggregated = {
            "timestamp": datetime.now(),
            "symbols": set(),
            "agent_signals": {},
            "combined_confidence": 0,
            "market_state": {},
        }

        for signal in signals:
            symbol = signal.data.get("symbol")
            if symbol:
                aggregated["symbols"].add(symbol)

            agent_type = signal.source
            if agent_type not in aggregated["agent_signals"]:
                aggregated["agent_signals"][agent_type] = []
            aggregated["agent_signals"][agent_type].append(signal)

            # Update market state based on signal type
            if signal.signal_type == "order_book_analysis":
                metrics = signal.data.get("metrics", {})
                aggregated["market_state"].update(
                    {
                        "order_book_imbalance": metrics.get(
                            "depth_imbalance"
                        ),
                        "spread": metrics.get("spread"),
                        "large_orders_detected": signal.data.get(
                            "large_order_detected"
                        ),
                    }
                )
            elif signal.signal_type == "tick_analysis":
                metrics = signal.data.get("metrics", {})
                aggregated["market_state"].update(
                    {
                        "price_momentum": metrics.get(
                            "price_momentum"
                        ),
                        "trade_intensity": metrics.get(
                            "trade_intensity"
                        ),
                        "kyle_lambda": metrics.get("kyle_lambda"),
                    }
                )
            elif signal.signal_type == "arbitrage_opportunity":
                opp = signal.data.get("best_opportunity")
                if opp:
                    aggregated["market_state"].update(
                        {
                            "arbitrage_spread": opp.get("spread"),
                            "arbitrage_return": opp.get("return"),
                        }
                    )

        # Calculate combined confidence as weighted average
        confidences = [s.confidence for s in signals]
        if confidences:
            aggregated["combined_confidence"] = np.mean(confidences)

        return aggregated

    def start(self, symbols: List[str], interval: float = 1.0):
        """Start the swarm monitoring system"""
        if self.running:
            logger.warning("Swarm is already running")
            return

        self.running = True

        def agent_loop(agent, symbol):
            while self.running:
                try:
                    if isinstance(agent, OrderBookAgent):
                        signal = agent.analyze_order_book(symbol)
                    elif isinstance(agent, TickDataAgent):
                        signal = agent.analyze_ticks(symbol)
                    elif isinstance(agent, LatencyArbitrageAgent):
                        signal = agent.find_arbitrage(symbol)

                    if signal:
                        agent.signal_queue.put(signal)
                except Exception as e:
                    logger.error(
                        f"Error in {agent.agent_name} loop: {e}"
                    )

                time.sleep(interval)

        def signal_collection_loop():
            while self.running:
                try:
                    current_signals = []

                    # Collect signals from all agents
                    for agent in self.agents.values():
                        while not agent.signal_queue.empty():
                            signal = agent.signal_queue.get_nowait()
                            if signal:
                                current_signals.append(signal)

                    if current_signals:
                        # Process current signals
                        self.process_signals(current_signals)

                        # Aggregate and analyze
                        aggregated = self.aggregate_signals(
                            current_signals
                        )
                        logger.info(
                            f"Aggregated market view: {aggregated}"
                        )

                except Exception as e:
                    logger.error(
                        f"Error in signal collection loop: {e}"
                    )

                time.sleep(interval)

        # Start agent threads
        self.threads = []
        for symbol in symbols:
            for agent in self.agents.values():
                thread = threading.Thread(
                    target=agent_loop,
                    args=(agent, symbol),
                    daemon=True,
                )
                thread.start()
                self.threads.append(thread)

        # Start signal collection thread
        collection_thread = threading.Thread(
            target=signal_collection_loop, daemon=True
        )
        collection_thread.start()
        self.threads.append(collection_thread)

    def stop(self):
        """Stop the swarm monitoring system"""
        self.running = False
        for thread in self.threads:
            thread.join(timeout=5.0)
        logger.info("Swarm stopped")


def market_making_processor(signals: List[MarketSignal]):
    """Enhanced signal processor with LLM analysis integration"""
    for signal in signals:
        if signal.confidence > 0.8:
            if signal.signal_type == "arbitrage_opportunity":
                opp = signal.data.get("best_opportunity")
                if (
                    opp and opp["return"] > 0.001
                ):  # 0.1% return threshold
                    logger.info(
                        "\nSignificant arbitrage opportunity detected:"
                    )
                    logger.info(f"Return: {opp['return']*100:.3f}%")
                    logger.info(f"Spread: ${opp['spread']:.2f}")
                    if "llm_analysis" in signal.data:
                        logger.info("\nLLM Analysis:")
                        logger.info(signal.data["llm_analysis"])

            elif signal.signal_type == "order_book_analysis":
                imbalance = signal.data["metrics"]["depth_imbalance"]
                if abs(imbalance) > 0.3:
                    logger.info(
                        f"\nSignificant order book imbalance detected: {imbalance:.3f}"
                    )
                    if "llm_analysis" in signal.data:
                        logger.info("\nLLM Analysis:")
                        logger.info(signal.data["llm_analysis"])

            elif signal.signal_type == "tick_analysis":
                momentum = signal.data["metrics"]["price_momentum"]
                if abs(momentum) > 0:
                    logger.info(
                        f"\nSignificant price momentum detected: {momentum:.3f}"
                    )
                    if "llm_analysis" in signal.data:
                        logger.info("\nLLM Analysis:")
                        logger.info(signal.data["llm_analysis"])


load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

coordinator = SwarmCoordinator(api_key)
coordinator.register_signal_processor(market_making_processor)

symbols = ["BTC/USDT", "ETH/USDT"]

logger.info(
    "Starting market microstructure analysis with LLM integration..."
)
logger.info(f"Monitoring symbols: {symbols}")
logger.info(
    f"CSV files will be written to: {os.path.abspath('market_data')}"
)

try:
    coordinator.start(symbols)
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    logger.info("Gracefully shutting down...")
    coordinator.stop()