trumpsignal-backend/app/services/kol_divergence.py

"""KOL talks-vs-trades cross-signal detector.

Compares B-tier content signals (Substack / Twitter post → ticker action)
against A-tier on-chain changes (wallet holding changes) for the same
KOL handle + ticker within a ±N-day window.

Two outcomes:
  divergence — KOL says bullish but on-chain is selling (or vice versa).
               On-chain action is the ground truth; word is noise/manipulation.
               Conclusion = opposite of what was said.

  alignment  — Post and chain agree. Conviction is reinforced.
               Conclusion = what was said (and done).

Logic:

  post side  → action ∈ {buy, bullish} = LONG intent
                        {sell, bearish} = SHORT intent
                        {mention}       = skip (no clear view)

  chain side → change_type ∈ {new_position, increased} = LONG action
                              {decreased, closed}        = SHORT action

  LONG intent  + SHORT action → divergence, direction='short'
  SHORT intent + LONG  action → divergence, direction='long'
  LONG  intent + LONG  action → alignment,  direction='long'
  SHORT intent + SHORT action → alignment,  direction='short'

Run cadence: after each onchain poll (02:05 UTC). Also callable manually.
Idempotent: UniqueConstraint(post_id, change_id) prevents double-writes.
"""

from __future__ import annotations

import hashlib
import json
import logging
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from typing import Optional

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import AsyncSessionLocal
from app.models import KolDivergence, KolHoldingChange, KolPost, KolWallet, Post, utcnow

logger = logging.getLogger(__name__)

# Match window: post and chain event must be within this many days of each other
WINDOW_DAYS = 7

# Only surface on-chain changes above this threshold (avoid noise from dust)
MIN_USD_CHANGE = 10_000

# Post actions that map to a directional view (skip 'mention')
_POST_LONG  = {"buy", "bullish"}
_POST_SHORT = {"sell", "bearish"}

# On-chain actions that map to a direction
_CHAIN_LONG  = {"new_position", "increased"}
_CHAIN_SHORT = {"decreased", "closed"}

# ── Telegram alert gating ────────────────────────────────────────────────────
# Only push divergences (the surprising "they say one thing, do another" case).
# Alignments are confirmations, not unique alpha — skip to keep volume sane.
ALERT_ON_TYPES = {"divergence"}
# Floor on the post's stated conviction. Without this, every "mention"-like
# soft view that happens to coincide with a buy/sell would page users.
ALERT_MIN_POST_CONVICTION = 0.5
# Minimum USD on the on-chain side to bother alerting. The 10K floor in
# MIN_USD_CHANGE keeps dust out of the table; this stricter floor avoids
# alerting on small rebalances by big wallets.
ALERT_MIN_USD = 50_000


def _post_for_divergence(handle: str, ticker: str, direction: str,
                         post_action: str, chain_action: str,
                         conviction: float, change_id: int,
                         usd_after: Optional[float],
                         days_apart: float) -> Post:
    """Build a Post row carrying a KOL divergence as an actionable signal.

    `direction` is the CONCLUSION direction from _classify: 'long' → buy,
    'short' → short. ai_confidence is derived from post conviction so the
    user's min_confidence floor in TelegramBinding gates noise.
    """
    signal = "buy" if direction == "long" else "short"
    usd_str = f"${(usd_after or 0)/1000:.0f}K" if usd_after else "?"
    text = (
        f"KOL {handle} says {post_action.upper()} {ticker} — "
        f"but on-chain shows {chain_action} ({usd_str}, Δ{days_apart:.1f}d). "
        f"Following the chain (the trade, not the talk): {signal.upper()} {ticker}."
    )
    # external_id must be unique-per-source. Use the change_id as the dedupe
    # key — at most one Post per (kol, ticker, chain-event).
    ext = hashlib.md5(f"kol_divergence:{handle}:{ticker}:{change_id}".encode()).hexdigest()
    return Post(
        external_id=ext,
        text=text,
        source="kol_divergence",
        published_at=datetime.now(timezone.utc).replace(tzinfo=None),
        sentiment="bullish" if signal == "buy" else "bearish",
        ai_confidence=int(round(conviction * 100)),
        relevant=True,
        signal=signal,
        target_asset=ticker,
        category=f"kol_divergence_{signal}",
        analysis_version="kol_divergence_v1",
        prefilter_reason="kol_divergence",
    )


def _classify(post_action: str, chain_action: str) -> Optional[tuple[str, str]]:
    """Returns (signal_type, direction) or None if no meaningful pair."""
    if post_action in _POST_LONG:
        if chain_action in _CHAIN_LONG:
            return "alignment", "long"
        if chain_action in _CHAIN_SHORT:
            return "divergence", "short"
    elif post_action in _POST_SHORT:
        if chain_action in _CHAIN_LONG:
            return "divergence", "long"
        if chain_action in _CHAIN_SHORT:
            return "alignment", "short"
    return None


async def run_divergence_scan(
    lookback_days: int = 30,
) -> list[dict]:
    """Scan recent posts × on-chain changes, write new KolDivergence rows.

    Returns list of newly written rows as dicts (for logging / API return).
    Already-stored pairs are silently skipped (idempotent).
    """
    since = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=lookback_days)
    results: list[dict] = []
    # Posts we created this scan — notify_signal runs AFTER commit so the
    # dispatcher's separate DB session can read the row.
    alert_post_ids: list[int] = []

    async with AsyncSessionLocal() as session:
        # ── 1. Load recent posts that have directional ticker signals ──────
        posts = (await session.execute(
            select(KolPost)
            .where(KolPost.published_at >= since)
            .where(KolPost.tickers_json.is_not(None))
        )).scalars().all()

        # ── 2. Load recent on-chain changes (keyed by handle via wallet) ───
        changes_rows = (await session.execute(
            select(KolHoldingChange, KolWallet.handle)
            .join(KolWallet, KolHoldingChange.wallet_id == KolWallet.id)
            .where(KolHoldingChange.detected_at >= since)
        )).all()

        # Pre-load all existing (post_id, change_id) pairs so we can skip
        # duplicates client-side instead of relying on a UniqueConstraint
        # violation. Catching the violation would force a session rollback
        # that wipes ALL pending writes (not just the offending one) — a
        # subtle data-loss bug. Pre-check is cheap (the table stays small).
        existing_pairs: set[tuple[int, int]] = {
            (pid, cid)
            for (pid, cid) in (await session.execute(
                select(KolDivergence.post_id, KolDivergence.change_id)
            )).all()
        }

        # Index changes by (handle, ticker) → list[KolHoldingChange]
        chain_index: dict[tuple[str, str], list[tuple[KolHoldingChange, str]]] = defaultdict(list)
        for change, handle in changes_rows:
            # Skip dust moves
            usd_delta = abs((change.usd_after or 0) - (change.usd_before or 0))
            if usd_delta < MIN_USD_CHANGE and (change.usd_after or 0) < MIN_USD_CHANGE:
                continue
            chain_index[(handle, change.ticker.upper())].append((change, handle))

        # ── 3. Match posts → changes ───────────────────────────────────────
        for post in posts:
            try:
                tickers = json.loads(post.tickers_json or "[]")
            except json.JSONDecodeError:
                continue

            for t in tickers:
                post_action = (t.get("action") or "").lower()
                if post_action not in (_POST_LONG | _POST_SHORT):
                    continue  # skip 'mention'

                ticker = (t.get("ticker") or "").upper()
                if not ticker:
                    continue

                candidates = chain_index.get((post.kol_handle, ticker), [])
                for change, handle in candidates:
                    # Time-window check
                    post_dt    = post.published_at
                    change_dt  = change.detected_at
                    days_apart = abs((change_dt - post_dt).total_seconds()) / 86400
                    if days_apart > WINDOW_DAYS:
                        continue

                    result = _classify(post_action, change.change_type)
                    if result is None:
                        continue
                    signal_type, direction = result

                    # Idempotency: skip if (post_id, change_id) already stored
                    pair_key = (post.id, change.id)
                    if pair_key in existing_pairs:
                        continue
                    existing_pairs.add(pair_key)

                    conviction = float(t.get("conviction") or 0)
                    row = KolDivergence(
                        handle          = post.kol_handle,
                        ticker          = ticker,
                        post_id         = post.id,
                        post_action     = post_action,
                        post_conviction = conviction,
                        post_at         = post_dt,
                        change_id       = change.id,
                        onchain_action  = change.change_type,
                        usd_before      = change.usd_before,
                        usd_after       = change.usd_after,
                        onchain_at      = change_dt,
                        signal_type     = signal_type,
                        direction       = direction,
                        days_apart      = round(days_apart, 2),
                    )
                    session.add(row)

                    # Emit a Post for Telegram fan-out — but only for the
                    # interesting case (divergences) with enough conviction
                    # and chain-size to be worth a push. See ALERT_* tunables.
                    if (signal_type in ALERT_ON_TYPES
                            and conviction >= ALERT_MIN_POST_CONVICTION
                            and (change.usd_after or 0) >= ALERT_MIN_USD):
                        alert_post = _post_for_divergence(
                            handle=post.kol_handle, ticker=ticker,
                            direction=direction, post_action=post_action,
                            chain_action=change.change_type,
                            conviction=conviction, change_id=change.id,
                            usd_after=change.usd_after, days_apart=days_apart,
                        )
                        session.add(alert_post)
                        await session.flush()  # populate alert_post.id
                        alert_post_ids.append(alert_post.id)
                    info = {
                        "handle": post.kol_handle,
                        "ticker": ticker,
                        "signal_type": signal_type,
                        "direction": direction,
                        "post_action": post_action,
                        "onchain_action": change.change_type,
                        "days_apart": round(days_apart, 2),
                        "usd_after": change.usd_after,
                    }
                    results.append(info)
                    emoji = "⚠️" if signal_type == "divergence" else "✅"
                    logger.info(
                        "[kol_divergence] %s %s %s: says %s, onchain %s → %s (%s) Δ%.1fd",
                        emoji, post.kol_handle, ticker,
                        post_action, change.change_type,
                        signal_type, direction, days_apart,
                    )

        await session.commit()

    # Fire Telegram fan-out AFTER commit so _dispatch's own session can
    # actually see the rows. _dispatch only needs the post_id, so we skip
    # notify_signal's Post-object wrapper and schedule it directly.
    if alert_post_ids:
        try:
            import asyncio
            from app.services.telegram import _dispatch
            for pid in alert_post_ids:
                asyncio.create_task(_dispatch(pid))
        except Exception as exc:
            logger.warning("[kol_divergence] notify_signal failed: %s", exc)

    logger.info("[kol_divergence] scan done: %d new pairs written, %d alerts queued",
                len(results), len(alert_post_ids))
    return results