trumpsignal-backend/app/main.py

import asyncio
import logging
from contextlib import asynccontextmanager

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.database import AsyncSessionLocal, engine
from app.models import Base
from app.scrapers.truth_social import poll_truth_social, backfill_history
from app.scrapers.trumpstruth import poll_trumpstruth
from app.services.binance import run_binance_ws
from app.ws.manager import manager

# Import all routers
from app.api.posts import router as posts_router
from app.api.prices import router as prices_router
from app.api.trades import router as trades_router
from app.api.performance import router as performance_router
from app.api.subscribe import router as subscribe_router
from app.api.user import router as user_router
from app.api.funding_signal import router as funding_signal_router
from app.api.funding_reversal import router as funding_reversal_router
from app.api.telegram import router as telegram_router
from app.api.signals import router as signals_router
from app.api.positions import router as positions_router
from app.api.scanners  import router as scanners_router
from app.api.kol       import router as kol_router
from app.api.macro     import router as macro_router

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

from typing import Optional
_binance_task: Optional[asyncio.Task] = None
_telegram_task: Optional[asyncio.Task] = None
_scheduler: Optional[AsyncIOScheduler] = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global _binance_task, _telegram_task, _scheduler

    # 1. Dev convenience only. Production should rely on Alembic so schema
    # ownership stays explicit and startup never mutates the DB implicitly.
    if settings.environment == "development":
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        logger.info("Database tables ensured (development mode).")
    else:
        logger.info("Skipping create_all; expecting schema managed by Alembic.")

    # 2. Backfill historical posts on startup (fast, no Claude API call)
    asyncio.create_task(backfill_history(AsyncSessionLocal, limit=500))

    # 2b. Rehydrate TP/SL + max-hold backstops for any positions still open in DB.
    # Critical: without this, a redeploy/crash silently drops protection on live trades.
    from app.services.recovery import rehydrate_open_trades
    asyncio.create_task(rehydrate_open_trades())

    # 3. Start Binance WebSocket background task
    _binance_task = asyncio.create_task(run_binance_ws(), name="binance_ws")
    logger.info("Binance WebSocket task started.")

    # 3. Start Truth Social poller via APScheduler
    _scheduler = AsyncIOScheduler()
    # Signal monitor — polls every 5 minutes
    from app.services.funding_signal import poll_funding_signal
    _scheduler.add_job(
        poll_funding_signal,
        "interval",
        minutes=5,
        id="funding_signal_poll",
        max_instances=1,
        coalesce=True,
    )
    logger.info("Breakout signal monitor scheduled every 5 minutes.")

    _scheduler.add_job(
        poll_truth_social,
        "interval",
        seconds=settings.truth_social_poll_seconds,
        args=[AsyncSessionLocal],
        id="truth_social_poll",
        max_instances=1,
        coalesce=True,
    )
    # Fallback poller — trumpstruth.org RSS. Same Truth Social originalId =
    # same external_id hash, so dedup is automatic. Whoever sees the post
    # first wins. Offset by half the interval so the two pollers don't
    # hammer the network at the same instant.
    # Offset the fallback by half the interval so the two pollers don't hit
    # upstream at the same instant. APScheduler treats next_run_time=None as
    # "paused" (job never fires) — must be an actual datetime.
    from datetime import datetime as _dt, timezone as _tz, timedelta as _td
    offset = settings.truth_social_poll_seconds // 2
    _scheduler.add_job(
        poll_trumpstruth,
        "interval",
        seconds=settings.truth_social_poll_seconds,
        args=[AsyncSessionLocal],
        id="trumpstruth_poll",
        max_instances=1,
        coalesce=True,
        next_run_time=_dt.now(_tz.utc) + _td(seconds=offset),
    )
    # HL <-> DB reconciliation — every 60s, detects state drift
    # (manual closes on HL UI, liquidations, orphan positions). See
    # app/services/reconciler.py. Critical for live trading safety.
    from app.services.reconciler import reconcile_all_once, RECONCILE_INTERVAL_SECONDS
    _scheduler.add_job(
        reconcile_all_once,
        "interval",
        seconds=RECONCILE_INTERVAL_SECONDS,
        id="hl_reconcile",
        max_instances=1,
        coalesce=True,
    )
    logger.info("HL <-> DB reconciler scheduled every %ds.", RECONCILE_INTERVAL_SECONDS)

    # ── System-2 bottom-reversal state machine ─────────────────────────────
    # Low-frequency, long-only: fires when ≥2 of 3 classic bottom signals
    # agree — AHR999 < 0.45, price ≤ 200-week MA ×1.05, Pi Cycle Bottom
    # (150d EMA ≤ 471d SMA × 0.745). Funding is a booster inside the scanner,
    # not an independent entry source. See app/services/scanners/btc_bottom_reversal.py.
    from app.services.scanners.btc_bottom_reversal import scan_once as btc_bottom_scan
    _scheduler.add_job(
        btc_bottom_scan, "cron", hour=0, minute=45,
        id="btc_bottom_reversal_scan", max_instances=1, coalesce=True,
    )
    logger.info("BTC bottom-reversal scanner scheduled daily at 00:45 UTC.")

    # ── BTC funding-rate reversal (hourly) ────────────────────────────────
    # Mean-reversion play on extreme perp positioning. Independent of the
    # bottom-reversal state machine but uses the same `evaluate_funding_reversal`
    # algorithm. Runs every hour because HL funding settles hourly.
    from app.services.scanners.funding_reversal import scan_once as funding_scan
    _scheduler.add_job(
        funding_scan, "cron", minute=7,   # :07 every hour, away from other jobs
        id="funding_reversal_scan", max_instances=1, coalesce=True,
    )
    logger.info("Funding reversal scanner scheduled hourly at :07.")

    # ── KOL Substack poller (daily) ──────────────────────────────────────
    # Hayes & co publish at most a few times a week. Daily poll is plenty;
    # RSS dedupe by URL is idempotent if it ever fires twice.
    from app.services.kol_substack import run_substack_poll
    _scheduler.add_job(
        run_substack_poll, "cron", hour=1, minute=15,
        id="kol_substack_poll", max_instances=1, coalesce=True,
    )
    logger.info("KOL Substack poller scheduled daily at 01:15 UTC.")

    # ── KOL A-tier: on-chain holdings snapshot (daily) ────────────────────
    # Polls HL public API (free) for perp positions; Arkham (key optional)
    # for full portfolio. Diffs against yesterday's snapshot → writes
    # kol_holding_changes. Runs at 02:00 UTC, after Substack poll finishes.
    from app.services.kol_onchain import run_onchain_poll
    _scheduler.add_job(
        run_onchain_poll, "cron", hour=2, minute=0,
        id="kol_onchain_poll", max_instances=1, coalesce=True,
    )
    logger.info("KOL on-chain holdings poller scheduled daily at 02:00 UTC.")

    # ── KOL talks-vs-trades divergence scan (daily) ───────────────────────
    # Runs after the on-chain poll finishes. Matches post ticker signals vs
    # holding changes for the same KOL+ticker within ±7 days.
    from app.services.kol_divergence import run_divergence_scan
    _scheduler.add_job(
        run_divergence_scan, "cron", hour=2, minute=15,
        id="kol_divergence_scan", max_instances=1, coalesce=True,
    )
    logger.info("KOL divergence scan scheduled daily at 02:15 UTC.")

    # ── Macro indicator daily snapshot ────────────────────────────────────
    # 8 indicators (AHR999, Fear & Greed, BTC dominance, ETH/BTC, stablecoin
    # supply, ETF flow, BTC OI, altcoin season). One row per calendar date.
    # Runs after KOL jobs so a slow KOL fetch can't make this one miss.
    from app.services.macro.poll import run_macro_poll
    _scheduler.add_job(
        run_macro_poll, "cron", hour=3, minute=0,
        id="macro_poll", max_instances=1, coalesce=True,
    )
    logger.info("Macro indicator snapshot scheduled daily at 03:00 UTC.")

    # Kick off an initial poll on startup IF today's row doesn't exist yet.
    # Otherwise a fresh deploy shows an empty macro panel until 03:00 UTC of
    # the next day. Fire-and-forget — never blocks startup.
    async def _macro_bootstrap():
        try:
            from datetime import datetime, timezone
            from sqlalchemy import select
            from app.models import MacroSnapshot
            today = datetime.now(timezone.utc).date()
            async with AsyncSessionLocal() as db:
                exists = (await db.execute(
                    select(MacroSnapshot).where(MacroSnapshot.snapshot_date == today)
                )).scalar_one_or_none()
            if exists is None:
                logger.info("Macro: no row for today, running one-shot bootstrap fetch.")
                await run_macro_poll()
        except Exception as exc:
            logger.warning("Macro bootstrap fetch failed: %s (%s)",
                           type(exc).__name__, exc)
    asyncio.create_task(_macro_bootstrap())

    # ── Telegram daily digest ─────────────────────────────────────────────
    # Runs every hour at :00. Each binding picks its own digest_hour_utc
    # (0–23, default 12); the sender only fans out to rows whose preferred
    # hour matches the current one — so we get one cron job covering all 24
    # delivery slots. The send function itself dedupes against
    # last_digest_sent_at (skips if < 23h), so a coalesced cron or worker
    # restart can't double-send a user in the same day.
    from app.services.telegram_digest import send_daily_digest
    _scheduler.add_job(
        send_daily_digest, "cron", minute=0,
        id="telegram_daily_digest",
        max_instances=1, coalesce=True,
    )
    logger.info("Telegram daily digest scheduled hourly at :00 UTC "
                "(per-user hour selection).")

    _scheduler.start()
    logger.info(
        "Truth Social pollers scheduled every %ds (CNN + trumpstruth.org).",
        settings.truth_social_poll_seconds,
    )

    # ── Telegram bot long-poll loop (optional) ────────────────────────────
    # Only started if TELEGRAM_BOT_TOKEN is set. Handles /start CODE bindings
    # and /stop, /status, /test commands. Survives transient network errors
    # via internal back-off.
    from app.services.telegram_bot import run_bot_loop
    _telegram_task = asyncio.create_task(run_bot_loop(), name="telegram_bot")
    logger.info("Telegram bot task created (will no-op if token missing).")

    yield

    # Shutdown
    logger.info("Shutting down background tasks...")
    if _scheduler:
        _scheduler.shutdown(wait=False)
    if _binance_task and not _binance_task.done():
        _binance_task.cancel()
        try:
            await _binance_task
        except asyncio.CancelledError:
            pass
    if _telegram_task and not _telegram_task.done():
        _telegram_task.cancel()
        try:
            await _telegram_task
        except asyncio.CancelledError:
            pass
    await engine.dispose()
    logger.info("Shutdown complete.")


app = FastAPI(
    title="TrumpSignal API",
    version="1.0.0",
    description="Crypto trading signals derived from Trump's Truth Social posts.",
    lifespan=lifespan,
)

# CORS
# In production we only allow the canonical frontend origin (FRONTEND_URL).
# In development we additionally permit the local Next dev server. NEVER
# permit "*" here — every endpoint either reads/writes user-personalised
# data or accepts signed envelopes, both of which require credentialled
# requests (and "*" is rejected by browsers in combination with credentials
# anyway).
allowed_origins = [settings.frontend_url]
if settings.environment == "development":
    allowed_origins.extend([
        "http://localhost:3000",
        "http://localhost:3001",
    ])
app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# REST routers
app.include_router(posts_router, prefix="/api")
app.include_router(prices_router, prefix="/api")
app.include_router(trades_router, prefix="/api")
app.include_router(performance_router, prefix="/api")
app.include_router(subscribe_router, prefix="/api")
app.include_router(user_router, prefix="/api")
app.include_router(funding_signal_router,   prefix="/api")
app.include_router(funding_reversal_router, prefix="/api")
app.include_router(telegram_router,         prefix="/api")
app.include_router(signals_router,        prefix="/api")
app.include_router(positions_router,      prefix="/api")
app.include_router(scanners_router,       prefix="/api")
app.include_router(kol_router,             prefix="/api")
app.include_router(macro_router,           prefix="/api")


@app.get("/api/health")
async def health():
    """Shallow liveness — used by load balancers / Docker healthcheck.
    Returns 200 as long as the process is alive and the event loop responds."""
    return {"status": "ok"}


@app.get("/api/health/deep")
async def health_deep():
    """Deep healthcheck — used by external uptime monitors (UptimeRobot, etc).

    Returns 503 if any of:
      - DB query fails
      - Scraper hasn't completed a successful poll in > 90s
        (poll runs every 15s; allow 6 misses before alarm)

    Body always includes diagnostic fields so the alarm payload is actionable.
    """
    from datetime import datetime, timezone
    from fastapi import Response
    from sqlalchemy import text
    from app.scrapers import truth_social as ts_scraper
    from app.scrapers import trumpstruth as ts_fallback

    now = datetime.now(timezone.utc)
    problems: list[str] = []

    # 1. DB ping
    db_ok = False
    db_error: Optional[str] = None
    try:
        async with AsyncSessionLocal() as db:
            await db.execute(text("SELECT 1"))
        db_ok = True
    except Exception as exc:
        db_error = str(exc)
        problems.append(f"db: {db_error}")

    # 2. Scraper liveness — system is healthy if AT LEAST ONE source polled
    # recently. Both stale = real problem (network down, or both upstreams dead).
    sources = [
        ("cnn",          ts_scraper.last_successful_poll_at,  ts_scraper.last_poll_error),
        ("trumpstruth",  ts_fallback.last_successful_poll_at, ts_fallback.last_poll_error),
    ]
    source_status = []
    freshest_age: Optional[int] = None
    for name, last, err in sources:
        age = int((now - last).total_seconds()) if last else None
        source_status.append({
            "name": name,
            "last_poll": last.isoformat() if last else None,
            "age_sec": age,
            "last_error": err,
        })
        if age is not None and (freshest_age is None or age < freshest_age):
            freshest_age = age

    if freshest_age is None:
        problems.append("scrapers: no source has ever completed a poll")
    elif freshest_age > 90:
        # 6× the 15s interval — both sources are silent
        problems.append(f"scrapers: all stale (freshest={freshest_age}s)")

    body = {
        "status": "ok" if not problems else "degraded",
        "now": now.isoformat(),
        "db_ok": db_ok,
        "db_error": db_error,
        "scrapers": source_status,
        "freshest_age_sec": freshest_age,
        "problems": problems,
    }

    if problems:
        return Response(
            content=__import__("json").dumps(body),
            status_code=503,
            media_type="application/json",
        )
    return body

if settings.environment == "development":
    from app.api.dev import router as dev_router
    app.include_router(dev_router, prefix="/api")


# ── WebSocket endpoints ────────────────────────────────────────────────────


@app.websocket("/ws/prices")
async def ws_prices(websocket: WebSocket):
    """Subscribe to live price ticks (type: 'price')."""
    await manager.connect(websocket)
    try:
        while True:
            # Keep connection alive; messages are pushed via manager.broadcast
            await websocket.receive_text()
    except WebSocketDisconnect:
        await manager.disconnect(websocket)
    except Exception as exc:
        logger.warning("WebSocket /ws/prices error: %s", exc)
        await manager.disconnect(websocket)


@app.websocket("/ws/posts")
async def ws_posts(websocket: WebSocket):
    """Subscribe to new post events (type: 'new_post')."""
    await manager.connect(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        await manager.disconnect(websocket)
    except Exception as exc:
        logger.warning("WebSocket /ws/posts error: %s", exc)
        await manager.disconnect(websocket)