Integrate Redis (10.30.20.70) for locks, cache, and decision state to maximize infra utilization

.env.example

@@ -33,6 +33,8 @@ TRILIUM_TOKEN=REPLACE_ME
 
 N8N_BOT_WEBHOOK=
 
+REDIS_URL=redis://10.30.20.70:6379/0
+
 APP_HOST=0.0.0.0
 APP_PORT=8089
 DB_PATH=sqlite:///./bot.db

bot.py

@@ -18,6 +18,9 @@ from services import (
     qdrant_add_memory,
     trilium_log,
     n8n_emit,
+    redis_set_json,
+    redis_get_json,
+    redis_lock,
 )
 
 scheduler = BackgroundScheduler(timezone=settings.timezone)
@@ -42,6 +45,9 @@ def curate_cycle():
 
 
 def run_cycle():
+    if not redis_lock('alpaca:run_cycle:lock', ttl=240):
+        return
+
     db = SessionLocal()
     try:
         acct = account_snapshot()
@@ -64,7 +70,12 @@ def run_cycle():
             if spent >= settings.max_daily_notional:
                 break
 
-            news = searx_news(symbol)
+            cached_news = redis_get_json(f'alpaca:news:{symbol}')
+            news = cached_news if cached_news else searx_news(symbol)
+            if news:
+                redis_set_json(f'alpaca:news:{symbol}', {'items': news} if isinstance(news, list) else news, ttl=900)
+                if isinstance(news, dict) and 'items' in news:
+                    news = news['items']
             strat = strategy_signals(symbol, news)
             memory_hits = qdrant_similar(symbol, json.dumps(news)[:2000], limit=5)
             decision = llm_final_decision(symbol, news, strat, memory_hits)
@@ -82,6 +93,19 @@ def run_cycle():
                 decision["order_usd"] = max(decision.get("order_usd", 0.0), settings.force_paper_min_usd)
                 decision["reason"] = f"{decision.get('reason','')} | force_paper_trades"
 
+            redis_set_json(
+                f"alpaca:last_decision:{symbol}",
+                {
+                    "symbol": symbol,
+                    "action": decision["action"],
+                    "confidence": decision["confidence"],
+                    "order_usd": decision["order_usd"],
+                    "reason": decision["reason"],
+                    "ts": datetime.utcnow().isoformat(),
+                },
+                ttl=86400,
+            )
+
             drow = BotDecision(
                 symbol=symbol,
                 action=decision["action"],

config.py

@@ -40,6 +40,8 @@ class Settings:
 
     n8n_webhook = os.getenv("N8N_BOT_WEBHOOK", "")
 
+    redis_url = os.getenv("REDIS_URL", "redis://10.30.20.70:6379/0")
+
     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

@@ -6,3 +6,4 @@ apscheduler==3.10.4
 python-dotenv==1.0.1
 sqlalchemy==2.0.36
 pydantic==2.10.3
+redis==5.2.1

services.py

@@ -3,8 +3,50 @@ import random
 import time
 import uuid
 import requests
+import redis
 from config import settings
 
+_redis = None
+
+def redis_client():
+    global _redis
+    if _redis is None:
+        try:
+            _redis = redis.Redis.from_url(settings.redis_url, decode_responses=True, socket_timeout=2)
+            _redis.ping()
+        except Exception:
+            _redis = None
+    return _redis
+
+def redis_set_json(key: str, value: dict, ttl: int = 3600):
+    r = redis_client()
+    if not r:
+        return False
+    try:
+        r.setex(key, ttl, json.dumps(value))
+        return True
+    except Exception:
+        return False
+
+def redis_get_json(key: str):
+    r = redis_client()
+    if not r:
+        return None
+    try:
+        v = r.get(key)
+        return json.loads(v) if v else None
+    except Exception:
+        return None
+
+def redis_lock(key: str, ttl: int = 180):
+    r = redis_client()
+    if not r:
+        return True
+    try:
+        return bool(r.set(key, str(int(time.time())), ex=ttl, nx=True))
+    except Exception:
+        return True
+
 
 def _ollama_generate(model: str, payload_obj: dict, timeout: int = 45):
     payload = {"model": model, "stream": False, "prompt": json.dumps(payload_obj), "format": "json"}