"""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, reduce} = SHORT intent {mention} = skip (no clear view) (reduce = partial profit-take / trimming a long → a risk-reducing, short-leaning stance. Counting it lets us catch "publicly trimming BTC but wallet is adding" divergence.) 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') # 'reduce' (kol-v2: partial profit-take / trimming a long) is a short-leaning # stance — include it so a "trimming publicly while accumulating on-chain" # mismatch is caught instead of silently dropped. _POST_LONG = {"buy", "bullish"} _POST_SHORT = {"sell", "bearish", "reduce"} # 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