"""Daily 3-section Telegram digest — Macro Vibes / KOL / Trump. Sent to every TelegramBinding where digest_enabled=True at their chosen digest_hour_utc. Independent of the per-signal real-time alert path (telegram._dispatch) — a user can turn one off and the other on. Architecture: cron (hourly :00) ↓ send_daily_digest(current_hour) ↓ build_global_digest() ← runs ONCE, content is global ↓ for each binding where digest_hour_utc == current_hour: format_digest(global, user_state) ← appends per-user line send_message(chat_id, text) update last_digest_sent_at The body is rule-based (no LLM). Each section reads one or two structured fields and picks a templated sentence. If we later want richer prose, the single place to hook DeepSeek/Claude is at the bottom of `format_digest` — everything else stays the same. Idempotency: the SELECT filters on last_digest_sent_at IS NULL OR last_digest_sent_at < now - 23h so a coalesced cron or worker restart can't double-send within a day. """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Optional from sqlalchemy import select, update, func, or_, and_ from app.config import settings from app.database import AsyncSessionLocal as async_session from app.models import ( TelegramBinding, MacroSnapshot, KolPost, KolDivergence, Post, BotTrade, Subscription, ) from app.services.telegram import send_message logger = logging.getLogger(__name__) # ── Data model ──────────────────────────────────────────────────────────── @dataclass class MacroBlock: available: bool composite: Optional[float] regime: Optional[str] # BULL / BULLISH / NEUTRAL / BEARISH / BEAR ahr999: Optional[float] fear_greed: Optional[int] bottom_signal_firing: bool # any Post(source=btc_bottom_reversal, last 24h) bottom_votes: Optional[int] # 2 or 3 if firing, else None funding_signal_firing: bool # any Post(source=funding_reversal, last 24h) @dataclass class KolBlock: posts_24h: int bullish: int bearish: int neutral: int divergences_24h: int sample_divergence: Optional[str] # one human-readable line, or None @dataclass class TrumpBlock: posts_24h: int actionable: list # list of (asset, side, confidence) @dataclass class GlobalDigest: """Same content for everyone. Per-user state (auto-trade, your trades) is appended in format_digest.""" date_label: str # "May 26 · Mon" macro: MacroBlock kol: KolBlock trump: TrumpBlock headline_emoji: str # 🟢 / 🟡 / ⚪ headline_text: str # "BTC bottom signal firing. Worth a look." # ── Builder ─────────────────────────────────────────────────────────────── async def build_global_digest(now: Optional[datetime] = None) -> GlobalDigest: """Single DB pass over the three sources. Pure read — never raises; returns a degraded digest if any source is missing.""" now = now or datetime.now(timezone.utc) now_naive = now.replace(tzinfo=None) cutoff_24h = now_naive - timedelta(hours=24) date_label = now.strftime("%b %-d · %a") async with async_session() as db: # ── Macro: today's row (or yesterday's if today's not in yet) ──── today = now.date() yesterday = today - timedelta(days=1) snap = (await db.execute( select(MacroSnapshot) .where(MacroSnapshot.snapshot_date.in_([today, yesterday])) .order_by(MacroSnapshot.snapshot_date.desc()) )).scalars().first() # Bottom + funding triggers in last 24h bottom_post = (await db.execute( select(Post) .where(Post.source == "btc_bottom_reversal", Post.created_at >= cutoff_24h) .order_by(Post.created_at.desc()) )).scalars().first() funding_post = (await db.execute( select(Post) .where(Post.source == "funding_reversal", Post.created_at >= cutoff_24h) )).scalars().first() macro = MacroBlock( available=snap is not None, composite=snap.composite_score if snap else None, regime=snap.regime_label if snap else None, ahr999=snap.ahr999 if snap else None, fear_greed=snap.fear_greed if snap else None, bottom_signal_firing=bottom_post is not None, bottom_votes=(bottom_post.ai_confidence and (3 if bottom_post.ai_confidence >= 90 else 2)) if bottom_post else None, funding_signal_firing=funding_post is not None, ) # ── KOL: last 24h posts + divergences ──────────────────────────── kol_posts = (await db.execute( select(KolPost).where(KolPost.published_at >= cutoff_24h) )).scalars().all() # action lives inside tickers_json; easier: classify on summary text. # The structured action is on KolDivergence rows; for the bare count # we just split posts by whether divergence rows reference them. bullish = bearish = neutral = 0 divergence_rows = (await db.execute( select(KolDivergence) .where(KolDivergence.post_at >= cutoff_24h) .order_by(KolDivergence.created_at.desc()) )).scalars().all() for d in divergence_rows: if d.direction == "long": bullish += 1 elif d.direction == "short": bearish += 1 else: neutral += 1 # Posts not yet classified into divergences just count as neutral. neutral += max(0, len(kol_posts) - (bullish + bearish + neutral)) divergences_24h = sum( 1 for d in divergence_rows if d.signal_type == "divergence" ) sample_divergence: Optional[str] = None for d in divergence_rows: if d.signal_type != "divergence": continue # "Hayes publicly bullish BTC, on-chain selling" sample_divergence = ( f"{d.handle} publicly {d.post_action} {d.ticker}, " f"on-chain {d.onchain_action}" ) break kol = KolBlock( posts_24h=len(kol_posts), bullish=bullish, bearish=bearish, neutral=neutral, divergences_24h=divergences_24h, sample_divergence=sample_divergence, ) # ── Trump: last 24h posts, list actionable ones ────────────────── trump_posts = (await db.execute( select(Post) .where(Post.source == "truth", Post.created_at >= cutoff_24h) .order_by(Post.created_at.desc()) )).scalars().all() actionable = [] for p in trump_posts: if p.signal in ("buy", "short") and (p.ai_confidence or 0) >= 70: actionable.append(( p.target_asset or "?", "LONG" if p.signal == "buy" else "SHORT", int(p.ai_confidence or 0), )) trump = TrumpBlock(posts_24h=len(trump_posts), actionable=actionable) headline_emoji, headline_text = _headline(macro, kol, trump) return GlobalDigest( date_label=date_label, macro=macro, kol=kol, trump=trump, headline_emoji=headline_emoji, headline_text=headline_text, ) def _headline(macro: MacroBlock, kol: KolBlock, trump: TrumpBlock) -> tuple[str, str]: """Pick a one-liner verdict + emoji from the three sections. Priority: macro bottom > macro funding > Trump actionable > KOL divergence > nothing. """ if macro.bottom_signal_firing: return "🟢", "BTC bottom signal firing. Worth a look." if macro.funding_signal_firing: return "🟢", "Funding extreme — mean-reversion setup forming." if len(trump.actionable) > 0: n = len(trump.actionable) return "🟢", f"Trump posted {n} actionable crypto signal{'s' if n > 1 else ''} today." if kol.divergences_24h > 0: return "🟡", f"{kol.divergences_24h} KOL talks-vs-trades divergence today." return "⚪", "Quiet day, nothing to act on." # ── Formatter ───────────────────────────────────────────────────────────── def format_digest(g: GlobalDigest, user_state: Optional[dict] = None) -> str: """Render the global digest into a single HTML message. `user_state` is a small dict built per-binding by send_daily_digest: { "wallet": str | None, "auto_trade": bool, "trades_today": int, "open_pnl_summary": str | None } None or empty → free-tier user, the Pro status line is suppressed. """ lines: list[str] = [] lines.append("📰 Trump Alpha · Daily Brief") lines.append(f"{g.date_label}") lines.append("") lines.append(f"{g.headline_emoji} Today: {g.headline_text}") lines.append("") # ── Macro ──────────────────────────────────────────────────────────── lines.append("━ Macro ━") m = g.macro if not m.available: lines.append("Macro snapshot unavailable today.") elif m.bottom_signal_firing: votes = m.bottom_votes or 2 lines.append(f"BTC bottom triggers: {votes}/3 firing.") if m.ahr999 is not None: lines.append(f"AHR999={m.ahr999:.2f}, F&G={m.fear_greed or '?'}.") elif m.funding_signal_firing: lines.append("Funding-rate extreme — mean-reversion alert active.") else: # Plain day: one sentence + 1-2 backing numbers regime_phrase = { "BULL": "strongly bullish", "BULLISH": "leaning bullish", "NEUTRAL": "in neutral zone, far from bottom", "BEARISH": "leaning bearish", "BEAR": "strongly bearish", }.get((m.regime or "NEUTRAL").upper(), "in neutral zone") lines.append(f"BTC {regime_phrase}.") bits = [] if m.ahr999 is not None: bits.append(f"AHR999={m.ahr999:.2f} (bottom <0.45)") if m.fear_greed is not None: bits.append(f"F&G={m.fear_greed}") if bits: lines.append(", ".join(bits) + ".") lines.append("") # ── KOL ────────────────────────────────────────────────────────────── lines.append("━ KOL ━") k = g.kol if k.posts_24h == 0: lines.append("No KOL posts in last 24h.") else: lines.append( f"{k.posts_24h} posts today: {k.bullish} bullish / " f"{k.bearish} bearish." ) if k.divergences_24h > 0 and k.sample_divergence: lines.append(f"⚠️ {k.sample_divergence}.") elif k.divergences_24h == 0: lines.append("No talks-vs-trades divergence.") lines.append("") # ── Trump ──────────────────────────────────────────────────────────── lines.append("━ Trump ━") t = g.trump if t.posts_24h == 0: lines.append("No posts in last 24h.") elif not t.actionable: lines.append(f"{t.posts_24h} posts today, none crypto-actionable.") else: lines.append( f"{t.posts_24h} posts today, " f"{len(t.actionable)} actionable:" ) for asset, side, conf in t.actionable[:3]: emoji = "🟢" if side == "LONG" else "🔴" lines.append(f" {emoji} {asset} · {side} · conf {conf}") if len(t.actionable) > 3: lines.append(f" …and {len(t.actionable) - 3} more.") lines.append("") # ── Per-user state (Pro only) ──────────────────────────────────────── if user_state and user_state.get("wallet"): at = "ON" if user_state.get("auto_trade") else "OFF" td = int(user_state.get("trades_today") or 0) bits = [f"auto-trade {at}", f"{td} trade{'s' if td != 1 else ''} today"] if user_state.get("open_pnl_summary"): bits.append(user_state["open_pnl_summary"]) lines.append("Your status: " + " · ".join(bits)) lines.append("") # ── Footer ─────────────────────────────────────────────────────────── fe = (settings.frontend_url or "").rstrip("/") if fe: lines.append(f'→ open dashboard') lines.append("") lines.append("Daily brief · reply /digest off to unsubscribe") return "\n".join(lines).strip() # ── Sender ──────────────────────────────────────────────────────────────── async def _build_user_state(db, binding: TelegramBinding, now_naive: datetime) -> dict: """Pro-only: fetch auto-trade flag + today's trade count for the wallet attached to this binding. Returns {} for free users.""" if not binding.wallet_address: return {} sub = (await db.execute( select(Subscription).where(Subscription.wallet_address == binding.wallet_address) )).scalar_one_or_none() if sub is None: return {} start_of_day = now_naive.replace(hour=0, minute=0, second=0, microsecond=0) trades_today = (await db.execute( select(func.count(BotTrade.id)).where( BotTrade.wallet_address == binding.wallet_address, BotTrade.opened_at >= start_of_day, ) )).scalar() or 0 # Open positions' aggregated unrealised — best-effort, skip on any error. open_pnl_summary: Optional[str] = None try: # Released trades aren't bot-managed any more; their unrealised # PnL belongs to the user's manual control, not to "your bot status". open_trades = (await db.execute( select(BotTrade).where( BotTrade.wallet_address == binding.wallet_address, BotTrade.closed_at.is_(None), BotTrade.released_at.is_(None), ) )).scalars().all() if open_trades: from app.services.price_store import price_store unreal = 0.0 for t in open_trades: px = price_store.latest_price(t.asset) if not px or not t.entry_price or not t.size_usd: continue pct = (px - t.entry_price) / t.entry_price signed = pct if t.side == "long" else -pct rem = t.remaining_fraction if t.remaining_fraction is not None else 1.0 unreal += t.size_usd * rem * signed if open_trades: sign = "+" if unreal >= 0 else "" open_pnl_summary = f"{len(open_trades)} open ({sign}{unreal:.1f} USD)" except Exception as exc: logger.debug("digest user_state: skip open pnl: %s", exc) return { "wallet": binding.wallet_address, "auto_trade": bool(sub.auto_trade), "trades_today": int(trades_today), "open_pnl_summary": open_pnl_summary, } async def send_daily_digest(current_hour: Optional[int] = None) -> int: """Send the digest to every binding whose digest_hour_utc matches the current UTC hour AND who hasn't received one in the last 23 hours. Returns the number of messages successfully sent. Never raises — the cron firing must not bring the worker down. """ if not settings.telegram_bot_token: return 0 now = datetime.now(timezone.utc) hour = current_hour if current_hour is not None else now.hour now_naive = now.replace(tzinfo=None) dedupe_cutoff = now_naive - timedelta(hours=23) # Build the body ONCE — it's identical for every user (only per-user # state line varies). One DB pass instead of one per binding. try: global_digest = await build_global_digest(now) except Exception as exc: logger.exception("Digest: build_global_digest failed: %s", exc) return 0 sent = 0 async with async_session() as db: bindings = (await db.execute( select(TelegramBinding).where( TelegramBinding.digest_enabled.is_(True), TelegramBinding.alerts_enabled.is_(True), TelegramBinding.digest_hour_utc == hour, or_( TelegramBinding.last_digest_sent_at.is_(None), TelegramBinding.last_digest_sent_at < dedupe_cutoff, ), ) )).scalars().all() if not bindings: return 0 for b in bindings: try: user_state = await _build_user_state(db, b, now_naive) text = format_digest(global_digest, user_state) ok = await send_message(b.chat_id, text) if ok: await db.execute( update(TelegramBinding) .where(TelegramBinding.id == b.id) .values(last_digest_sent_at=now_naive) ) sent += 1 except Exception as exc: logger.warning("Digest send failed for chat=%s: %s", b.chat_id, exc) await db.commit() logger.info("Daily digest: hour=%dz sent=%d/%d", hour, sent, len(bindings)) return sent # ── Manual preview (called from bot /digest command) ────────────────────── async def send_preview_for(chat_id: int) -> bool: """Render and immediately send a digest to one chat. Does NOT update last_digest_sent_at — preview is independent of the daily slot.""" if not settings.telegram_bot_token: return False try: g = await build_global_digest() except Exception as exc: logger.exception("Digest preview: build failed: %s", exc) return await send_message( chat_id, "⚠️ Digest preview unavailable — check back shortly.") async with async_session() as db: b = (await db.execute( select(TelegramBinding).where(TelegramBinding.chat_id == chat_id) )).scalar_one_or_none() user_state = (await _build_user_state(db, b, datetime.utcnow()) if b else {}) text = format_digest(g, user_state) return await send_message(chat_id, text)