Add optional Adanos market sentiment tool

CHANGELOG.md

@@ -6,6 +6,7 @@ Deploy marker: 4.5.32 release commit (`deploy 4.5.32`)
 
 ### Changed
 - **Lumibot README and docs now frame the agent examples as AI trading teams.** The public docs refresh the BotSpot CTAs, favicon/logo assets, and agent-flow visuals while keeping the existing examples and links intact.
+- **AI agents can optionally use Adanos Market Sentiment API data.** Strategies now get a `strategy.sentiment` helper and, when `ADANOS_API_KEY` is configured, agents can call `adanos_market_sentiment` for US-equity sentiment from Reddit, X / FinTwit, news, and Polymarket.
 
 ### Fixed
 - **Schwab position sync now skips unsupported mutual-fund and bond positions instead of crashing.** Accounts that contain asset classes LumiBot does not model can still sync supported stocks, ETFs, options, futures, and cash-like positions.

docsrc/agents_builtin_tools.rst

@@ -108,6 +108,18 @@ using realtime parameters. LumiBot's built-in FRED tools do not use public CSV
 fallbacks; macro tool output should either come from the official API or fail
 clearly.
 
+Adanos Market Sentiment
+-----------------------
+
+If ``ADANOS_API_KEY`` is configured, LumiBot exposes
+``adanos_market_sentiment`` as an optional agent evidence tool. It can request
+US-equity sentiment for a single stock or broad market sentiment from Adanos
+sources including Reddit, X / FinTwit, news, and Polymarket.
+
+In backtests, the tool defaults its ``end`` bound to the current simulated date
+so agents do not request data after the strategy datetime. Strategies can also
+call the same helper directly through ``self.sentiment``.
+
 News
 ----
 

docsrc/environment_variables.rst

@@ -738,8 +738,8 @@ ANTHROPIC_API_KEY
 
 Other providers (Groq, Mistral, Cohere, Fireworks, Together, etc.) use the provider-prefixed id format and the corresponding provider env var; see the LiteLLM documentation for the full list.
 
-SEC fundamentals and agent memory
----------------------------------
+SEC fundamentals, market sentiment, and agent memory
+----------------------------------------------------
 
 LUMIBOT_SEC_USER_AGENT
 ^^^^^^^^^^^^^^^^^^^^^^
@@ -770,6 +770,22 @@ LUMIBOT_FRED_CACHE_DIR
 - Purpose: Override local FRED macro data cache.
 - Default: ``~/.lumibot/cache/fred``.
 
+ADANOS_API_KEY
+^^^^^^^^^^^^^^
+
+- Purpose: Optional Adanos Market Sentiment API key for the
+  ``adanos_market_sentiment`` built-in agent tool and
+  ``strategy.sentiment`` helper.
+- Default: unset.
+- Notes: When set, LumiBot can fetch US-equity sentiment from Reddit,
+  X / FinTwit, news, and Polymarket via the Adanos API.
+
+ADANOS_API_BASE_URL
+^^^^^^^^^^^^^^^^^^^
+
+- Purpose: Override the Adanos API base URL.
+- Default: ``https://api.adanos.org``.
+
 LUMIBOT_MEMORY_DIR
 ^^^^^^^^^^^^^^^^^^
 

lumibot/__init__.py

@@ -129,6 +129,7 @@ def _default_cache_folder() -> str:
     "traders",
     "tools",
     "components",
+    "sentiment",
     "constants",
     "credentials",
     "trading_builtins",

lumibot/components/agents/builtins.py

@@ -383,6 +383,17 @@ def docs_search(*, query: str, max_results: int = 5, limit: int | None = None) -
 )
 
 
+ADANOS_MARKET_SENTIMENT_DESCRIPTION = (
+    "Fetch Adanos Market Sentiment API data for US equities using ADANOS_API_KEY. "
+    "Use this as an optional external sentiment signal alongside prices, indicators, news, "
+    "SEC filings, and macro data. "
+    "Sources can be reddit, x, news, polymarket, or a comma-separated subset. "
+    "Set mode='stock' with symbol for per-stock sentiment, or mode='market' for broad market sentiment. "
+    "In backtests, end defaults to the current simulated date so the tool does not request data "
+    "after the strategy datetime."
+)
+
+
 def _bind_alpaca_news(strategy: Any, manager: Any) -> BoundTool:
     def _warn_unavailable() -> None:
         message = (
@@ -595,6 +606,90 @@ def alpaca_news(
     )
 
 
+def _bind_adanos_market_sentiment(strategy: Any, manager: Any) -> BoundTool:
+    def _warn_unavailable() -> None:
+        message = (
+            "[agents] adanos_market_sentiment is not configured and will not be exposed. "
+            "Set ADANOS_API_KEY to use Adanos Market Sentiment API data."
+        )
+        if manager is not None:
+            warned = getattr(manager, "_warned_unavailable_builtin_tools", None)
+            if warned is None:
+                warned = set()
+                manager._warned_unavailable_builtin_tools = warned
+            if "adanos_market_sentiment" in warned:
+                return
+            warned.add("adanos_market_sentiment")
+        log_message = getattr(strategy, "log_message", None)
+        if callable(log_message):
+            try:
+                log_message(message, color="yellow")
+                return
+            except Exception:
+                pass
+        warning = getattr(manager, "warning", None) if manager is not None else None
+        if callable(warning):
+            warning(message)
+
+    sentiment_client = getattr(strategy, "sentiment", None)
+    api_key = str(getattr(sentiment_client, "api_key", "") or os.environ.get("ADANOS_API_KEY") or "").strip()
+
+    def unavailable_adanos_market_sentiment(**kwargs: Any) -> dict[str, Any]:
+        return {
+            "ok": False,
+            "tool_error": True,
+            "error": {
+                "type": "MissingCredentials",
+                "message": "ADANOS_API_KEY is required to fetch Adanos market sentiment data.",
+            },
+            "results": {},
+        }
+
+    if not api_key:
+        _warn_unavailable()
+        return BoundTool(
+            name="adanos_market_sentiment",
+            description=ADANOS_MARKET_SENTIMENT_DESCRIPTION,
+            function=unavailable_adanos_market_sentiment,
+            source="builtin",
+            metadata={
+                "kind": "sentiment",
+                "disabled": True,
+                "disabled_reason": "missing ADANOS_API_KEY",
+            },
+        )
+
+    if sentiment_client is None:
+        from lumibot.sentiment import AdanosMarketSentiment
+
+        sentiment_client = AdanosMarketSentiment(strategy)
+
+    def adanos_market_sentiment(
+        *,
+        symbol: str = "",
+        sources: str = "reddit,x,news,polymarket",
+        days: int = 7,
+        end: str | None = None,
+        mode: str = "stock",
+    ) -> dict[str, Any]:
+        mode_text = str(mode or "stock").strip().lower()
+        if mode_text == "market":
+            result = sentiment_client.get_market_sentiment(sources=sources, days=days, end=end)
+        elif mode_text == "stock":
+            result = sentiment_client.get_stock_sentiment(symbol, sources=sources, days=days, end=end)
+        else:
+            raise ValueError("mode must be 'stock' or 'market'.")
+        return {"ok": not bool(result.get("errors")), **result}
+
+    return BoundTool(
+        name="adanos_market_sentiment",
+        description=ADANOS_MARKET_SENTIMENT_DESCRIPTION,
+        function=adanos_market_sentiment,
+        source="builtin",
+        metadata={"kind": "sentiment"},
+    )
+
+
 def _bind_open_orders(strategy: Any, manager: Any) -> BoundTool:
     def open_orders() -> dict[str, Any]:
         orders = strategy.get_orders()
@@ -1252,6 +1347,15 @@ def alpaca_news(self) -> ToolDefinition:
         )
 
 
+class _SentimentTools:
+    def adanos_market_sentiment(self) -> ToolDefinition:
+        return ToolDefinition(
+            name="adanos_market_sentiment",
+            description=ADANOS_MARKET_SENTIMENT_DESCRIPTION,
+            binder=_bind_adanos_market_sentiment,
+        )
+
+
 class _IndicatorTools:
     def list_indicators(self) -> ToolDefinition:
         return ToolDefinition(name="list_indicators", description="List common technical indicators.", binder=_bind_list_indicators)
@@ -1363,6 +1467,7 @@ class _BuiltinTools:
     duckdb = _DuckDBTools()
     docs = _DocsTools()
     news = _NewsTools()
+    sentiment = _SentimentTools()
     indicators = _IndicatorTools()
     fundamentals = _FundamentalTools()
     macro = _MacroTools()
@@ -1380,6 +1485,7 @@ def all(self) -> list[ToolDefinition]:
             self.duckdb.query(),
             self.docs.search(),
             self.news.alpaca_news(),
+            self.sentiment.adanos_market_sentiment(),
             self.indicators.list_indicators(),
             self.indicators.get_indicator(),
             self.indicators.get_indicators(),

lumibot/components/agents/manager.py

@@ -1098,6 +1098,7 @@ def _derive_warnings(self, result: AgentRunResult, runtime_context: dict[str, An
                 "get_fred_series",
                 "get_fred_latest",
                 "get_fred_snapshot",
+                "adanos_market_sentiment",
             }
             for name in tool_names
         )

lumibot/sentiment/__init__.py

@@ -0,0 +1,5 @@
+"""Sentiment data helpers for LumiBot strategies and agents."""
+
+from .adanos import AdanosMarketSentiment
+
+__all__ = ["AdanosMarketSentiment"]

lumibot/sentiment/adanos.py

@@ -0,0 +1,168 @@
+import os
+from datetime import date, datetime, timezone
+from typing import Any
+
+import requests
+
+ADANOS_API_BASE_URL = "https://api.adanos.org"
+ADANOS_SOURCES = ("reddit", "x", "news", "polymarket")
+
+
+def _parse_dt(value: Any) -> datetime | None:
+    if isinstance(value, datetime):
+        return value
+    if isinstance(value, date):
+        return datetime(value.year, value.month, value.day)
+    text = str(value or "").strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = f"{text[:-1]}+00:00"
+    try:
+        return datetime.fromisoformat(text)
+    except ValueError:
+        try:
+            return datetime.strptime(text[:10], "%Y-%m-%d")
+        except ValueError:
+            return None
+
+
+def _as_of_datetime(value: Any) -> datetime:
+    parsed = _parse_dt(value)
+    if parsed is not None:
+        return parsed
+    return datetime.now(timezone.utc)
+
+
+def _date_text(value: Any | None) -> str | None:
+    parsed = _parse_dt(value)
+    if parsed is None:
+        return None
+    return parsed.date().isoformat()
+
+
+def _source_list(sources: list[str] | tuple[str, ...] | str | None) -> list[str]:
+    if sources is None:
+        requested = list(ADANOS_SOURCES)
+    elif isinstance(sources, str):
+        requested = [part.strip().lower() for part in sources.split(",") if part.strip()]
+    else:
+        requested = [str(part).strip().lower() for part in sources if str(part).strip()]
+    if not requested:
+        raise ValueError("No Adanos sources provided.")
+    invalid = [source for source in requested if source not in ADANOS_SOURCES]
+    if invalid:
+        raise ValueError(f"Unsupported Adanos source(s): {', '.join(invalid)}")
+    return requested
+
+
+class AdanosMarketSentiment:
+    """Small Adanos Market Sentiment API client for US-equity research.
+
+    The helper is intentionally transport-level and optional. Set
+    ``ADANOS_API_KEY`` or pass ``api_key=...`` explicitly before calling live
+    endpoints.
+    """
+
+    def __init__(
+        self,
+        strategy: Any | None = None,
+        *,
+        api_key: str | None = None,
+        base_url: str | None = None,
+        timeout: float = 30.0,
+    ) -> None:
+        self.strategy = strategy
+        self.api_key = api_key or os.environ.get("ADANOS_API_KEY")
+        self.base_url = (base_url or os.environ.get("ADANOS_API_BASE_URL") or ADANOS_API_BASE_URL).rstrip("/")
+        self.timeout = float(timeout)
+
+    def _strategy_as_of(self) -> datetime:
+        if self.strategy is not None and hasattr(self.strategy, "get_datetime"):
+            try:
+                return _as_of_datetime(self.strategy.get_datetime())
+            except Exception:
+                pass
+        return datetime.now(timezone.utc)
+
+    def _headers(self) -> dict[str, str]:
+        key = str(self.api_key or os.environ.get("ADANOS_API_KEY") or "").strip()
+        if not key:
+            raise ValueError("ADANOS_API_KEY is required to fetch Adanos market sentiment data.")
+        return {"X-API-Key": key}
+
+    def _get_json(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
+        response = requests.get(
+            f"{self.base_url}{path}",
+            headers=self._headers(),
+            params={key: value for key, value in params.items() if value not in (None, "")},
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        if isinstance(payload, dict):
+            return payload
+        return {"data": payload}
+
+    def get_stock_sentiment(
+        self,
+        symbol: str,
+        *,
+        sources: list[str] | tuple[str, ...] | str | None = None,
+        days: int = 7,
+        end: Any | None = None,
+    ) -> dict[str, Any]:
+        ticker = str(symbol or "").strip().upper()
+        if not ticker:
+            raise ValueError("symbol is required.")
+        day_count = max(int(days), 1)
+        end_text = _date_text(end) or self._strategy_as_of().date().isoformat()
+        results: dict[str, Any] = {}
+        errors: dict[str, str] = {}
+        for source in _source_list(sources):
+            try:
+                results[source] = self._get_json(
+                    f"/{source}/stocks/v1/stock/{ticker}",
+                    {"days": day_count, "to": end_text},
+                )
+            except Exception as exc:
+                errors[source] = str(exc)
+        return {
+            "source": "adanos",
+            "symbol": ticker,
+            "sources": list(results),
+            "requested_sources": _source_list(sources),
+            "days": day_count,
+            "end": end_text,
+            "results": results,
+            "errors": errors,
+        }
+
+    def get_market_sentiment(
+        self,
+        *,
+        sources: list[str] | tuple[str, ...] | str | None = None,
+        days: int = 7,
+        end: Any | None = None,
+    ) -> dict[str, Any]:
+        day_count = max(int(days), 1)
+        end_text = _date_text(end) or self._strategy_as_of().date().isoformat()
+        results: dict[str, Any] = {}
+        errors: dict[str, str] = {}
+        for source in _source_list(sources):
+            try:
+                results[source] = self._get_json(
+                    f"/{source}/stocks/v1/market-sentiment",
+                    {"days": day_count, "to": end_text},
+                )
+            except Exception as exc:
+                errors[source] = str(exc)
+        return {
+            "source": "adanos",
+            "sources": list(results),
+            "requested_sources": _source_list(sources),
+            "days": day_count,
+            "end": end_text,
+            "results": results,
+            "errors": errors,
+        }