trumpsignal-backend/scripts/rescore_v5.py

"""
One-shot migration: re-score every post in the DB with the v5 prompt.

Use this after deploying the v5 analysis.py to wipe the v4 transition state.
After running, every row in `posts` will have analysis_version='v5-extreme-alpha'.

Usage (run on the server, in the backend container):

    # 1. Dry-run — shows what would change, no DB writes, no AI calls
    python -m scripts.rescore_v5 --dry-run

    # 2. Live run — confirm cost, then re-score everything
    python -m scripts.rescore_v5

    # 3. Force re-run on already-v5 rows (rare; only if you change the prompt
    #    again without bumping ANALYSIS_VERSION)
    python -m scripts.rescore_v5 --force

What it does:
  • Reads every post not already at v5 (or every post if --force).
  • Calls analyze_post() — which internally applies the new prefilter and
    will skip AI for ~3-5% of posts (RTs, bare URLs, empty bodies).
  • Updates these columns in place: signal, ai_confidence, ai_reasoning,
    sentiment, relevant, prefilter_reason, analysis_version.
  • DOES NOT touch: price_at_post, price_impact_*, opened/closed_at on
    related trades. Those stay accurate; only the AI's interpretation
    of the post is rewritten.
  • Sleeps 0.6s between AI calls to stay under your provider's rate limit.
  • Logs progress every 10 posts. Continues on per-post errors.

Final report shows the signal-distribution diff so you can sanity-check
that v5 actually drops the actionable count by ~3-5×.
"""

import argparse
import asyncio
import logging
import sys
import time
from collections import Counter

from sqlalchemy import select

from app.database import AsyncSessionLocal
from app.models import Post
from app.services.analysis import analyze_post, ANALYSIS_VERSION

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("rescore")

SLEEP_BETWEEN_CALLS = 0.6   # seconds, ~1.6 req/s — under most provider quotas


async def fetch_targets(force: bool):
    async with AsyncSessionLocal() as db:
        if force:
            stmt = select(Post)
        else:
            stmt = select(Post).where(Post.analysis_version != ANALYSIS_VERSION)
        rows = (await db.execute(stmt)).scalars().all()
        return [
            {"id": r.id, "text": r.text, "old_signal": r.signal,
             "old_conf": r.ai_confidence, "old_version": r.analysis_version}
            for r in rows
        ]


async def update_post(post_id: int, new: dict) -> bool:
    """Write one re-scored row back. Returns True on success."""
    async with AsyncSessionLocal() as db:
        post = await db.get(Post, post_id)
        if post is None:
            return False
        post.signal = new["signal"]
        post.ai_confidence = new["confidence"]
        post.ai_reasoning = new.get("reasoning") or ""
        post.sentiment = new["sentiment"]
        post.relevant = new["relevant"]
        post.prefilter_reason = new.get("prefilter_reason")
        post.analysis_version = new["analysis_version"]
        # v5 routing fields — without these the bot can't route old posts
        # correctly. analyze_post() always returns these (None for hold).
        post.target_asset = new.get("target_asset")
        post.category = new.get("category")
        post.expected_move_pct = new.get("expected_move_pct")
        # Note: do NOT touch price_impact_asset / price_at_post / m5/m15/m1h.
        # Those represent actual market behavior (independent of AI's call)
        # and stay accurate. The signals/accuracy endpoint will recompute
        # directional correctness against the new signal automatically.
        await db.commit()
    return True


async def main(dry_run: bool, force: bool):
    targets = await fetch_targets(force)
    logger.info("Found %d posts to re-score (force=%s)", len(targets), force)
    if not targets:
        logger.info("Nothing to do. Exit.")
        return

    # Pre-flight: dry-run shows distribution we'd start from
    old_counts = Counter(t["old_signal"] for t in targets)
    logger.info("Current signal distribution: %s", dict(old_counts))

    if dry_run:
        logger.info("DRY-RUN: would call analyze_post() %d times.", len(targets))
        logger.info("DRY-RUN: re-run without --dry-run to actually write.")
        return

    # Confirmation guard for live runs (skipped if stdin is not a tty,
    # e.g. when piped from CI or Docker exec without -it)
    if sys.stdin.isatty():
        msg = (f"\nAbout to re-score {len(targets)} posts via the AI provider.\n"
               f"Estimated cost: ~${len(targets) * 0.0054:.2f} on Haiku.\n"
               f"Estimated time: ~{len(targets) * SLEEP_BETWEEN_CALLS / 60:.1f} min.\n"
               f"Type 'yes' to proceed: ")
        if input(msg).strip().lower() != "yes":
            logger.info("Cancelled.")
            return

    started = time.time()
    new_signals: list[str] = []
    errors = 0

    for i, t in enumerate(targets, 1):
        try:
            new = await analyze_post(t["text"])
            ok = await update_post(t["id"], new)
            if not ok:
                logger.warning("post %d disappeared mid-run", t["id"])
                continue
            new_signals.append(new["signal"])

            if t["old_signal"] != new["signal"]:
                logger.info(
                    "  id=%d  %s(%s) → %s(%d)",
                    t["id"], t["old_signal"], t["old_conf"],
                    new["signal"], new["confidence"],
                )
        except Exception as exc:
            errors += 1
            logger.error("Failed on post %d: %s", t["id"], exc)
            new_signals.append(t["old_signal"])  # keep old in counter

        if i % 10 == 0:
            rate = i / (time.time() - started)
            eta = (len(targets) - i) / rate
            logger.info("Progress: %d/%d  (%.1f/s, ETA %.1f min)",
                        i, len(targets), rate, eta / 60)

        await asyncio.sleep(SLEEP_BETWEEN_CALLS)

    # ── Final diff ─────────────────────────────────────────────────
    new_counts = Counter(new_signals)
    elapsed_min = (time.time() - started) / 60

    logger.info("=" * 60)
    logger.info("RESCORE COMPLETE")
    logger.info("=" * 60)
    logger.info("Elapsed:  %.1f min", elapsed_min)
    logger.info("Errors:   %d", errors)
    logger.info("")
    logger.info("Signal distribution diff:")
    logger.info("  %-8s  %8s  %8s  %8s", "signal", "before", "after", "delta")
    for sig in ("hold", "buy", "short", "sell", None):
        b = old_counts.get(sig, 0)
        a = new_counts.get(sig, 0)
        if b == 0 and a == 0:
            continue
        delta = a - b
        sign = "+" if delta > 0 else ""
        logger.info("  %-8s  %8d  %8d  %s%d",
                    str(sig), b, a, sign, delta)

    # Reality check: actionable rate should drop substantially
    actionable_before = old_counts.get("buy", 0) + old_counts.get("short", 0) + old_counts.get("sell", 0)
    actionable_after = new_counts.get("buy", 0) + new_counts.get("short", 0)
    pct_before = actionable_before / len(targets) * 100
    pct_after = actionable_after / len(targets) * 100
    logger.info("")
    logger.info("Actionable rate:  %.1f%% → %.1f%%   (target: 2-4%%)",
                pct_before, pct_after)
    if actionable_after >= actionable_before * 0.8:
        logger.warning("⚠️  Actionable rate barely dropped. Either v5 prompt isn't")
        logger.warning("   strict enough, or sample is unusual. Inspect a few posts.")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--dry-run", action="store_true",
                    help="Show what would change without calling AI or writing DB")
    ap.add_argument("--force", action="store_true",
                    help="Re-score even posts already at the current version")
    args = ap.parse_args()

    asyncio.run(main(dry_run=args.dry_run, force=args.force))