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