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>
224 lines
8.3 KiB
Python
224 lines
8.3 KiB
Python
"""Snapshot-style tests for the daily digest formatter.
|
|
|
|
These hit format_digest with hand-rolled GlobalDigest objects so we exercise
|
|
every branch (quiet day / bottom trigger / actionable Trump / divergence /
|
|
empty all-around) without standing up a DB. The actual build_global_digest
|
|
SQL is exercised by integration tests / manual /digest in the bot.
|
|
|
|
The point of these tests is to lock the user-facing wording — if a future
|
|
change accidentally turns "BTC bottom triggers: 3/3 firing" into a less
|
|
clear phrasing, CI catches it.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from app.services.telegram_digest import (
|
|
GlobalDigest, MacroBlock, KolBlock, TrumpBlock, format_digest,
|
|
)
|
|
|
|
|
|
def _empty_blocks():
|
|
return (
|
|
MacroBlock(available=False, composite=None, regime=None,
|
|
ahr999=None, fear_greed=None,
|
|
bottom_signal_firing=False, bottom_votes=None,
|
|
funding_signal_firing=False),
|
|
KolBlock(posts_24h=0, bullish=0, bearish=0, neutral=0,
|
|
divergences_24h=0, sample_divergence=None),
|
|
TrumpBlock(posts_24h=0, actionable=[]),
|
|
)
|
|
|
|
|
|
def test_quiet_day_renders_compact_and_mentions_no_activity():
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.regime = "NEUTRAL"
|
|
m.ahr999 = 0.62
|
|
m.fear_greed = 38
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="⚪",
|
|
headline_text="Quiet day, nothing to act on.")
|
|
out = format_digest(g)
|
|
assert "Trump Alpha · Daily Brief" in out
|
|
assert "⚪" in out
|
|
assert "Quiet day" in out
|
|
assert "neutral zone" in out
|
|
assert "AHR999=0.62" in out
|
|
assert "No KOL posts" in out
|
|
assert "No posts in last 24h" in out
|
|
# No per-user Pro line for free users
|
|
assert "Your status" not in out
|
|
# Always ends with the unsubscribe hint
|
|
assert "/digest off" in out
|
|
|
|
|
|
def test_bottom_trigger_day_promotes_macro_section_and_keeps_short():
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.bottom_signal_firing = True
|
|
m.bottom_votes = 3
|
|
m.ahr999 = 0.41
|
|
m.fear_greed = 14
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="🟢",
|
|
headline_text="BTC bottom signal firing. Worth a look.")
|
|
out = format_digest(g)
|
|
assert "🟢" in out
|
|
assert "BTC bottom signal firing" in out
|
|
# Macro section shows the trigger detail
|
|
assert "3/3 firing" in out
|
|
assert "AHR999=0.41" in out
|
|
|
|
|
|
def test_trump_actionable_list_is_capped_at_three_with_summary_overflow():
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.regime = "NEUTRAL"
|
|
t.posts_24h = 6
|
|
t.actionable = [
|
|
("BTC", "LONG", 92),
|
|
("SOL", "LONG", 81),
|
|
("ETH", "SHORT", 76),
|
|
("DOGE", "LONG", 71), # 4th — should be summarised
|
|
]
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="🟢",
|
|
headline_text="Trump posted 4 actionable crypto signals today.")
|
|
out = format_digest(g)
|
|
assert "BTC · LONG · conf 92" in out
|
|
assert "SOL · LONG · conf 81" in out
|
|
assert "ETH · SHORT · conf 76" in out
|
|
# The fourth must roll into the "and N more" line, not be listed verbatim
|
|
assert "DOGE" not in out
|
|
assert "1 more" in out
|
|
|
|
|
|
def test_divergence_warning_surfaces_sample_line():
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.regime = "NEUTRAL"
|
|
k.posts_24h = 14
|
|
k.bullish = 9
|
|
k.bearish = 1
|
|
k.divergences_24h = 1
|
|
k.sample_divergence = "ArthurHayes publicly bullish BTC, on-chain decreased"
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="🟡",
|
|
headline_text="1 KOL talks-vs-trades divergence today.")
|
|
out = format_digest(g)
|
|
assert "⚠️ ArthurHayes" in out
|
|
assert "9</b> bullish" in out
|
|
|
|
|
|
def test_pro_user_state_appended_for_wallet_bound():
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.regime = "NEUTRAL"
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="⚪", headline_text="Quiet day, nothing to act on.")
|
|
out = format_digest(g, user_state={
|
|
"wallet": "0xabc",
|
|
"auto_trade": True,
|
|
"trades_today": 2,
|
|
"open_pnl_summary": "1 open (+12.4 USD)",
|
|
})
|
|
assert "Your status" in out
|
|
assert "auto-trade ON" in out
|
|
assert "2 trades today" in out
|
|
assert "1 open (+12.4 USD)" in out
|
|
|
|
|
|
def test_pro_state_suppressed_for_free_user():
|
|
m, k, t = _empty_blocks()
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="⚪", headline_text="Quiet day, nothing to act on.")
|
|
out_free = format_digest(g, user_state={})
|
|
out_no_wallet = format_digest(g, user_state={"wallet": None})
|
|
assert "Your status" not in out_free
|
|
assert "Your status" not in out_no_wallet
|
|
|
|
|
|
def test_html_safe_against_ampersand_in_label():
|
|
"""The 'F&G' label must always emit as F&G so Telegram's HTML
|
|
parser doesn't reject the message."""
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.regime = "NEUTRAL"
|
|
m.ahr999 = 0.62
|
|
m.fear_greed = 38
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="⚪", headline_text="Quiet day, nothing to act on.")
|
|
out = format_digest(g)
|
|
assert "F&G=38" in out
|
|
assert "F&G=38" not in out
|
|
|
|
|
|
def test_total_length_well_under_telegram_cap():
|
|
"""Even the busiest realistic day must fit in one Telegram message
|
|
(4096 chars)."""
|
|
m, k, t = _empty_blocks()
|
|
m.available = True
|
|
m.bottom_signal_firing = True
|
|
m.bottom_votes = 3
|
|
m.ahr999 = 0.41
|
|
m.fear_greed = 14
|
|
k.posts_24h = 30
|
|
k.bullish = 20
|
|
k.bearish = 8
|
|
k.divergences_24h = 4
|
|
k.sample_divergence = "Hayes publicly bullish ETH, on-chain decreased"
|
|
t.posts_24h = 12
|
|
t.actionable = [("BTC", "LONG", 95), ("SOL", "LONG", 88), ("ETH", "SHORT", 80)]
|
|
g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t,
|
|
headline_emoji="🟢", headline_text="Lots happening today.")
|
|
out = format_digest(g, user_state={
|
|
"wallet": "0xabc", "auto_trade": True, "trades_today": 4,
|
|
"open_pnl_summary": "3 open (+125.3 USD)",
|
|
})
|
|
assert len(out) < 4096
|
|
|
|
|
|
# ── build_global_digest DB aggregation (twitter-noise exclusion) ───────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_global_digest_excludes_twitter_noise(monkeypatch):
|
|
"""KOL posts_24h must NOT count twitter noise (gm / RT / jokes). Substack
|
|
essays (tier=NULL) and twitter signal posts (tier!='noise') count; twitter
|
|
noise (tier='noise') is excluded. Regression guard for the digest count
|
|
being inflated by a KOL's gm/RT spam after X ingestion landed."""
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.ext.asyncio import (
|
|
AsyncSession, async_sessionmaker, create_async_engine,
|
|
)
|
|
from app.models import Base, KolPost
|
|
from app.services import telegram_digest
|
|
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
sf = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
now = datetime(2026, 6, 4, 12, 0, 0)
|
|
recent = now - timedelta(hours=2)
|
|
|
|
def _post(ext, source, tier=None, handle="cryptohayes"):
|
|
return KolPost(
|
|
kol_handle=handle, source=source, external_id=ext,
|
|
url=f"https://x.com/x/status/{ext}", published_at=recent,
|
|
raw_text="body", content_hash=ext, tier=tier,
|
|
)
|
|
|
|
async with sf() as s:
|
|
s.add_all([
|
|
_post("sub1", "substack", tier=None, handle="raoulpal"), # essay → count
|
|
_post("tw1", "twitter", tier="directional"), # signal → count
|
|
_post("tw2", "twitter", tier="noise"), # gm noise → exclude
|
|
_post("tw3", "twitter", tier="noise"), # RT noise → exclude
|
|
])
|
|
await s.commit()
|
|
|
|
monkeypatch.setattr(telegram_digest, "async_session", sf)
|
|
g = await telegram_digest.build_global_digest(now=now)
|
|
# 2 real posts (substack essay + twitter signal); 2 noise excluded
|
|
assert g.kol.posts_24h == 2
|