Three Local LLMs Walk Into a Pipeline: An LBD Deep‑Dive

Three local AI models get the same systems-engineering homework. The results are... educational.

When you feed the same complex systems-engineering prompt into three very different local LLMs, you learn a lot—not just about the models, but about how modern AI handles the difference between understanding an architecture and actually implementing one.

The prompt itself was non-trivial: build a fully asynchronous Python data-ingestion pipeline using asyncio, httpx, Pydantic v2, a custom token-bucket rate limiter, exponential-backoff retry logic, a bounded queue, and a worker pool that maintains thread-safe moving averages.

At first glance, there was a clear winner (at least according to Gemini's quick review). Then I looked more closely at the winner.

And that's where this little experiment got considerably more interesting.

Below, we break down how each model responded, what Gemini observed, what I found when I checked the generated code more closely, and what the results actually say about local LLM engineering.

---

📌 The Prompt (Summary)

The prompt required:

• Strict Pydantic v2 validation models
• A custom async token‑bucket rate limiter
• A retry decorator with exponential backoff + jitter
• An asyncio.Queue with bounded size
• A worker pool performing thread‑safe moving‑average aggregation
• No placeholders—full, functional code

This is the kind of prompt that exposes whether a model truly understands systems architecture, not just syntax.

---

🏆 Model #1 — Devstral‑24B

✔️ Where Devstral Shines

• Best structural imagination of the three
• Implements regex ticker validation (neither Qwen did)
• Clean separation of concerns: MockAPIClient, MovingAverageCalculator, worker pipeline
• Implements a proper sentinel‑value shutdown pattern

❌ Where Devstral Fails

• Broken rate limiter — uses async with rate_limiter: but never implements __aenter__ / __aexit__
• Calls self.last_checked.loop.time() on an asyncio.Event (invalid)
• Uses deprecated datetime.utcnow()
• Uses .dict() instead of .model_dump() for Pydantic v2

🧩 Verdict

Devstral produced the most ambitious architecture, but hallucinated several async primitives that would crash immediately. It “thinks big,” but sometimes invents APIs that don’t exist.

---

🏆 Model #2 — Qwen2.5‑Coder‑14B-instruct-q4_K_M

✔️ Strengths

• Clean, readable code
• Implements a correct async token bucket
• Implements a retry decorator with exponential backoff + jitter

❌ Weaknesses

• Completely bypasses the queue‑worker architecture
• Turns the pipeline into a simple “fetch URL → put into queue” loop
• Does not simulate ticks; instead polls a mock URL repeatedly
• Moving average logic is oversimplified and not thread‑safe

🧩 Verdict

The Qwen 2.5 14B coder model writes clean code but misses the architectural intent. It treats the pipeline like a sequential script with a queue bolted on as an afterthought.

---

🏆 Model #3 — Qwen3.5-9B

✔️ Where Qwen3.5 Shines

• Best overall architectural match of the three
• Implements the requested producer → bounded queue → worker flow
• Uses deque and asyncio.Lock() for shared moving-average state
• Uses timezone-aware UTC timestamps
• Attempts to handle transient failures with exponential backoff and jitter
• Implements a recognizable asynchronous token-bucket design

❌ Where Qwen3.5 Fails

• The generated code is not actually production-ready.
• TimeoutException is referenced but never defined or imported.
• _make_robust_api_call() is called with a limiter keyword argument that its function signature does not accept.
• A synchronous httpx.Client is used as though it were asynchronous, including await client.post(...).
• The HTTP client is given no apparent base_url, while the request uses a relative URL.
• The retry decorator is defined as an async def and is never actually used; retry logic is duplicated inline instead.
• Workers are cancelled rather than being cleanly drained after the producer finishes, so queued work can potentially be abandoned.

🧩 Verdict

Qwen3.5-9B produced the best-looking architecture of the three—and, importantly, it was the model whose structure most closely matched what the prompt actually requested.

But “best architecture” turned out not to mean “production-ready code.” A closer inspection of the raw output revealed several implementation-level errors that would prevent this code from working as written.

Which creates an amusing problem for the experiment: the model that looked best to another AI evaluator wasn't actually the winner we initially thought it was.

---

🤖 Gemini’s Comparative Analysis

Gemini evaluated the three outputs and reached a fairly straightforward conclusion:

1. Qwen3.5-9B showed the strongest structural reasoning and best matched the requested pipeline architecture.
2. Devstral-24B showed the strongest architectural imagination, but contained serious mistakes in its custom async primitives.
3. Qwen2.5-Coder-14B produced relatively clean code, but missed too much of the requested architecture.

Gemini's summary of Qwen3.5-9B was particularly enthusiastic:

“Despite having fewer parameters, its reasoning capability, structural awareness, and adherence to production patterns are substantially better.”

And that assessment is not unreasonable. Looking only at the architecture, Qwen3.5 really does come out ahead.

But then I did something that is occasionally useful when evaluating AI-generated code:

I read the code.

And I found several problems Gemini had missed.

🤦 The Evaluator Gets Evaluated

The Qwen3.5 output contains an undefined TimeoutException, calls _make_robust_api_call() with a limiter argument that the function doesn't accept, uses synchronous httpx.Client methods as though they were awaitable, and sends a relative URL without an apparent base URL. The retry decorator is also defined but isn't actually used by the pipeline.

In other words, the code that Gemini described as the most “production-ready” would not simply need polishing. It would fail at runtime in multiple places.

That doesn't make Gemini's architectural assessment useless. Quite the opposite: the architectural assessment was probably directionally correct. What it demonstrates is something more subtle:

An LLM can be very good at recognizing good architecture while still being bad at verifying whether the resulting code actually works.

And apparently, another LLM can make the same mistake when reviewing it.

---

📊 Final Ranking

1. 🥇 Qwen3.5-9B — Best match to the requested architecture, but with significant implementation bugs
2. 🥈 Devstral-24B — Most ambitious design, undermined by broken async primitives
3. 🥉 Qwen2.5-Coder-14B — Cleanest-looking implementation in places, but misses too much of the actual architectural requirement

There is an important asterisk next to that first-place finish:

None of the three produced code I would ship unchanged.

Qwen3.5 wins because it came closest to understanding the whole system. It does not win because its generated implementation is actually production-ready.

---

💬 Closing Thoughts

This test wasn't really about syntax. It was about systems reasoning.

The prompt forced each model to juggle concurrency, validation, rate limiting, retry logic, bounded queues, worker coordination, and stateful aggregation. The interesting part wasn't simply which model wrote the most code. It was which model appeared to understand how all those pieces were supposed to fit together.

On that measure, Qwen3.5-9B won this particular contest. Its architecture was the closest match to the requested design, despite having fewer parameters than Qwen2.5-Coder-14B and Devstral-24B.

But the second lesson may be more important than the first.

Good-looking AI-generated architecture is not the same thing as working software.

Qwen3.5's code was good enough to convince Gemini—and initially me—that it was the most production-ready implementation. A closer inspection exposed several basic runtime and API-contract errors. Devstral had a similar problem in a different form: it understood the shape of the architecture but invented async APIs that don't exist. Qwen2.5-Coder was more conservative, but in doing so largely sidestepped the architecture the prompt was asking for.

So perhaps the useful result isn't “Qwen3.5-9B is the best local coding model.”

It's this:

Local LLMs are getting remarkably good at producing plausible systems. We're still not at the point where “plausible” means “verified.”

And apparently, neither is the AI evaluating the AI.

That's probably a more useful conclusion than declaring a universal winner from one coding exercise.

---

📎 Appendix: Code Blocks (plus the prompt)

The Prompt:

Write an asynchronous Python data-ingestion pipeline using asyncio and httpx that consumes a simulated stream of high-frequency financial tick payloads.

Requirements:

Use Pydantic v2 to define strict validation models for incoming payloads (timestamp, ticker symbol, float price, volume, and a nested metadata dictionary).

Implement a custom asynchronous token-bucket rate limiter to throttle outgoing requests to an external mock API.

Implement a retry decorator with exponential backoff and jitter for transient network exceptions.

Process the stream concurrently using an asyncio.Queue with a bounded size, feeding a worker pool that aggregates moving averages in-memory thread-safely.

Do not use placeholder comments (# TODO); write out the complete, functional implementation code including imports, exception handling, and a main execution loop.

Below are the three raw outputs exactly as generated by the models.

Devstral‑24B Output

import asyncio
import httpx
import random
from datetime import datetime, timedelta
from collections import defaultdict
from functools import wraps
from pydantic import BaseModel, Field, ValidationError
from typing import Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Pydantic models
class Metadata(BaseModel):
    exchange: str
    source: str

class TickPayload(BaseModel):
    timestamp: datetime
    ticker: str = Field(..., regex=r'^[A-Z]+$')
    price: float
    volume: int
    metadata: Metadata

# Token bucket rate limiter
class TokenBucket:
    def __init__(self, tokens: int, refill_rate: float):
        self.tokens = tokens
        self.refill_rate = refill_rate
        self.last_checked = asyncio.Event()
        self.last_checked.set()

    async def consume(self, amount: int = 1):
        await self.last_checked.wait()
        if self.tokens < amount:
            sleep_time = (amount - self.tokens) / self.refill_rate
            logger.info(f"Rate limited. Sleeping for {sleep_time:.2f} seconds")
            await asyncio.sleep(sleep_time)
            self.tokens = min(self.tokens + sleep_time * self.refill_rate, self.tokens)
        self.tokens -= amount
        self.last_checked.clear()
        asyncio.create_task(self._refill())

    async def _refill(self):
        now = asyncio.get_event_loop().time()
        time_since_last_check = now - self.last_checked.loop.time()
        self.tokens = min(self.tokens + time_since_last_check * self.refill_rate, self.tokens)
        self.last_checked.set()

# Retry decorator with exponential backoff and jitter
def retry(exceptions, max_retries=5, backoff_factor=2):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            retries = 0
            while retries <= max_retries:
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    if retries == max_retries:
                        raise e
                    sleep_time = (backoff_factor ** retries) + (random.random() * backoff_factor ** retries)
                    logger.warning(f"Retrying in {sleep_time:.2f} seconds: {e}")
                    await asyncio.sleep(sleep_time)
                    retries += 1
        return wrapper
    return decorator

# Mock API client
class MockAPIClient:
    async def send_data(self, payload: Dict[str, Any]):
        # Simulate network delay
        await asyncio.sleep(random.uniform(0.05, 0.15))
        # Simulate random failures
        if random.random() < 0.05:
            raise httpx.NetworkError("Simulated network error")

# In-memory moving average calculator
class MovingAverageCalculator:
    def __init__(self, window_size: int = 10):
        self.window_size = window_size
        self.prices = defaultdict(lambda: [])
        self.lock = asyncio.Lock()

    async def update(self, ticker: str, price: float):
        async with self.lock:
            self.prices[ticker].append(price)
            if len(self.prices[ticker]) > self.window_size:
                self.prices[ticker].pop(0)

    async def get_average(self, ticker: str) -> float:
        async with self.lock:
            if not self.prices[ticker]:
                return 0.0
            return sum(self.prices[ticker]) / len(self.prices[ticker])

# Data ingestion pipeline
async def data_ingestion_pipeline(tick_queue: asyncio.Queue, api_client: MockAPIClient, rate_limiter: TokenBucket, moving_avg_calculator: MovingAverageCalculator):
    while True:
        payload = await tick_queue.get()
        if payload is None:  # Sentinel value to stop workers
            break

        try:
            tick = TickPayload(**payload)
            logger.info(f"Processing tick: {tick}")

            # Update moving average
            await moving_avg_calculator.update(tick.ticker, tick.price)

            # Send to external API with rate limiting
            async with rate_limiter:
                await rate_limiter.consume()

            @retry((httpx.NetworkError,), max_retries=5)
            async def send_with_retry():
                await api_client.send_data(tick.dict())

            await send_with_retry()

        except ValidationError as e:
            logger.error(f"Validation error: {e}")
        except Exception as e:
            logger.error(f"Error processing tick: {e}")
        finally:
            tick_queue.task_done()

# Simulated stream generator
async def generate_tick_stream(tick_queue: asyncio.Queue, tick_rate: float = 10.0):
    while True:
        tick = {
            "timestamp": datetime.utcnow(),
            "ticker": random.choice(["AAPL", "GOOGL", "AMZN", "MSFT"]),
            "price": round(random.uniform(100, 200), 2),
            "volume": random.randint(100, 1000),
            "metadata": {
                "exchange": "NASDAQ",
                "source": "Simulator"
            }
        }
        await tick_queue.put(tick)
        await asyncio.sleep(1 / tick_rate)

async def main():
    tick_queue = asyncio.Queue(max_size=100)

    # Initialize components
    api_client = MockAPIClient()
    rate_limiter = TokenBucket(tokens=5, refill_rate=1)  # 5 requests per second
    moving_avg_calculator = MovingAverageCalculator(window_size=10)

    # Start stream generator
    asyncio.create_task(generate_tick_stream(tick_queue))

    # Start worker pool
    workers = [asyncio.create_task(data_ingestion_pipeline(tick_queue, api_client, rate_limiter, moving_avg_calculator)) for _ in range(5)]

    # Periodically print moving averages
    try:
        while True:
            await asyncio.sleep(5)
            for ticker in ["AAPL", "GOOGL", "AMZN", "MSFT"]:
                avg = await moving_avg_calculator.get_average(ticker)
                logger.info(f"Moving average for {ticker}: {avg:.2f}")
    except asyncio.CancelledError:
        pass
    finally:
        # Clean up
        for _ in range(5):  # Number of workers
            await tick_queue.put(None)
        await asyncio.gather(*workers)

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

---

Qwen2.5‑Coder‑14B Output

import asyncio
import httpx
import logging
from pydantic.v2 import BaseModel, ValidationError
from decimal import Decimal
from typing import Dict, List, Optional
from functools import wraps

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Pydantic models for validation
class Metadata(BaseModel):
    source: str
    provider: str

class TickPayload(BaseModel):
    timestamp: int
    ticker: str
    price: Decimal
    volume: int
    metadata: Metadata

# Asynchronous token bucket rate limiter
class AsyncTokenBucket:
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate
        self.lock = asyncio.Lock()
        self.last_refill = asyncio.get_running_loop().time()

    async def refill(self):
        now = asyncio.get_running_loop().time()
        elapsed = now - self.last_refill
        if elapsed > 0:
            self.tokens = min(self.tokens + elapsed * self.refill_rate, self.capacity)
            self.last_refill = now

    async def consume(self, tokens: int) -> None:
        async with self.lock:
            await self.refill()
            while self.tokens < tokens:
                await asyncio.sleep((tokens - self.tokens) / self.refill_rate)
                await self.refill()
            self.tokens -= tokens

# Exponential backoff with jitter
def exponential_backoff_jitter(max_attempts: int, base_delay: float = 1.0, max_delay: float = 60.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            attempt = 0
            while attempt < max_attempts:
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    logger.error(f"Attempt {attempt + 1} failed: {e}")
                    attempt += 1
                    if attempt < max_attempts:
                        delay = base_delay * (2 ** (attempt - 1)) + (0.2 * (2 ** (attempt - 1)))  # Add jitter
                        delay = min(delay, max_delay)
                        await asyncio.sleep(delay)
            raise Exception("Max attempts reached")
        return wrapper
    return decorator

# Moving average calculator
class MovingAverage:
    def __init__(self, window_size: int):
        self.window_size = window_size
        self.values = []

    def add_value(self, value: float):
        self.values.append(value)
        if len(self.values) > self.window_size:
            self.values.pop(0)

    def get_average(self) -> float:
        if not self.values:
            return 0.0
        return sum(self.values) / len(self.values)

# Data ingestion pipeline
async def ingest_stream(queue: asyncio.Queue, rate_limiter: AsyncTokenBucket):
    async with httpx.AsyncClient() as client:
        url = "https://mockapi.com/ticks"
        while True:
            try:
                await rate_limiter.consume(1)
                response = await client.get(url)
                response.raise_for_status()
                tick_data = response.json()
                await queue.put(tick_data)
            except Exception as e:
                logger.error(f"Error fetching tick data: {e}")
                await asyncio.sleep(10)  # Retry after a delay

async def process_tick(queue: asyncio.Queue, moving_averages: Dict[str, MovingAverage]):
    while True:
        tick_data = await queue.get()
        try:
            tick = TickPayload(**tick_data)
            logger.info(f"Processing tick: {tick}")
            if tick.ticker not in moving_averages:
                moving_averages[tick.ticker] = MovingAverage(window_size=10)
            moving_averages[tick.ticker].add_value(tick.price)
            average_price = moving_averages[tick.ticker].get_average()
            logger.info(f"Moving average for {tick.ticker}: {average_price}")
        except ValidationError as e:
            logger.error(f"Validation error: {e}")
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=1000)
    rate_limiter = AsyncTokenBucket(capacity=10, refill_rate=1.0)
    moving_averages = {}

    # Start the ingestor
    asyncio.create_task(ingest_stream(queue, rate_limiter))

    # Start the processor pool
    num_workers = 5
    for _ in range(num_workers):
        asyncio.create_task(process_tick(queue, moving_averages))

    # Wait for all tasks to complete
    await queue.join()

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

---

Qwen3.5‑9B Output

import asyncio
import time
import random
from collections import deque
from datetime import datetime, timezone
from functools import wraps
from typing import Callable, Optional, Any

import httpx
from pydantic import BaseModel, Field


# -----------------------------------------------------------------------------
# 1. Pydantic V2 Models for Strict Validation
# -----------------------------------------------------------------------------

class Meta(BaseModel):
    """Nested metadata dictionary structure."""
    source: str = Field(default="internal-sim", description="Origin of the tick")
    exchange_latency_ms: Optional[int] = Field(None, ge=0, description="Optional latency info")

class Tick(BaseModel):
    """High-frequency financial tick payload."""
    timestamp: datetime
    ticker: str
    price: float = Field(ge=0.0, description="Current asset price")
    volume: float = Field(ge=0.0, description="Trade volume")
    metadata: Meta = Field(default_factory=Meta)


# -----------------------------------------------------------------------------
# 2. Async Token Bucket Rate Limiter
# -----------------------------------------------------------------------------

class AsyncTokenBucket:
    """Custom asynchronous token bucket implementation for rate limiting."""

    def __init__(self, capacity: int = 10, refill_rate: float = 5.0):
        """
        Args:
            capacity: Max tokens in bucket.
            refill_rate: Tokens added per second.
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = float(capacity)
        self.last_refill_time = time.monotonic()

    async def acquire(self, tokens: int = 1) -> None:
        """Acquire specified number of tokens. Wait if insufficient."""
        while True:
            now = time.monotonic()
            elapsed = now - self.last_refill_time
            # Calculate current available capacity, capped at bucket max
            available = min(self.capacity, self.tokens + elapsed * self.refill_rate)

            if available >= tokens:
                # Successfully acquired
                self.last_refill_time = now
                self.tokens = available - tokens
                break

            # Need to wait. Calculate time to acquire enough tokens.
            sleep_time = (tokens - available) / self.refill_rate
            await asyncio.sleep(sleep_time)


# -----------------------------------------------------------------------------
# 3. Retry Decorator with Exponential Backoff and Jitter
# -----------------------------------------------------------------------------

async def async_retry(
    func: Callable,
    *,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_backoff: float = 64.0,
):
    """
    Decorator for applying exponential backoff with jitter to async functions.
    Handles transient network exceptions.
    """
    
    @wraps(func)
    async def wrapper(*args, **kwargs):
        attempt = 0
        last_exception = None

        while attempt < max_retries:
            try:
                return await func(*args, **kwargs)
            except (httpx.HTTPError, httpx.ConnectError, TimeoutException) as e:
                last_exception = e
                # Transient error handling: wait before retry
                if attempt > 0:
                    # Calculate exponential backoff delay + jitter
                    jitter = random.uniform(0.0, base_delay * 0.5)
                    wait_time = min(base_delay * (2 ** (attempt - 1)), max_backoff) + jitter
                    
                    print(f"Transient error occurred ({e}). Retrying {attempt}/{max_retries} in {wait_time:.2f}s.")
                    await asyncio.sleep(wait_time)
                
                attempt += 1
        
        # If loop finishes without success, raise the last exception or a custom one
        if last_exception:
            raise last_exception

    return wrapper


# -----------------------------------------------------------------------------
# 4. Mock External API and Aggregation Logic
# -----------------------------------------------------------------------------

async def mock_external_api(ticker: str, payload: Any) -> dict:
    """Simulates an external API endpoint with random failures."""
    await asyncio.sleep(random.uniform(0.05, 0.2)) # Simulate network latency
    
    # Randomly fail to simulate transient errors (e.g., 10% chance)
    if random.random() < 0.1:
        raise httpx.ConnectError("Simulated connection reset by peer")

    return {
        "status": "processed",
        "ticker": ticker,
        "timestamp": payload.timestamp.isoformat(),
    }

class MovingAverageStore:
    """Thread-safe (asyncio safe) in-memory store for moving average calculations."""

    def __init__(self, window_size: int = 20):
        self.window_size = window_size
        # Dictionary of tickers -> deque of prices
        self._prices: dict[str, deque] = {}
        # Lock to protect shared state
        self._lock = asyncio.Lock()

    async def update(self, price: float, ticker: str) -> Optional[float]:
        """Update store with new price and return current MA."""
        async with self._lock:
            if ticker not in self._prices:
                # Initialize deque
                self._prices[ticker] = deque(maxlen=self.window_size)

            # Append new price (deque handles maxlen automatically)
            self._prices[ticker].append(price)

            prices_list = list(self._prices[ticker])
            
            if len(prices_list) > 0:
                current_ma = sum(prices_list) / len(prices_list)
                return current_ma
            
            return None

# -----------------------------------------------------------------------------
# 5. Pipeline Components
# -----------------------------------------------------------------------------

async def producer(queue: asyncio.Queue, duration_seconds: int):
    """Simulates a stream of high-frequency ticks."""
    end_time = time.monotonic() + duration_seconds
    
    while time.monotonic() < end_time:
        # Generate simulated tick data
        now = datetime.now(timezone.utc)
        ticker_pool = ["AAPL", "MSFT", "GOOGL", "AMZN", "TSLA"]
        ticker = random.choice(ticker_pool)
        
        # Slightly jitter price/volume
        base_price = random.uniform(100.0, 300.0)
        price = round(base_price * (1 + random.gauss(0, 0.005)), 4) # +/- 0.5% noise
        volume = round(random.uniform(1000, 50000), 2)

        try:
            tick = Tick(
                timestamp=now,
                ticker=ticker,
                price=price,
                volume=volume,
                metadata=Meta(source="generator")
            )
        except ValidationError as e:
            print(f"Validation Error: {e}")
            continue

        await queue.put(tick)
        
        # Time interval between ticks (1 tick per 10ms to simulate high freq)
        await asyncio.sleep(0.01)


async def worker(queue: asyncio.Queue, limiter: AsyncTokenBucket, ma_store: MovingAverageStore):
    """Worker task consuming from queue and sending to external API."""
    
    while True:
        try:
            # 1. Get item (blocks if queue empty)
            tick = await asyncio.wait_for(queue.get(), timeout=0.1)
            
            # 2. Rate Limiting
            await limiter.acquire()
            
            async with httpx.Client(timeout=httpx.Timeout(timeout=5.0, connect=2.0)) as client:
                # 3. Retry Logic is applied via decorator logic usually, 
                # but here we apply it manually to keep the worker clean or use a decorated func.
                # Let's apply retry logic inline for clarity within the loop
                
                # Attempt API Call with Retry logic wrapped locally
                try:
                    await _make_robust_api_call(limiter=None, client=client, payload=tick)
                except Exception as e:
                    print(f"Error processing tick {tick.ticker}: {e}")
                
            # 4. Aggregate Data (Safe inside lock via async method)
            await ma_store.update(tick.price, tick.ticker)
            
            # Release queue slot immediately
            queue.task_done()

        except asyncio.TimeoutError:
            continue


# Wrapper for API call with retry logic implemented as per requirement
async def _make_robust_api_call(payload: Any, client: httpx.Client):
    """Retryable API call using exponential backoff."""
    
    async def _try_send(tick):
        # Prepare payload
        resp = await client.post(f"/api/ticker/{tick.ticker}", json={
            "price": tick.price,
            "volume": tick.volume,
            "ts": tick.timestamp.isoformat(),
        })
        if not resp.is_success:
             raise httpx.HTTPError(response=resp)
        return True

    # Retry wrapper logic inline for the worker loop to be self-contained
    last_err = None
    attempt = 0
    
    while attempt < 5:
        try:
            await _try_send(payload)
            break
        except (httpx.HTTPError, ConnectionError, TimeoutException) as e:
            last_err = e
            attempt += 1
            if attempt >= 5:
                raise
            
            # Wait with jitter (simple implementation similar to decorator concept)
            delay = min(1.0 * (2 ** (attempt - 1)), 64.0) + random.uniform(0, 0.5)
            print(f"Retry {attempt}: waiting {delay}s for error: {str(last_err)[:20]}")
            await asyncio.sleep(delay)


async def main():
    """Main execution loop."""
    
    # Configuration
    QUEUE_SIZE = 50
    WORKERS = 4
    DURATION_SEC = 30
    
    # Initialize components
    queue = asyncio.Queue(maxsize=QUEUE_SIZE)
    limiter = AsyncTokenBucket(capacity=10, refill_rate=5.0) # 5 req/sec limit
    ma_store = MovingAverageStore(window_size=20)
    
    client = None
    
    async with httpx.AsyncClient() as http_client:
        await limiter.acquire(tokens=5) # Initial burst
        
        try:
            # Start Worker Pool
            workers = [
                asyncio.create_task(worker(queue, limiter, ma_store)) 
                for _ in range(WORKERS)
            ]
            
            # Start Producer
            producer_task = asyncio.create_task(producer(queue, duration_seconds=DURATION_SEC))
            
            # Wait for production to finish or cancellation
            await producer_task
            
            print("Stream ingestion finished. Workers exiting...")
            
        except asyncio.CancelledError:
            pass

        # Stop workers
        for w in workers:
            w.cancel()
        
        # Brief cleanup
        try:
            await asyncio.gather(*workers, return_exceptions=True)
        except (asyncio.CancelledError, Exception):
            pass

        # Print Aggregation Results
        print("\n--- Final Moving Average Stats ---")
        for ticker, prices in ma_store._prices.items():
            if len(prices) > 1:
                final_ma = sum(prices) / len(prices)
                print(f"{ticker}: Samples={len(prices)}, Avg={final_ma:.4f}, First={prices[0]:.4f}, Last={prices[-1]:.4f}")


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