Add AI Analyst service: Ollama (local) + HuggingFace (HF Spaces)

New ai_analyst/ai_analyst.py service consumes trades/snapshots from
Kafka, builds a market-context prompt every 30 min, calls Ollama
(llama3.1:8b) locally or HuggingFace Inference API (Mistral-7B) as
fallback, and publishes insights to the ai_insights Kafka topic.

Dashboard consumes ai_insights and broadcasts via SSE. New full-width
AI Analyst panel displays timestamped insights with animated cards.

- ai_analyst/ai_analyst.py: new service (Ollama-first, HF fallback)
- ai_analyst/Dockerfile: minimal Python image for local compose
- shared/config.py: add AI_INSIGHTS_TOPIC
- dashboard/dashboard.py: consume ai_insights, SSE broadcast
- dashboard/templates/index.html: AI Analyst panel + JS handlers
- Dockerfile: copy ai_analyst.py into HF single-container image
- entrypoint.sh: create ai_insights topic, start ai_analyst.py
- docker-compose.yml: ai_analyst service with OLLAMA_HOST + HF_TOKEN

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Dockerfile

@@ -70,6 +70,9 @@ COPY fix-ui-client/fix-ui-client.py   /app/fix_ui/fix_ui_client.py
 COPY fix-ui-client/templates/         /app/fix_ui/templates/
 COPY client_hf.cfg                    /app/fix_ui/client_hf.cfg
 
+# AI Analyst service
+COPY ai_analyst/ai_analyst.py         /app/ai_analyst.py
+
 # ── Kafka KRaft configuration ─────────────────────────────────────────────────
 COPY kafka-kraft.properties           /opt/kafka/config/kraft/server.properties
 

ai_analyst/Dockerfile

@@ -0,0 +1,10 @@
+FROM python:3.11-slim
+
+RUN pip install --no-cache-dir kafka-python==2.0.2 requests==2.31.0
+
+WORKDIR /app
+COPY ai_analyst.py .
+
+ENV PYTHONPATH=/app
+
+CMD ["python", "-u", "ai_analyst.py"]

ai_analyst/ai_analyst.py

@@ -0,0 +1,212 @@
+import sys
+sys.path.insert(0, "/app")
+
+import threading, time, os, json, datetime, requests
+from collections import deque
+
+from shared.config import Config
+from shared.kafka_utils import create_producer, create_consumer
+
+# ── Config ─────────────────────────────────────────────────────────────────────
+OLLAMA_HOST    = os.getenv("OLLAMA_HOST", "")          # e.g. http://host.docker.internal:11434
+OLLAMA_MODEL   = os.getenv("OLLAMA_MODEL", "llama3.1:8b")
+HF_TOKEN       = os.getenv("HF_TOKEN", "")
+HF_MODEL       = os.getenv("HF_MODEL", "mistralai/Mistral-7B-Instruct-v0.2")
+ANALYSIS_INTERVAL = int(os.getenv("ANALYSIS_INTERVAL", "1800"))  # 30 min default
+
+# ── Rolling market data buffers ────────────────────────────────────────────────
+recent_trades     = deque(maxlen=200)
+latest_snapshots  = {}          # symbol -> snapshot dict
+lock              = threading.Lock()
+
+_running   = False
+_suspended = False
+
+# ── LLM call ──────────────────────────────────────────────────────────────────
+
+def call_llm(prompt: str) -> str | None:
+    """Try Ollama first, fall back to HuggingFace Inference API."""
+
+    # 1. Ollama (local)
+    if OLLAMA_HOST:
+        try:
+            resp = requests.post(
+                f"{OLLAMA_HOST}/api/chat",
+                json={
+                    "model":    OLLAMA_MODEL,
+                    "messages": [{"role": "user", "content": prompt}],
+                    "stream":   False,
+                },
+                timeout=90,
+            )
+            if resp.status_code == 200:
+                text = resp.json().get("message", {}).get("content", "").strip()
+                if text:
+                    print(f"[AI-Analyst] Insight via Ollama ({OLLAMA_MODEL})")
+                    return text
+            else:
+                print(f"[AI-Analyst] Ollama HTTP {resp.status_code}: {resp.text[:200]}")
+        except Exception as e:
+            print(f"[AI-Analyst] Ollama unreachable: {e}")
+
+    # 2. HuggingFace Inference API
+    if HF_TOKEN:
+        try:
+            url = f"https://api-inference.huggingface.co/models/{HF_MODEL}/v1/chat/completions"
+            resp = requests.post(
+                url,
+                headers={"Authorization": f"Bearer {HF_TOKEN}"},
+                json={
+                    "model":       HF_MODEL,
+                    "messages":    [{"role": "user", "content": prompt}],
+                    "max_tokens":  220,
+                    "temperature": 0.7,
+                },
+                timeout=45,
+            )
+            if resp.status_code == 200:
+                text = resp.json()["choices"][0]["message"]["content"].strip()
+                if text:
+                    print(f"[AI-Analyst] Insight via HuggingFace ({HF_MODEL})")
+                    return text
+            else:
+                print(f"[AI-Analyst] HF HTTP {resp.status_code}: {resp.text[:300]}")
+        except Exception as e:
+            print(f"[AI-Analyst] HF API error: {e}")
+
+    return None
+
+
+# ── Prompt builder ─────────────────────────────────────────────────────────────
+
+def build_prompt() -> str:
+    now     = datetime.datetime.now().strftime("%H:%M:%S")
+    cutoff  = time.time() - ANALYSIS_INTERVAL
+
+    with lock:
+        trades_snap = list(recent_trades)
+        snaps_snap  = dict(latest_snapshots)
+        session_str = ("ACTIVE" if _running and not _suspended
+                       else "SUSPENDED" if _suspended else "IDLE")
+
+    # Recent trades summary per symbol
+    recent = [t for t in trades_snap if float(t.get("timestamp", 0)) >= cutoff]
+    interval_label = f"{ANALYSIS_INTERVAL // 60} min" if ANALYSIS_INTERVAL >= 60 else f"{ANALYSIS_INTERVAL}s"
+    if recent:
+        by_sym: dict = {}
+        for t in recent:
+            sym = t.get("symbol", "?")
+            by_sym.setdefault(sym, []).append(t)
+        trade_lines = []
+        for sym, ts in sorted(by_sym.items()):
+            prices = [float(t.get("price", 0)) for t in ts]
+            vol    = sum(int(t.get("quantity") or t.get("qty") or 0) for t in ts)
+            trade_lines.append(
+                f"  {sym}: {len(ts)} trade(s), "
+                f"range {min(prices):.2f}–{max(prices):.2f}, "
+                f"vol {vol}, last {prices[-1]:.2f}"
+            )
+        trades_block = "\n".join(trade_lines)
+    else:
+        trades_block = "  No trades in the last interval"
+
+    # Order book snapshot
+    if snaps_snap:
+        book_lines = []
+        for sym, snap in sorted(snaps_snap.items()):
+            bid = snap.get("best_bid")
+            ask = snap.get("best_ask")
+            if bid and ask:
+                spread = float(ask) - float(bid)
+                book_lines.append(f"  {sym}: Bid {bid} / Ask {ask} (spread {spread:.2f})")
+            else:
+                book_lines.append(f"  {sym}: Bid {bid or '-'} / Ask {ask or '-'}")
+        book_block = "\n".join(book_lines)
+    else:
+        book_block = "  No order book data yet"
+
+    return f"""You are a concise financial market analyst for a simulated stock exchange.
+Time: {now} | Session: {session_str}
+
+Trades in the last {interval_label}:
+{trades_block}
+
+Order book (best bid/ask):
+{book_block}
+
+In 3–4 sentences analyse: activity level, notable price moves or volume spikes, market sentiment.
+Be specific and data-driven. No headers, no bullet points, plain prose only."""
+
+
+# ── Kafka consumer (market data) ──────────────────────────────────────────────
+
+def consume_market_data():
+    global _running, _suspended
+    consumer = create_consumer(
+        topics=[
+            Config.TRADES_TOPIC,
+            Config.SNAPSHOTS_TOPIC,
+            Config.CONTROL_TOPIC,
+        ],
+        group_id="ai-analyst",
+        component_name="AI-Analyst",
+        auto_offset_reset="latest",
+    )
+    for msg in consumer:
+        with lock:
+            if msg.topic == Config.TRADES_TOPIC:
+                recent_trades.append(msg.value)
+            elif msg.topic == Config.SNAPSHOTS_TOPIC:
+                snap = msg.value
+                sym  = snap.get("symbol")
+                if sym:
+                    latest_snapshots[sym] = snap
+            elif msg.topic == Config.CONTROL_TOPIC:
+                cmd = msg.value.get("command", "")
+                if cmd == "start":
+                    _running   = True
+                    _suspended = False
+                elif cmd in ("end", "stop"):
+                    _running   = False
+                elif cmd == "suspend":
+                    _suspended = True
+                elif cmd == "resume":
+                    _suspended = False
+
+
+# ── Analysis loop ──────────────────────────────────────────────────────────────
+
+def analysis_loop(producer):
+    print(f"[AI-Analyst] Analysis loop started (interval={ANALYSIS_INTERVAL}s)")
+    if OLLAMA_HOST:
+        print(f"[AI-Analyst] Ollama: {OLLAMA_HOST}  model: {OLLAMA_MODEL}")
+    if HF_TOKEN:
+        print(f"[AI-Analyst] HuggingFace fallback: model={HF_MODEL}")
+    if not OLLAMA_HOST and not HF_TOKEN:
+        print("[AI-Analyst] WARNING: neither OLLAMA_HOST nor HF_TOKEN configured — no insights will be generated")
+
+    while True:
+        time.sleep(ANALYSIS_INTERVAL)
+
+        with lock:
+            active = _running and not _suspended
+
+        if not active:
+            continue
+
+        prompt = build_prompt()
+        text   = call_llm(prompt)
+
+        if text:
+            insight = {"text": text, "timestamp": time.time()}
+            producer.send(Config.AI_INSIGHTS_TOPIC, insight)
+            producer.flush()
+            print(f"[AI-Analyst] Published insight ({len(text)} chars)")
+
+
+# ── Entry point ─────────────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    producer = create_producer(component_name="AI-Analyst")
+    threading.Thread(target=consume_market_data, daemon=True).start()
+    analysis_loop(producer)

dashboard/dashboard.py

@@ -13,6 +13,7 @@ app.config["TEMPLATES_AUTO_RELOAD"] = True
 
 # Shared state
 orders, bbos, trades_cache = [], {}, []
+ai_insights_cache = []
 lock = threading.Lock()
 
 # SSE: list of queues for connected clients
@@ -128,7 +129,7 @@ def broadcast_event(event_type, data):
 
 def consume_kafka():
     consumer = create_consumer(
-        topics=[Config.ORDERS_TOPIC, Config.SNAPSHOTS_TOPIC, Config.TRADES_TOPIC],
+        topics=[Config.ORDERS_TOPIC, Config.SNAPSHOTS_TOPIC, Config.TRADES_TOPIC, Config.AI_INSIGHTS_TOPIC],
         group_id="dashboard",
         component_name="Dashboard",
     )
@@ -168,6 +169,12 @@ def consume_kafka():
                 ts = float(trade.get("timestamp") or time.time())
                 record_trade(sym, price, qty, ts)
 
+            elif msg.topic == Config.AI_INSIGHTS_TOPIC:
+                insight = msg.value
+                ai_insights_cache.insert(0, insight)
+                ai_insights_cache[:] = ai_insights_cache[:10]
+                broadcast_event("ai_insight", insight)
+
 
 # Initialise DB then start consumer thread
 init_history_db()
@@ -511,6 +518,7 @@ def stream():
                     f"event: init\ndata: "
                     f"{json.dumps({'orders': list(orders), 'bbos': dict(bbos), 'trades': list(trades_cache)})}\n\n"
                 )
+                yield f"event: ai_insights_init\ndata: {json.dumps(list(ai_insights_cache))}\n\n"
             # Also send current session state
             if not session_state["active"]:
                 _sess_status = "ended"

dashboard/templates/index.html

@@ -48,6 +48,30 @@
       from { background: yellow; }
       to   { background: transparent; }
     }
+
+    /* AI Insights panel */
+    .ai-panel {
+      background: #fff;
+      border-radius: 8px;
+      padding: 10px 14px;
+      box-shadow: 0 2px 4px rgba(0,0,0,0.1);
+      margin-top: 20px;
+    }
+    .insight-card {
+      padding: 9px 12px;
+      border-left: 3px solid #5c6bc0;
+      margin-bottom: 8px;
+      background: #f8f9ff;
+      border-radius: 0 4px 4px 0;
+      font-size: 13px;
+      line-height: 1.6;
+    }
+    .insight-time { font-size: 11px; color: #999; margin-bottom: 3px; }
+    @keyframes fadeInDown {
+      from { opacity: 0; transform: translateY(-5px); }
+      to   { opacity: 1; transform: translateY(0); }
+    }
+    .insight-new { animation: fadeInDown 0.4s ease; }
     /* Connection status indicator */
     .status {
       display: inline-flex;
@@ -352,6 +376,20 @@
 
   </div>
 
+  <!-- AI Analyst panel (full width) -->
+  <div class="ai-panel">
+    <h2 style="margin:0 0 8px; font-size:15px; display:flex; align-items:center; gap:10px;">
+      AI Analyst
+      <span id="ai-model-badge" style="font-size:10px; color:#fff; background:#5c6bc0; padding:2px 8px; border-radius:10px; font-weight:normal;"></span>
+      <span id="ai-status" style="font-size:11px; color:#999; font-weight:normal; margin-left:4px;">waiting for first insight…</span>
+    </h2>
+    <div id="ai-insights-list" style="max-height:220px; overflow-y:auto;">
+      <div class="insight-card" style="color:#bbb; border-left-color:#ddd; background:#fafafa;" id="ai-placeholder">
+        No insights yet — insights are generated every 30 min when the session is active.
+      </div>
+    </div>
+  </div>
+
   <script>
     // State
     const state = {
@@ -390,6 +428,18 @@
     // Selected order state
     let selectedOrder = null;
 
+    function addInsight(insight) {
+      const list = document.getElementById("ai-insights-list");
+      const ph   = document.getElementById("ai-placeholder");
+      if (ph) ph.remove();
+      const div  = document.createElement("div");
+      div.className = "insight-card insight-new";
+      const t = new Date(insight.timestamp * 1000).toLocaleTimeString();
+      div.innerHTML = `<div class="insight-time">${t}</div><div>${insight.text}</div>`;
+      list.prepend(div);
+      while (list.children.length > 10) list.removeChild(list.lastChild);
+    }
+
     function renderOrders() {
       const tbody = document.getElementById("orders-body");
       tbody.innerHTML = "";
@@ -1036,6 +1086,18 @@
         updateModeBtn(data.mode);
       });
 
+      eventSource.addEventListener("ai_insights_init", (e) => {
+        const insights = JSON.parse(e.data);
+        insights.forEach(addInsight);
+      });
+
+      eventSource.addEventListener("ai_insight", (e) => {
+        const insight = JSON.parse(e.data);
+        addInsight(insight);
+        document.getElementById("ai-status").textContent =
+          "Last update: " + new Date().toLocaleTimeString();
+      });
+
       eventSource.onerror = () => {
         setStatus("disconnected", "Disconnected");
         state.connected = false;

docker-compose.yml

@@ -146,6 +146,23 @@ services:
     depends_on:
       - fix_oeg
 
+  ai_analyst:
+    build: ./ai_analyst
+    container_name: ai_analyst
+    depends_on:
+      - kafka
+    volumes:
+      - ./shared:/app/shared
+    environment:
+      - KAFKA_BOOTSTRAP=kafka:9092
+      - OLLAMA_HOST=http://host.docker.internal:11434
+      - OLLAMA_MODEL=llama3.1:8b
+      - HF_TOKEN=${HF_TOKEN:-}
+      - HF_MODEL=${HF_MODEL:-mistralai/Mistral-7B-Instruct-v0.2}
+      - ANALYSIS_INTERVAL=1800
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+
   dashboard:
     build:
       context: ./dashboard

entrypoint.sh

@@ -39,7 +39,7 @@ done
 echo "[startup] Kafka ready."
 
 # Create topics
-for TOPIC in orders trades snapshots control; do
+for TOPIC in orders trades snapshots control ai_insights; do
   $KAFKA_DIR/bin/kafka-topics.sh \
     --create --if-not-exists \
     --topic "$TOPIC" \
@@ -65,6 +65,10 @@ echo "[startup] Starting FIX UI Client on port 5002..."
 python3 /app/fix_ui/fix_ui_client.py &
 sleep 3
 
+echo "[startup] Starting AI Analyst (interval=1800s)..."
+python3 /app/ai_analyst.py &
+sleep 1
+
 echo "[startup] Starting Frontend on port 5003..."
 PORT=$FRONTEND_PORT TEMPLATE_FOLDER=/app/frontend_templates python3 /app/frontend.py &
 sleep 2

shared/config.py

@@ -29,6 +29,9 @@ class Config:
     # Control topic for start/end of day signals
     CONTROL_TOPIC: str = os.getenv("CONTROL_TOPIC", "control")
 
+    # AI Analyst insights topic
+    AI_INSIGHTS_TOPIC: str = os.getenv("AI_INSIGHTS_TOPIC", "ai_insights")
+
     # Trading simulation
     TICK_SIZE: float = float(os.getenv("TICK_SIZE", "0.05"))
     ORDERS_PER_MIN: int = int(os.getenv("ORDERS_PER_MIN", "8"))