k
54884f3e24
Feed-health pass over KOL_FEEDS: - raoulpal: stale Substack (last 2024-05) → Real Vision podcast feed - dampedspring: paywalled (0 entries) → free "Damped Spring 101" Substack - unchained: Cloudflare 403 → canonical Megaphone podcast feed - lynalden: Cloudflare 202 → FeedBurner mirror - glassnode: recovered via httpx http2=True (was 403 on HTTP/1.1) - browser User-Agent + Accept headers on feed fetch - removed dead feeds with no active replacement: placeholder, dragonfly, niccarter, eugene - pin h2==4.3.0 (required by http2=True) All 25 remaining feeds verified fetching real body content; newest post per feed ≤88d. Bundles in-flight KOL-module work already in the working tree (kol_x ingest, migration 027, tests). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
497 lines
21 KiB
Python
497 lines
21 KiB
Python
"""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 html
|
||
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 ────────────────────────────
|
||
# Exclude twitter noise (gm / RT / jokes) from the count. Substack
|
||
# posts have tier=NULL (kept); twitter signal posts have tier!='noise'
|
||
# (kept); twitter noise is dropped. Without this filter a KOL's gm/RT
|
||
# spam inflates the daily "N KOL posts" stat. Mirrors the /kol/posts
|
||
# signals_only filter and the frontend.
|
||
kol_posts = (await db.execute(
|
||
select(KolPost).where(
|
||
KolPost.published_at >= cutoff_24h,
|
||
(KolPost.tier.is_(None)) | (KolPost.tier != "noise"),
|
||
)
|
||
)).scalars().all()
|
||
# Count bullish/bearish/neutral at POST level, not divergence-pair level.
|
||
# A single post can match multiple on-chain events (one per ticker/wallet),
|
||
# producing several KolDivergence rows. Counting rows would inflate the
|
||
# numbers — a post mentioning 3 tickers was previously counted 3× (bug).
|
||
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()
|
||
# Deduplicate by post_id; for posts with mixed directions take "long"
|
||
# first (a bullish call that also hedges short is net bullish).
|
||
post_direction: dict[int, str] = {}
|
||
for d in divergence_rows:
|
||
if d.post_id not in post_direction:
|
||
post_direction[d.post_id] = d.direction
|
||
elif d.direction == "long":
|
||
post_direction[d.post_id] = "long" # long overrides neutral
|
||
for direction in post_direction.values():
|
||
if direction == "long":
|
||
bullish += 1
|
||
elif direction == "short":
|
||
bearish += 1
|
||
else:
|
||
neutral += 1
|
||
# Posts not yet cross-referenced with on-chain data 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"
|
||
# Escape — ticker is AI-extracted (kol_analysis) so not guaranteed
|
||
# HTML-safe; the digest is sent with parse_mode=HTML.
|
||
sample_divergence = html.escape(
|
||
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("📰 <b>Trump Alpha · Daily Brief</b>")
|
||
lines.append(f"<i>{g.date_label}</i>")
|
||
lines.append("")
|
||
lines.append(f"{g.headline_emoji} <b>Today:</b> {g.headline_text}")
|
||
lines.append("")
|
||
|
||
# ── Macro ────────────────────────────────────────────────────────────
|
||
lines.append("<b>━ Macro ━</b>")
|
||
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: <b>{votes}/3 firing</b>.")
|
||
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("<b>━ KOL ━</b>")
|
||
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: <b>{k.bullish}</b> bullish / "
|
||
f"<b>{k.bearish}</b> 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("<b>━ Trump ━</b>")
|
||
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"<b>{len(t.actionable)}</b> 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("<b>Your status:</b> " + " · ".join(bits))
|
||
lines.append("")
|
||
|
||
# ── Footer ───────────────────────────────────────────────────────────
|
||
fe = (settings.frontend_url or "").rstrip("/")
|
||
if fe:
|
||
lines.append(f'<a href="{fe}/en/macro">→ open dashboard</a>')
|
||
lines.append("")
|
||
lines.append("<i>Daily brief · reply /digest off to unsubscribe</i>")
|
||
|
||
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:
|
||
# NOTE: alerts_enabled is intentionally NOT gated here. /stop pauses
|
||
# real-time signal alerts but the daily digest is a separate feature
|
||
# (/digest on|off). Filtering on alerts_enabled would silently kill
|
||
# the digest for any user who typed /stop, contradicting the bot's
|
||
# own message ("Alerts paused — send /start to re-enable").
|
||
bindings = (await db.execute(
|
||
select(TelegramBinding).where(
|
||
TelegramBinding.digest_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)
|