# Install RabbitMQ
brew install rabbitmq  # macOS
# or
sudo apt-get install rabbitmq-server  # Ubuntu

# Start RabbitMQ
rabbitmq-server

# Enable management interface
rabbitmq-plugins enable rabbitmq_management

import pika
import json
import time
from datetime import datetime

class RabbitMQClient:
    def __init__(self, host='localhost', port=5672, username='guest', password='guest'):
        self.credentials = pika.PlainCredentials(username, password)
        self.parameters = pika.ConnectionParameters(host, port, '/', self.credentials)
        self.connection = None
        self.channel = None
        self.connect()
    
    def connect(self):
        """Establish connection and create channel."""
        self.connection = pika.BlockingConnection(self.parameters)
        self.channel = self.connection.channel()
    
    def create_exchange(self, exchange_name, exchange_type='topic', durable=True):
        """Create an exchange."""
        self.channel.exchange_declare(
            exchange=exchange_name,
            exchange_type=exchange_type,
            durable=durable
        )
    
    def create_queue(self, queue_name, durable=True):
        """Create a queue."""
        self.channel.queue_declare(queue=queue_name, durable=durable)
    
    def bind_queue(self, queue_name, exchange_name, routing_key='#'):
        """Bind a queue to an exchange with a routing key."""
        self.channel.queue_bind(
            queue=queue_name,
            exchange=exchange_name,
            routing_key=routing_key
        )
    
    def publish(self, exchange_name, routing_key, message, persistent=True):
        """Publish a message to an exchange."""
        properties = pika.BasicProperties(
            delivery_mode=2 if persistent else 1,  # 2 = persistent
            content_type='application/json',
            timestamp=int(time.time())
        )
        
        self.channel.basic_publish(
            exchange=exchange_name,
            routing_key=routing_key,
            body=json.dumps(message),
            properties=properties
        )
    
    def consume(self, queue_name, callback, auto_ack=False):
        """Start consuming messages from a queue."""
        self.channel.basic_consume(
            queue=queue_name,
            on_message_callback=callback,
            auto_ack=auto_ack
        )
        self.channel.start_consuming()
    
    def close(self):
        """Close the connection."""
        if self.connection:
            self.connection.close()

# Usage example
def create_market_data_pipeline():
    """Create a market data distribution pipeline."""
    client = RabbitMQClient()
    
    # Create exchanges
    client.create_exchange('market_data', 'topic')
    client.create_exchange('orders', 'topic')
    client.create_exchange('executions', 'topic')
    
    # Create queues
    client.create_queue('price_updates')
    client.create_queue('trade_alerts')
    client.create_queue('order_events')
    client.create_queue('risk_events')
    
    # Bind queues to exchanges
    client.bind_queue('price_updates', 'market_data', 'price.*')
    client.bind_queue('trade_alerts', 'executions', 'trade.*')
    client.bind_queue('order_events', 'orders', 'order.*')
    
    return client

class MarketDataPublisher:
    """Publishes market data events."""
    
    def __init__(self, rabbitmq_client):
        self.client = rabbitmq_client
    
    def publish_price_update(self, symbol, price, volume):
        """Publish a price update event."""
        message = {
            'event_type': 'price_update',
            'symbol': symbol,
            'price': price,
            'volume': volume,
            'timestamp': datetime.utcnow().isoformat()
        }
        self.client.publish('market_data', f'price.{symbol}', message)
    
    def publish_trade(self, symbol, price, quantity, side):
        """Publish a trade execution event."""
        message = {
            'event_type': 'trade',
            'symbol': symbol,
            'price': price,
            'quantity': quantity,
            'side': side,
            'timestamp': datetime.utcnow().isoformat()
        }
        self.client.publish('executions', 'trade.completed', message)
    
    def publish_order(self, order_id, symbol, side, quantity, order_type):
        """Publish an order event."""
        message = {
            'event_type': 'order',
            'order_id': order_id,
            'symbol': symbol,
            'side': side,
            'quantity': quantity,
            'order_type': order_type,
            'timestamp': datetime.utcnow().isoformat()
        }
        self.client.publish('orders', f'order.{order_id}', message)

class MarketDataConsumer:
    """Consumes and processes market data events."""
    
    def __init__(self, rabbitmq_client):
        self.client = rabbitmq_client
    
    def start_consuming(self):
        """Start consuming events."""
        self.client.consume('price_updates', self.handle_price_update)
    
    def handle_price_update(self, channel, method, properties, body):
        """Handle a price update event."""
        data = json.loads(body)
        symbol = data['symbol']
        price = data['price']
        
        # Process the price update
        print(f"Price update: {symbol} ${price:.2f}")
        
        # Check for alerts
        self.check_alerts(symbol, price)
        
        # Update database
        self.update_price_history(symbol, price, data['timestamp'])
        
        # Acknowledge the message
        channel.basic_ack(delivery_tag=method.delivery_tag)
    
    def check_alerts(self, symbol, price):
        """Check if the price triggers any alerts."""
        # Implementation would check against user alerts
        pass
    
    def update_price_history(self, symbol, price, timestamp):
        """Update price history in database."""
        # Implementation would update the database
        pass

# Usage
client = RabbitMQClient()
publisher = MarketDataPublisher(client)
consumer = MarketDataConsumer(client)

# Publish price updates
publisher.publish_price_update('AAPL', 150.50, 1000000)
publisher.publish_price_update('GOOGL', 2800.00, 500000)

# Start consuming
# consumer.start_consuming()

# Install Kafka-python
# pip install kafka-python

from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError
import json
import time

class KafkaClient:
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self.consumers = {}
    
    def create_producer(self):
        """Create a Kafka producer."""
        self.producer = KafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None,
            acks='all',  # Wait for all replicas to acknowledge
            retries=5,
            max_in_flight_requests_per_connection=1  # Preserve ordering
        )
        return self.producer
    
    def create_consumer(self, topic, group_id, auto_offset_reset='earliest'):
        """Create a Kafka consumer."""
        consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.bootstrap_servers,
            group_id=group_id,
            auto_offset_reset=auto_offset_reset,
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            key_deserializer=lambda m: m.decode('utf-8') if m else None,
            enable_auto_commit=False,  # Manual commit for exactly-once semantics
            max_poll_records=100
        )
        self.consumers[group_id] = consumer
        return consumer
    
    def publish(self, topic, message, key=None, partition=None):
        """Publish a message to a topic."""
        if not self.producer:
            self.create_producer()
        
        future = self.producer.send(
            topic=topic,
            value=message,
            key=key,
            partition=partition
        )
        
        # Wait for the send to complete
        try:
            record_metadata = future.get(timeout=10)
            return {
                'topic': record_metadata.topic,
                'partition': record_metadata.partition,
                'offset': record_metadata.offset
            }
        except KafkaError as e:
            print(f"Failed to send message: {e}")
            return None
    
    def consume_loop(self, consumer_group, callback, max_messages=None):
        """Consume messages in a loop."""
        consumer = self.consumers.get(consumer_group)
        if not consumer:
            raise ValueError(f"Consumer group {consumer_group} not found")
        
        messages_processed = 0
        for message in consumer:
            callback(message)
            consumer.commit()
            messages_processed += 1
            if max_messages and messages_processed >= max_messages:
                break
    
    def close(self):
        """Close all connections."""
        if self.producer:
            self.producer.close()
        for consumer in self.consumers.values():
            consumer.close()

class MarketDataStream:
    """
    A complete market data streaming application using Kafka.
    """
    
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.client = KafkaClient(bootstrap_servers)
        self.producer = self.client.create_producer()
    
    def stream_price_data(self, symbol, price_generator):
        """Stream price data from a generator."""
        for price_data in price_generator:
            message = {
                'symbol': symbol,
                'price': price_data['price'],
                'volume': price_data.get('volume', 0),
                'timestamp': datetime.utcnow().isoformat()
            }
            self.client.publish('market-data', message, key=symbol)
            time.sleep(0.001)  # Simulate real-time
    
    def calculate_indicators(self, symbol, window=20):
        """Calculate real-time indicators."""
        consumer = self.client.create_consumer(
            topic='market-data',
            group_id=f'indicator-calculator-{symbol}'
        )
        
        prices = []
        
        def process_message(message):
            if message.key == symbol:
                prices.append(message.value['price'])
                if len(prices) > window:
                    prices.pop(0)
                
                if len(prices) >= window:
                    # Calculate SMA
                    sma = sum(prices) / len(prices)
                    
                    # Calculate volatility
                    returns = [(prices[i] - prices[i-1]) / prices[i-1] 
                              for i in range(1, len(prices))]
                    vol = np.std(returns) * np.sqrt(252)
                    
                    # Publish indicator
                    indicator_message = {
                        'symbol': symbol,
                        'timestamp': datetime.utcnow().isoformat(),
                        'sma_20': sma,
                        'volatility': vol,
                        'latest_price': prices[-1]
                    }
                    self.client.publish('indicators', indicator_message, key=symbol)
        
        # Start consuming
        self.client.consume_loop('indicator-calculator', process_message)
    
    def detect_patterns(self, symbols):
        """Detect trading patterns in real-time."""
        consumer = self.client.create_consumer(
            topic='indicators',
            group_id='pattern-detector'
        )
        
        def process_message(message):
            if message.key in symbols:
                data = message.value
                # Simple pattern detection
                if data.get('sma_20') and data.get('latest_price'):
                    if data['latest_price'] > data['sma_20'] * 1.02:
                        alert = {
                            'symbol': message.key,
                            'pattern': 'BREAKOUT',
                            'price': data['latest_price'],
                            'sma': data['sma_20'],
                            'timestamp': data['timestamp']
                        }
                        self.client.publish('alerts', alert, key=message.key)
        
        self.client.consume_loop('pattern-detector', process_message)

class ExactlyOnceProcessor:
    """
    Implements exactly-once processing for financial transactions.
    """
    
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.client = KafkaClient(bootstrap_servers)
        self.producer = self.client.create_producer()
        self.state = {}
    
    def process_transaction(self, transaction_id, amount, account_id):
        """
        Process a transaction with exactly-once semantics.
        """
        # Check if already processed
        if transaction_id in self.state:
            return self.state[transaction_id]
        
        # Process the transaction
        result = self._execute_transaction(amount, account_id)
        
        # Store the result
        self.state[transaction_id] = result
        
        return result
    
    def _execute_transaction(self, amount, account_id):
        """Execute the actual transaction."""
        # Implementation would update database
        return {'status': 'success', 'amount': amount}

class EventStore:
    """
    An event store for financial transactions.
    """
    
    def __init__(self, kafka_client):
        self.client = kafka_client
        self.events = []
    
    def append_event(self, aggregate_id, event_type, data):
        """Append an event to the store."""
        event = {
            'aggregate_id': aggregate_id,
            'event_type': event_type,
            'data': data,
            'timestamp': datetime.utcnow().isoformat(),
            'version': len(self.events) + 1
        }
        
        # Store in Kafka
        self.client.publish('event-store', event, key=aggregate_id)
        self.events.append(event)
        
        return event['version']
    
    def get_events(self, aggregate_id):
        """Get all events for an aggregate."""
        consumer = self.client.create_consumer(
            topic='event-store',
            group_id=f'event-reader-{aggregate_id}'
        )
        
        events = []
        for message in consumer:
            if message.key == aggregate_id:
                events.append(message.value)
        
        return sorted(events, key=lambda e: e['version'])

class AccountAggregate:
    """
    An aggregate that reconstructs state from events.
    """
    
    def __init__(self, account_id, event_store):
        self.account_id = account_id
        self.event_store = event_store
        self.balance = 0.0
        self.status = 'ACTIVE'
        self.rebuild()
    
    def rebuild(self):
        """Rebuild state from events."""
        events = self.event_store.get_events(self.account_id)
        for event in events:
            self.apply_event(event)
    
    def apply_event(self, event):
        """Apply an event to update state."""
        if event['event_type'] == 'ACCOUNT_CREATED':
            self.balance = 0.0
            self.status = 'ACTIVE'
        elif event['event_type'] == 'DEPOSIT':
            self.balance += event['data']['amount']
        elif event['event_type'] == 'WITHDRAWAL':
            self.balance -= event['data']['amount']
        elif event['event_type'] == 'TRADE_EXECUTED':
            self.balance += event['data']['pnl']
    
    def deposit(self, amount):
        """Process a deposit."""
        self.event_store.append_event(
            self.account_id,
            'DEPOSIT',
            {'amount': amount}
        )
        self.balance += amount
    
    def withdraw(self, amount):
        """Process a withdrawal."""
        if self.balance < amount:
            raise ValueError("Insufficient funds")
        
        self.event_store.append_event(
            self.account_id,
            'WITHDRAWAL',
            {'amount': amount}
        )
        self.balance -= amount
    
    def execute_trade(self, pnl):
        """Execute a trade and update balance."""
        self.event_store.append_event(
            self.account_id,
            'TRADE_EXECUTED',
            {'pnl': pnl}
        )
        self.balance += pnl

class PriceAlertSystem:
    """
    Real-time price alert system using Kafka streams.
    """
    
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.client = KafkaClient(bootstrap_servers)
        self.alerts = {}
    
    def set_alert(self, symbol, threshold_price, condition='above'):
        """
        Set a price alert.
        condition: 'above' or 'below'
        """
        self.alerts[f"{symbol}_{condition}"] = {
            'symbol': symbol,
            'threshold': threshold_price,
            'condition': condition
        }
    
    def start_monitoring(self):
        """Start monitoring price updates."""
        consumer = self.client.create_consumer(
            topic='market-data',
            group_id='price-alerts'
        )
        
        def process_message(message):
            data = message.value
            symbol = data['symbol']
            price = data['price']
            
            for alert_key, alert in self.alerts.items():
                if alert['symbol'] != symbol:
                    continue
                
                if alert['condition'] == 'above' and price > alert['threshold']:
                    self.trigger_alert(symbol, price, alert['threshold'], 'above')
                elif alert['condition'] == 'below' and price < alert['threshold']:
                    self.trigger_alert(symbol, price, alert['threshold'], 'below')
        
        self.client.consume_loop('price-alerts', process_message)
    
    def trigger_alert(self, symbol, price, threshold, condition):
        """Trigger and publish an alert."""
        alert_message = {
            'symbol': symbol,
            'price': price,
            'threshold': threshold,
            'condition': condition,
            'timestamp': datetime.utcnow().isoformat()
        }
        
        self.client.publish('alerts', alert_message, key=symbol)
        
        print(f"ALERT: {symbol} is {condition} ${threshold:.2f} at ${price:.2f}")

class DataEnrichmentPipeline:
    """
    Enriches raw market data with additional information.
    """
    
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.client = KafkaClient(bootstrap_servers)
        self.enrichment_data = {}
    
    def enrich_market_data(self, market_data):
        """Enrich market data with additional information."""
        symbol = market_data['symbol']
        
        # Add company information
        if symbol in self.enrichment_data:
            company_info = self.enrichment_data[symbol]
            market_data['sector'] = company_info.get('sector')
            market_data['market_cap'] = company_info.get('market_cap')
            market_data['pe_ratio'] = company_info.get('pe_ratio')
        
        # Calculate day range
        if symbol in self.daily_stats:
            market_data['day_high'] = self.daily_stats[symbol]['high']
            market_data['day_low'] = self.daily_stats[symbol]['low']
            market_data['day_change'] = (market_data['price'] - self.daily_stats[symbol]['open']) / self.daily_stats[symbol]['open']
        
        return market_data
    
    def process_stream(self):
        """Process the market data stream with enrichment."""
        consumer = self.client.create_consumer(
            topic='market-data',
            group_id='data-enricher'
        )
        
        def process_message(message):
            enriched = self.enrich_market_data(message.value)
            self.client.publish('enriched-data', enriched, key=message.key)
        
        self.client.consume_loop('data-enricher', process_message)

class EventDrivenTradingSystem:
    """
    A complete event-driven trading system.
    """
    
    def __init__(self, bootstrap_servers=['localhost:9092']):
        self.client = KafkaClient(bootstrap_servers)
        self.producer = self.client.create_producer()
        self.strategies = {}
        self.risk_manager = RiskManager()
    
    def add_strategy(self, symbol, strategy):
        """Add a trading strategy for a symbol."""
        self.strategies[symbol] = strategy
    
    def start(self):
        """Start the trading system."""
        # Start market data consumer
        market_consumer = self.client.create_consumer(
            topic='market-data',
            group_id='trading-system'
        )
        
        def handle_market_data(message):
            symbol = message.key
            price = message.value['price']
            
            if symbol in self.strategies:
                # Execute strategy
                signal = self.strategies[symbol].on_price(price)
                
                if signal:
                    # Check risk limits
                    if self.risk_manager.check_limits(symbol, signal):
                        # Publish order
                        order = {
                            'symbol': symbol,
                            'side': signal['side'],
                            'quantity': signal['quantity'],
                            'price': signal.get('price', None),
                            'order_type': signal.get('order_type', 'MARKET'),
                            'strategy_id': signal.get('strategy_id')
                        }
                        self.client.publish('orders', order, key=symbol)
        
        # Start consuming
        self.client.consume_loop('trading-system', handle_market_data)

class RiskManager:
    """
    Manages risk for the trading system.
    """
    
    def __init__(self):
        self.position_limits = {}
        self.max_position = 10000
        self.daily_loss_limit = 100000
        self.daily_loss = 0
    
    def check_limits(self, symbol, signal):
        """Check if the signal violates any risk limits."""
        # Check position limits
        if abs(self.get_position(symbol) + signal['quantity']) > self.max_position:
            return False
        
        # Check daily loss limit
        estimated_pnl = signal['quantity'] * signal.get('price', 0)
        if self.daily_loss + estimated_pnl < -self.daily_loss_limit:
            return False
        
        return True
    
    def get_position(self, symbol):
        """Get current position for a symbol."""
        # Implementation would query the database
        return 0

class RealTimeDashboard:
    """
    Updates a dashboard in real-time using WebSockets.
    """
    
    def __init__(self, kafka_client):
        self.client = kafka_client
        self.connections = {}
    
    def broadcast_price(self, symbol, price):
        """Broadcast a price update to all connected clients."""
        message = {
            'type': 'price_update',
            'symbol': symbol,
            'price': price,
            'timestamp': datetime.utcnow().isoformat()
        }
        
        # Publish to Kafka for WebSocket server
        self.client.publish('dashboard-updates', message, key=symbol)
    
    def broadcast_portfolio(self, account_id, portfolio_data):
        """Broadcast portfolio updates."""
        message = {
            'type': 'portfolio_update',
            'account_id': account_id,
            'data': portfolio_data,
            'timestamp': datetime.utcnow().isoformat()
        }
        
        self.client.publish('portfolio-updates', message, key=account_id)

# WebSocket server (simplified)
class DashboardWebSocket:
    def __init__(self, kafka_client):
        self.client = kafka_client
    
    def start_server(self):
        """Start the WebSocket server."""
        consumer = self.client.create_consumer(
            topic='dashboard-updates',
            group_id='websocket-server'
        )
        
        def process_message(message):
            # Broadcast to all connected WebSocket clients
            self._broadcast_to_clients(message.value)
        
        self.client.consume_loop('websocket-server', process_message)