diff --git a/alembic/versions/023_telegram_digest_prefs.py b/alembic/versions/023_telegram_digest_prefs.py new file mode 100644 index 0000000..a9497f0 --- /dev/null +++ b/alembic/versions/023_telegram_digest_prefs.py @@ -0,0 +1,53 @@ +"""Telegram per-user digest preferences. + +Adds three columns to telegram_bindings so each subscriber can opt in/out +of the daily three-section brief (Macro Vibes / KOL / Trump) and pick +which UTC hour it lands. + + digest_enabled : bool, default True. Toggleable via /digest on|off. + digest_hour_utc : int 0–23, default 12. Toggleable via /digest_time HH. + 12 UTC ≈ Asia evening / EU afternoon / US morning — + the only single hour that's reasonable for all three + big crypto-active zones, so it's our default. + last_digest_sent_at : nullable timestamp. Guards against double-send if the + cron coalesces or the worker restarts mid-firing. + Used as "skip if < 23h ago" in send_daily_digest. + +Backfills are safe: every existing binding gets digest_enabled=True so users +who subscribed before this feature start receiving the brief automatically. +If they don't want it, one /digest off keeps them off. + +Revision ID: 023 +Revises: 022 +Create Date: 2026-05-26 +""" +from alembic import op +import sqlalchemy as sa + + +revision = "023" +down_revision = "022" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("telegram_bindings") as batch: + batch.add_column(sa.Column( + "digest_enabled", sa.Boolean, nullable=False, + server_default=sa.true(), + )) + batch.add_column(sa.Column( + "digest_hour_utc", sa.Integer, nullable=False, + server_default="12", + )) + batch.add_column(sa.Column( + "last_digest_sent_at", sa.DateTime, nullable=True, + )) + + +def downgrade() -> None: + with op.batch_alter_table("telegram_bindings") as batch: + batch.drop_column("last_digest_sent_at") + batch.drop_column("digest_hour_utc") + batch.drop_column("digest_enabled") diff --git a/alembic/versions/024_bot_trades_released_at.py b/alembic/versions/024_bot_trades_released_at.py new file mode 100644 index 0000000..33eb93f --- /dev/null +++ b/alembic/versions/024_bot_trades_released_at.py @@ -0,0 +1,44 @@ +"""BotTrade.released_at — user 'release from bot management' marker. + +Adds a single nullable timestamp column. When set: + - recovery.py rehydration SKIPS this row (no watchdog re-register on + restart) — the bot has stopped managing this position + - reconciler / close paths don't touch it + - the HL position itself is NOT closed; the user has taken back manual + control of the open trade + +This is the escape hatch for "adopt → bot manages → user wants control back +without closing". The bot stops driving but the HL position keeps running +under the user's own decisions. + +Distinct from `closed_at`: a closed trade has a known exit_price/pnl and is +finalised. A released trade has neither — it's still open on HL, just no +longer in our watchdog. + +Backfill: every existing row gets released_at = NULL. No semantic change to +any historical or already-open trade. + +Revision ID: 024 +Revises: 023 +Create Date: 2026-05-26 +""" +from alembic import op +import sqlalchemy as sa + + +revision = "024" +down_revision = "023" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("bot_trades") as batch: + batch.add_column(sa.Column( + "released_at", sa.DateTime, nullable=True, + )) + + +def downgrade() -> None: + with op.batch_alter_table("bot_trades") as batch: + batch.drop_column("released_at") diff --git a/app/api/positions.py b/app/api/positions.py index 42a3445..ff9e5c7 100644 --- a/app/api/positions.py +++ b/app/api/positions.py @@ -158,10 +158,15 @@ async def get_open_positions( body=None, allow_replay=True, ) + # Exclude released trades — those rows are "closed_at IS NULL" but the + # user has explicitly taken back manual control of the HL position. The + # bot is no longer managing them, so they shouldn't appear in the live + # positions tile (which would suggest the bot is still driving stops). rows = await db.execute( select(BotTrade).where( BotTrade.wallet_address == wallet, BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), ).order_by(BotTrade.opened_at.desc()) ) trades = rows.scalars().all() @@ -203,10 +208,13 @@ async def get_today_stats( ) closed = closed_rows.scalars().all() + # See /positions/open: released trades aren't actively managed by the + # bot, so they shouldn't inflate the open-count tile. open_rows = await db.execute( select(BotTrade).where( BotTrade.wallet_address == wallet, BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), ) ) open_trades = open_rows.scalars().all() @@ -234,6 +242,8 @@ async def get_today_stats( ACTION_CLOSE_TRADE = "close_trade" ACTION_SET_GROW = "set_trade_grow" ACTION_VIEW_POSITIONS = "view_positions" +ACTION_ADOPT_POSITION = "adopt_position" +ACTION_RELEASE_TRADE = "release_trade" class CloseTradeResponse(BaseModel): @@ -390,3 +400,165 @@ async def set_trade_grow( logger.info("Grow %s for trade %d by %s", "ON" if enabled else "OFF", trade_id, wallet) return GrowResponse(trade_id=trade_id, grow_mode=enabled) + + +# ─── Adopt / release (System-2 manage-only flow) ──────────────────────────── + + +class HLPositionItem(BaseModel): + asset: str + side: str + size_coins: float + entry_price: float + leverage: int + size_usd: float + already_adopted: bool + + +class HLPositionsResponse(BaseModel): + wallet: str + count: int + positions: list[HLPositionItem] + + +@router.get("/positions/hl/{wallet}", response_model=HLPositionsResponse) +async def list_hl_positions_endpoint(wallet: str): + """Read the wallet's CURRENT Hyperliquid open positions, annotated with + 'already adopted' flag. Used by the Adopt picker on the frontend and by + the Telegram /adopt command. + + No auth — same trust model as /positions/open (wallet address IS the + access key). Reading HL state is read-only and harmless. + """ + from app.services.adoption import list_hl_positions, AdoptionError + try: + items = await list_hl_positions(wallet) + except AdoptionError as exc: + raise HTTPException(400, exc.message) + return HLPositionsResponse( + wallet=wallet.lower().strip(), + count=len(items), + positions=[HLPositionItem(**vars(it)) for it in items], + ) + + +class AdoptResponse(BaseModel): + status: str + trade_id: int + asset: str + side: str + entry_price: float + size_usd: float + leverage: int + mode: str + protective_stop: float + + +@router.post("/positions/adopt", response_model=AdoptResponse) +async def adopt_position_endpoint(request: Request, + db: AsyncSession = Depends(get_db)): + """Hand a manually-opened HL position to the bot. + + Body: { wallet, timestamp, signature, asset, mode } + asset : e.g. "BTC". Case-insensitive. + mode : "standard" | "aggressive" — picks ladder spacing. + + On success: a BotTrade row is created, sys2 stop ladder + de-risk + + pyramid + peak-trail registered with the live watchdog. From that + moment, the bot drives the trade. + + Failure surfaces an AdoptionError mapped to a 400 with .code in the + response body so the UI can switch on it (e.g. 'already_adopted', + 'concurrency_cap', 'not_on_hl'). + """ + raw = await request.json() + wallet = (raw.get("wallet") or "").lower().strip() + timestamp = raw.get("timestamp") + signature = raw.get("signature") + asset = (raw.get("asset") or "").upper().strip() + mode = (raw.get("mode") or "standard").strip() + + if not wallet: + raise HTTPException(422, "wallet required") + if not isinstance(timestamp, int): + raise HTTPException(422, "timestamp (ms) required") + if not isinstance(signature, str) or not signature: + raise HTTPException(422, "signature required") + if not asset: + raise HTTPException(422, "asset required (e.g. 'BTC')") + if mode not in ("standard", "aggressive"): + raise HTTPException(422, "mode must be 'standard' or 'aggressive'") + + verify_signed_request( + action=ACTION_ADOPT_POSITION, + wallet=wallet, timestamp_ms=timestamp, signature=signature, + body={"asset": asset, "mode": mode}, + ) + + from app.services.adoption import adopt_position, AdoptionError + try: + result = await adopt_position(wallet, asset, mode) + except AdoptionError as exc: + # 409 for state conflicts (already adopted / concurrency), 400 else. + status = 409 if exc.code in ( + "already_adopted", "concurrency_cap", "already_closed", + ) else 400 + raise HTTPException(status, {"code": exc.code, "message": exc.message}) + + return AdoptResponse( + status="ok", + trade_id=result.trade_id, + asset=result.asset, side=result.side, + entry_price=result.entry_price, size_usd=result.size_usd, + leverage=result.leverage, mode=result.mode, + protective_stop=result.protective_stop, + ) + + +class ReleaseResponse(BaseModel): + status: str + trade_id: int + asset: str + already_released: bool + + +@router.post("/positions/{trade_id}/release", response_model=ReleaseResponse) +async def release_trade_endpoint(trade_id: int, request: Request): + """Hand control of trade_id back to the user. + + Body: { wallet, timestamp, signature, trade_id } + + Marks released_at on the BotTrade row, unregisters the watchdog. + The HL position itself is NOT touched — it keeps running under the + user's manual control. Idempotent: releasing a released trade returns + success. + """ + raw = await request.json() + wallet = (raw.get("wallet") or "").lower().strip() + timestamp = raw.get("timestamp") + signature = raw.get("signature") + body_tid = raw.get("trade_id") + + if not wallet: + raise HTTPException(422, "wallet required") + if not isinstance(timestamp, int): + raise HTTPException(422, "timestamp (ms) required") + if not isinstance(signature, str) or not signature: + raise HTTPException(422, "signature required") + if body_tid != trade_id: + raise HTTPException(400, "trade_id mismatch (path vs signed body)") + + verify_signed_request( + action=ACTION_RELEASE_TRADE, + wallet=wallet, timestamp_ms=timestamp, signature=signature, + body={"trade_id": trade_id}, + ) + + from app.services.adoption import release_management, AdoptionError + try: + result = await release_management(wallet, trade_id) + except AdoptionError as exc: + status = 404 if exc.code == "not_found" else ( + 403 if exc.code == "not_owner" else 400) + raise HTTPException(status, {"code": exc.code, "message": exc.message}) + return ReleaseResponse(**result) diff --git a/app/main.py b/app/main.py index e40b9fc..538d694 100644 --- a/app/main.py +++ b/app/main.py @@ -209,6 +209,22 @@ async def lifespan(app: FastAPI): type(exc).__name__, exc) asyncio.create_task(_macro_bootstrap()) + # ── Telegram daily digest ───────────────────────────────────────────── + # Runs every hour at :00. Each binding picks its own digest_hour_utc + # (0–23, default 12); the sender only fans out to rows whose preferred + # hour matches the current one — so we get one cron job covering all 24 + # delivery slots. The send function itself dedupes against + # last_digest_sent_at (skips if < 23h), so a coalesced cron or worker + # restart can't double-send a user in the same day. + from app.services.telegram_digest import send_daily_digest + _scheduler.add_job( + send_daily_digest, "cron", minute=0, + id="telegram_daily_digest", + max_instances=1, coalesce=True, + ) + logger.info("Telegram daily digest scheduled hourly at :00 UTC " + "(per-user hour selection).") + _scheduler.start() logger.info( "Truth Social pollers scheduled every %ds (CNN + trumpstruth.org).", diff --git a/app/models.py b/app/models.py index 5332714..e81ac3f 100644 --- a/app/models.py +++ b/app/models.py @@ -98,6 +98,12 @@ class BotTrade(Base): wallet_address: Mapped[str] = mapped_column(String(64), nullable=False, index=True) opened_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=utcnow) closed_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + # User-initiated "stop managing" marker. When set: recovery skips + # re-register on restart, monitor stops driving, HL position keeps + # running under the user's own control. NOT the same as closed_at — + # there's no exit_price / pnl on a released row. Used by /release in + # the adopt-then-release flow for System-2 positions. + released_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) hl_order_id: Mapped[Optional[str]] = mapped_column(String(128), nullable=True) # Snapshot of user settings AT TIME OF ENTRY. Required so changes to # Subscription.position_size_usd / leverage after-the-fact don't corrupt @@ -414,6 +420,14 @@ class TelegramBinding(Base): total_alerts_sent: Mapped[int] = mapped_column(Integer, nullable=False, default=0) total_alerts_failed: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # Daily 3-section digest (Macro / KOL / Trump). Independent of the + # per-signal alerts above — a user can disable all real-time alerts but + # still want the once-a-day overview, or vice versa. See migration 023 + # and app/services/telegram_digest.py. + digest_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + digest_hour_utc: Mapped[int] = mapped_column(Integer, nullable=False, default=12) + last_digest_sent_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + class MacroSnapshot(Base): """Daily snapshot of all macro indicators surfaced on the BTC page. diff --git a/app/services/adoption.py b/app/services/adoption.py new file mode 100644 index 0000000..d4b1f54 --- /dev/null +++ b/app/services/adoption.py @@ -0,0 +1,439 @@ +"""Adopt / release flow for System-2 manage-only positions. + +The user opens a position MANUALLY on Hyperliquid (any size / leverage). +They then ask the bot to manage it — either via the dashboard adoption +button or the Telegram /adopt command. From that moment on, tp_sl_monitor +runs the full System-2 lifecycle against it: 5-rung stop ladder, downside +de-risk, pyramid, peak-trail, full close. + +`release` is the escape hatch. It marks released_at on the BotTrade row, +unregisters the watchdog, and leaves the HL position untouched — the user +takes back manual control without closing. + +The flow is intentionally LIGHT: + - We don't validate signal timing (any HL position can be adopted) + - We don't enforce direction (both long and short work; ladder is signed) + - We DO enforce the same concurrency cap as the old auto-open path + (SYS2_MAX_CONCURRENT) — adopting unlimited correlated crypto-beta is + the same disaster regardless of who pulled the trigger + - We DO block paper-mode subscribers (paper has no HL state to read) + +Race notes: + - Adopt creates the BotTrade row, then register_trade. If a price tick + arrives between those two events, no harm — register_trade is + idempotent on trade_id. + - Two concurrent adopts on the same wallet+asset are guarded by the + "already adopted?" check inside the wallet-keyed asyncio lock reused + from bot_engine. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Dict, Optional + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.database import AsyncSessionLocal +from app.models import BotTrade, Subscription +from app.config import settings +from app.services.crypto import decrypt_api_key +from app.services.hyperliquid import HyperliquidTrader +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, +) + +logger = logging.getLogger(__name__) + + +# Default category — the System-2 strategy is BTC bottom-reversal long. +# An adopted short or non-BTC asset still gets the same generous ladder +# (it's direction-agnostic), but we tag the category for telemetry. +ADOPTED_CATEGORY = "btc_bottom_reversal_long" + + +# Per-wallet asyncio lock to serialise adopt_position calls for the same +# wallet. Without this, two concurrent /adopt requests on the same wallet +# can both pass the duplicate-check and end up writing two BotTrade rows +# pointing at the same HL position — corrupting the ladder / de-risk math +# (each row would race to reduce the position). Process-local; with the +# atomic duplicate check inside the lock, multi-process deploys are still +# safe (losers would just see "already_adopted" on the second attempt). +_adopt_locks: Dict[str, asyncio.Lock] = {} + + +def _wallet_adopt_lock(wallet: str) -> asyncio.Lock: + lock = _adopt_locks.get(wallet) + if lock is None: + lock = asyncio.Lock() + _adopt_locks[wallet] = lock + return lock + + +# ── Errors ────────────────────────────────────────────────────────────────── + + +class AdoptionError(Exception): + """User-facing adoption failure. .code is a stable short token suitable + for switching on in the API / Telegram layers; .message is the + human-readable text.""" + def __init__(self, code: str, message: str): + super().__init__(message) + self.code = code + self.message = message + + +# ── Reads ─────────────────────────────────────────────────────────────────── + + +@dataclass +class HLPositionView: + """Lightweight projection of a Hyperliquid open position, for adoption + pickers. Kept thin so the UI / Telegram layers don't depend on the + HL SDK's dict shape.""" + asset: str + side: str # "long" | "short" + size_coins: float # absolute (always positive) + entry_price: float + leverage: int + size_usd: float # size_coins * entry_price + already_adopted: bool # there's a BotTrade open + not released for it + + +async def list_hl_positions(wallet: str) -> list[HLPositionView]: + """Read the wallet's current HL open positions, annotated with whether + each is already under bot management. Raises AdoptionError on missing + subscription / missing key / HL read failure.""" + wallet_l = wallet.lower().strip() + async with AsyncSessionLocal() as db: + sub = (await db.execute( + select(Subscription).where(Subscription.wallet_address == wallet_l) + )).scalar_one_or_none() + if sub is None: + raise AdoptionError("no_subscription", + "Wallet not subscribed. Open Settings on the dashboard first.") + if not sub.hl_api_key: + raise AdoptionError("no_hl_key", + "This wallet has no Hyperliquid API key on file. " + "Add one in Settings before adopting positions.") + # Adopted positions only — closed/released rows don't block re-adopt. + existing = (await db.execute( + select(BotTrade.asset).where( + BotTrade.wallet_address == wallet_l, + BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), + ) + )).scalars().all() + adopted_assets = {a.upper() for a in existing} + + try: + api_key = decrypt_api_key(sub.hl_api_key) + except Exception as exc: + raise AdoptionError("key_decrypt_failed", + f"Could not decrypt the HL API key: {exc}") + + trader = HyperliquidTrader( + api_private_key=api_key, + account_address=wallet_l, + leverage=int(sub.leverage or 2), + mainnet=settings.hl_mainnet, + ) + try: + raw = await trader.get_open_positions() + except Exception as exc: + raise AdoptionError("hl_read_failed", + f"Could not read HL positions: {exc}") + + out: list[HLPositionView] = [] + for p in raw: + coin = (p.get("coin") or "").upper() + szi = float(p.get("szi") or 0.0) + entry = float(p.get("entry_px") or 0.0) + lev_info = p.get("leverage") or {} + # HL position.leverage shape is {"type": "isolated"|"cross", "value": N} + lev = int(lev_info.get("value") or sub.leverage or 2) + if szi == 0 or entry <= 0: + continue + out.append(HLPositionView( + asset = coin, + side = "long" if szi > 0 else "short", + size_coins = abs(szi), + entry_price = entry, + leverage = lev, + size_usd = round(abs(szi) * entry, 2), + already_adopted = coin in adopted_assets, + )) + return out + + +# ── Adopt ─────────────────────────────────────────────────────────────────── + + +@dataclass +class AdoptionResult: + trade_id: int + asset: str + side: str + entry_price: float + size_usd: float + leverage: int + mode: str + protective_stop: float # positive % distance + derisk_ladder: list # the rungs we just registered + stop_ladder: list # the upside ratchet rungs + + +async def adopt_position(wallet: str, asset: str, mode: str) -> AdoptionResult: + """Read the user's current HL position for `asset`, create a BotTrade + row with the System-2 ladder for `mode`, and register it with the + watchdog. Raises AdoptionError on any failure path. + + Wallet-level locking serialises concurrent adopts for the same wallet + so the duplicate / concurrency-cap checks below are race-free. + """ + wallet_l = wallet.lower().strip() + asset_u = asset.upper().strip() + mode_n = sys2_normalize_mode(mode) + if mode_n not in SYS2_MODES: + raise AdoptionError("bad_mode", + f"mode must be one of {SYS2_MODES}; got {mode!r}") + + async with _wallet_adopt_lock(wallet_l): + return await _adopt_locked(wallet_l, asset_u, mode_n) + + +async def _adopt_locked(wallet_l: str, asset_u: str, mode_n: str) -> AdoptionResult: + async with AsyncSessionLocal() as db: + sub = (await db.execute( + select(Subscription).where(Subscription.wallet_address == wallet_l) + )).scalar_one_or_none() + if sub is None: + raise AdoptionError("no_subscription", + "Wallet not subscribed. Open Settings on the dashboard first.") + if not sub.hl_api_key: + raise AdoptionError("no_hl_key", + "This wallet has no Hyperliquid API key on file.") + # Paper-mode wallets have no real HL position to manage. The HL read + # below would return empty and surface a confusing "not_on_hl" error + # — better to reject explicitly with a clear cause up front. + if getattr(sub, "paper_mode", False): + raise AdoptionError("paper_mode", + "Adopt isn't available in paper mode — paper trades have " + "no Hyperliquid position to manage. Turn off paper mode in " + "Settings to use /adopt.") + + # System-2 circuit breaker. Same gate the auto-open path used to run: + # if recent losses tripped the sys2 breaker, block new adoptions for + # the lockout window. Otherwise the breaker would be useless under + # the manage-only model. + from app.services.circuit_breaker import is_tripped as _cb_tripped + sub_snapshot = { + "sys2_cb_tripped_at": sub.sys2_cb_tripped_at, + "sys2_cb_reason": sub.sys2_cb_reason, + } + tripped, cb_msg = _cb_tripped(sub_snapshot, "sys2") + if tripped: + raise AdoptionError("circuit_breaker", + f"Adopt blocked: {cb_msg}. Turning Auto-Trade ON or arming " + "Manual Window in Settings clears the breaker.") + + # Already managing this wallet+asset? + dup = (await db.execute( + select(BotTrade).where( + BotTrade.wallet_address == wallet_l, + BotTrade.asset == asset_u, + BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), + ) + )).scalar_one_or_none() + if dup is not None: + raise AdoptionError("already_adopted", + f"{asset_u} is already managed by the bot (trade #{dup.id}). " + "Send /release first if you want to re-adopt.") + + # Wallet-level concurrency cap. Same reasoning as the auto-open + # path: every System-2 asset is correlated crypto-beta and 3 + # simultaneous positions ≈ 1 leveraged macro bet, not 3 bets. + open_count = (await db.execute( + select(BotTrade).where( + BotTrade.wallet_address == wallet_l, + BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), + ) + )).scalars().all() + if len(open_count) >= SYS2_MAX_CONCURRENT: + raise AdoptionError("concurrency_cap", + f"You already have {len(open_count)} managed positions " + f"(cap is {SYS2_MAX_CONCURRENT}). /release one first.") + + try: + api_key = decrypt_api_key(sub.hl_api_key) + except Exception as exc: + raise AdoptionError("key_decrypt_failed", + f"Could not decrypt the HL API key: {exc}") + + trader = HyperliquidTrader( + api_private_key=api_key, + account_address=wallet_l, + leverage=int(sub.leverage or 2), + mainnet=settings.hl_mainnet, + ) + try: + raw = await trader.get_open_positions() + except Exception as exc: + raise AdoptionError("hl_read_failed", + f"Could not read HL positions: {exc}") + + target = next((p for p in raw if (p.get("coin") or "").upper() == asset_u), + None) + if target is None: + raise AdoptionError("not_on_hl", + f"No open {asset_u} position found on Hyperliquid. " + "Make sure you've filled the order, then try again.") + + szi = float(target.get("szi") or 0.0) + entry = float(target.get("entry_px") or 0.0) + if szi == 0 or entry <= 0: + raise AdoptionError("bad_hl_state", + f"HL returned a zero-size or zero-price {asset_u} position. " + "Refresh and try again in a moment.") + + side = "long" if szi > 0 else "short" + size_coins = abs(szi) + size_usd = round(size_coins * entry, 2) + lev_info = target.get("leverage") or {} + leverage = int(lev_info.get("value") or sub.leverage or 2) + + # Build the System-2 exit profile against the ACTUAL HL leverage so the + # protective stop is always inside the real liquidation line. Identical + # math to bot_engine's old sys2 open path; reused so behaviour stays + # consistent across the two entry points. + exit_profile = get_exit_profile(ADOPTED_CATEGORY) + prot_stop = sys2_protective_stop_pct(leverage) + derisk = sys2_derisk_ladder(leverage, mode_n) + addon = sys2_addon_ladder(mode_n) + peak_trail = sys2_peak_trail(mode_n) + stop_ladder = get_stop_ladder(ADOPTED_CATEGORY) + + now_naive = datetime.now(timezone.utc).replace(tzinfo=None) + async with AsyncSessionLocal() as db: + trade = BotTrade( + asset=asset_u, + side=side, + entry_price=entry, + wallet_address=wallet_l, + trigger_post_id=None, # adopted, not signal-triggered + opened_at=now_naive, + hl_order_id=f"adopted:{int(now_naive.timestamp())}", + size_usd=size_usd, + base_size_usd=size_usd, + sys2_mode=mode_n, + leverage=leverage, + # Freeze the System-2 exit profile on the row. No TP / no + # trailing-from-peak — pure staged stop + de-risk + peak-trail. + eff_take_profit_pct = None, + eff_stop_loss_pct = prot_stop, + eff_trailing_stop_pct = None, + eff_trailing_activate_pct = None, + eff_max_hold_hours = exit_profile.max_hold_hours, + eff_invalidation = None, + eff_invalidation_price = None, + eff_min_hold_until_ts = None, + ) + db.add(trade) + await db.commit() + await db.refresh(trade) + trade_id = trade.id + + # Register with the watchdog AFTER the row is committed so a tick fired + # in the next millisecond can find the trade on disk if it needs to. + from app.services.tp_sl_monitor import register_trade + register_trade( + trade_id=trade_id, + wallet=wallet_l, + api_key=api_key, + leverage=leverage, + asset=asset_u, + side=side, + entry_price=entry, + take_profit_pct=None, + stop_loss_pct=prot_stop, + trailing_stop_pct=None, + trailing_activate_at_pct=None, + invalidation=None, + invalidation_price=None, + min_hold_until_ts=None, + stop_ladder=stop_ladder, + derisk_ladder=derisk, + derisk_done=0, + addon_ladder=addon, + addon_done=0, + peak_trail=peak_trail, + grow_mode=False, # user opts a position into Grow explicitly + ) + + logger.warning( + "ADOPTED trade %d: wallet=%s asset=%s side=%s entry=%.4f size=%.2f " + "lev=%dx mode=%s prot_stop=%.2f%% derisk=%s", + trade_id, wallet_l, asset_u, side, entry, size_usd, + leverage, mode_n, prot_stop, derisk, + ) + + return AdoptionResult( + trade_id=trade_id, asset=asset_u, side=side, entry_price=entry, + size_usd=size_usd, leverage=leverage, mode=mode_n, + protective_stop=prot_stop, derisk_ladder=derisk, + stop_ladder=stop_ladder or [], + ) + + +# ── Release ───────────────────────────────────────────────────────────────── + + +async def release_management(wallet: str, trade_id: int) -> dict: + """Stop managing trade_id without closing the HL position. Marks + released_at, unregisters the watchdog. Idempotent — re-releasing a + released trade is a no-op.""" + wallet_l = wallet.lower().strip() + async with AsyncSessionLocal() as db: + trade = (await db.execute( + select(BotTrade).where(BotTrade.id == trade_id) + )).scalar_one_or_none() + if trade is None: + raise AdoptionError("not_found", f"trade {trade_id} not found") + if trade.wallet_address.lower() != wallet_l: + raise AdoptionError("not_owner", + f"trade {trade_id} belongs to a different wallet") + if trade.closed_at is not None: + raise AdoptionError("already_closed", + f"trade {trade_id} is already closed — nothing to release") + if trade.released_at is not None: + # Idempotent — return success. + return {"status": "ok", "trade_id": trade_id, + "asset": trade.asset, "already_released": True} + + now_naive = datetime.now(timezone.utc).replace(tzinfo=None) + await db.execute( + update(BotTrade) + .where(BotTrade.id == trade_id) + .values(released_at=now_naive) + ) + await db.commit() + + from app.services.tp_sl_monitor import unregister + unregister(trade_id) + + logger.warning( + "RELEASED trade %d: wallet=%s asset=%s — bot stops managing, " + "HL position untouched.", + trade_id, wallet_l, trade.asset, + ) + return {"status": "ok", "trade_id": trade_id, + "asset": trade.asset, "already_released": False} diff --git a/app/services/bot_engine.py b/app/services/bot_engine.py index 40f3cff..e6c37b0 100644 --- a/app/services/bot_engine.py +++ b/app/services/bot_engine.py @@ -117,29 +117,31 @@ async def process_post(post: Post, db: AsyncSession) -> None: sys2 = is_system_2(post.source) if sys2: - # SYSTEM 2 — reversal/breakout. The Trump-tuned regime filter - # (recent-move ≤5%, vol-contraction) actively REJECTS valid reversal - # setups, so it's bypassed entirely. The signal's structural gates - # (RSI<30 ×4wk, 200d reclaim, funding extreme) ARE the regime check. - exit_profile = get_exit_profile(post.category) - if exit_profile.stop_ladder: - logger.info( - "%s [%s/%s]: bypassing Trump regime filter. " - "Exit = STAGED stop (no TP/trailing) base sl=%.1f%% " - "ladder=%s maxhold=%dh", - system2_display_name(), post.source, post.category, - exit_profile.stop_loss_pct, exit_profile.stop_ladder, - exit_profile.max_hold_hours, - ) - else: - logger.info( - "%s [%s/%s]: bypassing Trump regime filter. " - "Exit profile sl=%.1f%% trail=%s@%s maxhold=%dh inval=%s", - system2_display_name(), post.source, post.category, - exit_profile.stop_loss_pct, exit_profile.trailing_stop_pct, - exit_profile.trailing_activate_at_pct, exit_profile.max_hold_hours, - exit_profile.invalidation, - ) + # SYSTEM 2 — manual-open + auto-manage model. + # + # We do NOT auto-open System-2 positions any more. The strategy is + # day-K level (AHR999 / 200WMA / Pi Cycle Bottom take weeks to play + # out) so a 24h entry delay is irrelevant — but the auto-open path + # carried real risk: leverage-clip math, daily-budget split, + # concurrency caps, sys2-CB pre-open gate, paper branches, key + # handling. All of that is execution surface for ~0 alpha vs giving + # the user the alert + letting them open on HL themselves. + # + # The valuable part of System-2 — the multi-month exit management + # (5-rung stop ladder + downside de-risk + pyramid + peak-trail) — + # is preserved in tp_sl_monitor and runs against positions the user + # adopts via /api/positions/adopt (or the Telegram /adopt command). + # + # Here we just record the signal (already done by signals.ingest → + # Post row) and let the Telegram fan-out send the alert with the + # "open on HL and /adopt" CTA. No subscriber iteration, no open. + logger.info( + "%s [%s/%s] signal recorded — manual-open + /adopt model " + "(no auto-open). Conf=%d", + system2_display_name(), post.source, post.category, + post.ai_confidence or 0, + ) + return else: # SYSTEM 1 — Trump. REPOSITIONED: low-frequency, selective, tight # stop, ≥30-min holds. See signal_categories.py. diff --git a/app/services/reconciler.py b/app/services/reconciler.py index 3003fb8..6aee38e 100644 --- a/app/services/reconciler.py +++ b/app/services/reconciler.py @@ -87,11 +87,14 @@ async def _reconcile_wallet(sub: Subscription) -> dict: hl_assets_open = {p["coin"]: p for p in hl_positions} async with AsyncSessionLocal() as db: - # All DB rows we believe are still open for this wallet. + # All DB rows we believe are still open AND not released for this + # wallet. A released trade left the HL position open but handed + # control back to the user — reconciler must not touch it. rows = await db.execute( select(BotTrade).where( BotTrade.wallet_address == sub.wallet_address, BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), ) ) db_open_trades = rows.scalars().all() diff --git a/app/services/recovery.py b/app/services/recovery.py index 4afe0d3..cb0288c 100644 --- a/app/services/recovery.py +++ b/app/services/recovery.py @@ -29,7 +29,15 @@ async def rehydrate_open_trades() -> None: from app.services.tp_sl_monitor import register_trade async with AsyncSessionLocal() as db: - result = await db.execute(select(BotTrade).where(BotTrade.closed_at.is_(None))) + # Skip released trades: the user has taken back manual control of + # the HL position; re-registering would re-attach the watchdog and + # silently start managing again after a restart. + result = await db.execute( + select(BotTrade).where( + BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), + ) + ) open_trades = result.scalars().all() if not open_trades: diff --git a/app/services/telegram.py b/app/services/telegram.py index e0bfd31..d7c9c02 100644 --- a/app/services/telegram.py +++ b/app/services/telegram.py @@ -128,7 +128,21 @@ def format_post(post: Post) -> str: }.get(post.source, "/en") link = f'\n\n→ open in dashboard' - msg = f"{head}\n{sub}\n\n{body}{extra}{link}" + # System-2 (Macro Vibes bottom / funding reversal) is MANUAL-OPEN now: + # we no longer auto-trade these on behalf of subscribers. The signal is + # day-K level so an entry delay doesn't matter — but the alert needs + # to make the next user action obvious. + cta = "" + if post.source in ("btc_bottom_reversal", "funding_reversal"): + cta = ( + "\n\nWant bot to manage your trade?\n" + "1. Open the position on Hyperliquid (your size / leverage).\n" + "2. Reply /adopt here to hand it to the bot.\n" + "Bot will run the 5-rung stop ladder, de-risk, pyramid, peak-trail. " + "You can take it back any time with /release." + ) + + msg = f"{head}\n{sub}\n\n{body}{extra}{cta}{link}" return msg[:MAX_LEN] @@ -137,20 +151,33 @@ def format_post(post: Post) -> str: async def send_message(chat_id: int, text: str, *, parse_mode: str = "HTML", - disable_preview: bool = True) -> bool: + disable_preview: bool = True, + reply_markup: Optional[dict] = None) -> bool: """Single HTTP POST to Telegram Bot API. Returns True on 200, False on - any failure (caller decides whether to bump the failure counter).""" + any failure (caller decides whether to bump the failure counter). + + `reply_markup` is the standard Bot-API inline keyboard / reply keyboard + payload, e.g. + {"inline_keyboard": [ + [{"text": "Yes", "callback_data": "x:y"}], + [{"text": "No", "callback_data": "x:n"}], + ]} + Callback_data is limited to 64 bytes by Telegram; keep payloads short. + """ token = settings.telegram_bot_token if not token: return False url = TG_API.format(token=token, method="sendMessage") + payload: dict = { + "chat_id": chat_id, "text": text, + "parse_mode": parse_mode, + "disable_web_page_preview": disable_preview, + } + if reply_markup is not None: + payload["reply_markup"] = reply_markup try: async with httpx.AsyncClient(timeout=10) as client: - r = await client.post(url, json={ - "chat_id": chat_id, "text": text, - "parse_mode": parse_mode, - "disable_web_page_preview": disable_preview, - }) + r = await client.post(url, json=payload) if r.status_code != 200: logger.warning("Telegram sendMessage failed chat=%d status=%d body=%s", chat_id, r.status_code, r.text[:200]) @@ -161,6 +188,59 @@ async def send_message(chat_id: int, text: str, *, return False +async def edit_message(chat_id: int, message_id: int, text: str, *, + parse_mode: str = "HTML", + reply_markup: Optional[dict] = None) -> bool: + """Edit an existing message (typically to swap out an inline keyboard + after the user pressed a button). Returns True on 200.""" + token = settings.telegram_bot_token + if not token: + return False + url = TG_API.format(token=token, method="editMessageText") + payload: dict = { + "chat_id": chat_id, "message_id": message_id, "text": text, + "parse_mode": parse_mode, + } + if reply_markup is not None: + payload["reply_markup"] = reply_markup + try: + async with httpx.AsyncClient(timeout=10) as client: + r = await client.post(url, json=payload) + if r.status_code != 200: + # Telegram returns 400 on "message is not modified" — harmless. + if "message is not modified" not in r.text: + logger.warning("Telegram editMessageText failed chat=%d msg=%d " + "status=%d body=%s", + chat_id, message_id, r.status_code, r.text[:200]) + return False + return True + except Exception as exc: + logger.warning("Telegram editMessageText exception: %s", exc) + return False + + +async def answer_callback(callback_query_id: str, text: str = "", + show_alert: bool = False) -> bool: + """Acknowledge an inline-button press. MUST be called within 30s of the + callback firing or the button stays in 'loading' state on the user's + client. Empty text → silent ack.""" + token = settings.telegram_bot_token + if not token: + return False + url = TG_API.format(token=token, method="answerCallbackQuery") + payload = {"callback_query_id": callback_query_id} + if text: + payload["text"] = text + payload["show_alert"] = show_alert + try: + async with httpx.AsyncClient(timeout=10) as client: + await client.post(url, json=payload) + return True + except Exception as exc: + logger.debug("Telegram answerCallback exception: %s", exc) + return False + + # ── Dispatcher ──────────────────────────────────────────────────────────── diff --git a/app/services/telegram_bot.py b/app/services/telegram_bot.py index 3c4b90f..c62ff8d 100644 --- a/app/services/telegram_bot.py +++ b/app/services/telegram_bot.py @@ -86,21 +86,29 @@ def _consume_code(code: str) -> Optional[str]: HELP_TEXT = ( "👋 Trump Alpha bot\n\n" - "Push alerts for high-conviction crypto signals — Trump posts, BTC bottom " - "reversals, funding extremes, KOL divergence.\n\n" + "Push alerts for high-conviction crypto signals — Trump posts, " + "Macro Vibes bottom signal, funding extremes, KOL divergence. " + "Plus a once-a-day 3-section brief.\n\n" "Just press /start — you're subscribed. No wallet needed.\n\n" - "Configure:\n" + "Real-time alerts:\n" " /trump on|off Trump · Truth Social posts\n" - " /btc on|off BTC · macro bottom\n" - " /funding on|off BTC · funding reversal\n" + " /btc on|off Macro Vibes · bottom signal\n" + " /funding on|off Macro Vibes · funding reversal\n" " /kol on|off KOL · talks vs trades\n" " /conf 0-100 min AI confidence (Trump only)\n" " /quiet 23 7 mute hours UTC (or 'off' to disable)\n\n" + "Daily brief:\n" + " /digest send me a preview right now\n" + " /digest on|off toggle the daily auto-send\n" + " /digest_time HH set the UTC hour (default 12)\n\n" "Status & control:\n" " /status — show your preferences\n" " /stop — pause all alerts\n" " /test — send a sample alert\n\n" - "Pro features (auto-trade on Hyperliquid):\n" + "Manage HL positions (Pro):\n" + " /adopt hand an HL position to the bot\n" + " /release stop managing (HL position stays open)\n\n" + "Pro features (manage + auto-trade on Hyperliquid):\n" "Open the dashboard → Settings → Connect Telegram to link " "your wallet." ) @@ -114,6 +122,10 @@ _DEFAULT_PREFS = { "alert_funding": True, "alert_kol_divergence": False, "min_confidence": 70, + # Daily digest defaults — on at 12 UTC (Asia evening / EU afternoon / + # US morning; only single hour reasonable for all three zones). + "digest_enabled": True, + "digest_hour_utc": 12, } @@ -398,19 +410,266 @@ async def _cmd_status(chat_id: int) -> None: mute = "" if b.mute_from_hour is not None and b.mute_until_hour is not None: mute = f"\n🌙 Quiet hours: {b.mute_from_hour:02d}–{b.mute_until_hour:02d} UTC" + digest_state = ( + f"🟢 ON @ {b.digest_hour_utc:02d}:00 UTC" + if b.digest_enabled else "🔴 OFF" + ) await send_message( chat_id, f"📡 Status\n\n" f"{tier_line}\n" f"Alerts: {on}\n" f"Sources: {src_line}\n" - f"Min confidence: {b.min_confidence}{mute}\n\n" + f"Min confidence: {b.min_confidence}{mute}\n" + f"Daily brief: {digest_state}\n\n" f"Sent: {b.total_alerts_sent} · Failed: {b.total_alerts_failed}\n\n" - f"Toggle anything with /trump /btc /funding /kol /conf /quiet — " - f"send /help for the full list.", + f"Toggle anything with /trump /btc /funding /kol /conf /quiet " + f"/digest /digest_time — send /help for the full list.", ) +async def _cmd_digest(chat_id: int, arg: str) -> None: + """Three dispatches in one command: + /digest → send a preview RIGHT NOW (manual fetch) + /digest on|off → toggle daily auto-send + Anything else → usage hint + Time changes go through /digest_time HH (separate command — easier to + parse and discover than overloading /digest with a numeric arg).""" + arg_l = arg.strip().lower() + if not arg_l: + from app.services.telegram_digest import send_preview_for + ok = await send_preview_for(chat_id) + if not ok: + await send_message(chat_id, + "Couldn't render a preview right now. Send /start first if " + "you haven't, otherwise try again in a minute.") + return + v = _parse_on_off(arg_l) + if v is None: + await send_message(chat_id, + "Usage:\n" + " /digest — preview right now\n" + " /digest on or /digest off\n" + " /digest_time 14 — set send hour (UTC)") + return + ok = await _set_pref(chat_id, "digest_enabled", v) + if not ok: + await send_message(chat_id, "No binding here. Send /start first.") + return + state = "🟢 ON" if v else "🔴 OFF" + await send_message(chat_id, f"{state} · daily brief") + + +async def _cmd_digest_time(chat_id: int, arg: str) -> None: + try: + h = int(arg.strip()) + if not 0 <= h <= 23: + raise ValueError + except ValueError: + await send_message(chat_id, + "Usage: /digest_time 14 (UTC hour, 0–23).\n" + "Example: /digest_time 0 = midnight UTC.") + return + ok = await _set_pref(chat_id, "digest_hour_utc", h) + if not ok: + await send_message(chat_id, "No binding here. Send /start first.") + return + await send_message(chat_id, + f"⏰ Daily brief now lands at {h:02d}:00 UTC.") + + +async def _wallet_for_chat(chat_id: int) -> Optional[str]: + """Return the bound wallet address for this chat_id, or None if the + chat hasn't linked one yet (free tier).""" + async with async_session() as db: + b = (await db.execute( + select(TelegramBinding).where(TelegramBinding.chat_id == chat_id) + )).scalar_one_or_none() + return b.wallet_address if (b and b.wallet_address) else None + + +async def _cmd_adopt(chat_id: int, arg: str) -> None: + """List the user's HL open positions and let them pick one to hand to + the bot. Pro-only — needs a wallet bound (we read HL state via the + subscription's API key). + + /adopt — show picker with inline keyboard + /adopt BTC — skip the picker, jump straight to mode pick for BTC + """ + wallet = await _wallet_for_chat(chat_id) + if not wallet: + await send_message( + chat_id, + "⚠️ /adopt is Pro-only.\n" + "It needs to read your Hyperliquid open positions, so the " + "Telegram chat must be linked to a wallet first.\n\n" + "Open the dashboard → Settings → Connect Telegram → paste the " + "code with /start CODE.", + ) + return + + from app.services.adoption import list_hl_positions, AdoptionError + try: + positions = await list_hl_positions(wallet) + except AdoptionError as exc: + await send_message(chat_id, f"❌ {exc.message}") + return + + if not positions: + await send_message( + chat_id, + "🟦 You have no open positions on Hyperliquid right now.\n\n" + "Open the trade on HL first, then come back and send /adopt.", + ) + return + + arg_u = arg.strip().upper() + if arg_u: + # Jump straight to mode picker for the requested asset. + match = next((p for p in positions if p.asset == arg_u), None) + if not match: + await send_message( + chat_id, + f"❌ No open {arg_u} position on Hyperliquid. " + f"Send /adopt with no args to see the list.", + ) + return + await _send_mode_picker(chat_id, match) + return + + # Build the asset picker. + rows: list[list[dict]] = [] + for p in positions: + if p.already_adopted: + label = f"✅ {p.asset} {p.side} ${p.size_usd:.0f} (already managed)" + # Disabled-looking row — still tap-able so they get feedback. + rows.append([{"text": label, + "callback_data": f"adopt:dup:{p.asset}"}]) + else: + arrow = "🟢" if p.side == "long" else "🔴" + label = (f"{arrow} {p.asset} {p.side} " + f"${p.size_usd:.0f} @ {p.entry_price:.2f} · {p.leverage}x") + rows.append([{"text": label, + "callback_data": f"adopt:pick:{p.asset}"}]) + rows.append([{"text": "✖ Cancel", "callback_data": "adopt:cancel"}]) + + await send_message( + chat_id, + "Pick a position to hand to the bot:\n\n" + "Bot will then run the 5-rung stop ladder + de-risk + pyramid + " + "peak-trail against it. You can release any time with " + "/release ID.", + reply_markup={"inline_keyboard": rows}, + ) + + +async def _send_mode_picker(chat_id: int, position) -> None: + """After an asset is picked, prompt for standard vs aggressive mode.""" + arrow = "🟢" if position.side == "long" else "🔴" + text = ( + f"{arrow} {position.asset} {position.side} " + f"${position.size_usd:.0f} @ {position.entry_price:.2f} · " + f"{position.leverage}x\n\n" + "Pick the risk mode the bot should use:\n\n" + "📈 Standard — conservative ladder, longer hold, slower pyramid. " + "Designed to survive bull-cycle corrections.\n" + "🚀 Aggressive — earlier + bigger pyramid adds, wider peak-trail. " + "Bigger reward on a clean multi-x, harder shakeouts.\n" + ) + rows = [ + [{"text": "📈 Standard", + "callback_data": f"adopt:mode:{position.asset}:standard"}], + [{"text": "🚀 Aggressive", + "callback_data": f"adopt:mode:{position.asset}:aggressive"}], + [{"text": "✖ Cancel", "callback_data": "adopt:cancel"}], + ] + await send_message(chat_id, text, reply_markup={"inline_keyboard": rows}) + + +async def _cmd_release(chat_id: int, arg: str) -> None: + """Hand a managed trade back to the user. + + /release — show picker with all currently managed trades + /release 42 — release trade_id 42 directly + """ + wallet = await _wallet_for_chat(chat_id) + if not wallet: + await send_message( + chat_id, + "⚠️ /release is Pro-only — link a wallet first via " + "Settings → Connect Telegram.") + return + + from app.models import BotTrade + async with async_session() as db: + managed = (await db.execute( + select(BotTrade).where( + BotTrade.wallet_address == wallet, + BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), + ).order_by(BotTrade.opened_at.desc()) + )).scalars().all() + + if not managed: + await send_message( + chat_id, + "🟦 No positions currently under bot management. " + "Nothing to release.", + ) + return + + arg_s = arg.strip() + if arg_s: + try: + tid = int(arg_s) + except ValueError: + await send_message(chat_id, + "Usage: /release 42 (trade id) — or just " + "/release to pick from a list.") + return + target = next((t for t in managed if t.id == tid), None) + if not target: + await send_message(chat_id, + f"❌ Trade {tid} isn't under bot management for this wallet.") + return + await _confirm_release(chat_id, target) + return + + # Picker. + rows = [] + for t in managed: + arrow = "🟢" if t.side == "long" else "🔴" + label = (f"{arrow} #{t.id} {t.asset} {t.side} " + f"${(t.size_usd or 0):.0f} @ {t.entry_price:.2f}") + rows.append([{"text": label, + "callback_data": f"release:pick:{t.id}"}]) + rows.append([{"text": "✖ Cancel", "callback_data": "release:cancel"}]) + + await send_message( + chat_id, + "Release a position from bot management:\n\n" + "The HL position stays open — bot just stops driving stops / " + "de-risk / pyramid on it.", + reply_markup={"inline_keyboard": rows}, + ) + + +async def _confirm_release(chat_id: int, trade) -> None: + arrow = "🟢" if trade.side == "long" else "🔴" + text = ( + f"{arrow} #{trade.id} {trade.asset} {trade.side} " + f"${(trade.size_usd or 0):.0f} @ {trade.entry_price:.2f}\n\n" + "Release this trade from bot management?\n" + "The HL position will stay open — you take back manual control." + ) + rows = [ + [{"text": "✅ Yes, release", + "callback_data": f"release:yes:{trade.id}"}], + [{"text": "✖ Cancel", "callback_data": "release:cancel"}], + ] + await send_message(chat_id, text, reply_markup={"inline_keyboard": rows}) + + async def _cmd_test(chat_id: int) -> None: async with async_session() as db: b = (await db.execute( @@ -477,6 +736,16 @@ async def _handle_message(msg: dict) -> None: await _cmd_conf(chat_id, arg) elif cmd == "/quiet": await _cmd_quiet(chat_id, arg) + # ── daily brief ────────────────────────────────────────────── + elif cmd == "/digest": + await _cmd_digest(chat_id, arg) + elif cmd in ("/digest_time", "/digesttime"): + await _cmd_digest_time(chat_id, arg) + # ── manage-only flow (System-2 reversal) ───────────────────── + elif cmd == "/adopt": + await _cmd_adopt(chat_id, arg) + elif cmd == "/release": + await _cmd_release(chat_id, arg) else: if cmd.startswith("/"): await send_message(chat_id, "Unknown command. Send /help for the list.") @@ -484,6 +753,202 @@ async def _handle_message(msg: dict) -> None: logger.exception("Telegram handler failed for chat=%s text=%r", chat_id, text[:80]) +async def _handle_callback(cb: dict) -> None: + """Route inline-button presses. Callback data is a tiny colon-delimited + payload (max 64 bytes per Telegram). All branches MUST end with an + answerCallbackQuery — otherwise the user's button spins forever.""" + from app.services.telegram import answer_callback, edit_message + from app.services.adoption import ( + list_hl_positions, adopt_position, release_management, AdoptionError, + ) + + cb_id = cb.get("id") + data = cb.get("data") or "" + msg = cb.get("message") or {} + chat = msg.get("chat") or {} + chat_id = chat.get("id") + msg_id = msg.get("message_id") + if not (cb_id and chat_id and msg_id): + return + + parts = data.split(":") + domain = parts[0] if parts else "" + + try: + # ── ADOPT flow ────────────────────────────────────────────────── + if domain == "adopt": + sub = parts[1] if len(parts) > 1 else "" + if sub == "cancel": + await edit_message(chat_id, msg_id, "✖ Adopt cancelled.") + await answer_callback(cb_id) + return + if sub == "dup": + await answer_callback(cb_id, + "This position is already managed.", + show_alert=True) + return + if sub == "pick" and len(parts) == 3: + asset = parts[2] + wallet = await _wallet_for_chat(chat_id) + if not wallet: + await answer_callback(cb_id, + "Wallet not bound.", show_alert=True) + return + try: + positions = await list_hl_positions(wallet) + except AdoptionError as exc: + await edit_message(chat_id, msg_id, f"❌ {exc.message}") + await answer_callback(cb_id) + return + target = next((p for p in positions if p.asset == asset), None) + if not target: + await edit_message(chat_id, msg_id, + f"❌ No open {asset} on HL anymore — refresh /adopt.") + await answer_callback(cb_id) + return + # Replace the picker with the mode picker in place. + arrow = "🟢" if target.side == "long" else "🔴" + text = ( + f"{arrow} {target.asset} {target.side} " + f"${target.size_usd:.0f} @ {target.entry_price:.2f} · " + f"{target.leverage}x\n\n" + "Pick the risk mode:\n\n" + "📈 Standard — conservative, designed to survive " + "bull-cycle corrections.\n" + "🚀 Aggressive — earlier + bigger pyramid adds, " + "wider peak-trail. Higher reward, harder shakeouts.\n" + ) + rows = [ + [{"text": "📈 Standard", + "callback_data": f"adopt:mode:{asset}:standard"}], + [{"text": "🚀 Aggressive", + "callback_data": f"adopt:mode:{asset}:aggressive"}], + [{"text": "✖ Cancel", "callback_data": "adopt:cancel"}], + ] + await edit_message(chat_id, msg_id, text, + reply_markup={"inline_keyboard": rows}) + await answer_callback(cb_id) + return + if sub == "mode" and len(parts) == 4: + asset, mode = parts[2], parts[3] + wallet = await _wallet_for_chat(chat_id) + if not wallet: + await answer_callback(cb_id, + "Wallet not bound.", show_alert=True) + return + try: + result = await adopt_position(wallet, asset, mode) + except AdoptionError as exc: + await edit_message(chat_id, msg_id, + f"❌ {exc.message}") + await answer_callback(cb_id, exc.code, show_alert=False) + return + + # Render success summary — mention the de-risk + ladder so + # the user understands what the bot will actually DO. + derisk_summary = " → ".join( + f"{thr:.0f}% close {int(frac * 100)}%" + for thr, frac, _ in (result.derisk_ladder or []) + ) + ladder_summary = " → ".join( + f"peak {trig:.0f}% lock {floor:+.0f}%" + for trig, floor in (result.stop_ladder or []) + ) + text = ( + f"✅ Adopted #{result.trade_id} · {result.asset} " + f"{result.side} · {result.leverage}x · " + f"{result.mode}\n\n" + f"Entry: {result.entry_price:.2f} · " + f"Size: ${result.size_usd:.0f}\n" + f"Protective stop: -{result.protective_stop:.1f}% " + f"(inside HL liquidation)\n\n" + f"De-risk ladder (underwater):\n{derisk_summary}\n\n" + f"Stop ratchet (in profit):\n{ladder_summary}\n\n" + f"Release any time: /release {result.trade_id}" + ) + await edit_message(chat_id, msg_id, text) + await answer_callback(cb_id, "Adopted ✓") + return + + # ── RELEASE flow ──────────────────────────────────────────────── + if domain == "release": + sub = parts[1] if len(parts) > 1 else "" + if sub == "cancel": + await edit_message(chat_id, msg_id, "✖ Release cancelled.") + await answer_callback(cb_id) + return + if sub == "pick" and len(parts) == 3: + try: + tid = int(parts[2]) + except ValueError: + await answer_callback(cb_id, "Bad trade id", + show_alert=True) + return + # Refetch + render the confirm sheet in place. + from app.models import BotTrade + async with async_session() as db: + t = (await db.execute( + select(BotTrade).where(BotTrade.id == tid) + )).scalar_one_or_none() + if t is None or t.closed_at is not None or t.released_at is not None: + await edit_message(chat_id, msg_id, + f"❌ Trade {tid} no longer managed.") + await answer_callback(cb_id) + return + arrow = "🟢" if t.side == "long" else "🔴" + text = ( + f"{arrow} #{t.id} {t.asset} {t.side} " + f"${(t.size_usd or 0):.0f} @ {t.entry_price:.2f}\n\n" + "Release this trade from bot management?\n" + "The HL position will stay open — you take back manual " + "control." + ) + rows = [ + [{"text": "✅ Yes, release", + "callback_data": f"release:yes:{tid}"}], + [{"text": "✖ Cancel", "callback_data": "release:cancel"}], + ] + await edit_message(chat_id, msg_id, text, + reply_markup={"inline_keyboard": rows}) + await answer_callback(cb_id) + return + if sub == "yes" and len(parts) == 3: + try: + tid = int(parts[2]) + except ValueError: + await answer_callback(cb_id, "Bad trade id", + show_alert=True) + return + wallet = await _wallet_for_chat(chat_id) + if not wallet: + await answer_callback(cb_id, "Wallet not bound.", + show_alert=True) + return + try: + res = await release_management(wallet, tid) + except AdoptionError as exc: + await edit_message(chat_id, msg_id, + f"❌ {exc.message}") + await answer_callback(cb_id) + return + await edit_message(chat_id, msg_id, + f"✅ Released #{res['trade_id']} · {res['asset']}\n\n" + "Bot is no longer managing this position. " + "HL state is untouched — you have manual control.\n\n" + "Re-adopt any time with /adopt.") + await answer_callback(cb_id, "Released ✓") + return + + # Unknown callback — just ack so the button doesn't spin. + await answer_callback(cb_id) + except Exception: + logger.exception("Telegram callback handler failed for data=%r", data) + try: + await answer_callback(cb_id, "Internal error", show_alert=True) + except Exception: + pass + + async def run_bot_loop() -> None: """Long-poll forever. Idempotent on offsets — Telegram serves the same update twice only if we don't acknowledge it. Sleep on errors to avoid @@ -517,6 +982,13 @@ async def run_bot_loop() -> None: msg = upd.get("message") or upd.get("edited_message") if msg: await _handle_message(msg) + continue + cb = upd.get("callback_query") + if cb: + # Inline-keyboard button press (/adopt and /release + # picker flows). Acknowledged inside the handler so the + # button never spins forever on the user's client. + await _handle_callback(cb) except asyncio.CancelledError: logger.info("Telegram bot loop cancelled.") raise diff --git a/app/services/telegram_digest.py b/app/services/telegram_digest.py new file mode 100644 index 0000000..f94a5ae --- /dev/null +++ b/app/services/telegram_digest.py @@ -0,0 +1,472 @@ +"""Daily 3-section Telegram digest — Macro Vibes / KOL / Trump. + +Sent to every TelegramBinding where digest_enabled=True at their chosen +digest_hour_utc. Independent of the per-signal real-time alert path +(telegram._dispatch) — a user can turn one off and the other on. + +Architecture: + + cron (hourly :00) + ↓ + send_daily_digest(current_hour) + ↓ + build_global_digest() ← runs ONCE, content is global + ↓ + for each binding where digest_hour_utc == current_hour: + format_digest(global, user_state) ← appends per-user line + send_message(chat_id, text) + update last_digest_sent_at + +The body is rule-based (no LLM). Each section reads one or two structured +fields and picks a templated sentence. If we later want richer prose, the +single place to hook DeepSeek/Claude is at the bottom of +`format_digest` — everything else stays the same. + +Idempotency: the SELECT filters on + last_digest_sent_at IS NULL OR last_digest_sent_at < now - 23h +so a coalesced cron or worker restart can't double-send within a day. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Optional + +from sqlalchemy import select, update, func, or_, and_ + +from app.config import settings +from app.database import AsyncSessionLocal as async_session +from app.models import ( + TelegramBinding, MacroSnapshot, KolPost, KolDivergence, + Post, BotTrade, Subscription, +) +from app.services.telegram import send_message + +logger = logging.getLogger(__name__) + + +# ── Data model ──────────────────────────────────────────────────────────── + + +@dataclass +class MacroBlock: + available: bool + composite: Optional[float] + regime: Optional[str] # BULL / BULLISH / NEUTRAL / BEARISH / BEAR + ahr999: Optional[float] + fear_greed: Optional[int] + bottom_signal_firing: bool # any Post(source=btc_bottom_reversal, last 24h) + bottom_votes: Optional[int] # 2 or 3 if firing, else None + funding_signal_firing: bool # any Post(source=funding_reversal, last 24h) + + +@dataclass +class KolBlock: + posts_24h: int + bullish: int + bearish: int + neutral: int + divergences_24h: int + sample_divergence: Optional[str] # one human-readable line, or None + + +@dataclass +class TrumpBlock: + posts_24h: int + actionable: list # list of (asset, side, confidence) + + +@dataclass +class GlobalDigest: + """Same content for everyone. Per-user state (auto-trade, your trades) + is appended in format_digest.""" + date_label: str # "May 26 · Mon" + macro: MacroBlock + kol: KolBlock + trump: TrumpBlock + headline_emoji: str # 🟢 / 🟡 / ⚪ + headline_text: str # "BTC bottom signal firing. Worth a look." + + +# ── Builder ─────────────────────────────────────────────────────────────── + + +async def build_global_digest(now: Optional[datetime] = None) -> GlobalDigest: + """Single DB pass over the three sources. Pure read — never raises; + returns a degraded digest if any source is missing.""" + now = now or datetime.now(timezone.utc) + now_naive = now.replace(tzinfo=None) + cutoff_24h = now_naive - timedelta(hours=24) + date_label = now.strftime("%b %-d · %a") + + async with async_session() as db: + # ── Macro: today's row (or yesterday's if today's not in yet) ──── + today = now.date() + yesterday = today - timedelta(days=1) + snap = (await db.execute( + select(MacroSnapshot) + .where(MacroSnapshot.snapshot_date.in_([today, yesterday])) + .order_by(MacroSnapshot.snapshot_date.desc()) + )).scalars().first() + + # Bottom + funding triggers in last 24h + bottom_post = (await db.execute( + select(Post) + .where(Post.source == "btc_bottom_reversal", + Post.created_at >= cutoff_24h) + .order_by(Post.created_at.desc()) + )).scalars().first() + funding_post = (await db.execute( + select(Post) + .where(Post.source == "funding_reversal", + Post.created_at >= cutoff_24h) + )).scalars().first() + + macro = MacroBlock( + available=snap is not None, + composite=snap.composite_score if snap else None, + regime=snap.regime_label if snap else None, + ahr999=snap.ahr999 if snap else None, + fear_greed=snap.fear_greed if snap else None, + bottom_signal_firing=bottom_post is not None, + bottom_votes=(bottom_post.ai_confidence and + (3 if bottom_post.ai_confidence >= 90 else 2)) + if bottom_post else None, + funding_signal_firing=funding_post is not None, + ) + + # ── KOL: last 24h posts + divergences ──────────────────────────── + kol_posts = (await db.execute( + select(KolPost).where(KolPost.published_at >= cutoff_24h) + )).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. + 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() + for d in divergence_rows: + if d.direction == "long": + bullish += 1 + elif d.direction == "short": + bearish += 1 + else: + neutral += 1 + # Posts not yet classified into divergences just count as neutral. + neutral += max(0, len(kol_posts) - (bullish + bearish + neutral)) + + divergences_24h = sum( + 1 for d in divergence_rows if d.signal_type == "divergence" + ) + sample_divergence: Optional[str] = None + for d in divergence_rows: + if d.signal_type != "divergence": + continue + # "Hayes publicly bullish BTC, on-chain selling" + sample_divergence = ( + f"{d.handle} publicly {d.post_action} {d.ticker}, " + f"on-chain {d.onchain_action}" + ) + break + + kol = KolBlock( + posts_24h=len(kol_posts), + bullish=bullish, bearish=bearish, neutral=neutral, + divergences_24h=divergences_24h, + sample_divergence=sample_divergence, + ) + + # ── Trump: last 24h posts, list actionable ones ────────────────── + trump_posts = (await db.execute( + select(Post) + .where(Post.source == "truth", Post.created_at >= cutoff_24h) + .order_by(Post.created_at.desc()) + )).scalars().all() + actionable = [] + for p in trump_posts: + if p.signal in ("buy", "short") and (p.ai_confidence or 0) >= 70: + actionable.append(( + p.target_asset or "?", + "LONG" if p.signal == "buy" else "SHORT", + int(p.ai_confidence or 0), + )) + trump = TrumpBlock(posts_24h=len(trump_posts), actionable=actionable) + + headline_emoji, headline_text = _headline(macro, kol, trump) + + return GlobalDigest( + date_label=date_label, + macro=macro, kol=kol, trump=trump, + headline_emoji=headline_emoji, headline_text=headline_text, + ) + + +def _headline(macro: MacroBlock, kol: KolBlock, trump: TrumpBlock) -> tuple[str, str]: + """Pick a one-liner verdict + emoji from the three sections. + + Priority: macro bottom > macro funding > Trump actionable > KOL + divergence > nothing. + """ + if macro.bottom_signal_firing: + return "🟢", "BTC bottom signal firing. Worth a look." + if macro.funding_signal_firing: + return "🟢", "Funding extreme — mean-reversion setup forming." + if len(trump.actionable) > 0: + n = len(trump.actionable) + return "🟢", f"Trump posted {n} actionable crypto signal{'s' if n > 1 else ''} today." + if kol.divergences_24h > 0: + return "🟡", f"{kol.divergences_24h} KOL talks-vs-trades divergence today." + return "⚪", "Quiet day, nothing to act on." + + +# ── Formatter ───────────────────────────────────────────────────────────── + + +def format_digest(g: GlobalDigest, user_state: Optional[dict] = None) -> str: + """Render the global digest into a single HTML message. + + `user_state` is a small dict built per-binding by send_daily_digest: + { "wallet": str | None, + "auto_trade": bool, + "trades_today": int, + "open_pnl_summary": str | None } + None or empty → free-tier user, the Pro status line is suppressed. + """ + lines: list[str] = [] + lines.append("📰 Trump Alpha · Daily Brief") + lines.append(f"{g.date_label}") + lines.append("") + lines.append(f"{g.headline_emoji} Today: {g.headline_text}") + lines.append("") + + # ── Macro ──────────────────────────────────────────────────────────── + lines.append("━ Macro ━") + m = g.macro + if not m.available: + lines.append("Macro snapshot unavailable today.") + elif m.bottom_signal_firing: + votes = m.bottom_votes or 2 + lines.append(f"BTC bottom triggers: {votes}/3 firing.") + if m.ahr999 is not None: + lines.append(f"AHR999={m.ahr999:.2f}, F&G={m.fear_greed or '?'}.") + elif m.funding_signal_firing: + lines.append("Funding-rate extreme — mean-reversion alert active.") + else: + # Plain day: one sentence + 1-2 backing numbers + regime_phrase = { + "BULL": "strongly bullish", + "BULLISH": "leaning bullish", + "NEUTRAL": "in neutral zone, far from bottom", + "BEARISH": "leaning bearish", + "BEAR": "strongly bearish", + }.get((m.regime or "NEUTRAL").upper(), "in neutral zone") + lines.append(f"BTC {regime_phrase}.") + bits = [] + if m.ahr999 is not None: + bits.append(f"AHR999={m.ahr999:.2f} (bottom <0.45)") + if m.fear_greed is not None: + bits.append(f"F&G={m.fear_greed}") + if bits: + lines.append(", ".join(bits) + ".") + lines.append("") + + # ── KOL ────────────────────────────────────────────────────────────── + lines.append("━ KOL ━") + k = g.kol + if k.posts_24h == 0: + lines.append("No KOL posts in last 24h.") + else: + lines.append( + f"{k.posts_24h} posts today: {k.bullish} bullish / " + f"{k.bearish} bearish." + ) + if k.divergences_24h > 0 and k.sample_divergence: + lines.append(f"⚠️ {k.sample_divergence}.") + elif k.divergences_24h == 0: + lines.append("No talks-vs-trades divergence.") + lines.append("") + + # ── Trump ──────────────────────────────────────────────────────────── + lines.append("━ Trump ━") + t = g.trump + if t.posts_24h == 0: + lines.append("No posts in last 24h.") + elif not t.actionable: + lines.append(f"{t.posts_24h} posts today, none crypto-actionable.") + else: + lines.append( + f"{t.posts_24h} posts today, " + f"{len(t.actionable)} actionable:" + ) + for asset, side, conf in t.actionable[:3]: + emoji = "🟢" if side == "LONG" else "🔴" + lines.append(f" {emoji} {asset} · {side} · conf {conf}") + if len(t.actionable) > 3: + lines.append(f" …and {len(t.actionable) - 3} more.") + lines.append("") + + # ── Per-user state (Pro only) ──────────────────────────────────────── + if user_state and user_state.get("wallet"): + at = "ON" if user_state.get("auto_trade") else "OFF" + td = int(user_state.get("trades_today") or 0) + bits = [f"auto-trade {at}", f"{td} trade{'s' if td != 1 else ''} today"] + if user_state.get("open_pnl_summary"): + bits.append(user_state["open_pnl_summary"]) + lines.append("Your status: " + " · ".join(bits)) + lines.append("") + + # ── Footer ─────────────────────────────────────────────────────────── + fe = (settings.frontend_url or "").rstrip("/") + if fe: + lines.append(f'→ open dashboard') + lines.append("") + lines.append("Daily brief · reply /digest off to unsubscribe") + + return "\n".join(lines).strip() + + +# ── Sender ──────────────────────────────────────────────────────────────── + + +async def _build_user_state(db, binding: TelegramBinding, + now_naive: datetime) -> dict: + """Pro-only: fetch auto-trade flag + today's trade count for the + wallet attached to this binding. Returns {} for free users.""" + if not binding.wallet_address: + return {} + sub = (await db.execute( + select(Subscription).where(Subscription.wallet_address == binding.wallet_address) + )).scalar_one_or_none() + if sub is None: + return {} + start_of_day = now_naive.replace(hour=0, minute=0, second=0, microsecond=0) + trades_today = (await db.execute( + select(func.count(BotTrade.id)).where( + BotTrade.wallet_address == binding.wallet_address, + BotTrade.opened_at >= start_of_day, + ) + )).scalar() or 0 + # Open positions' aggregated unrealised — best-effort, skip on any error. + open_pnl_summary: Optional[str] = None + try: + # Released trades aren't bot-managed any more; their unrealised + # PnL belongs to the user's manual control, not to "your bot status". + open_trades = (await db.execute( + select(BotTrade).where( + BotTrade.wallet_address == binding.wallet_address, + BotTrade.closed_at.is_(None), + BotTrade.released_at.is_(None), + ) + )).scalars().all() + if open_trades: + from app.services.price_store import price_store + unreal = 0.0 + for t in open_trades: + px = price_store.latest_price(t.asset) + if not px or not t.entry_price or not t.size_usd: + continue + pct = (px - t.entry_price) / t.entry_price + signed = pct if t.side == "long" else -pct + rem = t.remaining_fraction if t.remaining_fraction is not None else 1.0 + unreal += t.size_usd * rem * signed + if open_trades: + sign = "+" if unreal >= 0 else "" + open_pnl_summary = f"{len(open_trades)} open ({sign}{unreal:.1f} USD)" + except Exception as exc: + logger.debug("digest user_state: skip open pnl: %s", exc) + + return { + "wallet": binding.wallet_address, + "auto_trade": bool(sub.auto_trade), + "trades_today": int(trades_today), + "open_pnl_summary": open_pnl_summary, + } + + +async def send_daily_digest(current_hour: Optional[int] = None) -> int: + """Send the digest to every binding whose digest_hour_utc matches the + current UTC hour AND who hasn't received one in the last 23 hours. + + Returns the number of messages successfully sent. Never raises — the + cron firing must not bring the worker down. + """ + if not settings.telegram_bot_token: + return 0 + + now = datetime.now(timezone.utc) + hour = current_hour if current_hour is not None else now.hour + now_naive = now.replace(tzinfo=None) + dedupe_cutoff = now_naive - timedelta(hours=23) + + # Build the body ONCE — it's identical for every user (only per-user + # state line varies). One DB pass instead of one per binding. + try: + global_digest = await build_global_digest(now) + except Exception as exc: + logger.exception("Digest: build_global_digest failed: %s", exc) + return 0 + + sent = 0 + async with async_session() as db: + bindings = (await db.execute( + select(TelegramBinding).where( + TelegramBinding.digest_enabled.is_(True), + TelegramBinding.alerts_enabled.is_(True), + TelegramBinding.digest_hour_utc == hour, + or_( + TelegramBinding.last_digest_sent_at.is_(None), + TelegramBinding.last_digest_sent_at < dedupe_cutoff, + ), + ) + )).scalars().all() + + if not bindings: + return 0 + + for b in bindings: + try: + user_state = await _build_user_state(db, b, now_naive) + text = format_digest(global_digest, user_state) + ok = await send_message(b.chat_id, text) + if ok: + await db.execute( + update(TelegramBinding) + .where(TelegramBinding.id == b.id) + .values(last_digest_sent_at=now_naive) + ) + sent += 1 + except Exception as exc: + logger.warning("Digest send failed for chat=%s: %s", b.chat_id, exc) + await db.commit() + + logger.info("Daily digest: hour=%dz sent=%d/%d", hour, sent, len(bindings)) + return sent + + +# ── Manual preview (called from bot /digest command) ────────────────────── + + +async def send_preview_for(chat_id: int) -> bool: + """Render and immediately send a digest to one chat. Does NOT update + last_digest_sent_at — preview is independent of the daily slot.""" + if not settings.telegram_bot_token: + return False + try: + g = await build_global_digest() + except Exception as exc: + logger.exception("Digest preview: build failed: %s", exc) + return await send_message( + chat_id, "⚠️ Digest preview unavailable — check back shortly.") + async with async_session() as db: + b = (await db.execute( + select(TelegramBinding).where(TelegramBinding.chat_id == chat_id) + )).scalar_one_or_none() + user_state = (await _build_user_state(db, b, datetime.utcnow()) + if b else {}) + text = format_digest(g, user_state) + return await send_message(chat_id, text) diff --git a/tests/test_adoption.py b/tests/test_adoption.py new file mode 100644 index 0000000..2fafc25 --- /dev/null +++ b/tests/test_adoption.py @@ -0,0 +1,169 @@ +"""Validation-path tests for the adopt / release flow. + +These cover the in-memory branches that don't need a real DB / real HL — +mode normalisation, the error-code surface, and the structure of the +returned objects. End-to-end (real HL state → BotTrade row written → +watchdog registered) is verified by the manual /adopt Telegram flow, +not in CI. + +Why so light: every interesting failure case is a DB or HL state issue +that's better caught manually. The function shape + error codes are what +the API + Telegram layer switch on, so those we lock down here. +""" +import pytest + +from app.services.adoption import ( + AdoptionError, ADOPTED_CATEGORY, + HLPositionView, AdoptionResult, +) +from app.services.signal_categories import ( + sys2_normalize_mode, get_exit_profile, get_stop_ladder, + sys2_protective_stop_pct, sys2_derisk_ladder, sys2_addon_ladder, + sys2_peak_trail, SYS2_MODES, +) + + +def test_adopted_category_resolves_to_btc_bottom_profile(): + """The adopted-position category must point to a real exit profile — + otherwise adoption silently falls back to the System-2 default + (different stop / different max-hold).""" + profile = get_exit_profile(ADOPTED_CATEGORY) + assert profile.stop_ladder is not None, \ + "adopted category must use the staged stop ladder" + assert profile.max_hold_hours == 12960, \ + "BTC bottom-reversal is an 18-month-max-hold strategy" + # 5 upside rungs match the documented strategy. + ladder = get_stop_ladder(ADOPTED_CATEGORY) + assert len(ladder) == 5 + + +def test_adoption_error_carries_code_and_message(): + """API + Telegram layers switch on .code; ensure both attrs are set.""" + exc = AdoptionError("concurrency_cap", "You already have 3.") + assert exc.code == "concurrency_cap" + assert exc.message == "You already have 3." + assert "3" in str(exc) + + +def test_mode_normalisation_maps_unknown_to_standard(): + assert sys2_normalize_mode("standard") == "standard" + assert sys2_normalize_mode("aggressive") == "aggressive" + assert sys2_normalize_mode("STANDARD") == "standard" + assert sys2_normalize_mode("") == "standard" + assert sys2_normalize_mode(None) == "standard" + assert sys2_normalize_mode("yolo") == "standard" + + +def test_mode_set_matches_canonical_list(): + """SYS2_MODES is the source of truth for what adopt_position accepts; + if we add a third mode this test fails on purpose to remind us to + update the bot's mode picker buttons too.""" + assert SYS2_MODES == ("standard", "aggressive") + + +@pytest.mark.parametrize("lev,expected_max_prot", [ + (1, 35.0), # 1× → 85% of 100% liq = 85%, capped at 35% + (2, 35.0), # 2× → 85% of 50% = 42.5%, capped at 35% + (3, 28.33), # 3× → 85% of 33.3% ≈ 28.33 + (5, 17.0), # 5× → 85% of 20% = 17.0 + (10, 8.5), # 10× → 85% of 10% = 8.5 +]) +def test_protective_stop_scales_with_leverage(lev, expected_max_prot): + """The whole point of recomputing protective stop against HL's actual + leverage is that this number must stay INSIDE the liquidation distance. + A regression here is a real-money risk regression.""" + got = sys2_protective_stop_pct(lev) + assert got == pytest.approx(expected_max_prot, abs=0.1) + + +def test_derisk_ladder_uses_three_rungs_summing_to_protective(): + """At any leverage, the three de-risk rungs should land at + -0.60·P, -0.80·P, -1.00·P. The final rung IS the full protective + close — same level as a one-shot stop, but staged.""" + for lev in (2, 5, 10): + prot = sys2_protective_stop_pct(lev) + ladder = sys2_derisk_ladder(lev, "standard") + assert len(ladder) == 3 + thrs = [r[0] for r in ladder] + assert thrs[0] == pytest.approx(-0.60 * prot, abs=0.5) + assert thrs[1] == pytest.approx(-0.80 * prot, abs=0.5) + assert thrs[2] == pytest.approx(-1.00 * prot, abs=0.5) + # Last rung is the FULL close (is_final flag). + assert ladder[-1][2] is True + + +def test_addon_ladder_differs_between_modes(): + """Aggressive mode must fire earlier and add more — otherwise the mode + label is a lie.""" + std = sys2_addon_ladder("standard") + agg = sys2_addon_ladder("aggressive") + assert std[0][0] > agg[0][0], \ + "aggressive first rung must trigger at a LOWER peak gain" + assert agg[0][1] > std[0][1], \ + "aggressive first rung must add MORE notional" + + +def test_peak_trail_aggressive_more_permissive(): + """Aggressive peak-trail must let the trade run longer + give back + more drawdown — that's the explicit trade-off vs standard.""" + std_start, std_dd = sys2_peak_trail("standard") + agg_start, agg_dd = sys2_peak_trail("aggressive") + assert agg_start < std_start + assert agg_dd > std_dd + + +def test_hl_position_view_dataclass_shape(): + """The API layer rehydrates HLPositionView via **vars(it); guard the + field set so a silent rename doesn't break the HTTP shape.""" + v = HLPositionView(asset="BTC", side="long", size_coins=0.02, + entry_price=72000.0, leverage=2, size_usd=1440.0, + already_adopted=False) + d = vars(v) + assert set(d.keys()) == { + "asset", "side", "size_coins", "entry_price", + "leverage", "size_usd", "already_adopted", + } + + +def test_adoption_result_dataclass_shape(): + r = AdoptionResult( + trade_id=1, asset="BTC", side="long", entry_price=72000.0, + size_usd=1440.0, leverage=2, mode="standard", + protective_stop=35.0, derisk_ladder=[], stop_ladder=[], + ) + assert r.trade_id == 1 + assert r.mode == "standard" + + +def test_wallet_adopt_lock_is_per_wallet_and_reused(): + """The lock cache must hand back the SAME lock for the same wallet + (otherwise two concurrent adopts would acquire different locks and + bypass serialisation), and a DIFFERENT lock for a different wallet.""" + from app.services.adoption import _wallet_adopt_lock + a1 = _wallet_adopt_lock("0xaaa") + a2 = _wallet_adopt_lock("0xaaa") + b1 = _wallet_adopt_lock("0xbbb") + assert a1 is a2 + assert a1 is not b1 + + +def test_adoption_error_codes_used_by_callers_are_all_defined(): + """The API and Telegram layers switch on AdoptionError.code values. + If we ever silently rename one, the switch falls through to the + generic branch and the user loses the specific guidance. Lock the + set of codes we promise to emit.""" + # These are the codes referenced from app/api/positions.py + + # app/services/telegram_bot.py. Update both sites if you add one. + promised_codes = { + "no_subscription", "no_hl_key", "paper_mode", + "key_decrypt_failed", "hl_read_failed", + "bad_mode", "circuit_breaker", + "already_adopted", "concurrency_cap", + "not_on_hl", "bad_hl_state", + "not_found", "not_owner", "already_closed", + } + # Just exercising the constructor for each — proves the codes + # round-trip cleanly through __init__. + for code in promised_codes: + err = AdoptionError(code, f"sample for {code}") + assert err.code == code diff --git a/tests/test_telegram_digest.py b/tests/test_telegram_digest.py new file mode 100644 index 0000000..f6b5c5c --- /dev/null +++ b/tests/test_telegram_digest.py @@ -0,0 +1,176 @@ +"""Snapshot-style tests for the daily digest formatter. + +These hit format_digest with hand-rolled GlobalDigest objects so we exercise +every branch (quiet day / bottom trigger / actionable Trump / divergence / +empty all-around) without standing up a DB. The actual build_global_digest +SQL is exercised by integration tests / manual /digest in the bot. + +The point of these tests is to lock the user-facing wording — if a future +change accidentally turns "BTC bottom triggers: 3/3 firing" into a less +clear phrasing, CI catches it. +""" + +from app.services.telegram_digest import ( + GlobalDigest, MacroBlock, KolBlock, TrumpBlock, format_digest, +) + + +def _empty_blocks(): + return ( + MacroBlock(available=False, composite=None, regime=None, + ahr999=None, fear_greed=None, + bottom_signal_firing=False, bottom_votes=None, + funding_signal_firing=False), + KolBlock(posts_24h=0, bullish=0, bearish=0, neutral=0, + divergences_24h=0, sample_divergence=None), + TrumpBlock(posts_24h=0, actionable=[]), + ) + + +def test_quiet_day_renders_compact_and_mentions_no_activity(): + m, k, t = _empty_blocks() + m.available = True + m.regime = "NEUTRAL" + m.ahr999 = 0.62 + m.fear_greed = 38 + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="⚪", + headline_text="Quiet day, nothing to act on.") + out = format_digest(g) + assert "Trump Alpha · Daily Brief" in out + assert "⚪" in out + assert "Quiet day" in out + assert "neutral zone" in out + assert "AHR999=0.62" in out + assert "No KOL posts" in out + assert "No posts in last 24h" in out + # No per-user Pro line for free users + assert "Your status" not in out + # Always ends with the unsubscribe hint + assert "/digest off" in out + + +def test_bottom_trigger_day_promotes_macro_section_and_keeps_short(): + m, k, t = _empty_blocks() + m.available = True + m.bottom_signal_firing = True + m.bottom_votes = 3 + m.ahr999 = 0.41 + m.fear_greed = 14 + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="🟢", + headline_text="BTC bottom signal firing. Worth a look.") + out = format_digest(g) + assert "🟢" in out + assert "BTC bottom signal firing" in out + # Macro section shows the trigger detail + assert "3/3 firing" in out + assert "AHR999=0.41" in out + + +def test_trump_actionable_list_is_capped_at_three_with_summary_overflow(): + m, k, t = _empty_blocks() + m.available = True + m.regime = "NEUTRAL" + t.posts_24h = 6 + t.actionable = [ + ("BTC", "LONG", 92), + ("SOL", "LONG", 81), + ("ETH", "SHORT", 76), + ("DOGE", "LONG", 71), # 4th — should be summarised + ] + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="🟢", + headline_text="Trump posted 4 actionable crypto signals today.") + out = format_digest(g) + assert "BTC · LONG · conf 92" in out + assert "SOL · LONG · conf 81" in out + assert "ETH · SHORT · conf 76" in out + # The fourth must roll into the "and N more" line, not be listed verbatim + assert "DOGE" not in out + assert "1 more" in out + + +def test_divergence_warning_surfaces_sample_line(): + m, k, t = _empty_blocks() + m.available = True + m.regime = "NEUTRAL" + k.posts_24h = 14 + k.bullish = 9 + k.bearish = 1 + k.divergences_24h = 1 + k.sample_divergence = "ArthurHayes publicly bullish BTC, on-chain decreased" + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="🟡", + headline_text="1 KOL talks-vs-trades divergence today.") + out = format_digest(g) + assert "⚠️ ArthurHayes" in out + assert "9 bullish" in out + + +def test_pro_user_state_appended_for_wallet_bound(): + m, k, t = _empty_blocks() + m.available = True + m.regime = "NEUTRAL" + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="⚪", headline_text="Quiet day, nothing to act on.") + out = format_digest(g, user_state={ + "wallet": "0xabc", + "auto_trade": True, + "trades_today": 2, + "open_pnl_summary": "1 open (+12.4 USD)", + }) + assert "Your status" in out + assert "auto-trade ON" in out + assert "2 trades today" in out + assert "1 open (+12.4 USD)" in out + + +def test_pro_state_suppressed_for_free_user(): + m, k, t = _empty_blocks() + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="⚪", headline_text="Quiet day, nothing to act on.") + out_free = format_digest(g, user_state={}) + out_no_wallet = format_digest(g, user_state={"wallet": None}) + assert "Your status" not in out_free + assert "Your status" not in out_no_wallet + + +def test_html_safe_against_ampersand_in_label(): + """The 'F&G' label must always emit as F&G so Telegram's HTML + parser doesn't reject the message.""" + m, k, t = _empty_blocks() + m.available = True + m.regime = "NEUTRAL" + m.ahr999 = 0.62 + m.fear_greed = 38 + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="⚪", headline_text="Quiet day, nothing to act on.") + out = format_digest(g) + assert "F&G=38" in out + assert "F&G=38" not in out + + +def test_total_length_well_under_telegram_cap(): + """Even the busiest realistic day must fit in one Telegram message + (4096 chars).""" + m, k, t = _empty_blocks() + m.available = True + m.bottom_signal_firing = True + m.bottom_votes = 3 + m.ahr999 = 0.41 + m.fear_greed = 14 + k.posts_24h = 30 + k.bullish = 20 + k.bearish = 8 + k.divergences_24h = 4 + k.sample_divergence = "Hayes publicly bullish ETH, on-chain decreased" + t.posts_24h = 12 + t.actionable = [("BTC", "LONG", 95), ("SOL", "LONG", 88), ("ETH", "SHORT", 80)] + g = GlobalDigest(date_label="May 26 · Mon", macro=m, kol=k, trump=t, + headline_emoji="🟢", headline_text="Lots happening today.") + out = format_digest(g, user_state={ + "wallet": "0xabc", "auto_trade": True, "trades_today": 4, + "open_pnl_summary": "3 open (+125.3 USD)", + }) + assert len(out) < 4096