server.py

"""
Bloomberg Terminal-style Web Dashboard — FastAPI backend
Run: python server.py
Then open: http://localhost:8000
"""

import io
import json
import logging
import os
import sys
import threading
import time
from datetime import datetime, timezone
from pathlib import Path

# UTF-8 safety on Windows
if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

from dotenv import load_dotenv
load_dotenv()

import uvicorn
import secrets
from fastapi import FastAPI, Request, Header
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

import mailer

from collectors import (
    RSSCollector, RedditCollector, CoinGeckoCollector, FearGreedCollector,
    NewsDataCollector, OnChainCollector, dedupe_articles, cluster_consensus,
)
from collectors.dedupe import _normalize_title
from analyzers import (
    NarrativeDetector, EntityExtractor, TrendScorer, EnsembleAnalyzer,
    SentimentAggregator, SpikeDetector, Alerter, Scorecard,
)
from analyzers.velocity import compute_velocity, top_accelerating
from analyzers.meta_model import MetaModel, features_from_item
from analyzers.relevance import RelevanceClassifier
from analyzers.liquidation import LiquidationHeatmap
from analyzers.magnet_tracker import MagnetTracker
from analyzers.forecast import MarketImpactForecast
from collectors.economic_calendar import EconomicCalendar, fmp_configured
from analyzers.market_reaction import MarketReaction
from analyzers.embedding_narratives import EmbeddingNarratives
from monitoring import monitor
from storage import Database
import persistence
from config import DB_PATH

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)

# Persistent analyzers (loaded once, reused across cycles)
_meta_model = MetaModel()
_embed_narr = EmbeddingNarratives()

app = FastAPI(title="Crypto Narrative Terminal")
# Ensure the static dir exists (git doesn't track empty folders, so it may be
# absent on a fresh clone / cloud deploy).
Path("static").mkdir(exist_ok=True)
app.mount("/static", StaticFiles(directory="static"), name="static")

_HTML = Path("templates/index.html").read_text(encoding="utf-8")
_ADMIN_HTML = Path("templates/admin.html").read_text(encoding="utf-8")

# Restore the last DB backup (if HF persistence is configured) BEFORE opening it
persistence.restore_db(DB_PATH)
db = Database()
alerter = Alerter()

# Admin auth — simple token issued on password login (set ADMIN_PASSWORD in .env)
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin")
_admin_tokens: set[str] = set()


def _check_admin(token: str | None) -> bool:
    return bool(token and token in _admin_tokens)

# ---------------------------------------------------------------------------
# In-memory cache — updated by background thread
# ---------------------------------------------------------------------------
_cache: dict = {}
_cache_lock = threading.Lock()
_last_update: str = ""
_is_fetching: bool = False


def _maybe_train_meta_model() -> None:
    """#2 Train the meta-model on stories that now have a 24h reaction label."""
    try:
        labeled = db.labeled_reactions(5000)
        samples = []
        for r in labeled:
            lab = r.get("label_24h")
            if lab not in ("bullish", "bearish"):
                continue
            art = db.get_article(r["url"])
            if not art:
                continue
            samples.append((features_from_item(art), 1 if lab == "bullish" else 0))
        if samples:
            _meta_model.train(samples)   # no-ops if below the min sample threshold
    except Exception as exc:
        logger.debug(f"[MetaModel] training skipped: {exc}")


def run_pipeline() -> dict:
    global _is_fetching
    _is_fetching = True
    try:
        rss_items      = RSSCollector().fetch_all()
        newsdata_items = NewsDataCollector().fetch_all()
        reddit_items   = RedditCollector().fetch_all()

        cg = CoinGeckoCollector()
        trending_coins = cg.fetch_trending()
        global_market  = cg.fetch_global()
        coin_prices    = cg.fetch_prices()
        trending_ids   = cg.get_trending_ids()
        btc_now = next((c.get("price_usd") for c in coin_prices if c.get("id") == "bitcoin"), None)
        eth_now = next((c.get("price_usd") for c in coin_prices if c.get("id") == "ethereum"), None)

        fg = FearGreedCollector()
        fear_greed_history = fg.fetch()
        fear_greed_current = fg.current()

        db.save_fear_greed(fear_greed_history)
        db.save_prices(coin_prices)

        # #5 On-chain + derivatives signals
        onchain_raw = OnChainCollector().fetch(prev_oi_btc=db.prev_oi_btc())
        if onchain_raw:
            db.save_onchain(onchain_raw)
        onchain = OnChainCollector.context_summary(onchain_raw)

        all_items = rss_items + newsdata_items + reddit_items
        if not all_items:
            monitor.component_failure("collectors", "no items collected")
            return {}

        # Collapse the same story arriving from multiple feeds (keeps the most
        # credible copy) — cleaner feed + no double-counting in sentiment.
        before = len(all_items)
        all_items = dedupe_articles(all_items)
        logger.info(f"[Dedupe] {before} -> {len(all_items)} items")

        # Crypto Relevance Filter — classify CRYPTO / MACRO / IRRELEVANT and
        # DISCARD irrelevant noise (sports, entertainment, off-topic) BEFORE
        # any scoring so it never reaches the sentiment index.
        all_items = RelevanceClassifier().classify_batch(all_items)
        all_items, discarded = RelevanceClassifier.filter_relevant(all_items)
        if discarded:
            monitor.component_failure("relevance_filter", f"discarded {discarded} irrelevant")
        logger.info(f"[Relevance] kept {len(all_items)}, discarded {discarded}")
        relevance_counts = {
            "crypto": sum(1 for i in all_items if i.get("relevance") == "crypto"),
            "macro":  sum(1 for i in all_items if i.get("relevance") == "macro"),
            "discarded_irrelevant": discarded,
        }

        sa = EnsembleAnalyzer()
        all_items = sa.analyze_batch(all_items)

        nd = NarrativeDetector()
        all_items = nd.detect_batch(all_items)

        ee = EntityExtractor()
        all_items = ee.extract_batch(all_items)
        coin_mentions = ee.coin_mention_frequency(all_items)

        # #3 Consensus / contradiction across duplicate sources
        consensus = cluster_consensus(rss_items + newsdata_items + reddit_items)
        for it in all_items:
            c = consensus.get(_normalize_title(it.get("title", "")))
            if c:
                it["consensus_strength"] = c["consensus_strength"]
                it["sentiment_variance"] = c["sentiment_variance"]
                it["cluster_size"] = c["cluster_size"]
                if it.get("url"):
                    db.save_consensus(it["url"], c["cluster_size"],
                                      c["sentiment_variance"], c["consensus_strength"])

        # #6 Social attention velocity (acceleration of mentions over time)
        db.save_mentions(coin_mentions, kind="coin")
        velocity_map = compute_velocity(db.mention_series("coin", 400), coin_mentions)
        for it in all_items:
            coins = it.get("entities", {}).get("coins", [])
            vs = max((velocity_map.get(c, {}).get("velocity_score", 0.0) for c in coins),
                     default=0.0)
            it["velocity_score"] = vs
        accelerating = top_accelerating(velocity_map, 8)

        # #2 Meta-model probability of bullish reaction (per item)
        for it in all_items:
            it["bull_prob"] = round(_meta_model.predict_proba(it), 4)

        # #4 Embedding-based narrative discovery (replaces TF-IDF); fallback below
        emerging_narratives = _embed_narr.discover(all_items, top_n=12)
        if not emerging_narratives:
            monitor.model_fallback("embeddings", "using TF-IDF keywords")
            nd.fit_corpus(all_items)
            emerging_narratives = [
                {"label": kw, "size": 0, "terms": [kw], "known_narrative": None}
                for kw, _ in nd.top_emerging_keywords(all_items, top_n=12)
            ]

        scorer = TrendScorer(trending_coin_ids=trending_ids)
        scored_narratives = scorer.score_narratives(all_items)

        # Spike detection vs prior snapshots (before saving the new one, so
        # the current run is compared against history, not itself).
        prior_snaps = db.get_snapshots(limit=12)
        spike_alerts = SpikeDetector(prior_snaps).detect(scored_narratives)
        if spike_alerts:
            alerter.push(spike_alerts)   # external channels if configured

        # 24h aggregate sentiment — merge fresh items with cached recent ones
        # so the prior-24h trend comparison has history to work with.
        db.save_articles(all_items)
        db.save_snapshot(scored_narratives)

        # #1 Market reaction labeling: seed new stories, fill matured ones
        mr = MarketReaction()
        mr.seed(db, all_items, btc_now, eth_now)
        mr.fill(db)

        # #2 Retrain meta-model on accumulated 24h-labelled reactions
        _maybe_train_meta_model()

        recent_cached = db.get_recent_articles(hours=48)
        # Reliable "vs ~24h ago" using the real stored snapshot, not stale articles
        prior_snapshot = db.get_index_near(hours_ago=24, tolerance_hours=8)
        sentiment_24h = SentimentAggregator().aggregate(recent_cached or all_items, prior_snapshot)
        # #7 nudge volatility probability with the derivatives liquidation proxy
        if onchain.get("liq_proxy") is not None:
            ov = sentiment_24h.get("overall", {})
            extra = min(0.3, onchain["liq_proxy"] / 40.0)
            ov["vol_prob"] = round(min(1.0, ov.get("vol_prob", 0.0) + extra), 4)
        db.save_sentiment_history(sentiment_24h)
        sentiment_history = db.get_sentiment_history(limit=96)
        # Accuracy scorecard: did our directional calls match BTC's next-24h move?
        btc_history = db.get_price_history("bitcoin", limit=200)
        scorecard = Scorecard().compute(sentiment_history, btc_history)

        # Liquidation heatmap (estimated from OI + leverage tiers vs recent prices).
        # Fall back to the last stored BTC price if CoinGecko was rate-limited.
        liq_price = btc_now or (btc_history[-1]["price"] if btc_history else None)
        liquidation = LiquidationHeatmap().compute(
            liq_price, btc_history, oi_btc=(onchain_raw or {}).get("oi_btc"))

        # Magnet target tracker: hit-check + expire + register (every cycle),
        # each level tracked for 24h, never re-registered while active.
        _mt = MagnetTracker()
        _mt.update(db, liquidation, liq_price)
        magnet_scorecard = _mt.compute(db)

        # Economic calendar (FMP) + Market Impact Forecast for the next 24-48h
        calendar = EconomicCalendar().fetch(days_ahead=7)
        forecast = MarketImpactForecast().compute(
            sentiment_24h=sentiment_24h, onchain=onchain, liquidation=liquidation,
            calendar=calendar, fear_greed=fear_greed_current)

        # #8 flush system-health events for this cycle
        monitor.flush(db)
        health = db.health_summary(24)
        db.purge_old()

        # Persist the DB to the HF Dataset backup (throttled; no-op if unset)
        persistence.backup_db(db)

        # Build news feed (most recent 60 items with title + source)
        news_feed = sorted(
            [i for i in all_items if i.get("title")],
            key=lambda x: x.get("published", ""),
            reverse=True,
        )[:60]

        result = {
            "narratives":        scored_narratives,
            "global_market":     global_market,
            "fear_greed":        fear_greed_current,
            "fear_greed_history": fear_greed_history,
            "trending_coins":    trending_coins,
            "coin_prices":       coin_prices,
            "coin_mentions":     coin_mentions,
            "emerging_narratives": emerging_narratives,
            "accelerating":      accelerating,
            "onchain":           onchain,
            "liquidation":       liquidation,
            "magnet_scorecard":  magnet_scorecard,
            "calendar":          calendar,
            "forecast":          forecast,
            "fmp_configured":    fmp_configured(),
            "health":            health,
            "relevance_counts":  relevance_counts,
            "meta_model_trained": _meta_model.trained,
            "article_count":     len(all_items),
            "sentiment_24h":     sentiment_24h,
            "sentiment_history": sentiment_history,
            "spike_alerts":      spike_alerts,
            "scorecard":         scorecard,
            "news_feed":         [
                {
                    "title":     n.get("title", "")[:120],
                    "source":    n.get("source", ""),
                    "url":       n.get("url", ""),
                    "published": n.get("published", ""),
                    "sentiment": n.get("sentiment", {}).get("label", "neutral"),
                    "compound":  n.get("sentiment", {}).get("compound", 0),
                    "confidence": n.get("sentiment", {}).get("confidence", 0),
                    "engine":    n.get("sentiment", {}).get("engine", "vader"),
                    "bull_prob": n.get("bull_prob", 0.5),
                    "consensus_strength": n.get("consensus_strength"),
                    "narratives": n.get("narratives", []),
                    "is_macro":  bool(n.get("is_macro")),
                    "relevance": n.get("relevance", "crypto"),
                }
                for n in news_feed
            ],
            "fetch_time": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
        }
        return result
    except Exception as exc:
        logger.exception("Pipeline failed")
        return {"error": str(exc)}
    finally:
        _is_fetching = False


REFRESH_MINUTES = int(os.getenv("REFRESH_INTERVAL_MINUTES", "10"))


def refresh_loop():
    while True:
        data = run_pipeline()
        if data:
            with _cache_lock:
                _cache.update(data)
                global _last_update
                _last_update = datetime.now(timezone.utc).strftime("%H:%M:%S")
        time.sleep(max(3, REFRESH_MINUTES) * 60)   # default every 10 min (configurable)


def _send_brief_now(slot_label: str = "manual") -> tuple[bool, str]:
    """Build the current brief and email it to all active clients."""
    with _cache_lock:
        data = dict(_cache)
    if not data or not data.get("sentiment_24h"):
        # Cache not warmed yet (e.g. just after a restart). Kick off a fetch
        # so the next attempt succeeds, and tell the user clearly.
        if not _is_fetching:
            threading.Thread(target=lambda: _cache.update(run_pipeline()), daemon=True).start()
        return False, "Data still loading — wait ~2 min after restart, then try again"
    settings = db.get_email_settings()
    recipients = [c["email"] for c in db.get_clients(active_only=True)]
    if not recipients:
        db.log_email(slot_label, 0, "skipped", "no active clients")
        return False, "No active clients"
    html = mailer.build_brief_html(data)
    idx = data["sentiment_24h"]["overall"].get("index", "")
    label = data["sentiment_24h"]["overall"].get("label", "")
    subject = f"{settings.get('subject_prefix','Crypto Narrative Brief')} — {label} ({idx}/100)"
    ok, detail = mailer.send_brief(recipients, subject, html)
    db.log_email(slot_label, len(recipients), "sent" if ok else "failed", detail)
    return ok, detail


def email_scheduler():
    """Checks every 30s whether a scheduled send time has arrived (UTC)."""
    while True:
        try:
            settings = db.get_email_settings()
            if settings.get("enabled"):
                now = datetime.now(timezone.utc)
                hhmm = now.strftime("%H:%M")
                today = now.strftime("%Y-%m-%d")
                for slot_num, key in ((1, "send_time_1"), (2, "send_time_2")):
                    if settings.get(key) == hhmm:
                        slot_id = f"{today}#{slot_num}"
                        if settings.get("last_sent_slot") != slot_id:
                            ok, detail = _send_brief_now(slot_id)
                            db.update_email_settings(last_sent_slot=slot_id)
                            logger.info(f"[Scheduler] slot {slot_id}: {ok} {detail}")
        except Exception as exc:
            logger.warning(f"[Scheduler] error: {exc}")
        time.sleep(30)


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------

@app.get("/health")
async def health():
    return {"status": "ok"}

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(a: FastAPI):
    threading.Thread(target=refresh_loop, daemon=True).start()
    threading.Thread(target=email_scheduler, daemon=True).start()
    yield

app.router.lifespan_context = lifespan


@app.get("/", response_class=HTMLResponse)
async def index():
    return HTMLResponse(_HTML)


@app.get("/api/data")
async def get_data():
    with _cache_lock:
        if not _cache:
            return JSONResponse({"status": "loading", "is_fetching": _is_fetching})
        return JSONResponse({**_cache, "last_update": _last_update, "is_fetching": _is_fetching})


@app.get("/api/refresh")
async def trigger_refresh():
    """Manually trigger a data refresh in background."""
    if not _is_fetching:
        t = threading.Thread(target=lambda: _cache.update(run_pipeline()), daemon=True)
        t.start()
    return {"status": "refreshing"}


# ---------------------------------------------------------------------------
# Admin panel
# ---------------------------------------------------------------------------

@app.get("/admin", response_class=HTMLResponse)
async def admin_page():
    return HTMLResponse(_ADMIN_HTML)


@app.post("/api/admin/login")
async def admin_login(payload: dict):
    if payload.get("password") == ADMIN_PASSWORD:
        token = secrets.token_urlsafe(24)
        _admin_tokens.add(token)
        return {"ok": True, "token": token}
    return JSONResponse({"ok": False, "error": "Wrong password"}, status_code=401)


def _auth_or_401(token: str | None):
    if not _check_admin(token):
        return JSONResponse({"ok": False, "error": "Unauthorized"}, status_code=401)
    return None


@app.get("/api/admin/state")
async def admin_state(x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    return {
        "ok": True,
        "clients":  db.get_clients(),
        "settings": db.get_email_settings(),
        "log":      db.get_email_log(20),
        "smtp_configured": mailer.email_configured(),
        "email_provider": "Brevo (HTTP)" if mailer.brevo_configured() else ("SMTP" if mailer.smtp_configured() else "none"),
        "has_data": bool(_cache.get("sentiment_24h")),
    }


@app.post("/api/admin/clients")
async def admin_add_client(payload: dict, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    email = (payload.get("email") or "").strip()
    if not email or "@" not in email:
        return JSONResponse({"ok": False, "error": "Invalid email"}, status_code=400)
    ok = db.add_client(email, payload.get("name", ""))
    return {"ok": ok, "error": None if ok else "Already exists"}


@app.delete("/api/admin/clients/{client_id}")
async def admin_remove_client(client_id: int, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    db.remove_client(client_id)
    return {"ok": True}


@app.post("/api/admin/clients/{client_id}/toggle")
async def admin_toggle_client(client_id: int, payload: dict, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    db.set_client_active(client_id, bool(payload.get("active")))
    return {"ok": True}


@app.post("/api/admin/settings")
async def admin_settings(payload: dict, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    db.update_email_settings(
        enabled=1 if payload.get("enabled") else 0,
        send_time_1=payload.get("send_time_1", "09:00"),
        send_time_2=payload.get("send_time_2", "21:00"),
        subject_prefix=payload.get("subject_prefix", "Crypto Narrative Brief"),
    )
    return {"ok": True, "settings": db.get_email_settings()}


@app.post("/api/admin/send-now")
async def admin_send_now(x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    ok, detail = _send_brief_now("manual")
    return {"ok": ok, "detail": detail}


# ---------------------------------------------------------------------------
# Cron endpoints — hit by an external free scheduler (GitHub Actions) so
# briefs fire and the host stays awake even on sleepy free tiers.
# Secured by a shared secret (CRON_SECRET in .env).
# ---------------------------------------------------------------------------
CRON_SECRET = os.getenv("CRON_SECRET", "")


@app.get("/api/cron/ping")
async def cron_ping(key: str = ""):
    """Keep-alive + ensure data is fresh. Safe to call frequently."""
    if CRON_SECRET and key != CRON_SECRET:
        return JSONResponse({"ok": False, "error": "bad key"}, status_code=403)
    if not _is_fetching and not _cache:
        threading.Thread(target=lambda: _cache.update(run_pipeline()), daemon=True).start()
    return {"ok": True, "awake": True, "has_data": bool(_cache.get("sentiment_24h"))}


@app.get("/api/cron/send-brief")
async def cron_send_brief(key: str = ""):
    """Send the brief now to all active clients (called at scheduled times)."""
    if CRON_SECRET and key != CRON_SECRET:
        return JSONResponse({"ok": False, "error": "bad key"}, status_code=403)
    ok, detail = _send_brief_now("cron")
    return {"ok": ok, "detail": detail}


if __name__ == "__main__":
    print("\n  Crypto Narrative Terminal")
    print("  Open http://localhost:8000 in your browser\n")
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="warning")