trumpsignal-backend/app/main.py

import asyncio
import fcntl
import logging
import os
from contextlib import asynccontextmanager

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware

from app.config import settings
from app.ratelimit import limiter
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.services.hl_price_feed import run_hl_price_feed
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 datetime import datetime, timezone
from typing import Optional
_binance_task:    Optional[asyncio.Task] = None
_hl_price_task:   Optional[asyncio.Task] = None
_telegram_task:   Optional[asyncio.Task] = None
_scheduler:       Optional[AsyncIOScheduler] = None

# Process boot time — used by /api/health/deep to grant feeds a startup grace
# period before "no tick yet" counts as a failure (avoids a false 503 at boot).
_PROCESS_START_AT = datetime.now(timezone.utc)
_FEED_BOOT_GRACE_SEC = 180

# Single-instance guard. The backend is single-process BY DESIGN (APScheduler,
# scanner_state, the replay cache, tp_sl_monitor's open-trade table and
# price_store are all in-memory). Running >1 worker silently duplicates the
# scheduler, splits TP/SL state, and lets two processes both long-poll the
# Telegram token. This advisory file lock lets only ONE process ("the leader")
# start background tasks; any extra worker serves HTTP reads only and logs loud.
_singleton_lock_fd = None
_is_leader: bool = False


def _acquire_singleton_lock() -> bool:
    """Return True if this process won the advisory lock and may run background
    tasks. False means another process/worker already holds it (multi-worker
    misconfig). The OS releases the lock automatically when the holder exits,
    so a crashed leader frees it for the next start."""
    global _singleton_lock_fd
    path = os.environ.get("SINGLETON_LOCK_PATH", "/tmp/trumpsignal-backend.lock")
    try:
        fd = open(path, "w")
        fcntl.flock(fd.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        try:
            fd.write(str(os.getpid()))
            fd.flush()
        except OSError:
            pass
        _singleton_lock_fd = fd  # keep the fd alive for the process lifetime
        return True
    except BlockingIOError:
        # Lock held by another process — this is the multi-worker signal.
        return False
    except OSError as exc:
        # Lock file itself is unusable (perms, missing dir). Fail OPEN so a
        # lock-path problem never takes the whole service down — but log loudly.
        logger.warning("Singleton lock unavailable (%s) — proceeding as leader.", exc)
        return True


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

    # 0. Single-instance guard. If another worker already holds the lock, this
    # process must NOT start the scheduler / scrapers / price feeds / Telegram
    # poller — otherwise scans double-fire, TP/SL state splits across processes,
    # and two pollers fight over the Telegram token. Serve HTTP reads only.
    _is_leader = _acquire_singleton_lock()
    if not _is_leader:
        logger.critical(
            "SINGLETON LOCK NOT ACQUIRED — another backend process/worker is "
            "already the leader. This process will serve HTTP reads ONLY and "
            "start NO background tasks. You are almost certainly running with "
            ">1 worker, which is UNSUPPORTED (see deploy/supervisor.conf). "
            "Run uvicorn/gunicorn with --workers 1."
        )
        yield
        return

    # 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.")

    # 3b. Supplemental HL price feed for assets not on Binance (HYPE, PURR, …)
    _hl_price_task = asyncio.create_task(run_hl_price_feed(), name="hl_price_feed")
    logger.info("HL supplemental price feed 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 X (Twitter) tweet ingestion (daily) ───────────────────────────
    # Polls X-native KOLs (andrewkang/@Rewkang, murad — no Substack feed) plus
    # Hayes's real-time position statements. Writes KolPost(source="twitter")
    # that the divergence scan (02:15) joins against on-chain data — this is the
    # post-side feed that lights up the andrewkang/murad wallets. No-op if
    # twitterapi_io_key is unset. Runs 01:30 — between substack (01:15) and
    # on-chain (02:00) so fresh posts are present for the 02:15 divergence scan.
    from app.services.kol_x import run_x_poll
    _scheduler.add_job(
        run_x_poll, "cron", hour=1, minute=30,
        id="kol_x_poll", max_instances=1, coalesce=True,
    )
    logger.info("KOL X (Twitter) poller scheduled daily at 01:30 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 _hl_price_task and not _hl_price_task.done():
        _hl_price_task.cancel()
        try:
            await _hl_price_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.")


# Rate limiter is the shared instance from app.ratelimit — it keys on the
# real client IP (x-forwarded-for / x-real-ip) relayed by the Next.js proxy,
# NOT the proxy's own IP. See app/ratelimit.py for the BUG-02 rationale.
# Public read endpoints: 60 req/min. Signed mutations: 20 req/min (per-route).

app = FastAPI(
    title="TrumpSignal API",
    version="1.0.0",
    description="Crypto trading signals derived from Trump's Truth Social posts.",
    lifespan=lifespan,
)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# Register the middleware so default_limits actually applies to EVERY route
# (not just the two with an explicit @limiter.limit decorator). Without this,
# all signed-mutation routes had no rate limit at all. WebSocket scope is
# skipped by slowapi. Decorated routes keep their stricter explicit limit.
app.add_middleware(SlowAPIMiddleware)

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

    # 3. Price-feed liveness — INDEPENDENT of the scrapers. A dead price feed
    # means tp_sl_monitor never fires: stop-loss / take-profit / trailing stops
    # silently stop protecting live trades. This is a real-money incident the
    # scraper check above would completely miss, so it gets its own per-feed gate.
    from app.services import binance as _binance
    from app.services import hl_price_feed as _hl_feed

    boot_age = int((now - _PROCESS_START_AT).total_seconds())
    in_grace = boot_age < _FEED_BOOT_GRACE_SEC
    # (feed name, last_tick_at, max age before stale). Binance ticks multiple
    # times/sec for BTC/ETH; HL feed polls every 2s. Thresholds are generous.
    feeds = [
        ("binance_ws",    _binance.last_tick_at, 120),
        ("hl_price_feed", _hl_feed.last_tick_at,  60),
    ]
    feed_status = []
    for name, last, max_age in feeds:
        age = int((now - last).total_seconds()) if last else None
        feed_status.append({
            "name": name,
            "last_tick": last.isoformat() if last else None,
            "age_sec": age,
            "max_age_sec": max_age,
        })
        if last is None:
            # Never ticked: only a problem once past the boot grace window
            # (otherwise we'd 503 for the first ~3 min of every restart).
            if not in_grace:
                problems.append(f"{name}: no tick since boot ({boot_age}s ago)")
        elif age is not None and age > max_age:
            problems.append(f"{name}: stale ({age}s > {max_age}s)")

    # A follower (lost the singleton lock) runs no background tasks — surface it
    # explicitly so the 503 it already triggers has an obvious root cause.
    if not _is_leader:
        problems.append("instance: not the singleton leader (>1 worker?) — no background tasks running")

    body = {
        "status": "ok" if not problems else "degraded",
        "now": now.isoformat(),
        "is_leader": _is_leader,
        "db_ok": db_ok,
        "db_error": db_error,
        "scrapers": source_status,
        "freshest_age_sec": freshest_age,
        "price_feeds": feed_status,
        "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)