trumpsignal-backend/app/scrapers/truth_social.py

"""
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

import httpx
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


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() -> Optional[list]:
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)",
        "Accept": "application/json",
    }
    try:
        async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
            resp = await client.get(ARCHIVE_URL, headers=headers)
            resp.raise_for_status()
            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 _process_entry(entry: dict, db: AsyncSession) -> Optional[Post]:
    external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()

    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 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 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:
            new_posts = []
            for entry in recent:
                try:
                    post = await _process_entry(entry, db)
                    if post:
                        new_posts.append(post)
                except Exception as exc:
                    logger.error("Error processing entry %s: %s", entry.get("id"), exc)

            if new_posts:
                await db.commit()
                for post in new_posts:
                    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)
            else:
                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()
    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:
            for entry in to_process:
                external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()
                result = await db.execute(select(Post).where(Post.external_id == external_id))
                if result.scalar_one_or_none():
                    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()