trumpsignal-backend/scripts/launch_seed.py

#!/usr/bin/env python3
"""
Pre-launch data preparation — conservative version.

What it does:
  1. DROP obsolete test-only sources (test/phase1/breakout/rsi_reversal/sma_reclaim)
     and their orphan bot_trades. These are residue from past scanner experiments
     that no longer fire — they only confuse the UI/sitemap.
  2. TRUNCATE the KOL window to the last KOL_RETAIN_DAYS days (default 30).
     KOL data churns daily and a 30-day window is what the UI surfaces anyway.
  3. REFETCH every upstream source so the kept window is fresh:
       * Trump CNN archive (idempotent; only inserts genuinely new posts)
       * KOL Substack / podcast / blog polls
       * KOL on-chain snapshots (HL perps + Etherscan ERC-20 balances)
       * KOL divergence detector
       * Optional: BTC bottom-reversal + funding-reversal scanners

What it INTENTIONALLY DOES NOT touch:
  * posts where source IN ('truth', 'btc_bottom_reversal', 'funding_reversal',
    'kol_divergence') — full history kept. Trump posts are too valuable to
    truncate (analytics + accuracy backtest need them); the two BTC scanners
    fire too rarely (≤4×/cycle) to safely drop any historical fire.
  * bot_trades older than KOL_RETAIN_DAYS — real PnL history stays. Only
    bot_trades whose trigger_post has been deleted (i.e. the trade fired
    off an obsolete-source signal) are pruned.
  * subscriptions, telegram_bindings, kol_wallets, candles — untouched.
  * The most-recent KOL holdings snapshot per wallet — kept as the diff
    baseline for the next on-chain poll (without it, the next snapshot
    would diff against nothing and produce zero kol_holding_changes).

Usage:
    # Preview (no DB writes):
    DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\
        venv/bin/python scripts/launch_seed.py --dry-run

    # Seed only (pure pre-launch warmup, no truncation / deletion):
    DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\
        venv/bin/python scripts/launch_seed.py --seed-only

    # Execute:
    DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\
        venv/bin/python scripts/launch_seed.py --yes

    # Skip the SEED step (just clean — useful if you'd rather let the
    # scheduled poll jobs trigger naturally over the next 24h):
    DATABASE_URL='...' venv/bin/python scripts/launch_seed.py --yes --no-seed

NOT safe to re-run after real users are on the platform — the KOL truncation
will erase divergence/alignment context that you can't recover.
"""
from __future__ import annotations

import argparse
import asyncio
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from sqlalchemy import delete, select, func, text
from app.database import AsyncSessionLocal
from app.models import (
    Post, KolPost, KolDivergence, KolHoldingChange,
    KolHoldingSnapshot, KolWallet, BotTrade,
    Subscription, TelegramBinding,
)


# ── Tunables ────────────────────────────────────────────────────────────────
# Test-only sources to wipe from `posts` regardless of age. The first three are
# explicit fixtures; the latter two are old scanner experiments now archived
# under app/services/scanners/_archive/ but their historical fires linger.
OBSOLETE_SOURCES = {"test", "phase1", "breakout", "rsi_reversal", "sma_reclaim"}

# Live production sources — never truncated by this script. Trump's full
# history is needed for /analytics + /signals/accuracy; the two BTC scanners
# fire too rarely (<5 times per market cycle) to safely lose any past fire.
LIVE_SOURCES_PROTECTED = {"truth", "btc_bottom_reversal", "funding_reversal", "kol_divergence"}

# KOL data churns daily and the UI surfaces a max-30-day window. Anything older
# than this on the KOL side is dead weight in the DB.
KOL_RETAIN_DAYS = 30
# ─────────────────────────────────────────────────────────────────────────────


def green(s):  return f"\033[32m{s}\033[0m"
def yellow(s): return f"\033[33m{s}\033[0m"
def red(s):    return f"\033[31m{s}\033[0m"
def bold(s):   return f"\033[1m{s}\033[0m"


async def report_counts(label: str) -> dict:
    """Snapshot row counts across every signal-bearing table."""
    counts: dict = {}
    async with AsyncSessionLocal() as db:
        for tbl, name in [
            (Post,               "posts"),
            (KolPost,            "kol_posts"),
            (KolDivergence,      "kol_divergence"),
            (KolHoldingChange,   "kol_holding_changes"),
            (KolHoldingSnapshot, "kol_holdings_snapshots"),
            (KolWallet,          "kol_wallets"),
            (BotTrade,           "bot_trades"),
            (Subscription,       "subscriptions"),
            (TelegramBinding,    "telegram_bindings"),
        ]:
            n = (await db.execute(select(func.count(tbl.id)))).scalar() or 0
            counts[name] = n

        by_src = (await db.execute(
            select(Post.source, func.count(Post.id)).group_by(Post.source)
        )).all()
        counts["_posts_by_source"] = dict(by_src)

    print(bold(f"\n── {label} ──"))
    for k, v in counts.items():
        if k.startswith("_"): continue
        print(f"  {k:25s} {v}")
    print(f"  posts breakdown: {counts['_posts_by_source']}")
    return counts


async def wipe_phase(dry_run: bool) -> None:
    """Delete only obsolete sources + truncate KOL window. Live sources stay."""
    kol_cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=KOL_RETAIN_DAYS)

    print(bold("\n── WIPE ──"))
    print(f"  Obsolete-source posts to drop: {sorted(OBSOLETE_SOURCES)}")
    print(f"  KOL retention window: last {KOL_RETAIN_DAYS} days "
          f"(cutoff = {kol_cutoff.date()})")
    print(f"  Live sources kept in full: {sorted(LIVE_SOURCES_PROTECTED)}")

    # Quote source list once for repeated reuse — SQLite needs IN (...) with
    # parentheses even for a single value.
    obsolete_in = "(" + ",".join(f"'{s}'" for s in OBSOLETE_SOURCES) + ")"

    async with AsyncSessionLocal() as db:
        previews = [
            ("kol_divergence",
             f"divergences older than {KOL_RETAIN_DAYS}d",
             f"SELECT COUNT(*) FROM kol_divergence WHERE post_at < '{kol_cutoff.isoformat()}'"),
            ("kol_holding_changes",
             f"changes older than {KOL_RETAIN_DAYS}d",
             f"SELECT COUNT(*) FROM kol_holding_changes WHERE detected_at < '{kol_cutoff.isoformat()}'"),
            ("kol_holdings_snapshots",
             f"snapshots older than {KOL_RETAIN_DAYS}d (keeping latest-per-wallet as diff baseline)",
             f"""SELECT COUNT(*) FROM kol_holdings_snapshots
                 WHERE snapshot_date < '{kol_cutoff.date()}'
                 AND id NOT IN (
                   SELECT MAX(id) FROM kol_holdings_snapshots GROUP BY wallet_id
                 )"""),
            ("kol_posts",
             f"KOL posts older than {KOL_RETAIN_DAYS}d",
             f"SELECT COUNT(*) FROM kol_posts WHERE published_at < '{kol_cutoff.isoformat()}'"),
            ("posts",
             "obsolete-source posts (test/phase1/breakout/rsi/sma)",
             f"SELECT COUNT(*) FROM posts WHERE source IN {obsolete_in}"),
            ("bot_trades",
             "trades whose trigger_post is being deleted (orphan cleanup)",
             f"""SELECT COUNT(*) FROM bot_trades WHERE trigger_post_id IN (
                 SELECT id FROM posts WHERE source IN {obsolete_in}
             )"""),
        ]
        for table, desc, sql in previews:
            n = (await db.execute(text(sql))).scalar() or 0
            tag = yellow(f"  would delete {n:5d}") if dry_run else green(f"  delete {n:5d}")
            print(f"{tag}  from {table:25s}  ({desc})")

        if dry_run:
            print(yellow("\n  [DRY RUN] no DB changes made."))
            return

        # Execute in child→parent order to avoid FK violations on Postgres
        # (SQLite usually has FK enforcement off, but PG enforces it strictly).
        await db.execute(text(
            f"DELETE FROM kol_divergence WHERE post_at < '{kol_cutoff.isoformat()}'"
        ))
        await db.execute(text(
            f"DELETE FROM kol_holding_changes WHERE detected_at < '{kol_cutoff.isoformat()}'"
        ))
        await db.execute(text(f"""
            DELETE FROM kol_holdings_snapshots
            WHERE snapshot_date < '{kol_cutoff.date()}'
            AND id NOT IN (
              SELECT MAX(id) FROM kol_holdings_snapshots GROUP BY wallet_id
            )
        """))
        await db.execute(text(
            f"DELETE FROM kol_posts WHERE published_at < '{kol_cutoff.isoformat()}'"
        ))
        # bot_trades FIRST (it references posts.id) then posts.
        await db.execute(text(f"""
            DELETE FROM bot_trades WHERE trigger_post_id IN (
              SELECT id FROM posts WHERE source IN {obsolete_in}
            )
        """))
        await db.execute(text(
            f"DELETE FROM posts WHERE source IN {obsolete_in}"
        ))
        await db.commit()
        print(green("  ✓ wipe committed"))


async def abort_if_live_user_state() -> bool:
    """Refuse destructive launch seeding if the DB already looks user-live."""
    async with AsyncSessionLocal() as db:
        active_subs = (await db.execute(
            select(func.count(Subscription.id)).where(Subscription.active == True)
        )).scalar() or 0
        tg_bindings = (await db.execute(
            select(func.count(TelegramBinding.id))
        )).scalar() or 0
    if active_subs or tg_bindings:
        print(red(
            "Refusing to run launch_seed on a DB that already has live user state "
            f"(active_subscriptions={active_subs}, telegram_bindings={tg_bindings})."
        ))
        print(yellow(
            "Use --dry-run to inspect only. If you truly intend to clean this DB, "
            "do it manually with a one-off migration/backup plan."
        ))
        return True
    return False


async def seed_real_data(*, exercise_scanners: bool = False) -> None:
    """Re-fetch every upstream source. All are idempotent on (source, external_id)."""
    print(bold("\n── SEED (re-running upstream fetches) ──"))

    print("\n  [1/5] Trump backfill (CNN archive)...")
    from app.scrapers.truth_social import backfill_history
    try:
        await backfill_history(AsyncSessionLocal, limit=500)
        print(green("        ✓ done"))
    except Exception as e:
        print(red(f"        ✗ FAILED: {type(e).__name__}: {e}"))

    print("\n  [2/5] KOL Substack/podcast/blog polling...")
    from app.services.kol_substack import run_substack_poll
    try:
        results = await run_substack_poll(analyze=True)
        total_new = sum(r.get("new", 0) for r in results)
        total_err = sum(r.get("errors", 0) for r in results)
        print(green(f"        ✓ done — {total_new} new posts ({total_err} errors)"))
    except Exception as e:
        print(red(f"        ✗ FAILED: {type(e).__name__}: {e}"))

    print("\n  [3/5] KOL on-chain snapshot (HL perps + Etherscan)...")
    from app.services.kol_onchain import run_onchain_poll
    try:
        result = await run_onchain_poll()
        print(green(f"        ✓ done — {result}"))
    except Exception as e:
        print(red(f"        ✗ FAILED: {type(e).__name__}: {e}"))

    print("\n  [4/5] KOL divergence detection (talks vs trades)...")
    from app.services.kol_divergence import run_divergence_scan
    try:
        # Run against the kept KOL window so we catch every still-relevant pair.
        new = await run_divergence_scan(lookback_days=KOL_RETAIN_DAYS)
        print(green(f"        ✓ done — {len(new)} new divergence/alignment pairs"))
    except Exception as e:
        print(red(f"        ✗ FAILED: {type(e).__name__}: {e}"))

    if exercise_scanners:
        print("\n  [5/5] BTC bottom + funding reversal scanners (exercise the path)...")
        from app.services.scanners.btc_bottom_reversal import scan_once as btc_scan
        from app.services.scanners.funding_reversal import scan_once as funding_scan
        for name, fn in [("btc_bottom", btc_scan), ("funding_reversal", funding_scan)]:
            try:
                await fn()
                print(green(f"        ✓ {name} scan completed (fire conditional on market state)"))
            except Exception as e:
                print(red(f"        ✗ {name} FAILED: {type(e).__name__}: {e}"))
    else:
        print(yellow(
            "\n  [5/5] Scanner exercise skipped by default. "
            "Use --exercise-scanners only if you explicitly want signal emission side-effects."
        ))


async def main() -> int:
    p = argparse.ArgumentParser(description=__doc__,
                                formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--dry-run", action="store_true",
                   help="show what would be deleted, no DB writes")
    p.add_argument("--seed-only", action="store_true",
                   help="run upstream fetch/warmup only; skip all wipe/truncation steps")
    p.add_argument("--yes", action="store_true",
                   help="actually perform the wipe (required without --dry-run)")
    p.add_argument("--no-seed", action="store_true",
                   help="skip the upstream re-fetch step")
    p.add_argument("--exercise-scanners", action="store_true",
                   help="also run standalone scanners during seed (may emit signals/alerts)")
    args = p.parse_args()

    if args.seed_only and args.dry_run:
        print(red("Choose either --dry-run or --seed-only, not both."))
        return 2

    if args.seed_only and args.no_seed:
        print(red("--seed-only conflicts with --no-seed."))
        return 2

    if args.seed_only:
        before = await report_counts("BEFORE")
        await seed_real_data(exercise_scanners=args.exercise_scanners)
        after = await report_counts("AFTER")

        print(bold("\n── DELTA ──"))
        for k in sorted(before):
            if k.startswith("_"): continue
            d = after[k] - before[k]
            arrow = green(f"+{d}") if d > 0 else (red(str(d)) if d < 0 else "  ·")
            print(f"  {k:25s} {before[k]:6d} → {after[k]:6d}  ({arrow})")

        print(bold(green("\n✓ seed-only warmup complete. Next: start backend and run scripts/launch_smoke.py.")))
        return 0

    if not args.dry_run and not args.yes:
        print(red("Refusing to run without --yes (or use --dry-run to preview)."))
        return 2

    if not args.dry_run and await abort_if_live_user_state():
        return 2

    before = await report_counts("BEFORE")
    await wipe_phase(dry_run=args.dry_run)

    if args.dry_run:
        print(yellow("\n[DRY RUN COMPLETE] re-run with --yes to execute."))
        return 0

    if not args.no_seed:
        await seed_real_data(exercise_scanners=args.exercise_scanners)

    after = await report_counts("AFTER")

    print(bold("\n── DELTA ──"))
    for k in sorted(before):
        if k.startswith("_"): continue
        d = after[k] - before[k]
        arrow = green(f"+{d}") if d > 0 else (red(str(d)) if d < 0 else "  ·")
        print(f"  {k:25s} {before[k]:6d} → {after[k]:6d}  ({arrow})")

    print(bold(green("\n✓ reset complete. Next: run scripts/launch_smoke.py after the backend is up.")))
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))