k
54884f3e24
Feed-health pass over KOL_FEEDS: - raoulpal: stale Substack (last 2024-05) → Real Vision podcast feed - dampedspring: paywalled (0 entries) → free "Damped Spring 101" Substack - unchained: Cloudflare 403 → canonical Megaphone podcast feed - lynalden: Cloudflare 202 → FeedBurner mirror - glassnode: recovered via httpx http2=True (was 403 on HTTP/1.1) - browser User-Agent + Accept headers on feed fetch - removed dead feeds with no active replacement: placeholder, dragonfly, niccarter, eugene - pin h2==4.3.0 (required by http2=True) All 25 remaining feeds verified fetching real body content; newest post per feed ≤88d. Bundles in-flight KOL-module work already in the working tree (kol_x ingest, migration 027, tests). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
285 lines
11 KiB
Python
285 lines
11 KiB
Python
"""KOL X (Twitter) ingester.
|
||
|
||
Polls each tracked KOL's recent tweets via twitterapi.io, dedupes by tweet id,
|
||
runs `x_analysis.analyze_x_post`, and writes a `KolPost(source="twitter")`.
|
||
|
||
The `tickers_json` it stores uses the exact `{ticker, action, conviction}` shape
|
||
that `kol_divergence` already consumes (it reads `t["ticker"]` / `t["action"]` /
|
||
`t["conviction"]`) — so no mapping layer is needed. x_analysis was designed
|
||
against that contract; this module just connects the pipe.
|
||
|
||
WHY THIS EXISTS
|
||
`andrewkang` (@Rewkang) and `murad` publish ONLY on X — no Substack feed.
|
||
Their on-chain wallets are already seeded (seed_kol_wallets.py), but with no
|
||
post-side data the divergence scanner detects nothing for them — the wallets
|
||
sit dark. This ingester is the missing post-side feed that lights them up.
|
||
|
||
DESIGN (mirrors kol_substack.py)
|
||
- Disabled (full no-op) when `settings.twitterapi_io_key` is empty.
|
||
- `KolPost.kol_handle` is the CANONICAL handle (e.g. "cryptohayes"), NOT the
|
||
X screen name — it must match `KolWallet.handle` so divergence can join
|
||
post-side ↔ on-chain for the same person.
|
||
- Dedup by (source="twitter", external_id=<tweet id>).
|
||
- Bare retweets ("RT @…") are skipped before the AI call to save spend.
|
||
- Only the first page (~20 newest tweets) is fetched per run. Daily volume
|
||
for these KOLs is < 20/day, and dedup makes re-runs cheap. Bump
|
||
`max_pages` if you add a very high-frequency account.
|
||
|
||
COST
|
||
~$0.15 / 1k tweets (twitterapi.io). 4 KOLs × ~20 tweets daily ≈ $0.015/mo.
|
||
|
||
CADENCE
|
||
Daily 01:30 UTC — after substack (01:15), before on-chain (02:00) and
|
||
divergence (02:15), so fresh X posts are present when divergence runs.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
from email.utils import parsedate_to_datetime
|
||
from typing import AsyncGenerator, Optional
|
||
|
||
import httpx
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from app.database import AsyncSessionLocal
|
||
from app.models import KolPost, utcnow
|
||
from app.services import x_analysis
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
X_API = "https://api.twitterapi.io/twitter/user/last_tweets"
|
||
|
||
# Tracked X-native KOLs. `handle` MUST match the canonical handle used in
|
||
# KolWallet / KOL_FEEDS so divergence can join against on-chain data.
|
||
# `x_username` is the actual X screen name (no @).
|
||
#
|
||
# cryptohayes is also on Substack (long-form) — X adds his real-time position
|
||
# statements ("just dumped my $HYPE"), which Substack essays don't carry.
|
||
# andrewkang + murad are X-ONLY and have seeded wallets waiting for this feed.
|
||
X_KOLS: list[dict] = [
|
||
{"handle": "cryptohayes", "x_username": "CryptoHayes", "display_name": "Arthur Hayes"},
|
||
{"handle": "andrewkang", "x_username": "Rewkang", "display_name": "Andrew Kang"},
|
||
{"handle": "murad", "x_username": "MustStopMurad", "display_name": "Murad Mahmudov"},
|
||
# Raoul Pal — Real Vision / GMI founder. Macro/liquidity/cycle thinker
|
||
# (mostly DIRECTIONAL, rarely position TRADE_SIGNAL). Same canonical handle
|
||
# as his Substack feed below so his X takes + long-form theses aggregate.
|
||
{"handle": "raoulpal", "x_username": "RaoulGMI", "display_name": "Raoul Pal"},
|
||
]
|
||
|
||
|
||
def _parse_created_at(raw: Optional[str]) -> datetime:
|
||
"""X `createdAt` looks like 'Thu Jun 04 12:29:13 +0000 2026'. Normalise to
|
||
naive UTC (same convention as kol_substack._parse_pub — divergence window
|
||
matching and digest 'since' filters all assume naive UTC)."""
|
||
if not raw:
|
||
return utcnow()
|
||
try:
|
||
dt = parsedate_to_datetime(raw)
|
||
if dt.tzinfo:
|
||
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
|
||
return dt
|
||
except Exception:
|
||
return utcnow()
|
||
|
||
|
||
async def _fetch_tweet_page(
|
||
client: httpx.AsyncClient,
|
||
x_username: str,
|
||
cursor: Optional[str],
|
||
key: str,
|
||
) -> tuple[list[dict], Optional[str]]:
|
||
"""Fetch one page of tweets. Returns (tweets, next_cursor).
|
||
next_cursor is None when there are no more pages or on any error."""
|
||
params: dict = {"userName": x_username, "includeReplies": "false"}
|
||
if cursor:
|
||
params["cursor"] = cursor
|
||
try:
|
||
r = await client.get(X_API, params=params, headers={"x-api-key": key})
|
||
if r.status_code != 200:
|
||
logger.warning("[kol_x] fetch %s HTTP %d: %s",
|
||
x_username, r.status_code, r.text[:200])
|
||
return [], None
|
||
data = r.json()
|
||
# Shape: {status, data: {tweets: [...], next_cursor?: str}, ...}
|
||
if data.get("status") != "success":
|
||
logger.warning("[kol_x] fetch %s non-success: %s",
|
||
x_username, str(data.get("msg") or data)[:200])
|
||
return [], None
|
||
inner = data.get("data") or {}
|
||
tweets = inner.get("tweets") or []
|
||
next_cursor = inner.get("next_cursor") or None
|
||
return tweets, next_cursor
|
||
except Exception as e:
|
||
logger.warning("[kol_x] fetch %s exception: %s", x_username, e)
|
||
return [], None
|
||
|
||
|
||
async def _iter_tweet_pages(
|
||
x_username: str,
|
||
max_pages: int = 3,
|
||
) -> AsyncGenerator[list[dict], None]:
|
||
"""Async generator: yields one page of tweets at a time.
|
||
|
||
Callers can break out of the loop early (page-level early-stop) without
|
||
fetching unnecessary pages. max_pages=3 is the gap-fill ceiling — normal
|
||
daily runs stop after page 1 because _ingest_kol_x breaks on all-dedup.
|
||
|
||
Never raises — a network error yields nothing and ends the iteration.
|
||
"""
|
||
key = settings.twitterapi_io_key
|
||
if not key:
|
||
return
|
||
cursor: Optional[str] = None
|
||
async with httpx.AsyncClient(timeout=20.0) as client:
|
||
for _ in range(max_pages):
|
||
tweets, cursor = await _fetch_tweet_page(client, x_username, cursor, key)
|
||
if tweets:
|
||
yield tweets
|
||
if not cursor or not tweets:
|
||
break
|
||
|
||
|
||
async def _ingest_kol_x(
|
||
session: AsyncSession,
|
||
kol: dict,
|
||
*,
|
||
analyze: bool = True,
|
||
max_new: int = 20,
|
||
max_pages: int = 3,
|
||
) -> dict:
|
||
"""Ingest one KOL's recent tweets. Mirrors kol_substack._ingest_kol.
|
||
|
||
Pages are fetched one at a time. If an entire page consists only of
|
||
already-stored tweets (dedup hits), fetching stops — subsequent pages
|
||
would be even older and equally known, so the API call is skipped.
|
||
This keeps the normal daily run to a single page fetch.
|
||
"""
|
||
handle = kol["handle"]
|
||
x_username = kol["x_username"]
|
||
stats = {"handle": handle, "x_username": x_username,
|
||
"new": 0, "skipped": 0, "analyzed": 0, "errors": 0}
|
||
|
||
async for page_tweets in _iter_tweet_pages(x_username, max_pages=max_pages):
|
||
if stats["new"] >= max_new:
|
||
break
|
||
|
||
page_new = 0 # new tweets stored on this page
|
||
page_deduped = 0 # already-stored tweets hit on this page (dedup)
|
||
|
||
for tw in page_tweets:
|
||
if stats["new"] >= max_new:
|
||
break
|
||
|
||
tweet_id = str(tw.get("id") or "").strip()
|
||
text = (tw.get("text") or "").strip()
|
||
if not tweet_id or not text:
|
||
continue
|
||
|
||
# Bare retweet → noise. Skip before the AI call to save spend.
|
||
if text.startswith("RT @"):
|
||
stats["skipped"] += 1
|
||
continue
|
||
|
||
# Dedup by (source, external_id).
|
||
existing = await session.execute(
|
||
select(KolPost).where(
|
||
KolPost.source == "twitter",
|
||
KolPost.external_id == tweet_id,
|
||
)
|
||
)
|
||
if existing.scalar_one_or_none() is not None:
|
||
stats["skipped"] += 1
|
||
page_deduped += 1
|
||
continue
|
||
|
||
author = tw.get("author") or {}
|
||
follower_count = author.get("followers")
|
||
url = tw.get("url") or f"https://x.com/{x_username}/status/{tweet_id}"
|
||
pub = _parse_created_at(tw.get("createdAt"))
|
||
body_hash = hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||
|
||
row = KolPost(
|
||
kol_handle=handle,
|
||
source="twitter",
|
||
external_id=tweet_id,
|
||
url=url,
|
||
title=None,
|
||
published_at=pub,
|
||
raw_text=text,
|
||
content_hash=body_hash,
|
||
)
|
||
session.add(row)
|
||
await session.flush()
|
||
stats["new"] += 1
|
||
page_new += 1
|
||
logger.info("[kol_x] new tweet %s id=%s tweet_id=%s", handle, row.id, tweet_id)
|
||
|
||
if analyze:
|
||
try:
|
||
result = await x_analysis.analyze_x_post(
|
||
handle=handle,
|
||
text=text,
|
||
posted_at=pub.isoformat(),
|
||
follower_count=follower_count,
|
||
)
|
||
if result.get("error"):
|
||
stats["errors"] += 1
|
||
else:
|
||
row.summary = result.get("summary")
|
||
row.tickers_json = json.dumps(result.get("tickers") or [],
|
||
ensure_ascii=False)
|
||
row.analyzed_at = utcnow()
|
||
row.analysis_model = result.get("model")
|
||
row.analysis_version = result.get("version")
|
||
# Extended x_analysis fields (migration 027)
|
||
row.tier = result.get("tier")
|
||
row.post_type = result.get("post_type")
|
||
row.talks_vs_trades_flag = bool(result.get("talks_vs_trades_flag", False))
|
||
row.sentiment = result.get("sentiment")
|
||
stats["analyzed"] += 1
|
||
except Exception as e:
|
||
logger.warning("[kol_x] analysis failed %s tweet %s: %s",
|
||
handle, row.id, e)
|
||
stats["errors"] += 1
|
||
|
||
# Page-level early stop: this page yielded NO new tweets AND hit at
|
||
# least one dedup → we've caught up to already-stored data, so older
|
||
# pages are known too. A page that is ALL bare-RTs (page_deduped == 0)
|
||
# must NOT early-stop — the next page may still hold original posts.
|
||
if page_new == 0 and page_deduped > 0:
|
||
logger.debug("[kol_x] %s page fully deduped — stopping early", handle)
|
||
break
|
||
|
||
await session.commit()
|
||
return stats
|
||
|
||
|
||
async def run_x_poll(*, analyze: bool = True) -> list[dict]:
|
||
"""Poll every tracked X KOL once. Full no-op (returns []) when the
|
||
twitterapi.io key is unset. Returns per-KOL stats.
|
||
|
||
Each KOL gets its own session so a commit failure for one does not leave
|
||
a dirty session that breaks subsequent KOLs in the same run.
|
||
"""
|
||
if not settings.twitterapi_io_key:
|
||
logger.info("[kol_x] twitterapi_io_key not set — X ingestion disabled.")
|
||
return []
|
||
results = []
|
||
for kol in X_KOLS:
|
||
try:
|
||
async with AsyncSessionLocal() as session:
|
||
stats = await _ingest_kol_x(session, kol, analyze=analyze)
|
||
results.append(stats)
|
||
except Exception as e:
|
||
# One KOL's DB/commit failure must not abort the rest of the run.
|
||
logger.error("[kol_x] ingest failed for %s: %s", kol.get("handle"), e)
|
||
results.append({"handle": kol.get("handle"), "error": str(e)})
|
||
logger.info("[kol_x] poll done: %s", results)
|
||
return results
|