Integrate Meilisearch + Redis deeply for cache, locks, searchable decisions/insights, and health telemetry

.env.example

@@ -35,6 +35,9 @@ N8N_BOT_WEBHOOK=
 
 REDIS_URL=redis://10.30.20.70:6379/0
 
+MEILI_URL=http://10.30.20.142:7700
+MEILI_API_KEY=
+
 APP_HOST=0.0.0.0
 APP_PORT=8089
 DB_PATH=sqlite:///./bot.db

app.py

@@ -6,7 +6,7 @@ from datetime import datetime, timedelta
 import threading
 from db import init_db, SessionLocal, BotDecision, TradeExecution, CuratedInsight
 from bot import start_scheduler, run_cycle, curate_cycle
-from services import account_snapshot, qdrant_memory_health
+from services import account_snapshot, qdrant_memory_health, meili_health, meili_search
 
 app = FastAPI(title="alpaca-llm-bot-v1")
 templates = Jinja2Templates(directory="templates")
@@ -34,6 +34,10 @@ def curate_now():
 def api_account():
     return JSONResponse(account_snapshot())
 
+@app.get('/api/search')
+def api_search(q: str, index: str = 'decisions', limit: int = 20):
+    return JSONResponse(meili_search(index, q, limit))
+
 @app.get("/api/metrics")
 def api_metrics(hours: int = 72):
     db = SessionLocal()
@@ -95,6 +99,7 @@ def api_metrics(hours: int = 72):
             },
             "leaderboard": leaderboard,
             "qdrant": qdrant_memory_health(),
+            "meili": meili_health(),
         }
     finally:
         db.close()

bot.py

@@ -21,6 +21,7 @@ from services import (
     redis_set_json,
     redis_get_json,
     redis_lock,
+    meili_index_doc,
 )
 
 scheduler = BackgroundScheduler(timezone=settings.timezone)
@@ -39,6 +40,14 @@ def curate_cycle():
             summary = summarize_news_with_ollama(symbol, news)
             row = CuratedInsight(symbol=symbol, summary=summary, sources=json.dumps(news)[:60000])
             db.add(row)
+            meili_index_doc('insights', {
+                'id': f"ins-{int(datetime.utcnow().timestamp())}-{symbol}",
+                'ts': datetime.utcnow().isoformat(),
+                'symbol': symbol,
+                'summary': summary,
+                'sources': json.dumps(news)[:2000],
+                'kind': 'insight'
+            })
         db.commit()
     finally:
         db.close()
@@ -118,6 +127,17 @@ def run_cycle():
             db.add(drow)
             db.commit()
             db.refresh(drow)
+            meili_index_doc('decisions', {
+                'id': f"dec-{drow.id}",
+                'ts': drow.ts.isoformat() if drow.ts else datetime.utcnow().isoformat(),
+                'symbol': drow.symbol,
+                'action': drow.action,
+                'confidence': drow.confidence,
+                'status': drow.status,
+                'reason': drow.reason[:500],
+                'order_usd': drow.order_usd,
+                'kind': 'decision'
+            })
 
             should_trade = decision["action"] in {"buy", "sell"} and decision["confidence"] >= settings.min_confidence
 
@@ -145,6 +165,16 @@ def run_cycle():
                         raw=json.dumps({"decision": decision, "strategy": strat, "memory": memory_hits, "broker": res})[:60000],
                     )
                     db.add(trade)
+                    db.flush()
+                    meili_index_doc('trades', {
+                        'id': f"tr-{trade.id}",
+                        'ts': trade.ts.isoformat() if trade.ts else datetime.utcnow().isoformat(),
+                        'symbol': trade.symbol,
+                        'side': trade.side,
+                        'notional': trade.notional,
+                        'order_id': trade.alpaca_order_id,
+                        'kind': 'trade'
+                    })
                 db.commit()
 
                 # learning memory + notes + orchestration signal

config.py

@@ -42,6 +42,9 @@ class Settings:
 
     redis_url = os.getenv("REDIS_URL", "redis://10.30.20.70:6379/0")
 
+    meili_url = os.getenv("MEILI_URL", "http://10.30.20.142:7700")
+    meili_api_key = os.getenv("MEILI_API_KEY", "")
+
     db_path = os.getenv("DB_PATH", "sqlite:///./bot.db")
     host = os.getenv("APP_HOST", "0.0.0.0")
     port = int(os.getenv("APP_PORT", "8089"))

requirements.txt

@@ -7,3 +7,4 @@ python-dotenv==1.0.1
 sqlalchemy==2.0.36
 pydantic==2.10.3
 redis==5.2.1
+meilisearch-python-sdk==7.0.2

services.py

@@ -4,9 +4,63 @@ import time
 import uuid
 import requests
 import redis
+import meilisearch_python_sdk
 from config import settings
 
 _redis = None
+_meili = None
+
+def meili_client():
+    global _meili
+    if _meili is None:
+        try:
+            _meili = meilisearch_python_sdk.Client(settings.meili_url, settings.meili_api_key or None)
+        except Exception:
+            _meili = None
+    return _meili
+
+def meili_index_doc(index_uid: str, doc: dict):
+    c = meili_client()
+    if not c:
+        return False
+    try:
+        idx = c.index(index_uid)
+        idx.add_documents([doc], primary_key='id')
+        return True
+    except Exception:
+        return False
+
+def meili_search(index_uid: str, q: str, limit: int = 20):
+    c = meili_client()
+    if not c:
+        return {"hits": []}
+    try:
+        idx = c.index(index_uid)
+        res = idx.search(q, {'limit': limit})
+        if isinstance(res, dict):
+            return res
+        # sdk object fallback
+        return {
+            'hits': getattr(res, 'hits', []),
+            'estimatedTotalHits': getattr(res, 'estimated_total_hits', None),
+            'processingTimeMs': getattr(res, 'processing_time_ms', None),
+            'query': q,
+        }
+    except Exception:
+        return {"hits": []}
+
+def meili_health():
+    c = meili_client()
+    if not c:
+        return {"ok": False, "error": "client_unavailable"}
+    try:
+        h = c.health()
+        status = getattr(h, 'status', None)
+        if status is None and isinstance(h, dict):
+            status = h.get('status')
+        return {"ok": True, "status": status or 'available'}
+    except Exception as e:
+        return {"ok": False, "error": str(e)[:160]}
 
 def redis_client():
     global _redis
@@ -300,9 +354,9 @@ def trilium_log(title: str, body: str):
 
 def qdrant_memory_health():
     try:
+        # collection-specific
         r = requests.get(f"{settings.qdrant_url}/collections/{settings.qdrant_collection}", timeout=10)
-        if not r.ok:
+        if r.ok:
-            return {"ok": False, "status": r.status_code}
             j = r.json().get("result", {})
             return {
                 "ok": True,
@@ -310,6 +364,12 @@ def qdrant_memory_health():
                 "points_count": j.get("points_count"),
                 "indexed_vectors_count": j.get("indexed_vectors_count"),
             }
+        # fallback global health/list
+        r2 = requests.get(f"{settings.qdrant_url}/collections", timeout=10)
+        if r2.ok:
+            names=[c.get('name') for c in r2.json().get('result',{}).get('collections',[])]
+            return {"ok": True, "collection": settings.qdrant_collection, "exists": settings.qdrant_collection in names, "collections": names[:20]}
+        return {"ok": False, "status": r.status_code}
     except Exception as e:
         return {"ok": False, "error": str(e)[:160]}