feat(catalog+t28): wire all 5 stores + news intelligence product

backend/app/catalog/init_schema.py

@@ -0,0 +1,172 @@
+"""Catalog database schema — initial migration.
+
+Creates the tables referenced by the v4.0 catalog models.
+Idempotent: safe to run multiple times.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+
+import asyncpg
+
+
+SCHEMA = """
+-- Tokens (T27)
+CREATE TABLE IF NOT EXISTS tokens (
+    token_id           TEXT PRIMARY KEY,
+    chain              TEXT NOT NULL,
+    address            TEXT NOT NULL,
+    symbol             TEXT,
+    name               TEXT,
+    decimals           INT,
+    deployer_wallet_id TEXT,
+    deployed_at        TIMESTAMPTZ NOT NULL,
+    initial_supply     BIGINT,
+    current_supply     BIGINT,
+    is_honeypot        BOOLEAN,
+    is_mintable        BOOLEAN,
+    is_proxy           BOOLEAN,
+    tax_buy_bps        INT,
+    tax_sell_bps       INT,
+    risk_tier          TEXT,
+    risk_score         INT,
+    risk_factors       TEXT[],
+    rag_embedding_id   TEXT,
+    updated_at         TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS idx_tokens_chain ON tokens(chain);
+CREATE INDEX IF NOT EXISTS idx_tokens_deployer ON tokens(deployer_wallet_id);
+CREATE INDEX IF NOT EXISTS idx_tokens_risk ON tokens(risk_score DESC) WHERE risk_score IS NOT NULL;
+
+-- Alerts
+CREATE TABLE IF NOT EXISTS alerts (
+    alert_id     TEXT PRIMARY KEY,
+    token_id     TEXT,
+    wallet_id    TEXT,
+    chain        TEXT,
+    alert_type   TEXT NOT NULL,
+    severity     TEXT NOT NULL,
+    title        TEXT NOT NULL,
+    description  TEXT,
+    evidence     JSONB DEFAULT '{}'::jsonb,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    resolved_at  TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS idx_alerts_token ON alerts(token_id);
+CREATE INDEX IF NOT EXISTS idx_alerts_wallet ON alerts(wallet_id);
+CREATE INDEX IF NOT EXISTS idx_alerts_created ON alerts(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_alerts_severity ON alerts(severity);
+
+-- News items (T28)
+CREATE TABLE IF NOT EXISTS news_items (
+    news_id          TEXT PRIMARY KEY,
+    url              TEXT NOT NULL,
+    title            TEXT NOT NULL,
+    summary          TEXT,
+    body_markdown    TEXT,
+    source           TEXT NOT NULL,
+    published_at     TIMESTAMPTZ NOT NULL,
+    ingested_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    chains_mentioned TEXT[] DEFAULT ARRAY[]::TEXT[],
+    tokens_mentioned TEXT[] DEFAULT ARRAY[]::TEXT[],
+    wallets_mentioned TEXT[] DEFAULT ARRAY[]::TEXT[],
+    sentiment_score  REAL,
+    ai_analysis      TEXT,
+    rag_embedding_id TEXT,
+    body_tsv         tsvector
+);
+CREATE INDEX IF NOT EXISTS idx_news_published ON news_items(published_at DESC);
+CREATE INDEX IF NOT EXISTS idx_news_source ON news_items(source);
+CREATE INDEX IF NOT EXISTS idx_news_sentiment ON news_items(sentiment_score);
+CREATE INDEX IF NOT EXISTS idx_news_body_tsv ON news_items USING gin(body_tsv);
+
+-- Auto-update the body_tsv column on insert/update
+CREATE OR REPLACE FUNCTION news_items_tsv_update() RETURNS trigger AS $$
+BEGIN
+    NEW.body_tsv := to_tsvector('english', COALESCE(NEW.title, '') || ' ' ||
+                                COALESCE(NEW.summary, '') || ' ' ||
+                                COALESCE(NEW.body_markdown, ''));
+    RETURN NEW;
+END
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS news_items_tsv_trigger ON news_items;
+CREATE TRIGGER news_items_tsv_trigger
+BEFORE INSERT OR UPDATE ON news_items
+FOR EACH ROW EXECUTE FUNCTION news_items_tsv_update();
+
+-- Scan reports (T29)
+CREATE TABLE IF NOT EXISTS scan_reports (
+    report_id          TEXT PRIMARY KEY,
+    subject_type       TEXT NOT NULL,
+    subject_id         TEXT NOT NULL,
+    generated_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    generated_by_model TEXT NOT NULL,
+    risk_score         INT NOT NULL,
+    risk_tier          TEXT NOT NULL,
+    sections           JSONB DEFAULT '{}'::jsonb,
+    markdown_url       TEXT,
+    paid_via_x402      TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_reports_subject ON scan_reports(subject_type, subject_id);
+CREATE INDEX IF NOT EXISTS idx_reports_generated ON scan_reports(generated_at DESC);
+
+-- RAG findings metadata (Qdrant has the vectors; this is metadata for query)
+CREATE TABLE IF NOT EXISTS rag_findings (
+    finding_id      TEXT PRIMARY KEY,
+    source_type     TEXT NOT NULL,
+    source_url      TEXT,
+    source_token_id TEXT,
+    source_wallet_id TEXT,
+    claim           TEXT NOT NULL,
+    confidence      REAL NOT NULL,
+    extracted_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    qdrant_point_id TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_findings_token ON rag_findings(source_token_id);
+CREATE INDEX IF NOT EXISTS idx_findings_wallet ON rag_findings(source_wallet_id);
+
+-- x402 receipts (T34)
+CREATE TABLE IF NOT EXISTS x402_receipts (
+    tx_hash     TEXT PRIMARY KEY,
+    agent_id    TEXT,
+    tool        TEXT NOT NULL,
+    amount_usd  REAL NOT NULL,
+    chain       TEXT,
+    paid_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    tier        TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_x402_paid_at ON x402_receipts(paid_at DESC);
+CREATE INDEX IF NOT EXISTS idx_x402_tool ON x402_receipts(tool);
+CREATE INDEX IF NOT EXISTS idx_x402_agent ON x402_receipts(agent_id);
+"""
+
+
+async def main():
+    """Run the schema migration."""
+    cfg = {
+        "host": os.getenv("POSTGRES_HOST", "rmi-postgres"),
+        "port": int(os.getenv("POSTGRES_PORT", "5432")),
+        "user": os.getenv("POSTGRES_USER", "rmi"),
+        "password": os.getenv("POSTGRES_PASSWORD", "RMI_PROD_POSTGRES_2026"),
+        "database": os.getenv("POSTGRES_DB", "rmi"),
+    }
+    print(f"Connecting to postgres at {cfg['host']}:{cfg['port']} db={cfg['database']}")
+    conn = await asyncpg.connect(**cfg)
+    try:
+        await conn.execute(SCHEMA)
+        print("✓ Schema applied")
+        # Verify
+        tables = await conn.fetch(
+            "SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY tablename"
+        )
+        print(f"Tables ({len(tables)}):")
+        for t in tables:
+            print(f"  {t['tablename']}")
+    finally:
+        await conn.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

backend/app/catalog/service.py

@@ -155,23 +155,38 @@ class CatalogService:
                 log.warning("catalog_postgres_fail: %s", e)
                 self._health.postgres = False
 
-            # Neo4j
+            # Neo4j (NEO4J_AUTH=none means no password)
             try:
                 from neo4j import GraphDatabase
 
                 cfg = self.config["neo4j"]
-                if cfg["password"]:
-                    self._neo_driver = GraphDatabase.driver(
-                        cfg["uri"], auth=(cfg["user"], cfg["password"])
-                    )
-                    with self._neo_driver.session() as s:
-                        s.run("RETURN 1").consume()
-                    self._health.neo4j = True
-                    log.info("catalog_neo4j_ok")
+                auth = (
+                    (cfg["user"], cfg["password"]) if cfg["password"] else None
+                )
+                self._neo_driver = GraphDatabase.driver(cfg["uri"], auth=auth)
+                with self._neo_driver.session() as s:
+                    s.run("RETURN 1").consume()
+                self._health.neo4j = True
+                log.info("catalog_neo4j_ok")
             except Exception as e:
                 log.warning("catalog_neo4j_fail: %s", e)
                 self._health.neo4j = False
 
+            # MinIO (HTTP health probe)
+            try:
+                import socket
+
+                host, port = self.config["minio"]["endpoint"].rsplit(":", 1)
+                s = socket.socket()
+                s.settimeout(3)
+                s.connect((host, int(port)))
+                s.close()
+                self._health.minio = True
+                log.info("catalog_minio_ok")
+            except Exception as e:
+                log.warning("catalog_minio_fail: %s", e)
+                self._health.minio = False
+
             # Qdrant (HTTP)
             try:
                 cfg = self.config["qdrant"]

backend/app/domain/__init__.py

@@ -1,15 +1 @@
-"""Pure business logic. NO FastAPI imports allowed.
-
-Per-domain packages (each follows: models.py + service.py + helpers):
-    - scanner:  token scanning, honeypot, rugcheck, holders, contract
-    - wallet:   wallet analysis, labels, behavior
-    - token:    token discovery, supply, metadata
-    - rag:      embeddings, search, ingest, firehose, feedback, agentic
-    - x402:     facilitator, settlement, enforcement
-    - intel:    feeds, narratives, graph
-    - scam:     classifier, patterns
-    - databus:  client, chain registry (96 chains)
-    - bulletin: user-generated content board
-
-Migration order: alerts (smoke test) → wallet → token → scanner → x402 → rag.
-"""
+"""domain package — HTTP layer per v4.0 (thin routes, no business logic)."""

backend/app/domain/news/__init__.py

@@ -0,0 +1,4 @@
+"""T28 News Intelligence — thin HTTP layer."""
+from .router import router
+
+__all__ = ["router"]

backend/app/domain/news/router.py

@@ -0,0 +1,375 @@
+"""T28 News Intelligence Product.
+
+Per v4.0 §T28. Four endpoints surface the news pipeline as a product:
+  POST /api/v1/news              list, paginated, filterable
+  GET  /api/v1/news/trending     time-decay weighted
+  GET  /api/v1/news/{news_id}    single item
+  POST /api/v1/news/{news_id}/analyze  LLM analysis (LiteLLM)
+
+Data sources:
+  - crypto_news (legacy table, 1750+ items from RSS feeds)
+  - news_items (new catalog table, populated by RSS ingest)
+  - Qdrant embeddings for semantic search (rag_embedding_id)
+
+Time-decay scoring for /trending:
+  score = recency_decay * source_authority * social_velocity * abs(sentiment)
+  recency_decay = 0.5 ** (hours_old / 6)        half-life 6h
+  source_authority: tier-1 (CoinDesk, The Block) = 1.0, tier-2 = 0.5, tier-3 = 0.2
+  social_velocity: tweets/shares in last hour (1 + n/100, capped at 2x)
+  abs(sentiment): polarizing news ranks higher
+"""
+from __future__ import annotations
+
+import math
+import os
+from datetime import datetime, UTC, timedelta
+from typing import Any, Optional
+
+from fastapi import APIRouter, HTTPException, Query
+from pydantic import BaseModel, ConfigDict, Field
+
+from app.catalog.llm_router import LLMRouter
+from app.catalog.models import utcnow
+from app.catalog.service import get_catalog
+
+router = APIRouter(prefix="/api/v1/news", tags=["news"])
+
+
+# ── Time-decay scoring (per v4.0 §T28) ────────────────────────────
+SOURCE_AUTHORITY: dict[str, float] = {
+    # Tier 1 — major crypto-native outlets
+    "coindesk": 1.0,
+    "the block": 1.0,
+    "decrypt": 1.0,
+    "cointelegraph": 1.0,
+    # Tier 2 — solid crypto coverage
+    "beincrypto": 0.7,
+    "u.today": 0.7,
+    "crypto.news": 0.7,
+    "blockworks": 0.8,
+    # Tier 3 — general / RSS aggregators
+    "google-crypto": 0.5,
+    "reddit-crypto": 0.4,
+    "twitter-crypto": 0.3,
+}
+
+
+def recency_decay(hours_old: float, half_life: float = 6.0) -> float:
+    """Exponential decay: 1.0 at 0h, 0.5 at half_life, 0.25 at 2*half_life."""
+    if hours_old < 0:
+        return 1.0
+    return 0.5 ** (hours_old / half_life)
+
+
+def source_authority(source: str) -> float:
+    s = (source or "").lower().strip()
+    for key, val in SOURCE_AUTHORITY.items():
+        if key in s:
+            return val
+    return 0.3  # unknown source
+
+
+def trend_score(
+    hours_old: float, source: str, sentiment: float | None, social_velocity: int = 0
+) -> float:
+    """Composite trending score per v4.0 formula."""
+    s_auth = source_authority(source)
+    s_vel = min(2.0, 1.0 + social_velocity / 100.0)
+    s_sent = 1.0 + abs(sentiment or 0.0)
+    return recency_decay(hours_old) * s_auth * s_vel * s_sent
+
+
+# ── Response models ────────────────────────────────────────────────
+class NewsItemOut(BaseModel):
+    model_config = ConfigDict(strict=False)  # accept None for any field with default
+
+    news_id: Optional[str] = None
+    url: Optional[str] = None
+    title: Optional[str] = None
+    summary: Optional[str] = ""
+    source: Optional[str] = "unknown"
+    published_at: Optional[datetime] = None
+    chains_mentioned: list[str] = Field(default_factory=list)
+    tokens_mentioned: list[str] = Field(default_factory=list)
+    sentiment_score: Optional[float] = None
+    score: Optional[float] = None  # only set for /trending
+
+
+class NewsListResponse(BaseModel):
+    items: list[NewsItemOut]
+    total: int
+    offset: int
+
+
+class NewsAnalysisResponse(BaseModel):
+    news_id: str
+    analysis: str | None
+    model: str | None = None
+    error: str | None = None
+
+
+# ── Row adapter (legacy crypto_news → NewsItemOut) ───────────────
+def _adapt_legacy_row(row: dict) -> NewsItemOut:
+    """Convert a crypto_news row to NewsItemOut shape."""
+    # published is a text field in legacy; try ISO parse
+    pub_str = row.get("published") or row.get("ingested_at")
+    pub_dt = utcnow()
+    if pub_str:
+        try:
+            if isinstance(pub_str, (int, float)):
+                pub_dt = datetime.fromtimestamp(float(pub_str), tz=UTC)
+            else:
+                # Try common formats
+                for fmt in (
+                    "%Y-%m-%dT%H:%M:%S.%fZ",
+                    "%Y-%m-%dT%H:%M:%SZ",
+                    "%Y-%m-%dT%H:%M:%S",
+                    "%Y-%m-%d %H:%M:%S",
+                ):
+                    try:
+                        pub_dt = datetime.strptime(str(pub_str)[:19], fmt).replace(tzinfo=UTC)
+                        break
+                    except ValueError:
+                        continue
+        except Exception:
+            pass
+    return NewsItemOut(
+        news_id=row.get("id", ""),
+        url=row.get("url", ""),
+        title=row.get("title", ""),
+        summary=(row.get("content") or "")[:500],
+        source=row.get("source", "unknown"),
+        published_at=pub_dt,
+        chains_mentioned=[],
+        tokens_mentioned=row.get("tickers") or [],
+        sentiment_score=row.get("sentiment"),
+    )
+
+
+# ── POST /api/v1/news (list with filters) ─────────────────────────
+@router.post("", response_model=NewsListResponse)
+async def list_news(
+    chain: str | None = None,
+    token: str | None = None,
+    category: str | None = None,
+    since_hours: int = Query(24, ge=1, le=720),
+    limit: int = Query(20, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+    sort: str = Query("recency", pattern="^(recency|relevance|sentiment)$"),
+) -> NewsListResponse:
+    """List news items with filters. Reads from both news_items (new) and crypto_news (legacy)."""
+    catalog = get_catalog()
+    await catalog._init_stores()
+    items: list[NewsItemOut] = []
+
+    # New table
+    if catalog._health.postgres:
+        try:
+            cutoff = utcnow() - timedelta(hours=since_hours)
+            query = "SELECT news_id, url, title, summary, source, published_at, sentiment_score, chains_mentioned, tokens_mentioned FROM news_items WHERE published_at > $1"
+            params: list[Any] = [cutoff]
+            if chain:
+                query += f" AND ${len(params)+1} = ANY(chains_mentioned)"
+                params.append(chain)
+            if token:
+                query += f" AND ${len(params)+1} = ANY(tokens_mentioned)"
+                params.append(token)
+            if sort == "sentiment":
+                query += " ORDER BY sentiment_score ASC NULLS LAST"
+            else:
+                query += " ORDER BY published_at DESC"
+            query += f" LIMIT {limit} OFFSET {offset}"
+            async with catalog._pg_pool.acquire() as conn:
+                rows = await conn.fetch(query, *params)
+            for r in rows:
+                items.append(
+                    NewsItemOut(
+                        news_id=r["news_id"],
+                        url=r["url"],
+                        title=r["title"],
+                        summary=r["summary"] or "",
+                        source=r["source"],
+                        published_at=r["published_at"],
+                        chains_mentioned=list(r["chains_mentioned"] or []),
+                        tokens_mentioned=list(r["tokens_mentioned"] or []),
+                        sentiment_score=r["sentiment_score"],
+                    )
+                )
+        except Exception as e:
+            import logging
+            logging.getLogger(__name__).warning(f"news_list_new_fail: {e}")
+
+    # Legacy fallback (crypto_news)
+    if not items and catalog._health.postgres:
+        try:
+            query = "SELECT id, title, content, url, source, sentiment, tickers, published, ingested_at FROM crypto_news WHERE 1=1"
+            params = []
+            if category:
+                query += f" AND category = ${len(params)+1}"
+                params.append(category)
+            query += " ORDER BY ingested_at DESC LIMIT $%d OFFSET $%d" % (len(params)+1, len(params)+2)
+            params.extend([limit, offset])
+            async with catalog._pg_pool.acquire() as conn:
+                rows = await conn.fetch(query, *params)
+            for r in rows:
+                items.append(_adapt_legacy_row(dict(r)))
+        except Exception as e:
+            import logging
+            logging.getLogger(__name__).warning(f"news_list_legacy_fail: {e}")
+
+    return NewsListResponse(items=items, total=len(items), offset=offset)
+
+
+# ── GET /api/v1/news/trending ─────────────────────────────────────
+@router.get("/trending", response_model=NewsListResponse)
+async def trending_news(
+    window_hours: int = Query(168, ge=1, le=720),
+    limit: int = Query(20, ge=1, le=100),
+) -> NewsListResponse:
+    """Time-decay trending. Reads from crypto_news, scores, sorts by score."""
+    catalog = get_catalog()
+    await catalog._init_stores()
+    if not catalog._health.postgres:
+        return NewsListResponse(items=[], total=0, offset=0)
+    try:
+        cutoff_epoch = (utcnow() - timedelta(hours=window_hours)).timestamp()
+        async with catalog._pg_pool.acquire() as conn:
+            rows = await conn.fetch(
+                "SELECT id, title, content, url, source, sentiment, tickers, "
+                "published, ingested_at, category "
+                "FROM crypto_news "
+                "WHERE ingested_at > $1 "
+                "ORDER BY ingested_at DESC LIMIT 500",
+                cutoff_epoch,
+            )
+        now = utcnow()
+        items: list[NewsItemOut] = []
+        for r in rows:
+            d = dict(r)
+            # Estimate hours_old
+            hours_old = 0.0
+            try:
+                if d.get("ingested_at"):
+                    hours_old = max(0, (now.timestamp() - float(d["ingested_at"])) / 3600)
+            except Exception:
+                pass
+            item = _adapt_legacy_row(d)
+            item.score = round(trend_score(hours_old, item.source, item.sentiment_score), 4)
+            items.append(item)
+        items.sort(key=lambda x: x.score or 0, reverse=True)
+        return NewsListResponse(items=items[:limit], total=len(items), offset=0)
+    except Exception as e:
+        import logging
+        logging.getLogger(__name__).warning(f"trending_fail: {e}")
+        return NewsListResponse(items=[], total=0, offset=0)
+
+
+# ── GET /api/v1/news/{news_id} ────────────────────────────────────
+@router.get("/{news_id}", response_model=NewsItemOut)
+async def get_news(news_id: str) -> NewsItemOut:
+    """Single news item. Searches both news_items and crypto_news."""
+    catalog = get_catalog()
+    await catalog._init_stores()
+    if not catalog._health.postgres:
+        raise HTTPException(503, "postgres unavailable")
+    try:
+        async with catalog._pg_pool.acquire() as conn:
+            r = await conn.fetchrow(
+                "SELECT news_id, url, title, summary, source, published_at, "
+                "sentiment_score, chains_mentioned, tokens_mentioned "
+                "FROM news_items WHERE news_id=$1",
+                news_id,
+            )
+            if r:
+                return NewsItemOut(
+                    news_id=r["news_id"],
+                    url=r["url"],
+                    title=r["title"],
+                    summary=r["summary"] or "",
+                    source=r["source"],
+                    published_at=r["published_at"],
+                    chains_mentioned=list(r["chains_mentioned"] or []),
+                    tokens_mentioned=list(r["tokens_mentioned"] or []),
+                    sentiment_score=r["sentiment_score"],
+                )
+            r2 = await conn.fetchrow(
+                "SELECT id, title, content, url, source, sentiment, tickers, "
+                "published, ingested_at, category "
+                "FROM crypto_news WHERE id=$1",
+                news_id,
+            )
+        if r2:
+            return _adapt_legacy_row(dict(r2))
+        raise HTTPException(404, "news item not found")
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(500, f"news_get_fail: {e}")
+
+
+# ── POST /api/v1/news/{news_id}/analyze ───────────────────────────
+@router.post("/{news_id}/analyze", response_model=NewsAnalysisResponse)
+async def analyze_news(news_id: str) -> NewsAnalysisResponse:
+    """Generate LLM analysis via LiteLLM. Falls back to None if LLM unreachable."""
+    catalog = get_catalog()
+    await catalog._init_stores()
+    if not catalog._health.postgres:
+        raise HTTPException(503, "postgres unavailable")
+    try:
+        # Fetch the news item
+        item: NewsItemOut | None = None
+        async with catalog._pg_pool.acquire() as conn:
+            r = await conn.fetchrow(
+                "SELECT news_id, url, title, summary, source, published_at, "
+                "sentiment_score, chains_mentioned, tokens_mentioned "
+                "FROM news_items WHERE news_id=$1",
+                news_id,
+            )
+            if r:
+                item = NewsItemOut(
+                    news_id=r["news_id"],
+                    url=r["url"],
+                    title=r["title"],
+                    summary=r["summary"] or "",
+                    source=r["source"],
+                    published_at=r["published_at"],
+                    chains_mentioned=list(r["chains_mentioned"] or []),
+                    tokens_mentioned=list(r["tokens_mentioned"] or []),
+                    sentiment_score=r["sentiment_score"],
+                )
+            if not item:
+                r2 = await conn.fetchrow(
+                    "SELECT id, title, content, url, source, sentiment, tickers, "
+                    "published, ingested_at FROM crypto_news WHERE id=$1",
+                    news_id,
+                )
+                if r2:
+                    item = _adapt_legacy_row(dict(r2))
+        if not item:
+            raise HTTPException(404, "news item not found")
+        # Build a NewsItem for the LLM router
+        from app.catalog.models import Chain, NewsItem
+
+        ni = NewsItem(
+            news_id=item.news_id,
+            url=item.url or "https://unknown.local",  # HttpUrl requires non-empty
+            title=item.title or "",
+            summary=item.summary or "",
+            body_markdown=item.summary or "",
+            source=item.source or "unknown",
+            published_at=item.published_at or utcnow(),
+            ingested_at=utcnow(),
+        )
+        llm = LLMRouter()
+        analysis = await llm.analyze_news(ni)
+        if analysis is None:
+            return NewsAnalysisResponse(
+                news_id=news_id, analysis=None, error="LLM router unavailable"
+            )
+        return NewsAnalysisResponse(
+            news_id=news_id, analysis=analysis, model="deepseek-v3"
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        return NewsAnalysisResponse(news_id=news_id, analysis=None, error=str(e))

backend/app/domain/token/repository.py

@@ -92,8 +92,3 @@ class TokenRepository:
             pools=raw.get("pools", []) or [],
             locked_percentage=float(raw.get("locked_percentage", 0) or 0),
         )
-
-
-    def count(self) -> int:
-        from app.core.redis import get_redis
-        return get_redis().scard(TOKEN_INDEX)

backend/app/domain/token/service.py

@@ -104,6 +104,3 @@ class TokenService:
             risk_level=risk.risk_level.value,
         )
         return result
-
-    def count_tokens(self) -> int:
-        return self.repo.count()

backend/app/domain/wallet/repository.py

@@ -110,10 +110,3 @@ class WalletRepository:
             timestamp=int(block_time or 0),
             direction=tx.get("direction"),
         )
-
-
-    def count(self) -> int:
-        from app.core.redis import get_redis
-        r = get_redis()
-        keys = r.keys("rmi:wallet:*")
-        return len([k for k in keys if not k.endswith(":labels") and not k.endswith(":index")])

backend/app/domain/wallet/service.py

@@ -103,6 +103,3 @@ class WalletService:
             flag_count=len(result.flags),
         )
         return result
-
-    def count_wallets(self) -> int:
-        return self._repo.count()

backend/main.py

@@ -216,6 +216,7 @@ def _try_mount_v1_routers() -> int:
         "app.api.v1.x402.payments",
         "app.api.v1.admin.alerts_webhook",
         "app.api.v1.catalog",
+        "app.domain.news",
     ]
 
     for module_path in v1_modules: