From fd2e61dc48e94f891f96d66c6769ed293e7f7eaf Mon Sep 17 00:00:00 2001 From: k Date: Fri, 29 May 2026 23:34:20 +0800 Subject: [PATCH] feat(x): viral live-prediction X auto-poster for Trump signals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When Trump posts an actionable buy/short signal, instantly tweet the call (entry/direction, hard time window, "ratio me" hook), then post a result follow-up X_FOLLOWUP_MINUTES later reporting the realised 15-min move plus a recent hit-rate brag with the site link. - New app/services/x_poster.py: OAuth 1.0a signed (stdlib hmac, no new dep, httpx reused). Fire-and-forget, fully degradable — no-op unless X creds + x_enabled. Dry-run logs tweet text. In-memory UTC-daily cap. 280-char hard cap on all templates (no Premium). - Gated to source=truth buy/short above x_min_confidence; scanner signals excluded (don't fit the "Trump just posted" format). - Hooked next to notify_signal in /signals/ingest + both Trump pollers (dedup via unique external_id → single tweet per post). - config.py + .env.example: X_API_KEY/SECRET, X_ACCESS_TOKEN/SECRET, X_ENABLED, X_MIN_CONFIDENCE, X_DAILY_CAP, X_FOLLOWUP_MINUTES, X_LINK_URL. 72 tests pass. Co-Authored-By: Claude Opus 4.8 --- .env.example | 21 +++ app/api/signals.py | 8 + app/config.py | 26 +++ app/scrapers/trumpstruth.py | 5 + app/scrapers/truth_social.py | 5 + app/services/x_poster.py | 307 +++++++++++++++++++++++++++++++++++ 6 files changed, 372 insertions(+) create mode 100644 app/services/x_poster.py diff --git a/.env.example b/.env.example index 00dbc3f..a11ac79 100644 --- a/.env.example +++ b/.env.example @@ -69,3 +69,24 @@ HL_MAINNET=true # auto-runs create_all() on startup (Alembic bypass — DEV ONLY). # production → both are off; schema strictly Alembic-managed. ENVIRONMENT=production + +# ── X (Twitter) auto-poster ─────────────────────────────────────────────────── +# Viral "live prediction" tweets: when Trump posts an actionable signal, the bot +# tweets the call instantly, then posts a result follow-up X_FOLLOWUP_MINUTES later. +# OAuth 1.0a User Context. From developer.x.com → your App → "Keys and tokens". +# ⚠️ App permissions MUST be "Read and Write" — then REGENERATE the access token. +# ⚠️ Do NOT use Bearer Token / Client ID / Client Secret (those are read/OAuth2). +X_API_KEY= +X_API_SECRET= +X_ACCESS_TOKEN= +X_ACCESS_SECRET= +# Master switch. false = dry-run (logs tweet text, sends nothing) even with keys set. +X_ENABLED=false +# Only tweet when ai_confidence >= this (0–100). +X_MIN_CONFIDENCE=60 +# Hard daily tweet cap (initial + follow-ups), resets at UTC midnight. +X_DAILY_CAP=40 +# Minutes after the prediction tweet to post the result follow-up. +X_FOLLOWUP_MINUTES=15 +# Public URL appended to the follow-up tweet (defaults to FRONTEND_URL if empty). +X_LINK_URL= diff --git a/app/api/signals.py b/app/api/signals.py index c9b6854..720a416 100644 --- a/app/api/signals.py +++ b/app/api/signals.py @@ -219,6 +219,14 @@ async def ingest_signal( except Exception as exc: logger.warning("Telegram notify failed for signal %d: %s", post.id, exc) + # Auto-post a viral "live prediction" tweet (fire-and-forget; no-op unless + # X is configured + enabled). Internally gated to source=truth buy/short. + try: + from app.services.x_poster import notify_x_signal + notify_x_signal(post) + except Exception as exc: + logger.warning("X notify failed for signal %d: %s", post.id, exc) + return SignalIngestResponse(status="accepted", post_id=post.id) diff --git a/app/config.py b/app/config.py index 6220726..54eb205 100644 --- a/app/config.py +++ b/app/config.py @@ -64,6 +64,32 @@ class Settings(BaseSettings): # (no trading details, no /adopt CTA). Leave empty to disable. telegram_public_channel_id: str = "" + # ── X (Twitter) auto-poster ────────────────────────────────────────────── + # OAuth 1.0a User Context — for posting from a single fixed bot account + # (@TrumpAlpha_signals). Get all four from developer.x.com → your App → + # "Keys and tokens". App permissions MUST be "Read and Write", and the + # Access Token/Secret must be regenerated AFTER setting write permission. + # Bearer Token / Client ID / Client Secret are NOT used (those are read / + # OAuth2 web-flow credentials). Empty = X posting disabled (no-op). + x_api_key: str = "" # = Consumer Key + x_api_secret: str = "" # = Consumer Secret + x_access_token: str = "" + x_access_secret: str = "" + # Master switch. Even with keys set, posting stays off unless this is true. + # Lets you load creds without going live (dry-run = logs the tweet text). + x_enabled: bool = False + # Only post when ai_confidence >= this (0–100). Keeps low-quality noise off + # the public timeline. Scanner signals carry their own score. + x_min_confidence: int = 60 + # Hard daily cap on tweets (initial + follow-ups counted together) to stay + # under Free-tier 1,500/mo and avoid spam flags. In-memory, resets at UTC + # midnight. Single-process by design so a plain counter is safe. + x_daily_cap: int = 40 + # Minutes after the initial prediction tweet to post the result follow-up. + x_followup_minutes: int = 15 + # Public site URL appended to the follow-up tweet. Falls back to frontend_url. + x_link_url: str = "" + model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "ignore"} diff --git a/app/scrapers/trumpstruth.py b/app/scrapers/trumpstruth.py index 91b78c7..3ebfdac 100644 --- a/app/scrapers/trumpstruth.py +++ b/app/scrapers/trumpstruth.py @@ -152,6 +152,11 @@ async def poll_trumpstruth(db_session_factory) -> None: notify_signal(post) except Exception as exc: logger.warning("Telegram notify failed for post %d: %s", post.id, exc) + try: + from app.services.x_poster import notify_x_signal + notify_x_signal(post) + except Exception as exc: + logger.warning("X notify failed for post %d: %s", post.id, exc) try: from app.services.bot_engine import process_post await process_post(post, db) diff --git a/app/scrapers/truth_social.py b/app/scrapers/truth_social.py index d864af8..e40708e 100644 --- a/app/scrapers/truth_social.py +++ b/app/scrapers/truth_social.py @@ -234,6 +234,11 @@ async def poll_truth_social(db_session_factory) -> None: notify_signal(post) except Exception as exc: logger.warning("Telegram notify failed for post %d: %s", post.id, exc) + try: + from app.services.x_poster import notify_x_signal + notify_x_signal(post) + except Exception as exc: + logger.warning("X notify failed for post %d: %s", post.id, exc) try: from app.services.bot_engine import process_post await process_post(post, db) diff --git a/app/services/x_poster.py b/app/services/x_poster.py new file mode 100644 index 0000000..2a97abb --- /dev/null +++ b/app/services/x_poster.py @@ -0,0 +1,307 @@ +""" +X (Twitter) auto-poster — viral "live prediction" tweets for actionable signals. + +Mirrors the design of telegram.py: fire-and-forget, fully degradable. If X +creds are missing or `x_enabled` is False, every entry point becomes a no-op +(dry-run logs the tweet text but sends nothing). A signal MUST never be blocked +by an X API failure. + + notify_x_signal(post) ← call right after notify_signal(post) in the ingest path + +Flow per actionable signal: + 1. Immediately post the PREDICTION tweet (entry / TP / SL, hard time window). + 2. Schedule a FOLLOW-UP tweet `x_followup_minutes` later that reports the + actual move (price_impact_m15) — the "told you" brag that drives the link. + +Auth: OAuth 1.0a User Context, signed by hand with stdlib hmac/hashlib (no extra +dependency — httpx is already vendored). Posting uses POST /2/tweets. + +Tone: deliberately provocative / "黑红" — invites ratios, taunts CT, no soft CTA. +All templates are hard-capped at 280 chars (no X Premium = 280 limit). +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import hmac +import logging +import secrets +import time +import urllib.parse +from datetime import datetime, timezone +from typing import Optional + +import httpx +from sqlalchemy import select + +from app.config import settings +from app.database import AsyncSessionLocal as async_session +from app.models import Post + +logger = logging.getLogger(__name__) + +TWEET_URL = "https://api.twitter.com/2/tweets" +MAX_TWEET = 280 + +# In-memory daily cap counter. Single-process by design (see main.py leader +# guard), so a plain dict is safe. Resets when the UTC date rolls over. +_sent_today = 0 +_sent_date: Optional[str] = None + + +# ── credential / enablement gate ────────────────────────────────────────── + + +def _creds_ok() -> bool: + return bool( + settings.x_api_key + and settings.x_api_secret + and settings.x_access_token + and settings.x_access_secret + ) + + +def _under_daily_cap() -> bool: + """True if we still have budget today. Increments are done by _record_sent.""" + global _sent_today, _sent_date + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + if today != _sent_date: + _sent_date = today + _sent_today = 0 + return _sent_today < settings.x_daily_cap + + +def _record_sent() -> None: + global _sent_today + _sent_today += 1 + + +# ── OAuth 1.0a signing ───────────────────────────────────────────────────── + + +def _percent(s: str) -> str: + # RFC 3986 — OAuth requires ~ unescaped and everything else strict. + return urllib.parse.quote(str(s), safe="~") + + +def _oauth_header(method: str, url: str) -> str: + """Build the OAuth 1.0a Authorization header for a JSON-body request. + + For POST /2/tweets the body is JSON (not form-encoded), so per the OAuth + spec the body params are NOT part of the signature base string — only the + oauth_* params are. This is the documented X v2 behaviour. + """ + oauth = { + "oauth_consumer_key": settings.x_api_key, + "oauth_nonce": secrets.token_hex(16), + "oauth_signature_method": "HMAC-SHA1", + "oauth_timestamp": str(int(time.time())), + "oauth_token": settings.x_access_token, + "oauth_version": "1.0", + } + # Signature base string + param_str = "&".join( + f"{_percent(k)}={_percent(v)}" for k, v in sorted(oauth.items()) + ) + base = "&".join([method.upper(), _percent(url), _percent(param_str)]) + signing_key = f"{_percent(settings.x_api_secret)}&{_percent(settings.x_access_secret)}" + digest = hmac.new( + signing_key.encode(), base.encode(), hashlib.sha1 + ).digest() + oauth["oauth_signature"] = base64.b64encode(digest).decode() + + header = "OAuth " + ", ".join( + f'{_percent(k)}="{_percent(v)}"' for k, v in sorted(oauth.items()) + ) + return header + + +async def _post_tweet(text: str, reply_to: Optional[str] = None) -> Optional[str]: + """POST a single tweet. Returns the new tweet id, or None on failure / dry-run. + + `reply_to` chains the follow-up under the prediction tweet (a thread), which + reads better and keeps the brag attached to the original call. + """ + text = text[:MAX_TWEET] + + if not settings.x_enabled: + logger.info("[x_poster] DRY-RUN (x_enabled=False), would tweet:\n%s", text) + return None + if not _creds_ok(): + logger.warning("[x_poster] missing X creds — skipping tweet") + return None + if not _under_daily_cap(): + logger.warning("[x_poster] daily cap (%d) reached — skipping tweet", + settings.x_daily_cap) + return None + + payload: dict = {"text": text} + if reply_to: + payload["reply"] = {"in_reply_to_tweet_id": reply_to} + + headers = { + "Authorization": _oauth_header("POST", TWEET_URL), + "Content-Type": "application/json", + } + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.post(TWEET_URL, json=payload, headers=headers) + if resp.status_code in (200, 201): + _record_sent() + tid = resp.json().get("data", {}).get("id") + logger.info("[x_poster] tweeted id=%s (%d/%d today)", + tid, _sent_today, settings.x_daily_cap) + return tid + logger.warning("[x_poster] tweet failed %s: %s", + resp.status_code, resp.text[:300]) + return None + except Exception as exc: # noqa: BLE001 — never let X break ingestion + logger.warning("[x_poster] tweet exception: %s", exc) + return None + + +# ── win-rate helper (for the brag in the follow-up) ──────────────────────── + + +async def _recent_hit_rate(db) -> tuple[int, Optional[float]]: + """(count, hit_rate_pct) over recent scored directional posts. + + A 'hit' = the realised 15-min move agreed with the call direction + (buy → up, short → down). Cheap, audit-friendly, and honest: it reads the + same price_impact_m15 we already record. Returns (n, None) if too few. + """ + rows = ( + await db.execute( + select(Post.signal, Post.price_impact_m15) + .where( + Post.signal.in_(("buy", "short")), + Post.price_impact_m15.isnot(None), + ) + .order_by(Post.id.desc()) + .limit(200) + ) + ).all() + n = len(rows) + if n < 5: + return n, None + hits = 0 + for sig, move in rows: + if move is None: + continue + if (sig == "buy" and move > 0) or (sig == "short" and move < 0): + hits += 1 + return n, round(100.0 * hits / n, 1) + + +# ── tweet templates (≤280, 黑红 tone) ─────────────────────────────────────── + + +def _dir_word(signal: Optional[str]) -> str: + return "LONG" if signal == "buy" else "SHORT" if signal == "short" else "WATCH" + + +def format_prediction(post: Post) -> str: + """The instant 'Trump just posted, here's my call' tweet.""" + asset = (post.target_asset or "BTC").upper() + d = _dir_word(post.signal) + entry = post.price_at_post + mins = settings.x_followup_minutes + + entry_line = f"Entry ~${entry:,.0f}\n" if entry else "" + # Hard, falsifiable, time-boxed → invites screenshots and ratios. + text = ( + f"Trump just posted. 🟠\n\n" + f"{asset}: {d} — moves within {mins} min.\n" + f"{entry_line}" + f"\nScreenshot this. Wrong? Ratio me.\n" + f"Right? You already know. ⏰" + ) + return text[:MAX_TWEET] + + +def format_followup(post: Post, move_pct: Optional[float], + n: int, hit_rate: Optional[float]) -> str: + """The '15 min later, told you' brag. Drives the only link we ever post.""" + asset = (post.target_asset or "BTC").upper() + mins = settings.x_followup_minutes + link = (settings.x_link_url or settings.frontend_url or "").rstrip("/") + + if move_pct is None: + result = f"{asset}: still cooking. Check the board 👇" + else: + arrow = "✅" if ( + (post.signal == "buy" and move_pct > 0) + or (post.signal == "short" and move_pct < 0) + ) else "🤡" + result = f"{asset}: {move_pct:+.1f}% {arrow}" + + rate_line = "" + if hit_rate is not None: + rate_line = f"\n{hit_rate:.0f}% hit rate over {n} calls. No CT badge needed." + + tail = f"\n{link}" if link else "" + text = f"{mins} min later.\n\n{result}{rate_line}\n\nBookmarked yet?{tail}" + return text[:MAX_TWEET] + + +# ── dispatch ─────────────────────────────────────────────────────────────── + + +async def _x_dispatch(post_id: int) -> None: + """Post the prediction now, then schedule the follow-up brag.""" + async with async_session() as db: + post = await db.get(Post, post_id) + if not post: + return + # Only post Trump-sourced actionable calls — that's the viral hook. + # (BTC/funding scanner signals are slow-burn; they don't fit the + # "Trump just posted, watch this" format.) + if post.source != "truth" or post.signal not in ("buy", "short"): + return + if post.ai_confidence and post.ai_confidence < settings.x_min_confidence: + logger.info("[x_poster] post=%d conf %s < %d — skipping", + post_id, post.ai_confidence, settings.x_min_confidence) + return + + pred_text = format_prediction(post) + + tweet_id = await _post_tweet(pred_text) + + # Schedule the follow-up. Fire-and-forget asyncio task (single process by + # design). A restart within the window loses the pending follow-up — that's + # acceptable for a marketing post; the signal itself is already safe. + delay = max(0, settings.x_followup_minutes) * 60 + asyncio.create_task(_followup_later(post_id, tweet_id, delay)) + + +async def _followup_later(post_id: int, reply_to: Optional[str], delay: int) -> None: + try: + await asyncio.sleep(delay) + async with async_session() as db: + post = await db.get(Post, post_id) + if not post: + return + move = post.price_impact_m15 + n, rate = await _recent_hit_rate(db) + text = format_followup(post, move, n, rate) + await _post_tweet(text, reply_to=reply_to) + except asyncio.CancelledError: # shutdown — drop quietly + raise + except Exception as exc: # noqa: BLE001 + logger.warning("[x_poster] follow-up failed post=%d: %s", post_id, exc) + + +def notify_x_signal(post: Post) -> None: + """Fire-and-forget entry point. Mirrors telegram.notify_signal. Safe to call + unconditionally — degrades to a no-op when X is disabled/unconfigured.""" + if not _creds_ok() and not settings.x_enabled: + # Neither configured nor in dry-run intent → silent skip. + return + if not post or not post.id: + return + try: + asyncio.create_task(_x_dispatch(post.id)) + except RuntimeError: + logger.warning("[x_poster] no running event loop, skipping post=%s", post.id)