v1.3 infra integration: Qdrant memory RAG, Trilium journaling, n8n emit hooks

2026-02-26 08:29:59 -08:00
parent cb5edf694d
commit ddc42254fa
5 changed files with 171 additions and 163 deletions
--- a/.env.example
+++ b/.env.example
@@ -1,9 +1,7 @@
-# Alpaca
 ALPACA_API_KEY=REPLACE_ME
 ALPACA_API_SECRET=REPLACE_ME
 ALPACA_BASE_URL=https://paper-api.alpaca.markets

-# Runtime
 PAPER_MODE=true
 STARTING_CAPITAL_USD=100
 MAX_ORDER_USD=5
@@ -16,16 +14,22 @@ TRADE_INTERVAL_HOURS=2
 CURATE_INTERVAL_MINUTES=30
 TIMEZONE=America/Los_Angeles

-# Ollama
 OLLAMA_URL=http://10.30.20.110:11434
 OLLAMA_CURATOR_MODEL=gemma3:latest
 OLLAMA_DECISION_MODEL=agent-oss:latest
+OLLAMA_EMBED_MODEL=nomic-embed-text:latest

-# Data sources
 SEARX_URL=http://10.30.20.35:6969/search
 SCRAPER_API_URL=http://10.30.20.115:24125

-# App
+QDRANT_URL=http://10.30.20.68:6333
+QDRANT_COLLECTION=alpaca_memory
+
+TRILIUM_URL=http://10.30.20.152:8080
+TRILIUM_TOKEN=REPLACE_ME
+
+N8N_BOT_WEBHOOK=
+
 APP_HOST=0.0.0.0
 APP_PORT=8089
 DB_PATH=sqlite:///./bot.db
--- a/README.md
+++ b/README.md
@@ -2,30 +2,25 @@

 Autonomous LLM trading system (paper-first) powered by:
 - Alpaca trading API
- Ollama local model inference
- Searx web data ingestion
+- Ollama (curator + decision models)
+- Searx ingestion + targeted deep research
+- Qdrant memory retrieval for prior similar setups
+- Trilium auto-journaling of decisions/trades
+- Optional n8n webhook emission for orchestration
 - FastAPI dark dashboard
- APScheduler autonomous loops

-## Autonomous behavior
- Curates market/news context every `CURATE_INTERVAL_MINUTES` (default 30)
- Runs trade decision cycle every `TRADE_INTERVAL_HOURS` (default 2)
- LLM decides buy/sell/hold per symbol
- Executes only if confidence >= `MIN_CONFIDENCE`
- Hard risk caps:
-  - max `$5` order notional (default)
-  - max daily notional cap
-  - max open positions cap
-  - market-open gate
+## v1.3 additions
+- Retrieval-augmented decisions via Qdrant (`QDRANT_URL`)
+- Auto-write trade logs to Trilium (`TRILIUM_URL`, `TRILIUM_TOKEN`)
+- n8n signal hook (`N8N_BOT_WEBHOOK`)
+- Two-model pipeline: cheap curator + stronger final decision model
+- Fee/slippage-aware tiny-bankroll controls

-## Quick start
+## Run
 ```bash
 cp .env.example .env
-# fill ALPACA_API_KEY / ALPACA_API_SECRET
-python3 -m venv .venv
-source .venv/bin/activate
-pip install -r requirements.txt
-uvicorn app:app --host 0.0.0.0 --port 8089
+# fill Alpaca keys + optional Trilium/N8N vars
+python3 -m uvicorn app:app --host 0.0.0.0 --port 8089
 ```

 Dashboard:
@@ -37,14 +32,6 @@ curl -X POST http://127.0.0.1:8089/curate-now
 curl -X POST http://127.0.0.1:8089/run-now
 ```

-## Production service (systemd)
-```bash
-sudo cp systemd/alpaca-llm-bot-v1.service /etc/systemd/system/
-sudo systemctl daemon-reload
-sudo systemctl enable --now alpaca-llm-bot-v1
-sudo systemctl status alpaca-llm-bot-v1
-```
-
-## Important
- Keep `PAPER_MODE=true` until you observe stable behavior.
- This is experimental and not financial advice.
+## Safety
+- Keep `PAPER_MODE=true` until stable
+- This is experimental software, not financial advice
--- a/bot.py
+++ b/bot.py
@@ -13,6 +13,10 @@ from services import (
    place_order,
    market_open,
    positions_snapshot,
+    qdrant_similar,
+    qdrant_add_memory,
+    trilium_log,
+    n8n_emit,
 )

 scheduler = BackgroundScheduler(timezone=settings.timezone)
@@ -55,14 +59,15 @@ def run_cycle():

            news = searx_news(symbol)
            strat = strategy_signals(symbol, news)
-            decision = llm_final_decision(symbol, news, strat)
+            memory_hits = qdrant_similar(symbol, json.dumps(news)[:2000], limit=5)
+            decision = llm_final_decision(symbol, news, strat, memory_hits)

-            # Escalate to deeper research when model asks or confidence weak
            if decision.get("needs_more_research") or decision.get("confidence", 0) < settings.min_confidence:
                more = extra_research(symbol, decision.get("research_topics", []))
                if more:
                    news = news + more
-                    decision = llm_final_decision(symbol, news, strat)
+                    memory_hits = qdrant_similar(symbol, json.dumps(news)[:2000], limit=5)
+                    decision = llm_final_decision(symbol, news, strat, memory_hits)

            drow = BotDecision(
                symbol=symbol,
@@ -77,19 +82,11 @@ def run_cycle():
            db.commit()
            db.refresh(drow)

-            should_trade = (
-                decision["action"] in {"buy", "sell"}
-                and decision["confidence"] >= settings.min_confidence
-            )
+            should_trade = decision["action"] in {"buy", "sell"} and decision["confidence"] >= settings.min_confidence

            if should_trade:
-                # Fee/slippage-aware cap for tiny bankroll
                effective_cost = settings.fee_per_trade_usd + (settings.slippage_bps / 10000.0) * decision["order_usd"]
-                notional = min(
-                    settings.max_order_usd,
-                    decision["order_usd"],
-                    settings.max_daily_notional - spent,
-                )
+                notional = min(settings.max_order_usd, decision["order_usd"], settings.max_daily_notional - spent)
                if notional <= effective_cost:
                    drow.status = "risk_blocked"
                    db.add(drow)
@@ -100,21 +97,46 @@ def run_cycle():
                ok = bool(res and res.get("ok"))
                drow.status = "executed" if ok else "failed"
                db.add(drow)
-                db.add(TradeExecution(
+
+                trade = TradeExecution(
                    symbol=symbol,
                    side=decision["action"],
                    qty=float((res or {}).get("json", {}).get("qty", 0) or 0),
                    notional=notional,
                    alpaca_order_id=(res or {}).get("json", {}).get("id", ""),
-                    raw=json.dumps({"decision": decision, "strategy": strat, "broker": res})[:60000],
-                ))
+                    raw=json.dumps({"decision": decision, "strategy": strat, "memory": memory_hits, "broker": res})[:60000],
+                )
+                db.add(trade)
                db.commit()
+
+                # learning memory + notes + orchestration signal
+                qdrant_add_memory(symbol, f"{symbol} {decision['action']} conf={decision['confidence']} reason={decision['reason']}", {
+                    "status": drow.status,
+                    "action": decision["action"],
+                    "confidence": decision["confidence"],
+                    "ts": datetime.utcnow().isoformat(),
+                })
+                trilium_log(
+                    f"Trade {symbol} {decision['action']} {drow.status}",
+                    f"## Decision\n- symbol: {symbol}\n- action: {decision['action']}\n- confidence: {decision['confidence']:.2f}\n- status: {drow.status}\n- notional: ${notional:.2f}\n\n## Reason\n{decision['reason']}\n\n## Strategy\n{json.dumps(strat, indent=2)}\n"
+                )
+                n8n_emit({"event": "trade", "symbol": symbol, "status": drow.status, "decision": decision, "notional": notional})
+
                if ok:
                    spent += notional
            else:
                drow.status = "skipped"
                db.add(drow)
                db.commit()
+
+            # store non-trade decisions too for memory
+            qdrant_add_memory(symbol, f"{symbol} decision={decision['action']} conf={decision['confidence']} status={drow.status}", {
+                "status": drow.status,
+                "action": decision["action"],
+                "confidence": decision["confidence"],
+                "ts": datetime.utcnow().isoformat(),
+            })
+
    finally:
        db.close()

--- a/config.py
+++ b/config.py
@@ -9,32 +9,34 @@ class Settings:
    alpaca_base = os.getenv("ALPACA_BASE_URL", "https://paper-api.alpaca.markets")
    paper_mode = os.getenv("PAPER_MODE", "true").lower() == "true"

-    # Capital/risk profile
    starting_capital_usd = float(os.getenv("STARTING_CAPITAL_USD", "100"))
    max_order_usd = float(os.getenv("MAX_ORDER_USD", "5"))
    max_daily_notional = float(os.getenv("MAX_DAILY_NOTIONAL", "40"))
    max_open_positions = int(os.getenv("MAX_OPEN_POSITIONS", "8"))
    min_confidence = float(os.getenv("MIN_CONFIDENCE", "0.60"))
-
-    # Approx fee model for small-size optimization
    fee_per_trade_usd = float(os.getenv("FEE_PER_TRADE_USD", "0.00"))
    slippage_bps = float(os.getenv("SLIPPAGE_BPS", "5"))

-    # Scheduling
    trade_interval_hours = int(os.getenv("TRADE_INTERVAL_HOURS", "2"))
    curate_interval_minutes = int(os.getenv("CURATE_INTERVAL_MINUTES", "30"))
    timezone = os.getenv("TIMEZONE", "America/Los_Angeles")

-    # LLM stack (small for curation, larger for final decision)
    ollama_url = os.getenv("OLLAMA_URL", "http://10.30.20.110:11434")
    ollama_curator_model = os.getenv("OLLAMA_CURATOR_MODEL", "gemma3:latest")
    ollama_decision_model = os.getenv("OLLAMA_DECISION_MODEL", "agent-oss:latest")
+    ollama_embed_model = os.getenv("OLLAMA_EMBED_MODEL", "nomic-embed-text:latest")

-    # Data sources
    searx_url = os.getenv("SEARX_URL", "http://10.30.20.35:6969/search")
    scraper_api = os.getenv("SCRAPER_API_URL", "http://10.30.20.115:24125")

-    # App
+    qdrant_url = os.getenv("QDRANT_URL", "http://10.30.20.68:6333")
+    qdrant_collection = os.getenv("QDRANT_COLLECTION", "alpaca_memory")
+
+    trilium_url = os.getenv("TRILIUM_URL", "http://10.30.20.152:8080")
+    trilium_token = os.getenv("TRILIUM_TOKEN", "")
+
+    n8n_webhook = os.getenv("N8N_BOT_WEBHOOK", "")
+
    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"))
--- a/services.py
+++ b/services.py
@@ -1,20 +1,66 @@
 import json
 import random
+import time
+import uuid
 import requests
 from config import settings


 def _ollama_generate(model: str, payload_obj: dict, timeout: int = 45):
-    payload = {
-        "model": model,
-        "stream": False,
-        "prompt": json.dumps(payload_obj),
-        "format": "json",
-    }
+    payload = {"model": model, "stream": False, "prompt": json.dumps(payload_obj), "format": "json"}
    r = requests.post(f"{settings.ollama_url}/api/generate", json=payload, timeout=timeout)
    r.raise_for_status()
-    resp = r.json().get("response", "{}")
-    return json.loads(resp)
+    return json.loads(r.json().get("response", "{}"))
+
+
+def _embed(text: str):
+    try:
+        r = requests.post(f"{settings.ollama_url}/api/embeddings", json={"model": settings.ollama_embed_model, "prompt": text[:8000]}, timeout=30)
+        r.raise_for_status()
+        return r.json().get("embedding", [])
+    except Exception:
+        return []
+
+
+def qdrant_ensure_collection(vector_size=768):
+    try:
+        requests.put(f"{settings.qdrant_url}/collections/{settings.qdrant_collection}", json={"vectors": {"size": vector_size, "distance": "Cosine"}}, timeout=10)
+    except Exception:
+        pass
+
+
+def qdrant_add_memory(symbol: str, text: str, payload: dict):
+    vec = _embed(text)
+    if not vec:
+        return False
+    qdrant_ensure_collection(len(vec))
+    body = {
+        "points": [{"id": str(uuid.uuid4()), "vector": vec, "payload": {"symbol": symbol, "text": text[:2000], **payload}}]
+    }
+    try:
+        r = requests.put(f"{settings.qdrant_url}/collections/{settings.qdrant_collection}/points", json=body, timeout=15)
+        return r.ok
+    except Exception:
+        return False
+
+
+def qdrant_similar(symbol: str, query_text: str, limit: int = 5):
+    vec = _embed(query_text)
+    if not vec:
+        return []
+    body = {"vector": vec, "limit": limit, "with_payload": True}
+    try:
+        r = requests.post(f"{settings.qdrant_url}/collections/{settings.qdrant_collection}/points/search", json=body, timeout=15)
+        if not r.ok:
+            return []
+        out = []
+        for p in r.json().get("result", []):
+            pl = p.get("payload", {})
+            if pl.get("symbol") in (symbol, None):
+                out.append({"score": p.get("score", 0), "text": pl.get("text", ""), "status": pl.get("status", "")})
+        return out
+    except Exception:
+        return []


 def searx_news(symbol: str, limit: int = 12):
@@ -24,50 +70,25 @@ def searx_news(symbol: str, limit: int = 12):
        r = requests.get(settings.searx_url, params=params, timeout=20)
        r.raise_for_status()
        data = r.json()
-        out = []
-        for it in data.get("results", [])[:limit]:
-            out.append({
-                "title": it.get("title", ""),
-                "url": it.get("url", ""),
-                "content": (it.get("content", "") or "")[:700],
-            })
-        return out
+        return [{"title": it.get("title", ""), "url": it.get("url", ""), "content": (it.get("content", "") or "")[:700]} for it in data.get("results", [])[:limit]]
    except Exception:
        return []


 def extra_research(symbol: str, weak_points: list, limit: int = 6):
-    """Second-pass targeted research when confidence/coverage is weak."""
    q = f"{symbol} {' '.join(weak_points[:3])} SEC filing guidance risks competition"
    params = {"q": q, "format": "json", "language": "en"}
    try:
        r = requests.get(settings.searx_url, params=params, timeout=20)
        r.raise_for_status()
        data = r.json()
-        out = []
-        for it in data.get("results", [])[:limit]:
-            out.append({
-                "title": it.get("title", ""),
-                "url": it.get("url", ""),
-                "content": (it.get("content", "") or "")[:700],
-            })
-        return out
+        return [{"title": it.get("title", ""), "url": it.get("url", ""), "content": (it.get("content", "") or "")[:700]} for it in data.get("results", [])[:limit]]
    except Exception:
        return []


 def summarize_news_with_ollama(symbol: str, context_items: list):
-    prompt = {
-        "task": "Summarize market-moving info into a concise, neutral brief.",
-        "symbol": symbol,
-        "news": context_items,
-        "format": {
-            "summary": "<=140 words",
-            "bullish_points": ["..."],
-            "bearish_points": ["..."],
-            "uncertainties": ["..."]
-        }
-    }
+    prompt = {"task": "Summarize market-moving info into a concise brief", "symbol": symbol, "news": context_items}
    try:
        parsed = _ollama_generate(settings.ollama_curator_model, prompt)
        return parsed.get("summary", "no-summary")
@@ -76,100 +97,50 @@ def summarize_news_with_ollama(symbol: str, context_items: list):


 def strategy_signals(symbol: str, context_items: list):
-    """Proven-ish small-capital rules encoded as interpretable signals."""
    text_blob = " ".join((x.get("title", "") + " " + x.get("content", "")) for x in context_items).lower()
    bullish = sum(k in text_blob for k in ["beat", "raise guidance", "upgrade", "buyback", "record revenue"])
    bearish = sum(k in text_blob for k in ["miss", "downgrade", "lawsuit", "probe", "cut guidance", "recall"])
-
-    # simple event momentum score
    score = bullish - bearish
-    # conservative policy for tiny capital: only trade stronger score edges
-    if score >= 2:
-        action = "buy"
-    elif score <= -2:
-        action = "sell"
-    else:
-        action = "hold"
-
+    action = "buy" if score >= 2 else ("sell" if score <= -2 else "hold")
    conf = min(0.85, 0.50 + abs(score) * 0.08)
-    return {
-        "strategy": "event-momentum-v1",
-        "score": score,
-        "action": action,
-        "confidence": conf,
-        "signals": {"bullish": bullish, "bearish": bearish},
-    }
+    return {"strategy": "event-momentum-v1", "score": score, "action": action, "confidence": conf, "signals": {"bullish": bullish, "bearish": bearish}}


-def llm_final_decision(symbol: str, context_items: list, strategy: dict):
+def llm_final_decision(symbol: str, context_items: list, strategy: dict, memory_hits: list):
    prompt = {
-        "task": "Final trading decision using all context and a conservative small-capital profile. Return strict JSON.",
+        "task": "Final trading decision. Return strict JSON.",
        "symbol": symbol,
-        "constraints": {
-            "actions": ["buy", "sell", "hold"],
-            "max_order_usd": settings.max_order_usd,
-            "min_expected_edge_after_fees": "positive",
-            "fee_per_trade_usd": settings.fee_per_trade_usd,
-            "slippage_bps": settings.slippage_bps,
-            "avoid_overtrading": True,
-        },
+        "constraints": {"actions": ["buy", "sell", "hold"], "max_order_usd": settings.max_order_usd, "fee_per_trade_usd": settings.fee_per_trade_usd, "slippage_bps": settings.slippage_bps, "avoid_overtrading": True},
        "strategy_prior": strategy,
+        "memory_hits": memory_hits,
        "news": context_items,
-        "output_schema": {
-            "action": "buy|sell|hold",
-            "confidence": "0-1",
-            "order_usd": f"<= {settings.max_order_usd}",
-            "reason": "short rationale",
-            "needs_more_research": True,
-            "research_topics": ["..."]
-        },
+        "output_schema": {"action": "buy|sell|hold", "confidence": "0-1", "order_usd": f"<= {settings.max_order_usd}", "reason": "short rationale", "needs_more_research": True, "research_topics": ["..."]},
    }
-
    try:
        d = _ollama_generate(settings.ollama_decision_model, prompt, timeout=60)
        action = str(d.get("action", "hold")).lower()
        if action not in {"buy", "sell", "hold"}:
            action = "hold"
-        confidence = max(0.0, min(1.0, float(d.get("confidence", 0.5))))
-        order_usd = min(float(d.get("order_usd", settings.max_order_usd)), settings.max_order_usd)
-        reason = d.get("reason", "fallback")
        return {
            "action": action,
-            "confidence": confidence,
-            "order_usd": order_usd,
-            "reason": reason,
+            "confidence": max(0.0, min(1.0, float(d.get("confidence", 0.5)))),
+            "order_usd": min(float(d.get("order_usd", settings.max_order_usd)), settings.max_order_usd),
+            "reason": d.get("reason", "fallback"),
            "needs_more_research": bool(d.get("needs_more_research", False)),
            "research_topics": d.get("research_topics", []) or [],
        }
    except Exception:
-        return {
-            "action": strategy.get("action", "hold"),
-            "confidence": min(strategy.get("confidence", 0.5), 0.55),
-            "order_usd": min(1.0, settings.max_order_usd),
-            "reason": "decision-fallback-strategy",
-            "needs_more_research": False,
-            "research_topics": [],
-        }
+        return {"action": strategy.get("action", "hold"), "confidence": min(strategy.get("confidence", 0.5), 0.55), "order_usd": min(1.0, settings.max_order_usd), "reason": "decision-fallback-strategy", "needs_more_research": False, "research_topics": []}


 def alpaca_headers():
-    return {
-        "APCA-API-KEY-ID": settings.alpaca_key,
-        "APCA-API-SECRET-KEY": settings.alpaca_secret,
-        "Content-Type": "application/json",
-    }
+    return {"APCA-API-KEY-ID": settings.alpaca_key, "APCA-API-SECRET-KEY": settings.alpaca_secret, "Content-Type": "application/json"}


 def place_order(symbol: str, action: str, order_usd: float):
    if action not in {"buy", "sell"}:
        return None
-    payload = {
-        "symbol": symbol,
-        "side": action,
-        "type": "market",
-        "time_in_force": "day",
-        "notional": round(order_usd, 2),
-    }
+    payload = {"symbol": symbol, "side": action, "type": "market", "time_in_force": "day", "notional": round(order_usd, 2)}
    try:
        r = requests.post(f"{settings.alpaca_base}/v2/orders", headers=alpaca_headers(), json=payload, timeout=20)
        return {"ok": r.ok, "status": r.status_code, "json": r.json() if r.text else {}}
@@ -189,18 +160,40 @@ def account_snapshot():
 def positions_snapshot():
    try:
        r = requests.get(f"{settings.alpaca_base}/v2/positions", headers=alpaca_headers(), timeout=20)
-        if r.ok:
-            return r.json()
+        return r.json() if r.ok else []
    except Exception:
-        pass
-    return []
+        return []


 def market_open():
    try:
        r = requests.get(f"{settings.alpaca_base}/v2/clock", headers=alpaca_headers(), timeout=20)
-        if r.ok:
-            return bool(r.json().get("is_open", False))
+        return bool(r.json().get("is_open", False)) if r.ok else False
+    except Exception:
+        return False
+
+
+def trilium_log(title: str, body: str):
+    if not settings.trilium_token:
+        return False
+    headers = {"Authorization": settings.trilium_token, "Content-Type": "application/json"}
+    payload = {"title": title[:120], "type": "text", "mime": "text/markdown", "content": body}
+    try:
+        # best-effort endpoints across Trilium variants
+        for ep in ["/etapi/create-note", "/etapi/notes"]:
+            r = requests.post(settings.trilium_url.rstrip("/") + ep, headers=headers, json=payload, timeout=15)
+            if r.ok:
+                return True
    except Exception:
        pass
    return False
+
+
+def n8n_emit(event: dict):
+    if not settings.n8n_webhook:
+        return False
+    try:
+        r = requests.post(settings.n8n_webhook, json=event, timeout=10)
+        return r.ok
+    except Exception:
+        return False