From 4442e97f281484dca98cf2f2943137347916dc7e Mon Sep 17 00:00:00 2001 From: k Date: Tue, 26 May 2026 01:04:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(macro):=20Macro=20Vibes=20=E2=80=94=208-in?= =?UTF-8?q?dicator=20daily=20snapshot=20+=20composite=20score?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- alembic/versions/022_macro_snapshots.py | 66 +++++ app/api/macro.py | 86 ++++++ app/main.py | 34 +++ app/models.py | 34 ++- app/services/macro/__init__.py | 0 app/services/macro/fetchers.py | 335 ++++++++++++++++++++++++ app/services/macro/poll.py | 102 ++++++++ app/services/macro/scoring.py | 149 +++++++++++ scripts/launch_seed.py | 290 ++++++++++++++++++++ scripts/launch_smoke.py | 301 +++++++++++++++++++++ scripts/verify.sh | 15 ++ tests/test_macro_fetchers_timing.py | 50 ++++ 12 files changed, 1461 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/022_macro_snapshots.py create mode 100644 app/api/macro.py create mode 100644 app/services/macro/__init__.py create mode 100644 app/services/macro/fetchers.py create mode 100644 app/services/macro/poll.py create mode 100644 app/services/macro/scoring.py create mode 100755 scripts/launch_seed.py create mode 100755 scripts/launch_smoke.py create mode 100755 scripts/verify.sh create mode 100644 tests/test_macro_fetchers_timing.py diff --git a/alembic/versions/022_macro_snapshots.py b/alembic/versions/022_macro_snapshots.py new file mode 100644 index 0000000..79e2ffc --- /dev/null +++ b/alembic/versions/022_macro_snapshots.py @@ -0,0 +1,66 @@ +"""Macro indicator snapshots (one row per day). + +Adds the `macro_snapshots` table backing the BTC page's macro panel. Wide +table — one column per indicator — because every panel view fetches them +all at once and a long EAV table would just need an immediate pivot. Daily +snapshot uniqueness is enforced by a UNIQUE(snapshot_date) constraint so the +fetcher can run safely on cron without producing dupes. + +Indicators (all free / public): + ahr999 : derived from price + age + altcoin_season_index : % of top-50 alts beating BTC over 90d (blockchaincenter formula) + fear_greed : alternative.me, 0-100 + btc_dominance_pct : CoinGecko global + eth_btc_ratio : Binance ETHBTC + stablecoin_supply_usd : DeFiLlama (USDT+USDC+DAI total) + etf_flow_net_usd_1d : Farside Investors (BTC spot ETF daily net flow) + btc_open_interest_usd : Binance fapi open interest + composite_score : -100..100 weighted blend computed at insert time + regime_label : "BULL" | "BULLISH" | "NEUTRAL" | "BEARISH" | "BEAR" + +`raw_json` retains the full upstream payload per indicator in case we want +to recompute or audit historical fetches without re-hitting the APIs. + +Revision ID: 022 +Revises: 021 +Create Date: 2026-05-25 +""" +from alembic import op +import sqlalchemy as sa + +revision = "022" +down_revision = "021" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "macro_snapshots", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("snapshot_date", sa.Date, nullable=False, unique=True, index=True), + sa.Column("captured_at", sa.DateTime, nullable=False), + + # The indicators. All nullable — any single upstream failure shouldn't + # block the whole row; we record what we got and leave the rest NULL. + sa.Column("ahr999", sa.Float, nullable=True), + sa.Column("altcoin_season_index", sa.Float, nullable=True), + sa.Column("fear_greed", sa.Integer, nullable=True), + sa.Column("fear_greed_label", sa.String(32), nullable=True), + sa.Column("btc_dominance_pct", sa.Float, nullable=True), + sa.Column("eth_btc_ratio", sa.Float, nullable=True), + sa.Column("stablecoin_supply_usd", sa.Float, nullable=True), + sa.Column("etf_flow_net_usd_1d", sa.Float, nullable=True), + sa.Column("btc_open_interest_usd", sa.Float, nullable=True), + + # Optional composite — see app/services/macro/scoring.py. + sa.Column("composite_score", sa.Float, nullable=True), + sa.Column("regime_label", sa.String(16), nullable=True), + + # Raw payload per indicator for debugging / re-scoring. + sa.Column("raw_json", sa.Text, nullable=True), + ) + + +def downgrade() -> None: + op.drop_table("macro_snapshots") diff --git a/app/api/macro.py b/app/api/macro.py new file mode 100644 index 0000000..d136f1a --- /dev/null +++ b/app/api/macro.py @@ -0,0 +1,86 @@ +"""Macro indicators API. + + GET /api/macro/snapshot + Latest macro snapshot (today's row, or the most recent available). + + GET /api/macro/history?days=30 + Time series of every indicator over the last N days. Used by the + sparklines on the BTC page macro panel. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Optional + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import get_db +from app.models import MacroSnapshot + +router = APIRouter(prefix="/macro", tags=["macro"]) + + +def _row_to_dict(r: MacroSnapshot) -> dict: + """Snapshot row → JSON-friendly dict, in the canonical indicator order + the frontend lists them in. + + Order rule (see UI mock-up in BtcPageClient MacroPanel): + 1. AHR999 + 2. Altcoin Season Index + 3. Fear & Greed + 4. BTC Dominance + 5. ETH/BTC Ratio + 6. Stablecoin Total Supply + 7. ETF Net Flow (1d) + 8. BTC Open Interest + """ + return { + "date": r.snapshot_date.isoformat() if r.snapshot_date else None, + "captured_at": r.captured_at.replace(tzinfo=timezone.utc).isoformat() if r.captured_at else None, + "indicators": { + "ahr999": r.ahr999, + "altcoin_season_index": r.altcoin_season_index, + "fear_greed": r.fear_greed, + "fear_greed_label": r.fear_greed_label, + "btc_dominance_pct": r.btc_dominance_pct, + "eth_btc_ratio": r.eth_btc_ratio, + "stablecoin_supply_usd": r.stablecoin_supply_usd, + "etf_flow_net_usd_1d": r.etf_flow_net_usd_1d, + "btc_open_interest_usd": r.btc_open_interest_usd, + }, + "composite_score": r.composite_score, + "regime_label": r.regime_label, + } + + +@router.get("/snapshot") +async def get_snapshot(db: AsyncSession = Depends(get_db)) -> dict: + """Latest macro snapshot. May be null if poll hasn't run yet.""" + row = (await db.execute( + select(MacroSnapshot).order_by(MacroSnapshot.snapshot_date.desc()).limit(1) + )).scalar_one_or_none() + if not row: + return {"ok": False, "error": "no snapshots yet — poll has not run"} + return {"ok": True, **_row_to_dict(row)} + + +@router.get("/history") +async def get_history( + days: int = Query(default=30, ge=1, le=365), + db: AsyncSession = Depends(get_db), +) -> dict: + """Time series across the last N days — for the panel sparklines.""" + cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).date() + rows = (await db.execute( + select(MacroSnapshot) + .where(MacroSnapshot.snapshot_date >= cutoff) + .order_by(MacroSnapshot.snapshot_date.asc()) + )).scalars().all() + return { + "ok": True, + "days": days, + "count": len(rows), + "items": [_row_to_dict(r) for r in rows], + } diff --git a/app/main.py b/app/main.py index db9c74d..e40b9fc 100644 --- a/app/main.py +++ b/app/main.py @@ -28,6 +28,7 @@ from app.api.signals import router as signals_router from app.api.positions import router as positions_router from app.api.scanners import router as scanners_router from app.api.kol import router as kol_router +from app.api.macro import router as macro_router logging.basicConfig( level=logging.INFO, @@ -176,6 +177,38 @@ async def lifespan(app: FastAPI): ) logger.info("KOL divergence scan scheduled daily at 02:15 UTC.") + # ── Macro indicator daily snapshot ──────────────────────────────────── + # 8 indicators (AHR999, Fear & Greed, BTC dominance, ETH/BTC, stablecoin + # supply, ETF flow, BTC OI, altcoin season). One row per calendar date. + # Runs after KOL jobs so a slow KOL fetch can't make this one miss. + from app.services.macro.poll import run_macro_poll + _scheduler.add_job( + run_macro_poll, "cron", hour=3, minute=0, + id="macro_poll", max_instances=1, coalesce=True, + ) + logger.info("Macro indicator snapshot scheduled daily at 03:00 UTC.") + + # Kick off an initial poll on startup IF today's row doesn't exist yet. + # Otherwise a fresh deploy shows an empty macro panel until 03:00 UTC of + # the next day. Fire-and-forget — never blocks startup. + async def _macro_bootstrap(): + try: + from datetime import datetime, timezone + from sqlalchemy import select + from app.models import MacroSnapshot + today = datetime.now(timezone.utc).date() + async with AsyncSessionLocal() as db: + exists = (await db.execute( + select(MacroSnapshot).where(MacroSnapshot.snapshot_date == today) + )).scalar_one_or_none() + if exists is None: + logger.info("Macro: no row for today, running one-shot bootstrap fetch.") + await run_macro_poll() + except Exception as exc: + logger.warning("Macro bootstrap fetch failed: %s (%s)", + type(exc).__name__, exc) + asyncio.create_task(_macro_bootstrap()) + _scheduler.start() logger.info( "Truth Social pollers scheduled every %ds (CNN + trumpstruth.org).", @@ -254,6 +287,7 @@ app.include_router(signals_router, prefix="/api") app.include_router(positions_router, prefix="/api") app.include_router(scanners_router, prefix="/api") app.include_router(kol_router, prefix="/api") +app.include_router(macro_router, prefix="/api") @app.get("/api/health") diff --git a/app/models.py b/app/models.py index cbef3c9..5332714 100644 --- a/app/models.py +++ b/app/models.py @@ -1,9 +1,10 @@ -from datetime import datetime, timezone +from datetime import date, datetime, timezone from typing import List, Optional from sqlalchemy import ( BigInteger, Boolean, + Date, DateTime, Float, ForeignKey, @@ -412,3 +413,34 @@ class TelegramBinding(Base): last_alert_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) total_alerts_sent: Mapped[int] = mapped_column(Integer, nullable=False, default=0) total_alerts_failed: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + +class MacroSnapshot(Base): + """Daily snapshot of all macro indicators surfaced on the BTC page. + + One row per calendar date. Every indicator column is nullable — any single + upstream API failing must not block the rest from being recorded. + Composite score is computed at insert time by app/services/macro/scoring.py + against whichever indicators successfully fetched (the formula degrades + gracefully if a few are missing). + """ + __tablename__ = "macro_snapshots" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + snapshot_date: Mapped[date] = mapped_column(Date, nullable=False, unique=True, index=True) + captured_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=utcnow) + + ahr999: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + altcoin_season_index: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + fear_greed: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + fear_greed_label: Mapped[Optional[str]] = mapped_column(String(32), nullable=True) + btc_dominance_pct: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + eth_btc_ratio: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + stablecoin_supply_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + etf_flow_net_usd_1d: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + btc_open_interest_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + + composite_score: Mapped[Optional[float]] = mapped_column(Float, nullable=True) + regime_label: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + + raw_json: Mapped[Optional[str]] = mapped_column(Text, nullable=True) diff --git a/app/services/macro/__init__.py b/app/services/macro/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/macro/fetchers.py b/app/services/macro/fetchers.py new file mode 100644 index 0000000..df9ed9f --- /dev/null +++ b/app/services/macro/fetchers.py @@ -0,0 +1,335 @@ +"""Individual fetchers for each macro indicator. + +Each function is an async coroutine that returns a dict shaped like: + { "value": float | int | None, + "label": Optional[str], # only some indicators + "raw": } # for debugging / re-scoring + +Every fetcher MUST tolerate upstream failure — return {"value": None} rather +than raise — so one dead API can't take down the whole snapshot. + +Public, free, no-key sources only: + + AHR999 : derived from BTC daily closes (Binance fapi) + Altcoin Season Index : CoinGecko top-50 90-day relative performance + Fear & Greed : api.alternative.me/fng (no auth) + BTC Dominance : CoinGecko /global + ETH/BTC Ratio : Binance kline ETHBTC daily + Stablecoin Supply : DeFiLlama /stablecoins + ETF Net Flow (1d) : Farside Investors HTML scrape + BTC Open Interest : Binance fapi /futures/data/openInterestHist +""" +from __future__ import annotations + +import logging +import math +import re +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +import httpx + +logger = logging.getLogger(__name__) + +# A vanilla User-Agent. CoinGecko + alternative.me + DeFiLlama all happily +# serve "Mozilla/5.0"; some get suspicious of anything that looks bot-like +# (e.g. python-httpx default UA returns 400 on /global). +UA = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) AppleWebKit/605.1.15"} +DEFAULT_TIMEOUT = 20 + + +def _none_on_fail(name: str): + """Decorator: log+swallow exceptions from a fetcher and return {value: None}.""" + def deco(fn): + async def wrapper(*a, **kw): + try: + return await fn(*a, **kw) + except Exception as exc: + logger.warning("macro fetch %s failed: %s (%s)", + name, type(exc).__name__, exc) + return {"value": None, "raw": {"error": f"{type(exc).__name__}: {exc}"}} + return wrapper + return deco + + +def _utc_midnight_ms(now: Optional[datetime] = None) -> int: + dt = now or datetime.now(timezone.utc) + midnight = dt.replace(hour=0, minute=0, second=0, microsecond=0) + return int(midnight.timestamp() * 1000) + + +def _drop_in_progress_daily_klines(rows: list[list], now: Optional[datetime] = None) -> list[list]: + """Binance daily klines are keyed by OPEN time. If the latest row opened at + today's 00:00 UTC, that candle is still in progress and should not be used + for daily snapshots.""" + if not rows: + return rows + cutoff = _utc_midnight_ms(now) + return [row for row in rows if int(row[0]) < cutoff] + + +def _latest_closed_daily_point(rows: list[dict], now: Optional[datetime] = None) -> Optional[dict]: + """Same idea as `_drop_in_progress_daily_klines`, but for daily point + series keyed by `timestamp`.""" + if not rows: + return None + cutoff = _utc_midnight_ms(now) + closed = [row for row in rows if int(row.get("timestamp", 0)) < cutoff] + return closed[-1] if closed else None + + +def _parse_farside_latest_total(html: str) -> dict: + """Extract the most recent dated row from Farside's historical table. + + The all-data table is chronological from oldest to newest, so the first + date row is NOT the latest one. + """ + m = re.search(r"]*>(.*?)", html, re.DOTALL | re.IGNORECASE) + if not m: + return {"value": None, "raw": {"error": "tbody not found"}} + body = m.group(1) + rows = re.findall(r"]*>(.*?)", body, re.DOTALL | re.IGNORECASE) + latest: Optional[dict] = None + for row in rows: + cells = re.findall(r"]*>(.*?)", row, re.DOTALL | re.IGNORECASE) + if not cells: + continue + date_text = re.sub(r"<[^>]+>", "", cells[0]).strip() + if not re.match(r"\d{1,2}\s+[A-Za-z]+\s+\d{4}", date_text): + continue + last_text = re.sub(r"<[^>]+>", "", cells[-1]).strip() + num = last_text.replace(",", "").replace("(", "-").replace(")", "") + try: + millions = float(num) + row_date = datetime.strptime(date_text, "%d %b %Y").date() + except ValueError: + continue + candidate = { + "value": round(millions * 1_000_000, 2), + "raw": {"date": date_text, "millions_usd": millions}, + "_date": row_date, + } + if latest is None or candidate["_date"] > latest["_date"]: + latest = candidate + + if latest is None: + return {"value": None, "raw": {"error": "no parseable rows"}} + latest.pop("_date", None) + return latest + + +# ── 1. AHR999 ─────────────────────────────────────────────────────────────── +# Formula: AHR999 = (price / 200d MA) × (price / age_fit_price) +# age_fit_price = 10 ** (5.84 * log10(days_since_2009_01_03) - 17.01) +# Below 0.45 historically marks accumulation zones; above 1.2 marks +# "expensive" regime that invalidates a bottom thesis. + +_AHR999_GENESIS = datetime(2009, 1, 3, tzinfo=timezone.utc) + + +@_none_on_fail("ahr999") +async def fetch_ahr999() -> dict: + """Compute AHR999 from the last 200 daily BTC closes (Binance fapi).""" + end_ms = int(datetime.now(timezone.utc).timestamp() * 1000) + start_ms = end_ms - 260 * 24 * 3600 * 1000 # extra buffer after dropping in-progress day + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as c: + r = await c.get( + "https://fapi.binance.com/fapi/v1/klines", + params={"symbol": "BTCUSDT", "interval": "1d", + "startTime": start_ms, "endTime": end_ms, "limit": 300}, + ) + r.raise_for_status() + rows = _drop_in_progress_daily_klines(r.json()) + closes = [float(row[4]) for row in rows] + if len(closes) < 200: + return {"value": None, "raw": {"error": "insufficient candles", "have": len(closes)}} + price = closes[-1] + ma200 = sum(closes[-200:]) / 200 + + days = (datetime.now(timezone.utc) - _AHR999_GENESIS).total_seconds() / 86400 + age_fit = 10 ** (5.84 * math.log10(days) - 17.01) + + ahr = (price / ma200) * (price / age_fit) + return { + "value": round(ahr, 4), + "raw": {"price": price, "ma200": round(ma200, 2), + "age_fit": round(age_fit, 2), "days": round(days, 1)}, + } + + +# ── 2. Altcoin Season Index (blockchaincenter.net formula) ─────────────────── +# Take top-50 coins by market cap (excluding stablecoins + wrapped). Count how +# many beat BTC's 90-day return. Result is the count, projected to 0-100. +# 75+ = altseason, <25 = bitcoin season, middle = neutral. + +_STABLE_OR_WRAPPED = { + "USDT", "USDC", "DAI", "BUSD", "TUSD", "USDD", "FDUSD", "PYUSD", "USDE", + "WBTC", "WETH", "STETH", "WSTETH", "WEETH", "RETH", +} + + +@_none_on_fail("altcoin_season_index") +async def fetch_altcoin_season_index() -> dict: + """Compute the Altcoin Season Index from CoinGecko /coins/markets. + + Original blockchaincenter.net formula uses a 90-day window, but + CoinGecko's /coins/markets `price_change_percentage` parameter only + accepts 1h/24h/7d/14d/30d/200d/1y — 90d returns HTTP 400. We use 30d + as the closest practical proxy. Long-horizon altseason (which 90d + captures better) would need per-coin /market_chart calls — 50× the + API budget for a marginal definition improvement. + """ + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=UA) as c: + r = await c.get( + "https://api.coingecko.com/api/v3/coins/markets", + params={"vs_currency": "usd", "order": "market_cap_desc", + "per_page": 60, "page": 1, + "price_change_percentage": "30d"}, + ) + r.raise_for_status() + rows = r.json() + + # Drop stablecoins + wrapped, keep top 50 of the remainder. + eligible = [ + row for row in rows + if (row.get("symbol") or "").upper() not in _STABLE_OR_WRAPPED + and row.get("price_change_percentage_30d_in_currency") is not None + ][:50] + if len(eligible) < 30: + return {"value": None, "raw": {"error": "insufficient eligible coins", + "have": len(eligible)}} + + btc_row = next((x for x in rows if x.get("symbol", "").upper() == "BTC"), None) + btc_30d = btc_row.get("price_change_percentage_30d_in_currency") if btc_row else None + if btc_30d is None: + return {"value": None, "raw": {"error": "BTC 30d return missing"}} + + n_outperform = sum( + 1 for row in eligible + if (row["price_change_percentage_30d_in_currency"] or -999) > btc_30d + ) + # Project the count over `len(eligible)` to a 0–100 scale. + index = (n_outperform / len(eligible)) * 100 + return { + "value": round(index, 1), + "raw": {"n_outperform": n_outperform, "of": len(eligible), + "btc_30d_pct": round(btc_30d, 2), "window": "30d"}, + } + + +# ── 3. Fear & Greed (alternative.me) ──────────────────────────────────────── + + +@_none_on_fail("fear_greed") +async def fetch_fear_greed() -> dict: + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=UA) as c: + r = await c.get("https://api.alternative.me/fng/?limit=1") + r.raise_for_status() + data = r.json() + item = (data.get("data") or [None])[0] + if not item: + return {"value": None, "raw": data} + return { + "value": int(item.get("value", 0)), + "label": item.get("value_classification"), + "raw": item, + } + + +# ── 4. BTC Dominance (CoinGecko /global) ──────────────────────────────────── + + +@_none_on_fail("btc_dominance") +async def fetch_btc_dominance() -> dict: + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=UA) as c: + r = await c.get("https://api.coingecko.com/api/v3/global") + r.raise_for_status() + data = r.json() + pct = (data.get("data", {}).get("market_cap_percentage", {}) or {}).get("btc") + if pct is None: + return {"value": None, "raw": data} + return {"value": round(float(pct), 2), + "raw": {"total_mcap_usd": data["data"]["total_market_cap"].get("usd")}} + + +# ── 5. ETH/BTC Ratio (Binance ETHBTC daily) ────────────────────────────────── + + +@_none_on_fail("eth_btc_ratio") +async def fetch_eth_btc_ratio() -> dict: + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as c: + r = await c.get( + "https://fapi.binance.com/fapi/v1/klines", + params={"symbol": "ETHBTC", "interval": "1d", "limit": 3}, + ) + # fapi may 404 ETHBTC; fall back to spot kline endpoint via data-api host. + if r.status_code == 400 or r.status_code == 404: + r = await c.get( + "https://data-api.binance.vision/api/v3/klines", + params={"symbol": "ETHBTC", "interval": "1d", "limit": 3}, + ) + r.raise_for_status() + rows = _drop_in_progress_daily_klines(r.json()) + if not rows: + return {"value": None, "raw": rows} + close = float(rows[-1][4]) + return {"value": round(close, 6), "raw": {"close": close, "n_rows": len(rows)}} + + +# ── 6. Stablecoin Total Supply (DeFiLlama) ─────────────────────────────────── + + +@_none_on_fail("stablecoin_supply") +async def fetch_stablecoin_supply() -> dict: + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=UA) as c: + r = await c.get( + "https://stablecoins.llama.fi/stablecoins", + params={"includePrices": "true"}, + ) + r.raise_for_status() + data = r.json() + # Sum circulating peggedUSD across all stables. + total = 0.0 + for stable in data.get("peggedAssets", []): + circ = stable.get("circulating", {}).get("peggedUSD") + if isinstance(circ, (int, float)): + total += float(circ) + if total <= 0: + return {"value": None, "raw": {"error": "no peggedUSD totals found"}} + return {"value": round(total, 2), + "raw": {"n_stables": len(data.get("peggedAssets", []))}} + + +# ── 7. BTC Spot ETF Net Flow 1d (Farside) ──────────────────────────────────── +# Farside doesn't have a JSON API but their daily flow page is parseable. We +# pull the most recent row from the All ETFs sum. + + +@_none_on_fail("etf_flow") +async def fetch_etf_flow() -> dict: + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, headers=UA, + follow_redirects=True) as c: + r = await c.get("https://farside.co.uk/bitcoin-etf-flow-all-data/") + r.raise_for_status() + return _parse_farside_latest_total(r.text) + + +# ── 8. BTC Open Interest (Binance fapi) ────────────────────────────────────── + + +@_none_on_fail("btc_open_interest") +async def fetch_btc_open_interest() -> dict: + async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as c: + r = await c.get( + "https://fapi.binance.com/futures/data/openInterestHist", + params={"symbol": "BTCUSDT", "period": "1d", "limit": 4}, + ) + r.raise_for_status() + rows = r.json() + latest = _latest_closed_daily_point(rows) + if not latest: + return {"value": None, "raw": rows} + notional = float(latest.get("sumOpenInterestValue", 0)) + return {"value": round(notional, 2), + "raw": {"contracts": float(latest.get("sumOpenInterest", 0)), + "timestamp": latest.get("timestamp")}} diff --git a/app/services/macro/poll.py b/app/services/macro/poll.py new file mode 100644 index 0000000..b978bac --- /dev/null +++ b/app/services/macro/poll.py @@ -0,0 +1,102 @@ +"""Daily macro snapshot orchestrator. + +Runs every fetcher in parallel (each is independently fail-tolerant), computes +the composite score against whatever came back, and UPSERTs one row keyed by +today's date. Subsequent same-day invocations overwrite the row with newer +data — i.e. you can re-run by hand to refresh after fixing a fetcher without +duplicating snapshots. + +Schedule: app/main.py wires this to cron at 03:00 UTC (after KOL polls). +""" +from __future__ import annotations + +import asyncio +import json +import logging +from datetime import datetime, timezone + +from sqlalchemy import select + +from app.database import AsyncSessionLocal +from app.models import MacroSnapshot + +from . import fetchers, scoring + +logger = logging.getLogger(__name__) + + +async def run_macro_poll() -> dict: + """Fetch every indicator and write today's snapshot row. Idempotent per + calendar day — re-runs overwrite the existing row instead of inserting.""" + logger.info("[macro] poll starting") + + # Run every fetcher in parallel — they're all independent. + results = await asyncio.gather( + fetchers.fetch_ahr999(), + fetchers.fetch_altcoin_season_index(), + fetchers.fetch_fear_greed(), + fetchers.fetch_btc_dominance(), + fetchers.fetch_eth_btc_ratio(), + fetchers.fetch_stablecoin_supply(), + fetchers.fetch_etf_flow(), + fetchers.fetch_btc_open_interest(), + return_exceptions=False, # the decorator already catches everything + ) + (ahr, alt, fg, dom, ebr, stab, etf, oi) = results + + # Pack into a values dict matching MacroSnapshot's columns. + values = { + "ahr999": ahr["value"], + "altcoin_season_index": alt["value"], + "fear_greed": fg["value"], + "fear_greed_label": fg.get("label"), + "btc_dominance_pct": dom["value"], + "eth_btc_ratio": ebr["value"], + "stablecoin_supply_usd": stab["value"], + "etf_flow_net_usd_1d": etf["value"], + "btc_open_interest_usd": oi["value"], + } + composite, regime = scoring.compute_composite(values) + values["composite_score"] = composite + values["regime_label"] = regime + + raw_blob = json.dumps({ + "ahr999": ahr.get("raw"), + "altcoin_season": alt.get("raw"), + "fear_greed": fg.get("raw"), + "btc_dominance": dom.get("raw"), + "eth_btc_ratio": ebr.get("raw"), + "stablecoin_supply": stab.get("raw"), + "etf_flow": etf.get("raw"), + "btc_open_interest": oi.get("raw"), + }, default=str)[:8000] # cap to a sane Text column size + + today = datetime.now(timezone.utc).date() + now = datetime.now(timezone.utc).replace(tzinfo=None) + + async with AsyncSessionLocal() as db: + existing = (await db.execute( + select(MacroSnapshot).where(MacroSnapshot.snapshot_date == today) + )).scalar_one_or_none() + + if existing: + # Update in place — same-day re-run. + for k, v in values.items(): + setattr(existing, k, v) + existing.captured_at = now + existing.raw_json = raw_blob + else: + db.add(MacroSnapshot( + snapshot_date=today, + captured_at=now, + raw_json=raw_blob, + **values, + )) + await db.commit() + + n_alive = sum(1 for v in [ahr, alt, fg, dom, ebr, stab, etf, oi] if v["value"] is not None) + logger.info("[macro] poll done: %d/8 indicators OK, composite=%s (%s)", + n_alive, composite, regime) + return {"date": today.isoformat(), "alive": n_alive, "of": 8, + "composite": composite, "regime": regime, + "values": values} diff --git a/app/services/macro/scoring.py b/app/services/macro/scoring.py new file mode 100644 index 0000000..977de0f --- /dev/null +++ b/app/services/macro/scoring.py @@ -0,0 +1,149 @@ +"""Composite "regime" score over the macro indicators. + +Goal: one -100..+100 number that says whether the overall macro setup tilts +risk-on (positive) or risk-off (negative). Used purely as advisory display on +the BTC page — does NOT (yet) feed sizing or trade decisions. + +Design choices: + * Each indicator contributes a per-indicator signal in [-1, +1]. + * Weights sum to 1.0 across the indicators we got. Missing indicators are + excluded and the remaining weights are renormalised — so a single dead + API doesn't drag the whole score toward zero. + * The contrarian indicators (fear/greed, AHR999, funding) are intentionally + inverted: extreme fear / cheap AHR999 / extreme funding all add positive + score (buy-the-fear logic). + +The weights here are seeded by hand from rough conviction. Re-tune after a +few weeks of live data — see `composite_score` history vs realised forward +returns. +""" +from __future__ import annotations + +from typing import Optional + +# (weight, signal_function) per indicator. signal_function takes the raw value +# and returns a number in [-1, +1]. +# +# Inputs may be None if the upstream fetch failed; the orchestrator filters +# them out and renormalises before summing. + + +def _ahr999_signal(v: Optional[float]) -> Optional[float]: + """< 0.45 cheap → +1; 0.45–1.2 neutral; > 1.2 expensive → -1.""" + if v is None: return None + if v < 0.45: return 1.0 + if v > 1.2: return -1.0 + # Linear interpolation between 0.45 and 1.2 → +1 to -1. + return 1.0 - 2 * ((v - 0.45) / (1.2 - 0.45)) + + +def _altseason_signal(v: Optional[float]) -> Optional[float]: + """High altseason index = risk-on (+1), low = BTC season (defensive but + not bearish, so a mild positive).""" + if v is None: return None + if v >= 75: return 0.7 # altseason — generally bullish risk + if v <= 25: return 0.3 # BTC-only — defensive but not a bear signal + return (v - 50) / 50 # linear, -0.5 to +0.5 + + +def _fear_greed_signal(v: Optional[int]) -> Optional[float]: + """Contrarian: extreme fear → buy (+1), extreme greed → sell (-1).""" + if v is None: return None + # Map 0..100 → +1..-1 (inverted, contrarian). + return (50 - v) / 50 + + +def _btc_dominance_signal(v: Optional[float]) -> Optional[float]: + """Hard to read in isolation — only inform on extremes. + Very high dominance often signals fear (risk-off into BTC) → mild bearish. + Very low = altseason froth → also mild bearish (cycle late). Mid = neutral.""" + if v is None: return None + if v >= 60: return -0.3 + if v <= 40: return -0.3 + return 0.0 + + +def _eth_btc_signal(v: Optional[float]) -> Optional[float]: + """Rising ETH/BTC = risk-on. No persistent absolute level matters; this is + really a trend indicator. We approximate with absolute thresholds for the + current cycle (2025-2026): > 0.04 risk-on, < 0.025 risk-off.""" + if v is None: return None + if v >= 0.04: return 0.7 + if v <= 0.025: return -0.7 + return (v - 0.0325) / 0.0075 * 0.7 # linear in the middle + + +def _stablecoin_supply_signal(v: Optional[float]) -> Optional[float]: + """Absolute supply tells us little day-over-day; we need the delta. Since + this scorer sees only the snapshot, we treat presence as 0 and let the + visual chart show the trend. Returns 0 if we have any value at all.""" + if v is None: return None + return 0.0 # contribution = 0 until we wire in a trend lookup + + +def _etf_flow_signal(v: Optional[float]) -> Optional[float]: + """Net inflow = institutional bid → +1, outflow → -1. Scale by magnitude.""" + if v is None: return None + # Daily prints over $200M are notable; over $500M unusually large. + abs_v = abs(v) + sign = 1 if v > 0 else (-1 if v < 0 else 0) + if abs_v >= 500_000_000: return 1.0 * sign + if abs_v >= 200_000_000: return 0.7 * sign + if abs_v >= 50_000_000: return 0.4 * sign + return 0.1 * sign + + +def _open_interest_signal(v: Optional[float]) -> Optional[float]: + """OI in isolation doesn't tell us direction — we'd need OI vs price + correlation. Until we have a trend window, contribute 0.""" + if v is None: return None + return 0.0 + + +# Weights (sum to 1.0 across all). When an indicator is missing, we drop its +# weight and renormalise the rest. +WEIGHTS = { + "ahr999": 0.20, + "altcoin_season": 0.10, + "fear_greed": 0.20, + "btc_dominance": 0.05, + "eth_btc": 0.15, + "stablecoin_supply": 0.05, + "etf_flow": 0.20, + "btc_open_interest": 0.05, +} + + +def compute_composite(values: dict) -> tuple[Optional[float], Optional[str]]: + """Return (score in [-100, +100], regime_label) or (None, None) if there + isn't enough data to score. + + `values` keys must match WEIGHTS keys (without "_signal" suffix). + """ + pairs = [ + ("ahr999", _ahr999_signal(values.get("ahr999"))), + ("altcoin_season", _altseason_signal(values.get("altcoin_season_index"))), + ("fear_greed", _fear_greed_signal(values.get("fear_greed"))), + ("btc_dominance", _btc_dominance_signal(values.get("btc_dominance_pct"))), + ("eth_btc", _eth_btc_signal(values.get("eth_btc_ratio"))), + ("stablecoin_supply", _stablecoin_supply_signal(values.get("stablecoin_supply_usd"))), + ("etf_flow", _etf_flow_signal(values.get("etf_flow_net_usd_1d"))), + ("btc_open_interest", _open_interest_signal(values.get("btc_open_interest_usd"))), + ] + alive = [(k, s) for k, s in pairs if s is not None] + if not alive: + return None, None + + total_w = sum(WEIGHTS[k] for k, _ in alive) + if total_w <= 0: + return None, None + + score = sum(WEIGHTS[k] * s for k, s in alive) / total_w * 100 + + if score >= 60: label = "BULL" + elif score >= 20: label = "BULLISH" + elif score > -20: label = "NEUTRAL" + elif score > -60: label = "BEARISH" + else: label = "BEAR" + + return round(score, 1), label diff --git a/scripts/launch_seed.py b/scripts/launch_seed.py new file mode 100755 index 0000000..0a23169 --- /dev/null +++ b/scripts/launch_seed.py @@ -0,0 +1,290 @@ +#!/usr/bin/env python3 +""" +Pre-launch data preparation — conservative version. + +What it does: + 1. DROP obsolete test-only sources (test/phase1/breakout/rsi_reversal/sma_reclaim) + and their orphan bot_trades. These are residue from past scanner experiments + that no longer fire — they only confuse the UI/sitemap. + 2. TRUNCATE the KOL window to the last KOL_RETAIN_DAYS days (default 30). + KOL data churns daily and a 30-day window is what the UI surfaces anyway. + 3. REFETCH every upstream source so the kept window is fresh: + * Trump CNN archive (idempotent; only inserts genuinely new posts) + * KOL Substack / podcast / blog polls + * KOL on-chain snapshots (HL perps + Etherscan ERC-20 balances) + * KOL divergence detector + * BTC bottom-reversal + funding-reversal scanners + +What it INTENTIONALLY DOES NOT touch: + * posts where source IN ('truth', 'btc_bottom_reversal', 'funding_reversal', + 'kol_divergence') — full history kept. Trump posts are too valuable to + truncate (analytics + accuracy backtest need them); the two BTC scanners + fire too rarely (≤4×/cycle) to safely drop any historical fire. + * bot_trades older than KOL_RETAIN_DAYS — real PnL history stays. Only + bot_trades whose trigger_post has been deleted (i.e. the trade fired + off an obsolete-source signal) are pruned. + * subscriptions, telegram_bindings, kol_wallets, candles — untouched. + * The most-recent KOL holdings snapshot per wallet — kept as the diff + baseline for the next on-chain poll (without it, the next snapshot + would diff against nothing and produce zero kol_holding_changes). + +Usage: + # Preview (no DB writes): + DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\ + venv/bin/python scripts/launch_seed.py --dry-run + + # Execute: + DATABASE_URL='sqlite+aiosqlite:///./trumpsignal.db' \\ + venv/bin/python scripts/launch_seed.py --yes + + # Skip the SEED step (just clean — useful if you'd rather let the + # scheduled poll jobs trigger naturally over the next 24h): + DATABASE_URL='...' venv/bin/python scripts/launch_seed.py --yes --no-seed + +NOT safe to re-run after real users are on the platform — the KOL truncation +will erase divergence/alignment context that you can't recover. +""" +from __future__ import annotations + +import argparse +import asyncio +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from sqlalchemy import delete, select, func, text +from app.database import AsyncSessionLocal +from app.models import ( + Post, KolPost, KolDivergence, KolHoldingChange, + KolHoldingSnapshot, KolWallet, BotTrade, + Subscription, TelegramBinding, +) + + +# ── Tunables ──────────────────────────────────────────────────────────────── +# Test-only sources to wipe from `posts` regardless of age. The first three are +# explicit fixtures; the latter two are old scanner experiments now archived +# under app/services/scanners/_archive/ but their historical fires linger. +OBSOLETE_SOURCES = {"test", "phase1", "breakout", "rsi_reversal", "sma_reclaim"} + +# Live production sources — never truncated by this script. Trump's full +# history is needed for /analytics + /signals/accuracy; the two BTC scanners +# fire too rarely (<5 times per market cycle) to safely lose any past fire. +LIVE_SOURCES_PROTECTED = {"truth", "btc_bottom_reversal", "funding_reversal", "kol_divergence"} + +# KOL data churns daily and the UI surfaces a max-30-day window. Anything older +# than this on the KOL side is dead weight in the DB. +KOL_RETAIN_DAYS = 30 +# ───────────────────────────────────────────────────────────────────────────── + + +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" +def bold(s): return f"\033[1m{s}\033[0m" + + +async def report_counts(label: str) -> dict: + """Snapshot row counts across every signal-bearing table.""" + counts: dict = {} + async with AsyncSessionLocal() as db: + for tbl, name in [ + (Post, "posts"), + (KolPost, "kol_posts"), + (KolDivergence, "kol_divergence"), + (KolHoldingChange, "kol_holding_changes"), + (KolHoldingSnapshot, "kol_holdings_snapshots"), + (KolWallet, "kol_wallets"), + (BotTrade, "bot_trades"), + (Subscription, "subscriptions"), + (TelegramBinding, "telegram_bindings"), + ]: + n = (await db.execute(select(func.count(tbl.id)))).scalar() or 0 + counts[name] = n + + by_src = (await db.execute( + select(Post.source, func.count(Post.id)).group_by(Post.source) + )).all() + counts["_posts_by_source"] = dict(by_src) + + print(bold(f"\n── {label} ──")) + for k, v in counts.items(): + if k.startswith("_"): continue + print(f" {k:25s} {v}") + print(f" posts breakdown: {counts['_posts_by_source']}") + return counts + + +async def wipe_phase(dry_run: bool) -> None: + """Delete only obsolete sources + truncate KOL window. Live sources stay.""" + kol_cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=KOL_RETAIN_DAYS) + + print(bold("\n── WIPE ──")) + print(f" Obsolete-source posts to drop: {sorted(OBSOLETE_SOURCES)}") + print(f" KOL retention window: last {KOL_RETAIN_DAYS} days " + f"(cutoff = {kol_cutoff.date()})") + print(f" Live sources kept in full: {sorted(LIVE_SOURCES_PROTECTED)}") + + # Quote source list once for repeated reuse — SQLite needs IN (...) with + # parentheses even for a single value. + obsolete_in = "(" + ",".join(f"'{s}'" for s in OBSOLETE_SOURCES) + ")" + + async with AsyncSessionLocal() as db: + previews = [ + ("kol_divergence", + f"divergences older than {KOL_RETAIN_DAYS}d", + f"SELECT COUNT(*) FROM kol_divergence WHERE post_at < '{kol_cutoff.isoformat()}'"), + ("kol_holding_changes", + f"changes older than {KOL_RETAIN_DAYS}d", + f"SELECT COUNT(*) FROM kol_holding_changes WHERE detected_at < '{kol_cutoff.isoformat()}'"), + ("kol_holdings_snapshots", + f"snapshots older than {KOL_RETAIN_DAYS}d (keeping latest-per-wallet as diff baseline)", + f"""SELECT COUNT(*) FROM kol_holdings_snapshots + WHERE snapshot_date < '{kol_cutoff.date()}' + AND id NOT IN ( + SELECT MAX(id) FROM kol_holdings_snapshots GROUP BY wallet_id + )"""), + ("kol_posts", + f"KOL posts older than {KOL_RETAIN_DAYS}d", + f"SELECT COUNT(*) FROM kol_posts WHERE published_at < '{kol_cutoff.isoformat()}'"), + ("posts", + "obsolete-source posts (test/phase1/breakout/rsi/sma)", + f"SELECT COUNT(*) FROM posts WHERE source IN {obsolete_in}"), + ("bot_trades", + "trades whose trigger_post is being deleted (orphan cleanup)", + f"""SELECT COUNT(*) FROM bot_trades WHERE trigger_post_id IN ( + SELECT id FROM posts WHERE source IN {obsolete_in} + )"""), + ] + for table, desc, sql in previews: + n = (await db.execute(text(sql))).scalar() or 0 + tag = yellow(f" would delete {n:5d}") if dry_run else green(f" delete {n:5d}") + print(f"{tag} from {table:25s} ({desc})") + + if dry_run: + print(yellow("\n [DRY RUN] no DB changes made.")) + return + + # Execute in child→parent order to avoid FK violations on Postgres + # (SQLite usually has FK enforcement off, but PG enforces it strictly). + await db.execute(text( + f"DELETE FROM kol_divergence WHERE post_at < '{kol_cutoff.isoformat()}'" + )) + await db.execute(text( + f"DELETE FROM kol_holding_changes WHERE detected_at < '{kol_cutoff.isoformat()}'" + )) + await db.execute(text(f""" + DELETE FROM kol_holdings_snapshots + WHERE snapshot_date < '{kol_cutoff.date()}' + AND id NOT IN ( + SELECT MAX(id) FROM kol_holdings_snapshots GROUP BY wallet_id + ) + """)) + await db.execute(text( + f"DELETE FROM kol_posts WHERE published_at < '{kol_cutoff.isoformat()}'" + )) + # bot_trades FIRST (it references posts.id) then posts. + await db.execute(text(f""" + DELETE FROM bot_trades WHERE trigger_post_id IN ( + SELECT id FROM posts WHERE source IN {obsolete_in} + ) + """)) + await db.execute(text( + f"DELETE FROM posts WHERE source IN {obsolete_in}" + )) + await db.commit() + print(green(" ✓ wipe committed")) + + +async def seed_real_data() -> None: + """Re-fetch every upstream source. All are idempotent on (source, external_id).""" + print(bold("\n── SEED (re-running upstream fetches) ──")) + + print("\n [1/5] Trump backfill (CNN archive)...") + from app.scrapers.truth_social import backfill_history + try: + await backfill_history(AsyncSessionLocal, limit=500) + print(green(" ✓ done")) + except Exception as e: + print(red(f" ✗ FAILED: {type(e).__name__}: {e}")) + + print("\n [2/5] KOL Substack/podcast/blog polling...") + from app.services.kol_substack import run_substack_poll + try: + results = await run_substack_poll(analyze=True) + total_new = sum(r.get("new", 0) for r in results) + total_err = sum(r.get("errors", 0) for r in results) + print(green(f" ✓ done — {total_new} new posts ({total_err} errors)")) + except Exception as e: + print(red(f" ✗ FAILED: {type(e).__name__}: {e}")) + + print("\n [3/5] KOL on-chain snapshot (HL perps + Etherscan)...") + from app.services.kol_onchain import run_onchain_poll + try: + result = await run_onchain_poll() + print(green(f" ✓ done — {result}")) + except Exception as e: + print(red(f" ✗ FAILED: {type(e).__name__}: {e}")) + + print("\n [4/5] KOL divergence detection (talks vs trades)...") + from app.services.kol_divergence import run_divergence_scan + try: + # Run against the kept KOL window so we catch every still-relevant pair. + new = await run_divergence_scan(lookback_days=KOL_RETAIN_DAYS) + print(green(f" ✓ done — {len(new)} new divergence/alignment pairs")) + except Exception as e: + print(red(f" ✗ FAILED: {type(e).__name__}: {e}")) + + print("\n [5/5] BTC bottom + funding reversal scanners (exercise the path)...") + from app.services.scanners.btc_bottom_reversal import scan_once as btc_scan + from app.services.scanners.funding_reversal import scan_once as funding_scan + for name, fn in [("btc_bottom", btc_scan), ("funding_reversal", funding_scan)]: + try: + await fn() + print(green(f" ✓ {name} scan completed (fire conditional on market state)")) + except Exception as e: + print(red(f" ✗ {name} FAILED: {type(e).__name__}: {e}")) + + +async def main() -> int: + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--dry-run", action="store_true", + help="show what would be deleted, no DB writes") + p.add_argument("--yes", action="store_true", + help="actually perform the wipe (required without --dry-run)") + p.add_argument("--no-seed", action="store_true", + help="skip the upstream re-fetch step") + args = p.parse_args() + + if not args.dry_run and not args.yes: + print(red("Refusing to run without --yes (or use --dry-run to preview).")) + return 2 + + before = await report_counts("BEFORE") + await wipe_phase(dry_run=args.dry_run) + + if args.dry_run: + print(yellow("\n[DRY RUN COMPLETE] re-run with --yes to execute.")) + return 0 + + if not args.no_seed: + await seed_real_data() + + after = await report_counts("AFTER") + + print(bold("\n── DELTA ──")) + for k in sorted(before): + if k.startswith("_"): continue + d = after[k] - before[k] + arrow = green(f"+{d}") if d > 0 else (red(str(d)) if d < 0 else " ·") + print(f" {k:25s} {before[k]:6d} → {after[k]:6d} ({arrow})") + + print(bold(green("\n✓ reset complete. Next: run scripts/launch_smoke.py after the backend is up."))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/scripts/launch_smoke.py b/scripts/launch_smoke.py new file mode 100755 index 0000000..6da6862 --- /dev/null +++ b/scripts/launch_smoke.py @@ -0,0 +1,301 @@ +#!/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())) diff --git a/scripts/verify.sh b/scripts/verify.sh new file mode 100755 index 0000000..7a6a35b --- /dev/null +++ b/scripts/verify.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT_DIR" + +echo "[verify] Running backend tests" +PYTHONPATH=. venv/bin/pytest tests -q + +if [[ -n "${BASE_URL:-}" ]]; then + echo "[verify] Running launch smoke against ${BASE_URL}" + PYTHONPATH=. venv/bin/python scripts/launch_smoke.py --base "$BASE_URL" +else + echo "[verify] Skipping launch smoke (set BASE_URL to enable it)" +fi diff --git a/tests/test_macro_fetchers_timing.py b/tests/test_macro_fetchers_timing.py new file mode 100644 index 0000000..5b1815c --- /dev/null +++ b/tests/test_macro_fetchers_timing.py @@ -0,0 +1,50 @@ +from datetime import datetime, timezone + +from app.services.macro.fetchers import ( + _drop_in_progress_daily_klines, + _latest_closed_daily_point, + _parse_farside_latest_total, +) + + +def test_drop_in_progress_daily_klines_removes_today_open_bar(): + now = datetime(2026, 5, 25, 8, 0, tzinfo=timezone.utc) + rows = [ + [1779494400000, 0, 0, 0, "76715.20"], + [1779580800000, 0, 0, 0, "77030.30"], + [1779667200000, 0, 0, 0, "77404.80"], # 2026-05-25 00:00 UTC, still open + ] + + filtered = _drop_in_progress_daily_klines(rows, now=now) + + assert [row[0] for row in filtered] == [1779494400000, 1779580800000] + + +def test_latest_closed_daily_point_skips_today_point(): + now = datetime(2026, 5, 25, 8, 0, tzinfo=timezone.utc) + rows = [ + {"timestamp": 1779494400000, "sumOpenInterestValue": "1"}, + {"timestamp": 1779580800000, "sumOpenInterestValue": "2"}, + {"timestamp": 1779667200000, "sumOpenInterestValue": "3"}, + ] + + latest = _latest_closed_daily_point(rows, now=now) + + assert latest == rows[1] + + +def test_parse_farside_latest_total_uses_newest_date_not_first_row(): + html = """ + + + + + + +
11 Jan 20240.0655.3
24 May 20260.0(12.5)
25 May 20260.0321.0
+ """ + + parsed = _parse_farside_latest_total(html) + + assert parsed["value"] == 321_000_000.0 + assert parsed["raw"]["date"] == "25 May 2026"