consensus.py

"""
Multi-account consensus (v0.7.0).
When 2+ Tier 1 accounts tweet same asset/direction within time window → amplified alert.
"""

import logging
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Optional, Set, Tuple

logger = logging.getLogger(__name__)


@dataclass
class SignalRecord:
    asset: str
    direction: str
    handle: str
    tier: int
    ts: datetime


class ConsensusTracker:
    """Tracks recent signals per (asset, direction). Returns event dict when 2+ Tier1 align."""

    def __init__(self, window_hours: int = 4, min_tier1: int = 2):
        self.window_minutes = window_hours * 60
        self.min_tier1 = min_tier1
        self._records: Dict[Tuple[str, str], List[SignalRecord]] = defaultdict(list)

    def add(
        self,
        asset: str,
        direction: str,
        handle: str,
        tier: int,
        ts: str = None,
    ) -> Optional[dict]:
        """
        Add signal. Returns event dict when consensus reached (amplified), else None.
        ts: ISO timestamp string from tweet.
        """
        key = (asset, direction)
        now = datetime.now(timezone.utc)
        if ts:
            try:
                t = datetime.fromisoformat(ts.replace("Z", "+00:00"))
                if t.tzinfo is None:
                    t = t.replace(tzinfo=timezone.utc)
                now = t
            except Exception:
                pass
        cutoff = now - timedelta(minutes=self.window_minutes)
        self._records[key] = [r for r in self._records[key] if r.ts > cutoff]
        self._records[key].append(
            SignalRecord(asset=asset, direction=direction, handle=handle, tier=tier, ts=now)
        )
        tier1_handles = {r.handle for r in self._records[key] if r.tier == 1}
        if len(tier1_handles) >= self.min_tier1:
            return {"asset": asset, "direction": direction, "handles": list(tier1_handles)}
        return None


def extract_keywords(text: str, direction: str) -> List[str]:
    """Extract bullish/bearish keywords for display."""
    text_lower = text.lower()
    bull = ["accumulat", "load the boat", "buy", "long", "bullish", "dip", "bottom", "nfa", "not financial advice"]
    bear = ["distribution", "reducing", "careful", "sell", "short", "bearish", "top", "resistance"]
    pool = bull if direction == "BUY" else bear
    found = [w for w in pool if w in text_lower]
    return found[:8]