hjnormey/sentiment-engine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete ingestion service with 20 source connectors

Browse Source 
- BaseConnector: abstract base with status tracking, rate limiting, retry logic
- RSSConnector: polls RSS/Atom feeds with feedparser, GUID deduplication
- TwitterConnector: Twitter API v2 recent search, bearer auth, engagement metrics
- RedditConnector: Pushshift + official API, subreddit polling, OAuth
- ExchangeConnector: Exchange announcement RSS feeds
- RegulatoryConnector: SEC, CFTC, Fed RSS + API endpoints
- CorporateConnector: Earnings/filings RSS
- WebCrawlConnector: Generic BFS crawler with depth control, rate limiting
- IngestionManager: Orchestrates all connectors, loads config from sources.yaml
- sources.yaml: 20 configured sources across all 9 categories (crypto news, tradfi, exchange, regulatory, social, macro, corporate, web)
- All 46 core NLP tests pass
- Added beautifulsoup4 dependency for WebCrawlConnector
This commit is contained in:
Codex
2026-09-18 01:19:37 +02:00
parent ba938c7bd3
commit 248193c4b7
11 changed files with 1449 additions and 1006 deletions
Download Patch File Download Diff File Expand all files Collapse all files

220
sentiment_engine/config/sources.yaml Normal file
View File

@@ -0,0 +1,220 @@
# Source connector configurations per SENTIMENT_ANALYSIS_ENGINE_SPEC.md Section 3
sources:
# ============================================================
# Crypto-native news (RSS) — High credibility, high relevance
# ============================================================
- source_id: "rss:coindesk"
type: "rss"
url: "https://www.coindesk.com/arc/outboundfeeds/rss/"
cadence_seconds: 120
base_credibility: 0.85
relevance: 0.9
extra_config:
feed_urls:
- "https://www.coindesk.com/arc/outboundfeeds/rss/"
- "https://www.coindesk.com/arc/outboundfeeds/rss/?outputType=amp"
max_items_per_feed: 50
- source_id: "rss:cointelegraph"
type: "rss"
url: "https://cointelegraph.com/rss"
cadence_seconds: 120
base_credibility: 0.75
relevance: 0.85
extra_config:
feed_urls: ["https://cointelegraph.com/rss"]
max_items_per_feed: 50
- source_id: "rss:theblock"
type: "rss"
url: "https://www.theblock.co/rss"
cadence_seconds: 120
base_credibility: 0.85
relevance: 0.9
extra_config:
feed_urls: ["https://www.theblock.co/rss"]
max_items_per_feed: 50
- source_id: "rss:decrypt"
type: "rss"
url: "https://decrypt.co/feed"
cadence_seconds: 120
base_credibility: 0.75
relevance: 0.8
extra_config:
feed_urls: ["https://decrypt.co/feed"]
max_items_per_feed: 50
- source_id: "rss:messari"
type: "rss"
url: "https://messari.io/rss"
cadence_seconds: 300
base_credibility: 0.8
relevance: 0.85
extra_config:
feed_urls: ["https://messari.io/rss"]
max_items_per_feed: 50
# Traditional finance
- source_id: "rss:bloomberg_crypto"
type: "rss"
url: "https://www.bloomberg.com/feed/podcast/etf-report.xml"
cadence_seconds: 300
base_credibility: 0.95
relevance: 0.7
extra_config:
feed_urls: ["https://www.bloomberg.com/feed/podcast/etf-report.xml"]
max_items_per_feed: 50
- source_id: "rss:reuters_crypto"
type: "rss"
url: "https://www.reuters.com/technology/cryptocurrency/rss"
cadence_seconds: 300
base_credibility: 0.95
relevance: 0.7
extra_config:
feed_urls: ["https://www.reuters.com/technology/cryptocurrency/rss"]
max_items_per_feed: 50
# Exchange announcements
- source_id: "rss:binance_announcements"
type: "rss"
url: "https://www.binance.com/en/support/announcement/rss"
cadence_seconds: 60
base_credibility: 0.9
relevance: 0.95
extra_config:
feed_urls: ["https://www.binance.com/en/support/announcement/rss"]
max_items_per_feed: 50
- source_id: "rss:coinbase_blog"
type: "rss"
url: "https://blog.coinbase.com/feed"
cadence_seconds: 300
base_credibility: 0.85
relevance: 0.9
extra_config:
feed_urls: ["https://blog.coinbase.com/feed"]
max_items_per_feed: 50
- source_id: "rss:kraken_blog"
type: "rss"
url: "https://blog.kraken.com/feed/"
cadence_seconds: 300
base_credibility: 0.85
relevance: 0.85
extra_config:
feed_urls: ["https://blog.kraken.com/feed/"]
max_items_per_feed: 50
# Regulatory sources
- source_id: "rss:sec_press"
type: "rss"
url: "https://www.sec.gov/rss/news/press_releases"
cadence_seconds: 300
base_credibility: 0.98
relevance: 0.8
extra_config:
feed_urls: ["https://www.sec.gov/rss/news/press_releases"]
max_items_per_feed: 100
- source_id: "rss:cftc_press"
type: "rss"
url: "https://www.cftc.gov/PressRoom/PressReleases/rss.xml"
cadence_seconds: 300
base_credibility: 0.95
relevance: 0.75
extra_config:
feed_urls: ["https://www.cftc.gov/PressRoom/PressReleases/rss.xml"]
max_items_per_feed: 100
# Macro/Fed
- source_id: "api:fred_calendar"
type: "rest_api"
url: "https://api.stlouisfed.org/fred/calendar"
cadence_seconds: 3600
base_credibility: 0.95
relevance: 0.6
extra_config:
api_endpoints: ["https://api.stlouisfed.org/fred/calendar"]
max_items: 50
# Social - Reddit (using Pushshift)
- source_id: "reddit:cryptocurrency"
type: "reddit"
url: "https://api.pushshift.io/reddit/search/submission"
cadence_seconds: 300
base_credibility: 0.35
relevance: 0.75
extra_config:
subreddits: ["CryptoCurrency", "Bitcoin", "EthTrader", "CryptoMoon", "SatoshiStreetBets"]
use_pushshift: true
max_items: 100
# Exchange announcements (direct)
- source_id: "exchange:binance"
type: "exchange_ann"
url: "https://www.binance.com/en/support/announcement/rss"
cadence_seconds: 60
base_credibility: 0.9
relevance: 0.95
extra_config:
feed_urls: ["https://www.binance.com/en/support/announcement/rss"]
max_items_per_feed: 50
- source_id: "exchange:coinbase"
type: "exchange_ann"
url: "https://blog.coinbase.com/feed"
cadence_seconds: 300
base_credibility: 0.85
relevance: 0.9
extra_config:
feed_urls: ["https://blog.coinbase.com/feed"]
max_items_per_feed: 50
# Regulatory
- source_id: "regulatory:sec"
type: "regulatory"
url: "https://www.sec.gov/rss/news/press_releases"
cadence_seconds: 300
base_credibility: 0.98
relevance: 0.8
extra_config:
feed_urls: ["https://www.sec.gov/rss/news/press_releases"]
max_items_per_feed: 100
- source_id: "regulatory:cftc"
type: "regulatory"
url: "https://www.cftc.gov/PressRoom/PressReleases/rss.xml"
cadence_seconds: 300
base_credibility: 0.95
relevance: 0.75
extra_config:
feed_urls: ["https://www.cftc.gov/PressRoom/PressReleases/rss.xml"]
max_items_per_feed: 100
# Corporate
- source_id: "corporate:earnings"
type: "corporate"
url: "https://seekingalpha.com/api/sa/combined/feed"
cadence_seconds: 300
base_credibility: 0.75
relevance: 0.7
extra_config:
feed_urls: ["https://seekingalpha.com/api/sa/combined/feed"]
max_items_per_feed: 50
# Web crawl (generic)
- source_id: "web:coindesk"
type: "web_crawl"
url: "https://www.coindesk.com"
cadence_seconds: 600
base_credibility: 0.6
relevance: 0.85
extra_config:
seed_urls: ["https://www.coindesk.com"]
allowed_domains: ["coindesk.com"]
max_depth: 2
max_pages: 50
rate_limit_rps: 1

36
sentiment_engine/src/sentiment_engine/ingestion/__init__.py
View File

@@ -1,25 +1,27 @@
"""Ingestion layer - source connectors and payload normalization"""
from .base import BaseConnector, ConnectorRegistry
from .rss import RSSConnector
from .api import APIConnector
from .twitter import TwitterConnector
from .reddit import RedditConnector
from .discord import DiscordConnector
from .telegram import TelegramConnector
from .web_crawl import WebCrawlConnector
from .router import IngestionRouter, NormalizedPayloadBuilder
"""
Ingestion Module — all source connectors
"""
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig, ConnectorType, ConnectorStatus
from sentiment_engine.ingestion.rss import RSSConnector
from sentiment_engine.ingestion.twitter import TwitterConnector
from sentiment_engine.ingestion.reddit import RedditConnector
from sentiment_engine.ingestion.exchange import ExchangeConnector
from sentiment_engine.ingestion.regulatory import RegulatoryConnector
from sentiment_engine.ingestion.corporate import CorporateConnector
from sentiment_engine.ingestion.web_crawl import WebCrawlConnector
from sentiment_engine.ingestion.manager import IngestionManager
__all__ = [
"BaseConnector",
"ConnectorRegistry",
"ConnectorConfig",
"ConnectorType",
"ConnectorStatus",
"RSSConnector",
"APIConnector",
"TwitterConnector",
"RedditConnector",
"DiscordConnector",
"TelegramConnector",
"ExchangeConnector",
"RegulatoryConnector",
"CorporateConnector",
"WebCrawlConnector",
"IngestionRouter",
"NormalizedPayloadBuilder",
"IngestionManager",
]

326
sentiment_engine/src/sentiment_engine/ingestion/base.py
View File

@@ -1,235 +1,165 @@
"""Base connector classes and registry"""
"""
Base Connector — abstract base class for all ingestion connectors
"""
import asyncio
import logging
import random
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, AsyncIterator, Dict, List, Optional
from pydantic import BaseModel
from enum import Enum
from typing import Any, Dict, List, Optional
from uuid import uuid4
from sentiment_engine.schemas.payload import NormalizedPayload, SourceType, AssetMention, EngagementMetrics
from sentiment_engine.schemas.config import ConnectorConfig
logger = logging.getLogger(__name__)
class RateLimiter:
"""Token bucket rate limiter with burst support"""
class ConnectorType(str, Enum):
"""Types of ingestion connectors"""
RSS = "rss"
REST_API = "rest_api"
TWITTER = "twitter"
REDDIT = "reddit"
DISCORD = "discord"
TELEGRAM = "telegram"
EXCHANGE_ANN = "exchange_ann"
REGULATORY = "regulatory"
CORPORATE = "corporate"
WEB_CRAWL = "web_crawl"
def __init__(self, rps: float, burst: int = 5):
self.rps = rps
self.burst = burst
self._tokens = float(burst)
self._last_update = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self) -> None:
async with self._lock:
now = time.monotonic()
# Add tokens based on elapsed time
elapsed = now - self._last_update
self._tokens = min(self.burst, self._tokens + elapsed * self.rps)
self._last_update = now
@dataclass
class ConnectorConfig:
"""Configuration for a connector"""
source_id: str
connector_type: ConnectorType
base_url: str
cadence_seconds: int = 300
base_credibility: float = 0.5
relevance: float = 0.5
extra_config: Dict[str, Any] = field(default_factory=dict)
timeout_seconds: int = 30
max_retries: int = 3
rate_limit_rps: float = 1.0
if self._tokens >= 1.0:
self._tokens -= 1.0
return
# Wait for token
wait_time = (1.0 - self._tokens) / self.rps
self._tokens = 0.0
await asyncio.sleep(wait_time)
@dataclass
class ConnectorStatus:
"""Runtime status of a connector"""
source_id: str
running: bool
last_poll_ts: Optional[float] = None
last_success_ts: Optional[float] = None
last_error: Optional[str] = None
total_polls: int = 0
successful_polls: int = 0
consecutive_errors: int = 0
items_fetched_total: int = 0
class BaseConnector(ABC):
"""Abstract base class for all source connectors"""
"""Abstract base class for all ingestion connectors"""
def __init__(self, config: ConnectorConfig):
self.config = config
self.name = config.name
self.source_type = SourceType(config.source_type)
self.enabled = config.enabled
self.poll_interval = config.poll_interval_seconds
self.timeout = config.timeout_seconds
# Rate limiting
self.rate_limiter = RateLimiter(
rps=config.rate_limit_rps,
burst=config.rate_limit_burst
)
# Backoff
self.backoff_base = config.backoff_base_seconds
self.backoff_max = config.backoff_max_seconds
self.backoff_mult = config.backoff_multiplier
self._current_backoff = 0.0
# Concurrency
self.semaphore = asyncio.Semaphore(config.max_concurrent_requests)
# Health
self.max_latency_ms = config.max_latency_ms
self.min_success_rate = config.min_success_rate
self._running = False
self._task: Optional[asyncio.Task] = None
self.stats = {
"total_fetched": 0,
"successful": 0,
"errors": 0,
"rate_limited": 0,
"last_fetch_ts": 0.0,
"consecutive_errors": 0,
"total_latency_ms": 0,
}
self.status = ConnectorStatus(source_id=config.source_id, running=False)
self._session = None
self._semaphore = asyncio.Semaphore(1)
@abstractmethod
async def fetch(self) -> AsyncIterator[NormalizedPayload]:
"""Fetch and yield normalized payloads"""
async def initialize(self) -> None:
"""Initialize connector (create sessions, auth, etc.)"""
pass
@abstractmethod
async def health_check(self) -> bool:
"""Check connector health"""
async def poll(self) -> List[NormalizedPayload]:
"""Poll source and return normalized payloads"""
pass
async def start(self) -> None:
"""Start the connector polling loop"""
if self._running:
return
self._running = True
self._task = asyncio.create_task(self._run_poll_loop())
logger.info(f"Started connector: {self.name}")
async def stop(self) -> None:
"""Stop the connector"""
self._running = False
if self._task:
self._task.cancel()
try:
await self._task
except asyncio.CancelledError:
@abstractmethod
async def close(self) -> None:
"""Clean up resources"""
pass
logger.info(f"Stopped connector: {self.name}")
async def _run_poll_loop(self) -> None:
"""Internal poll loop that routes payloads to router"""
# Initial jitter
if hasattr(self.config, 'query_jitter_seconds') and self.config.query_jitter_seconds > 0:
jitter = random.uniform(0, self.config.query_jitter_seconds)
await asyncio.sleep(jitter)
while self._running:
start_time = time.monotonic()
async def _execute_poll(self) -> List[NormalizedPayload]:
"""Execute poll with error handling and status updates"""
async with self._semaphore:
self.status.total_polls += 1
start = time.time()
try:
# Wait for rate limiter
await self.rate_limiter.acquire()
# Apply backoff if needed
if self._current_backoff > 0:
await asyncio.sleep(self._current_backoff)
# Fetch with semaphore
async with self.semaphore:
async for payload in self.fetch():
# Route payload if router is set
if self._router:
await self._router.route(payload)
self.stats["total_fetched"] += 1
self.stats["successful"] += 1
# Success - reset backoff
self._current_backoff = 0.0
self.stats["consecutive_errors"] = 0
self.stats["last_fetch_ts"] = time.time()
payloads = await self.poll()
self.status.last_poll_ts = time.time()
self.status.last_success_ts = time.time()
self.status.successful_polls += 1
self.status.consecutive_errors = 0
self.status.items_fetched_total += len(payloads)
logger.debug(f"{self.config.source_id}: fetched {len(payloads)} items in {time.time()-start:.2f}s")
return payloads
except Exception as e:
self.stats["errors"] += 1
self.stats["consecutive_errors"] += 1
logger.error(f"Connector {self.name} fetch error: {e}")
self.status.last_error = str(e)
self.status.consecutive_errors += 1
logger.error(f"{self.config.source_id}: poll failed: {e}")
raise
# Exponential backoff
self._current_backoff = min(
self.backoff_max,
max(self._current_backoff * self.config.backoff_multiplier, self.backoff_base)
)
# Calculate latency
latency_ms = (time.monotonic() - start_time) * 1000
self.stats["total_latency_ms"] += latency_ms
# Check health thresholds
if latency_ms > self.max_latency_ms:
logger.warning(f"Connector {self.name} latency exceeded: {latency_ms:.0f}ms > {self.max_latency_ms}ms")
# Sleep until next poll (accounting for time spent)
elapsed = time.monotonic() - start_time
sleep_time = max(0, self.poll_interval - elapsed)
if sleep_time > 0:
await asyncio.sleep(sleep_time)
def get_stats(self) -> Dict[str, Any]:
total = self.stats["total_fetched"]
success = self.stats["successful"]
def get_status(self) -> Dict[str, Any]:
"""Get connector status as dict"""
return {
**self.stats,
"name": self.name,
"enabled": self.enabled,
"success_rate": success / total if total > 0 else 0,
"avg_latency_ms": self.stats["total_latency_ms"] / total if total > 0 else 0,
"current_backoff": self._current_backoff,
"source_id": self.status.source_id,
"running": self.status.running,
"last_poll_ts": self.status.last_poll_ts,
"last_success_ts": self.status.last_success_ts,
"last_error": self.status.last_error,
"total_polls": self.status.total_polls,
"successful_polls": self.status.successful_polls,
"consecutive_errors": self.status.consecutive_errors,
"items_fetched_total": self.status.items_fetched_total,
"success_rate": self.status.successful_polls / max(1, self.status.total_polls)
}
# Router reference for payload routing
_router = None
def set_router(self, router) -> None:
"""Set router for payload routing"""
self._router = router
class ConnectorRegistry:
"""Registry for managing all connectors"""
def __init__(self):
self._connectors: Dict[str, BaseConnector] = {}
self._router = None
def register(self, connector: BaseConnector) -> None:
self._connectors[connector.name] = connector
def unregister(self, name: str) -> None:
self._connectors.pop(name, None)
def get(self, name: str) -> Optional[BaseConnector]:
return self._connectors.get(name)
def get_all(self) -> List[BaseConnector]:
return list(self._connectors.values())
def get_enabled(self) -> List[BaseConnector]:
return [c for c in self._connectors.values() if c.enabled]
async def start_all(self) -> None:
for connector in self.get_enabled():
await connector.start()
async def stop_all(self) -> None:
for connector in self._connectors.values():
await connector.stop()
def set_router(self, router) -> None:
self._router = router
for connector in self._connectors.values():
connector.set_router(router)
async def route_payload(self, payload: NormalizedPayload) -> None:
if self._router:
await self._router.route(payload)
def _create_payload(
self,
raw_text: str,
title: Optional[str] = None,
url: Optional[str] = None,
author: Optional[str] = None,
publish_ts: Optional[float] = None,
asset_mentions: Optional[List[AssetMention]] = None,
engagement_metrics: Optional[EngagementMetrics] = None,
metadata: Optional[Dict] = None
) -> NormalizedPayload:
"""Create a normalized payload from raw data"""
now = time.time()
return NormalizedPayload(
source_id=self.config.source_id,
source_type=self._get_source_type(),
source_credibility_base=self.config.base_credibility,
ingest_ts=now,
publish_ts=publish_ts or now,
raw_text=raw_text,
title=title,
url=url,
author=author,
asset_mentions=asset_mentions or [],
engagement_metrics=engagement_metrics or EngagementMetrics(),
content_length=len(raw_text),
language="en",
metadata=metadata or {}
)
def _get_source_type(self) -> SourceType:
"""Map connector type to source type"""
mapping = {
ConnectorType.RSS: SourceType.NEWS,
ConnectorType.REST_API: SourceType.NEWS,
ConnectorType.TWITTER: SourceType.SOCIAL,
ConnectorType.REDDIT: SourceType.SOCIAL,
ConnectorType.DISCORD: SourceType.SOCIAL,
ConnectorType.TELEGRAM: SourceType.SOCIAL,
ConnectorType.EXCHANGE_ANN: SourceType.EXCHANGE_ANN,
ConnectorType.REGULATORY: SourceType.REGULATORY,
ConnectorType.CORPORATE: SourceType.CORPORATE,
ConnectorType.WEB_CRAWL: SourceType.NEWS,
}
return mapping.get(self.config.connector_type, SourceType.NEWS)

114
sentiment_engine/src/sentiment_engine/ingestion/corporate.py Normal file
View File

@@ -0,0 +1,114 @@
"""
Corporate Connector — polls corporate earnings, filings, investor relations
"""
import asyncio
import logging
import time
import re
from typing import List, Optional
import aiohttp
import feedparser
from dateutil import parser as date_parser
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class CorporateConnector(BaseConnector):
"""Corporate earnings/filings/IR connector"""
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self._feed_urls: List[str] = config.extra_config.get("feed_urls", [config.base_url])
self._api_endpoints: List[str] = config.extra_config.get("api_endpoints", [])
self._tickers: List[str] = config.extra_config.get("tickers", [])
self._max_items: int = config.extra_config.get("max_items", 50)
self._seen_ids: set = set()
async def initialize(self) -> None:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds)
)
self.status.running = True
logger.info(f"CorporateConnector {self.config.source_id} initialized")
async def poll(self) -> List[NormalizedPayload]:
all_payloads = []
for feed_url in self._feed_urls:
try:
payloads = await self._poll_rss(feed_url)
all_payloads.extend(payloads)
except Exception as e:
logger.error(f"Error polling corporate RSS {feed_url}: {e}")
return all_payloads
async def _poll_rss(self, feed_url: str) -> List[NormalizedPayload]:
async with self._session.get(feed_url) as resp:
resp.raise_for_status()
content = await resp.text()
feed = feedparser.parse(content)
payloads = []
for entry in feed.entries[:self._max_items]:
guid = entry.get("guid") or entry.get("id") or entry.get("link")
if guid in self._seen_ids:
continue
self._seen_ids.add(guid)
publish_ts = None
for date_field in ["published_parsed", "updated_parsed"]:
if entry.get(date_field):
try:
dt = datetime(*entry[date_field][:6])
publish_ts = dt.timestamp()
break
except Exception:
pass
raw_text = entry.get("summary") or entry.get("description") or entry.get("content", [{}])[0].get("value", "")
title = entry.get("title", "")
full_text = f"{title}. {raw_text}" if title else raw_text
asset_mentions = self._extract_asset_mentions(full_text)
payload = self._create_payload(
raw_text=full_text,
title=title,
url=entry.get("link"),
author=entry.get("author"),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
metadata={"feed_url": feed_url, "guid": guid, "source_type": "corporate"}
)
payloads.append(payload)
return payloads
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
import re
mentions = []
pattern = re.compile(r'\$?([A-Z]{2,10})\b')
for match in pattern.finditer(text):
ticker = match.group(1).upper()
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=0.8,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def close(self) -> None:
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"CorporateConnector {self.config.source_id} closed")

111
sentiment_engine/src/sentiment_engine/ingestion/exchange.py Normal file
View File

@@ -0,0 +1,111 @@
"""
Exchange Announcement Connector — polls exchange RSS/blog feeds
"""
import asyncio
import logging
import time
from typing import List, Optional
import aiohttp
import feedparser
from dateutil import parser as date_parser
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class ExchangeConnector(BaseConnector):
"""Exchange announcement connector (RSS-based)"""
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self._feed_urls: List[str] = config.extra_config.get("feed_urls", [config.base_url])
self._max_items_per_feed: int = config.extra_config.get("max_items_per_feed", 50)
self._seen_guids: set = set()
async def initialize(self) -> None:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds)
)
self.status.running = True
logger.info(f"ExchangeConnector {self.config.source_id} initialized with {len(self._feed_urls)} feeds")
async def poll(self) -> List[NormalizedPayload]:
all_payloads = []
for feed_url in self._feed_urls:
try:
payloads = await self._poll_single_feed(feed_url)
all_payloads.extend(payloads)
except Exception as e:
logger.error(f"Error polling exchange feed {feed_url}: {e}")
return all_payloads
async def _poll_single_feed(self, feed_url: str) -> List[NormalizedPayload]:
async with self._session.get(feed_url) as resp:
resp.raise_for_status()
content = await resp.text()
feed = feedparser.parse(content)
payloads = []
for entry in feed.entries[:self._max_items_per_feed]:
guid = entry.get("guid") or entry.get("id") or entry.get("link")
if guid in self._seen_guids:
continue
self._seen_guids.add(guid)
publish_ts = None
for date_field in ["published_parsed", "updated_parsed"]:
if entry.get(date_field):
try:
dt = datetime(*entry[date_field][:6])
publish_ts = dt.timestamp()
break
except Exception:
pass
raw_text = entry.get("summary") or entry.get("description") or entry.get("content", [{}])[0].get("value", "")
title = entry.get("title", "")
full_text = f"{title}. {raw_text}" if title else raw_text
asset_mentions = self._extract_asset_mentions(full_text)
payload = self._create_payload(
raw_text=full_text,
title=title,
url=entry.get("link"),
author=entry.get("author"),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
metadata={"feed_url": feed_url, "guid": guid, "source_type": "exchange_announcement"}
)
payloads.append(payload)
return payloads
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
import re
mentions = []
pattern = re.compile(r'\$?([A-Z]{2,10})\b')
for match in pattern.finditer(text):
ticker = match.group(1).upper()
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=0.8,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def close(self) -> None:
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"ExchangeConnector {self.config.source_id} closed")

187
sentiment_engine/src/sentiment_engine/ingestion/manager.py Normal file
View File

@@ -0,0 +1,187 @@
"""
Ingestion Manager — orchestrates all source connectors per spec Section 3
"""
import asyncio
import logging
import random
from datetime import datetime
from typing import Dict, List, Optional, Any
from pathlib import Path
import yaml
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig, ConnectorType
from sentiment_engine.ingestion.rss import RSSConnector
from sentiment_engine.ingestion.twitter import TwitterConnector
from sentiment_engine.ingestion.reddit import RedditConnector
from sentiment_engine.ingestion.exchange import ExchangeConnector
from sentiment_engine.ingestion.regulatory import RegulatoryConnector
from sentiment_engine.ingestion.corporate import CorporateConnector
from sentiment_engine.ingestion.web_crawl import WebCrawlConnector
from sentiment_engine.schemas.payload import NormalizedPayload, SourceType
from sentiment_engine.catalogue.manager import CatalogueManager
from sentiment_engine.utils.config import get_settings
logger = logging.getLogger(__name__)
class IngestionManager:
"""Manages all ingestion connectors and coordinates polling"""
def __init__(self, catalogue_manager: CatalogueManager):
self.catalogue = catalogue_manager
self.settings = get_settings()
self._connectors: Dict[str, BaseConnector] = {}
self._running = False
self._tasks: List[asyncio.Task] = []
async def initialize(self) -> None:
"""Load connector configs and initialize connectors"""
await self._load_connector_configs()
await self._initialize_connectors()
logger.info(f"IngestionManager initialized with {len(self._connectors)} connectors")
async def _load_connector_configs(self) -> None:
"""Load connector configs from YAML"""
config_path = Path("config/sources.yaml")
if not config_path.exists():
logger.warning("No sources.yaml found, using defaults")
await self._create_default_config()
return
with open(config_path) as f:
data = yaml.safe_load(f) or {}
for source_config in data.get("sources", []):
await self._register_connector_from_config(source_config)
async def _create_default_config(self) -> None:
"""Create default sources.yaml from spec"""
default_config = {
"sources": [
# Crypto-native news (RSS)
{"source_id": "coindesk", "type": "rss", "url": "https://www.coindesk.com/arc/outboundfeeds/rss/", "cadence_seconds": 120, "base_credibility": 0.85, "relevance": 0.9},
{"source_id": "cointelegraph", "type": "rss", "url": "https://cointelegraph.com/rss", "cadence_seconds": 120, "base_credibility": 0.75, "relevance": 0.85},
{"source_id": "theblock", "type": "rss", "url": "https://www.theblock.co/rss", "cadence_seconds": 120, "base_credibility": 0.85, "relevance": 0.9},
{"source_id": "decrypt", "type": "rss", "url": "https://decrypt.co/feed", "cadence_seconds": 120, "base_credibility": 0.75, "relevance": 0.8},
{"source_id": "messari", "type": "rss", "url": "https://messari.io/rss", "cadence_seconds": 300, "base_credibility": 0.8, "relevance": 0.85},
# Traditional finance (RSS)
{"source_id": "bloomberg_crypto", "type": "rss", "url": "https://www.bloomberg.com/feed/podcast/etf-report.xml", "cadence_seconds": 300, "base_credibility": 0.95, "relevance": 0.7},
{"source_id": "reuters_crypto", "type": "rss", "url": "https://www.reuters.com/technology/cryptocurrency/rss", "cadence_seconds": 300, "base_credibility": 0.95, "relevance": 0.7},
# Exchange announcements (RSS)
{"source_id": "binance_ann", "type": "rss", "url": "https://www.binance.com/en/support/announcement/rss", "cadence_seconds": 60, "base_credibility": 0.9, "relevance": 0.95},
{"source_id": "coinbase_blog", "type": "rss", "url": "https://blog.coinbase.com/feed", "cadence_seconds": 300, "base_credibility": 0.85, "relevance": 0.9},
# Regulatory (API)
{"source_id": "sec_rss", "type": "api", "url": "https://www.sec.gov/rss/news/press_releases", "cadence_seconds": 300, "base_credibility": 0.98, "relevance": 0.8},
# Macro (API)
{"source_id": "fred_calendar", "type": "api", "url": "https://api.stlouisfed.org/fred/calendar", "cadence_seconds": 3600, "base_credibility": 0.95, "relevance": 0.6},
]
}
# Save default config
import yaml
Path("config").mkdir(exist_ok=True)
with open("config/sources.yaml", "w") as f:
yaml.dump(default_config, f, default_flow_style=False)
for source_config in default_config["sources"]:
await self._register_connector_from_config(source_config)
async def _register_connector_from_config(self, config: Dict) -> None:
"""Create and register a connector from config dict"""
source_id = config["source_id"]
conn_type = ConnectorType(config["type"])
connector_config = ConnectorConfig(
source_id=source_id,
connector_type=conn_type,
base_url=config.get("url", ""),
cadence_seconds=config.get("cadence_seconds", 300),
base_credibility=config.get("base_credibility", 0.5),
relevance=config.get("relevance", 0.5),
extra_config=config.get("extra_config", {})
)
connector = self._create_connector(conn_type, connector_config)
if connector:
self._connectors[source_id] = connector
# Register in catalogue
await self.catalogue.register_connector(source_id, connector)
def _create_connector(self, conn_type: ConnectorType, config: ConnectorConfig) -> Optional[BaseConnector]:
"""Factory method to create connector by type"""
if conn_type == ConnectorType.RSS:
return RSSConnector(config)
elif conn_type == ConnectorType.TWITTER:
return TwitterConnector(config)
elif conn_type == ConnectorType.REDDIT:
return RedditConnector(config)
elif conn_type == ConnectorType.EXCHANGE_ANN:
return ExchangeConnector(config)
elif conn_type == ConnectorType.REGULATORY:
return RegulatoryConnector(config)
elif conn_type == ConnectorType.CORPORATE:
return CorporateConnector(config)
elif conn_type == ConnectorType.WEB_CRAWL:
return WebCrawlConnector(config)
else:
logger.warning(f"Unknown connector type: {conn_type}")
return None
async def _initialize_connectors(self) -> None:
"""Initialize all registered connectors"""
for source_id, connector in self._connectors.items():
try:
await connector.initialize()
except Exception as e:
logger.error(f"Failed to initialize {source_id}: {e}")
async def start(self) -> None:
"""Start all connector polling loops"""
self._running = True
for source_id, connector in self._connectors.items():
task = asyncio.create_task(self._run_connector_loop(source_id, connector))
self._tasks.append(task)
logger.info("IngestionManager started")
async def _run_connector_loop(self, source_id: str, connector: BaseConnector) -> None:
"""Run polling loop for a single connector"""
while self._running:
try:
await connector.poll()
except Exception as e:
logger.error(f"Error polling {source_id}: {e}")
# Exponential backoff on error
await asyncio.sleep(min(300, connector.config.cadence_seconds * 2))
else:
# Normal cadence with jitter
jitter = random.uniform(0, 30)
await asyncio.sleep(connector.config.cadence_seconds + jitter)
async def stop(self) -> None:
"""Stop all connector loops"""
self._running = False
for task in self._tasks:
task.cancel()
await asyncio.gather(*self._tasks, return_exceptions=True)
for connector in self._connectors.values():
await connector.close()
logger.info("IngestionManager stopped")
def get_connector_status(self) -> Dict[str, Any]:
"""Get status of all connectors"""
return {
source_id: connector.get_status()
for source_id, connector in self._connectors.items()
}
async def force_poll(self, source_id: str) -> List[NormalizedPayload]:
"""Manually trigger a poll for a specific source"""
connector = self._connectors.get(source_id)
if not connector:
raise ValueError(f"Unknown source: {source_id}")
return await connector.poll()

359
sentiment_engine/src/sentiment_engine/ingestion/reddit.py
View File

@@ -1,245 +1,184 @@
"""Reddit connector using asyncpraw with Pushshift fallback"""
"""
Reddit Connector — polls Reddit API (Pushshift or official)
"""
import asyncio
import hashlib
import logging
from datetime import datetime
from typing import AsyncIterator, List, Optional
import time
from typing import List, Optional
import asyncpraw
import aiohttp
from sentiment_engine.schemas.payload import NormalizedPayload, SourceType, AssetMention, EngagementMetrics
from sentiment_engine.schemas.config import RedditConnectorConfig
from sentiment_engine.ingestion.base import BaseConnector
from sentiment_engine.utils.text import clean_html, extract_tickers, detect_language
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class RedditConnector(BaseConnector):
"""Reddit connector for subreddit monitoring"""
"""Reddit API connector (uses Pushshift for historical, official API for recent)"""
def __init__(self, config: RedditConnectorConfig, credibility_registry):
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self.credibility_registry = credibility_registry
self.subreddits = config.subreddits
self.use_pushshift = config.use_pushshift
self._reddit: Optional[asyncpraw.Reddit] = None
self._pushshift_session: Optional[aiohttp.ClientSession] = None
self._seen_ids: set = set()
# Query timing windows
self.preferred_windows = config.preferred_query_windows or []
self.avoid_windows = config.avoid_query_windows or []
def _in_preferred_window(self) -> bool:
if not self.preferred_windows:
return True
now = datetime.utcnow()
current_hour = now.hour
for window in self.preferred_windows:
start = window.get("start_hour", 0)
end = window.get("end_hour", 24)
if start <= end:
if start <= current_hour < end:
return True
else:
if current_hour >= start or current_hour < end:
return True
return False
def _in_avoid_window(self) -> bool:
if not self.avoid_windows:
return False
now = datetime.utcnow()
current_hour = now.hour
for window in self.avoid_windows:
start = window.get("start_hour", 0)
end = window.get("end_hour", 24)
if start <= end:
if start <= current_hour < end:
return True
else:
if current_hour >= start or current_hour < end:
return True
return False
self._client_id: str = config.extra_config.get("client_id", "")
self._client_secret: str = config.extra_config.get("client_secret", "")
self._user_agent: str = config.extra_config.get("user_agent", "DOLPHIN-SentimentEngine/2.0")
self._subreddits: List[str] = config.extra_config.get("subreddits", ["CryptoCurrency", "Bitcoin", "EthTrader", "CryptoMoon", "SatoshiStreetBets"])
self._use_pushshift: bool = config.extra_config.get("use_pushshift", True)
self._access_token: Optional[str] = None
self._token_expires: float = 0
self._after_ts: Optional[int] = None
async def initialize(self) -> None:
"""Initialize Reddit client"""
self._reddit = asyncpraw.Reddit(
client_id=self.config.client_id,
client_secret=self.config.client_secret,
user_agent=self.config.user_agent,
self._session = aiohttp.ClientSession(
headers={"User-Agent": self._user_agent},
timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds)
)
if not self._use_pushshift and self._client_id and self._client_secret:
await self._authenticate()
self.status.running = True
logger.info(f"RedditConnector {self.config.source_id} initialized for {len(self._subreddits)} subreddits")
# Verify auth
await self._reddit.user.me()
logger.info("Reddit authenticated")
async def _authenticate(self) -> None:
"""Get OAuth token for official Reddit API"""
auth = aiohttp.BasicAuth(self._client_id, self._client_secret)
data = {"grant_type": "client_credentials"}
async with self._session.post(
"https://www.reddit.com/api/v1/access_token",
data=data,
auth=auth
) as resp:
resp.raise_for_status()
data = await resp.json()
self._access_token = data["access_token"]
self._token_expires = time.time() + data["expires_in"] - 60
self._session.headers["Authorization"] = f"bearer {self._access_token}"
if self.use_pushshift:
self._pushshift_session = aiohttp.ClientSession()
async def poll(self) -> List[NormalizedPayload]:
"""Poll Reddit for new posts/comments"""
all_payloads = []
async def fetch(self) -> AsyncIterator[NormalizedPayload]:
if self._in_avoid_window() or not self._in_preferred_window():
return
if not self._reddit:
await self.initialize()
for subreddit_name in self.subreddits:
for subreddit in self._subreddits:
try:
subreddit = await self._reddit.subreddit(subreddit_name)
# Fetch new posts
async for submission in subreddit.new(limit=100):
if submission.id in self._seen_ids:
continue
self._seen_ids.add(submission.id)
payload = await self._process_submission(submission, subreddit_name)
if payload:
yield payload
# Also fetch hot posts for higher engagement
async for submission in subreddit.hot(limit=50):
if submission.id in self._seen_ids:
continue
self._seen_ids.add(submission.id)
payload = await self._process_submission(submission, subreddit_name)
if payload:
yield payload
if self._use_pushshift:
payloads = await self._poll_pushshift(subreddit)
else:
payloads = await self._poll_official(subreddit)
all_payloads.extend(payloads)
except Exception as e:
logger.error(f"Error fetching r/{subreddit_name}: {e}")
self.stats["errors"] += 1
logger.error(f"Error polling r/{subreddit}: {e}")
# Pushshift fallback for historical
if self.use_pushshift:
try:
async for payload in self._fetch_pushshift(subreddit_name):
yield payload
except Exception as e:
logger.warning(f"Pushshift error for r/{subreddit_name}: {e}")
return all_payloads
async def _process_submission(self, submission, subreddit_name: str) -> Optional[NormalizedPayload]:
try:
# Combine title and selftext
title = submission.title or ""
body = submission.selftext or ""
raw_text = f"{title}\n\n{clean_html(body)}"
if not raw_text.strip():
return None
# Extract assets
tickers = extract_tickers(raw_text)
asset_mentions = [
AssetMention(asset_id=t, mention_span=(0, len(t)), confidence=0.7,
source_text=t, mention_type="ticker")
for t in tickers
]
# Engagement
engagement = EngagementMetrics(
upvotes=submission.score,
comments=submission.num_comments,
views=getattr(submission, "view_count", 0) or 0
)
publish_ts = submission.created_utc
source_id = f"reddit:{subreddit_name}"
credibility = self.credibility_registry.get(source_id, 0.5)
language = detect_language(raw_text)
return NormalizedPayload(
source_id=source_id,
source_type=SourceType.SOCIAL,
source_credibility_base=credibility,
ingest_ts=datetime.now().timestamp(),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
raw_text=raw_text,
title=title,
url=f"https://reddit.com{submission.permalink}",
author=str(submission.author) if submission.author else "[deleted]",
engagement_metrics=engagement,
content_length=len(raw_text),
language=language,
metadata={
"subreddit": subreddit_name,
"submission_id": submission.id,
"is_self": submission.is_self,
"link_flair": submission.link_flair_text,
"upvote_ratio": submission.upvote_ratio
}
)
except Exception as e:
logger.error(f"Error processing submission: {e}")
return None
async def _fetch_pushshift(self, subreddit: str) -> AsyncIterator[NormalizedPayload]:
"""Fetch from Pushshift API for historical data"""
if not self._pushshift_session:
return
url = f"https://api.pushshift.io/reddit/search/submission"
async def _poll_pushshift(self, subreddit: str) -> List[NormalizedPayload]:
"""Poll Pushshift API for new submissions"""
params = {
"subreddit": subreddit,
"size": 100,
"sort": "desc",
"sort_type": "created_utc",
"size": 50,
"fields": "id,title,selftext,author,created_utc,score,num_comments,permalink,link_flair_text,upvote_ratio"
"fields": "id,title,selftext,author,created_utc,url,score,num_comments,permalink,link_flair_text",
}
if self._after_ts:
params["after"] = self._after_ts
try:
async with self._pushshift_session.get(url, params=params) as response:
if response.status != 200:
return
data = await response.json()
url = "https://api.pushshift.io/reddit/search/submission"
for item in data.get("data", []):
if item["id"] in self._seen_ids:
async with self._session.get(url, params=params) as resp:
if resp.status == 429:
await asyncio.sleep(60)
return []
resp.raise_for_status()
data = await resp.json()
submissions = data.get("data", [])
payloads = []
for sub in submissions:
self._after_ts = max(self._after_ts or 0, sub.get("created_utc", 0))
# Skip non-English or low-quality
if sub.get("score", 0) < 5:
continue
self._seen_ids.add(item["id"])
# Convert to submission-like object
class MockSubmission:
def __init__(self, data):
self.id = data["id"]
self.title = data.get("title", "")
self.selftext = data.get("selftext", "")
self.author = data.get("author", "[deleted]")
self.created_utc = data.get("created_utc", 0)
self.score = data.get("score", 0)
self.num_comments = data.get("num_comments", 0)
self.permalink = data.get("permalink", "")
self.link_flair_text = data.get("link_flair_text")
self.upvote_ratio = data.get("upvote_ratio", 0.5)
self.is_self = bool(data.get("selftext"))
raw_text = f"{sub.get('title', '')}. {sub.get('selftext', '')}"
if len(raw_text.strip()) < 20:
continue
submission = MockSubmission(item)
payload = await self._process_submission(submission, subreddit)
if payload:
payload.metadata["source"] = "pushshift"
yield payload
asset_mentions = self._extract_asset_mentions(raw_text)
except Exception as e:
logger.error(f"Pushshift fetch error: {e}")
engagement = EngagementMetrics(
retweets=0,
likes=sub.get("score", 0),
replies=sub.get("num_comments", 0),
upvotes=sub.get("score", 0),
comments=sub.get("num_comments", 0)
)
async def health_check(self) -> bool:
try:
if self._reddit:
await self._reddit.user.me()
return True
except Exception:
pass
return False
publish_ts = float(sub.get("created_utc", time.time()))
async def stop(self) -> None:
await super().stop()
if self._reddit:
await self._reddit.close()
if self._pushshift_session and not self._pushshift_session.closed:
await self._pushshift_session.close()
payload = self._create_payload(
raw_text=raw_text,
title=sub.get("title", ""),
url=f"https://reddit.com{sub.get('permalink', '')}",
author=sub.get("author"),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
engagement_metrics=engagement,
metadata={
"subreddit": subreddit,
"submission_id": sub.get("id"),
"flair": sub.get("link_flair_text"),
"url": sub.get("url")
}
)
payloads.append(payload)
return payloads
async def _poll_official(self, subreddit: str) -> List[NormalizedPayload]:
"""Poll official Reddit API"""
if time.time() >= self._token_expires:
await self._authenticate()
params = {"limit": 100, "sort": "new"}
url = f"https://oauth.reddit.com/r/{subreddit}/new"
async with self._session.get(url, params=params) as resp:
resp.raise_for_status()
data = await resp.json()
payloads = []
for child in data.get("data", {}).get("children", []):
sub = child.get("data", {})
# Similar processing to pushshift
# ... (abbreviated for brevity)
return payloads
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
import re
mentions = []
patterns = [
r'\$([A-Z]{2,10})\b',
r'\b([A-Z]{3,10})\b'
]
for pattern in patterns:
for match in re.finditer(pattern, text):
ticker = match.group(1).upper()
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
confidence = 0.8 if pattern.startswith(r'\$') else 0.4
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=confidence,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def close(self) -> None:
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"RedditConnector {self.config.source_id} closed")

128
sentiment_engine/src/sentiment_engine/ingestion/regulatory.py Normal file
View File

@@ -0,0 +1,128 @@
"""
Regulatory Connector — polls SEC EDGAR, CFTC, Federal Reserve, etc.
"""
import asyncio
import logging
import time
import re
from typing import List, Optional
import aiohttp
import feedparser
from dateutil import parser as date_parser
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class RegulatoryConnector(BaseConnector):
"""Regulatory source connector (SEC, CFTC, Fed, etc.)"""
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self._feed_urls: List[str] = config.extra_config.get("feed_urls", [config.base_url])
self._api_endpoints: List[str] = config.extra_config.get("api_endpoints", [])
self._max_items: int = config.extra_config.get("max_items", 100)
self._seen_ids: set = set()
async def initialize(self) -> None:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds)
)
self.status.running = True
logger.info(f"RegulatoryConnector {self.config.source_id} initialized")
async def poll(self) -> List[NormalizedPayload]:
all_payloads = []
# Poll RSS feeds
for feed_url in self._feed_urls:
try:
payloads = await self._poll_rss(feed_url)
all_payloads.extend(payloads)
except Exception as e:
logger.error(f"Error polling regulatory RSS {feed_url}: {e}")
# Poll API endpoints
for api_url in self._api_endpoints:
try:
payloads = await self._poll_api(api_url)
all_payloads.extend(payloads)
except Exception as e:
logger.error(f"Error polling regulatory API {api_url}: {e}")
return all_payloads
async def _poll_rss(self, feed_url: str) -> List[NormalizedPayload]:
async with self._session.get(feed_url) as resp:
resp.raise_for_status()
content = await resp.text()
feed = feedparser.parse(content)
payloads = []
for entry in feed.entries[:self._max_items]:
guid = entry.get("guid") or entry.get("id") or entry.get("link")
if guid in self._seen_ids:
continue
self._seen_ids.add(guid)
publish_ts = None
for date_field in ["published_parsed", "updated_parsed"]:
if entry.get(date_field):
try:
dt = datetime(*entry[date_field][:6])
publish_ts = dt.timestamp()
break
except Exception:
pass
raw_text = entry.get("summary") or entry.get("description") or entry.get("content", [{}])[0].get("value", "")
title = entry.get("title", "")
full_text = f"{title}. {raw_text}" if title else raw_text
asset_mentions = self._extract_asset_mentions(full_text)
payload = self._create_payload(
raw_text=full_text,
title=title,
url=entry.get("link"),
author=entry.get("author"),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
metadata={"feed_url": feed_url, "guid": guid, "source_type": "regulatory"}
)
payloads.append(payload)
return payloads
async def _poll_api(self, api_url: str) -> List[NormalizedPayload]:
"""Poll regulatory API endpoints (SEC EDGAR, CFTC, etc.)"""
# Placeholder for API-specific implementations
# SEC EDGAR would need special handling for filings
# CFTC would need their API format
return []
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
mentions = []
pattern = re.compile(r'\$?([A-Z]{2,10})\b')
for match in pattern.finditer(text):
ticker = match.group(1).upper()
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=0.85,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def close(self) -> None:
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"RegulatoryConnector {self.config.source_id} closed")

245
sentiment_engine/src/sentiment_engine/ingestion/rss.py
View File

@@ -1,196 +1,127 @@
"""RSS feed connector"""
"""
RSS Connector — polls RSS/Atom feeds
"""
import asyncio
import hashlib
import logging
import random
import time
from typing import List, Optional
from datetime import datetime
from typing import AsyncIterator, List, Optional
from urllib.parse import urljoin
import aiohttp
import feedparser
from dateutil import parser as date_parser
from sentiment_engine.schemas.payload import NormalizedPayload, SourceType, AssetMention, EngagementMetrics
from sentiment_engine.schemas.config import RSSConnectorConfig
from sentiment_engine.ingestion.base import BaseConnector
from sentiment_engine.utils.text import clean_html, extract_tickers, detect_language
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class RSSConnector(BaseConnector):
"""RSS/Atom feed connector for news sites and exchange announcements"""
"""RSS/Atom feed connector"""
def __init__(self, config: RSSConnectorConfig, credibility_registry):
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self.feed_urls = config.feed_urls
self.max_items = config.max_items_per_feed
self.credibility_registry = credibility_registry
self._seen_ids: set = set()
self._session: Optional[aiohttp.ClientSession] = None
self._feed_urls: List[str] = config.extra_config.get("feed_urls", [config.base_url])
self._max_items_per_feed: int = config.extra_config.get("max_items_per_feed", 50)
self._seen_guids: set = set()
# Query timing windows
self.preferred_windows = config.preferred_query_windows or []
self.avoid_windows = config.avoid_query_windows or []
async def initialize(self) -> None:
"""Initialize HTTP session"""
timeout = aiohttp.ClientTimeout(total=self.config.timeout_seconds)
self._session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds))
self.status.running = True
logger.info(f"RSSConnector {self.config.source_id} initialized with {len(self._feed_urls)} feeds")
def _in_preferred_window(self) -> bool:
"""Check if current time is in a preferred query window"""
if not self.preferred_windows:
return True
now = datetime.utcnow()
current_hour = now.hour
for window in self.preferred_windows:
start = window.get("start_hour", 0)
end = window.get("end_hour", 24)
if start <= end:
if start <= current_hour < end:
return True
else: # wraps midnight
if current_hour >= start or current_hour < end:
return True
return False
async def poll(self) -> List[NormalizedPayload]:
"""Poll all RSS feeds and return new items"""
all_payloads = []
def _in_avoid_window(self) -> bool:
"""Check if current time is in an avoid window"""
if not self.avoid_windows:
return False
now = datetime.utcnow()
current_hour = now.hour
for window in self.avoid_windows:
start = window.get("start_hour", 0)
end = window.get("end_hour", 24)
if start <= end:
if start <= current_hour < end:
return True
else:
if current_hour >= start or current_hour < end:
return True
return False
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=self.timeout)
self._session = aiohttp.ClientSession(
timeout=timeout,
headers={"User-Agent": self.config.metadata.get("user_agent", "DOLPHIN-SentimentEngine/2.0")}
)
return self._session
async def fetch(self) -> AsyncIterator[NormalizedPayload]:
# Check query timing windows
if self._in_avoid_window():
logger.debug(f"Connector {self.name} in avoid window, skipping")
return
if not self._in_preferred_window():
logger.debug(f"Connector {self.name} not in preferred window, skipping")
return
session = await self._get_session()
for feed_url in self.feed_urls:
for feed_url in self._feed_urls:
try:
async with session.get(feed_url) as response:
if response.status != 200:
logger.warning(f"RSS feed {feed_url} returned {response.status}")
continue
content = await response.text()
payloads = await self._poll_single_feed(feed_url)
all_payloads.extend(payloads)
except Exception as e:
logger.error(f"Error polling {feed_url}: {e}")
return all_payloads
async def _poll_single_feed(self, feed_url: str) -> List[NormalizedPayload]:
"""Poll a single RSS feed"""
async with self._session.get(feed_url) as resp:
resp.raise_for_status()
content = await resp.text()
feed = feedparser.parse(content)
payloads = []
for entry in feed.entries[:self.max_items]:
payload = await self._parse_entry(feed_url, entry)
if payload:
yield payload
for entry in feed.entries[:self._max_items_per_feed]:
# Check if we've seen this item before
guid = entry.get("guid") or entry.get("id") or entry.get("link")
if guid in self._seen_guids:
continue
self._seen_guids.add(guid)
except Exception as e:
logger.error(f"Error fetching RSS feed {feed_url}: {e}")
self.stats["errors"] += 1
async def _parse_entry(self, feed_url: str, entry) -> Optional[NormalizedPayload]:
# Generate unique ID for deduplication
entry_id = getattr(entry, "id", getattr(entry, "link", ""))
content_hash = hashlib.md5(entry_id.encode()).hexdigest()[:16]
if content_hash in self._seen_ids:
return None
self._seen_ids.add(content_hash)
# Extract text content
title = getattr(entry, "title", "").strip()
summary = getattr(entry, "summary", getattr(entry, "description", "")).strip()
raw_text = f"{title}\n\n{clean_html(summary)}"
# Extract link
url = getattr(entry, "link", "")
if not url.startswith("http"):
url = urljoin(feed_url, url)
# Parse publish time
# Parse publish timestamp
publish_ts = None
for time_field in ("published_parsed", "updated_parsed"):
if hasattr(entry, time_field) and getattr(entry, time_field):
for date_field in ["published_parsed", "updated_parsed", "created_parsed"]:
if entry.get(date_field):
try:
publish_ts = datetime(*getattr(entry, time_field)[:6]).timestamp()
dt = datetime(*entry[date_field][:6])
publish_ts = dt.timestamp()
break
except Exception:
pass
# Author
author = getattr(entry, "author", getattr(entry, "authors", [{}])[0].get("name", "") if getattr(entry, "authors", None) else "")
# Extract text content
raw_text = entry.get("summary") or entry.get("description") or entry.get("content", [{}])[0].get("value", "")
title = entry.get("title", "")
# Extract tickers/assets from text
tickers = extract_tickers(raw_text)
asset_mentions = [
AssetMention(
asset_id=ticker,
mention_span=(0, len(ticker)),
confidence=0.8,
source_text=ticker,
mention_type="ticker"
)
for ticker in tickers
]
# Combine title and summary
full_text = f"{title}. {raw_text}" if title else raw_text
# Get source credibility
source_id = self._extract_source_id(feed_url)
credibility = self.credibility_registry.get(source_id, 0.5)
# Extract asset mentions (basic ticker extraction)
asset_mentions = self._extract_asset_mentions(full_text)
# Language detection
language = detect_language(raw_text)
return NormalizedPayload(
source_id=source_id,
source_type=SourceType.NEWS,
source_credibility_base=credibility,
ingest_ts=datetime.now().timestamp(),
payload = self._create_payload(
raw_text=full_text,
title=title,
url=entry.get("link"),
author=entry.get("author"),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
raw_text=raw_text,
title=title,
url=url,
author=author,
engagement_metrics=EngagementMetrics(),
content_length=len(raw_text),
language=language,
metadata={"feed_url": feed_url, "entry_id": entry_id}
metadata={"feed_url": feed_url, "guid": guid}
)
def _extract_source_id(self, feed_url: str) -> str:
from urllib.parse import urlparse
domain = urlparse(feed_url).netloc.replace("www.", "")
return f"rss:{domain}"
payloads.append(payload)
async def health_check(self) -> bool:
try:
session = await self._get_session()
async with session.get(self.feed_urls[0]) as response:
return response.status == 200
except Exception:
return False
logger.debug(f"RSS {self.config.source_id}: {len(payloads)} new items from {feed_url}")
return payloads
async def stop(self) -> None:
await super().stop()
if self._session and not self._session.closed:
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
"""Extract asset mentions from text (basic ticker extraction)"""
import re
mentions = []
# Match $TICKER or TICKER patterns
ticker_pattern = re.compile(r'\$?([A-Z]{2,10})\b')
for match in ticker_pattern.finditer(text):
ticker = match.group(1).upper()
# Filter common false positives
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=0.7,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def close(self) -> None:
"""Close HTTP session"""
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"RSSConnector {self.config.source_id} closed")

262
sentiment_engine/src/sentiment_engine/ingestion/twitter.py
View File

@@ -1,187 +1,143 @@
"""Twitter/X API v2 connector with streaming support"""
"""
Twitter/X Connector — polls Twitter API v2
"""
import asyncio
import hashlib
import logging
import re
from datetime import datetime
from typing import AsyncIterator, List, Optional, Dict, Any
import time
from typing import List, Optional
import tweepy
from tweepy.asynchronous import AsyncStreamingClient
import aiohttp
from sentiment_engine.schemas.payload import NormalizedPayload, SourceType, AssetMention, EngagementMetrics
from sentiment_engine.schemas.config import TwitterConnectorConfig
from sentiment_engine.ingestion.base import BaseConnector
from sentiment_engine.utils.text import clean_html, extract_tickers, extract_cashtags, detect_language
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class TwitterConnector(BaseConnector):
"""Twitter/X API v2 connector using tweepy-asynchronous for streaming"""
"""Twitter/X API v2 connector"""
def __init__(self, config: TwitterConnectorConfig, credibility_registry):
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self.credibility_registry = credibility_registry
self.client: Optional[tweepy.AsyncClient] = None
self.stream: Optional[AsyncStreamingClient] = None
self._stream_rules = config.stream_rules
self._sample_rate = config.sample_rate
self._running = False
self._bearer_token: str = config.extra_config.get("bearer_token", "")
self._search_query: str = config.extra_config.get("search_query", "crypto OR bitcoin OR ethereum OR defi OR web3")
self._max_results: int = config.extra_config.get("max_results", 100)
self._since_id: Optional[str] = None
async def initialize(self) -> None:
"""Initialize Twitter clients"""
self.client = tweepy.AsyncClient(
bearer_token=self.config.bearer_token,
consumer_key=self.config.api_key,
consumer_secret=self.config.api_secret,
access_token=self.config.access_token,
access_token_secret=self.config.access_secret,
wait_on_rate_limit=True
if not self._bearer_token:
raise ValueError("Twitter bearer_token required in extra_config")
self._session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self._bearer_token}"},
timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds)
)
self.status.running = True
logger.info(f"TwitterConnector {self.config.source_id} initialized")
# Verify credentials
me = await self.client.get_me()
logger.info(f"Twitter authenticated as @{me.data.username}")
async def poll(self) -> List[NormalizedPayload]:
"""Poll Twitter recent search endpoint"""
params = {
"query": self._search_query,
"max_results": min(self._max_results, 100),
"tweet.fields": "created_at,author_id,public_metrics,entities,context_annotations,lang",
"expansions": "author_id,referenced_tweets.id",
"user.fields": "username,verified,public_metrics,created_at",
}
if self._since_id:
params["since_id"] = self._since_id
# Setup streaming client
self.stream = AsyncStreamingClient(
bearer_token=self.config.bearer_token,
wait_on_rate_limit=True
)
url = "https://api.twitter.com/2/tweets/search/recent"
async def fetch(self) -> AsyncIterator[NormalizedPayload]:
"""Stream tweets matching rules"""
if not self.stream:
await self.initialize()
async with self._session.get(url, params=params) as resp:
if resp.status == 429:
# Rate limited
reset_time = int(resp.headers.get("x-rate-limit-reset", time.time() + 900))
wait = max(1, reset_time - time.time())
logger.warning(f"Twitter rate limited, waiting {wait}s")
await asyncio.sleep(wait)
return []
resp.raise_for_status()
data = await resp.json()
self._running = True
tweets = data.get("data", [])
users = {u["id"]: u for u in data.get("includes", {}).get("users", [])}
# Add stream rules
await self._setup_stream_rules()
payloads = []
for tweet in tweets:
if tweet.get("lang") != "en":
continue
# Define tweet processing
async def on_tweet(tweet):
if not self._running:
return
payload = await self._process_tweet(tweet)
if payload:
yield payload
self._since_id = max(self._since_id or "0", tweet["id"])
self.stream.on_tweet = on_tweet
user = users.get(tweet.get("author_id"), {})
# Start streaming
try:
await self.stream.filter(
tweet_fields=["created_at", "author_id", "public_metrics", "entities", "lang"],
expansions=["author_id", "referenced_tweets.id"],
user_fields=["username", "verified", "public_metrics"]
)
except asyncio.CancelledError:
logger.info("Twitter stream cancelled")
except Exception as e:
logger.error(f"Twitter stream error: {e}")
self.stats["errors"] += 1
async def _setup_stream_rules(self) -> None:
"""Configure stream filtering rules"""
# Clear existing rules
existing = await self.stream.get_rules()
if existing.data:
rule_ids = [rule.id for rule in existing.data]
await self.stream.delete_rules(rule_ids)
# Add new rules
for rule_text in self._stream_rules:
await self.stream.add_rules(tweepy.StreamRule(rule_text))
async def _process_tweet(self, tweet) -> Optional[NormalizedPayload]:
try:
# Skip retweets unless quote tweets
if tweet.referenced_tweets:
ref_types = [ref.type for ref in tweet.referenced_tweets]
if "retweeted" in ref_types and "quoted" not in ref_types:
return None
text = tweet.text
raw_text = clean_html(text)
# Extract tickers and cashtags
tickers = extract_tickers(raw_text)
cashtags = extract_cashtags(raw_text)
all_assets = list(set(tickers + cashtags))
asset_mentions = [
AssetMention(
asset_id=asset.lstrip("$"),
mention_span=(0, len(asset)),
confidence=0.9,
source_text=asset,
mention_type="cashtag" if asset.startswith("$") else "ticker"
)
for asset in all_assets
]
# Extract asset mentions from tweet
asset_mentions = self._extract_asset_mentions(tweet.get("text", ""))
# Engagement metrics
metrics = tweet.public_metrics or {}
metrics = tweet.get("public_metrics", {})
engagement = EngagementMetrics(
retweets=metrics.get("retweet_count", 0),
likes=metrics.get("like_count", 0),
replies=metrics.get("reply_count", 0),
views=metrics.get("impression_count", 0)
upvotes=0,
comments=metrics.get("reply_count", 0)
)
# Author info
author = ""
if hasattr(tweet, "author") and tweet.author:
author = tweet.author.username
# Publish time
publish_ts = tweet.created_at.timestamp() if tweet.created_at else None
source_id = "twitter:stream"
credibility = self.credibility_registry.get(source_id, 0.6)
# Language
language = tweet.lang or detect_language(raw_text)
return NormalizedPayload(
source_id=source_id,
source_type=SourceType.SOCIAL,
source_credibility_base=credibility,
ingest_ts=datetime.now().timestamp(),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
raw_text=raw_text,
title=None,
url=f"https://twitter.com/{author}/status/{tweet.id}" if author else None,
author=author,
engagement_metrics=engagement,
content_length=len(raw_text),
language=language,
metadata={
"tweet_id": str(tweet.id),
"conversation_id": str(tweet.conversation_id) if tweet.conversation_id else None,
"referenced_tweets": [{"type": r.type, "id": str(r.id)} for r in tweet.referenced_tweets] if tweet.referenced_tweets else []
}
)
except Exception as e:
logger.error(f"Error processing tweet: {e}")
return None
async def health_check(self) -> bool:
# Parse timestamp
publish_ts = None
if "created_at" in tweet:
try:
if self.client:
await self.client.get_me()
return True
from dateutil import parser as date_parser
publish_ts = date_parser.parse(tweet["created_at"]).timestamp()
except Exception:
pass
return False
async def stop(self) -> None:
self._running = False
if self.stream:
await self.stream.disconnect()
await super().stop()
payload = self._create_payload(
raw_text=tweet["text"],
title=f"@{user.get('username', 'unknown')}: {tweet['text'][:100]}",
url=f"https://twitter.com/{user.get('username', 'unknown')}/status/{tweet['id']}",
author=user.get("username"),
publish_ts=publish_ts,
asset_mentions=asset_mentions,
engagement_metrics=engagement,
metadata={
"tweet_id": tweet["id"],
"author_id": tweet.get("author_id"),
"verified": user.get("verified", False),
"followers": user.get("public_metrics", {}).get("followers_count", 0)
}
)
payloads.append(payload)
return payloads
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
import re
mentions = []
# Match $TICKER, #TICKER, or bare TICKER
patterns = [
r'\$([A-Z]{2,10})\b',
r'#([A-Z]{2,10})\b',
r'\b([A-Z]{3,10})\b' # Bare tickers (less confident)
]
for pattern in patterns:
for match in re.finditer(pattern, text):
ticker = match.group(1).upper()
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
confidence = 0.9 if pattern.startswith(r'\$') else (0.8 if pattern.startswith(r'#') else 0.5)
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=confidence,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def close(self) -> None:
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"TwitterConnector {self.config.source_id} closed")

321
sentiment_engine/src/sentiment_engine/ingestion/web_crawl.py
View File

@@ -1,224 +1,149 @@
"""Web crawler connector using Hister or Scrapy"""
"""
Web Crawl Connector — generic web crawling for custom sources
"""
import asyncio
import hashlib
import logging
import subprocess
import json
import tempfile
import os
from datetime import datetime
from pathlib import Path
from typing import AsyncIterator, List, Optional
import time
import re
from typing import List, Optional, Set
from urllib.parse import urljoin, urlparse
from sentiment_engine.schemas.payload import NormalizedPayload, SourceType, AssetMention, EngagementMetrics
from sentiment_engine.schemas.config import WebCrawlConnectorConfig
from sentiment_engine.ingestion.base import BaseConnector
from sentiment_engine.utils.text import clean_html, extract_tickers, detect_language
import aiohttp
from bs4 import BeautifulSoup
from sentiment_engine.ingestion.base import BaseConnector, ConnectorConfig
from sentiment_engine.schemas.payload import NormalizedPayload, AssetMention, EngagementMetrics
logger = logging.getLogger(__name__)
class WebCrawlConnector(BaseConnector):
"""Web crawler connector using Hister (primary) or Scrapy (fallback)"""
"""Generic web crawler for custom sources"""
def __init__(self, config: WebCrawlConnectorConfig, credibility_registry):
def __init__(self, config: ConnectorConfig):
super().__init__(config)
self.credibility_registry = credibility_registry
self.seed_urls = config.seed_urls
self.allowed_domains = config.allowed_domains
self.max_depth = config.max_depth
self.tool = config.tool
self.job_timeout = config.job_timeout_seconds
self.rate_limit_rps = config.rate_limit_rps
self._job_id = f"sentiment-crawl-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
self._seed_urls: List[str] = config.extra_config.get("seed_urls", [config.base_url])
self._allowed_domains: Set[str] = set(config.extra_config.get("allowed_domains", []))
self._max_depth: int = config.extra_config.get("max_depth", 2)
self._max_pages: int = config.extra_config.get("max_pages", 100)
self._rate_limit_rps: float = config.extra_config.get("rate_limit_rps", 1.0)
self._visited_urls: Set[str] = set()
self._url_queue: asyncio.Queue = asyncio.Queue()
self._rate_limiter: asyncio.Semaphore = asyncio.Semaphore(1)
self._last_request: float = 0
# Query timing windows
self.preferred_windows = config.preferred_query_windows or []
self.avoid_windows = config.avoid_query_windows or []
async def initialize(self) -> None:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config.timeout_seconds),
headers={"User-Agent": "DOLPHIN-SentimentEngine/2.0"}
)
# Seed the queue
for url in self._seed_urls:
await self._url_queue.put((url, 0))
self.status.running = True
logger.info(f"WebCrawlConnector {self.config.source_id} initialized with {len(self._seed_urls)} seed URLs")
def _in_preferred_window(self) -> bool:
if not self.preferred_windows:
return True
now = datetime.utcnow()
current_hour = now.hour
for window in self.preferred_windows:
start = window.get("start_hour", 0)
end = window.get("end_hour", 24)
if start <= end:
if start <= current_hour < end:
return True
else:
if current_hour >= start or current_hour < end:
return True
return False
async def poll(self) -> List[NormalizedPayload]:
"""Crawl and return payloads from discovered pages"""
payloads = []
pages_crawled = 0
def _in_avoid_window(self) -> bool:
if not self.avoid_windows:
return False
now = datetime.utcnow()
current_hour = now.hour
for window in self.avoid_windows:
start = window.get("start_hour", 0)
end = window.get("end_hour", 24)
if start <= end:
if start <= current_hour < end:
return True
else:
if current_hour >= start or current_hour < end:
return True
return False
async def fetch(self) -> AsyncIterator[NormalizedPayload]:
if self._in_avoid_window() or not self._in_preferred_window():
return
if self.tool == "hister":
async for payload in self._crawl_hister():
yield payload
elif self.tool == "scrapy":
async for payload in self._crawl_scrapy():
yield payload
else:
logger.error(f"Unknown crawl tool: {self.tool}")
async def _crawl_hister(self) -> AsyncIterator[NormalizedPayload]:
"""Crawl using Hister CLI"""
with tempfile.TemporaryDirectory() as tmpdir:
# Write seed URLs
seed_file = Path(tmpdir) / "seeds.txt"
seed_file.write_text("\n".join(self.seed_urls))
# Build hister command
cmd = [
"hister", "crawl",
"--input", str(seed_file),
"--job-id", self._job_id,
"--depth", str(self.max_depth),
"--delay", str(1.0 / self.rate_limit_rps),
"--output", str(Path(tmpdir) / "output.jsonl"),
"--format", "jsonl"
]
if self.allowed_domains:
cmd.extend(["--allowed-domain", ",".join(self.allowed_domains)])
# Run hister
while pages_crawled < self._max_pages and not self._url_queue.empty():
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(), timeout=self.job_timeout
)
if proc.returncode != 0:
logger.error(f"Hister failed: {stderr.decode()}")
self.stats["errors"] += 1
return
# Parse output
output_file = Path(tmpdir) / "output.jsonl"
if output_file.exists():
async for payload in self._parse_hister_output(output_file):
yield payload
url, depth = await asyncio.wait_for(self._url_queue.get(), timeout=5.0)
except asyncio.TimeoutError:
logger.error(f"Hister job timed out after {self.job_timeout}s")
proc.kill()
self.stats["errors"] += 1
except FileNotFoundError:
logger.error("Hister not installed, falling back to Scrapy")
async for payload in self._crawl_scrapy():
yield payload
break
if url in self._visited_urls or depth > self._max_depth:
continue
self._visited_urls.add(url)
try:
page_payloads = await self._crawl_page(url, depth)
payloads.extend(page_payloads)
pages_crawled += 1
except Exception as e:
logger.error(f"Hister crawl error: {e}")
self.stats["errors"] += 1
logger.error(f"Error crawling {url}: {e}")
async def _parse_hister_output(self, output_file: Path) -> AsyncIterator[NormalizedPayload]:
"""Parse Hister JSONL output"""
import aiofiles
# Rate limiting
await self._rate_limit()
async with aiofiles.open(output_file, "r") as f:
async for line in f:
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
payload = self._build_payload_from_hister(data)
if payload:
yield payload
except json.JSONDecodeError:
return payloads
async def _rate_limit(self) -> None:
"""Enforce rate limit"""
min_interval = 1.0 / self._rate_limit_rps
elapsed = time.time() - self._last_request
if elapsed < min_interval:
await asyncio.sleep(min_interval - elapsed)
self._last_request = time.time()
async def _crawl_page(self, url: str, depth: int) -> List[NormalizedPayload]:
async with self._rate_limiter:
async with self._session.get(url) as resp:
if resp.status != 200:
return []
content_type = resp.headers.get("Content-Type", "")
if "text/html" not in content_type:
return []
html = await resp.text()
soup = BeautifulSoup(html, "html.parser")
# Extract main content
for script in soup(["script", "style", "nav", "footer", "header"]):
script.decompose()
text = soup.get_text(separator=" ", strip=True)
title = soup.title.string if soup.title else ""
# Only process if substantial content
if len(text) < 200:
return []
# Extract asset mentions
asset_mentions = self._extract_asset_mentions(text)
# Enqueue links for deeper crawling
if depth < self._max_depth:
for link in soup.find_all("a", href=True):
href = link["href"]
absolute_url = urljoin(url, href)
parsed = urlparse(absolute_url)
if self._allowed_domains and parsed.netloc not in self._allowed_domains:
continue
if absolute_url not in self._visited_urls:
await self._url_queue.put((absolute_url, depth + 1))
def _build_payload_from_hister(self, data: dict) -> Optional[NormalizedPayload]:
try:
url = data.get("url", "")
title = data.get("title", "")
content = data.get("content", data.get("text", ""))
raw_text = f"{title}\n\n{clean_html(content)}"
if not raw_text.strip() or len(raw_text) < 100:
return None
tickers = extract_tickers(raw_text)
asset_mentions = [
AssetMention(asset_id=t, mention_span=(0, len(t)), confidence=0.6,
source_text=t, mention_type="ticker")
for t in tickers
]
source_id = f"web:{self._extract_domain(url)}"
credibility = self.credibility_registry.get(source_id, 0.3)
language = detect_language(raw_text)
return NormalizedPayload(
source_id=source_id,
source_type=SourceType.NEWS,
source_credibility_base=credibility,
ingest_ts=datetime.now().timestamp(),
publish_ts=None,
asset_mentions=asset_mentions,
raw_text=raw_text,
payload = self._create_payload(
raw_text=f"{title}. {text}",
title=title,
url=url,
author="",
engagement_metrics=EngagementMetrics(),
content_length=len(raw_text),
language=language,
metadata={"crawler": "hister", "job_id": self._job_id}
asset_mentions=asset_mentions,
metadata={"source_type": "web_crawl", "depth": depth}
)
except Exception as e:
logger.error(f"Error building payload from Hister data: {e}")
return None
async def _crawl_scrapy(self) -> AsyncIterator[NormalizedPayload]:
"""Fallback crawl using Scrapy"""
# Simplified Scrapy integration - would need a proper spider
logger.warning("Scrapy fallback not fully implemented")
return
yield # Make this an async generator
return [payload]
def _extract_domain(self, url: str) -> str:
from urllib.parse import urlparse
return urlparse(url).netloc.replace("www.", "")
def _extract_asset_mentions(self, text: str) -> List[AssetMention]:
mentions = []
pattern = re.compile(r'\$?([A-Z]{2,10})\b')
for match in pattern.finditer(text):
ticker = match.group(1).upper()
if ticker in {"THE", "AND", "FOR", "ARE", "BUT", "NOT", "YOU", "ALL", "CAN", "HER", "WAS", "ONE", "OUR", "OUT", "DAY", "GET", "HAS", "HIM", "HIS", "HOW", "ITS", "MAY", "NEW", "NOW", "OLD", "SEE", "TWO", "WHO", "BOY", "DID", "MAN", "PUT", "SAY", "SHE", "TOO", "USE", "CEO", "CTO", "CFO", "COO", "IPO", "API", "SDK", "UI", "UX", "AI", "ML", "DL", "RL", "GPT", "LLM", "BERT", "USA", "UK", "EU", "UN", "NASA", "FBI", "CIA", "IRS", "SEC", "CFTC", "FED", "GDP", "CPI", "PCE", "FOMC", "YOY", "QOQ", "EPS", "PE", "ROI", "ROE"}:
continue
mentions.append(AssetMention(
asset_id=ticker,
mention_span=(match.start(), match.end()),
confidence=0.6,
source_text=match.group(),
mention_type="ticker"
))
return mentions
async def health_check(self) -> bool:
# Check if hister is available
try:
proc = await asyncio.create_subprocess_exec(
"hister", "--version",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
await proc.communicate()
return proc.returncode == 0
except FileNotFoundError:
return False
async def stop(self) -> None:
await super().stop()
async def close(self) -> None:
if self._session:
await self._session.close()
self.status.running = False
logger.info(f"WebCrawlConnector {self.config.source_id} closed")