refactor(databus): move app.databus engine files to app.domains.databus (P4.5 cont)

app/core/lifespan.py

@@ -59,7 +59,7 @@ async def _start_background_tasks(app: FastAPI) -> None:
 
     # Cache warmer (passes app instance)
     try:
-        from app.databus.core import cache_warm_loop
+        from app.domains.databus.core import cache_warm_loop
 
         asyncio.create_task(cache_warm_loop(app))
         logger.info("background_task_started", task="databus_cache_warmer", interval="")

app/databus/__init__.py

@@ -1,69 +1,23 @@
+"""databus - DEPRECATED shim. Use app.domains.databus.
+
+Phase 4.5 of AUDIT-2026-Q3.md moved app/databus/ to app.domains.databus/.
+This shim re-exports the canonical public surface so legacy imports like
+`from app.databus import databus` keep working.
+
+MIGRATION: replace `from app.databus import ...` with
+`from app.domains.databus import ...`.
 """
-RMI DataBus - The Single Source of Truth for ALL Data
-=====================================================
-
-Every API call, MCP tool, x402 tool, scanner, and frontend hook routes through here.
-No raw HTTP calls to external APIs anywhere else. Period.
-
-Architecture (request flow):
-  Request → SecurityGate → CreditGate → CacheLayer → ProviderChain → Result
-        ↓                      ↓             ↓              ↓            ↓
-   admin check         free-first logic   L1→L2→L3      local-first!   RAG index
-   vault keys          auto-rotate      Redis+R2      OUR data 1st    WS broadcast
-
-What it REPLACES (do NOT use these anymore):
-  - app/cache_manager.py (RMICache)          → databus.cache
-  - app/caching_shield/unified_layer.py       → databus.fetch()
-  - app/caching_shield/data_fallback.py       → databus.fetch()
-  - app/caching_shield/api_registry.py        → databus.key_pool
-  - app/caching_shield/rate_limiter.py         → databus.rate_limiter
-  - app/arkham_connector.py                    → databus.providers.arkham
-  - app/coingecko_connector.py                → databus.providers.coingecko
-  - Direct .env reads for API keys            → databus.vault (encrypted in-memory)
-
-OWN DATA FIRST:
-  Our crown jewels - Wallet Memory Bank, RAG (17K docs), SENTINEL scanner,
-  Consensus RPC, Funding Tracer, ClickHouse, Price Consensus, News Network,
-  Bundle Detection, Label Import (169K+) - these are FAST, FREE, and OURS.
-  They go FIRST in every fallback chain. External APIs only augment or fill gaps.
-
-Usage:
-    from app.databus import databus
-
-    # Simple fetch - auto-selects best provider chain
-    result = await databus.fetch("token_price", mint="So11111111111111111111111111111111111111111")
-
-    # Explicit chain override
-    result = await databus.fetch("wallet_labels", address="7EcD...",
-                                  chain="local_first")
-
-    # Admin-only data (requires admin key in request)
-    result = await databus.fetch("arkham_entity", address="0x...",
-                                  admin_key="...")
-
-    # Check system health
-    health = await databus.health()
-
-    # Get capacity report (credits, rate limits, recommendations)
-    report = databus.capacity_report()
-"""
-
-from app.databus.access_control import AccessController, ConsumerType, access_controller
-from app.databus.core import DataBus, databus
-from app.databus.key_affinity import KeyAffinitySelector, key_affinity
-from app.databus.response_schema import SchemaValidator, schema_validator
-from app.databus.social import SocialDataAggregator, XTwitterProvider
-
-__all__ = [
-    "AccessController",
-    "ConsumerType",
-    "DataBus",
-    "KeyAffinitySelector",
-    "SchemaValidator",
-    "SocialDataAggregator",
-    "XTwitterProvider",
-    "access_controller",
-    "databus",
-    "key_affinity",
-    "schema_validator",
-]
+from app.domains.databus import *  # noqa: F403
+from app.domains.databus import (  # noqa: F401
+    AccessController,
+    ConsumerType,
+    DataBus,
+    KeyAffinitySelector,
+    SchemaValidator,
+    SocialDataAggregator,
+    XTwitterProvider,
+    access_controller,
+    databus,
+    key_affinity,
+    schema_validator,
+)

app/domains/databus/__init__.py

@@ -0,0 +1,69 @@
+"""
+RMI DataBus - The Single Source of Truth for ALL Data
+=====================================================
+
+Every API call, MCP tool, x402 tool, scanner, and frontend hook routes through here.
+No raw HTTP calls to external APIs anywhere else. Period.
+
+Architecture (request flow):
+  Request → SecurityGate → CreditGate → CacheLayer → ProviderChain → Result
+        ↓                      ↓             ↓              ↓            ↓
+   admin check         free-first logic   L1→L2→L3      local-first!   RAG index
+   vault keys          auto-rotate      Redis+R2      OUR data 1st    WS broadcast
+
+What it REPLACES (do NOT use these anymore):
+  - app/cache_manager.py (RMICache)          → databus.cache
+  - app/caching_shield/unified_layer.py       → databus.fetch()
+  - app/caching_shield/data_fallback.py       → databus.fetch()
+  - app/caching_shield/api_registry.py        → databus.key_pool
+  - app/caching_shield/rate_limiter.py         → databus.rate_limiter
+  - app/arkham_connector.py                    → databus.providers.arkham
+  - app/coingecko_connector.py                → databus.providers.coingecko
+  - Direct .env reads for API keys            → databus.vault (encrypted in-memory)
+
+OWN DATA FIRST:
+  Our crown jewels - Wallet Memory Bank, RAG (17K docs), SENTINEL scanner,
+  Consensus RPC, Funding Tracer, ClickHouse, Price Consensus, News Network,
+  Bundle Detection, Label Import (169K+) - these are FAST, FREE, and OURS.
+  They go FIRST in every fallback chain. External APIs only augment or fill gaps.
+
+Usage:
+    from app.databus import databus
+
+    # Simple fetch - auto-selects best provider chain
+    result = await databus.fetch("token_price", mint="So11111111111111111111111111111111111111111")
+
+    # Explicit chain override
+    result = await databus.fetch("wallet_labels", address="7EcD...",
+                                  chain="local_first")
+
+    # Admin-only data (requires admin key in request)
+    result = await databus.fetch("arkham_entity", address="0x...",
+                                  admin_key="...")
+
+    # Check system health
+    health = await databus.health()
+
+    # Get capacity report (credits, rate limits, recommendations)
+    report = databus.capacity_report()
+"""
+
+from app.domains.databus.access_control import AccessController, ConsumerType, access_controller
+from app.domains.databus.core import DataBus, databus
+from app.domains.databus.key_affinity import KeyAffinitySelector, key_affinity
+from app.domains.databus.response_schema import SchemaValidator, schema_validator
+from app.domains.databus.social import SocialDataAggregator, XTwitterProvider
+
+__all__ = [
+    "AccessController",
+    "ConsumerType",
+    "DataBus",
+    "KeyAffinitySelector",
+    "SchemaValidator",
+    "SocialDataAggregator",
+    "XTwitterProvider",
+    "access_controller",
+    "databus",
+    "key_affinity",
+    "schema_validator",
+]

app/domains/databus/_generated/provider_chains.py

@@ -8,7 +8,7 @@ Source: app/databus/provider_chains.py (legacy shim).
 
 import logging
 
-from app.databus.provider_core import (
+from app.domains.databus.provider_core import (
     Provider,
     ProviderChain,
     ProviderTier,
@@ -17,7 +17,7 @@ from app.databus.provider_core import (
     _rate_limiters,
     _RateLimiter,
 )
-from app.databus.provider_implementations import (
+from app.domains.databus.provider_implementations import (
     _alchemy_token_balances,
     _alchemy_token_metadata,
     _arkham_counterparties,
@@ -63,7 +63,7 @@ from app.databus.provider_implementations import (
     _solana_tracker_trending,
     _virustotal_url_scan,
 )
-from app.databus.spl_metadata_decoder import _spl_metadata_decoder_provider
+from app.domains.databus.spl_metadata_decoder import _spl_metadata_decoder_provider
 
 logger = logging.getLogger("databus.providers.chains")
 def build_provider_chains() -> dict[str, ProviderChain]:
@@ -271,7 +271,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── Bitquery chains (activate free plan at bitquery.io/pricing) ──
     try:
-        from app.databus.bitquery_provider import BitqueryProvider
+        from app.domains.databus.bitquery_provider import BitqueryProvider
 
         _bq = BitqueryProvider()
 
@@ -758,7 +758,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── PREMIUM SCANNER - 10 high-value detection chains ──
     try:
-        from app.databus.premium_scanner import (
+        from app.domains.databus.premium_scanner import (
             detect_bot_farms,
             detect_bundles,
             detect_copy_trading,
@@ -842,7 +842,11 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── WEBHOOK SYSTEM ──
     try:
-        from app.databus.webhooks import handle_webhook, list_webhooks, setup_webhook  # noqa: F401
+        from app.domains.databus.webhooks import (  # noqa: F401
+            handle_webhook,
+            list_webhooks,
+            setup_webhook,
+        )
 
         chains["webhook_handler"] = ProviderChain(
             "webhook_handler",
@@ -866,7 +870,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── ARKHAM WEBSOCKET - real-time intelligence streaming ──
     try:
-        from app.databus.arkham_ws import arkham_ws_subscribe
+        from app.domains.databus.arkham_ws import arkham_ws_subscribe
 
         chains["arkham_ws"] = ProviderChain(
             "arkham_ws",
@@ -890,7 +894,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Volume Authenticity (Fake Volume %) ──
     try:
-        from app.databus.volume_authenticity import analyze_volume_authenticity
+        from app.domains.databus.volume_authenticity import analyze_volume_authenticity
 
         chains["volume_authenticity"] = ProviderChain(
             "volume_authenticity",
@@ -911,7 +915,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: OHLCV Engine ──
     try:
-        from app.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
+        from app.domains.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
 
         chains["ohlcv"] = ProviderChain(
             "ohlcv",
@@ -939,7 +943,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Token Security Matrix (37+ checks) ──
     try:
-        from app.databus.token_security import get_check_matrix_endpoint, run_full_scan
+        from app.domains.databus.token_security import get_check_matrix_endpoint, run_full_scan
 
         chains["token_security"] = ProviderChain(
             "token_security",
@@ -967,8 +971,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS INTELLIGENCE - 10 Premium Features ──
     try:
-        from app.databus.data_quality import enhanced_token_report, get_tier_comparison
-        from app.databus.rugcharts_intel import (
+        from app.domains.databus.data_quality import enhanced_token_report, get_tier_comparison
+        from app.domains.databus.rugcharts_intel import (
             cross_chain_entity,
             developer_reputation,
             holder_health_score,
@@ -1140,7 +1144,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS & MARKET DATA - Free APIs ──
     try:
-        from app.databus.news_provider import (
+        from app.domains.databus.news_provider import (
             get_fear_greed,
             get_full_news_feed,
             get_market_brief,
@@ -1241,7 +1245,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS INTELLIGENCE ENGINE - multi-source, quality-scored, sentiment-tagged ──
     try:
-        from app.databus.news_intel import (
+        from app.domains.databus.news_intel import (
             add_comment,
             add_reaction,
             aggregate_all_news,
@@ -1365,7 +1369,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── X/CT INTELLIGENCE - Crypto Twitter Rundown ──
     try:
-        from app.databus.x_intel import fetch_ct_rundown, track_ct_accounts
+        from app.domains.databus.x_intel import fetch_ct_rundown, track_ct_accounts
 
         chains["ct_rundown"] = ProviderChain(
             "ct_rundown",
@@ -1401,8 +1405,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── SOCIAL INTELLIGENCE - KOL tracking, shill detection, Daily Intel ──
     try:
-        from app.databus.daily_intel import generate_daily_intel
-        from app.databus.social_intel import (
+        from app.domains.databus.daily_intel import generate_daily_intel
+        from app.domains.databus.social_intel import (
             detect_shill_campaigns,
             get_kol_leaderboard,
             get_kol_profile,
@@ -1525,7 +1529,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── MODEL REGISTRY - Smart free model routing, quality review ──
     try:
-        from app.databus.model_registry import ai_call, get_usage_stats, review_content
+        from app.domains.databus.model_registry import ai_call, get_usage_stats, review_content
 
         chains["ai_task"] = ProviderChain(
             "ai_task",
@@ -1581,7 +1585,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RAG INGESTION - nightly indexing, health checks ──
     try:
-        from app.databus.rag_ingestion import nightly_rag_index, rag_health_check
+        from app.domains.databus.rag_ingestion import nightly_rag_index, rag_health_check
 
         chains["rag_nightly"] = ProviderChain(
             "rag_nightly",

app/domains/databus/arkham_ws.py

@@ -97,7 +97,7 @@ async def broadcast_ws_update(address: str, update: dict):
 
     # Push to DataBus WebSocket for frontend subscribers
     try:
-        from app.databus.ws_stream import ws_manager
+        from app.domains.databus.ws_stream import ws_manager
 
         await ws_manager.broadcast(
             "arkham_realtime",

app/domains/databus/bitquery_provider.py

@@ -36,7 +36,7 @@ import time
 
 import httpx
 
-from app.databus.cache import CacheLayer, get_cache
+from app.domains.databus.cache import CacheLayer, get_cache
 
 logger = logging.getLogger("databus.bitquery")
 

app/domains/databus/core.py

@@ -32,11 +32,11 @@ import time
 from collections import defaultdict
 from typing import Any
 
-from app.databus.access_control import access_controller
-from app.databus.cache import CacheLayer, get_cache
-from app.databus.providers import ProviderChain, ProviderTier, build_provider_chains
-from app.databus.security import security
-from app.databus.vault import DataBusVault, get_vault
+from app.domains.databus.access_control import access_controller
+from app.domains.databus.cache import CacheLayer, get_cache
+from app.domains.databus.providers import ProviderChain, ProviderTier, build_provider_chains
+from app.domains.databus.security import security
+from app.domains.databus.vault import DataBusVault, get_vault
 
 logger = logging.getLogger("databus.core")
 
@@ -223,7 +223,7 @@ class DataBus:
             self.chains = build_provider_chains()
             # Register Bitquery provider for blockchain data
             try:
-                from app.databus.bitquery_provider import BitqueryProvider
+                from app.domains.databus.bitquery_provider import BitqueryProvider
 
                 self.bitquery = BitqueryProvider(self.cache)
                 logger.info("Bitquery provider registered")
@@ -485,7 +485,7 @@ class DataBus:
             "market_movers",
         ):
             try:
-                from app.databus.ws_stream import databus_ws_broadcast
+                from app.domains.databus.ws_stream import databus_ws_broadcast
 
                 await databus_ws_broadcast(data_type, result)
             except Exception:

app/domains/databus/daily_intel.py

@@ -89,7 +89,7 @@ async def _gather_all_data() -> dict:
 
     # Market data
     try:
-        from app.databus.news_provider import get_market_brief
+        from app.domains.databus.news_provider import get_market_brief
 
         data["market"] = await get_market_brief()
     except Exception:
@@ -97,7 +97,7 @@ async def _gather_all_data() -> dict:
 
     # News intel
     try:
-        from app.databus.news_intel import aggregate_all_news
+        from app.domains.databus.news_intel import aggregate_all_news
 
         data["news"] = await aggregate_all_news(limit=20)
     except Exception:
@@ -105,7 +105,7 @@ async def _gather_all_data() -> dict:
 
     # CT Rundown
     try:
-        from app.databus.x_intel import fetch_ct_rundown
+        from app.domains.databus.x_intel import fetch_ct_rundown
 
         data["ct"] = await fetch_ct_rundown(limit=15)
     except Exception:
@@ -113,7 +113,7 @@ async def _gather_all_data() -> dict:
 
     # Social metrics
     try:
-        from app.databus.social_intel import get_social_metrics
+        from app.domains.databus.social_intel import get_social_metrics
 
         data["social"] = await get_social_metrics()
     except Exception:
@@ -121,7 +121,7 @@ async def _gather_all_data() -> dict:
 
     # Fear & Greed
     try:
-        from app.databus.news_provider import get_fear_greed
+        from app.domains.databus.news_provider import get_fear_greed
 
         data["fear_greed"] = await get_fear_greed()
     except Exception:
@@ -129,7 +129,7 @@ async def _gather_all_data() -> dict:
 
     # Prediction markets
     try:
-        from app.databus.news_provider import get_prediction_markets
+        from app.domains.databus.news_provider import get_prediction_markets
 
         data["prediction_markets"] = await get_prediction_markets(limit=5)
     except Exception:
@@ -234,7 +234,7 @@ async def generate_daily_intel(publish: bool = False, **kw) -> dict | None:
     Args:
         publish: If True, publish to Ghost (primary) + X/Telegram (syndication)
     """
-    from app.databus.model_registry import ai_call, review_content
+    from app.domains.databus.model_registry import ai_call, review_content
 
     # ── PHASE 0: Gather all data ──
     logger.info("Daily Intel: gathering data...")

app/domains/databus/data_quality.py

@@ -570,7 +570,7 @@ async def enhanced_token_report(address: str, chain: str = "solana", **kw) -> di
 
     # Get the base token report
     try:
-        from app.databus.rugcharts_intel import instant_token_report
+        from app.domains.databus.rugcharts_intel import instant_token_report
 
         base_report = await instant_token_report(address, chain, **kw)
     except Exception as e:

app/domains/databus/ohlcv_engine.py

@@ -340,7 +340,7 @@ async def fetch_ohlcv(
     # Also compute authenticity if we have volume data
     authenticity = None
     try:
-        from app.databus.volume_authenticity import quick_authenticity_score
+        from app.domains.databus.volume_authenticity import quick_authenticity_score
 
         auth = quick_authenticity_score(
             volume_24h=summary.get("volume_24h", 0),

app/domains/databus/providers/__init__.py

@@ -38,7 +38,7 @@ import httpx
 logger = logging.getLogger("databus.providers")
 
 # Import SPL metadata decoder
-from app.databus.spl_metadata_decoder import _spl_metadata_decoder_provider  # noqa: E402
+from app.domains.databus.spl_metadata_decoder import _spl_metadata_decoder_provider  # noqa: E402
 
 
 class ProviderTier(Enum):
@@ -1539,7 +1539,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── Bitquery chains (activate free plan at bitquery.io/pricing) ──
     try:
-        from app.databus.bitquery_provider import BitqueryProvider
+        from app.domains.databus.bitquery_provider import BitqueryProvider
 
         _bq = BitqueryProvider()
 
@@ -2026,7 +2026,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── PREMIUM SCANNER - 10 high-value detection chains ──
     try:
-        from app.databus.premium_scanner import (
+        from app.domains.databus.premium_scanner import (
             detect_bot_farms,
             detect_bundles,
             detect_copy_trading,
@@ -2110,7 +2110,11 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── WEBHOOK SYSTEM ──
     try:
-        from app.databus.webhooks import handle_webhook, list_webhooks, setup_webhook  # noqa: F401
+        from app.domains.databus.webhooks import (  # noqa: F401
+            handle_webhook,
+            list_webhooks,
+            setup_webhook,
+        )
 
         chains["webhook_handler"] = ProviderChain(
             "webhook_handler",
@@ -2134,7 +2138,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── ARKHAM WEBSOCKET - real-time intelligence streaming ──
     try:
-        from app.databus.arkham_ws import arkham_ws_subscribe
+        from app.domains.databus.arkham_ws import arkham_ws_subscribe
 
         chains["arkham_ws"] = ProviderChain(
             "arkham_ws",
@@ -2158,7 +2162,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Volume Authenticity (Fake Volume %) ──
     try:
-        from app.databus.volume_authenticity import analyze_volume_authenticity
+        from app.domains.databus.volume_authenticity import analyze_volume_authenticity
 
         chains["volume_authenticity"] = ProviderChain(
             "volume_authenticity",
@@ -2179,7 +2183,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: OHLCV Engine ──
     try:
-        from app.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
+        from app.domains.databus.ohlcv_engine import fetch_ohlcv, ingest_trade_data
 
         chains["ohlcv"] = ProviderChain(
             "ohlcv",
@@ -2207,7 +2211,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS: Token Security Matrix (37+ checks) ──
     try:
-        from app.databus.token_security import get_check_matrix_endpoint, run_full_scan
+        from app.domains.databus.token_security import get_check_matrix_endpoint, run_full_scan
 
         chains["token_security"] = ProviderChain(
             "token_security",
@@ -2235,8 +2239,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RUGCHARTS INTELLIGENCE - 10 Premium Features ──
     try:
-        from app.databus.data_quality import enhanced_token_report, get_tier_comparison
-        from app.databus.rugcharts_intel import (
+        from app.domains.databus.data_quality import enhanced_token_report, get_tier_comparison
+        from app.domains.databus.rugcharts_intel import (
             cross_chain_entity,
             developer_reputation,
             holder_health_score,
@@ -2408,7 +2412,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS & MARKET DATA - Free APIs ──
     try:
-        from app.databus.news_provider import (
+        from app.domains.databus.news_provider import (
             get_fear_greed,
             get_full_news_feed,
             get_market_brief,
@@ -2509,7 +2513,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── NEWS INTELLIGENCE ENGINE - multi-source, quality-scored, sentiment-tagged ──
     try:
-        from app.databus.news_intel import (
+        from app.domains.databus.news_intel import (
             add_comment,
             add_reaction,
             aggregate_all_news,
@@ -2633,7 +2637,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── X/CT INTELLIGENCE - Crypto Twitter Rundown ──
     try:
-        from app.databus.x_intel import fetch_ct_rundown, track_ct_accounts
+        from app.domains.databus.x_intel import fetch_ct_rundown, track_ct_accounts
 
         chains["ct_rundown"] = ProviderChain(
             "ct_rundown",
@@ -2669,8 +2673,8 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── SOCIAL INTELLIGENCE - KOL tracking, shill detection, Daily Intel ──
     try:
-        from app.databus.daily_intel import generate_daily_intel
-        from app.databus.social_intel import (
+        from app.domains.databus.daily_intel import generate_daily_intel
+        from app.domains.databus.social_intel import (
             detect_shill_campaigns,
             get_kol_leaderboard,
             get_kol_profile,
@@ -2793,7 +2797,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── MODEL REGISTRY - Smart free model routing, quality review ──
     try:
-        from app.databus.model_registry import ai_call, get_usage_stats, review_content
+        from app.domains.databus.model_registry import ai_call, get_usage_stats, review_content
 
         chains["ai_task"] = ProviderChain(
             "ai_task",
@@ -2849,7 +2853,7 @@ def build_provider_chains() -> dict[str, ProviderChain]:
 
     # ── RAG INGESTION - nightly indexing, health checks ──
     try:
-        from app.databus.rag_ingestion import nightly_rag_index, rag_health_check
+        from app.domains.databus.rag_ingestion import nightly_rag_index, rag_health_check
 
         chains["rag_nightly"] = ProviderChain(
             "rag_nightly",

app/domains/databus/providers/btc.py

@@ -8,7 +8,7 @@ Re-exports the Bitcoin-scoped provider functions from
 Providers:
 - _blockchair_address, _blockchair_stats
 """
-from app.databus.providers import (  # noqa: F401
+from app.domains.databus.providers import (  # noqa: F401
     _blockchair_address,
     _blockchair_stats,
 )

app/domains/databus/providers/evm.py

@@ -12,7 +12,7 @@ Providers:
 - _moralis_wallet_net_worth
 - _etherscan_tx_trace
 """
-from app.databus.providers import (  # noqa: F401
+from app.domains.databus.providers import (  # noqa: F401
     _alchemy_token_balances,
     _alchemy_token_metadata,
     _etherscan_tx_trace,

app/domains/databus/providers/free_tier.py

@@ -15,7 +15,8 @@ Providers:
 - _passthrough_market_overview, _passthrough_trending, _passthrough_news,
   _passthrough_alerts, _passthrough_scanner, _passthrough_rag
 """
-from app.databus.providers import (  # noqa: F401
+from app.domains.databus.providers import (  # noqa: F401
+    _coindesk_news,
     _coingecko_price,
     _defillama_chains,
     _defillama_tvl,
@@ -26,12 +27,11 @@ from app.databus.providers import (  # noqa: F401
     _dexscreener_trades,
     _local_token_price,
     _local_wallet_labels,
+    _messari_news,
     _passthrough_alerts,
     _passthrough_market_overview,
     _passthrough_news,
     _passthrough_rag,
     _passthrough_scanner,
     _passthrough_trending,
-    _coindesk_news,
-    _messari_news,
 )

app/domains/databus/providers/mcp_servers.py

@@ -5,6 +5,6 @@ Phase 3B of AUDIT-2026-Q3.md.
 Re-exports the MCP bridge provider function from
 ``app.databus.providers`` for cleaner import paths.
 """
-from app.databus.providers import (  # noqa: F401
+from app.domains.databus.providers import (  # noqa: F401
     _mcp_bridge,
 )

app/domains/databus/providers/paid_tier.py

@@ -12,7 +12,7 @@ Providers:
 - _dune_early_buyers
 - _virustotal_url_scan
 """
-from app.databus.providers import (  # noqa: F401
+from app.domains.databus.providers import (  # noqa: F401
     _arkham_counterparties,
     _arkham_entity,
     _arkham_intel_search,

app/domains/databus/providers/solana.py

@@ -9,7 +9,7 @@ Providers:
 - _birdeye_overview, _birdeye_price (premium Solana)
 - _solana_tracker_price, _solana_tracker_token, _solana_tracker_trending
 """
-from app.databus.providers import (  # noqa: F401
+from app.domains.databus.providers import (  # noqa: F401
     _birdeye_overview,
     _birdeye_price,
     _solana_tracker_price,

app/domains/databus/rag_ingestion.py

@@ -26,7 +26,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 1. Index recent news articles ──
-        from app.databus.news_intel import aggregate_all_news
+        from app.domains.databus.news_intel import aggregate_all_news
 
         news = await aggregate_all_news(limit=100)
         news_docs = []
@@ -55,7 +55,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 2. Index CT Rundown ──
-        from app.databus.x_intel import fetch_ct_rundown
+        from app.domains.databus.x_intel import fetch_ct_rundown
 
         ct = await fetch_ct_rundown(limit=30)
         ct_docs = []
@@ -84,7 +84,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 3. Index market data snapshot ──
-        from app.databus.news_provider import get_fear_greed, get_market_brief
+        from app.domains.databus.news_provider import get_fear_greed, get_market_brief
 
         market = await get_market_brief()
         fear = await get_fear_greed()
@@ -107,7 +107,7 @@ async def nightly_rag_index(**kw) -> dict:
 
     try:
         # ── 4. Index social metrics ──
-        from app.databus.social_intel import get_social_metrics
+        from app.domains.databus.social_intel import get_social_metrics
 
         social = await get_social_metrics()
         social_doc = {

app/domains/databus/rag_provider.py

@@ -30,7 +30,7 @@ async def _rag_search_provider(**kwargs) -> dict | None:
         }
 
     try:
-        from app.databus.rag_engine import ai_enrich, hybrid_search
+        from app.domains.databus.rag_engine import ai_enrich, hybrid_search
 
         if isinstance(collections, str):
             collections = [c.strip() for c in collections.split(",") if c.strip()]

app/domains/databus/router.py

@@ -20,8 +20,8 @@ import os
 
 from fastapi import APIRouter, HTTPException, Query, Request
 
-from app.databus.core import databus
-from app.databus.social import (
+from app.domains.databus.core import databus
+from app.domains.databus.social import (
     X_DAILY_READ_BUDGET,
     X_FREE_MONTHLY_READ_LIMIT,
 )
@@ -133,11 +133,11 @@ async def databus_chains():
 async def databus_access_matrix(request: Request):
     """View the full access control matrix. Admin key required."""
     if not _verify_admin(request):
-        from app.databus.access_control import access_controller
+        from app.domains.databus.access_control import access_controller
 
         # Non-admin sees a limited view - only their tier's access
         return {"note": "Full matrix requires admin key. Your access depends on your consumer type."}
-    from app.databus.access_control import access_controller
+    from app.domains.databus.access_control import access_controller
 
     return access_controller.list_access_matrix()
 
@@ -145,7 +145,7 @@ async def databus_access_matrix(request: Request):
 @router.get("/mcp-scope/{tool_id}")
 async def databus_mcp_scope(tool_id: str):
     """Get the data types an MCP tool is authorized to access."""
-    from app.databus.access_control import access_controller
+    from app.domains.databus.access_control import access_controller
 
     allowed = access_controller.get_mcp_allowed_types(tool_id)
     if not allowed:
@@ -156,7 +156,7 @@ async def databus_mcp_scope(tool_id: str):
 @router.get("/x402-scope/{tier}")
 async def databus_x402_scope(tier: str):
     """Get the data types an x402 pricing tier can access."""
-    from app.databus.access_control import access_controller
+    from app.domains.databus.access_control import access_controller
 
     allowed = access_controller.get_x402_allowed_types(tier)
     return {"tier": tier, "allowed_types": allowed}
@@ -465,7 +465,7 @@ async def databus_schema_validate(request: Request):
     body = await request.json()
     data_type = body.get("data_type", "")
     data = body.get("data", {})
-    from app.databus.response_schema import schema_validator
+    from app.domains.databus.response_schema import schema_validator
 
     is_valid, missing = schema_validator.validate(data_type, data)
     return {"data_type": data_type, "valid": is_valid, "missing_fields": missing}
@@ -477,7 +477,7 @@ async def databus_schema_validate(request: Request):
 @router.get("/social/x/profile/{username}")
 async def social_x_profile(username: str):
     """Get X/Twitter user profile - cached 24h."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.x.get_user(username)
@@ -489,7 +489,7 @@ async def social_x_profile(username: str):
 @router.get("/social/x/tweets/{username}")
 async def social_x_tweets(username: str, count: int = Query(20, ge=1, le=100)):
     """Get recent tweets from user - cached 15min."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.get_our_tweets(count=count) if username.lower() == "cryptorugmunch" else None
@@ -506,7 +506,7 @@ async def social_x_tweets(username: str, count: int = Query(20, ge=1, le=100)):
 @router.get("/social/x/mentions")
 async def social_x_mentions(count: int = Query(20, ge=1, le=100)):
     """Get mentions of @CryptoRugMunch - cached 15min."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.get_our_mentions(count=count)
@@ -518,7 +518,7 @@ async def social_x_mentions(count: int = Query(20, ge=1, le=100)):
 @router.get("/social/x/engagement")
 async def social_x_engagement(tweet_ids: str = Query(..., description="Comma-separated tweet IDs")):
     """Get engagement metrics for tweets - cached 1h."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     ids = [tid.strip() for tid in tweet_ids.split(",")[:100]]
@@ -529,7 +529,7 @@ async def social_x_engagement(tweet_ids: str = Query(..., description="Comma-sep
 @router.get("/social/x/search")
 async def social_x_search(q: str = Query(..., description="Search query"), count: int = Query(10, ge=1, le=100)):
     """Search X/Twitter - VERY expensive, cached 24h. Use sparingly."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     result = await agg.search_mentions(q, count=count)
@@ -541,7 +541,7 @@ async def social_x_search(q: str = Query(..., description="Search query"), count
 @router.get("/social/kol/{username}")
 async def social_kol_reputation(username: str):
     """KOL reputation score - cached 24h."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     return await agg.get_kol_reputation(username)
@@ -550,7 +550,7 @@ async def social_kol_reputation(username: str):
 @router.get("/social/sentiment")
 async def social_sentiment(username: str = Query("CryptoRugMunch")):
     """Brand sentiment analysis - cached 1h."""
-    from app.databus.social import SocialDataAggregator
+    from app.domains.databus.social import SocialDataAggregator
 
     agg = SocialDataAggregator(databus.cache)
     return await agg.get_sentiment(username)
@@ -559,7 +559,7 @@ async def social_sentiment(username: str = Query("CryptoRugMunch")):
 @router.get("/social/x/budget")
 async def social_x_budget():
     """Current X API read budget status."""
-    from app.databus.social import XTwitterProvider
+    from app.domains.databus.social import XTwitterProvider
 
     provider = XTwitterProvider(databus.cache)
     return {
@@ -577,7 +577,7 @@ X_HANDLE = "CryptoRugMunch"
 @router.get("/social/x/discover")
 async def social_x_discover(handle: str = Query(X_HANDLE), limit: int = Query(50, ge=1, le=100)):
     """Discover tweets via web search - no API needed, works for free."""
-    from app.databus.social_scraper import XWebScraper
+    from app.domains.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     tweets = await scraper.discover_tweets(handle=handle, limit=limit)
@@ -587,7 +587,7 @@ async def social_x_discover(handle: str = Query(X_HANDLE), limit: int = Query(50
 @router.get("/social/x/engagement-report")
 async def social_x_engagement_report(handle: str = Query(X_HANDLE)):
     """Get engagement report for a handle - avg likes, best tweets, posting frequency."""
-    from app.databus.social_scraper import XWebScraper
+    from app.domains.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     report = await scraper.get_engagement_report(handle=handle)
@@ -597,7 +597,7 @@ async def social_x_engagement_report(handle: str = Query(X_HANDLE)):
 @router.get("/social/x/mentions-discover")
 async def social_x_mentions_discover(handle: str = Query(X_HANDLE), limit: int = Query(20, ge=1, le=50)):
     """Find tweets mentioning a handle - via web search."""
-    from app.databus.social_scraper import XWebScraper
+    from app.domains.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     mentions = await scraper.get_mentions(handle=handle, limit=limit)
@@ -607,7 +607,7 @@ async def social_x_mentions_discover(handle: str = Query(X_HANDLE), limit: int =
 @router.get("/social/x/trending")
 async def social_x_trending():
     """Get current crypto trending topics from web search."""
-    from app.databus.social_scraper import XWebScraper
+    from app.domains.databus.social_scraper import XWebScraper
 
     scraper = XWebScraper(databus.cache)
     topics = await scraper.get_trending_topics()
@@ -620,7 +620,7 @@ async def social_x_trending():
 @router.get("/bitquery/health")
 async def bitquery_health():
     """Check Bitquery provider health and billing status."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     return await provider.health()
@@ -629,7 +629,7 @@ async def bitquery_health():
 @router.get("/bitquery/token-price/{network}/{token_address}")
 async def bitquery_token_price(network: str, token_address: str):
     """Get DEX token price from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_token_price(network, token_address)
@@ -641,7 +641,7 @@ async def bitquery_token_price(network: str, token_address: str):
 @router.get("/bitquery/holders/{network}/{token_address}")
 async def bitquery_holders(network: str, token_address: str, limit: int = Query(100, ge=1, le=500)):
     """Get token holder distribution from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_holder_distribution(network, token_address, limit)
@@ -653,7 +653,7 @@ async def bitquery_holders(network: str, token_address: str, limit: int = Query(
 @router.get("/bitquery/tx-trace/{network}/{tx_hash}")
 async def bitquery_tx_trace(network: str, tx_hash: str):
     """Get full transaction trace from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_transaction_trace(network, tx_hash)
@@ -665,7 +665,7 @@ async def bitquery_tx_trace(network: str, tx_hash: str):
 @router.get("/bitquery/dex-volume/{network}")
 async def bitquery_dex_volume(network: str, pool: str = Query(None), timeframe: str = Query("24h")):
     """Get DEX trading volume from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_dex_volume(network, pool, timeframe)
@@ -683,7 +683,7 @@ async def bitquery_dex_volume(network: str, pool: str = Query(None), timeframe:
 @router.get("/bitquery/balance/{network}/{address}")
 async def bitquery_balance(network: str, address: str):
     """Get address token balances from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_address_balance(network, address)
@@ -695,7 +695,7 @@ async def bitquery_balance(network: str, address: str):
 @router.get("/bitquery/cross-chain/{address}")
 async def bitquery_cross_chain(address: str, networks: str = Query("ethereum,bsc,solana")):
     """Track token transfers across chains from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     net_list = [n.strip() for n in networks.split(",")]
@@ -710,7 +710,7 @@ async def bitquery_contract_events(
     network: str, contract: str, event: str = Query(None), limit: int = Query(50, ge=1, le=200)
 ):
     """Get smart contract events from Bitquery."""
-    from app.databus.bitquery_provider import BitqueryProvider
+    from app.domains.databus.bitquery_provider import BitqueryProvider
 
     provider = BitqueryProvider(databus.cache)
     result = await provider.get_smart_contract_events(network, contract, event, limit)
@@ -734,7 +734,7 @@ async def receive_webhook(service: str, request: Request):
     indexes in RAG, triggers premium scanner, pushes alerts.
     """
     try:
-        from app.databus.webhooks import handle_webhook
+        from app.domains.databus.webhooks import handle_webhook
     except ImportError:
         raise HTTPException(501, "Webhook system not available") from None
 
@@ -758,7 +758,7 @@ async def receive_webhook(service: str, request: Request):
 async def list_webhooks_endpoint():
     """List all recent webhook events."""
     try:
-        from app.databus.webhooks import list_webhooks
+        from app.domains.databus.webhooks import list_webhooks
 
         return await list_webhooks()
     except ImportError:
@@ -777,7 +777,7 @@ async def setup_webhook_endpoint(service: str, request: Request):
     }
     """
     try:
-        from app.databus.webhooks import setup_webhook
+        from app.domains.databus.webhooks import setup_webhook
     except ImportError:
         raise HTTPException(501, "Webhook system not available") from None
 

app/domains/databus/rugcharts_intel.py

@@ -1198,7 +1198,7 @@ async def instant_token_report(address: str = "", chain: str = "solana", **kw) -
     try:
         # ── Section 1: Security Scan ──
         try:
-            from app.databus.token_security import run_full_scan
+            from app.domains.databus.token_security import run_full_scan
 
             security = await run_full_scan(address, chain, api_key=api_key)
             if security:
@@ -1288,7 +1288,7 @@ async def instant_token_report(address: str = "", chain: str = "solana", **kw) -
             liq = float(kw.get("liquidity_usd", 0))
             uw = int(kw.get("unique_wallets", 0))
             if vol_24h > 0 or liq > 0:
-                from app.databus.volume_authenticity import quick_authenticity_score
+                from app.domains.databus.volume_authenticity import quick_authenticity_score
 
                 auth = quick_authenticity_score(vol_24h, liq, uw, 0)
                 report["sections"]["volume"] = {

app/domains/databus/social.py

@@ -20,7 +20,7 @@ from datetime import UTC, datetime
 
 import httpx
 
-from app.databus.cache import CacheLayer
+from app.domains.databus.cache import CacheLayer
 
 logger = logging.getLogger("databus.social")
 

app/domains/databus/social_intel.py

@@ -295,21 +295,21 @@ async def generate_daily_intel(**kw) -> dict:
     """
     # Gather all data sources
     try:
-        from app.databus.news_provider import get_market_brief
+        from app.domains.databus.news_provider import get_market_brief
 
         brief = await get_market_brief()
     except Exception:
         brief = {}
 
     try:
-        from app.databus.news_intel import aggregate_all_news
+        from app.domains.databus.news_intel import aggregate_all_news
 
         news = await aggregate_all_news(limit=15)
     except Exception:
         news = {"articles": []}
 
     try:
-        from app.databus.x_intel import fetch_ct_rundown
+        from app.domains.databus.x_intel import fetch_ct_rundown
 
         ct = await fetch_ct_rundown(limit=10)
     except Exception:
@@ -411,7 +411,7 @@ async def get_social_metrics(**kw) -> dict:
     """
     # Gather data
     try:
-        from app.databus.news_intel import aggregate_all_news
+        from app.domains.databus.news_intel import aggregate_all_news
 
         news = await aggregate_all_news(limit=50)
     except Exception:

app/domains/databus/social_scraper.py

@@ -16,7 +16,7 @@ Cron: Every 6 hours, discover new tweets, extract content, update metrics.
 import logging
 from datetime import UTC, datetime
 
-from app.databus.cache import CacheLayer, get_cache
+from app.domains.databus.cache import CacheLayer, get_cache
 
 logger = logging.getLogger("databus.social_scraper")
 

app/domains/databus/spl_metadata_decoder.py

@@ -257,7 +257,6 @@ async def fetch_metaplex_metadata(mint: str, rpc_url: str) -> dict[str, Any] | N
         # Check if it's off the ed25519 curve (valid PDA)
         # For simplicity, we'll just use the first one that doesn't have the high bit set
         # Actually, proper PDA derivation is complex. Let's use a known endpoint or skip if too complex.
-        pass
 
     # Simpler approach: use getProgramAccounts with filters
     payload = {

app/domains/databus/token_security.py

@@ -613,7 +613,7 @@ async def run_quick_scan(address: str, chain: str = "ethereum", **kw) -> dict[st
         tx_count = int(kw.get("tx_count", 0))
 
         if volume_24h > 0 or liquidity > 0:
-            from app.databus.volume_authenticity import quick_authenticity_score
+            from app.domains.databus.volume_authenticity import quick_authenticity_score
 
             auth = quick_authenticity_score(volume_24h, liquidity, unique_wallets, tx_count)
             fake_pct = auth.get("fake_volume_pct", 0)

app/domains/token/__init__.py

@@ -8,7 +8,7 @@ from app.core.health import DomainHealth
 async def _health_check() -> DomainHealth:
     """Token health: DataBus is importable."""
     try:
-        import app.databus  # noqa: F401
+        import app.domains.databus  # noqa: F401
         return DomainHealth(
             name="token",
             healthy=True,

app/domains/token/repository.py

@@ -19,7 +19,7 @@ class TokenRepository:
     async def get_detail(self, address: str, chain: str) -> TokenDetail:
         """Fetch token metadata + supply."""
         try:
-            from app.databus.client import get_databus_client
+            from app.domains.databus.client import get_databus_client
 
             client = get_databus_client()
             raw = await client.get_token_info(address, chain=chain)
@@ -36,7 +36,7 @@ class TokenRepository:
     ) -> list[TokenHolder]:
         """Fetch top holders."""
         try:
-            from app.databus.client import get_databus_client
+            from app.domains.databus.client import get_databus_client
 
             client = get_databus_client()
             raw = await client.get_token_holders(address, chain=chain, limit=limit)
@@ -48,7 +48,7 @@ class TokenRepository:
     async def get_liquidity(self, address: str, chain: str) -> TokenLiquidity:
         """Fetch liquidity + pool info."""
         try:
-            from app.databus.client import get_databus_client
+            from app.domains.databus.client import get_databus_client
 
             client = get_databus_client()
             raw = await client.get_token_liquidity(address, chain=chain)

app/routers/x402_databus_tools.py

@@ -1123,7 +1123,7 @@ async def x402_catalog(request: Request):
     """Full catalog of available x402 DataBus tools with pricing."""
     catalog = []
     for tool_id, info in sorted(X402_TOOL_PRICING.items(), key=lambda x: (x[1]["category"], x[1]["price_usd"])):
-        from app.databus.access_control import access_controller
+        from app.domains.databus.access_control import access_controller
 
         allowed_free = access_controller.get_x402_allowed_types("free")
         allowed_basic = access_controller.get_x402_allowed_types("basic")
@@ -1195,7 +1195,7 @@ async def x402_catalog(request: Request):
 @router.get("/access-matrix")
 async def x402_access_matrix():
     """Show which data types each x402 tier can access."""
-    from app.databus.access_control import access_controller
+    from app.domains.databus.access_control import access_controller
 
     return {
         "free": access_controller.get_x402_allowed_types("free"),

databus_warm_cron.py

@@ -15,8 +15,8 @@ os.environ["REDIS_PORT"] = "6379"
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
-from app.databus.cache import get_cache
-from app.databus.core import databus
+from app.domains.databus.cache import get_cache
+from app.domains.databus.core import databus
 
 
 async def warm():

scripts/fix_x402_pipeline.py

@@ -11,7 +11,7 @@ import sys
 
 sys.path.insert(0, "/app")
 
-from app.databus.providers import build_provider_chains
+from app.domains.databus.providers import build_provider_chains
 from app.routers.x402_databus_tools import X402_TOOL_PRICING
 
 chains = build_provider_chains()

tests/integration/test_databus.py

@@ -80,7 +80,7 @@ class TestDataBusSmoke:
     @pytest.mark.parametrize("chain", CHAINS)
     async def test_chain_exists_in_providers(self, chain):
         """Every chain should exist in build_provider_chains()."""
-        from app.databus.providers import build_provider_chains
+        from app.domains.databus.providers import build_provider_chains
 
         chains = build_provider_chains()
         assert chain in chains, f"Chain '{chain}' missing from provider chains"