k
041f010a30
kol-v2 added the 'reduce' action (KOL partially trimming a long), but kol_divergence._POST_LONG/_POST_SHORT predate it, so any post whose ticker action was 'reduce' was skipped by the 'not in (_POST_LONG|_POST_SHORT)' guard — the divergence scan never saw it. That dropped a high-value case: a KOL publicly trimming/taking profit on an asset while their on-chain wallet is ADDING is exactly the talks-vs-trades mismatch this module exists to surface. 'reduce' is a short-leaning stance, so it now joins _POST_SHORT: reduce + chain adding → divergence (long) reduce + chain reducing → alignment (short) 72 tests pass. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""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
|