k
353 lines
14 KiB
Python
353 lines
14 KiB
Python
"""
|
|
Trump Truth Social scraper — CNN public archive
|
|
Source: https://ix.cnn.io/data/truth-social/truth_archive.json
|
|
Updated every ~5 minutes by CNN.
|
|
"""
|
|
|
|
import hashlib
|
|
import html
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import Post, iso_utc
|
|
from app.services.analysis import analyze_post
|
|
from app.services.price_store import price_store
|
|
from app.ws.manager import manager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ARCHIVE_URL = "https://ix.cnn.io/data/truth-social/truth_archive.json"
|
|
|
|
# Liveness tracker — updated every successful poll (even when no new posts).
|
|
# Read by /api/health to detect a dead scraper. None = never ran yet.
|
|
last_successful_poll_at: Optional[datetime] = None
|
|
last_poll_error: Optional[str] = None
|
|
|
|
# Conditional-GET validators from the last 200 response. The archive is 30k+
|
|
# posts (~MBs of JSON); CNN only regenerates it every ~5 min, so most polls
|
|
# can be answered with a 304 and skip download + parse entirely.
|
|
_etag: Optional[str] = None
|
|
_last_modified: Optional[str] = None
|
|
|
|
# Sentinel returned by _fetch_archive on HTTP 304 — distinct from None
|
|
# (fetch failure) so the poller can count it as a successful, no-work cycle.
|
|
NOT_MODIFIED = object()
|
|
|
|
|
|
def _strip_html(text: str) -> str:
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
text = html.unescape(text)
|
|
return re.sub(r"\s+", " ", text).strip()
|
|
|
|
|
|
def _parse_dt(iso: str) -> datetime:
|
|
"""Parse Truth Social's ISO timestamp into a naive-UTC datetime.
|
|
|
|
IMPORTANT: must convert to UTC *before* stripping tzinfo. Otherwise an
|
|
input like '2026-04-24T15:07:48-04:00' would be stored as naive
|
|
15:07:48 and later mis-read as UTC — a silent 4-hour shift.
|
|
"""
|
|
try:
|
|
dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
|
|
if dt.tzinfo is not None:
|
|
dt = dt.astimezone(timezone.utc)
|
|
return dt.replace(tzinfo=None)
|
|
except Exception:
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
async def _fetch_archive(conditional: bool = True):
|
|
"""Fetch the archive JSON. Returns a list, NOT_MODIFIED (304), or None
|
|
on error. `conditional=False` (backfill) always downloads the full body
|
|
so a poller-set ETag can't starve the startup backfill."""
|
|
global _etag, _last_modified
|
|
headers = {
|
|
"User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)",
|
|
"Accept": "application/json",
|
|
}
|
|
if conditional:
|
|
if _etag:
|
|
headers["If-None-Match"] = _etag
|
|
if _last_modified:
|
|
headers["If-Modified-Since"] = _last_modified
|
|
try:
|
|
from app.services.http_client import get_client
|
|
resp = await get_client().get(ARCHIVE_URL, headers=headers, timeout=30)
|
|
if conditional and resp.status_code == 304:
|
|
return NOT_MODIFIED
|
|
resp.raise_for_status()
|
|
_etag = resp.headers.get("etag")
|
|
_last_modified = resp.headers.get("last-modified")
|
|
return resp.json()
|
|
except Exception as exc:
|
|
# Include type name — httpx often raises bare ConnectError/TimeoutException
|
|
# with empty .args, which used to log as just "Failed to fetch CNN archive:"
|
|
# with no body, making outages impossible to diagnose.
|
|
logger.error("Failed to fetch CNN archive: %s (%s)",
|
|
type(exc).__name__, exc)
|
|
return None
|
|
|
|
|
|
async def _known_external_ids(entries: list, db: AsyncSession) -> set:
|
|
"""One batch query for the dedup pass. On a typical poll all ~50 entries
|
|
already exist, so this replaces 50 per-entry SELECTs with 1."""
|
|
ids = [hashlib.md5(str(e["id"]).encode()).hexdigest() for e in entries]
|
|
if not ids:
|
|
return set()
|
|
rows = await db.execute(
|
|
select(Post.external_id).where(Post.external_id.in_(ids)))
|
|
return {r[0] for r in rows}
|
|
|
|
|
|
async def _process_entry(entry: dict, db: AsyncSession,
|
|
known_ids: Optional[set] = None) -> Optional[Post]:
|
|
external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()
|
|
|
|
# Fast path: pre-fetched batch dedup set. New (unseen) ids still get the
|
|
# confirming SELECT below — protects against a concurrent insert by the
|
|
# other poller between the batch query and this entry.
|
|
if known_ids is not None and external_id in known_ids:
|
|
return None
|
|
|
|
result = await db.execute(select(Post).where(Post.external_id == external_id))
|
|
if result.scalar_one_or_none():
|
|
return None
|
|
|
|
text = _strip_html(entry.get("content") or "").strip()
|
|
if not text:
|
|
return None
|
|
|
|
published_at = _parse_dt(entry.get("created_at", ""))
|
|
|
|
# ── Deterministic entry pre-filter (saves AI spend + blocks 80% of noise) ──
|
|
# The 13-trade backtest showed AI confidence ≥ 85 still lets through pure
|
|
# rhetoric and second-derivative news. Hard-coded action-marker + future-
|
|
# tense + dedup check rejects those BEFORE the AI call. Failing posts are
|
|
# still saved to DB (so we have a record) but stamped as non-actionable.
|
|
from app.database import AsyncSessionLocal
|
|
from app.services.entry_filter import passes_entry_filter
|
|
filter_ok, filter_reason = await passes_entry_filter(text, AsyncSessionLocal)
|
|
if not filter_ok:
|
|
logger.info("Entry filter rejected post (id=%s): %s — %s",
|
|
entry.get("id"), filter_reason, text[:80])
|
|
# Insert a stub row so we don't keep re-fetching the same post from
|
|
# the upstream archive. Signal=hold, no AI call, no analysis.
|
|
stub = Post(
|
|
external_id=external_id, text=text, source="truth",
|
|
published_at=published_at,
|
|
sentiment="neutral", ai_confidence=0,
|
|
relevant=False, signal="hold",
|
|
prefilter_reason=filter_reason[:32],
|
|
analysis_version="entry_filter_rejected",
|
|
)
|
|
db.add(stub)
|
|
await db.flush()
|
|
return None # don't broadcast, don't trade
|
|
|
|
analysis = await analyze_post(text)
|
|
|
|
asset = analysis["asset"]
|
|
# `tracked_asset`: the asset whose price impact we measure and display.
|
|
# Use target_asset (the perp we actually trade — may be SOL/TRUMP/etc.)
|
|
# when available; fall back to the sentiment asset (BTC/ETH) otherwise.
|
|
# Bug fix: previously always used `asset` (BTC/ETH), which measured the
|
|
# wrong price move when the bot traded a different perp.
|
|
tracked_asset = analysis.get("target_asset") or asset
|
|
|
|
# Only capture the price AT post time. The m5/m15/m1h peaks are filled in
|
|
# asynchronously by price_impact_monitor as the windows elapse — avoids
|
|
# recording 0.00% because future candles don't exist yet at entry time.
|
|
price_at_post = None
|
|
if tracked_asset and analysis["relevant"]:
|
|
price_at_post = price_store.get_price_at(tracked_asset, published_at)
|
|
|
|
post = Post(
|
|
external_id=external_id,
|
|
text=text,
|
|
source="truth",
|
|
published_at=published_at,
|
|
sentiment=analysis["sentiment"],
|
|
signal=analysis.get("signal"),
|
|
ai_confidence=analysis["confidence"],
|
|
ai_reasoning=analysis.get("reasoning"),
|
|
prefilter_reason=analysis.get("prefilter_reason"),
|
|
analysis_version=analysis.get("analysis_version"),
|
|
relevant=analysis["relevant"],
|
|
# Track the actually-traded asset (target_asset ?? sentiment_asset).
|
|
price_impact_asset=tracked_asset if analysis["relevant"] else None,
|
|
price_impact_m5=None, # filled by price_impact_monitor after 5 m
|
|
price_impact_m15=None, # filled by price_impact_monitor after 15 m
|
|
price_impact_m1h=None, # filled by price_impact_monitor after 1 h
|
|
price_at_post=price_at_post,
|
|
# v5 routing: AI decides the actual perp to trade. May be SOL/TRUMP/etc.
|
|
target_asset=analysis.get("target_asset"),
|
|
category=analysis.get("category"),
|
|
expected_move_pct=analysis.get("expected_move_pct"),
|
|
)
|
|
db.add(post)
|
|
await db.flush()
|
|
|
|
# Register with the live peak tracker so it starts watching immediately.
|
|
if tracked_asset and analysis["relevant"] and price_at_post:
|
|
from app.services.price_impact_monitor import register_post
|
|
register_post(
|
|
post_id=post.id,
|
|
asset=tracked_asset,
|
|
signal=analysis.get("signal"),
|
|
entry_price=price_at_post,
|
|
published_at=published_at,
|
|
)
|
|
|
|
return post
|
|
|
|
|
|
def _post_to_ws_payload(post: Post) -> dict:
|
|
price_impact = None
|
|
if post.price_impact_asset and post.price_at_post is not None:
|
|
# At broadcast time all windows are open — values are null until
|
|
# price_impact_monitor fills them in. Frontend treats null as "pending".
|
|
price_impact = {
|
|
"asset": post.price_impact_asset,
|
|
"m5": post.price_impact_m5, # None = not yet measured
|
|
"m15": post.price_impact_m15,
|
|
"m1h": post.price_impact_m1h,
|
|
"price_at_post": post.price_at_post,
|
|
}
|
|
return {
|
|
"type": "new_post",
|
|
"post": {
|
|
"id": post.id,
|
|
"text": post.text,
|
|
"source": post.source,
|
|
"published_at": iso_utc(post.published_at),
|
|
"sentiment": post.sentiment,
|
|
"signal": post.signal,
|
|
"ai_confidence": post.ai_confidence,
|
|
"ai_reasoning": post.ai_reasoning,
|
|
"relevant": post.relevant,
|
|
"target_asset": post.target_asset,
|
|
"category": post.category,
|
|
"expected_move_pct": post.expected_move_pct,
|
|
"price_impact": price_impact,
|
|
},
|
|
}
|
|
|
|
|
|
async def dispatch_post(post: Post, db: AsyncSession) -> None:
|
|
"""Broadcast + fan-out + trade for one freshly committed post. Shared by
|
|
both pollers so delivery doesn't depend on which source wins the race."""
|
|
await manager.broadcast(_post_to_ws_payload(post))
|
|
logger.info("Saved new post id=%d: %s", post.id, post.text[:60])
|
|
# Telegram fan-out (fire-and-forget). _dispatch filters internally:
|
|
# buy/short → per-subscriber + public channel; relevant-but-hold →
|
|
# public channel only; noise → dropped.
|
|
try:
|
|
from app.services.telegram import notify_signal
|
|
notify_signal(post)
|
|
except Exception as exc:
|
|
logger.warning("Telegram notify failed for post %d: %s", post.id, exc)
|
|
try:
|
|
from app.services.x_poster import notify_x_signal
|
|
notify_x_signal(post)
|
|
except Exception as exc:
|
|
logger.warning("X notify failed for post %d: %s", post.id, exc)
|
|
try:
|
|
from app.services.bot_engine import process_post
|
|
await process_post(post, db)
|
|
except Exception as exc:
|
|
logger.error("process_post failed for post %d: %s", post.id, exc)
|
|
|
|
|
|
async def poll_truth_social(db_session_factory) -> None:
|
|
global last_successful_poll_at, last_poll_error
|
|
logger.info("Polling CNN Truth Social archive...")
|
|
entries = await _fetch_archive()
|
|
if entries is NOT_MODIFIED:
|
|
# Archive unchanged since last poll — successful cycle, nothing to do.
|
|
last_successful_poll_at = datetime.now(timezone.utc)
|
|
last_poll_error = None
|
|
return
|
|
if not entries:
|
|
last_poll_error = "fetch_archive returned empty"
|
|
return
|
|
|
|
# Only process the latest 50 entries each poll (archive has 30k+ posts)
|
|
recent = entries[:50]
|
|
logger.info("Checking %d recent entries...", len(recent))
|
|
|
|
async with db_session_factory() as db:
|
|
try:
|
|
known_ids = await _known_external_ids(recent, db)
|
|
found_new = False
|
|
# Entries are newest-first. Commit + dispatch each new post
|
|
# IMMEDIATELY rather than after the whole batch — an actionable
|
|
# post must not wait behind the AI analysis of older entries.
|
|
for entry in recent:
|
|
try:
|
|
post = await _process_entry(entry, db, known_ids)
|
|
if post:
|
|
found_new = True
|
|
await db.commit()
|
|
await dispatch_post(post, db)
|
|
except Exception as exc:
|
|
logger.error("Error processing entry %s: %s", entry.get("id"), exc)
|
|
|
|
# Capture any remaining writes (entry-filter stub rows).
|
|
await db.commit()
|
|
if not found_new:
|
|
logger.info("No new posts found.")
|
|
# Mark a successful poll cycle (separate from "found new posts").
|
|
last_successful_poll_at = datetime.now(timezone.utc)
|
|
last_poll_error = None
|
|
except Exception as exc:
|
|
logger.error("Transaction error: %s", exc)
|
|
last_poll_error = f"transaction_error: {exc}"
|
|
await db.rollback()
|
|
|
|
|
|
async def backfill_history(db_session_factory, limit: int = 500) -> None:
|
|
"""One-time backfill of historical posts (no Claude analysis, no price impact)."""
|
|
logger.info("Starting historical backfill (limit=%d)...", limit)
|
|
entries = await _fetch_archive(conditional=False)
|
|
if not entries:
|
|
logger.error("Backfill failed: could not fetch archive")
|
|
return
|
|
|
|
to_process = entries[:limit]
|
|
saved = 0
|
|
|
|
async with db_session_factory() as db:
|
|
try:
|
|
known_ids = await _known_external_ids(to_process, db)
|
|
for entry in to_process:
|
|
external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()
|
|
if external_id in known_ids:
|
|
continue
|
|
|
|
text = _strip_html(entry.get("content") or "").strip()
|
|
if not text:
|
|
continue
|
|
|
|
post = Post(
|
|
external_id=external_id,
|
|
text=text,
|
|
source="truth",
|
|
published_at=_parse_dt(entry.get("created_at", "")),
|
|
sentiment="neutral",
|
|
ai_confidence=0,
|
|
relevant=False,
|
|
)
|
|
db.add(post)
|
|
saved += 1
|
|
|
|
await db.commit()
|
|
logger.info("Backfill complete: saved %d posts", saved)
|
|
except Exception as exc:
|
|
logger.error("Backfill error: %s", exc)
|
|
await db.rollback()
|