k
4442e97f28
New backend pipeline: 8 free public macro signals fetched in parallel,
upserted once per calendar day, served via /api/macro/{snapshot,history}.
- AHR999 (computed from Binance 200d klines)
- Altcoin Season Index (CoinGecko top-50 30d)
- Fear & Greed (alternative.me)
- BTC dominance, ETH/BTC ratio
- Stablecoin supply (DeFiLlama)
- Spot BTC ETF net flow (Farside)
- BTC perp open interest (Binance fapi)
Each fetcher is independently @_none_on_fail decorated so one outage
can't take down the snapshot; scoring renormalises across whichever
indicators returned a value. Daily cron at 03:00 UTC; on startup a
fire-and-forget bootstrap fills today's row if missing.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
302 lines
11 KiB
Python
Executable File
302 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Post-launch end-to-end smoke test.
|
|
|
|
Runs every observable verification we have about whether the system is
|
|
actually doing its job AFTER you've put the new code live. Safe to run
|
|
on production at any time (read-only except for one round-trip test signal
|
|
that's tagged 'smoke_test' and immediately deleted).
|
|
|
|
Output: one line per check, ✓/✗/!, exits non-zero if anything ✗.
|
|
Wire it into a cron + alert webhook to get paged when something rots.
|
|
|
|
What it verifies (the four "is X working?" questions you'd ask manually):
|
|
|
|
1. Backend alive — /api/health/deep returns ok
|
|
2. Scrapers up-to-date — both Trump pollers fresh (< 60s)
|
|
3. Binance feed alive — WS connected + recent price tick in DB
|
|
4. Public API serves data — every /api/* the dashboard uses returns 200 with non-empty body
|
|
5. KOL pipeline producing — kol_posts has a row in the last 24h
|
|
6. Funding snapshot working — /api/funding/snapshot returns ok=true
|
|
7. Telegram bot reachable — getMe authenticates (skipped if no token)
|
|
8. AI provider reachable — /models lists at least one model
|
|
9. Signal ingest round-trip — POST a 'smoke_test' signal, GET it back, DELETE it
|
|
10. WebSocket broadcasting — connect + receive a tick within 8 s
|
|
|
|
Defaults to localhost:8000 — pass --base for production:
|
|
venv/bin/python scripts/launch_smoke.py --base https://api.trumpalpha.io
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import httpx
|
|
import websockets
|
|
from sqlalchemy import select, func
|
|
from app.config import settings
|
|
from app.database import AsyncSessionLocal
|
|
from app.models import Post, KolPost
|
|
|
|
|
|
def green(s): return f"\033[32m{s}\033[0m"
|
|
def yellow(s): return f"\033[33m{s}\033[0m"
|
|
def red(s): return f"\033[31m{s}\033[0m"
|
|
|
|
|
|
class Checker:
|
|
def __init__(self):
|
|
self.errors: list[str] = []
|
|
self.warnings: list[str] = []
|
|
|
|
def ok(self, name: str, detail: str = ""):
|
|
print(green(f" ✓ {name:36s}") + (f" {detail}" if detail else ""))
|
|
|
|
def warn(self, name: str, detail: str):
|
|
print(yellow(f" ! {name:36s} {detail}"))
|
|
self.warnings.append(f"{name}: {detail}")
|
|
|
|
def fail(self, name: str, detail: str):
|
|
print(red(f" ✗ {name:36s} {detail}"))
|
|
self.errors.append(f"{name}: {detail}")
|
|
|
|
|
|
async def check_health(c: Checker, base: str):
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as cli:
|
|
r = await cli.get(f"{base}/api/health/deep")
|
|
if r.status_code != 200:
|
|
c.fail("health/deep", f"HTTP {r.status_code}")
|
|
return
|
|
body = r.json()
|
|
if body.get("status") != "ok":
|
|
c.fail("health/deep", f"status={body.get('status')} problems={body.get('problems')}")
|
|
return
|
|
c.ok("health/deep", f"db_ok, freshest_age={body.get('freshest_age_sec')}s")
|
|
except Exception as e:
|
|
c.fail("health/deep", f"{type(e).__name__}: {e}")
|
|
|
|
|
|
async def check_scrapers(c: Checker, base: str):
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as cli:
|
|
r = await cli.get(f"{base}/api/health/deep")
|
|
body = r.json()
|
|
for s in body.get("scrapers", []):
|
|
age = s.get("age_sec")
|
|
name = f"scraper:{s['name']}"
|
|
if age is None:
|
|
c.fail(name, "never polled")
|
|
elif age > 60:
|
|
c.warn(name, f"stale ({age}s old)")
|
|
else:
|
|
c.ok(name, f"fresh ({age}s)")
|
|
except Exception as e:
|
|
c.fail("scrapers", f"{type(e).__name__}: {e}")
|
|
|
|
|
|
async def check_public_apis(c: Checker, base: str):
|
|
endpoints = [
|
|
("/api/posts?limit=5", lambda b: isinstance(b, list)),
|
|
("/api/funding/snapshot", lambda b: b.get("ok") is True),
|
|
("/api/scanners", lambda b: "scanners" in b),
|
|
("/api/signals/sources", lambda b: "sources" in b and len(b["sources"]) > 0),
|
|
("/api/kol/posts?limit=5", lambda b: "items" in b),
|
|
("/api/kol/digest?days=7", lambda b: True),
|
|
]
|
|
async with httpx.AsyncClient(timeout=15) as cli:
|
|
for path, validator in endpoints:
|
|
name = f"GET {path[:34]}"
|
|
try:
|
|
r = await cli.get(f"{base}{path}")
|
|
if r.status_code != 200:
|
|
c.fail(name, f"HTTP {r.status_code}: {r.text[:80]}")
|
|
continue
|
|
if not validator(r.json()):
|
|
c.warn(name, "200 but payload shape unexpected")
|
|
continue
|
|
c.ok(name)
|
|
except Exception as e:
|
|
c.fail(name, f"{type(e).__name__}: {e}")
|
|
|
|
|
|
async def check_kol_freshness(c: Checker):
|
|
cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=2)
|
|
async with AsyncSessionLocal() as db:
|
|
n = (await db.execute(
|
|
select(func.count(KolPost.id)).where(KolPost.published_at >= cutoff)
|
|
)).scalar() or 0
|
|
if n > 0:
|
|
c.ok("kol_posts (last 48h)", f"{n} rows")
|
|
else:
|
|
c.warn("kol_posts (last 48h)", "0 rows — daily poll might not have run yet (cron at 01:15 UTC)")
|
|
|
|
|
|
async def check_telegram(c: Checker):
|
|
if not settings.telegram_bot_token:
|
|
c.warn("telegram", "no token — feature disabled")
|
|
return
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as cli:
|
|
r = await cli.get(
|
|
f"https://api.telegram.org/bot{settings.telegram_bot_token}/getMe"
|
|
)
|
|
if r.status_code != 200 or not r.json().get("ok"):
|
|
c.fail("telegram getMe", f"HTTP {r.status_code}: {r.text[:80]}")
|
|
return
|
|
u = r.json()["result"]["username"]
|
|
c.ok("telegram getMe", f"@{u}")
|
|
except Exception as e:
|
|
c.fail("telegram getMe", f"{type(e).__name__}: {e}")
|
|
|
|
|
|
async def check_ai(c: Checker):
|
|
if settings.anthropic_api_key:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as cli:
|
|
r = await cli.get(
|
|
"https://api.anthropic.com/v1/models",
|
|
headers={"x-api-key": settings.anthropic_api_key,
|
|
"anthropic-version": "2023-06-01"},
|
|
)
|
|
if r.status_code == 200:
|
|
c.ok("anthropic /models", "auth OK")
|
|
else:
|
|
c.fail("anthropic /models", f"HTTP {r.status_code}")
|
|
except Exception as e:
|
|
c.fail("anthropic", f"{type(e).__name__}: {e}")
|
|
elif settings.ai_api_key:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as cli:
|
|
r = await cli.get(
|
|
f"{settings.ai_base_url.rstrip('/')}/models",
|
|
headers={"Authorization": f"Bearer {settings.ai_api_key}"},
|
|
)
|
|
if r.status_code == 200:
|
|
c.ok("AI /models", f"auth OK at {settings.ai_base_url}")
|
|
else:
|
|
c.fail("AI /models", f"HTTP {r.status_code}")
|
|
except Exception as e:
|
|
c.fail("AI provider", f"{type(e).__name__}: {e}")
|
|
else:
|
|
c.warn("AI provider", "no key configured — Trump signals will not be scored")
|
|
|
|
|
|
async def check_signal_ingest_roundtrip(c: Checker, base: str):
|
|
"""End-to-end: POST a smoke_test signal, confirm DB+API see it, then delete."""
|
|
if not settings.ingest_api_key:
|
|
c.warn("signal ingest", "INGEST_API_KEY empty — endpoint fail-closed (correct), can't round-trip")
|
|
return
|
|
ts_tag = int(time.time())
|
|
ext_id = f"smoke-{ts_tag}"
|
|
payload = {
|
|
"source": "smoke_test",
|
|
"external_id": ext_id,
|
|
"text": "Synthetic smoke-test signal from scripts/launch_smoke.py — IGNORE.",
|
|
"signal": "buy",
|
|
"target_asset": "BTC",
|
|
"confidence": 50,
|
|
"category": "smoke_test",
|
|
}
|
|
posted_id = None
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as cli:
|
|
r = await cli.post(
|
|
f"{base}/api/signals/ingest",
|
|
json=payload,
|
|
headers={"X-Ingest-Key": settings.ingest_api_key},
|
|
)
|
|
if r.status_code != 200:
|
|
c.fail("ingest POST", f"HTTP {r.status_code}: {r.text[:120]}")
|
|
return
|
|
result = r.json()
|
|
posted_id = result.get("post_id")
|
|
if not posted_id:
|
|
c.fail("ingest POST", f"no post_id in response: {result}")
|
|
return
|
|
# Read back via DB (the public /api/posts may not return non-trading sources)
|
|
async with AsyncSessionLocal() as db:
|
|
row = await db.get(Post, posted_id)
|
|
if not row:
|
|
c.fail("ingest round-trip", f"posted id={posted_id} not in DB")
|
|
return
|
|
c.ok("ingest round-trip", f"post_id={posted_id}, status={result.get('status')}")
|
|
except Exception as e:
|
|
c.fail("ingest round-trip", f"{type(e).__name__}: {e}")
|
|
finally:
|
|
# Always clean up the smoke-test row regardless of whether the check passed.
|
|
if posted_id is not None:
|
|
try:
|
|
async with AsyncSessionLocal() as db:
|
|
p = await db.get(Post, posted_id)
|
|
if p:
|
|
await db.delete(p)
|
|
await db.commit()
|
|
except Exception as e:
|
|
c.warn("ingest cleanup",
|
|
f"could not delete smoke-test post_id={posted_id}: {e}")
|
|
|
|
|
|
async def check_ws(c: Checker, base: str):
|
|
# base is http(s)://...; switch scheme
|
|
ws_url = base.replace("http://", "ws://").replace("https://", "wss://") + "/ws/prices"
|
|
try:
|
|
async with websockets.connect(ws_url, open_timeout=5) as ws:
|
|
try:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=8)
|
|
c.ok("websocket /ws/prices", f"got tick: {msg[:80]}")
|
|
except asyncio.TimeoutError:
|
|
c.warn("websocket /ws/prices",
|
|
"connected but no tick in 8s — possible if Binance WS just reconnected")
|
|
except Exception as e:
|
|
c.fail("websocket /ws/prices", f"{type(e).__name__}: {e}")
|
|
|
|
|
|
async def main() -> int:
|
|
p = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
p.add_argument("--base", default="http://localhost:8000",
|
|
help="Backend base URL (default: localhost:8000)")
|
|
args = p.parse_args()
|
|
base = args.base.rstrip("/")
|
|
|
|
print(f"Smoke test against: {base}\n")
|
|
c = Checker()
|
|
|
|
await check_health(c, base)
|
|
await check_scrapers(c, base)
|
|
await check_public_apis(c, base)
|
|
await check_kol_freshness(c)
|
|
await check_telegram(c)
|
|
await check_ai(c)
|
|
await check_signal_ingest_roundtrip(c, base)
|
|
await check_ws(c, base)
|
|
|
|
print()
|
|
if c.errors:
|
|
print(red(f"❌ {len(c.errors)} failure(s):"))
|
|
for e in c.errors:
|
|
print(red(f" - {e}"))
|
|
if c.warnings:
|
|
print(yellow(f"⚠️ {len(c.warnings)} warning(s):"))
|
|
for w in c.warnings:
|
|
print(yellow(f" - {w}"))
|
|
return 1
|
|
if c.warnings:
|
|
print(yellow(f"⚠️ {len(c.warnings)} warning(s) — but all critical checks passed:"))
|
|
for w in c.warnings:
|
|
print(yellow(f" - {w}"))
|
|
print(green("✅ All critical checks passed."))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(main()))
|