feat: add social sentiment intelligence for US stocks (Reddit/X/Polymarket) (#705)

Integrate real-time social media sentiment data from api.adanos.org as an
additional intelligence source for US stock analysis. When configured, the
system fetches Reddit community sentiment, X/Twitter buzz, and Polymarket
prediction market data — and injects it alongside news context into the
LLM analysis prompt.

- New SocialSentimentService with retry, TTL caching, and graceful fallback
- Only activates for US stock tickers (A-shares/HK stocks are unaffected)
- Fully optional: requires SOCIAL_SENTIMENT_API_KEY env var
- Works in both standard and Agent analysis pipelines (via news_context)
- 22 unit tests covering all code paths including zero-value edge cases
- Updated README.md and docs/CHANGELOG.md

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>

.env.example

@@ -98,6 +98,15 @@ BRAVE_API_KEYS=
 # SearXNG 实例地址（逗号分隔，私有部署无配额，需在 settings.yml 启用 format: json）
 SEARXNG_BASE_URLS=
 
+# ===================================
+# Social Sentiment Intelligence (US stocks only)
+# ===================================
+# Reddit / X (Twitter) / Polymarket sentiment data from api.adanos.org
+# Only activates for US stock tickers (AAPL, TSLA, etc.), ignored for A-shares/HK stocks
+# Free tier: 250 requests/month | Register: https://api.adanos.org
+# SOCIAL_SENTIMENT_API_KEY=sk_live_your_key_here
+# SOCIAL_SENTIMENT_API_URL=https://api.adanos.org
+
 # ===================================
 # 新闻时效与分析筛选配置
 # ===================================

README.md

@@ -55,6 +55,7 @@
 | AI 模型 | [AIHubMix](https://aihubmix.com/?aff=CfMq)、Gemini、OpenAI 兼容、DeepSeek、通义千问、Claude 等（统一通过 [LiteLLM](https://github.com/BerriAI/litellm) 调用，支持多 Key 负载均衡）|
 | 行情数据 | AkShare、Tushare、Pytdx、Baostock、YFinance |
 | 新闻搜索 | Tavily、SerpAPI、Bocha、Brave、MiniMax |
+| 社交舆情 | [Stock Sentiment API](https://api.adanos.org/docs)（Reddit / X / Polymarket，仅美股，可选） |
 
 > 注：美股历史数据与实时行情统一使用 YFinance，确保复权一致性
 
@@ -158,6 +159,8 @@
 | `BOCHA_API_KEYS` | [博查搜索](https://open.bocha.cn/) Web Search API（中文搜索优化，支持AI摘要，多个key用逗号分隔） | 可选 |
 | `BRAVE_API_KEYS` | [Brave Search](https://brave.com/search/api/) API（隐私优先，美股优化，多个key用逗号分隔） | 可选 |
 | `SEARXNG_BASE_URLS` | SearXNG 自建实例（无配额兜底，需在 settings.yml 启用 format: json） | 可选 |
+| `SOCIAL_SENTIMENT_API_KEY` | [Stock Sentiment API](https://api.adanos.org/docs)（Reddit/X/Polymarket 社交舆情，仅美股） | 可选 |
+| `SOCIAL_SENTIMENT_API_URL` | 自定义社交舆情 API 地址（默认 `https://api.adanos.org`） | 可选 |
 | `TUSHARE_TOKEN` | [Tushare Pro](https://tushare.pro/weborder/#/login?reg=834638 ) Token | 可选 |
 | `PREFETCH_REALTIME_QUOTES` | 实时行情预取开关：设为 `false` 可禁用全市场预取（默认 `true`） | 可选 |
 | `WECHAT_MSG_TYPE` | 企微消息类型，默认 markdown，支持配置 text 类型，发送纯 markdown 文本 | 可选 |

docs/CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### 新功能
+
+- 📱 **Social Sentiment Intelligence (US stocks)** — 新增 Reddit / X (Twitter) / Polymarket 社交媒体情绪数据源，为美股分析提供实时社交舆情情报。数据来自 api.adanos.org，包含 Buzz Score、情绪评分、提及量等指标。完全可选（需配置 `SOCIAL_SENTIMENT_API_KEY`），仅对美股生效，A 股 / 港股不受影响。
 ### 文档
 
 - 新增云服务器 Web 界面部署与访问教程 (Fixes #686)

src/agent/executor.py

@@ -454,6 +454,8 @@ class AgentExecutor:
                 parts.append(f"\n[系统已获取的实时行情]\n{json.dumps(context['realtime_quote'], ensure_ascii=False)}")
             if context.get("chip_distribution"):
                 parts.append(f"\n[系统已获取的筹码分布]\n{json.dumps(context['chip_distribution'], ensure_ascii=False)}")
+            if context.get("news_context"):
+                parts.append(f"\n[系统已获取的新闻与舆情情报]\n{context['news_context']}")
 
         parts.append("\n请使用可用工具获取缺失的数据（如历史K线、新闻等），然后以决策仪表盘 JSON 格式输出分析结果。")
         return "\n".join(parts)

src/config.py

@@ -329,6 +329,10 @@ class Config:
     serpapi_keys: List[str] = field(default_factory=list)  # SerpAPI Keys
     searxng_base_urls: List[str] = field(default_factory=list)  # SearXNG instance URLs (self-hosted, no quota)
 
+    # === Social Sentiment (US stocks only, api.adanos.org) ===
+    social_sentiment_api_key: Optional[str] = None
+    social_sentiment_api_url: str = "https://api.adanos.org"
+
     # === 新闻与分析筛选配置 ===
     news_max_age_days: int = 3   # 新闻最大时效（天）
     bias_threshold: float = 5.0  # 乖离率阈值（%），超过此值提示不追高
@@ -889,6 +893,8 @@ class Config:
             brave_api_keys=brave_api_keys,
             serpapi_keys=serpapi_keys,
             searxng_base_urls=searxng_base_urls,
+            social_sentiment_api_key=os.getenv('SOCIAL_SENTIMENT_API_KEY') or None,
+            social_sentiment_api_url=os.getenv('SOCIAL_SENTIMENT_API_URL', 'https://api.adanos.org').rstrip('/'),
             news_max_age_days=max(1, int(os.getenv('NEWS_MAX_AGE_DAYS', '3'))),
             bias_threshold=max(1.0, float(os.getenv('BIAS_THRESHOLD', '5.0'))),
             agent_mode=os.getenv('AGENT_MODE', 'false').lower() == 'true',

src/core/pipeline.py

@@ -29,9 +29,11 @@ from src.analyzer import GeminiAnalyzer, AnalysisResult, fill_chip_structure_if_
 from src.data.stock_mapping import STOCK_NAME_MAP
 from src.notification import NotificationService, NotificationChannel
 from src.search_service import SearchService
+from src.services.social_sentiment_service import SocialSentimentService
 from src.enums import ReportType
 from src.stock_analyzer import StockTrendAnalyzer, TrendAnalysisResult
 from src.core.trading_calendar import get_market_for_stock, is_market_open
+from data_provider.us_index_mapping import is_us_stock_code
 from bot.models import BotMessage
 
 
@@ -107,6 +109,14 @@ class StockAnalysisPipeline:
         else:
             logger.warning("搜索服务未启用（未配置 API Key）")
 
+        # 初始化社交舆情服务（仅美股）
+        self.social_sentiment_service = SocialSentimentService(
+            api_key=self.config.social_sentiment_api_key,
+            api_url=self.config.social_sentiment_api_url,
+        )
+        if self.social_sentiment_service.is_available:
+            logger.info("Social sentiment service enabled (Reddit/X/Polymarket, US stocks only)")
+
     def fetch_and_save_stock_data(
         self, 
         code: str,
@@ -327,6 +337,19 @@ class StockAnalysisPipeline:
             else:
                 logger.info(f"{stock_name}({code}) 搜索服务不可用，跳过情报搜索")
 
+            # Step 4.5: Social sentiment intelligence (US stocks only)
+            if self.social_sentiment_service.is_available and is_us_stock_code(code):
+                try:
+                    social_context = self.social_sentiment_service.get_social_context(code)
+                    if social_context:
+                        logger.info(f"{stock_name}({code}) Social sentiment data retrieved")
+                        if news_context:
+                            news_context = news_context + "\n\n" + social_context
+                        else:
+                            news_context = social_context
+                except Exception as e:
+                    logger.warning(f"{stock_name}({code}) Social sentiment fetch failed: {e}")
+
             # Step 5: 获取分析上下文（技术面数据）
             context = self.db.get_analysis_context(code)
 
@@ -591,6 +614,22 @@ class StockAnalysisPipeline:
             if trend_result:
                 initial_context["trend_result"] = self._safe_to_dict(trend_result)
 
+            # Agent path: inject social sentiment as news_context so both
+            # executor (_build_user_message) and orchestrator (ctx.set_data)
+            # can consume it through the existing news_context channel
+            if self.social_sentiment_service.is_available and is_us_stock_code(code):
+                try:
+                    social_context = self.social_sentiment_service.get_social_context(code)
+                    if social_context:
+                        existing = initial_context.get("news_context")
+                        if existing:
+                            initial_context["news_context"] = existing + "\n\n" + social_context
+                        else:
+                            initial_context["news_context"] = social_context
+                        logger.info(f"[{code}] Agent mode: social sentiment data injected into news_context")
+                except Exception as e:
+                    logger.warning(f"[{code}] Agent mode: social sentiment fetch failed: {e}")
+
             # 运行 Agent
             message = f"请分析股票 {code} ({stock_name})，并生成决策仪表盘报告。"
             agent_result = executor.run(message, context=initial_context)

src/services/social_sentiment_service.py

@@ -0,0 +1,302 @@
+# -*- coding: utf-8 -*-
+"""
+===================================
+Social Sentiment Intelligence Service
+===================================
+
+Fetches Reddit / X (Twitter) / Polymarket social sentiment data
+from api.adanos.org for US stock tickers.
+
+Optional — requires SOCIAL_SENTIMENT_API_KEY.
+Only activates for US stock codes (AAPL, TSLA, etc.).
+"""
+
+import logging
+import time
+from typing import Any, Dict, List, Optional
+
+import requests
+from tenacity import (
+    retry,
+    stop_after_attempt,
+    wait_exponential,
+    retry_if_exception_type,
+    before_sleep_log,
+)
+
+logger = logging.getLogger(__name__)
+
+_TRANSIENT_EXCEPTIONS = (
+    requests.exceptions.SSLError,
+    requests.exceptions.ConnectionError,
+    requests.exceptions.Timeout,
+    requests.exceptions.ChunkedEncodingError,
+)
+
+_REQUEST_TIMEOUT = 8  # seconds
+
+
+@retry(
+    stop=stop_after_attempt(2),
+    wait=wait_exponential(multiplier=1, min=1, max=5),
+    retry=retry_if_exception_type(_TRANSIENT_EXCEPTIONS),
+    before_sleep=before_sleep_log(logger, logging.WARNING),
+    reraise=True,
+)
+def _get_with_retry(url: str, *, headers: Dict[str, str], params: Optional[Dict[str, Any]] = None,
+                    timeout: int = _REQUEST_TIMEOUT) -> requests.Response:
+    """GET with retry on transient network errors."""
+    return requests.get(url, headers=headers, params=params or {}, timeout=timeout)
+
+
+class SocialSentimentService:
+    """
+    Social Sentiment Intelligence — Reddit / X / Polymarket.
+
+    Fetches social-media sentiment data from api.adanos.org and formats
+    it as a text block suitable for injection into the LLM analysis prompt.
+
+    Usage::
+
+        svc = SocialSentimentService(api_key="sk_live_...", api_url="https://api.adanos.org")
+        if svc.is_available:
+            context = svc.get_social_context("TSLA")
+    """
+
+    # Cache TTL for trending endpoints (seconds)
+    _TRENDING_CACHE_TTL = 600  # 10 minutes
+
+    def __init__(self, api_key: Optional[str] = None, api_url: str = "https://api.adanos.org"):
+        self._api_key = (api_key or "").strip() or None
+        self._api_url = (api_url or "https://api.adanos.org").rstrip("/")
+        # Simple in-memory cache: {"key": (timestamp, data)}
+        self._cache: Dict[str, tuple] = {}
+
+    @property
+    def is_available(self) -> bool:
+        return self._api_key is not None
+
+    @property
+    def _headers(self) -> Dict[str, str]:
+        return {"X-API-Key": self._api_key or "", "Accept": "application/json"}
+
+    # ------------------------------------------------------------------
+    # API calls
+    # ------------------------------------------------------------------
+
+    def _fetch_json(self, url: str, params: Optional[Dict[str, Any]] = None) -> Optional[Dict]:
+        """Fetch JSON from API, return None on any error."""
+        try:
+            resp = _get_with_retry(url, headers=self._headers, params=params)
+            if resp.status_code == 200:
+                return resp.json()
+            logger.warning("Social sentiment API %s returned %s", url, resp.status_code)
+        except _TRANSIENT_EXCEPTIONS as e:
+            logger.warning("Social sentiment API %s network error: %s", url, e)
+        except Exception as e:
+            logger.warning("Social sentiment API %s unexpected error: %s", url, e)
+        return None
+
+    def _fetch_cached(self, cache_key: str, url: str, params: Optional[Dict[str, Any]] = None) -> Optional[Any]:
+        """Fetch with simple TTL cache (for trending endpoints)."""
+        now = time.monotonic()
+        cached = self._cache.get(cache_key)
+        if cached and (now - cached[0]) < self._TRENDING_CACHE_TTL:
+            return cached[1]
+        data = self._fetch_json(url, params)
+        if data is not None:
+            self._cache[cache_key] = (now, data)
+        return data
+
+    def fetch_reddit_report(self, ticker: str) -> Optional[Dict]:
+        """Fetch detailed Reddit report for a single ticker."""
+        url = f"{self._api_url}/reddit/stocks/v1/report/{ticker.upper()}"
+        return self._fetch_json(url)
+
+    def fetch_reddit_trending(self) -> Optional[List[Dict]]:
+        """Fetch Reddit trending stocks (cached)."""
+        url = f"{self._api_url}/reddit/stocks/v1/trending"
+        data = self._fetch_cached("reddit_trending", url)
+        if isinstance(data, dict):
+            return data.get("trending", data.get("data", []))
+        if isinstance(data, list):
+            return data
+        return None
+
+    def fetch_x_trending(self) -> Optional[List[Dict]]:
+        """Fetch X/Twitter trending stocks (cached)."""
+        url = f"{self._api_url}/x/stocks/v1/trending"
+        data = self._fetch_cached("x_trending", url)
+        if isinstance(data, dict):
+            return data.get("trending", data.get("data", []))
+        if isinstance(data, list):
+            return data
+        return None
+
+    def fetch_polymarket_trending(self) -> Optional[List[Dict]]:
+        """Fetch Polymarket trending stocks (cached)."""
+        url = f"{self._api_url}/polymarket/stocks/v1/trending"
+        data = self._fetch_cached("polymarket_trending", url)
+        if isinstance(data, dict):
+            return data.get("trending", data.get("data", []))
+        if isinstance(data, list):
+            return data
+        return None
+
+    # ------------------------------------------------------------------
+    # Main entry point
+    # ------------------------------------------------------------------
+
+    def get_social_context(self, ticker: str) -> Optional[str]:
+        """
+        Fetch social sentiment from all platforms and return a formatted
+        text block for the LLM prompt.  Returns None if no data found.
+        """
+        if not self.is_available:
+            return None
+
+        ticker_upper = ticker.upper()
+
+        # 1. Reddit per-ticker report (richest data)
+        reddit_data = self.fetch_reddit_report(ticker_upper)
+
+        # 2. X trending (filter for this ticker)
+        x_entry = None
+        x_trending = self.fetch_x_trending()
+        if x_trending:
+            x_entry = self._find_ticker_in_trending(x_trending, ticker_upper)
+
+        # 3. Polymarket trending (filter for this ticker)
+        poly_entry = None
+        poly_trending = self.fetch_polymarket_trending()
+        if poly_trending:
+            poly_entry = self._find_ticker_in_trending(poly_trending, ticker_upper)
+
+        # If no data from any source, skip
+        if not reddit_data and not x_entry and not poly_entry:
+            return None
+
+        return self._format_social_intel(ticker_upper, reddit_data, x_entry, poly_entry)
+
+    # ------------------------------------------------------------------
+    # Formatting
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _find_ticker_in_trending(trending: List[Dict], ticker: str) -> Optional[Dict]:
+        """Find a ticker entry in a trending list."""
+        for entry in trending:
+            code = (entry.get("ticker") or entry.get("symbol") or entry.get("code") or "").upper()
+            if code == ticker:
+                return entry
+        return None
+
+    @staticmethod
+    def _coalesce(*values):
+        """Return the first value that is not None (preserves 0 and 0.0)."""
+        for v in values:
+            if v is not None:
+                return v
+        return None
+
+    @staticmethod
+    def _format_social_intel(
+        ticker: str,
+        reddit_data: Optional[Dict],
+        x_entry: Optional[Dict],
+        poly_entry: Optional[Dict],
+    ) -> str:
+        """Format social sentiment data as a prompt-ready text block."""
+        lines = [f"📱 Social Sentiment Intelligence for {ticker} (Reddit / X / Polymarket)"]
+        lines.append("=" * 60)
+
+        # --- Reddit ---
+        if reddit_data:
+            lines.append("\n🔴 Reddit Community Sentiment:")
+            report = reddit_data.get("report", reddit_data)
+
+            # Buzz score
+            buzz = SocialSentimentService._coalesce(report.get("buzz_score"), report.get("buzz"))
+            if buzz is not None:
+                trend_label = report.get("trend", "")
+                lines.append(f"  Buzz Score: {buzz}/100 ({trend_label})" if trend_label
+                             else f"  Buzz Score: {buzz}/100")
+
+            # Sentiment (0 is a valid neutral value, must not be dropped)
+            sentiment = SocialSentimentService._coalesce(report.get("sentiment_score"), report.get("sentiment"))
+            if sentiment is not None:
+                lines.append(f"  Sentiment Score: {sentiment}")
+
+            # Mentions
+            mentions = SocialSentimentService._coalesce(report.get("total_mentions"), report.get("mentions"))
+            if mentions is not None:
+                subs = SocialSentimentService._coalesce(report.get("subreddit_count"), report.get("subreddits"))
+                sub_str = f" across {subs} subreddits" if subs else ""
+                lines.append(f"  Mentions: {mentions}{sub_str} (7-day)")
+
+            # Top mentions
+            top_mentions = report.get("top_mentions", [])
+            if top_mentions:
+                lines.append("  Top Mentions:")
+                for i, m in enumerate(top_mentions[:5], 1):
+                    text = (m.get("text") or m.get("title") or "")[:120]
+                    sub = m.get("subreddit", "")
+                    score = SocialSentimentService._coalesce(m.get("sentiment_score"), m.get("sentiment"))
+                    upvotes = m.get("upvotes", "")
+                    meta_parts = []
+                    if score is not None:
+                        meta_parts.append(f"sentiment: {score}")
+                    if sub:
+                        meta_parts.append(f"r/{sub}")
+                    if upvotes:
+                        meta_parts.append(f"{upvotes} upvotes")
+                    meta = f" ({', '.join(meta_parts)})" if meta_parts else ""
+                    lines.append(f"    {i}. \"{text}\"{meta}")
+
+            # Daily stats
+            daily = report.get("daily_stats", [])
+            if daily:
+                lines.append("  Recent Daily Activity:")
+                for d in daily[:5]:
+                    day = d.get("date", "")
+                    day_mentions = d.get("mentions", "?")
+                    day_sentiment = d.get("avg_sentiment", "?")
+                    lines.append(f"    {day}: {day_mentions} mentions, avg sentiment {day_sentiment}")
+        else:
+            lines.append("\n🔴 Reddit: No data available")
+
+        # --- X / Twitter ---
+        if x_entry:
+            lines.append("\n🐦 X (Twitter) Sentiment:")
+            x_buzz = SocialSentimentService._coalesce(x_entry.get("buzz_score"), x_entry.get("buzz"))
+            x_sentiment = SocialSentimentService._coalesce(x_entry.get("sentiment_score"), x_entry.get("sentiment"))
+            x_mentions = SocialSentimentService._coalesce(x_entry.get("total_mentions"), x_entry.get("mentions"))
+            x_trend = x_entry.get("trend", "")
+            if x_buzz is not None:
+                lines.append(f"  Buzz Score: {x_buzz}/100 ({x_trend})" if x_trend
+                             else f"  Buzz Score: {x_buzz}/100")
+            if x_sentiment is not None:
+                lines.append(f"  Sentiment Score: {x_sentiment}")
+            if x_mentions is not None:
+                lines.append(f"  Mentions: {x_mentions} (7-day)")
+        else:
+            lines.append("\n🐦 X (Twitter): No data available")
+
+        # --- Polymarket ---
+        if poly_entry:
+            lines.append("\n🔮 Polymarket (Prediction Markets):")
+            poly_buzz = SocialSentimentService._coalesce(poly_entry.get("buzz_score"), poly_entry.get("buzz"))
+            poly_sentiment = SocialSentimentService._coalesce(poly_entry.get("sentiment_score"), poly_entry.get("sentiment"))
+            poly_trades = SocialSentimentService._coalesce(poly_entry.get("trade_count"), poly_entry.get("trades"))
+            if poly_buzz is not None:
+                lines.append(f"  Buzz Score: {poly_buzz}/100")
+            if poly_sentiment is not None:
+                lines.append(f"  Market Sentiment: {poly_sentiment}")
+            if poly_trades is not None:
+                lines.append(f"  Trade Count: {poly_trades}")
+        else:
+            lines.append("\n🔮 Polymarket: No active prediction markets found")
+
+        lines.append("")
+        lines.append("Source: api.adanos.org — Real-time social sentiment aggregation")
+        return "\n".join(lines)

tests/test_social_sentiment_service.py

@@ -0,0 +1,269 @@
+# -*- coding: utf-8 -*-
+"""Tests for SocialSentimentService."""
+
+import time
+import unittest
+from unittest.mock import patch, MagicMock
+
+from src.services.social_sentiment_service import SocialSentimentService
+
+
+class TestServiceAvailability(unittest.TestCase):
+    """Tests for is_available property."""
+
+    def test_unavailable_without_key(self):
+        svc = SocialSentimentService(api_key=None)
+        self.assertFalse(svc.is_available)
+
+    def test_unavailable_with_empty_key(self):
+        svc = SocialSentimentService(api_key="  ")
+        self.assertFalse(svc.is_available)
+
+    def test_available_with_key(self):
+        svc = SocialSentimentService(api_key="sk_live_test123")
+        self.assertTrue(svc.is_available)
+
+
+class TestFetchRedditReport(unittest.TestCase):
+    """Tests for fetch_reddit_report."""
+
+    def setUp(self):
+        self.svc = SocialSentimentService(api_key="sk_live_test", api_url="https://api.example.com")
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_success(self, mock_get):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {
+            "report": {
+                "buzz_score": 85.5,
+                "sentiment_score": 0.23,
+                "total_mentions": 342,
+                "subreddit_count": 8,
+                "trend": "rising",
+            }
+        }
+        mock_get.return_value = mock_resp
+
+        result = self.svc.fetch_reddit_report("TSLA")
+        self.assertIsNotNone(result)
+        self.assertEqual(result["report"]["buzz_score"], 85.5)
+        mock_get.assert_called_once()
+        call_args = mock_get.call_args
+        self.assertIn("/reddit/stocks/v1/report/TSLA", call_args[0][0])
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_http_error_returns_none(self, mock_get):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 404
+        mock_get.return_value = mock_resp
+
+        result = self.svc.fetch_reddit_report("UNKNOWN")
+        self.assertIsNone(result)
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_timeout_returns_none(self, mock_get):
+        import requests
+        mock_get.side_effect = requests.exceptions.Timeout("timed out")
+
+        result = self.svc.fetch_reddit_report("AAPL")
+        self.assertIsNone(result)
+
+
+class TestFetchTrending(unittest.TestCase):
+    """Tests for trending endpoints with caching."""
+
+    def setUp(self):
+        self.svc = SocialSentimentService(api_key="sk_live_test", api_url="https://api.example.com")
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_x_trending_returns_list(self, mock_get):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {
+            "trending": [
+                {"ticker": "AAPL", "buzz_score": 72.1, "sentiment_score": 0.15},
+                {"ticker": "TSLA", "buzz_score": 65.0, "sentiment_score": -0.05},
+            ]
+        }
+        mock_get.return_value = mock_resp
+
+        result = self.svc.fetch_x_trending()
+        self.assertIsNotNone(result)
+        self.assertEqual(len(result), 2)
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_trending_cache_prevents_duplicate_calls(self, mock_get):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {"trending": [{"ticker": "AAPL"}]}
+        mock_get.return_value = mock_resp
+
+        # First call hits API
+        self.svc.fetch_x_trending()
+        # Second call should use cache
+        self.svc.fetch_x_trending()
+
+        self.assertEqual(mock_get.call_count, 1)
+
+
+class TestGetSocialContext(unittest.TestCase):
+    """Tests for get_social_context (main entry point)."""
+
+    def test_returns_none_when_unavailable(self):
+        svc = SocialSentimentService(api_key=None)
+        result = svc.get_social_context("AAPL")
+        self.assertIsNone(result)
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_returns_none_when_no_data(self, mock_get):
+        mock_resp = MagicMock()
+        mock_resp.status_code = 404
+        mock_get.return_value = mock_resp
+
+        svc = SocialSentimentService(api_key="sk_live_test")
+        result = svc.get_social_context("XYZZY")
+        self.assertIsNone(result)
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_formats_reddit_data(self, mock_get):
+        def side_effect(url, **kwargs):
+            resp = MagicMock()
+            if "/report/" in url:
+                resp.status_code = 200
+                resp.json.return_value = {
+                    "report": {
+                        "buzz_score": 87.5,
+                        "sentiment_score": 0.23,
+                        "total_mentions": 342,
+                        "subreddit_count": 8,
+                        "trend": "rising",
+                        "top_mentions": [
+                            {"text": "TSLA looking strong", "subreddit": "wallstreetbets", "upvotes": 1234}
+                        ],
+                    }
+                }
+            else:
+                resp.status_code = 200
+                resp.json.return_value = {"trending": []}
+            return resp
+
+        mock_get.side_effect = side_effect
+
+        svc = SocialSentimentService(api_key="sk_live_test")
+        result = svc.get_social_context("TSLA")
+
+        self.assertIsNotNone(result)
+        self.assertIn("Social Sentiment Intelligence", result)
+        self.assertIn("Reddit", result)
+        self.assertIn("87.5", result)
+        self.assertIn("342", result)
+        self.assertIn("TSLA looking strong", result)
+
+    @patch("src.services.social_sentiment_service._get_with_retry")
+    def test_includes_all_platforms(self, mock_get):
+        def side_effect(url, **kwargs):
+            resp = MagicMock()
+            resp.status_code = 200
+            if "/report/" in url:
+                resp.json.return_value = {"report": {"buzz_score": 80, "trend": "rising"}}
+            elif "/x/" in url:
+                resp.json.return_value = {"trending": [{"ticker": "AAPL", "buzz_score": 65}]}
+            elif "/polymarket/" in url:
+                resp.json.return_value = {"trending": [{"ticker": "AAPL", "buzz_score": 45, "trade_count": 120}]}
+            else:
+                resp.json.return_value = {"trending": []}
+            return resp
+
+        mock_get.side_effect = side_effect
+
+        svc = SocialSentimentService(api_key="sk_live_test")
+        result = svc.get_social_context("AAPL")
+
+        self.assertIsNotNone(result)
+        self.assertIn("Reddit", result)
+        self.assertIn("X (Twitter)", result)
+        self.assertIn("Polymarket", result)
+        self.assertIn("65", result)  # X buzz
+        self.assertIn("120", result)  # Polymarket trades
+
+
+class TestZeroValueHandling(unittest.TestCase):
+    """Verify that zero-valued numeric fields (e.g. neutral sentiment) are preserved."""
+
+    def test_zero_sentiment_preserved_in_reddit(self):
+        result = SocialSentimentService._format_social_intel(
+            "AAPL",
+            reddit_data={"report": {"buzz_score": 50, "sentiment_score": 0, "total_mentions": 10}},
+            x_entry=None,
+            poly_entry=None,
+        )
+        self.assertIn("Sentiment Score: 0", result)
+        self.assertIn("Buzz Score: 50", result)
+        self.assertIn("Mentions: 10", result)
+
+    def test_zero_buzz_preserved_in_x(self):
+        result = SocialSentimentService._format_social_intel(
+            "AAPL",
+            reddit_data=None,
+            x_entry={"buzz_score": 0, "sentiment_score": 0.0, "total_mentions": 0},
+            poly_entry=None,
+        )
+        self.assertIn("Buzz Score: 0/100", result)
+        self.assertIn("Sentiment Score: 0.0", result)
+        self.assertIn("Mentions: 0", result)
+
+    def test_coalesce_preserves_zero(self):
+        self.assertEqual(SocialSentimentService._coalesce(0, 5), 0)
+        self.assertEqual(SocialSentimentService._coalesce(0.0, 1.0), 0.0)
+        self.assertEqual(SocialSentimentService._coalesce(None, 0), 0)
+        self.assertIsNone(SocialSentimentService._coalesce(None, None))
+
+
+class TestFindTickerInTrending(unittest.TestCase):
+    """Tests for _find_ticker_in_trending helper."""
+
+    def test_finds_by_ticker_field(self):
+        trending = [{"ticker": "AAPL", "buzz": 80}, {"ticker": "TSLA", "buzz": 60}]
+        result = SocialSentimentService._find_ticker_in_trending(trending, "TSLA")
+        self.assertIsNotNone(result)
+        self.assertEqual(result["buzz"], 60)
+
+    def test_finds_by_symbol_field(self):
+        trending = [{"symbol": "AAPL", "buzz": 80}]
+        result = SocialSentimentService._find_ticker_in_trending(trending, "AAPL")
+        self.assertIsNotNone(result)
+
+    def test_returns_none_when_not_found(self):
+        trending = [{"ticker": "AAPL", "buzz": 80}]
+        result = SocialSentimentService._find_ticker_in_trending(trending, "MSFT")
+        self.assertIsNone(result)
+
+    def test_case_insensitive_match(self):
+        trending = [{"ticker": "aapl", "buzz": 80}]
+        result = SocialSentimentService._find_ticker_in_trending(trending, "AAPL")
+        self.assertIsNotNone(result)
+
+
+class TestUSStockGating(unittest.TestCase):
+    """Verify that only US stock codes are processed."""
+
+    def test_a_share_code_not_us(self):
+        from data_provider.us_index_mapping import is_us_stock_code
+        self.assertFalse(is_us_stock_code("600519"))
+        self.assertFalse(is_us_stock_code("000001"))
+        self.assertFalse(is_us_stock_code("300750"))
+
+    def test_hk_code_not_us(self):
+        from data_provider.us_index_mapping import is_us_stock_code
+        self.assertFalse(is_us_stock_code("HK00700"))
+
+    def test_us_code_detected(self):
+        from data_provider.us_index_mapping import is_us_stock_code
+        self.assertTrue(is_us_stock_code("AAPL"))
+        self.assertTrue(is_us_stock_code("TSLA"))
+        self.assertTrue(is_us_stock_code("NVDA"))
+
+
+if __name__ == "__main__":
+    unittest.main()