refactored data pipeline to async - 3x faster market data processing

# Old synchronous approach
def fetch_market_data(symbols):
    results = []
    for symbol in symbols:
        data = fetch_from_api(symbol)  # Blocks here
        results.append(data)
    return results

# With 10 symbols, takes 10 * 180ms = 1,800ms total

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

class AsyncMarketData:
    """Async market data fetcher with connection pooling"""

    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.base_url = "https://api.polygon.io"
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session = None

    async def __aenter__(self):
        """Context manager entry"""
        connector = aiohttp.TCPConnector(
            limit=100,  # Max connections
            ttl_dns_cache=300  # DNS cache TTL
        )
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=aiohttp.ClientTimeout(total=30)
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Context manager exit"""
        if self.session:
            await self.session.close()

    async def fetch_quote(self, symbol: str) -> Dict[str, Any]:
        """Fetch single quote with rate limiting"""
        async with self.semaphore:  # Rate limit concurrent requests
            url = f"{self.base_url}/v2/last/trade/{symbol}"
            params = {"apiKey": self.api_key}

            try:
                async with self.session.get(url, params=params) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        return {
                            'symbol': symbol,
                            'price': data['results']['p'],
                            'size': data['results']['s'],
                            'timestamp': data['results']['t'],
                            'success': True
                        }
                    else:
                        return {
                            'symbol': symbol,
                            'success': False,
                            'error': f"HTTP {resp.status}"
                        }
            except Exception as e:
                return {
                    'symbol': symbol,
                    'success': False,
                    'error': str(e)
                }

    async def fetch_multiple(self, symbols: List[str]) -> List[Dict[str, Any]]:
        """Fetch multiple quotes concurrently"""
        tasks = [self.fetch_quote(symbol) for symbol in symbols]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # Handle any exceptions
        clean_results = []
        for result in results:
            if isinstance(result, Exception):
                clean_results.append({
                    'success': False,
                    'error': str(result)
                })
            else:
                clean_results.append(result)

        return clean_results


class StrategyDataFeed:
    """High-level interface for strategy data needs"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.cache = {}  # Simple in-memory cache

    async def get_current_prices(self, symbols: List[str]) -> Dict[str, float]:
        """Get current prices for list of symbols"""

        # Check cache first (1 second TTL)
        now = time.time()
        cached = {}
        to_fetch = []

        for symbol in symbols:
            if symbol in self.cache:
                cached_time, cached_price = self.cache[symbol]
                if now - cached_time < 1.0:  # Cache valid
                    cached[symbol] = cached_price
                    continue
            to_fetch.append(symbol)

        # Fetch uncached symbols
        if to_fetch:
            async with AsyncMarketData(self.api_key) as fetcher:
                results = await fetcher.fetch_multiple(to_fetch)

                for result in results:
                    if result['success']:
                        symbol = result['symbol']
                        price = result['price']
                        cached[symbol] = price
                        self.cache[symbol] = (now, price)

        return cached


# Usage in strategy
class TradingStrategy:
    def __init__(self):
        self.data_feed = StrategyDataFeed(
            api_key=os.getenv("POLYGON_API_KEY")
        )

    async def evaluate_signals(self):
        """Main strategy loop"""
        symbols = ['SPY', 'QQQ', 'IWM', 'TLT', 'GLD',
                   'VXX', 'XLF', 'XLE', 'XLK', 'XLV']

        # Fetch all prices concurrently
        start = time.time()
        prices = await self.data_feed.get_current_prices(symbols)
        elapsed = time.time() - start

        print(f"Fetched {len(prices)} prices in {elapsed*1000:.1f}ms")

        # Strategy logic here
        for symbol, price in prices.items():
            signal = self.calculate_signal(symbol, price)
            if signal:
                await self.execute_trade(symbol, signal)


# Main event loop
async def main():
    strategy = TradingStrategy()

    while True:
        await strategy.evaluate_signals()
        await asyncio.sleep(1)  # Run every second


if __name__ == "__main__":
    asyncio.run(main())

self.semaphore = asyncio.Semaphore(10)  # Max 10 concurrent

timeout=aiohttp.ClientTimeout(total=30)

return_exceptions=True  # Don't fail entire batch if one symbol fails

# Old sync
prices = self.data_feed.get_prices(symbols)

# New async
prices = await self.data_feed.get_current_prices(symbols)

# Check redis first (0.3ms)
# Then async fetch if cache miss (220ms)
# Then cache result for next request