diff --git a/.env.example b/.env.example index a11ac79..ce538ef 100644 --- a/.env.example +++ b/.env.example @@ -90,3 +90,9 @@ X_DAILY_CAP=40 X_FOLLOWUP_MINUTES=15 # Public URL appended to the follow-up tweet (defaults to FRONTEND_URL if empty). X_LINK_URL= + +# ── X (Twitter) READ — KOL tweet ingestion (kol_x.py) ──────────────────────── +# twitterapi.io managed-scraper key (x-api-key). Reads other people's tweets — +# unrelated to the OAuth poster creds above. Empty = X ingestion disabled. +# Get a pay-as-you-go key at twitterapi.io (~$0.15 / 1k tweets). +TWITTERAPI_IO_KEY= diff --git a/CLAUDE.md b/CLAUDE.md index 37615a8..e7ae73b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,6 +63,7 @@ Four signal sources → one bot → optional Hyperliquid execution │ 1. Trump Truth Social │── auto-classify (DeepSeek) → "buy"/"short"/"noise" │ (every post, <3s) │ if actionable: Trump scalp auto-open (System 1) └──────────────────────────┘ Tight 1.5% SL, 12h cooldown, ≥30min min-hold + Optional: post prediction tweet via x_poster.py ┌──────────────────────────┐ │ 2. Macro Vibes │── 8 daily macro indicators (AHR999, F&G, etc.) @@ -72,10 +73,11 @@ Four signal sources → one bot → optional Hyperliquid execution 5-rung stop ladder, de-risk, pyramid, peak-trail. ┌──────────────────────────┐ -│ 3. KOL talks-vs-trades │── Substack/podcast ingest + ETH on-chain diff -│ (19 KOLs, daily) │ Divergence (publicly bullish, secretly selling) +│ 3. KOL talks-vs-trades │── Substack/podcast/X ingest + ETH on-chain diff +│ (29 feeds, daily) │ Divergence (publicly bullish, secretly selling) └──────────────────────────┘ is the platform's highest-conviction signal. Telegram alert only — never auto-trades. + x_analysis.py adds real-time X post scoring. ┌──────────────────────────┐ │ 4. Funding extreme │── Hourly BTC perp funding scan @@ -102,13 +104,18 @@ linked) = Trump auto-trade + /adopt manage-only flow for sys2. - Batch / reanalysis uses `AI_MODEL` (quality, ~10s) - **Trading**: Hyperliquid SDK; API-wallet keys are envelope-encrypted with `ENCRYPTION_KEY` (KEK), per-user DEK derivation via `crypto.py`. -- **Prices**: Binance WS (`binance.py`) feeds `price_store` + powers the - `tp_sl_monitor` per-tick evaluator. +- **Prices**: Two feeds: + - `binance.py` WebSocket → BTC, ETH, SOL, TRUMP, BNB, DOGE, LINK, AAVE + - `hl_price_feed.py` polls HL `allMids` every 2s → HYPE, PURR (HL-native assets not on Binance) + Both pump `price_store` + `tp_sl_monitor` on every tick. - **Telegram**: long-poll mode (single instance), HTML messages, inline keyboards. `telegram.py` send/edit/answer + `telegram_bot.py` commands. Public channel broadcast (`TELEGRAM_PUBLIC_CHANNEL_ID` env) sends a sanitised `format_public_post` version (no execution details, tier label instead of raw confidence) after every per-user fan-out in `_dispatch`. +- **X (Twitter)**: `x_poster.py` — optional viral prediction tweets after each + actionable Trump signal. Gated by `x_enabled=False` (off by default). Full + no-op if creds missing. OAuth 1.0a hand-signed with stdlib hmac — no extra deps. --- @@ -122,13 +129,18 @@ app/ │ ├── user.py /subscribe|settings|manual-window|auto-trade │ ├── telegram.py /telegram/{preferences,bind,unbind,test} │ ├── macro.py /macro/{snapshot,history} -│ └── kol.py /kol/{posts,digest,wallets,divergence} +│ ├── kol.py /kol/{posts,digest,wallets,divergence} +│ ├── performance.py /performance ← wallet-scoped real-money stats (30d) +│ ├── funding_reversal.py /funding/snapshot ← live funding state + 7d history +│ ├── funding_signal.py /signal/{status,toggle,history} ← breakout monitor +│ └── dev.py Dev-only routes (only mounted in development env) ├── services/ │ ├── bot_engine.py ★ TRADING CORE — process_post, _execute_for_subscriber, +│ │ _broadcast_trade_alert (WS failure notifications), │ │ close_and_finalize, partial_derisk, pyramid_add │ ├── adoption.py ★ /adopt + /release flow (sys2 manage-only) │ ├── tp_sl_monitor.py Per-price-tick close evaluator. on_price_tick is -│ │ called from binance.py once per second +│ │ called from binance.py + hl_price_feed.py once/sec │ ├── hyperliquid.py HL trader (open/close/reduce/leverage) │ ├── recovery.py Startup rehydration of open BotTrades into watchdog │ ├── reconciler.py Every 60s: compare DB ↔ HL state, mark drift @@ -136,13 +148,31 @@ app/ │ ├── signal_categories.py CRITICAL CONFIG — sys1/sys2 sources, ladders, │ │ leverage clamping, protective stop formulas │ ├── regime_filter.py Sys1 only — recent-move / vol-contraction gates -│ ├── analysis.py AI signal scoring (DeepSeek) +│ ├── analysis.py AI signal scoring (DeepSeek) for Trump posts +│ ├── x_analysis.py AI scoring for X (Twitter) KOL posts. Three tiers: +│ │ TRADE_SIGNAL / DIRECTIONAL / NOISE. Strict NOISE +│ │ default — most X posts should be filtered out. +│ │ Consumed by kol_x.py. tickers come out in the +│ │ {ticker,action,conviction} shape kol_divergence reads. +│ ├── x_poster.py X (Twitter) auto-poster for Trump signals. +│ │ Fires a prediction tweet then a follow-up at +│ │ +x_followup_minutes with the actual move. Gated by +│ │ x_enabled env var (False by default). Full no-op +│ │ if creds missing — never blocks signal flow. │ ├── entry_filter.py Cheap text-based pre-filter (skip RT/URL-only) │ ├── telegram.py send_message / edit_message / answer_callback │ ├── telegram_bot.py Long-poll loop + /start /digest /adopt /release ... │ ├── telegram_digest.py Daily 3-section brief (rule-based; no LLM) │ ├── price_store.py In-memory latest price per asset │ ├── price_backfill.py Backfill historical 5min bars from Binance +│ ├── hl_price_feed.py Supplemental HL price feed for HL-native assets +│ │ (HYPE, PURR). Polls allMids every 2s. Runs +│ │ alongside binance.py. Without this, TP/SL +│ │ silently stops protecting HL-native trades. +│ ├── backtest.py Single-post backtest harness. Fetches 1m Binance +│ │ candles for [published_at, +max_hold_h] and +│ │ replays current exit rules. Conservative (uses +│ │ HIGH/LOW within bar). No fees. Batch runner on top. │ ├── crypto.py HL API-key envelope encryption │ ├── scanner_state.py In-memory toggle + observability for scanners │ ├── macro/ @@ -153,31 +183,58 @@ app/ │ │ ├── btc_bottom_reversal.py 2-of-3 AHR999 + 200WMA + Pi Bottom │ │ ├── funding_reversal.py Hourly funding extreme │ │ └── sma_reclaim.py (archive — not scheduled) -│ ├── kol_substack.py RSS ingest for 19 KOL feeds +│ ├── kol_substack.py RSS ingest for 29 KOL feeds (substack/blog/podcast) +│ ├── kol_x.py X (Twitter) ingest via twitterapi.io → x_analysis → +│ │ KolPost(source="twitter"). Daily 01:30 UTC. No-op +│ │ if twitterapi_io_key unset. Provides the post-side +│ │ feed for X-only KOLs (andrewkang, murad). │ ├── kol_onchain.py HL public API + Etherscan diff │ ├── kol_divergence.py Cross-ref talks vs trades within ±7d -│ ├── kol_analysis.py AI ticker/direction/conviction extract +│ ├── kol_analysis.py AI ticker/direction/conviction extract (Substack) │ ├── bottom_indicators.py AHR999 / Pi Cycle / 200WMA math -│ └── funding_signal.py Real-time funding extreme detector +│ ├── funding_signal.py Real-time funding extreme detector +│ └── signed_request.py EIP-191 signature verification (+ replay cache) ├── scrapers/ -│ └── truth_social.py CNN + trumpstruth.org pollers (5s interval) +│ ├── truth_social.py CNN archive poller (5s interval) — primary +│ └── trumpstruth.py trumpstruth.org RSS fallback poller. Same post id +│ hash → automatic dedup. Whoever sees first wins. +│ Offset by half the interval so the two pollers +│ don't hit upstream simultaneously. ├── ws/ -│ └── manager.py WebSocket fan-out for live UI updates +│ └── manager.py WebSocket fan-out for live UI updates. +│ Broadcasts trade_alert events (execution_failed / +│ insufficient_balance / budget_reached) via +│ _broadcast_trade_alert() in bot_engine.py. ├── models.py ★ All SQLAlchemy models in one file ├── database.py Async engine + session factory ├── config.py Pydantic Settings — reads .env └── main.py FastAPI lifespan, scheduler setup, route mount + Includes: singleton lock guard (one-leader, multi- + worker safe), deep health check /api/health/deep, + boot-grace window for price feeds. alembic/versions/ Migrations (numbered NNN). Latest = 026 026 = composite index (wallet_address, closed_at) on bot_trades scripts/ One-shot ops ├── preflight.py Pre-launch readiness gate (env / DB / TG / AI) ├── launch_smoke.py End-to-end smoke (14 checks against running API) +├── launch_seed.py Pre-launch data prep: drops test sources, trims KOL +│ window to last 30d, refetches all upstream sources. +│ Run ONCE before flipping traffic to a fresh DB. +├── seed_kol_wallets.py Seeds the KOL wallet table with known addresses. +│ Idempotent (INSERT OR IGNORE). Run once at first deploy. ├── rescore_v5.py Re-score every Post with current AI prompt ├── backfill_signals.py Fill in signal for posts missing it └── verify_sys2_lifecycle.py Manual System-2 lifecycle walk-through -tests/ pytest, 64 tests, fast (<1s total) +tests/ pytest, 83 tests, fast (<3s total) +├── test_adoption.py Adoption + release flow (snapshot-style, no real HL/AI) +├── test_telegram_digest.py Daily digest formatting +├── test_kol_x.py X ingest: dedup / mapping / no-op (mocked fetch+AI) +├── test_ratelimit.py Rate limit coverage (BUG-02 fix) +├── test_bottom_reversal_strategy.py btc_bottom_reversal 2-of-3 logic +├── test_macro_fetchers_timing.py Macro fetcher timeout + error handling +└── test_production_readiness.py Environment / config sanity checks ``` --- @@ -206,6 +263,7 @@ Confidence min 88 (platform) / user 85 (platform) Circuit breaker sys1_* sys2_* (independent) Daily budget Full daily_budget_usd n/a — user controls notional on HL Telegram alert Trump alert format Macro/funding alert + /adopt CTA +X tweet Optional prediction tweet n/a ``` **If you're tempted to put sys2 logic in `_execute_for_subscriber`**: stop. @@ -237,11 +295,11 @@ minimalism — don't extend them. circuit_breaker (sys2 CB still gates adopt!) / already_adopted / concurrency_cap (3) c. Re-read HL state inside lock (fresh entry/size/lev) - d. Resolve sys2_protective_stop_pct(HL_actual_leverage) - e. INSERT BotTrade with eff_* frozen + sys2_mode + hl_order_id="adopted:" + d. Reject if leverage > SYS2_MAX_LEVERAGE (BUG-09 fix) + e. Resolve sys2_protective_stop_pct(HL_actual_leverage) + f. INSERT BotTrade with eff_* frozen + sys2_mode + hl_order_id="adopted:" + trigger_post_id=NULL - f. register_trade() with full ladder/de-risk/addon/peak_trail - └─ Telegram confirmation w/ ladder summary + g. register_trade() with full ladder/de-risk/addon/peak_trail 4. tp_sl_monitor drives the position └─ Stop ratchet, downside de-risk partial reduces, pyramid add-ons, @@ -249,17 +307,11 @@ minimalism — don't extend them. lock-protected partial_derisk / pyramid_add / close_and_finalize. 5a. User wants out: /release - └─ release_management(wallet, trade_id): - a. Sets BotTrade.released_at = now - b. unregister(trade_id) from watchdog + └─ Sets BotTrade.released_at = now; unregister(trade_id) from watchdog └─ HL position UNTOUCHED — bot stops driving, user has manual control - └─ Across restarts: recovery skips released rows (released_at filter) - └─ Reconciler skips them. Released trades don't appear in - /positions/open or the digest "your status" line. 5b. Bot drives the close (ladder / max-hold) └─ close_and_finalize() atomic claim sets closed_at, computes pnl - └─ Trade is now CLOSED (closed_at set, exit_price + pnl_usd written) 5c. User force-closes via UI: POST /api/positions/{id}/close └─ manual_close calls close_and_finalize(force=True) — bypasses @@ -268,13 +320,30 @@ minimalism — don't extend them. 6. Recovery on restart: └─ recovery.rehydrate_open_trades reads BotTrade WHERE closed_at IS NULL AND released_at IS NULL - └─ For each: rebuild ladder from sys2_mode + (category OR adopted fallback). - The fallback for trigger_post_id IS NULL is the critical fix — - without it adopted trades lose their entire sys2 ladder on restart. + └─ For each sys2 trade: rebuild ladder from sys2_mode + adopted fallback. + Also reschedules _time_stop_check with remaining seconds (elapsed windows + fire immediately with delay=0). This is the critical fix — without it + adopted trades lose their entire sys2 ladder on restart. ``` --- +## Singleton lock guard (critical infrastructure detail) + +`main.py:lifespan` calls `_acquire_singleton_lock()` at boot using an +advisory file lock (`/tmp/trumpsignal-backend.lock`, configurable via +`SINGLETON_LOCK_PATH`). Only ONE process — "the leader" — starts the +scheduler, scrapers, price feeds, and Telegram poller. Any additional +worker (e.g. accidental `--workers 2`) serves HTTP reads only and logs +`SINGLETON LOCK NOT ACQUIRED`. The OS releases the lock automatically +when the leader exits, so a crashed leader unblocks the next start. + +The `/api/health/deep` endpoint surfaces `"is_leader": false` and adds +it to `problems[]` so uptime monitors catch mis-configured multi-worker +deployments. + +--- + ## Critical invariants checklist (when reviewing any trading change) - [ ] Does this code path respect `released_at IS NULL`? @@ -308,8 +377,7 @@ minimalism — don't extend them. 7. Add a `/yoursource on|off` command in `telegram_bot.py`. 8. If sys2: extend `signal_categories._CATEGORY_EXITS` if it needs a custom exit profile (otherwise default works). -9. Add the deep-link path in `telegram.format_post` AND `telegram.format_public_post` - (both the per-user private alert and the public channel use the same path map). +9. Add the deep-link path in `telegram.format_post` AND `telegram.format_public_post`. ### Add a new bot command @@ -328,6 +396,15 @@ minimalism — don't extend them. 2. Mirror the field on the SQLAlchemy model in `app/models.py`. 3. Apply locally: `DATABASE_URL= alembic upgrade head`. +### Enable X (Twitter) posting + +1. Create a Twitter developer app with OAuth 1.0a user context permissions. +2. Set in `.env`: `X_API_KEY`, `X_API_SECRET`, `X_ACCESS_TOKEN`, + `X_ACCESS_SECRET`, `X_ENABLED=true`. +3. Optionally tune: `X_DAILY_CAP` (default 40), `X_FOLLOWUP_MINUTES` (default 15). +4. Verify with a dry-run: set `X_ENABLED=false` and check logs — the poster + logs what it WOULD tweet without sending. + ### Deploy ```bash @@ -340,16 +417,27 @@ python scripts/launch_smoke.py --base https://api.trumpalpha.io --- +## Running it + +```bash +cd backend && python -m venv venv && source venv/bin/activate +pip install -r requirements.txt +cp .env.example .env # fill in: ENCRYPTION_KEY, AI_API_KEY, INGEST_API_KEY +uvicorn app.main:app --reload # dev only; prod uses --workers 1, no --reload +``` + +--- + ## Testing ```bash source venv/bin/activate -python -m pytest tests/ -q # full suite, ~0.5s +python -m pytest tests/ -q # 79 tests, ~2s python scripts/preflight.py # env + DB + TG + AI auth checks python scripts/launch_smoke.py # 14 end-to-end checks vs running API ``` -64 tests. Adoption + telegram_digest are snapshot-style (no real HL/AI). +Adoption + telegram_digest are snapshot-style (no real HL/AI). End-to-end trading is verified manually via the bot. --- @@ -395,6 +483,12 @@ this for you" to "you opened it, bot manages your discipline". - **`_execute_for_subscriber` has lots of `if sub["_is_system_2"]` branches.** Dead code under v2.0 (process_post early-returns for sys2). Kept for diff minimalism — don't extend or re-enable. +- **`kol_x.X_KOLS` handle ≠ X username.** `handle` is the CANONICAL key + (e.g. "cryptohayes") that MUST match `KolWallet.handle` so divergence can + join post-side ↔ on-chain; `x_username` is the screen name fetched from X. + andrewkang/murad wallets stay dark (zero divergence detections) until X + ingestion supplies their post side — that's the whole reason kol_x exists. + Empty `twitterapi_io_key` → kol_x is a full no-op (no error). - **`funding_reversal` source is in `SYSTEM_2_SOURCES`? No.** It's intentionally NOT in either supported set — it ingests as a Post for audit + sends Telegram alert via the CTA path, but doesn't trigger any @@ -407,119 +501,44 @@ this for you" to "you opened it, bot manages your discipline". Useful for telemetry filtering. - **`telegram.send_message` accepts `int | str` for `chat_id`.** Intentional. Integer = private chat, string = public channel username (e.g. `"@trumpalpha"`). - The public channel uses the string form; per-user alerts still pass integers. - **`format_public_post` deliberately omits `expected_move_pct`, `invalidation_price`, and `/adopt` CTA.** Execution-sensitive data stays - private (per-user). The public version shows confidence tier (HIGH/MED/LOW) - instead of the raw score for readability. -- **`_adopt_locks` in adoption.py looks like it should have a cap like - `_wallet_open_locks` (512).** It doesn't yet — see Known Issues below. + private. The public version shows confidence tier (HIGH/MED/LOW) instead + of the raw score. +- **`_adopt_locks` in adoption.py** is an `OrderedDict` capped at 512 with + LRU eviction — matches the `_WALLET_LOCK_MAX` pattern in `bot_engine`. +- **`trumpstruth.py` runs at half-interval offset** — the two CNN + trumpstruth + pollers are deliberately staggered so they don't hammer upstream simultaneously. + The offset is `truth_social_poll_seconds // 2`, set via APScheduler + `next_run_time` at boot. --- ## Open known issues (not blocking launch but worth fixing later) -- ~~**`_time_stop_check` tasks not rehydrated on restart** (BUG-01, FIXED 2026-05-29):~~ - `recovery.py` now imports `_time_stop_check` + `_background_tasks` from - `bot_engine` and `get_exit_profile` from `signal_categories`. After each - `register_trade()` call for a sys2 trade, it checks `exit_profile.time_stop_hours`, - computes remaining seconds, and reschedules the task. Elapsed windows (backend - was down longer than the time-stop period) fire immediately with `delay_seconds=0`. - -- ~~**Rate limit bypass via proxy x-forwarded-for** (BUG-02, HIGH, FIXED 2026-05-29):~~ - Two-part fix. (1) Frontend proxy `app/api/proxy/[...path]/route.ts` now - relays the real client IP via `x-forwarded-for` / `x-real-ip`. (2) Backend - had the matching gap: `slowapi`'s default `get_remote_address` reads - `request.client.host` (the proxy IP, since uvicorn runs without - `--proxy-headers`), so the relayed header was ignored and all users still - shared one bucket. Now a shared `app/ratelimit.py` exposes `client_ip_key` - (reads `x-forwarded-for[0]` → `x-real-ip` → peer) used by ONE shared - `limiter` across `main.py`, `posts.py`, `prices.py`. Also registered - `SlowAPIMiddleware` so `default_limits` (60/min) actually applies to every - route — previously only the 2 decorated read endpoints were limited and all - signed-mutation routes had no limit at all. Covered by `tests/test_ratelimit.py`. - -- ~~**`close_and_finalize` double-failure leaves DB/HL state inconsistent** - (BUG-03, MITIGATED 2026-05-29):~~ Full two-phase-commit is out of scope. - Mitigation: `reconciler._reconcile_wallet` now runs a "ghost position" pass — - queries DB-closed trades from the last 2h and cross-checks against HL open - positions. Mismatches are logged at ERROR level and broadcast via WS - (`reconcile_drift.ghost_positions`). Manual close required on HL UI. - -- ~~**`_adopt_locks` has no capacity cap** (BUG-04, FIXED 2026-05-29):~~ - `adoption._adopt_locks` is now an `OrderedDict` capped at `_ADOPT_LOCK_MAX=512` - with LRU eviction — matches the `_WALLET_LOCK_MAX` pattern in `bot_engine`. - -- ~~**Reconciler runs wallets sequentially** (BUG-05, FIXED 2026-05-29):~~ - `reconcile_all_once` now fans out with `asyncio.gather` + `asyncio.Semaphore(10)`. - Worst-case tail latency is `ceil(N/10) × 30s` instead of `N × 30s`. - -- ~~**`/stop` also silently disables daily digest** (BUG-07, FIXED 2026-05-29):~~ - `send_daily_digest` no longer filters on `alerts_enabled` — digest and - real-time alerts are independent. `/stop` reply updated to say "Real-time - alerts paused … send `/digest off` to stop the daily brief separately". - -- ~~**`binance.py` ASSET_MAP only had BTC + ETH** (BUG-08, CRITICAL, FIXED 2026-05-29):~~ - Trump's AI can set `target_asset` to SOL/TRUMP/BNB/DOGE/LINK/AAVE. - `ASSET_MAP` now covers all those; `BINANCE_WS_URL` is derived from it - automatically so the stream list and routing table can never diverge. - **Still missing**: HYPE (HL-native, not on Binance) — trades on HYPE fall - back to max-hold only until a supplemental HL price feed is added. - -- ~~**`_close_locks` leaked one entry per concurrent-close loser** (BUG-11, FIXED 2026-05-29):~~ - `close_and_finalize` now pops `_close_locks[trade_id]` in the - `rowcount == 0` early-return path, not just on the success paths. - -- ~~**`signed_request._seen` O(n) purge at 5000 entries** (BUG-12, FIXED 2026-05-29):~~ - Threshold reduced to `_SEEN_PURGE_THRESHOLD = 1000` and the purge now - builds an `expired` list in one pass rather than calling `dict.pop` in a loop. - -- ~~**`_get_max_leverage` makes a fresh HL `meta()` call on every trade open** - (BUG-10, FIXED 2026-05-29):~~ `_MAX_LEV_CACHE` added (same 300s TTL as - `_SZ_DECIMALS_CACHE`). Cache miss populates ALL coins from the single - `meta()` response, so concurrent opens pay one API call, not N. - -- ~~**Trump daily budget split**~~ (FIXED 2026-05-29): System-2 is manage-only - so the `sys2_pct` reservation no longer makes sense for auto-opens. - `daily_cap` for System-1 (Trump) is now `total_cap × 1.0` instead of - `total_cap × (1 - sys2_pct)`. The split logic remains intact so it would - work correctly if sys2 auto-open is ever re-enabled via ADR. - -- ~~**HL high-leverage adoption**~~ (BUG-09, FIXED 2026-05-29): `adopt_position` - now rejects positions where `leverage > SYS2_MAX_LEVERAGE` with error code - `leverage_too_high`. Previously the protective stop was computed for 10× - but applied to a 25× position — the stop was OUTSIDE the liquidation band. - - **`adopt:choose:BTC` callback may show stale prices** if user takes >60s to tap (HL fees, partial fills can change entry/size). adopt_position re-reads HL at mode-tap time so the FROZEN BotTrade is always fresh, but the picker label could be outdated. -- **Telegram bot offset on restart**: `getUpdates` may replay last 24h of - messages on backend restart. Stale `/adopt` could re-fire. User can - /release to recover. +- ~~**Telegram bot offset on restart**~~ **FIXED 2026-06-01**: startup drain + added. Stale `/adopt` replays suppressed. -- ~~**`pyramid_add` double-add on DB commit failure** (BUG-13, FIXED 2026-05-29):~~ - Same pattern as BUG-03 / partial_derisk. HL `open_position` succeeds but the - subsequent `db.commit()` of `addon_steps_done` fails → retrigger sees same - `step_idx` and double-adds. Fix: pre-claim `addon_steps_done = step_idx + 1` - with a conditional UPDATE (WHERE `addon_steps_done == step_idx`) BEFORE calling - HL. If `rowcount == 0`, return early. Second UPDATE writes `entry_price` and - `size_usd` after fill confirmation. +- **`/adopt` picker label** can show stale price if user waits >60s to tap + (frozen BotTrade is always fresh, but the Telegram picker label may be outdated). -- ~~**`price_impact_monitor` measured wrong asset** (BUG-14, FIXED 2026-05-29):~~ - `truth_social.py` passed `analysis["asset"]` (BTC/ETH sentiment proxy) to - `register_post` instead of `target_asset` (SOL/TRUMP/etc. — the perp we actually - trade). Impact % was measuring BTC/ETH not the traded coin. Fix: introduced - `tracked_asset = analysis.get("target_asset") or asset` and used it for - `price_at_post`, `price_impact_asset`, and `register_post(asset=…)`. +**Deferred security items (need interface change or DB migration):** +- **C3**: Read endpoints pass `ts`/`sig` as URL query params → access log exposure. +- **H4**: HL API key KEK uses single unsalted SHA-256. Needs re-encryption migration. +- **M1**: Adopted sys2 positions miscounted against sys1 daily budget. +- **M5**: `GET /telegram/{wallet}/status` unauthenticated — exposes chat_id/tg_username. --- ## Repos in this project -- **This repo** (`/Users/k/Public/Claude/backend`) — Python/FastAPI backend -- **Sibling frontend** (`/Users/k/Public/Claude/trumpsignal`) — Next.js 16 +- **This repo** (`/Users/k/Public/trumpsignal/backend`) — Python/FastAPI backend +- **Sibling frontend** (`/Users/k/Public/trumpsignal/frontend`) — Next.js 16 dashboard at trumpsignal.com. See its own CLAUDE.md. Both deployed independently. Backend serves the JSON API + Telegram bot. diff --git a/alembic/versions/027_kol_post_analysis_fields.py b/alembic/versions/027_kol_post_analysis_fields.py new file mode 100644 index 0000000..135d9f1 --- /dev/null +++ b/alembic/versions/027_kol_post_analysis_fields.py @@ -0,0 +1,30 @@ +"""Add tier / post_type / talks_vs_trades_flag / sentiment to kol_posts + +Revision ID: 027 +Revises: 026 +Create Date: 2026-06-05 +""" +from alembic import op +import sqlalchemy as sa + +revision = '027' +down_revision = '026' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table('kol_posts') as batch: + batch.add_column(sa.Column('tier', sa.String(16), nullable=True)) + batch.add_column(sa.Column('post_type', sa.String(16), nullable=True)) + batch.add_column(sa.Column('talks_vs_trades_flag', sa.Boolean, + nullable=True, server_default='0')) + batch.add_column(sa.Column('sentiment', sa.String(16), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table('kol_posts') as batch: + batch.drop_column('sentiment') + batch.drop_column('talks_vs_trades_flag') + batch.drop_column('post_type') + batch.drop_column('tier') diff --git a/app/api/funding_signal.py b/app/api/funding_signal.py index edf60b7..6ed0935 100644 --- a/app/api/funding_signal.py +++ b/app/api/funding_signal.py @@ -2,12 +2,14 @@ API endpoints for the breakout signal monitor. GET /api/signal/status — current state (enabled, recent signals) -POST /api/signal/toggle — flip the on/off switch +POST /api/signal/toggle — flip the on/off switch (requires X-Ingest-Key) GET /api/signal/history — last N signals (fired regardless of enabled state) """ -from fastapi import APIRouter +from fastapi import APIRouter, Header, HTTPException +from typing import Optional +from app.config import settings from app.services.funding_signal import ( set_enabled, is_enabled, @@ -18,17 +20,29 @@ from app.services.funding_signal import ( router = APIRouter(prefix="/signal", tags=["signal"]) +def _require_ingest_key(x_ingest_key: Optional[str]) -> None: + """Fail-closed operator-only guard (same pattern as signals.py).""" + expected = settings.ingest_api_key + if not expected: + raise HTTPException(503, "toggle disabled (INGEST_API_KEY not configured)") + if not x_ingest_key: + raise HTTPException(401, "missing X-Ingest-Key header") + if x_ingest_key != expected: + raise HTTPException(401, "invalid X-Ingest-Key") + + @router.get("/status") async def status(): return get_status() @router.post("/toggle") -async def toggle(enabled: bool): +async def toggle(enabled: bool, x_ingest_key: Optional[str] = Header(default=None)): """ - Body: ?enabled=true or ?enabled=false - Example: POST /api/signal/toggle?enabled=true + Operator-only toggle — requires X-Ingest-Key header (same secret as signal ingest). + Example: POST /api/signal/toggle?enabled=true -H 'X-Ingest-Key: …' """ + _require_ingest_key(x_ingest_key) set_enabled(enabled) return {"enabled": is_enabled()} diff --git a/app/api/kol.py b/app/api/kol.py index cb8efea..dbacd48 100644 --- a/app/api/kol.py +++ b/app/api/kol.py @@ -13,7 +13,7 @@ from datetime import datetime, timedelta, timezone from typing import Any, List, Optional from fastapi import APIRouter, Depends, Header, HTTPException, Query -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings @@ -59,6 +59,11 @@ def _summary_dto(post: KolPost) -> dict: "tickers": _parse_tickers(post.tickers_json), "analyzed_at": iso_utc(post.analyzed_at), "analysis_model": post.analysis_model, + # Extended x_analysis fields (migration 027) + "tier": post.tier, + "post_type": post.post_type, + "talks_vs_trades_flag": post.talks_vs_trades_flag or False, + "sentiment": post.sentiment, } @@ -71,19 +76,47 @@ def _detail_dto(post: KolPost) -> dict: @router.get("/kol/posts") async def list_kol_posts( handle: Optional[str] = Query(default=None, description="filter by kol_handle"), - source: Optional[str] = Query(default=None, description="substack | twitter"), + source: Optional[str] = Query(default=None, description="substack | blog | podcast"), + signals_only: bool = Query(default=False, + description="exclude noise posts (tier='noise')"), + ticker: Optional[str] = Query(default=None, description="filter by ticker symbol e.g. BTC"), + days: Optional[int] = Query(default=None, ge=1, le=365, description="restrict to last N days"), limit: int = Query(default=50, ge=1, le=200), page: int = Query(default=1, ge=1), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: - stmt = select(KolPost) + base = select(KolPost) if handle: - stmt = stmt.where(KolPost.kol_handle == handle) + base = base.where(KolPost.kol_handle == handle) if source: - stmt = stmt.where(KolPost.source == source) - stmt = stmt.order_by(KolPost.published_at.desc()).offset((page - 1) * limit).limit(limit) - rows = (await db.execute(stmt)).scalars().all() - return {"items": [_summary_dto(p) for p in rows], "page": page, "limit": limit} + base = base.where(KolPost.source == source) + if signals_only: + base = base.where( + (KolPost.tier.is_(None)) | (KolPost.tier != "noise") + ) + if ticker: + # tickers_json stores [{"ticker":"BTC",...}] — match the key-value pair + base = base.where(KolPost.tickers_json.like(f'%"ticker": "{ticker.upper()}"%')) + if days: + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + base = base.where(KolPost.published_at >= cutoff) + + total: int = (await db.execute( + select(func.count()).select_from(base.subquery()) + )).scalar_one() + + rows = (await db.execute( + base.order_by(KolPost.published_at.desc()) + .offset((page - 1) * limit) + .limit(limit) + )).scalars().all() + + return { + "items": [_summary_dto(p) for p in rows], + "page": page, + "limit": limit, + "total": total, + } @router.get("/kol/posts/{post_id}") @@ -365,14 +398,23 @@ async def list_divergence( signal_type=alignment → KOL's words matched their on-chain action (reinforced signal). """ since = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days) - stmt = select(KolDivergence).where(KolDivergence.created_at >= since) + # Use post_at (when the KOL post was published) rather than created_at + # (when the divergence row was written to DB). created_at can lag by + # days or weeks when bulk scans are run, causing old event-pairs to + # appear as "recent" divergences long after they occurred. + stmt = select(KolDivergence).where(KolDivergence.post_at >= since) if handle: stmt = stmt.where(KolDivergence.handle == handle) if ticker: stmt = stmt.where(KolDivergence.ticker == ticker.upper()) if signal_type: stmt = stmt.where(KolDivergence.signal_type == signal_type) - stmt = stmt.order_by(KolDivergence.created_at.desc()).limit(200) + # Order by post_at (when the KOL actually published), consistent with the + # post_at window filter above. Ordering by created_at (DB write time) put + # a backfilled older post (post_at=05-18, created_at=05-28) ahead of a + # newer one (post_at=05-23), so the "latest divergence" read wrong. + # Tie-break on created_at so same-day posts have a stable order. + stmt = stmt.order_by(KolDivergence.post_at.desc(), KolDivergence.created_at.desc()).limit(200) rows = (await db.execute(stmt)).scalars().all() return { "window_days": days, diff --git a/app/api/performance.py b/app/api/performance.py index 071a18b..ff6d45f 100644 --- a/app/api/performance.py +++ b/app/api/performance.py @@ -76,14 +76,29 @@ async def get_performance( max_drawdown_pct=0.0, ) - winning = sum(1 for t in trades if (t.pnl_usd or 0) > 0) - win_rate = winning / total_trades + # Only include trades with a known PnL in financial statistics. + # Trades with pnl_usd=NULL were externally closed or unsettled — treating + # them as 0 silently inflates trade count and distorts win rate / net PnL. + settled = [t for t in trades if t.pnl_usd is not None] + if not settled: + return BotPerformance( + period_days=PERIOD_DAYS, + total_trades=total_trades, + win_rate=0.0, + net_pnl_usd=0.0, + avg_hold_seconds=0.0, + max_drawdown_pct=0.0, + ) - pnl_values = [(t.pnl_usd or 0.0) for t in trades] + winning = sum(1 for t in settled if t.pnl_usd > 0) # type: ignore[operator] + win_rate = winning / len(settled) + + pnl_values = [t.pnl_usd for t in settled] # type: ignore[misc] net_pnl = sum(pnl_values) - hold_values = [(t.hold_seconds or 0) for t in trades] - avg_hold = sum(hold_values) / len(hold_values) + # For hold time use all trades (we always have opened_at + closed_at when closed) + hold_values = [t.hold_seconds for t in trades if t.hold_seconds is not None] + avg_hold = sum(hold_values) / len(hold_values) if hold_values else 0.0 # Max drawdown: running peak → trough of cumulative PnL cumulative = 0.0 diff --git a/app/api/positions.py b/app/api/positions.py index 253bb19..8b3322a 100644 --- a/app/api/positions.py +++ b/app/api/positions.py @@ -199,11 +199,15 @@ async def get_today_stats( hour=0, minute=0, second=0, microsecond=0, tzinfo=None ) + # Exclude released trades — these were released back to user control and + # closed by the user on HL directly, not by the bot. Including them would + # inflate the bot's "today" P&L with trades the bot didn't manage. closed_rows = await db.execute( select(BotTrade).where( BotTrade.wallet_address == wallet, BotTrade.closed_at >= midnight, BotTrade.pnl_usd.is_not(None), + BotTrade.released_at.is_(None), ) ) closed = closed_rows.scalars().all() @@ -280,6 +284,15 @@ async def manual_close( if not wallet or not isinstance(timestamp, int) or not isinstance(signature, str): raise HTTPException(422, "wallet, timestamp, signature required") + # Verify signature first — prevents trade_id enumeration via 403/409 before auth. + verify_signed_request( + action=ACTION_CLOSE_TRADE, + wallet=wallet, + timestamp_ms=timestamp, + signature=signature, + body={"trade_id": trade_id}, + ) + # Load the trade. Must be open AND owned by the signing wallet. trade = (await db.execute(select(BotTrade).where(BotTrade.id == trade_id))).scalar_one_or_none() if trade is None: @@ -289,14 +302,6 @@ async def manual_close( if trade.closed_at is not None: raise HTTPException(409, f"trade {trade_id} is already closed") - verify_signed_request( - action=ACTION_CLOSE_TRADE, - wallet=wallet, - timestamp_ms=timestamp, - signature=signature, - body={"trade_id": trade_id}, - ) - # Find the API key from the subscription. sub = (await db.execute( select(Subscription).where(Subscription.wallet_address == wallet) @@ -330,7 +335,26 @@ async def manual_close( force=True, ) - closed = (await db.execute(select(BotTrade).where(BotTrade.id == trade_id))).scalar_one() + # B45: close_and_finalize uses its own AsyncSessionLocal session, so the + # route's `db` session has a stale identity-map cache for this trade row. + # `populate_existing=True` forces SQLAlchemy to overwrite the cached + # instance with the freshly-committed values (exit_price, pnl_usd, etc.) + # rather than returning the pre-close snapshot from the identity map. + closed = (await db.execute( + select(BotTrade).where(BotTrade.id == trade_id).execution_options(populate_existing=True) + )).scalar_one() + + # B46: close_and_finalize returns silently on certain failures (no price + # for paper close, HL returns no fill, etc.) without raising an exception. + # Detect the failure by checking whether closed_at was actually written. + if closed.closed_at is None: + raise HTTPException( + 500, + "Close command issued but the position could not be closed " + "(no price feed, HL fill failure, or another caller closed it first). " + "Refresh open positions — if it still shows, retry or close manually on HL." + ) + return CloseTradeResponse( status="ok", trade_id=trade_id, @@ -373,14 +397,6 @@ async def set_trade_grow( if body_tid != trade_id: raise HTTPException(400, "trade_id mismatch (path vs signed body)") - trade = (await db.execute(select(BotTrade).where(BotTrade.id == trade_id))).scalar_one_or_none() - if trade is None: - raise HTTPException(404, f"trade {trade_id} not found") - if trade.wallet_address.lower() != wallet: - raise HTTPException(403, "trade belongs to a different wallet") - if trade.closed_at is not None: - raise HTTPException(409, f"trade {trade_id} is already closed") - verify_signed_request( action=ACTION_SET_GROW, wallet=wallet, @@ -389,6 +405,14 @@ async def set_trade_grow( body={"trade_id": trade_id, "enabled": enabled}, ) + trade = (await db.execute(select(BotTrade).where(BotTrade.id == trade_id))).scalar_one_or_none() + if trade is None: + raise HTTPException(404, f"trade {trade_id} not found") + if trade.wallet_address.lower() != wallet: + raise HTTPException(403, "trade belongs to a different wallet") + if trade.closed_at is not None: + raise HTTPException(409, f"trade {trade_id} is already closed") + trade.grow_mode = enabled await db.commit() diff --git a/app/api/posts.py b/app/api/posts.py index 9bf3515..2f7f19a 100644 --- a/app/api/posts.py +++ b/app/api/posts.py @@ -5,16 +5,29 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import Response from app.ratelimit import limiter -from sqlalchemy import select +from sqlalchemy import case, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.models import Post, iso_utc -from app.schemas import PriceImpact, TrumpPost +from app.schemas import PostFilterCounts, PostListResponse, PriceImpact, SourceCount, TrumpPost router = APIRouter() logger = logging.getLogger(__name__) +_ARCHIVE_EXCLUDED_SOURCES = ( + "truth", + "btc_bottom_reversal", + "funding_reversal", + "kol_divergence", +) + + +_AI_SCORED_EXPR = ( + (func.coalesce(Post.ai_confidence, 0) > 0) | + Post.ai_reasoning.is_not(None) +) + def _direction_correct(signal: Optional[str], pct: Optional[float]) -> Optional[bool]: if pct is None or signal is None: @@ -98,11 +111,161 @@ async def get_posts( return [_post_to_schema(p) for p in posts] +@router.get("/posts-paged", response_model=PostListResponse) +@limiter.limit("60/minute") +async def get_posts_page( + request: Request, + limit: int = Query(default=20, ge=1, le=500), + page: int = Query(default=1, ge=1), + source: Optional[str] = Query( + default=None, + description="Filter to a single source (e.g. 'truth', 'btc_bottom_reversal').", + ), + source_in: Optional[str] = Query( + default=None, + description="Comma-separated allowlist of sources.", + ), + source_not_in: Optional[str] = Query( + default=None, + description="Comma-separated denylist of sources.", + ), + archive_only: bool = Query( + default=False, + description="When true, return only archived/retired sources (exclude live modules).", + ), + sentiment: Optional[str] = Query( + default=None, + pattern="^(bullish|bearish|neutral)$", + description="Optional sentiment filter.", + ), + signal: Optional[str] = Query( + default=None, + pattern="^(buy|short|actionable)$", + description="Optional signal filter. 'actionable' = buy or short.", + ), + ai_scored_only: bool = Query( + default=False, + description="When true, exclude off-topic rows that were skipped before AI scoring.", + ), + db: AsyncSession = Depends(get_db), + response: Response = None, +): + offset = (page - 1) * limit + stmt = select(Post) + count_stmt = select(func.count()).select_from(Post) + counts_stmt = select( + func.count().label("all_count"), + func.sum(case((Post.signal.in_(("buy", "short")), 1), else_=0)).label("actionable_count"), + func.sum(case((Post.signal == "buy", 1), else_=0)).label("buy_count"), + func.sum(case((Post.signal == "short", 1), else_=0)).label("short_count"), + func.sum(case((_AI_SCORED_EXPR, 0), else_=1)).label("off_topic_count"), + ).select_from(Post) + source_counts_stmt = select( + Post.source.label("source"), + func.count(Post.id).label("count"), + func.max(Post.published_at).label("latest"), + ).select_from(Post) + + included_sources = [s.strip() for s in (source_in or "").split(",") if s.strip()] + excluded_sources = [s.strip() for s in (source_not_in or "").split(",") if s.strip()] + if archive_only: + excluded_sources = list(dict.fromkeys([*excluded_sources, *_ARCHIVE_EXCLUDED_SOURCES])) + + if source: + stmt = stmt.where(Post.source == source) + count_stmt = count_stmt.where(Post.source == source) + counts_stmt = counts_stmt.where(Post.source == source) + source_counts_stmt = source_counts_stmt.where(Post.source == source) + elif included_sources: + stmt = stmt.where(Post.source.in_(included_sources)) + count_stmt = count_stmt.where(Post.source.in_(included_sources)) + counts_stmt = counts_stmt.where(Post.source.in_(included_sources)) + # NOTE: source_counts is the chip/source breakdown the UI renders the + # filter bar from. It must reflect every source available in the + # current view scope — NOT just the one the user has selected. So + # `source_in` (the chip-selection narrowing) is deliberately NOT + # applied here; only the exclusion filters (archive_only / + # source_not_in) below scope it. Applying it would collapse the chip + # bar to the single selected source with no way back to "all". + + if excluded_sources: + stmt = stmt.where(~Post.source.in_(excluded_sources)) + count_stmt = count_stmt.where(~Post.source.in_(excluded_sources)) + counts_stmt = counts_stmt.where(~Post.source.in_(excluded_sources)) + source_counts_stmt = source_counts_stmt.where(~Post.source.in_(excluded_sources)) + + if sentiment: + stmt = stmt.where(Post.sentiment == sentiment) + count_stmt = count_stmt.where(Post.sentiment == sentiment) + counts_stmt = counts_stmt.where(Post.sentiment == sentiment) + source_counts_stmt = source_counts_stmt.where(Post.sentiment == sentiment) + if ai_scored_only: + stmt = stmt.where(_AI_SCORED_EXPR) + count_stmt = count_stmt.where(_AI_SCORED_EXPR) + source_counts_stmt = source_counts_stmt.where(_AI_SCORED_EXPR) + + if signal == "actionable": + stmt = stmt.where(Post.signal.in_(("buy", "short"))) + count_stmt = count_stmt.where(Post.signal.in_(("buy", "short"))) + elif signal: + stmt = stmt.where(Post.signal == signal) + count_stmt = count_stmt.where(Post.signal == signal) + + stmt = stmt.order_by(Post.published_at.desc()).offset(offset).limit(limit) + result = await db.execute(stmt) + total_result = await db.execute(count_stmt) + counts_result = await db.execute(counts_stmt) + source_counts_result = await db.execute( + source_counts_stmt.group_by(Post.source).order_by(func.count(Post.id).desc(), Post.source.asc()) + ) + posts = result.scalars().all() + total = int(total_result.scalar_one() or 0) + counts_row = counts_result.one() + off_topic = int(counts_row.off_topic_count or 0) + all_count = int(counts_row.all_count or 0) + if response is not None: + response.headers["Cache-Control"] = "public, max-age=30, stale-while-revalidate=60" + return PostListResponse( + items=[_post_to_schema(p) for p in posts], + total=total, + page=page, + limit=limit, + counts=PostFilterCounts( + all=max(0, all_count - off_topic) if ai_scored_only else all_count, + actionable=int(counts_row.actionable_count or 0), + buy=int(counts_row.buy_count or 0), + short=int(counts_row.short_count or 0), + off_topic=off_topic, + ), + source_counts=[ + SourceCount( + source=row.source, + count=int(row.count or 0), + latest=iso_utc(row.latest), + ) + for row in source_counts_result.all() + ], + ) + + @router.get("/signals/accuracy") async def signal_accuracy(db: AsyncSession = Depends(get_db)): - """Aggregate accuracy of directional signals (buy/sell/short) against realised price moves.""" + """Aggregate accuracy of directional signals against realised price moves. + + Scoped to the CURRENT signal taxonomy the live bot actually trades: + * only buy/short (the retired "sell" vocabulary is excluded — the bot + emits buy/short now, and 42 legacy truth/sell rows would otherwise + pollute the public scoreboard), + * only production sources (SUPPORTED_TRADING_SOURCES) — retired/test + ingest sources like rsi_reversal, sma_reclaim, breakout, phase1 and + `test` must not appear in the public accuracy stats. + """ + from app.services.signal_categories import SUPPORTED_TRADING_SOURCES result = await db.execute( - select(Post).where(Post.signal.in_(["buy", "sell", "short"])) + select(Post).where( + Post.signal.in_(["buy", "short"]), + func.lower(Post.source).in_(SUPPORTED_TRADING_SOURCES), + ) ) posts = result.scalars().all() diff --git a/app/api/signals.py b/app/api/signals.py index 720a416..ccd1c6f 100644 --- a/app/api/signals.py +++ b/app/api/signals.py @@ -31,6 +31,7 @@ from typing import Optional from fastapi import APIRouter, Depends, Header, HTTPException from pydantic import BaseModel, Field, field_validator from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings @@ -174,8 +175,22 @@ async def ingest_signal( prefilter_reason="external_signal", # bypasses entry-filter audit ) db.add(post) - await db.commit() - await db.refresh(post) + try: + await db.commit() + await db.refresh(post) + except IntegrityError: + # Concurrent request beat us to the INSERT — fetch the existing row. + await db.rollback() + existing2 = await db.execute(select(Post).where(Post.external_id == ext_id_hashed)) + prior2 = existing2.scalar_one_or_none() + if prior2: + return SignalIngestResponse( + status="duplicate", + post_id=prior2.id, + dedup_against=prior2.id, + note=f"concurrent ingest: external_id {body.external_id!r} already exists", + ) + raise # unexpected — re-raise if we still can't find the row logger.info( "Ingested signal: source=%s id=%s → post_id=%d, %s/%s conf=%d category=%s", diff --git a/app/api/telegram.py b/app/api/telegram.py index bc72b24..c4d0c11 100644 --- a/app/api/telegram.py +++ b/app/api/telegram.py @@ -87,14 +87,21 @@ class InitResponse(BaseModel): expires_in_seconds: int -# ── Endpoints ──────────────────────────────────────────────────────────── +# ── Helpers ────────────────────────────────────────────────────────────── -@router.get("/{wallet}/status", response_model=StatusResponse) -async def status(wallet: str, db: AsyncSession = Depends(get_db)) -> StatusResponse: - """Public-by-wallet read. Returns whether server is configured AND - whether this wallet has bound a Telegram chat.""" - wallet = wallet.lower().strip() +async def _build_status( + wallet: str, + db: AsyncSession, + authenticated: bool = False, +) -> StatusResponse: + """Shared status-building logic used by both the GET endpoint and + update_preferences so both always return a consistent StatusResponse. + + B37: previously update_preferences called `await status(wallet, db)`, + which passed db as the `timestamp` positional arg and left the DI-managed + `db` param as a raw Depends() wrapper — crashing on `.execute()`. + """ configured = bool(settings.telegram_bot_token and settings.telegram_bot_username) b = (await db.execute( select(TelegramBinding).where(TelegramBinding.wallet_address == wallet) @@ -104,6 +111,14 @@ async def status(wallet: str, db: AsyncSession = Depends(get_db)) -> StatusRespo return StatusResponse(configured=configured, bot_username=settings.telegram_bot_username or None, bound=False) + + if not authenticated: + return StatusResponse( + configured=configured, + bot_username=settings.telegram_bot_username or None, + bound=True, + ) + return StatusResponse( configured=configured, bot_username=settings.telegram_bot_username or None, @@ -123,6 +138,49 @@ async def status(wallet: str, db: AsyncSession = Depends(get_db)) -> StatusRespo ) +# ── Endpoints ──────────────────────────────────────────────────────────── + + +@router.get("/{wallet}/status", response_model=StatusResponse) +async def status( + wallet: str, + timestamp: Optional[int] = None, + signature: Optional[str] = None, + db: AsyncSession = Depends(get_db), +) -> StatusResponse: + """Wallet Telegram status. + + Unauthenticated: returns configured + bound (boolean only). + Authenticated (timestamp + signature): returns full binding details. + This prevents third parties from de-anonymising wallets via tg_username/chat_id. + """ + wallet = wallet.lower().strip() + + # Verify ownership if credentials provided (best-effort — never block on failure). + # Accept either "view_telegram_status" (dedicated action) or "view_user" + # (broad read action that the frontend already caches for other endpoints). + # Accepting view_user avoids requiring a separate wallet signature just to + # see Telegram status — the frontend reuses its cached view_user envelope. + authenticated = False + if timestamp is not None and signature: + for _action in ("view_telegram_status", "view_user"): + try: + verify_signed_request( + action=_action, + wallet=wallet, + timestamp_ms=timestamp, + signature=signature, + body=None, + allow_replay=True, + ) + authenticated = True + break + except Exception: + pass # Try next action; fall through to redacted response if all fail + + return await _build_status(wallet, db, authenticated=authenticated) + + @router.post("/{wallet}/init", response_model=InitResponse) async def init_binding( wallet: str, body: SignedEnvelope, @@ -192,7 +250,10 @@ async def update_preferences( await db.commit() await db.refresh(b) - return await status(wallet, db) + # Return full authenticated status — the caller already proved ownership + # via the signed request verified above (B37: was `await status(wallet, db)` + # which passed db as the timestamp arg, crashing inside status()). + return await _build_status(wallet, db, authenticated=True) @router.post("/{wallet}/unbind") diff --git a/app/api/trades.py b/app/api/trades.py index 99973c0..398463a 100644 --- a/app/api/trades.py +++ b/app/api/trades.py @@ -19,27 +19,34 @@ ACTION_VIEW_USER = "view_user" def _trade_to_schema(trade: BotTrade) -> BotTradeSchema: - # Join against trigger_post to surface the source tag. When a trade was - # opened by an ingested signal (VCP scanner, user's module, etc.) the - # source reveals WHICH module produced it — critical for "is module X - # actually making money?" attribution analysis. trigger_source = None + trigger_post_text = None if trade.trigger_post is not None: trigger_source = trade.trigger_post.source - # Paper trades are tagged via hl_order_id at open time; that's the only - # stable signal we have to distinguish them in aggregate views. + trigger_post_text = trade.trigger_post.text + elif trade.hl_order_id and str(trade.hl_order_id).startswith("adopted:"): + # Adopted (sys2 manage-only) trades have trigger_post_id=NULL by + # construction, so trigger_source would otherwise fall through to + # "Unknown" in the UI. The hl_order_id prefix is the reliable marker + # (see adoption.py / CLAUDE.md) — surface it as a proper attribution. + trigger_source = "adopted" is_paper = (trade.hl_order_id == "paper") + # Preserve None for nullable fields — do NOT coerce to 0 / "". + # The frontend uses null to distinguish "unknown/not yet settled" from + # a genuine zero (e.g. a break-even trade, a 0-second hold). + # `or 0` was silently masking externally-closed trades and unsettled PnL. return BotTradeSchema( id=trade.id, asset=trade.asset, side=trade.side, entry_price=trade.entry_price, - exit_price=trade.exit_price or 0.0, - pnl_usd=trade.pnl_usd or 0.0, - hold_seconds=trade.hold_seconds or 0, - trigger_post_id=trade.trigger_post_id or 0, + exit_price=trade.exit_price, # None = position still open or extern-closed + pnl_usd=trade.pnl_usd, # None = unsettled / extern-closed + hold_seconds=trade.hold_seconds, # None = not yet computed + trigger_post_id=trade.trigger_post_id, # None = adopted/manual, no trigger post opened_at=iso_utc(trade.opened_at) or "", - closed_at=iso_utc(trade.closed_at) or "", + closed_at=iso_utc(trade.closed_at), # None = still open (shouldn't happen here) + trigger_post_text=trigger_post_text, trigger_source=trigger_source, is_paper=is_paper, ) @@ -50,7 +57,7 @@ async def get_trades( wallet: str = Query(..., description="Wallet address (lower-cased internally)"), ts: int = Query(..., description="Signed timestamp (ms)"), sig: str = Query(..., description="EIP-191 signature"), - limit: int = Query(default=20, ge=1, le=100), + limit: int = Query(default=20, ge=1, le=500), # raised from 100: Analytics needs 500 for full history page: int = Query(default=1, ge=1), db: AsyncSession = Depends(get_db), ): @@ -71,7 +78,7 @@ async def get_trades( .options(joinedload(BotTrade.trigger_post)) .where(BotTrade.wallet_address == wallet) .where(BotTrade.closed_at.is_not(None)) - .order_by(BotTrade.opened_at.desc()) + .order_by(BotTrade.closed_at.desc()) # closed_at = settlement time; opened_at would bury recently-closed old trades .offset(offset) .limit(limit) ) diff --git a/app/api/user.py b/app/api/user.py index 455fe93..e3156c7 100644 --- a/app/api/user.py +++ b/app/api/user.py @@ -42,17 +42,18 @@ async def verify_hl_api_key_can_trade(api_key: str, account_address: str) -> Non def _trade_to_schema(trade: BotTrade) -> BotTradeSchema: + # Preserve None — do NOT coerce to 0 / "". See trades.py comment. return BotTradeSchema( id=trade.id, asset=trade.asset, side=trade.side, entry_price=trade.entry_price, - exit_price=trade.exit_price or 0.0, - pnl_usd=trade.pnl_usd or 0.0, - hold_seconds=trade.hold_seconds or 0, - trigger_post_id=trade.trigger_post_id or 0, + exit_price=trade.exit_price, + pnl_usd=trade.pnl_usd, + hold_seconds=trade.hold_seconds, + trigger_post_id=trade.trigger_post_id, opened_at=iso_utc(trade.opened_at) or "", - closed_at=iso_utc(trade.closed_at) or "", + closed_at=iso_utc(trade.closed_at), ) @@ -107,7 +108,7 @@ async def get_user_public(wallet: str, db: AsyncSession = Depends(get_db)): "wallet_address": wallet, "active": False, "hl_api_key_set": False, "paper_mode": False, "manual_window_until": None, "circuit_breaker_tripped_at": None, "circuit_breaker_reason": None, - "auto_trade": False, + "auto_trade": False, "trump_enabled": False, "macro_enabled": False, } return { "wallet_address": wallet, @@ -119,6 +120,12 @@ async def get_user_public(wallet: str, db: AsyncSession = Depends(get_db)): "circuit_breaker_tripped_at": iso_utc(sub.circuit_breaker_tripped_at), "circuit_breaker_reason": sub.circuit_breaker_reason, "auto_trade": bool(sub.auto_trade), + # Per-system enable flags. bot_engine gates System-1 on trump_enabled + # and System-2 on macro_enabled (see _execute_for_subscriber). Without + # these, the UI claims "next Trump signal auto-opens" even when + # trump_enabled=0, where the backend actually skips the trade. + "trump_enabled": bool(sub.trump_enabled), + "macro_enabled": bool(sub.macro_enabled), } @@ -179,6 +186,7 @@ async def get_user( hl_api_key_set=hl_api_key_set, hl_api_key_masked=masked, paper_mode=bool(sub.paper_mode), + auto_trade=bool(sub.auto_trade), trades=[_trade_to_schema(t) for t in trades], settings=UserSettings( leverage=sub.leverage, diff --git a/app/config.py b/app/config.py index 54eb205..c05a5cf 100644 --- a/app/config.py +++ b/app/config.py @@ -90,6 +90,15 @@ class Settings(BaseSettings): # Public site URL appended to the follow-up tweet. Falls back to frontend_url. x_link_url: str = "" + # ── X (Twitter) READ — KOL tweet ingestion (separate from the poster above) ── + # twitterapi.io managed-scraper key, sent as the `x-api-key` header. Powers + # kol_x.py: polls each tracked KOL's recent tweets → x_analysis.analyze_x_post + # → KolPost(source="twitter") → kol_divergence. This is READ-ONLY third-party + # data and is UNRELATED to the OAuth 1.0a poster creds above (those write our + # own tweets; this reads other people's). Empty = X ingestion disabled (no-op). + # Get a pay-as-you-go key at twitterapi.io (~$0.15 / 1k tweets). + twitterapi_io_key: str = "" + model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "ignore"} diff --git a/app/main.py b/app/main.py index 2b84baa..a4dd6db 100644 --- a/app/main.py +++ b/app/main.py @@ -226,6 +226,20 @@ async def lifespan(app: FastAPI): ) logger.info("KOL Substack poller scheduled daily at 01:15 UTC.") + # ── KOL X (Twitter) tweet ingestion (daily) ─────────────────────────── + # Polls X-native KOLs (andrewkang/@Rewkang, murad — no Substack feed) plus + # Hayes's real-time position statements. Writes KolPost(source="twitter") + # that the divergence scan (02:15) joins against on-chain data — this is the + # post-side feed that lights up the andrewkang/murad wallets. No-op if + # twitterapi_io_key is unset. Runs 01:30 — between substack (01:15) and + # on-chain (02:00) so fresh posts are present for the 02:15 divergence scan. + from app.services.kol_x import run_x_poll + _scheduler.add_job( + run_x_poll, "cron", hour=1, minute=30, + id="kol_x_poll", max_instances=1, coalesce=True, + ) + logger.info("KOL X (Twitter) poller scheduled daily at 01:30 UTC.") + # ── KOL A-tier: on-chain holdings snapshot (daily) ──────────────────── # Polls HL public API (free) for perp positions; Arkham (key optional) # for full portfolio. Diffs against yesterday's snapshot → writes diff --git a/app/models.py b/app/models.py index b6f3396..c1739c7 100644 --- a/app/models.py +++ b/app/models.py @@ -276,11 +276,17 @@ class KolPost(Base): raw_text: Mapped[str] = mapped_column(Text, nullable=False) content_hash: Mapped[str] = mapped_column(String(64), nullable=False) - summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True) - tickers_json: Mapped[Optional[str]] = mapped_column(Text, nullable=True) - analyzed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) - analysis_model: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) - analysis_version: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + tickers_json: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + analyzed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + analysis_model: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + analysis_version: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + + # x_analysis extended output (migration 027) + tier: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + post_type: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + talks_vs_trades_flag: Mapped[Optional[bool]] = mapped_column(Boolean, nullable=True, default=False) + sentiment: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=utcnow) diff --git a/app/schemas.py b/app/schemas.py index 8316ea5..ac91b9b 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -42,6 +42,36 @@ class TrumpPost(BaseModel): model_config = {"from_attributes": True} +class PostFilterCounts(BaseModel): + # Count of posts matching the current non-signal filters. When the caller + # sets ai_scored_only=true this already excludes off-topic/noise rows. + all: int + actionable: int + buy: int + short: int + # Off-topic rows hidden by the "Signals only" toggle. Computed from the + # same source/sentiment scope but ignoring ai_scored_only so the toggle + # can stay visible while active. + off_topic: int + + +class SourceCount(BaseModel): + source: str + count: int + latest: Optional[str] = None + + +class PostListResponse(BaseModel): + items: list[TrumpPost] + total: int + page: int + limit: int + counts: PostFilterCounts + source_counts: list[SourceCount] = [] + + + + class Candle(BaseModel): time: int open: float @@ -56,12 +86,18 @@ class BotTrade(BaseModel): asset: str side: str entry_price: float - exit_price: float - pnl_usd: float - hold_seconds: int - trigger_post_id: int + # Nullable fields: None = "not yet known / externally closed / not applicable". + # Do NOT coerce to 0 — the frontend uses null to distinguish genuine zeros + # (break-even trade) from "we don't have this data yet". + exit_price: Optional[float] = None # None while still open / extern-closed + pnl_usd: Optional[float] = None # None = unsettled + hold_seconds: Optional[int] = None # None = not yet computed + trigger_post_id: Optional[int] = None # None = adopted/manual (no trigger post) opened_at: str - closed_at: str + closed_at: Optional[str] = None # None for still-open positions returned from /user + # Optional short trigger snippet for table/list UIs. Comes from the joined + # trigger Post row when present; omitted for adopted / deleted-post trades. + trigger_post_text: Optional[str] = None # Source tag of the originating signal (e.g. 'truth', 'breakout', 'my_strategy'). # Joined from posts.source on read; not stored on BotTrade itself. # 'unknown' when the trigger post has been deleted or trigger_post_id is null. @@ -152,6 +188,7 @@ class UserResponse(BaseModel): hl_api_key_set: bool hl_api_key_masked: Optional[str] = None paper_mode: bool = False + auto_trade: bool = False # B50: was silently dropped, causing Settings to show ON as OFF trades: list[BotTrade] settings: UserSettings # Convex-strategy: ISO-UTC timestamp until which the bot is manually armed. diff --git a/app/services/adoption.py b/app/services/adoption.py index 449b77f..2ab831d 100644 --- a/app/services/adoption.py +++ b/app/services/adoption.py @@ -48,7 +48,7 @@ from app.services.signal_categories import ( get_exit_profile, get_stop_ladder, sys2_normalize_mode, sys2_protective_stop_pct, sys2_derisk_ladder, sys2_addon_ladder, sys2_peak_trail, - SYS2_MAX_CONCURRENT, SYS2_MODES, + SYS2_MAX_CONCURRENT, SYS2_MODES, SYS2_MAX_LEVERAGE, ) logger = logging.getLogger(__name__) diff --git a/app/services/binance.py b/app/services/binance.py index 1f42e3f..2dc2bf2 100644 --- a/app/services/binance.py +++ b/app/services/binance.py @@ -31,14 +31,40 @@ last_tick_at: Optional[datetime] = None # on Binance; those require a separate HL price feed (see BUG-08 note in CLAUDE.md). # Until that feed is added, HYPE trades fall back to max-hold only. ASSET_MAP: dict[str, str] = { - "btcusdt": "BTC", - "ethusdt": "ETH", - "solusdt": "SOL", + # Original 8 + "btcusdt": "BTC", + "ethusdt": "ETH", + "solusdt": "SOL", "trumpusdt": "TRUMP", - "bnbusdt": "BNB", - "dogeusdt": "DOGE", - "linkusdt": "LINK", - "aaveusdt": "AAVE", + "bnbusdt": "BNB", + "dogeusdt": "DOGE", + "linkusdt": "LINK", + "aaveusdt": "AAVE", + # Extended: all mainstream HL_PERPS that have Binance spot/perp pairs. + # Without these, any bot trade on these assets loses TP/SL/trailing + # protection — only max_hold remains as an exit. + "avaxusdt": "AVAX", + "arbusdt": "ARB", + "opusdt": "OP", + "suiusdt": "SUI", + "aptusdt": "APT", + "injusdt": "INJ", + "atomusdt": "ATOM", + "xrpusdt": "XRP", + "ltcusdt": "LTC", + "adausdt": "ADA", + "maticusdt": "MATIC", + "shibusdt": "SHIB", + "pepeusdt": "PEPE", + "wifusdt": "WIF", + "bonkusdt": "BONK", + "taousdt": "TAO", + "jupusdt": "JUP", + "renderusdt": "RENDER", + "fetusdt": "FET", + "tiausdt": "TIA", + "seiusdt": "SEI", + "pendleusdt": "PENDLE", } # Build the combined-stream WS URL from ASSET_MAP so the two are always in sync. diff --git a/app/services/bot_engine.py b/app/services/bot_engine.py index d68b9e1..b2dfca1 100644 --- a/app/services/bot_engine.py +++ b/app/services/bot_engine.py @@ -21,6 +21,16 @@ from app.services.price_store import price_store # noqa: F401 (used elsewhere) logger = logging.getLogger(__name__) + +async def _broadcast_trade_alert(wallet: str, event: str, **kwargs) -> None: + """Broadcast a trade lifecycle event over WebSocket. Fire-and-forget — never raises.""" + try: + from app.ws.manager import manager + await manager.broadcast({"type": "trade_alert", "wallet": wallet, "event": event, **kwargs}) + except Exception as exc: + logger.warning("trade_alert broadcast failed: %s", exc) + + # Platform-wide thresholds (per-user values in Subscription override where applicable) # No global confidence floor — users pick their own 0–100 threshold via settings. # Per-user max hold lives on Subscription.max_hold_hours (default 168h = 7 days @@ -230,6 +240,13 @@ async def process_post(post: Post, db: AsyncSession) -> None: sys2_leverage=s.sys2_leverage, sys2_mode=s.sys2_mode, auto_trade=bool(s.auto_trade), + # B38: module toggles — must be in snapshot so the execution gate + # can read them. trump_enabled gates System-1 (Trump); macro_enabled + # gates System-2 (Macro Vibes / bottom reversal). Both default to + # False (new subscribers start with bot idle) so a NULL in old rows + # continues the old conservative behaviour. + trump_enabled=bool(s.trump_enabled), + macro_enabled=bool(s.macro_enabled), ) for s in subscribers ] @@ -298,6 +315,63 @@ async def process_post(post: Post, db: AsyncSession) -> None: await asyncio.gather(*tasks, return_exceptions=True) +def _is_in_active_window(sub: dict) -> bool: + """Return True if the current UTC time falls inside the subscriber's + active window (schedule or manual override). + + Priority (highest wins): + 1. manual_window_until — operator-armed override; if set and in the + future, the bot is ALWAYS armed regardless of the schedule. + 2. active_from / active_until — user-configured recurring schedule. + Both must be set; if only one is set we treat the schedule as + unconfigured and return True (don't gate). + 3. If neither is set, always active. + + B30: was defined but never called — entire feature was dead. + """ + now = datetime.now(timezone.utc).replace(tzinfo=None) + + # 1. Manual override window (e.g. "arm for the next 4 hours") + mwu = sub.get("manual_window_until") + if mwu is not None: + mwu_dt = mwu if isinstance(mwu, datetime) else None + if mwu_dt is None: + try: + from datetime import datetime as _dt + mwu_dt = _dt.fromisoformat(str(mwu).replace("Z", "")) + except Exception: + mwu_dt = None + if mwu_dt is not None and mwu_dt > now: + return True # manual override wins + + # 2. Recurring schedule + af = sub.get("active_from") + au = sub.get("active_until") + if af is None or au is None: + return True # no schedule configured → always active + + # Convert to time-of-day for comparison (schedule repeats daily) + def _to_time(val): + if isinstance(val, datetime): + return val.time() + try: + from datetime import datetime as _dt + return _dt.fromisoformat(str(val).replace("Z", "")).time() + except Exception: + return None + + af_t = _to_time(af) + au_t = _to_time(au) + if af_t is None or au_t is None: + return True # can't parse → don't block + + now_t = now.time() + if af_t <= au_t: + return af_t <= now_t <= au_t # same-day window + else: + return now_t >= af_t or now_t <= au_t # crosses midnight + + async def _execute_for_subscriber( sub: dict, post_id: int, @@ -306,9 +380,27 @@ async def _execute_for_subscriber( side: str, ) -> None: wallet = sub["wallet"] - if not sub["hl_api_key"]: + + # B38: module on/off switches gate BEFORE any expensive work. + # trump_enabled → System-1 (source="truth"); macro_enabled → System-2. + # Defaults: False. A NULL column (pre-migration row) is treated as False, + # preserving the existing conservative behaviour. + is_sys2 = bool(sub.get("_is_system_2")) + if is_sys2: + if not sub.get("macro_enabled"): + logger.info("Sub %s: macro_enabled OFF — post %d not traded", wallet, post_id) + return + else: + if not sub.get("trump_enabled"): + logger.info("Sub %s: trump_enabled OFF — post %d not traded", wallet, post_id) + return + + # B29: paper users have no HL API key by design — skip the key check for + # them. The paper-mode branch later uses price_store instead of HL. + if not sub["hl_api_key"] and not sub.get("paper_mode"): logger.warning("Subscriber %s has no HL API key, skipping", wallet) return + # Required-setup guard. System 1 (Trump) needs take-profit and stop-loss. # System 2 (reversal) supplies its own stop + trailing from the category # profile, so only the two Trump exit fields are user-required. @@ -339,6 +431,15 @@ async def _execute_for_subscriber( wallet, post_id) return + # ── Schedule / manual-window gate (B30) ─────────────────────────────── + # active_from/active_until define a recurring daily window (e.g. 09:00-17:00 UTC). + # manual_window_until is an operator override that arms the bot for a + # fixed duration regardless of the schedule (e.g. "trade for the next 4h"). + # System-2 (manage-only / adopt model) is exempt — it never auto-opens. + if _should_apply_schedule(sub) and not _is_in_active_window(sub): + logger.info("Sub %s: outside active window — post %d not traded", wallet, post_id) + return + # ── Circuit breaker gate (P1.1) ──────────────────────────────────────── # Checked BEFORE key decryption (cheap fast-fail). If tripped, the wallet # has lost too much today or hit a losing streak — block until manual reset. @@ -410,9 +511,19 @@ async def _execute_for_subscriber( ) ) # Only count spend from THIS system against THIS system's slice. + # M1 fix: adopted trades have trigger_post_id=NULL so the + # outerjoin yields src=NULL → (src or "").lower()="" → not in + # SYSTEM_2_SOURCES → t_is_s2=False. That miscounts every + # adopted sys2 position against the sys1 budget, inflating + # "spent" and prematurely blocking new Trump scalp opens. + # Adopted trades are always sys2 — their hl_order_id starts + # with "adopted:" by construction (see adoption.py). spent = 0.0 for t, src in spent_result.all(): - t_is_s2 = (src or "").lower() in SYSTEM_2_SOURCES + if t.hl_order_id and str(t.hl_order_id).startswith("adopted:"): + t_is_s2 = True + else: + t_is_s2 = (src or "").lower() in SYSTEM_2_SOURCES if t_is_s2 != is_s2: continue spent += (t.size_usd if t.size_usd is not None else sub["position_size_usd"]) @@ -420,6 +531,11 @@ async def _execute_for_subscriber( logger.info("Sub %s [%s] daily budget reached: spent=%.2f + new=%.2f > cap=%.2f (%.0f%% of %.2f)", wallet, _system, spent, sized_position_usd, daily_cap, (sys2_pct if is_s2 else 100.0), total_cap) + asyncio.create_task(_broadcast_trade_alert( + wallet, "budget_reached", + asset=asset, size_usd=sized_position_usd, + spent_usd=round(spent, 2), cap_usd=round(daily_cap, 2), + )) return # ── System-2 correlation / concentration cap ─────────────────────── @@ -505,6 +621,24 @@ async def _execute_for_subscriber( logger.info("Subscriber %s already has open %s position, skipping", wallet, asset) return + balance = await trader.get_balance() + # HL isolated-margin requires notional / leverage as collateral, + # not the full notional value. Add a 10% buffer for fees + slippage. + leverage = sub.get("leverage") or 1 + required_margin = round((sized_position_usd / max(leverage, 1)) * 1.1, 2) + if balance < required_margin: + logger.warning( + "Sub %s: insufficient balance %.2f < required margin %.2f " + "(notional=%.2f lev=%dx) for %s — skipping", + wallet, balance, required_margin, sized_position_usd, leverage, asset, + ) + asyncio.create_task(_broadcast_trade_alert( + wallet, "insufficient_balance", + asset=asset, balance_usd=round(balance, 2), + required_usd=required_margin, + )) + return + result = await trader.open_position(asset, side, sized_position_usd) entry_price = result.get('fill_price', 0.0) @@ -529,7 +663,7 @@ async def _execute_for_subscriber( # for its stop math, but we still record the true value so # BotTrade.leverage reflects what HL actually applied. sub["leverage"] = effective_leverage - if sys2: + if sub.get("_is_system_2"): from app.services.signal_categories import ( sys2_protective_stop_pct as _sps, ) @@ -562,7 +696,7 @@ async def _execute_for_subscriber( # constructor before the later assignment used to throw # UnboundLocalError on every sys2 fire, leaving the HL # position open with NO DB record and NO watchdog. Critical. - _sys2_mode = sub.get("_sys2_mode", "standard") if sys2 else None + _sys2_mode = sub.get("_sys2_mode", "standard") if sub.get("_is_system_2") else None trade = BotTrade( asset=asset, @@ -601,7 +735,8 @@ async def _execute_for_subscriber( _derisk = None _addon = None _peak_trail = None - if sys2: + _is_sys2 = bool(sub.get("_is_system_2")) + if _is_sys2: _mode_for_ladders = _sys2_mode or "standard" from app.services.signal_categories import ( sys2_derisk_ladder, sys2_addon_ladder, sys2_peak_trail, @@ -624,7 +759,7 @@ async def _execute_for_subscriber( invalidation=eff["invalidation"], invalidation_price=eff.get("invalidation_price"), min_hold_until_ts=eff["min_hold_until_ts"], - stop_ladder=get_stop_ladder(post.category) if sys2 else None, + stop_ladder=get_stop_ladder(sub.get("_category", "")) if _is_sys2 else None, derisk_ladder=_derisk, derisk_done=0, addon_ladder=_addon, @@ -656,6 +791,10 @@ async def _execute_for_subscriber( except Exception as e: logger.error("Trade execution failed for %s: %s", wallet, e) + asyncio.create_task(_broadcast_trade_alert( + wallet, "execution_failed", + asset=asset, reason=str(e)[:200], + )) async def partial_derisk( @@ -885,6 +1024,16 @@ async def pyramid_add( fill = r.get("fill_price") filled_coins = float(r.get("size_coins") or 0.0) if not fill or filled_coins <= 0: + # Revert the pre-claim — HL didn't fill so the rung must + # remain retryable. Without this, addon_steps_done stays + # incremented and the rung is permanently burned (B48). + await db.execute( + update(BotTrade) + .where(BotTrade.id == trade_id) + .where(BotTrade.addon_steps_done == step_idx + 1) + .values(addon_steps_done=step_idx) + ) + await db.commit() return (False, None, None) # Use the ACTUAL filled notional, not the intended amount — # an IOC add can under-fill on thin/volatile books; assuming diff --git a/app/services/circuit_breaker.py b/app/services/circuit_breaker.py index 5c196f9..a694b9b 100644 --- a/app/services/circuit_breaker.py +++ b/app/services/circuit_breaker.py @@ -129,7 +129,7 @@ async def check_and_trip( today_trades = await _trades_for_system(db, wallet, start_of_day, system) today_pnl = sum(t.pnl_usd or 0 for t in today_trades) - dd_limit = -CB_DAILY_DD_USD_PER_BASE * sub.position_size_usd * 10 + dd_limit = -CB_DAILY_DD_USD_PER_BASE * sub.position_size_usd if today_pnl < dd_limit: reason = "daily_dd" logger.warning("CB[%s] trip [daily_dd] %s: pnl=%.2f < %.2f", diff --git a/app/services/funding_signal.py b/app/services/funding_signal.py index 76cc6f0..7f73f6e 100644 --- a/app/services/funding_signal.py +++ b/app/services/funding_signal.py @@ -11,7 +11,9 @@ Broadcasts alert via WebSocket. Gated by user on/off toggle. """ import collections +import json import logging +import os from datetime import datetime, timezone from typing import Deque, Optional @@ -40,12 +42,49 @@ _enabled: bool = False _recent_signals: Deque[dict] = collections.deque(maxlen=50) _last_fired: dict[str, Optional[datetime]] = {s: None for s in WATCH_SYMBOLS} +# B52: persist _enabled across process restarts without a DB migration. +# The file survives `systemctl restart` but is cleared by OS reboots (which is +# acceptable — an operator who reboots the server is expected to re-arm the +# monitor). Path is configurable via env so staging vs prod can differ. +_STATE_FILE = os.environ.get( + "BREAKOUT_MONITOR_STATE_FILE", + "/tmp/trumpsignal-breakout-state.json", +) + + +def _load_persisted_state() -> None: + """Read _enabled from disk on startup. Called once at module import time.""" + global _enabled + try: + with open(_STATE_FILE) as f: + data = json.load(f) + _enabled = bool(data.get("enabled", False)) + logger.info("Breakout monitor: loaded persisted state enabled=%s", _enabled) + except FileNotFoundError: + pass # first run — start disabled + except Exception as exc: + logger.warning("Breakout monitor: failed to load state file: %s", exc) + + +def _persist_state() -> None: + """Write _enabled to disk so it survives process restarts.""" + try: + with open(_STATE_FILE, "w") as f: + json.dump({"enabled": _enabled, "updated_at": datetime.now(timezone.utc).isoformat()}, f) + except Exception as exc: + logger.warning("Breakout monitor: failed to persist state: %s", exc) + + +# Load persisted state at import time (called when main.py registers the scheduler). +_load_persisted_state() + # ── Public API ──────────────────────────────────────────────────────────────── def set_enabled(value: bool) -> None: global _enabled _enabled = value + _persist_state() # B52: survive restarts logger.info("Funding signal monitor: %s", "ENABLED" if value else "DISABLED") diff --git a/app/services/hyperliquid.py b/app/services/hyperliquid.py index 8ea3a6d..53582f3 100644 --- a/app/services/hyperliquid.py +++ b/app/services/hyperliquid.py @@ -185,7 +185,12 @@ class HyperliquidTrader: ) logger.info("Set leverage %dx for %s (isolated)", effective, coin) except Exception as exc: - logger.warning("set_leverage error (non-fatal): %s", exc) + # DO NOT swallow — if leverage isn't set, the position would open + # at whatever HL had previously. Downstream stop math + # (sys2_protective_stop_pct) uses the return value; a wrong + # value puts the stop outside the real liquidation line. + logger.error("set_leverage FAILED for %s %dx: %s — aborting open", coin, effective, exc) + raise RuntimeError(f"set_leverage failed for {coin} at {effective}×: {exc}") from exc return effective async def open_position( diff --git a/app/services/kol_substack.py b/app/services/kol_substack.py index dd4177f..c80a4b9 100644 --- a/app/services/kol_substack.py +++ b/app/services/kol_substack.py @@ -43,6 +43,16 @@ logger = logging.getLogger(__name__) # of body per post or it just hallucinates a topic line. (Headlines-only # feeds like Vitalik's blog need a follow-up HTML fetch, deferred.) # 3. Add with a sensible handle + display_name. +# 4. ⚠️ The KOL feed COUNT is hardcoded in the FRONTEND for SEO/marketing +# (it can't read this list cross-repo). If len(KOL_FEEDS) changes, update +# every "N KOL feeds" mention in the frontend repo: +# app/layout.tsx (JSON-LD ×2), app/page.tsx (×3 incl. metric), +# app/[locale]/kol/page.tsx (×4), app/[locale]/kol/KolPageClient.tsx +# (subtitle "and N more" = count-3), app/[locale]/glossary/page.tsx, +# public/llms.txt + llms-full.txt, app/opengraph-image.tsx. +# Currently len(KOL_FEEDS) == 25. (X-only KOLs live in kol_x.X_KOLS.) +# (2026-06-09: dropped from 29 → 25; removed placeholder, dragonfly, +# niccarter, eugene as dead feeds. Frontend count mentions need updating.) KOL_FEEDS: list[dict] = [ # ── Substack essayists (long-form thesis pieces) ───────────────────── { @@ -50,38 +60,40 @@ KOL_FEEDS: list[dict] = [ "display_name": "Arthur Hayes", "feed_url": "https://cryptohayes.substack.com/feed", }, - # Placeholder VC (Joel Monegro / Chris Burniske). Token-focused VC, posts - # long-form thesis pieces every 1-3 months that map directly to their - # portfolio bets (Solana staking, L1 monetary premium, etc.). + # Raoul Pal — Real Vision / Global Macro Investor founder. "Short Excerpts + # Raoul Pal — his public Substack (raoulpal.substack.com/feed) went STALE + # (last post 2024-05). Replaced 2026-06-09 with the Real Vision official + # podcast feed (feeds.megaphone.fm/realvision) — daily, free, 2000+ eps, + # full episode descriptions with macro/crypto thesis. It's the whole Real + # Vision channel (not Raoul-only), but high signal + fresh. Same canonical + # handle as his X (@RaoulGMI in kol_x) so long-form + real-time aggregate. { - "handle": "placeholder", - "display_name": "Placeholder VC", - "feed_url": "https://www.placeholder.vc/blog?format=rss", + "handle": "raoulpal", + "display_name": "Raoul Pal (Real Vision)", + "feed_url": "https://feeds.megaphone.fm/realvision", + "source": "podcast", }, - # Dragonfly Capital research blog on Medium — free, active (10+ posts). - # dragonfly.xyz/blog/rss.xml returns 0 (paywall). medium.com/dragonfly-research - # is the team's public research arm: airdrops, DeFi, protocol deep-dives. - { - "handle": "dragonfly", - "display_name": "Dragonfly Capital", - "feed_url": "https://medium.com/feed/dragonfly-research", - }, - # Andy Constan's Substack is paywalled (RSS returns 0). Keeping for any - # occasional public teaser. Forward Guidance podcast (Blockworks) features - # him weekly but is macro/equities-focused — not crypto-coin-specific enough - # to extract ticker signals from episode descriptions. + # REMOVED 2026-06-09 — dead feeds, no active replacement exists: + # • placeholder (Placeholder VC) — last post 2025-09, blog is their only + # public source, no newsletter. ~266d stale. + # • dragonfly (Dragonfly Research, Medium) — last post 2025-03 (~454d). + # dragonfly.xyz has no working RSS (JS-rendered, all endpoints empty); + # Haseeb's medium/@hosseeb is even older (2024). + # If either resumes regular publishing with a real RSS, re-add here. + # Andy Constan — his old Substack (dampedspring.substack.com) returned 0 + # entries (paywalled). Replaced 2026-06-09 with his FREE long-form Substack + # "Damped Spring 101" (dampedspring101.substack.com), which he's publicly + # committed to keeping free — active, real macro essays (liquidity, rates, + # positioning). His deepest paid research stays gated, but this carries + # genuine thesis content suitable for ticker/direction extraction. { "handle": "dampedspring", "display_name": "Damped Spring / Andy Constan", - "feed_url": "https://dampedspring.substack.com/feed", - }, - # Nic Carter's Substack is paywalled (RSS returns 0). His Medium feed is - # FREE and active — different URL, same author, real content. - { - "handle": "niccarter", - "display_name": "Nic Carter (Castle Island)", - "feed_url": "https://medium.com/feed/@nic__carter", + "feed_url": "https://dampedspring101.substack.com/feed", }, + # REMOVED 2026-06-09 — niccarter (Nic Carter / Castle Island). Substack + # paywalled (0 entries); Medium feed went stale (last 2025-07, ~316d). + # No active free replacement. Re-add if he resumes a public RSS. # Delphi Digital podcast (Buzzsprout) — 478 episodes, active May 2025. # Public, free. Episode descriptions name specific protocols / tokens with # thesis framing — good extraction signal. delphidigital.io/feed returns 0. @@ -104,12 +116,8 @@ KOL_FEEDS: list[dict] = [ "display_name": "The DeFi Edge", "feed_url": "https://thedefiedge.com/feed/", }, - # Eugene Ng Ah Sio — trader/analyst, sporadic but specific. - { - "handle": "eugene", - "display_name": "Eugene Ng Ah Sio", - "feed_url": "https://eugene.substack.com/feed", - }, + # REMOVED 2026-06-09 — eugene (Eugene Ng Ah Sio). Substack effectively + # dormant: only 5 entries, last 2025-05 (~399d). No alternative source. # ── DeFi journalism (Substack-style RSS) ───────────────────────────── # The Defiant — Camila Russo's team. DeFi-focused news with frequent # protocol + token mentions. Free RSS, ~100 entries. @@ -149,12 +157,15 @@ KOL_FEEDS: list[dict] = [ "feed_url": "https://feeds.megaphone.fm/lightspeed", "source": "podcast", }, - # Unchained — Laura Shin. Long interview format with founders and - # traders. Show notes are 6K+ chars (near-transcript). + # Unchained — Laura Shin. Long interview format with founders and traders. + # The website RSS (unchainedcrypto.com/feed) is Cloudflare-blocked (403, + # even HTTP/2 + browser UA). Switched 2026-06-09 to the canonical Megaphone + # podcast feed (resolved via iTunes lookup id=1123922160) — 1100+ episodes, + # active daily, 1.5-2K char show notes naming protocols/tokens. { "handle": "unchained", "display_name": "Unchained (Laura Shin)", - "feed_url": "https://www.unchainedcrypto.com/feed/", + "feed_url": "https://feeds.megaphone.fm/LSHML4761942757", "source": "podcast", }, # Bankless podcast — Ryan Sean Adams + David Hoffman. ETH-focused but @@ -240,8 +251,12 @@ KOL_FEEDS: list[dict] = [ # but extremely high quality. Mostly BTC-only signals. { "handle": "lynalden", + # lynalden.com/feed is Cloudflare-blocked (202, no body). Switched + # 2026-06-09 to her FeedBurner mirror, which serves fine. Note: bodies + # are article EXCERPTS (~400 chars) not full text, but enough for the + # AI to extract the BTC/macro thesis + direction. "display_name": "Lyn Alden", - "feed_url": "https://www.lynalden.com/feed/", + "feed_url": "https://feeds.feedburner.com/lynalden", "source": "blog", }, # Bitcoin Magazine — high-frequency news + institutional adoption analysis. @@ -336,8 +351,19 @@ async def _fetch_feed(feed_url: str) -> list: """feedparser is sync; do the HTTP fetch through httpx for timeout control + uniformity with the rest of the codebase, then hand bytes to feedparser.""" - async with httpx.AsyncClient(timeout=20.0, follow_redirects=True) as client: - r = await client.get(feed_url, headers={"User-Agent": "TrumpSignal/1.0 KOL-tracker"}) + # http2=True matters: some CDN-fronted feeds (e.g. Glassnode) return 403 + # to HTTP/1.1 requests but serve fine over HTTP/2 (curl defaults to h2, + # which is why they worked manually but not here). Requires the `h2` pkg. + async with httpx.AsyncClient(timeout=20.0, follow_redirects=True, http2=True) as client: + # Use a real browser UA. Several feeds (Glassnode, others behind a CDN) + # return 403/202 to non-browser agents. A plain bot UA was silently + # losing those feeds. This recovers them without per-feed special-casing. + r = await client.get(feed_url, headers={ + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/124.0 Safari/537.36", + "Accept": "application/rss+xml, application/atom+xml, application/xml, text/xml, */*", + }) r.raise_for_status() parsed = feedparser.parse(r.content) return list(parsed.entries or []) @@ -439,6 +465,9 @@ async def _ingest_kol( row.analyzed_at = utcnow() row.analysis_model = result.get("model") row.analysis_version = result.get("version") + # Extended analysis fields (migration 027) + row.post_type = result.get("post_type") + row.talks_vs_trades_flag = bool(result.get("talks_vs_trades_flag", False)) stats["analyzed"] += 1 except Exception as e: logger.warning("[kol_substack] analysis failed for %s post %s: %s", @@ -452,11 +481,15 @@ async def _ingest_kol( async def run_substack_poll(*, analyze: bool = True) -> list[dict]: """Poll every configured KOL feed once. Despite the legacy name this now covers Substack essays, Medium blogs, and major crypto podcasts via RSS. - Returns per-KOL stats.""" + Returns per-KOL stats. + + Each KOL gets its own session so a commit failure for one does not leave + a dirty session that breaks subsequent KOLs in the same run. + """ results = [] - async with AsyncSessionLocal() as session: - for kol in KOL_FEEDS: + for kol in KOL_FEEDS: + async with AsyncSessionLocal() as session: stats = await _ingest_kol(session, kol, analyze=analyze) - results.append(stats) + results.append(stats) logger.info("[kol_substack] poll done: %s", results) return results diff --git a/app/services/kol_x.py b/app/services/kol_x.py new file mode 100644 index 0000000..1803c19 --- /dev/null +++ b/app/services/kol_x.py @@ -0,0 +1,284 @@ +"""KOL X (Twitter) ingester. + +Polls each tracked KOL's recent tweets via twitterapi.io, dedupes by tweet id, +runs `x_analysis.analyze_x_post`, and writes a `KolPost(source="twitter")`. + +The `tickers_json` it stores uses the exact `{ticker, action, conviction}` shape +that `kol_divergence` already consumes (it reads `t["ticker"]` / `t["action"]` / +`t["conviction"]`) — so no mapping layer is needed. x_analysis was designed +against that contract; this module just connects the pipe. + +WHY THIS EXISTS + `andrewkang` (@Rewkang) and `murad` publish ONLY on X — no Substack feed. + Their on-chain wallets are already seeded (seed_kol_wallets.py), but with no + post-side data the divergence scanner detects nothing for them — the wallets + sit dark. This ingester is the missing post-side feed that lights them up. + +DESIGN (mirrors kol_substack.py) + - Disabled (full no-op) when `settings.twitterapi_io_key` is empty. + - `KolPost.kol_handle` is the CANONICAL handle (e.g. "cryptohayes"), NOT the + X screen name — it must match `KolWallet.handle` so divergence can join + post-side ↔ on-chain for the same person. + - Dedup by (source="twitter", external_id=). + - Bare retweets ("RT @…") are skipped before the AI call to save spend. + - Only the first page (~20 newest tweets) is fetched per run. Daily volume + for these KOLs is < 20/day, and dedup makes re-runs cheap. Bump + `max_pages` if you add a very high-frequency account. + +COST + ~$0.15 / 1k tweets (twitterapi.io). 4 KOLs × ~20 tweets daily ≈ $0.015/mo. + +CADENCE + Daily 01:30 UTC — after substack (01:15), before on-chain (02:00) and + divergence (02:15), so fresh X posts are present when divergence runs. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime +from typing import AsyncGenerator, Optional + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.database import AsyncSessionLocal +from app.models import KolPost, utcnow +from app.services import x_analysis + +logger = logging.getLogger(__name__) + +X_API = "https://api.twitterapi.io/twitter/user/last_tweets" + +# Tracked X-native KOLs. `handle` MUST match the canonical handle used in +# KolWallet / KOL_FEEDS so divergence can join against on-chain data. +# `x_username` is the actual X screen name (no @). +# +# cryptohayes is also on Substack (long-form) — X adds his real-time position +# statements ("just dumped my $HYPE"), which Substack essays don't carry. +# andrewkang + murad are X-ONLY and have seeded wallets waiting for this feed. +X_KOLS: list[dict] = [ + {"handle": "cryptohayes", "x_username": "CryptoHayes", "display_name": "Arthur Hayes"}, + {"handle": "andrewkang", "x_username": "Rewkang", "display_name": "Andrew Kang"}, + {"handle": "murad", "x_username": "MustStopMurad", "display_name": "Murad Mahmudov"}, + # Raoul Pal — Real Vision / GMI founder. Macro/liquidity/cycle thinker + # (mostly DIRECTIONAL, rarely position TRADE_SIGNAL). Same canonical handle + # as his Substack feed below so his X takes + long-form theses aggregate. + {"handle": "raoulpal", "x_username": "RaoulGMI", "display_name": "Raoul Pal"}, +] + + +def _parse_created_at(raw: Optional[str]) -> datetime: + """X `createdAt` looks like 'Thu Jun 04 12:29:13 +0000 2026'. Normalise to + naive UTC (same convention as kol_substack._parse_pub — divergence window + matching and digest 'since' filters all assume naive UTC).""" + if not raw: + return utcnow() + try: + dt = parsedate_to_datetime(raw) + if dt.tzinfo: + dt = dt.astimezone(timezone.utc).replace(tzinfo=None) + return dt + except Exception: + return utcnow() + + +async def _fetch_tweet_page( + client: httpx.AsyncClient, + x_username: str, + cursor: Optional[str], + key: str, +) -> tuple[list[dict], Optional[str]]: + """Fetch one page of tweets. Returns (tweets, next_cursor). + next_cursor is None when there are no more pages or on any error.""" + params: dict = {"userName": x_username, "includeReplies": "false"} + if cursor: + params["cursor"] = cursor + try: + r = await client.get(X_API, params=params, headers={"x-api-key": key}) + if r.status_code != 200: + logger.warning("[kol_x] fetch %s HTTP %d: %s", + x_username, r.status_code, r.text[:200]) + return [], None + data = r.json() + # Shape: {status, data: {tweets: [...], next_cursor?: str}, ...} + if data.get("status") != "success": + logger.warning("[kol_x] fetch %s non-success: %s", + x_username, str(data.get("msg") or data)[:200]) + return [], None + inner = data.get("data") or {} + tweets = inner.get("tweets") or [] + next_cursor = inner.get("next_cursor") or None + return tweets, next_cursor + except Exception as e: + logger.warning("[kol_x] fetch %s exception: %s", x_username, e) + return [], None + + +async def _iter_tweet_pages( + x_username: str, + max_pages: int = 3, +) -> AsyncGenerator[list[dict], None]: + """Async generator: yields one page of tweets at a time. + + Callers can break out of the loop early (page-level early-stop) without + fetching unnecessary pages. max_pages=3 is the gap-fill ceiling — normal + daily runs stop after page 1 because _ingest_kol_x breaks on all-dedup. + + Never raises — a network error yields nothing and ends the iteration. + """ + key = settings.twitterapi_io_key + if not key: + return + cursor: Optional[str] = None + async with httpx.AsyncClient(timeout=20.0) as client: + for _ in range(max_pages): + tweets, cursor = await _fetch_tweet_page(client, x_username, cursor, key) + if tweets: + yield tweets + if not cursor or not tweets: + break + + +async def _ingest_kol_x( + session: AsyncSession, + kol: dict, + *, + analyze: bool = True, + max_new: int = 20, + max_pages: int = 3, +) -> dict: + """Ingest one KOL's recent tweets. Mirrors kol_substack._ingest_kol. + + Pages are fetched one at a time. If an entire page consists only of + already-stored tweets (dedup hits), fetching stops — subsequent pages + would be even older and equally known, so the API call is skipped. + This keeps the normal daily run to a single page fetch. + """ + handle = kol["handle"] + x_username = kol["x_username"] + stats = {"handle": handle, "x_username": x_username, + "new": 0, "skipped": 0, "analyzed": 0, "errors": 0} + + async for page_tweets in _iter_tweet_pages(x_username, max_pages=max_pages): + if stats["new"] >= max_new: + break + + page_new = 0 # new tweets stored on this page + page_deduped = 0 # already-stored tweets hit on this page (dedup) + + for tw in page_tweets: + if stats["new"] >= max_new: + break + + tweet_id = str(tw.get("id") or "").strip() + text = (tw.get("text") or "").strip() + if not tweet_id or not text: + continue + + # Bare retweet → noise. Skip before the AI call to save spend. + if text.startswith("RT @"): + stats["skipped"] += 1 + continue + + # Dedup by (source, external_id). + existing = await session.execute( + select(KolPost).where( + KolPost.source == "twitter", + KolPost.external_id == tweet_id, + ) + ) + if existing.scalar_one_or_none() is not None: + stats["skipped"] += 1 + page_deduped += 1 + continue + + author = tw.get("author") or {} + follower_count = author.get("followers") + url = tw.get("url") or f"https://x.com/{x_username}/status/{tweet_id}" + pub = _parse_created_at(tw.get("createdAt")) + body_hash = hashlib.sha256(text.encode("utf-8")).hexdigest() + + row = KolPost( + kol_handle=handle, + source="twitter", + external_id=tweet_id, + url=url, + title=None, + published_at=pub, + raw_text=text, + content_hash=body_hash, + ) + session.add(row) + await session.flush() + stats["new"] += 1 + page_new += 1 + logger.info("[kol_x] new tweet %s id=%s tweet_id=%s", handle, row.id, tweet_id) + + if analyze: + try: + result = await x_analysis.analyze_x_post( + handle=handle, + text=text, + posted_at=pub.isoformat(), + follower_count=follower_count, + ) + if result.get("error"): + stats["errors"] += 1 + else: + row.summary = result.get("summary") + row.tickers_json = json.dumps(result.get("tickers") or [], + ensure_ascii=False) + row.analyzed_at = utcnow() + row.analysis_model = result.get("model") + row.analysis_version = result.get("version") + # Extended x_analysis fields (migration 027) + row.tier = result.get("tier") + row.post_type = result.get("post_type") + row.talks_vs_trades_flag = bool(result.get("talks_vs_trades_flag", False)) + row.sentiment = result.get("sentiment") + stats["analyzed"] += 1 + except Exception as e: + logger.warning("[kol_x] analysis failed %s tweet %s: %s", + handle, row.id, e) + stats["errors"] += 1 + + # Page-level early stop: this page yielded NO new tweets AND hit at + # least one dedup → we've caught up to already-stored data, so older + # pages are known too. A page that is ALL bare-RTs (page_deduped == 0) + # must NOT early-stop — the next page may still hold original posts. + if page_new == 0 and page_deduped > 0: + logger.debug("[kol_x] %s page fully deduped — stopping early", handle) + break + + await session.commit() + return stats + + +async def run_x_poll(*, analyze: bool = True) -> list[dict]: + """Poll every tracked X KOL once. Full no-op (returns []) when the + twitterapi.io key is unset. Returns per-KOL stats. + + Each KOL gets its own session so a commit failure for one does not leave + a dirty session that breaks subsequent KOLs in the same run. + """ + if not settings.twitterapi_io_key: + logger.info("[kol_x] twitterapi_io_key not set — X ingestion disabled.") + return [] + results = [] + for kol in X_KOLS: + try: + async with AsyncSessionLocal() as session: + stats = await _ingest_kol_x(session, kol, analyze=analyze) + results.append(stats) + except Exception as e: + # One KOL's DB/commit failure must not abort the rest of the run. + logger.error("[kol_x] ingest failed for %s: %s", kol.get("handle"), e) + results.append({"handle": kol.get("handle"), "error": str(e)}) + logger.info("[kol_x] poll done: %s", results) + return results diff --git a/app/services/macro/fetchers.py b/app/services/macro/fetchers.py index 35add80..f508396 100644 --- a/app/services/macro/fetchers.py +++ b/app/services/macro/fetchers.py @@ -161,63 +161,54 @@ async def fetch_ahr999() -> dict: } -# ── 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. +# ── 2. Altcoin Season Index (blockchaincenter.net — official source) ───────── +# Scrape the value directly from blockchaincenter.net, which is the canonical +# publisher of this index (90-day window: how many of the top 50 alts beat BTC +# over 90 days). 75+ = altseason, <25 = bitcoin season. +# +# Previous implementation computed the index from CoinGecko /coins/markets +# using a 30d window (CoinGecko doesn't return 90d per-coin data on that +# endpoint). The 30d vs 90d discrepancy caused readings up to ~30 points +# higher than the official index during BTC-dominated markets. Scraping the +# actual page is more reliable than re-implementing the formula. +# +# Fallback: if the page scrape fails, return None (the @_none_on_fail +# decorator handles that gracefully). -_STABLE_OR_WRAPPED = { - "USDT", "USDC", "DAI", "BUSD", "TUSD", "USDD", "FDUSD", "PYUSD", "USDE", - "WBTC", "WETH", "STETH", "WSTETH", "WEETH", "RETH", -} +_BCC_URL = "https://www.blockchaincenter.net/altcoin-season-index/" +# Regex for the server-rendered value in the Next.js HTML: +# "Season41" or "Season (41" inside the page markup. +_BCC_RE = re.compile(r"Season[^<]{0,20}\s*(\d{1,3})") @_none_on_fail("altcoin_season_index") async def fetch_altcoin_season_index() -> dict: - """Compute the Altcoin Season Index from CoinGecko /coins/markets. + """Fetch the Altcoin Season Index from blockchaincenter.net. - 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. + The site is Next.js SSR — the value is embedded in the initial HTML as a + server-rendered text node. We parse it with a tight regex and fall back to + None on any parse failure so the rest of the snapshot is unaffected. """ - 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"}, - ) + async with httpx.AsyncClient( + timeout=DEFAULT_TIMEOUT, + headers={**UA, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"}, + follow_redirects=True, + ) as c: + r = await c.get(_BCC_URL) r.raise_for_status() - rows = r.json() + html = r.text - # 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)}} + m = _BCC_RE.search(html) + if not m: + return {"value": None, "raw": {"error": "regex did not match", "url": _BCC_URL}} - 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"}} + value = int(m.group(1)) + if not 0 <= value <= 100: + return {"value": None, "raw": {"error": f"parsed value out of range: {value}"}} - 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"}, + "value": float(value), + "raw": {"source": "blockchaincenter.net", "window": "90d", "parsed": value}, } diff --git a/app/services/signed_request.py b/app/services/signed_request.py index 353629b..0d4c62b 100644 --- a/app/services/signed_request.py +++ b/app/services/signed_request.py @@ -94,7 +94,7 @@ def verify_signed_request( # 1. Freshness now_ms = int(time.time() * 1000) - if abs(now_ms - timestamp_ms) > MAX_SKEW_SECONDS * 1000: + if timestamp_ms > now_ms + 30_000 or now_ms - timestamp_ms > MAX_SKEW_SECONDS * 1000: # allow 30s future drift for clock skew raise HTTPException(401, "Signed request expired or clock skew too large") # 2. Recover signer diff --git a/app/services/telegram.py b/app/services/telegram.py index 5d5a6ca..a7d7351 100644 --- a/app/services/telegram.py +++ b/app/services/telegram.py @@ -22,6 +22,7 @@ Source → user-toggle mapping: from __future__ import annotations import asyncio +import html import logging from datetime import datetime, timezone from typing import Optional @@ -100,13 +101,21 @@ def format_post(post: Post) -> str: asset = post.target_asset or "?" conf = post.ai_confidence or 0 - # Heading: emoji + asset + direction + confidence - head = f"{emoji} {asset} · {sig} · conf {conf}" + # Heading: emoji + asset + direction + confidence. + # asset comes from post.target_asset (scanner-written) — escape defensively; + # sig/conf/src are controlled enums/ints/labels. + head = f"{emoji} {html.escape(asset)} · {sig} · conf {conf}" sub = f"{src}" + # Truncate the RAW text first, THEN escape — escaping first could let a + # 5-char entity (&) get sliced mid-sequence by the length cap. body = (post.text or "").strip() if len(body) > 600: body = body[:600].rstrip() + "…" + # parse_mode=HTML: Trump posts routinely contain & < > ("Law & Order"). + # Without escaping, Telegram rejects the whole message (400 can't parse + # entities) and every subscriber silently misses the alert. + body = html.escape(body) # Move size hint if present (BTC bottom & funding emit expected_move_pct) extra = "" @@ -157,6 +166,7 @@ def format_trump_mention(post: Post) -> str: body = (post.text or "").strip() if len(body) > 300: body = body[:300].rstrip() + "…" + body = html.escape(body) # see format_post — escape user text for HTML mode fe = (settings.frontend_url or "").rstrip("/") link = f'\n\n→ view on TrumpAlpha' if fe else "" @@ -197,7 +207,7 @@ def format_public_post(post: Post) -> str: else: conf_label = "—" - head = f"{emoji} {asset} · {sig} · conf {conf_label}" + head = f"{emoji} {html.escape(asset)} · {sig} · conf {conf_label}" sub = f"{src}" body = (post.text or "").strip() @@ -212,6 +222,10 @@ def format_public_post(post: Post) -> str: reason = reason[:200].rstrip() + "…" body = reason + # Escape AFTER choosing text-vs-reason — both are user/AI content going + # into an HTML-parse-mode message. (See format_post.) + body = html.escape(body) + fe = (settings.frontend_url or "").rstrip("/") link = "" if fe: @@ -447,6 +461,13 @@ async def _dispatch(post_id: int) -> None: post_id, channel_id) +# Strong references to in-flight dispatch tasks. asyncio.create_task() only +# keeps a WEAK reference, so a fire-and-forget task can be garbage-collected +# mid-await — "Task was destroyed but it is pending!" — silently dropping a +# subscriber fan-out. We hold the task until it completes, then auto-discard. +_dispatch_tasks: set[asyncio.Task] = set() + + def notify_signal(post: Post) -> None: """Fire-and-forget. Schedules `_dispatch(post.id)` on the running loop and returns immediately. Safe to call from any async context — falls @@ -459,7 +480,9 @@ def notify_signal(post: Post) -> None: if not post or not post.id: return try: - asyncio.create_task(_dispatch(post.id)) + task = asyncio.create_task(_dispatch(post.id)) + _dispatch_tasks.add(task) + task.add_done_callback(_dispatch_tasks.discard) except RuntimeError: # No running loop — extremely unusual in our FastAPI context. logger.warning("notify_signal: no running event loop, skipping post=%s", post.id) diff --git a/app/services/telegram_bot.py b/app/services/telegram_bot.py index 964e773..c68303c 100644 --- a/app/services/telegram_bot.py +++ b/app/services/telegram_bot.py @@ -998,6 +998,31 @@ async def run_bot_loop() -> None: return logger.info("Telegram bot loop starting (long-poll mode).") + + # ── Startup drain ──────────────────────────────────────────────────── + # On restart, Telegram may replay up to 24h of unacknowledged updates. + # The most dangerous replays are stale /adopt callback_query items that + # would re-fire an adoption the user already dealt with. Fix: pull all + # pending updates with timeout=0 (non-blocking), advance offset past + # them without processing, then start the real loop fresh. + try: + drain_url = TG_API.format(token=token, method="getUpdates") + async with httpx.AsyncClient(timeout=10) as client: + r = await client.get(drain_url, params={"timeout": 0, "limit": 100}) + if r.status_code == 200: + pending = r.json().get("result", []) + if pending: + drain_offset = pending[-1]["update_id"] + 1 + # ACK by sending offset back — Telegram won't re-deliver these. + async with httpx.AsyncClient(timeout=10) as client: + await client.get(drain_url, params={"timeout": 0, "offset": drain_offset}) + logger.info( + "Startup drain: skipped %d stale update(s), offset now %d", + len(pending), drain_offset, + ) + except Exception as exc: + logger.warning("Startup drain failed (non-fatal): %s", exc) + offset: Optional[int] = None backoff = 1.0 diff --git a/app/services/telegram_digest.py b/app/services/telegram_digest.py index cd624f1..7ea496e 100644 --- a/app/services/telegram_digest.py +++ b/app/services/telegram_digest.py @@ -29,6 +29,7 @@ so a coalesced cron or worker restart can't double-send within a day. from __future__ import annotations +import html import logging from dataclasses import dataclass from datetime import datetime, timedelta, timezone @@ -138,26 +139,43 @@ async def build_global_digest(now: Optional[datetime] = None) -> GlobalDigest: ) # ── KOL: last 24h posts + divergences ──────────────────────────── + # Exclude twitter noise (gm / RT / jokes) from the count. Substack + # posts have tier=NULL (kept); twitter signal posts have tier!='noise' + # (kept); twitter noise is dropped. Without this filter a KOL's gm/RT + # spam inflates the daily "N KOL posts" stat. Mirrors the /kol/posts + # signals_only filter and the frontend. kol_posts = (await db.execute( - select(KolPost).where(KolPost.published_at >= cutoff_24h) + select(KolPost).where( + KolPost.published_at >= cutoff_24h, + (KolPost.tier.is_(None)) | (KolPost.tier != "noise"), + ) )).scalars().all() - # action lives inside tickers_json; easier: classify on summary text. - # The structured action is on KolDivergence rows; for the bare count - # we just split posts by whether divergence rows reference them. + # Count bullish/bearish/neutral at POST level, not divergence-pair level. + # A single post can match multiple on-chain events (one per ticker/wallet), + # producing several KolDivergence rows. Counting rows would inflate the + # numbers — a post mentioning 3 tickers was previously counted 3× (bug). bullish = bearish = neutral = 0 divergence_rows = (await db.execute( select(KolDivergence) .where(KolDivergence.post_at >= cutoff_24h) .order_by(KolDivergence.created_at.desc()) )).scalars().all() + # Deduplicate by post_id; for posts with mixed directions take "long" + # first (a bullish call that also hedges short is net bullish). + post_direction: dict[int, str] = {} for d in divergence_rows: - if d.direction == "long": + if d.post_id not in post_direction: + post_direction[d.post_id] = d.direction + elif d.direction == "long": + post_direction[d.post_id] = "long" # long overrides neutral + for direction in post_direction.values(): + if direction == "long": bullish += 1 - elif d.direction == "short": + elif direction == "short": bearish += 1 else: neutral += 1 - # Posts not yet classified into divergences just count as neutral. + # Posts not yet cross-referenced with on-chain data count as neutral. neutral += max(0, len(kol_posts) - (bullish + bearish + neutral)) divergences_24h = sum( @@ -168,7 +186,9 @@ async def build_global_digest(now: Optional[datetime] = None) -> GlobalDigest: if d.signal_type != "divergence": continue # "Hayes publicly bullish BTC, on-chain selling" - sample_divergence = ( + # Escape — ticker is AI-extracted (kol_analysis) so not guaranteed + # HTML-safe; the digest is sent with parse_mode=HTML. + sample_divergence = html.escape( f"{d.handle} publicly {d.post_action} {d.ticker}, " f"on-chain {d.onchain_action}" ) diff --git a/app/services/tp_sl_monitor.py b/app/services/tp_sl_monitor.py index 413f871..a988e2a 100644 --- a/app/services/tp_sl_monitor.py +++ b/app/services/tp_sl_monitor.py @@ -158,6 +158,19 @@ def register_trade( and not derisk_ladder): return + # Warn when the price feed doesn't cover this asset — TP/SL/trailing + # stops will NEVER fire; only max_hold will exit the position. + from app.services.binance import ASSET_MAP + from app.services.hl_price_feed import HL_PRICE_ASSETS + covered = set(ASSET_MAP.values()) | HL_PRICE_ASSETS + if asset not in covered: + logger.error( + "PRICE FEED GAP: trade %d asset=%s has no price coverage. " + "TP/SL/trailing-stop will NOT trigger — only max_hold protects " + "this position. Add %susdt to ASSET_MAP in binance.py.", + trade_id, asset, asset.lower(), + ) + _watched[trade_id] = WatchedTrade( trade_id=trade_id, wallet=wallet, api_key=api_key, leverage=leverage, asset=asset, side=side, entry_price=entry_price, diff --git a/app/services/x_analysis.py b/app/services/x_analysis.py index 134db03..626ac97 100644 --- a/app/services/x_analysis.py +++ b/app/services/x_analysis.py @@ -35,7 +35,7 @@ from app.config import settings logger = logging.getLogger(__name__) -ANALYSIS_VERSION = "x-v1" +ANALYSIS_VERSION = "x-v2" # v2: tone-is-not-content rule + meme calibration example ANTHROPIC_MODEL = "claude-haiku-4-5-20251001" _anthropic_client = None @@ -89,6 +89,13 @@ ALL of the above → tier: "noise", tickers: [] Only extract signals when the post contains CLEAR, EXPLICIT information about what the KOL is doing or believes RIGHT NOW. Ambiguity → noise. +TONE IS NOT CONTENT. A joke, meme, celebratory, or emoji-spam tone +("Arise Chikun!", "Yachtzee", "Meow", "😘😘😘", "WAGMI lfg") does NOT make a +post noise when it still carries an EXPLICIT ticker + direction. Score the +claim, ignore the wrapper. "$WLD initiate bull market 😘😘😘" is a directional +bullish call on WLD — NOT noise. Only drop to noise when the ticker or the +direction is genuinely absent/vague, never merely because the wording is silly. + ═══════════════════════════════════════════════════════════════════════ THREE SIGNAL TIERS ═══════════════════════════════════════════════════════════════════════ @@ -194,6 +201,11 @@ POST: "ETH dominance will crush alts this cycle. The flippening is real." → tier: "directional", action: "bullish", asset: ETH, conviction: 0.52, timeframe: "months" +POST: "Arise Chikun! $WLD initiate bull market. Yachtzee 😘😘😘😘" +→ tier: "directional", action: "bullish", asset: WLD, conviction: 0.7, + timeframe: "immediate" (meme/celebration tone does NOT override the + explicit "$WLD initiate bull market" call — score the claim, not the vibe) + [NOISE — output noise, empty tickers] POST: "gm everyone 🌅" diff --git a/requirements.txt b/requirements.txt index 37e260a..1d48e18 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,7 @@ alembic==1.16.5 pydantic==2.13.2 pydantic-settings==2.11.0 httpx==0.28.1 +h2==4.3.0 # enables httpx http2=True in kol_substack (Glassnode etc. 403 on HTTP/1.1) feedparser==6.0.12 websockets==15.0.1 websocket-client==1.9.0 diff --git a/tests/test_kol_x.py b/tests/test_kol_x.py new file mode 100644 index 0000000..8ceb1f5 --- /dev/null +++ b/tests/test_kol_x.py @@ -0,0 +1,272 @@ +"""Tests for the X (Twitter) KOL ingester (kol_x). + +The twitterapi.io fetch and the AI scorer are mocked so we exercise the +storage / dedup / mapping logic deterministically — no network, no AI spend. + +The contract these lock down: + - tweets land as KolPost(source="twitter") + - kol_handle is the CANONICAL handle (joins to KolWallet), not the X username + - bare retweets are skipped before scoring + - tickers_json keeps the {ticker, action, conviction} shape kol_divergence reads + - tier / post_type / talks_vs_trades_flag / sentiment are persisted (migration 027) + - re-running is idempotent (dedup by tweet id) + - page-level early-stop fires when a full page is all-dedup + - no twitterapi key → full no-op +""" + +from __future__ import annotations + +import json +from typing import AsyncGenerator + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import ( + AsyncSession, async_sessionmaker, create_async_engine, +) + +from app.config import settings +from app.models import Base, KolPost +from app.services import kol_x + + +_FAKE_TWEETS = [ + { # actionable position statement → should be stored + scored + "id": "111", + "text": "I just dumped my entire $HYPE position", + "url": "https://x.com/CryptoHayes/status/111", + "createdAt": "Thu Jun 04 05:49:13 +0000 2026", + "author": {"followers": 797330}, + }, + { # bare retweet → skipped before the AI call + "id": "222", + "text": "RT @someone: not my words", + "createdAt": "Thu Jun 04 06:00:00 +0000 2026", + "author": {"followers": 797330}, + }, + { # low-signal but original → stored (AI decides noise/not) + "id": "333", + "text": "gm crypto fam", + "createdAt": "Thu Jun 04 07:00:00 +0000 2026", + "author": {"followers": 797330}, + }, +] + +_FAKE_SCORE = { + "post_type": "original", + "tier": "trade_signal", + "summary": "Dumped HYPE", + "tickers": [{"ticker": "HYPE", "action": "sell", "conviction": 0.95}], + "talks_vs_trades_flag": True, + "sentiment": "bearish", + "model": "test-model", + "version": "x-test", + "error": None, +} + +_KOL = {"handle": "cryptohayes", "x_username": "CryptoHayes", "display_name": "Hayes"} + + +async def _fresh_session_factory(): + """Each test gets its own isolated in-memory DB.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + +def _make_page_iter(pages: list[list[dict]]): + """Return an async generator that yields one page at a time from `pages`.""" + async def fake_iter(_username, max_pages=3) -> AsyncGenerator[list[dict], None]: + for page in pages: + yield page + return fake_iter + + +def _patch_io(monkeypatch, pages: list[list[dict]] | None = None): + if pages is None: + pages = [list(_FAKE_TWEETS)] # one page containing all fake tweets + + async def fake_score(**_kw): + return dict(_FAKE_SCORE) + + monkeypatch.setattr(settings, "twitterapi_io_key", "test-key") + monkeypatch.setattr(kol_x, "_iter_tweet_pages", _make_page_iter(pages)) + monkeypatch.setattr(kol_x.x_analysis, "analyze_x_post", fake_score) + + +# ── Core contract ───────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_run_x_poll_noop_without_key(monkeypatch): + monkeypatch.setattr(settings, "twitterapi_io_key", "") + assert await kol_x.run_x_poll() == [] + + +@pytest.mark.asyncio +async def test_ingest_writes_divergence_ready_kolposts(monkeypatch): + _patch_io(monkeypatch) + sf = await _fresh_session_factory() + + async with sf() as s: + stats = await kol_x._ingest_kol_x(s, _KOL) + + # 111 + 333 stored; 222 (bare RT) skipped before scoring + assert stats["new"] == 2 + assert stats["skipped"] == 1 + assert stats["analyzed"] == 2 + assert stats["errors"] == 0 + + rows = (await s.execute( + select(KolPost).where(KolPost.source == "twitter") + )).scalars().all() + assert {r.external_id for r in rows} == {"111", "333"} + # canonical handle (joins to KolWallet), NOT the X screen name + assert all(r.kol_handle == "cryptohayes" for r in rows) + + signal = next(r for r in rows if r.external_id == "111") + tk = json.loads(signal.tickers_json) + assert tk[0]["ticker"] == "HYPE" + assert tk[0]["action"] == "sell" # in kol_divergence._POST_SHORT + assert tk[0]["conviction"] == 0.95 + assert signal.analysis_model == "test-model" + assert signal.analysis_version == "x-test" + # published_at parsed from the X date string into a naive datetime + assert signal.published_at.year == 2026 and signal.published_at.month == 6 + + +@pytest.mark.asyncio +async def test_ingest_stores_extended_analysis_fields(monkeypatch): + """migration 027 fields — tier / post_type / talks_vs_trades_flag / sentiment.""" + _patch_io(monkeypatch) + sf = await _fresh_session_factory() + + async with sf() as s: + await kol_x._ingest_kol_x(s, _KOL) + signal = (await s.execute( + select(KolPost).where(KolPost.external_id == "111") + )).scalar_one() + + assert signal.tier == "trade_signal" + assert signal.post_type == "original" + assert signal.talks_vs_trades_flag is True + assert signal.sentiment == "bearish" + + +@pytest.mark.asyncio +async def test_ingest_is_idempotent_on_rerun(monkeypatch): + _patch_io(monkeypatch) + sf = await _fresh_session_factory() + + async with sf() as s: + first = await kol_x._ingest_kol_x(s, _KOL) + second = await kol_x._ingest_kol_x(s, _KOL) + + assert first["new"] == 2 + assert second["new"] == 0 # everything already stored + assert second["skipped"] == 3 # 2 existing + 1 bare RT + + +@pytest.mark.asyncio +async def test_fetch_failure_does_not_raise(monkeypatch): + """A dead/blocked KOL must not abort the run — empty pages → empty stats.""" + monkeypatch.setattr(settings, "twitterapi_io_key", "test-key") + _patch_io(monkeypatch, pages=[]) # generator yields nothing + sf = await _fresh_session_factory() + async with sf() as s: + stats = await kol_x._ingest_kol_x(s, _KOL) + assert stats["new"] == 0 and stats["errors"] == 0 + + +# ── Multi-page & early-stop ─────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_multipage_fetches_all_new_tweets(monkeypatch): + """Two pages of distinct tweets → both pages are stored.""" + page1 = [{"id": "p1_1", "text": "SOL is the play", "createdAt": "Thu Jun 04 05:00:00 +0000 2026", "author": {"followers": 100000}}] + page2 = [{"id": "p2_1", "text": "ETH too slow ngmi", "createdAt": "Wed Jun 03 10:00:00 +0000 2026", "author": {"followers": 100000}}] + _patch_io(monkeypatch, pages=[page1, page2]) + sf = await _fresh_session_factory() + + async with sf() as s: + stats = await kol_x._ingest_kol_x(s, _KOL) + + assert stats["new"] == 2 + + +@pytest.mark.asyncio +async def test_early_stop_when_page_fully_deduped(monkeypatch): + """After page1 is stored, a second run should detect all-dedup on page1 + and break before page2 is consumed — verified by keeping a 'new' tweet + on page2 that must NOT appear in the DB if early-stop fired correctly.""" + page1 = [{"id": "old1", "text": "BTC to the moon", + "createdAt": "Thu Jun 04 05:00:00 +0000 2026", "author": {}}] + page2 = [{"id": "brand_new", "text": "Sold everything", + "createdAt": "Wed Jun 03 09:00:00 +0000 2026", "author": {}}] + + run = 0 + pages_yielded: list[list] = [] + + async def counting_iter(_username, max_pages=3): + nonlocal run + run += 1 + if run == 1: + # First run: only page1 (simulates single-page normal daily run) + pages_yielded.append(page1) + yield page1 + else: + # Second run: both pages available, but early-stop should fire at page1 + for page in [page1, page2]: + pages_yielded.append(page) + yield page + + async def fake_score(**_kw): + return dict(_FAKE_SCORE) + + monkeypatch.setattr(settings, "twitterapi_io_key", "test-key") + monkeypatch.setattr(kol_x, "_iter_tweet_pages", counting_iter) + monkeypatch.setattr(kol_x.x_analysis, "analyze_x_post", fake_score) + + sf = await _fresh_session_factory() + async with sf() as s: + first = await kol_x._ingest_kol_x(s, _KOL) + assert first["new"] == 1 # page1 stored (1 tweet) + + pages_yielded.clear() + second = await kol_x._ingest_kol_x(s, _KOL) + assert second["new"] == 0 # page1 all deduped → early stop + # Only page1 should have been yielded — early-stop broke before page2 + assert pages_yielded == [page1] + + # page2's tweet must NOT be in the DB (generator closed before it was yielded) + rows = (await s.execute( + select(KolPost).where(KolPost.source == "twitter") + )).scalars().all() + assert {r.external_id for r in rows} == {"old1"} + + +@pytest.mark.asyncio +async def test_all_retweet_page_does_not_early_stop(monkeypatch): + """A page that is ALL bare retweets (page_new=0, page_deduped=0) must NOT + trigger early-stop — the next page can still hold original posts. Regression + guard: previously 'page_new == 0' alone stopped paging on RT-heavy pages, + silently dropping originals on older pages.""" + page1 = [ + {"id": "rt1", "text": "RT @a: gm", "createdAt": "Thu Jun 04 05:00:00 +0000 2026", "author": {}}, + {"id": "rt2", "text": "RT @b: wagmi", "createdAt": "Thu Jun 04 04:00:00 +0000 2026", "author": {}}, + ] + page2 = [ + {"id": "orig1", "text": "Just sold all my $SOL", "createdAt": "Wed Jun 03 09:00:00 +0000 2026", "author": {}}, + ] + _patch_io(monkeypatch, pages=[page1, page2]) + sf = await _fresh_session_factory() + + async with sf() as s: + stats = await kol_x._ingest_kol_x(s, _KOL) + # page1 all-RT → skipped 2, NO early-stop → page2 consumed → orig1 stored + assert stats["new"] == 1 + assert stats["skipped"] == 2 + rows = (await s.execute( + select(KolPost).where(KolPost.source == "twitter") + )).scalars().all() + assert {r.external_id for r in rows} == {"orig1"} diff --git a/tests/test_posts_paged.py b/tests/test_posts_paged.py new file mode 100644 index 0000000..27f0542 --- /dev/null +++ b/tests/test_posts_paged.py @@ -0,0 +1,232 @@ +from __future__ import annotations + +from datetime import datetime, timedelta + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.api.posts import router as posts_router +from app.database import get_db +from app.models import Base, Post + + +@pytest.mark.asyncio +async def test_posts_paged_filters_and_counts(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:", future=True) + session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + now = datetime(2026, 5, 30, 12, 0, 0) + async with session_factory() as session: + session.add_all([ + Post( + external_id="truth-buy", + text="Bullish signal", + source="truth", + published_at=now, + sentiment="bullish", + signal="buy", + ai_confidence=91, + ai_reasoning="AI saw upside", + relevant=True, + ), + Post( + external_id="truth-short", + text="Bearish signal", + source="truth", + published_at=now - timedelta(minutes=1), + sentiment="bearish", + signal="short", + ai_confidence=88, + ai_reasoning="AI saw downside", + relevant=True, + ), + Post( + external_id="truth-hold", + text="Neutral but scored", + source="truth", + published_at=now - timedelta(minutes=2), + sentiment="neutral", + signal="hold", + ai_confidence=45, + ai_reasoning="No trade edge", + relevant=True, + ), + Post( + external_id="truth-noise", + text="Off-topic golf post", + source="truth", + published_at=now - timedelta(minutes=3), + sentiment="neutral", + signal=None, + ai_confidence=0, + ai_reasoning=None, + relevant=False, + ), + Post( + external_id="macro-buy", + text="Macro buy", + source="btc_bottom_reversal", + published_at=now - timedelta(minutes=4), + sentiment="bullish", + signal="buy", + ai_confidence=97, + ai_reasoning="Macro setup", + relevant=True, + ), + Post( + external_id="archive-breakout", + text="Legacy breakout signal", + source="breakout", + published_at=now - timedelta(minutes=5), + sentiment="bullish", + signal="buy", + ai_confidence=73, + ai_reasoning="Old scanner", + relevant=True, + ), + Post( + external_id="archive-sma", + text="Legacy sma reclaim signal", + source="sma_reclaim", + published_at=now - timedelta(minutes=6), + sentiment="bullish", + signal="short", + ai_confidence=64, + ai_reasoning="Old scanner 2", + relevant=True, + ), + ]) + await session.commit() + + app = FastAPI() + app.include_router(posts_router, prefix="/api") + + async def override_get_db(): + async with session_factory() as session: + yield session + + app.dependency_overrides[get_db] = override_get_db + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://testserver") as client: + res = await client.get("/api/posts-paged", params={"source": "truth", "limit": 10, "page": 1}) + assert res.status_code == 200 + data = res.json() + assert data["total"] == 4 + assert len(data["items"]) == 4 + assert data["counts"] == { + "all": 4, + "actionable": 2, + "buy": 1, + "short": 1, + "off_topic": 1, + } + assert data["source_counts"] == [{"source": "truth", "count": 4, "latest": "2026-05-30T12:00:00.000Z"}] + + actionable = await client.get( + "/api/posts-paged", + params={"source": "truth", "signal": "actionable", "limit": 10, "page": 1}, + ) + assert actionable.status_code == 200 + actionable_data = actionable.json() + assert actionable_data["total"] == 2 + assert {item["signal"] for item in actionable_data["items"]} == {"buy", "short"} + assert actionable_data["counts"]["all"] == 4 + + scored_only = await client.get( + "/api/posts-paged", + params={"source": "truth", "ai_scored_only": "true", "limit": 10, "page": 1}, + ) + assert scored_only.status_code == 200 + scored_data = scored_only.json() + assert scored_data["total"] == 3 + assert all(item["ai_confidence"] > 0 or item["ai_reasoning"] for item in scored_data["items"]) + assert scored_data["counts"] == { + "all": 3, + "actionable": 2, + "buy": 1, + "short": 1, + "off_topic": 1, + } + + bearish_buy = await client.get( + "/api/posts-paged", + params={ + "source": "truth", + "sentiment": "bearish", + "signal": "buy", + "limit": 10, + "page": 1, + }, + ) + assert bearish_buy.status_code == 200 + bearish_buy_data = bearish_buy.json() + assert bearish_buy_data["total"] == 0 + assert bearish_buy_data["counts"] == { + "all": 1, + "actionable": 1, + "buy": 0, + "short": 1, + "off_topic": 0, + } + + archive = await client.get( + "/api/posts-paged", + params={ + "archive_only": "true", + "limit": 10, + "page": 1, + }, + ) + assert archive.status_code == 200 + archive_data = archive.json() + assert archive_data["total"] == 2 + assert len(archive_data["items"]) == 2 + assert {item["source"] for item in archive_data["items"]} == {"breakout", "sma_reclaim"} + assert archive_data["source_counts"] == [ + {"source": "breakout", "count": 1, "latest": "2026-05-30T11:55:00.000Z"}, + {"source": "sma_reclaim", "count": 1, "latest": "2026-05-30T11:54:00.000Z"}, + ] + + # Regression: selecting one archive source via source_in must narrow the + # paged items/total, but source_counts (the chip bar) must STILL list + # every archived source so the UI can offer a way back to "all". + archive_one = await client.get( + "/api/posts-paged", + params={ + "archive_only": "true", + "source_in": "breakout", + "limit": 10, + "page": 1, + }, + ) + assert archive_one.status_code == 200 + archive_one_data = archive_one.json() + assert archive_one_data["total"] == 1 + assert {item["source"] for item in archive_one_data["items"]} == {"breakout"} + # Chip bar is NOT collapsed to the selected source. + assert archive_one_data["source_counts"] == [ + {"source": "breakout", "count": 1, "latest": "2026-05-30T11:55:00.000Z"}, + {"source": "sma_reclaim", "count": 1, "latest": "2026-05-30T11:54:00.000Z"}, + ] + + archive_compat = await client.get( + "/api/posts-paged", + params={ + "archive_only": "true", + "source_not_in": "truth", + "limit": 10, + "page": 1, + }, + ) + assert archive_compat.status_code == 200 + archive_compat_data = archive_compat.json() + assert archive_compat_data["total"] == 2 + assert {item["source"] for item in archive_compat_data["items"]} == {"breakout", "sma_reclaim"} + + await engine.dispose() diff --git a/tests/test_production_readiness.py b/tests/test_production_readiness.py index 2a4e858..1a07281 100644 --- a/tests/test_production_readiness.py +++ b/tests/test_production_readiness.py @@ -41,6 +41,12 @@ class _Trade: pnl_usd = 2.25 +class _ClosedTrade(_Trade): + """Same as _Trade but with closed_at set, simulating a committed close.""" + from datetime import datetime, timezone + closed_at = datetime(2026, 1, 1, 0, 0, 0) + + class _Sub: leverage = 3 hl_api_key = None @@ -65,7 +71,10 @@ async def test_manual_close_returns_close_result(monkeypatch): monkeypatch.setattr(bot_engine, "close_and_finalize", fake_close_and_finalize) - db = _Db([_Trade(), _Sub(), _Trade()]) + # Responses in order: load trade → load sub → populate_existing re-read + # (B45 fix: one query with populate_existing=True instead of old stale cache). + # _ClosedTrade has closed_at set so the B46 success guard passes. + db = _Db([_Trade(), _Sub(), _ClosedTrade()]) result = await positions.manual_close(7, _Request(), db) diff --git a/tests/test_telegram_digest.py b/tests/test_telegram_digest.py index f6b5c5c..996f54a 100644 --- a/tests/test_telegram_digest.py +++ b/tests/test_telegram_digest.py @@ -10,6 +10,8 @@ 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, ) @@ -174,3 +176,48 @@ def test_total_length_well_under_telegram_cap(): "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 diff --git a/tests/test_trade_alerts.py b/tests/test_trade_alerts.py new file mode 100644 index 0000000..655761b --- /dev/null +++ b/tests/test_trade_alerts.py @@ -0,0 +1,150 @@ +"""Tests for trade-alert broadcasts and balance pre-check logic added 2026-06-01. + +Coverage targets: + 1. _broadcast_trade_alert — fire-and-forget, must not raise + 2. Balance pre-check maths — required_margin = (notional / leverage) * 1.1 + 3. Startup drain in the Telegram bot loop — offset advances past pending updates +""" +from __future__ import annotations + +import asyncio +import pytest + +# ── 1. _broadcast_trade_alert ───────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_broadcast_trade_alert_no_connections(monkeypatch): + """broadcast_trade_alert must silently succeed when no WS clients are connected.""" + from app.services.bot_engine import _broadcast_trade_alert + from app.ws import manager as mgr_mod + + # Patch manager.broadcast to verify it's called with correct payload + calls: list[dict] = [] + async def fake_broadcast(msg: dict): + calls.append(msg) + + monkeypatch.setattr(mgr_mod.manager, "broadcast", fake_broadcast) + + await _broadcast_trade_alert("0xabc", "execution_failed", asset="BTC", reason="test error") + + assert len(calls) == 1 + assert calls[0]["type"] == "trade_alert" + assert calls[0]["wallet"] == "0xabc" + assert calls[0]["event"] == "execution_failed" + assert calls[0]["asset"] == "BTC" + + +@pytest.mark.asyncio +async def test_broadcast_trade_alert_swallows_exceptions(monkeypatch): + """broadcast_trade_alert must not propagate exceptions — trade flow must continue.""" + from app.services.bot_engine import _broadcast_trade_alert + from app.ws import manager as mgr_mod + + async def exploding_broadcast(msg: dict): + raise RuntimeError("WS layer crashed") + + monkeypatch.setattr(mgr_mod.manager, "broadcast", exploding_broadcast) + + # Must not raise + await _broadcast_trade_alert("0xabc", "budget_reached", asset="BTC") + + +# ── 2. Balance pre-check maths ──────────────────────────────────────────────── + +def test_required_margin_formula(): + """required_margin = (notional / leverage) * 1.1 — verify key scenarios.""" + def required_margin(notional: float, leverage: int) -> float: + return round((notional / max(leverage, 1)) * 1.1, 2) + + # 2× leverage, $100 position → $50 margin + 10% buffer = $55 + assert required_margin(100, 2) == 55.0 + + # 5× leverage, $500 position → $100 margin + 10% = $110 + assert required_margin(500, 5) == 110.0 + + # 1× leverage, $20 position → $20 + 10% = $22 + assert required_margin(20, 1) == 22.0 + + # edge: leverage=0 treated as 1 (no divide-by-zero) + assert required_margin(100, 0) == 110.0 + + +def test_balance_check_does_not_block_low_leverage(): + """With $100 balance and 10× leverage on a $200 notional, margin = $22 — should PASS.""" + balance = 100.0 + notional = 200.0 + leverage = 10 + required = (notional / max(leverage, 1)) * 1.1 + assert balance >= required, ( + f"Balance ${balance} should cover ${required:.2f} margin " + f"(${notional} notional at {leverage}×)" + ) + + +def test_balance_check_fires_at_truly_insufficient_balance(): + """With $5 balance and 2× on $20 notional, margin = $11 — should BLOCK.""" + balance = 5.0 + notional = 20.0 + leverage = 2 + required = (notional / max(leverage, 1)) * 1.1 + assert balance < required, ( + f"Balance ${balance} should NOT cover ${required:.2f} margin" + ) + + +# ── 3. Telegram startup drain ──────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_telegram_startup_drain_advances_offset(monkeypatch): + """Verify the drain calls getUpdates with timeout=0 and ACKs the last update_id.""" + import httpx + + drain_calls: list[dict] = [] + + class FakeResponse: + status_code = 200 + def json(self): + if len(drain_calls) == 1: # first call: return pending updates + return {"result": [{"update_id": 100}, {"update_id": 101}]} + return {"result": []} # second call (ACK): no pending + + class FakeClient: + async def __aenter__(self): return self + async def __aexit__(self, *a): pass + async def get(self, url, params=None): + drain_calls.append(params or {}) + return FakeResponse() + + monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: FakeClient()) + + # Simulate only the drain portion (extract the logic) + from app.config import settings + monkeypatch.setattr(settings, "telegram_bot_token", "test-token") + + import httpx as _httpx + + # Re-run the drain logic inline (mirrors telegram_bot.py startup drain) + token = "test-token" + TG_API = "https://api.telegram.org/bot{token}/{method}" + + async with _httpx.AsyncClient(timeout=10) as client: + r = await client.get( + TG_API.format(token=token, method="getUpdates"), + params={"timeout": 0, "limit": 100}, + ) + if r.status_code == 200: + pending = r.json().get("result", []) + if pending: + drain_offset = pending[-1]["update_id"] + 1 + async with _httpx.AsyncClient(timeout=10) as client: + await client.get( + TG_API.format(token=token, method="getUpdates"), + params={"timeout": 0, "offset": drain_offset}, + ) + + # First call: drain request (timeout=0, limit=100) + assert drain_calls[0].get("timeout") == 0 + assert drain_calls[0].get("limit") == 100 + + # Second call: ACK with offset = last_update_id + 1 + assert drain_calls[1].get("offset") == 102 # 101 + 1 diff --git a/tests/test_x_analysis.py b/tests/test_x_analysis.py new file mode 100644 index 0000000..4ab7945 --- /dev/null +++ b/tests/test_x_analysis.py @@ -0,0 +1,217 @@ +"""Tests for x_analysis — the DETERMINISTIC normalization layer. + +The LLM call is mocked so each test feeds a controlled raw model response and +asserts how analyze_x_post normalizes it. This locks down the enforcement rules +that protect downstream consumers (kol_x → kol_divergence), independent of +whatever the model actually returns: + + - retweet post_type → forced noise + - trade_signal requires a buy/sell/reduce ticker with conviction ≥ 0.7, + else it downgrades to directional (never silently dropped) + - noise → tickers cleared + talks_vs_trades_flag forced false + - ticker hygiene: conviction clamped 0..1, bad action → mention, + overlong symbol dropped + - invalid tier/post_type → safe defaults + - bad JSON / empty text → graceful fallback, never raises + +These are pure logic (no network, no AI spend) so they're fast and stable. +""" + +from __future__ import annotations + +import json + +import pytest + +from app.config import settings +from app.services import x_analysis + + +# ── helpers ──────────────────────────────────────────────────────────────── + +def _patch_llm(monkeypatch, raw: dict): + """Force analyze_x_post's LLM call to return `raw` (serialized). We patch + the OpenAI path (anthropic key blanked) since that's the default in CI.""" + payload = json.dumps(raw) + + class _Msg: + content = payload + + class _Choice: + message = _Msg() + + class _Resp: + choices = [_Choice()] + + class _Completions: + async def create(self, **_kw): + return _Resp() + + class _Chat: + completions = _Completions() + + class _Client: + chat = _Chat() + + monkeypatch.setattr(settings, "anthropic_api_key", "") # → OpenAI path + monkeypatch.setattr(x_analysis, "_oai", lambda: _Client()) + + +def _raw(**over): + """Minimal well-formed raw model response; override per test.""" + base = { + "post_type": "original", + "tier": "directional", + "summary": "x", + "tickers": [], + "talks_vs_trades_flag": False, + "has_price_target": False, + "price_targets": [], + "sentiment": "neutral", + "reasoning": "y", + } + base.update(over) + return base + + +# ── tier enforcement ─────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_retweet_post_type_forced_to_noise(monkeypatch): + _patch_llm(monkeypatch, _raw( + post_type="retweet", tier="trade_signal", + tickers=[{"ticker": "BTC", "action": "buy", "conviction": 0.9}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="some original-looking text body") + assert r["tier"] == "noise" + assert r["tickers"] == [] + + +@pytest.mark.asyncio +async def test_trade_signal_downgrades_without_strong_action(monkeypatch): + # tier=trade_signal but only a 'bullish' ticker (not buy/sell/reduce) + _patch_llm(monkeypatch, _raw( + tier="trade_signal", + tickers=[{"ticker": "SOL", "action": "bullish", "conviction": 0.9}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="SOL looking strong") + assert r["tier"] == "directional" + assert r["tickers"][0]["ticker"] == "SOL" + + +@pytest.mark.asyncio +async def test_trade_signal_downgrades_on_low_conviction(monkeypatch): + _patch_llm(monkeypatch, _raw( + tier="trade_signal", + tickers=[{"ticker": "SOL", "action": "buy", "conviction": 0.5}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="bought a little SOL maybe") + assert r["tier"] == "directional" # 0.5 < 0.7 floor + + +@pytest.mark.asyncio +async def test_trade_signal_kept_when_strong(monkeypatch): + _patch_llm(monkeypatch, _raw( + tier="trade_signal", + tickers=[{"ticker": "SOL", "action": "buy", "conviction": 0.9}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="aped SOL full size lfg") + assert r["tier"] == "trade_signal" + + +@pytest.mark.asyncio +async def test_invalid_tier_falls_back_to_noise(monkeypatch): + _patch_llm(monkeypatch, _raw(tier="超级买入")) + r = await x_analysis.analyze_x_post(handle="k", text="ambiguous content here") + assert r["tier"] == "noise" + + +# ── noise enforcement ────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_noise_clears_tickers_and_flag(monkeypatch): + _patch_llm(monkeypatch, _raw( + tier="noise", talks_vs_trades_flag=True, + tickers=[{"ticker": "BTC", "action": "buy", "conviction": 0.9}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="gm frens beautiful day") + assert r["tickers"] == [] + assert r["talks_vs_trades_flag"] is False + + +# ── ticker hygiene ───────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_conviction_clamped_to_unit_interval(monkeypatch): + _patch_llm(monkeypatch, _raw( + tier="directional", + tickers=[{"ticker": "BTC", "action": "bullish", "conviction": 1.8}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="BTC going parabolic") + assert r["tickers"][0]["conviction"] == 1.0 + + +@pytest.mark.asyncio +async def test_invalid_action_becomes_mention(monkeypatch): + _patch_llm(monkeypatch, _raw( + tier="directional", + tickers=[{"ticker": "BTC", "action": "yolo", "conviction": 0.5}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="BTC yolo time") + assert r["tickers"][0]["action"] == "mention" + + +@pytest.mark.asyncio +async def test_overlong_ticker_dropped(monkeypatch): + _patch_llm(monkeypatch, _raw( + tier="directional", + tickers=[{"ticker": "THISISWAYTOOLONG", "action": "bullish", "conviction": 0.5}], + )) + r = await x_analysis.analyze_x_post(handle="k", text="some long token mention") + assert r["tickers"] == [] + + +# ── graceful failure ─────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_empty_text_short_circuits_without_llm(monkeypatch): + # no _patch_llm → if it called the LLM it would error; it must not. + r = await x_analysis.analyze_x_post(handle="k", text=" ") + assert r["tier"] == "noise" + assert r["error"] == "empty post" + + +@pytest.mark.asyncio +async def test_bare_retweet_prefiltered_without_llm(monkeypatch): + r = await x_analysis.analyze_x_post(handle="k", text="RT @someone: gm") + assert r["tier"] == "noise" + assert r["post_type"] == "retweet" + + +@pytest.mark.asyncio +async def test_bad_json_returns_fallback(monkeypatch): + class _Msg: + content = "not json at all {{{ " + + class _Choice: + message = _Msg() + + class _Resp: + choices = [_Choice()] + + class _Completions: + async def create(self, **_kw): + return _Resp() + + class _Chat: + completions = _Completions() + + class _Client: + chat = _Chat() + + monkeypatch.setattr(settings, "anthropic_api_key", "") + monkeypatch.setattr(x_analysis, "_oai", lambda: _Client()) + + r = await x_analysis.analyze_x_post(handle="k", text="real content that triggers llm") + assert r["tier"] == "noise" + assert r["error"] and "parse_error" in r["error"]