using python async for real-time market data

def get_market_data():
    # Get data from multiple sources sequentially
    polygon_data = fetch_polygon()      # 200ms
    ib_data = fetch_interactive_brokers()  # 300ms
    greeks = calculate_greeks()         # 150ms

    return combine_data(polygon_data, ib_data, greeks)

# Total time: 650ms per symbol
# For 10 symbols: 6.5 seconds

import asyncio
import aiohttp

async def get_market_data():
    # Fetch everything in parallel
    polygon_task = fetch_polygon_async()
    ib_task = fetch_ib_async()
    greeks_task = calculate_greeks_async()

    # Wait for all to complete
    polygon_data, ib_data, greeks = await asyncio.gather(
        polygon_task,
        ib_task,
        greeks_task
    )

    return combine_data(polygon_data, ib_data, greeks)

# Total time: 300ms (max of the three)
# For 10 symbols: 1.2 seconds (with batching)

import asyncio
import aiohttp
from typing import List, Dict
import time

class AsyncMarketData:
    def __init__(self, polygon_key: str, ib_client):
        self.polygon_key = polygon_key
        self.ib_client = ib_client
        self.session = None

    async def __aenter__(self):
        # Create aiohttp session for connection pooling
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    async def fetch_polygon_quote(self, symbol: str) -> Dict:
        """Fetch real-time quote from Polygon"""
        url = f"https://api.polygon.io/v2/last/trade/{symbol}"
        params = {"apiKey": self.polygon_key}

        async with self.session.get(url, params=params) as response:
            return await response.json()

    async def fetch_ib_quote(self, symbol: str) -> Dict:
        """Fetch quote from Interactive Brokers"""
        # IB client already async via ib_insync
        contract = Stock(symbol, 'SMART', 'USD')
        ticker = await self.ib_client.reqTickersAsync(contract)
        return {
            'bid': ticker.bid,
            'ask': ticker.ask,
            'last': ticker.last
        }

    async def calculate_greeks(self, symbol: str, price: float) -> Dict:
        """Calculate option greeks asynchronously"""
        # Run CPU-intensive calculation in thread pool
        loop = asyncio.get_event_loop()
        greeks = await loop.run_in_executor(
            None,
            self._calculate_greeks_sync,
            symbol,
            price
        )
        return greeks

    def _calculate_greeks_sync(self, symbol: str, price: float) -> Dict:
        """Synchronous greeks calculation (CPU bound)"""
        # Black-Scholes calculation here
        # This runs in separate thread via run_in_executor
        import numpy as np
        from scipy.stats import norm

        # ... greeks math ...

        return {
            'delta': delta,
            'gamma': gamma,
            'theta': theta,
            'vega': vega
        }

    async def get_complete_data(self, symbols: List[str]) -> Dict[str, Dict]:
        """Fetch complete market data for multiple symbols"""
        tasks = []

        for symbol in symbols:
            # Create task for each symbol
            task = self._fetch_symbol_data(symbol)
            tasks.append(task)

        # Execute all tasks in parallel
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Handle results and exceptions
        data = {}
        for symbol, result in zip(symbols, results):
            if isinstance(result, Exception):
                print(f"Error fetching {symbol}: {result}")
                continue
            data[symbol] = result

        return data

    async def _fetch_symbol_data(self, symbol: str) -> Dict:
        """Fetch all data sources for single symbol"""
        # Parallel fetch from all sources
        polygon_task = self.fetch_polygon_quote(symbol)
        ib_task = self.fetch_ib_quote(symbol)

        polygon_data, ib_data = await asyncio.gather(
            polygon_task,
            ib_task
        )

        # Calculate greeks using IB price
        price = ib_data['last']
        greeks = await self.calculate_greeks(symbol, price)

        return {
            'symbol': symbol,
            'polygon': polygon_data,
            'ib': ib_data,
            'greeks': greeks,
            'timestamp': time.time()
        }

async def main():
    symbols = ['SPY', 'QQQ', 'IWM', 'AAPL', 'MSFT',
               'GOOGL', 'AMZN', 'TSLA', 'NVDA', 'META']

    async with AsyncMarketData(POLYGON_KEY, ib_client) as client:
        start = time.time()
        data = await client.get_complete_data(symbols)
        elapsed = time.time() - start

        print(f"Fetched {len(data)} symbols in {elapsed:.2f}s")

        for symbol, info in data.items():
            print(f"{symbol}: ${info['ib']['last']:.2f}")

# Run it
asyncio.run(main())

class StreamingMarketData:
    def __init__(self, polygon_key: str):
        self.polygon_key = polygon_key
        self.ws = None
        self.callbacks = {}

    async def connect(self):
        """Connect to Polygon websocket"""
        import websockets

        url = f"wss://socket.polygon.io/stocks"
        self.ws = await websockets.connect(url)

        # Authenticate
        await self.ws.send(f'{{"action":"auth","params":"{self.polygon_key}"}}')

        # Start processing messages
        asyncio.create_task(self._process_messages())

    async def subscribe(self, symbols: List[str], callback):
        """Subscribe to real-time updates"""
        # Subscribe to trades
        for symbol in symbols:
            msg = {
                "action": "subscribe",
                "params": f"T.{symbol}"
            }
            await self.ws.send(str(msg))
            self.callbacks[symbol] = callback

    async def _process_messages(self):
        """Process incoming websocket messages"""
        async for message in self.ws:
            data = json.loads(message)

            # Handle trade message
            if data[0].get('ev') == 'T':
                symbol = data[0]['sym']
                price = data[0]['p']
                size = data[0]['s']

                # Call registered callback
                if symbol in self.callbacks:
                    callback = self.callbacks[symbol]
                    await callback(symbol, price, size)

    async def close(self):
        if self.ws:
            await self.ws.close()

async def handle_trade(symbol, price, size):
    print(f"{symbol}: ${price:.2f} x {size}")

    # Update strategy state
    await strategy.on_price_update(symbol, price)

async def main():
    stream = StreamingMarketData(POLYGON_KEY)
    await stream.connect()

    # Subscribe to symbols
    symbols = ['SPY', 'QQQ', 'AAPL']
    await stream.subscribe(symbols, handle_trade)

    # Run forever
    await asyncio.Event().wait()

asyncio.run(main())

# Limit to 5 concurrent requests
semaphore = asyncio.Semaphore(5)

async def fetch_with_limit(symbol):
    async with semaphore:
        return await fetch_data(symbol)

try:
    data = await asyncio.wait_for(fetch_data(symbol), timeout=5.0)
except asyncio.TimeoutError:
    print(f"Timeout fetching {symbol}")

async def fetch_with_retry(symbol, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await fetch_data(symbol)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # exponential backoff

# BAD - blocks entire event loop
async def bad_example():
    result = some_blocking_api_call()  # freezes everything
    return result

# GOOD - run blocking call in thread
async def good_example():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, some_blocking_api_call)
    return result

# BAD - one failure kills everything
results = await asyncio.gather(task1, task2, task3)

# GOOD - handle exceptions per task
results = await asyncio.gather(
    task1,
    task2,
    task3,
    return_exceptions=True
)

for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"Task {i} failed: {result}")

# BAD - leaks connections
async def fetch():
    session = aiohttp.ClientSession()
    return await session.get(url)

# GOOD - use context manager
async def fetch():
    async with aiohttp.ClientSession() as session:
        return await session.get(url)