Pre-launch hardening: KOL module, Telegram, scanners, WS resilience

Big-picture changes since b941223:

KOL pipeline (new) — Substack/podcast/blog RSS → AI ticker extraction →
on-chain wallet diff → talks-vs-trades divergence detection. Daily polls,
19 feeds, divergence emits Post + Telegram fan-out.

Telegram push (new) — walletless free tier + wallet-linked Pro upgrade,
in-bot preference commands (/trump /btc /funding /kol /conf /quiet),
signed-envelope API for dashboard. Disconnect-wallet keeps free
subscription.

BTC funding-rate reversal scanner (new) — hourly cron, 30d cumulative
funding threshold + mean-revert + 7d price confirm, emits via
/api/signals/ingest. BTC bottom-reversal scanner promoted to System 2.

WS broadcast rewrite — per-client send timeout + parallel fan-out
(asyncio.gather). Fixes "Binance WS no close frame" reconnect storms +
APScheduler 11-min job misses, both caused by one slow client stalling
the kline loop.

Error visibility — three silent-error sites (trumpstruth/truth_social
fetchers, funding_reversal scanner) now include exception type name so
httpx ConnectError-style empty-message errors stop logging blank lines.
Telegram bot loop now classifies ReadTimeout vs network vs unknown +
logger.exception for the unknown bucket.

Security hygiene — trumpsignal.db untracked from git (held subscriber
wallets + encrypted HL keys + 22 bot trades); .gitignore now blocks
*.db/.next/backups. CORS only allows FRONTEND_URL in production.

New ops scripts —
  - scripts/preflight.py: env/DB/Telegram/AI auth verification gate
  - scripts/backup_db.sh: cron-friendly daily DB backup (SQLite + Postgres)
  - scripts/seed_kol_wallets.py: idempotent KOL on-chain wallet seeder

15 new Alembic migrations (007-021) covering convex strategy fields,
phase-1 safety, two-system frozen exits, invalidation prices, dynamic
SYS2 leverage, staged de-risk + pyramiding, peak gain tracking, risk
mode, auto-trade + grow flags, KOL module, KOL on-chain, KOL divergence,
Telegram bindings + walletless.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

app/services/bot_engine.py

@@ -6,7 +6,7 @@ Iterates all active subscribers and executes trades on their behalf.
 
 import asyncio
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
 from typing import Dict
 
 from sqlalchemy import select, update
@@ -23,7 +23,9 @@ logger = logging.getLogger(__name__)
 
 # Platform-wide thresholds (per-user values in Subscription override where applicable)
 # No global confidence floor — users pick their own 0–100 threshold via settings.
-MAX_HOLD_SECONDS = 3600          # force-close after 1 hour (shared across users)
+# Per-user max hold lives on Subscription.max_hold_hours (default 168h = 7 days
+# in the convex-strategy redesign). Old 1-hour cap killed every potential
+# runner before it could compound, so it's been removed.
 
 # Hyperliquid perp fees (mainnet, base tier).
 # IOC orders always cross the book → taker fee both on open and close.
@@ -63,6 +65,17 @@ def _lock_for(trade_id: int) -> asyncio.Lock:
     return lock
 
 
+def _confidence_floor_for(sub: dict) -> int:
+    if sub.get("_is_system_2"):
+        from app.services.signal_categories import system2_min_confidence
+        return system2_min_confidence()
+    return sub["min_confidence"]
+
+
+def _should_apply_schedule(sub: dict) -> bool:
+    return not bool(sub.get("_is_system_2"))
+
+
 async def process_post(post: Post, db: AsyncSession) -> None:
     """
     Entry point called by the scraper after a new post is saved.
@@ -91,6 +104,87 @@ async def process_post(post: Post, db: AsyncSession) -> None:
 
     logger.info("Signal: post=%d asset=%s side=%s confidence=%d", post.id, asset, side, post.ai_confidence)
 
+    # ── System routing ─────────────────────────────────────────────────────
+    # Two independent trading systems share this execution layer but NOT
+    # their strategy logic. See app/services/signal_categories.py.
+    from app.services.signal_categories import (
+        get_exit_profile, get_stop_ladder, is_supported_trading_source,
+        is_system_2, system2_display_name,
+    )
+    if not is_supported_trading_source(post.source):
+        logger.info("Post %d skipped: unsupported trading source=%s", post.id, post.source)
+        return
+    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,
+            )
+    else:
+        # SYSTEM 1 — Trump. REPOSITIONED: low-frequency, selective, tight
+        # stop, ≥30-min holds. See signal_categories.py.
+        exit_profile = None
+        from app.services.signal_categories import (
+            TRUMP_MIN_CONFIDENCE, TRUMP_COOLDOWN_HOURS,
+        )
+
+        # (a) Confidence floor — only the very top posts clear the bar.
+        if (post.ai_confidence or 0) < TRUMP_MIN_CONFIDENCE:
+            logger.info(
+                "Post %d skipped: Trump confidence %d < floor %d (low-freq mode)",
+                post.id, post.ai_confidence or 0, TRUMP_MIN_CONFIDENCE,
+            )
+            return
+
+        # (b) Cooldown — "don't trade every opportunity". If we opened ANY
+        #     Trump trade in the last TRUMP_COOLDOWN_HOURS, skip this post.
+        #     DB-backed so it survives restarts (same pattern as scanners).
+        from app.services.scanner_state import last_signal_at
+        from sqlalchemy import select as _sel, func as _fn
+        from app.models import Post as _Post, BotTrade as _BT
+        cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=TRUMP_COOLDOWN_HOURS)
+        recent_trump = await db.execute(
+            _sel(_fn.count(_BT.id))
+            .select_from(_BT)
+            .join(_Post, _Post.id == _BT.trigger_post_id)
+            .where(_Post.source == "truth", _BT.opened_at >= cutoff)
+        )
+        if (recent_trump.scalar() or 0) > 0:
+            logger.info(
+                "Post %d skipped: Trump cooldown active (a Trump trade opened "
+                "within the last %dh)", post.id, TRUMP_COOLDOWN_HOURS,
+            )
+            return
+
+        # (c) Regime filter — still applies; tuned for short-term behaviour.
+        from app.services.regime_filter import passes_regime_filter
+        regime_ok, regime_reasons = passes_regime_filter(asset, side)
+        for r in regime_reasons:
+            logger.info("Regime [%s]: %s", asset, r)
+        if not regime_ok:
+            logger.info("Post %d skipped: regime filter rejected (see ✗ lines above)", post.id)
+            return
+
     result = await db.execute(select(Subscription).where(Subscription.active == True))  # noqa: E712
     subscribers = result.scalars().all()
 
@@ -111,9 +205,80 @@ async def process_post(post: Post, db: AsyncSession) -> None:
             daily_budget_usd=s.daily_budget_usd,
             active_from=s.active_from,
             active_until=s.active_until,
+            # Convex-strategy fields (default to legacy values if column null).
+            trailing_stop_pct=s.trailing_stop_pct,
+            trailing_activate_at_pct=s.trailing_activate_at_pct,
+            max_hold_hours=s.max_hold_hours or 168,
+            manual_window_until=s.manual_window_until,
+            # Phase 1 safety
+            paper_mode=bool(s.paper_mode),
+            circuit_breaker_tripped_at=s.circuit_breaker_tripped_at,
+            circuit_breaker_reason=s.circuit_breaker_reason,
+            # Two-system separation
+            sys2_budget_pct=(s.sys2_budget_pct if s.sys2_budget_pct is not None else 0.7),
+            sys2_cb_tripped_at=s.sys2_cb_tripped_at,
+            sys2_cb_reason=s.sys2_cb_reason,
+            sys2_leverage=s.sys2_leverage,
+            sys2_mode=s.sys2_mode,
+            auto_trade=bool(s.auto_trade),
         )
         for s in subscribers
     ]
+
+    # SYSTEM 2: override the user's Trump exit params with the category's
+    # exit profile. The user configures risk for their Trump scalp; the
+    # reversal system's risk is a property of the SIGNAL, not the user.
+    if sys2 and exit_profile is not None:
+        from app.services.signal_categories import (
+            sys2_effective_leverage, sys2_protective_stop_pct,
+            sys2_normalize_mode,
+        )
+        for snap in subs_snapshot:
+            mode = sys2_normalize_mode(snap.get("sys2_mode"))
+            snap["_sys2_mode"] = mode
+            # Dynamic System-2 leverage (independent of the Trump `leverage`);
+            # default depends on the risk mode (standard 2× / aggressive 8×).
+            lev = sys2_effective_leverage(snap.get("sys2_leverage"), mode)
+            # Protective stop auto-scaled INSIDE the liquidation line for THIS
+            # leverage → the position is de-risked by our monitor, never
+            # liquidated by the exchange.
+            prot = sys2_protective_stop_pct(lev)
+
+            snap["_is_system_2"]             = True
+            snap["_category"]                = post.category
+            snap["leverage"]                 = lev   # HL opens at sys2 leverage
+            snap["take_profit_pct"]          = None   # pure trailing, no fixed TP
+            # Base catastrophic floor = leverage-aware protective stop. The
+            # upside ladder rungs (get_stop_ladder) still ratchet this UP as
+            # peak gain grows; they never loosen it.
+            snap["stop_loss_pct"]            = prot
+            snap["trailing_activate_at_pct"] = exit_profile.trailing_activate_at_pct
+            snap["trailing_stop_pct"]        = exit_profile.trailing_stop_pct
+            snap["max_hold_hours"]           = exit_profile.max_hold_hours
+            # Carried through for the monitor (time-stop / invalidation).
+            snap["_sys2_time_stop_hours"]    = exit_profile.time_stop_hours
+            snap["_sys2_invalidation"]       = exit_profile.invalidation
+            snap["_sys2_invalidation_price"] = post.invalidation_price
+            logger.info(
+                "System-2 dynamic leverage: %dx → protective stop %.2f%% "
+                "(approx liquidation %.2f%%) for %s",
+                lev, prot, 100.0 / lev, snap["wallet"],
+            )
+    elif not sys2:
+        # SYSTEM 1 — Trump, repositioned. System-enforced now (not pure
+        # user params): tight SL always on, but TP/trailing suppressed for
+        # the first TRUMP_MIN_HOLD_MINUTES so we don't scalp out early.
+        # User still controls size / leverage / daily budget.
+        from app.services.signal_categories import (
+            TRUMP_STOP_LOSS_PCT, TRUMP_MAX_HOLD_HOURS, TRUMP_MIN_HOLD_MINUTES,
+        )
+        for snap in subs_snapshot:
+            snap["stop_loss_pct"]      = TRUMP_STOP_LOSS_PCT
+            snap["max_hold_hours"]     = TRUMP_MAX_HOLD_HOURS
+            snap["_min_hold_minutes"]  = TRUMP_MIN_HOLD_MINUTES
+            # Keep user's take_profit_pct / trailing config — they still pick
+            # the profit target; we only gate WHEN it can fire.
+
     post_id = post.id
     post_confidence = post.ai_confidence or 0
 
@@ -135,30 +300,47 @@ async def _execute_for_subscriber(
     if not sub["hl_api_key"]:
         logger.warning("Subscriber %s has no HL API key, skipping", wallet)
         return
-    # Required-setup guard: the bot is only allowed to trade when the user
-    # has explicitly set take-profit, stop-loss, and a daily budget. Prevents
-    # accidental trading on partially-configured accounts.
-    missing = [k for k in ("take_profit_pct", "stop_loss_pct", "daily_budget_usd")
-               if sub.get(k) is None]
+    # Required-setup guard. System 1 (Trump) needs the user's take-profit,
+    # stop-loss, and daily budget. System 2 (reversal) supplies its own
+    # stop-loss + (no) take-profit from the category profile — only the
+    # daily budget remains a user-required risk cap there.
+    if sub.get("_is_system_2"):
+        required = ("daily_budget_usd",)
+    else:
+        required = ("take_profit_pct", "stop_loss_pct", "daily_budget_usd")
+    missing = [k for k in required if sub.get(k) is None]
     if missing:
         logger.info("Sub %s skipped post %d: setup incomplete (missing %s)",
                     wallet, post_id, ", ".join(missing))
         return
 
-    if post_confidence < sub["min_confidence"]:
+    confidence_floor = _confidence_floor_for(sub)
+    if post_confidence < confidence_floor:
         logger.info("Sub %s filters out post %d: conf %d < user min %d",
-                    wallet, post_id, post_confidence, sub["min_confidence"])
+                    wallet, post_id, post_confidence, confidence_floor)
         return
 
-    # Active-window check. Both values stored as naive-UTC.
-    af = sub.get("active_from")
-    au = sub.get("active_until")
-    if af is not None and au is not None:
-        now_utc_naive = datetime.now(timezone.utc).replace(tzinfo=None)
-        if now_utc_naive < af or now_utc_naive >= au:
-            logger.info("Sub %s skipped post %d: outside active window [%s, %s)",
-                        wallet, post_id, af, au)
-            return
+    # ── Master Auto-Trade gate (D) ─────────────────────────────────────────
+    # The ONE user-facing switch. The signal has ALREADY been ingested as a
+    # Post by the ingest endpoint (so it shows in the feed regardless); here
+    # we only decide whether to OPEN a trade. OFF (default) = monitor-only.
+    # Replaces the old scanner-toggle / timed manual-window / schedule trio.
+    # NOTE: this gates NEW entries only — already-open positions keep their
+    # protective de-risk / stop (tp_sl_monitor runs independently of this).
+    if not sub.get("auto_trade"):
+        logger.info("Sub %s: Auto-Trade OFF — post %d recorded, no trade opened",
+                    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.
+    from app.services.circuit_breaker import is_tripped as _cb_tripped
+    _system = "sys2" if sub.get("_is_system_2") else "sys1"
+    tripped, cb_reason = _cb_tripped(sub, _system)
+    if tripped:
+        logger.info("Sub %s skipped post %d: %s", wallet, post_id, cb_reason)
+        return
 
     try:
         api_key = decrypt_api_key(sub["hl_api_key"])
@@ -166,49 +348,170 @@ async def _execute_for_subscriber(
         logger.error("Cannot decrypt key for %s: %s", wallet, exc)
         return
 
+    # ── Position sizing (C) ────────────────────────────────────────────────
+    # Score-based multiplier on the user's base size. Bigger bet when more
+    # confluences are in place. Computed BEFORE the lock so the value is
+    # available for both the budget check and the open call.
+    # System 2 has its OWN sizing — regime_filter's multiplier would shrink
+    # reversal bets (it penalises volatility expansion / prior movement,
+    # which are the SIGNAL for a reversal). See signal_categories.
+    if sub.get("_is_system_2"):
+        from app.services.signal_categories import system2_size_multiplier
+        size_mult = system2_size_multiplier(sub.get("_category"), post_confidence)
+        size_logic = f"sys2 cat={sub.get('_category')}"
+    else:
+        from app.services.regime_filter import calculate_size_multiplier
+        size_mult = calculate_size_multiplier(post_confidence, asset, side)
+        size_logic = "sys1 regime-scored"
+    sized_position_usd = round(sub["position_size_usd"] * size_mult, 2)
+    logger.info(
+        "Sub %s sizing [%s]: base=%.2f × %.2fx = %.2f (asset=%s, conf=%d)",
+        wallet, size_logic, sub["position_size_usd"], size_mult,
+        sized_position_usd, asset, post_confidence,
+    )
+
     # Per-wallet lock wraps BOTH the budget re-check and the open, so two
     # concurrent signals can't both pass the daily-budget gate before either
     # trade is written to DB (TOCTOU race under asyncio.gather).
     async with _wallet_lock(wallet):
         # Re-check daily budget INSIDE the lock so it's atomic with the open.
-        daily_cap = sub.get("daily_budget_usd")
-        if daily_cap is not None and daily_cap > 0:
+        # Budget is SPLIT between the two systems so a Trump scalp can't eat
+        # the allocation reserved for a once-a-year reversal signal.
+        total_cap = sub.get("daily_budget_usd")
+        if total_cap is not None and total_cap > 0:
+            sys2_pct = sub.get("sys2_budget_pct", 0.7)
+            is_s2 = bool(sub.get("_is_system_2"))
+            daily_cap = total_cap * (sys2_pct if is_s2 else (1.0 - sys2_pct))
             start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=None)
+            from app.models import Post as _P
+            from app.services.signal_categories import SYSTEM_2_SOURCES
             async with AsyncSessionLocal() as budget_db:
                 spent_result = await budget_db.execute(
-                    select(BotTrade).where(
+                    select(BotTrade, _P.source)
+                    .outerjoin(_P, _P.id == BotTrade.trigger_post_id)
+                    .where(
                         BotTrade.wallet_address == wallet,
                         BotTrade.opened_at >= start_of_day,
                     )
                 )
-                spent_trades = spent_result.scalars().all()
-                spent = sum(
-                    (t.size_usd if t.size_usd is not None else sub["position_size_usd"])
-                    for t in spent_trades
+                # Only count spend from THIS system against THIS system's slice.
+                spent = 0.0
+                for t, src in spent_result.all():
+                    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"])
+            if spent + sized_position_usd > daily_cap:
+                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 1.0 - sys2_pct) * 100, total_cap)
+                return
+
+        # ── System-2 correlation / concentration cap ───────────────────────
+        # Inside the wallet lock so it's atomic with the open. The whole
+        # System-2 basket is correlated crypto-beta; cap CONCURRENT open
+        # positions + total open notional so a synchronised "crypto bottomed"
+        # week can't stack into one 10x leveraged macro bet.
+        if sub.get("_is_system_2"):
+            from app.services.signal_categories import (
+                SYS2_MAX_CONCURRENT, SYS2_MAX_OPEN_NOTIONAL_MULT, SYSTEM_2_SOURCES,
+            )
+            from app.models import Post as _P2
+            async with AsyncSessionLocal() as conc_db:
+                open_rows = await conc_db.execute(
+                    select(BotTrade, _P2.source)
+                    .outerjoin(_P2, _P2.id == BotTrade.trigger_post_id)
+                    .where(
+                        BotTrade.wallet_address == wallet,
+                        BotTrade.closed_at.is_(None),
+                    )
                 )
-            if spent + sub["position_size_usd"] > daily_cap:
-                logger.info("Sub %s daily budget reached: spent=%.2f + new=%.2f > cap=%.2f",
-                            wallet, spent, sub["position_size_usd"], daily_cap)
+                open_sys2 = [
+                    t for t, src in open_rows.all()
+                    if (src or "").lower() in SYSTEM_2_SOURCES
+                ]
+            open_count = len(open_sys2)
+            open_notional = sum(
+                (t.size_usd if t.size_usd is not None else sub["position_size_usd"])
+                for t in open_sys2
+            )
+            notional_ceiling = SYS2_MAX_OPEN_NOTIONAL_MULT * sub["position_size_usd"]
+            if open_count >= SYS2_MAX_CONCURRENT:
+                logger.info(
+                    "Sub %s sys2 concentration cap: %d open positions ≥ max %d "
+                    "(correlated crypto-beta — skipping post %d)",
+                    wallet, open_count, SYS2_MAX_CONCURRENT, post_id)
+                return
+            if open_notional + sized_position_usd > notional_ceiling:
+                logger.info(
+                    "Sub %s sys2 notional cap: open=%.2f + new=%.2f > ceiling=%.2f "
+                    "(%.1f× base) — skipping post %d",
+                    wallet, open_notional, sized_position_usd, notional_ceiling,
+                    SYS2_MAX_OPEN_NOTIONAL_MULT, post_id)
                 return
 
         # Each subscriber gets its own session — AsyncSession is NOT concurrency-safe.
         async with AsyncSessionLocal() as db:
             try:
-                trader = HyperliquidTrader(
-                    api_private_key=api_key,
-                    account_address=wallet,
-                    leverage=sub["leverage"],
-                    mainnet=settings.hl_mainnet,
-                )
+                # ── Paper-mode branch (P1.3) ──────────────────────────────
+                # Skip Hyperliquid entirely. Use the current Binance price as
+                # the synthetic fill. DB row gets hl_order_id="paper" so close
+                # logic + reconciliation know to skip HL too.
+                if sub["paper_mode"]:
+                    from app.services.price_store import price_store
+                    entry_price = price_store.latest_price(asset)
+                    if not entry_price:
+                        logger.warning("Paper mode: no price for %s, skipping", asset)
+                        return
+                    # Check DB (not HL) for already-open paper position on same asset.
+                    open_dup = await db.execute(
+                        select(BotTrade).where(
+                            BotTrade.wallet_address == wallet,
+                            BotTrade.asset == asset,
+                            BotTrade.closed_at.is_(None),
+                        )
+                    )
+                    if open_dup.scalar_one_or_none():
+                        logger.info("Paper mode: %s already has open %s, skipping", wallet, asset)
+                        return
+                    result = {"fill_price": entry_price, "order_id": "paper"}
+                    logger.info("[PAPER] Open %s %s for %s @ %.4f size=%.2f",
+                                side, asset, wallet, entry_price, sized_position_usd)
+                else:
+                    trader = HyperliquidTrader(
+                        api_private_key=api_key,
+                        account_address=wallet,
+                        leverage=sub["leverage"],
+                        mainnet=settings.hl_mainnet,
+                    )
 
-                open_positions = await trader.get_open_positions()
-                if any(p.get('coin') == asset for p in open_positions):
-                    logger.info("Subscriber %s already has open %s position, skipping", wallet, asset)
-                    return
+                    open_positions = await trader.get_open_positions()
+                    if any(p.get('coin') == asset for p in open_positions):
+                        logger.info("Subscriber %s already has open %s position, skipping", wallet, asset)
+                        return
 
-                result = await trader.open_position(asset, side, sub["position_size_usd"])
+                    result = await trader.open_position(asset, side, sized_position_usd)
                 entry_price = result.get('fill_price', 0.0)
 
+                # Resolve the FROZEN exit profile ONCE. Everything downstream
+                # (the watchdog AND recovery-after-restart) reads these — never
+                # the mutable Subscription/category config again.
+                import time as _time
+                eff_min_hold_ts = (
+                    _time.time() + sub["_min_hold_minutes"] * 60
+                    if sub.get("_min_hold_minutes") else None
+                )
+                eff = dict(
+                    take_profit_pct       = sub["take_profit_pct"],
+                    stop_loss_pct         = sub["stop_loss_pct"],
+                    trailing_stop_pct     = sub.get("trailing_stop_pct"),
+                    trailing_activate_pct = sub.get("trailing_activate_at_pct"),
+                    max_hold_hours        = int(sub["max_hold_hours"]),
+                    invalidation          = sub.get("_sys2_invalidation"),
+                    invalidation_price    = sub.get("_sys2_invalidation_price"),
+                    min_hold_until_ts     = eff_min_hold_ts,
+                )
+
                 trade = BotTrade(
                     asset=asset,
                     side=side,
@@ -217,17 +520,40 @@ async def _execute_for_subscriber(
                     trigger_post_id=post_id,
                     opened_at=datetime.now(timezone.utc).replace(tzinfo=None),
                     hl_order_id=result.get('order_id'),
-                    size_usd=sub["position_size_usd"],
+                    size_usd=sized_position_usd,
+                    base_size_usd=sized_position_usd,   # immutable; pyramiding grows size_usd
+                    sys2_mode=(_sys2_mode if sys2 else None),
                     leverage=sub["leverage"],
+                    # Freeze the exit profile on the row.
+                    eff_take_profit_pct       = eff["take_profit_pct"],
+                    eff_stop_loss_pct         = eff["stop_loss_pct"],
+                    eff_trailing_stop_pct     = eff["trailing_stop_pct"],
+                    eff_trailing_activate_pct = eff["trailing_activate_pct"],
+                    eff_max_hold_hours        = eff["max_hold_hours"],
+                    eff_invalidation          = eff["invalidation"],
+                    eff_invalidation_price    = eff["invalidation_price"],
+                    eff_min_hold_until_ts     = eff["min_hold_until_ts"],
                 )
                 db.add(trade)
                 await db.commit()
                 await db.refresh(trade)
 
-                logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)",
-                            side, asset, wallet, entry_price, trade.id)
+                logger.info("Opened %s %s for %s @ %.4f size=%.2f (trade_id=%d)",
+                            side, asset, wallet, entry_price, sized_position_usd, trade.id)
 
+                # Register with the watchdog using the FROZEN profile.
                 from app.services.tp_sl_monitor import register_trade
+                _derisk = None
+                _addon = None
+                _peak_trail = None
+                _sys2_mode = sub.get("_sys2_mode", "standard")
+                if sys2:
+                    from app.services.signal_categories import (
+                        sys2_derisk_ladder, sys2_addon_ladder, sys2_peak_trail,
+                    )
+                    _derisk = sys2_derisk_ladder(sub["leverage"], _sys2_mode)
+                    _addon = sys2_addon_ladder(_sys2_mode)
+                    _peak_trail = sys2_peak_trail(_sys2_mode)
                 register_trade(
                     trade_id=trade.id,
                     wallet=wallet,
@@ -236,20 +562,235 @@ async def _execute_for_subscriber(
                     asset=asset,
                     side=side,
                     entry_price=entry_price,
-                    take_profit_pct=sub["take_profit_pct"],
-                    stop_loss_pct=sub["stop_loss_pct"],
+                    take_profit_pct=eff["take_profit_pct"],
+                    stop_loss_pct=eff["stop_loss_pct"],
+                    trailing_stop_pct=eff["trailing_stop_pct"],
+                    trailing_activate_at_pct=eff["trailing_activate_pct"],
+                    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,
+                    derisk_ladder=_derisk,
+                    derisk_done=0,
+                    addon_ladder=_addon,
+                    addon_done=0,
+                    peak_trail=_peak_trail,
+                    grow_mode=bool(getattr(trade, "grow_mode", False)),
                 )
 
+                # Max hold backstop (per-user for System 1, per-category for
+                # System 2 — already injected into the snapshot upstream).
+                max_hold_seconds = int(sub["max_hold_hours"]) * 3600
                 task = asyncio.create_task(_close_after_hold(
-                    trade.id, api_key, sub["leverage"], asset, wallet
+                    trade.id, api_key, sub["leverage"], asset, wallet, max_hold_seconds,
                 ))
                 _background_tasks.add(task)
                 task.add_done_callback(_background_tasks.discard)
 
+                # System-2 time-stop: if the position is still ~flat after
+                # `time_stop_hours`, the thesis is slow/dead — close early
+                # instead of tying up capital for the full max-hold.
+                ts_hours = sub.get("_sys2_time_stop_hours")
+                if ts_hours:
+                    ts_task = asyncio.create_task(_time_stop_check(
+                        trade.id, api_key, sub["leverage"], asset, wallet,
+                        int(ts_hours) * 3600,
+                    ))
+                    _background_tasks.add(ts_task)
+                    ts_task.add_done_callback(_background_tasks.discard)
+
             except Exception as e:
                 logger.error("Trade execution failed for %s: %s", wallet, e)
 
 
+async def partial_derisk(
+    trade_id: int,
+    api_key: str,
+    asset: str,
+    wallet: str,
+    step_idx: int,
+    frac_of_original: float,
+    reason: str = "derisk",
+) -> bool:
+    """Staged de-risk: partially close a System-2 trade, realise the slice's
+    PnL, and KEEP the trade open. Idempotent per step_idx (guarded by the
+    persisted `derisk_steps_done` counter under the per-trade lock).
+
+    Returns True if the step was executed (or already done), False on a
+    recoverable no-op so the monitor can retry next tick.
+    """
+    async with _lock_for(trade_id):
+        async with AsyncSessionLocal() as db:
+            try:
+                row = await db.execute(select(BotTrade).where(BotTrade.id == trade_id))
+                trade = row.scalar_one_or_none()
+                if trade is None or trade.closed_at is not None:
+                    return True  # gone / already fully closed — nothing to do
+                if (trade.derisk_steps_done or 0) > step_idx:
+                    return True  # this step already executed (idempotent)
+                if (trade.derisk_steps_done or 0) != step_idx:
+                    # Steps must run in order; let the monitor catch up.
+                    return False
+
+                size_usd = trade.size_usd if trade.size_usd is not None else 0.0
+                rem_frac = trade.remaining_fraction if trade.remaining_fraction is not None else 1.0
+                if size_usd <= 0 or rem_frac <= 0:
+                    return True
+                # frac of the CURRENT open position needed to shed
+                # `frac_of_original` of the ORIGINAL notional.
+                frac_of_current = max(0.0, min(1.0, frac_of_original / rem_frac))
+
+                if trade.hl_order_id == "paper":
+                    from app.services.price_store import price_store
+                    fill = price_store.latest_price(asset)
+                    if not fill:
+                        logger.error("Paper de-risk: no %s price, retry trade %d", asset, trade_id)
+                        return False
+                    closed_frac_of_current = frac_of_current
+                else:
+                    trader = HyperliquidTrader(
+                        api_private_key=api_key,
+                        account_address=wallet,
+                        leverage=(trade.leverage or 1),
+                        mainnet=settings.hl_mainnet,
+                    )
+                    r = await trader.reduce_position(asset, frac_of_current)
+                    if r.get("already_closed"):
+                        # Position vanished (manual close / liquidation race).
+                        # Let the reconciler / full-close path handle it.
+                        logger.warning("De-risk: %s position gone for trade %d", asset, trade_id)
+                        return True
+                    fill = r.get("fill_price")
+                    closed_frac_of_current = r.get("closed_fraction") or 0.0
+                    if not fill or closed_frac_of_current <= 0:
+                        return False  # no fill — retry next tick
+
+                # Fraction of ORIGINAL notional actually shed this step.
+                closed_frac_of_original = closed_frac_of_current * rem_frac
+                slice_usd = size_usd * closed_frac_of_original
+                pct = ((fill - trade.entry_price) / trade.entry_price) if trade.entry_price else 0.0
+                signed_pct = pct if trade.side == "long" else -pct
+                # Round-trip taker fee on this slice (open-share + this close).
+                fees = slice_usd * HL_TAKER_FEE_RATE * 2
+                slice_pnl = slice_usd * signed_pct - fees
+
+                trade.realized_partial_pnl_usd = round((trade.realized_partial_pnl_usd or 0.0) + slice_pnl, 4)
+                trade.remaining_fraction = max(0.0, round(rem_frac - closed_frac_of_original, 6))
+                trade.derisk_steps_done = step_idx + 1
+                await db.commit()
+
+                logger.warning(
+                    "De-risk step %d trade %d (%s): closed %.1f%% of original "
+                    "@ %.2f, slice PnL %.2f, remaining %.1f%% (reason=%s)",
+                    step_idx + 1, trade_id, asset, closed_frac_of_original * 100,
+                    fill, slice_pnl, trade.remaining_fraction * 100, reason,
+                )
+                return True
+            except Exception as e:
+                logger.error("partial_derisk failed for trade %d step %d: %s",
+                             trade_id, step_idx, e)
+                return False
+
+
+async def pyramid_add(
+    trade_id: int,
+    api_key: str,
+    asset: str,
+    wallet: str,
+    step_idx: int,
+    frac_of_base: float,
+    reason: str = "pyramid",
+) -> tuple:
+    """Pyramiding: add to a CONFIRMED System-2 winner. Returns
+    (ok: bool, new_entry: float|None, ref_price: float|None).
+
+    Guards: only while derisk_steps_done==0 (clean uptrend), idempotent per
+    step_idx, structural confirmation re-checked at execution, per-trade
+    notional cap. On success blends the average entry and grows size_usd.
+    """
+    async with _lock_for(trade_id):
+        async with AsyncSessionLocal() as db:
+            try:
+                row = await db.execute(select(BotTrade).where(BotTrade.id == trade_id))
+                trade = row.scalar_one_or_none()
+                if trade is None or trade.closed_at is not None:
+                    return (True, None, None)
+                if (trade.derisk_steps_done or 0) > 0:
+                    return (True, None, None)   # went underwater — no pyramiding
+                if (trade.addon_steps_done or 0) > step_idx:
+                    return (True, None, None)   # already done (idempotent)
+                if (trade.addon_steps_done or 0) != step_idx:
+                    return (False, None, None)  # out of order — retry later
+
+                base = trade.base_size_usd or trade.size_usd or 0.0
+                if base <= 0:
+                    return (True, None, None)
+                from app.services.signal_categories import SYS2_MAX_OPEN_NOTIONAL_MULT
+                add_usd = round(base * frac_of_base, 2)
+                cur_notional = trade.size_usd or base
+                if cur_notional + add_usd > base * SYS2_MAX_OPEN_NOTIONAL_MULT:
+                    logger.warning("Pyramid: notional cap hit trade %d (%.2f+%.2f > %.1fx base)",
+                                   trade_id, cur_notional, add_usd, SYS2_MAX_OPEN_NOTIONAL_MULT)
+                    trade.addon_steps_done = step_idx + 1     # stop retrying
+                    await db.commit()
+                    return (True, None, None)
+
+                # Structural confirmation — fetched ONCE here, not per tick.
+                from app.services.market_data import for_asset, drop_in_progress_bar
+                from app.services.bottom_indicators import trend_confirmed
+                from app.services.price_store import price_store
+                prov = for_asset(asset)
+                raw = await prov.fetch_1d(asset, days=260)
+                daily = drop_in_progress_bar(raw, "1d")
+                closes = [float(c["close"]) for c in daily if c.get("close")]
+                highs = [float(c["high"]) for c in daily if c.get("high")]
+                px = price_store.latest_price(asset)
+                if not px:
+                    return (False, None, None)
+                if not trend_confirmed(closes, highs, px):
+                    return (False, None, None)   # not a confirmed uptrend yet
+
+                if trade.hl_order_id == "paper":
+                    fill = px
+                    actual_add_usd = add_usd          # synthetic full fill
+                else:
+                    trader = HyperliquidTrader(
+                        api_private_key=api_key,
+                        account_address=wallet,
+                        leverage=(trade.leverage or 1),
+                        mainnet=settings.hl_mainnet,
+                    )
+                    r = await trader.open_position(asset, trade.side, add_usd)
+                    fill = r.get("fill_price")
+                    filled_coins = float(r.get("size_coins") or 0.0)
+                    if not fill or filled_coins <= 0:
+                        return (False, None, None)
+                    # Use the ACTUAL filled notional, not the intended amount —
+                    # an IOC add can under-fill on thin/volatile books; assuming
+                    # the full add_usd would corrupt the blended entry + size.
+                    actual_add_usd = filled_coins * fill
+
+                old_notional = trade.size_usd or base
+                new_notional = old_notional + actual_add_usd
+                blended = (old_notional * trade.entry_price + actual_add_usd * fill) / new_notional
+                trade.entry_price = round(blended, 6)
+                trade.size_usd = round(new_notional, 2)
+                trade.addon_steps_done = step_idx + 1
+                await db.commit()
+
+                logger.warning(
+                    "Pyramid add step %d trade %d (%s): +$%.2f filled @ %.2f → "
+                    "notional $%.2f, blended entry %.4f (reason=%s)",
+                    step_idx + 1, trade_id, asset, actual_add_usd, fill,
+                    new_notional, blended, reason,
+                )
+                return (True, round(blended, 6), float(fill))
+            except Exception as e:
+                logger.error("pyramid_add failed trade %d step %d: %s",
+                             trade_id, step_idx, e)
+                return (False, None, None)
+
+
 async def close_and_finalize(
     trade_id: int,
     api_key: str,
@@ -297,23 +838,46 @@ async def close_and_finalize(
                     size_usd = sub.position_size_usd if sub else 20.0
                 trade_leverage = trade.leverage if trade.leverage is not None else leverage
 
-                trader = HyperliquidTrader(
-                    api_private_key=api_key,
-                    account_address=wallet,
-                    leverage=trade_leverage,
-                    mainnet=settings.hl_mainnet,
-                )
-                close_result = await trader.close_position(asset)
+                # ── Paper-mode close (P1.3) ──────────────────────────────
+                # No Hyperliquid call. Synthesize a fill at the current Binance
+                # price. Same downstream PnL math as a real trade.
+                if trade.hl_order_id == "paper":
+                    from app.services.price_store import price_store
+                    paper_exit = price_store.latest_price(asset)
+                    if not paper_exit:
+                        logger.error("Paper close: no price for %s, can't close trade %d",
+                                     asset, trade_id)
+                        # Roll back the closed_at claim so we can retry later.
+                        await db.execute(
+                            update(BotTrade).where(BotTrade.id == trade_id).values(closed_at=None)
+                        )
+                        await db.commit()
+                        return
+                    close_result = {"fill_price": paper_exit, "already_closed": False}
+                else:
+                    trader = HyperliquidTrader(
+                        api_private_key=api_key,
+                        account_address=wallet,
+                        leverage=trade_leverage,
+                        mainnet=settings.hl_mainnet,
+                    )
+                    close_result = await trader.close_position(asset)
 
                 # Detect "position was already closed externally" before we even tried
                 if close_result.get("already_closed"):
+                    # PnL of the (now-gone) open remainder is unknown, BUT any
+                    # PnL already BANKED by staged de-risk is real money — keep
+                    # it in pnl_usd instead of discarding it as None, else
+                    # analytics / "today realised" silently undercount.
+                    banked = trade.realized_partial_pnl_usd or 0.0
                     logger.warning(
-                        "Trade %d: no open %s position found on HL — "
-                        "user likely closed it manually. Marking trade closed with no PnL.",
-                        trade_id, asset,
+                        "Trade %d: no open %s position found on HL — user likely "
+                        "closed it manually. Marking closed; pnl=%s (banked de-risk only).",
+                        trade_id, asset, round(banked, 2) if banked else None,
                     )
                     trade.exit_price = None
-                    trade.pnl_usd = None
+                    trade.pnl_usd = round(banked, 2) if banked else None
+                    trade.remaining_fraction = 0.0
                     trade.hold_seconds = int(
                         (datetime.now(timezone.utc) -
                          trade.opened_at.replace(tzinfo=timezone.utc)).total_seconds()
@@ -334,13 +898,24 @@ async def close_and_finalize(
                 pct = ((exit_price - trade.entry_price) / trade.entry_price) if trade.entry_price else 0.0
                 signed_pct = pct if trade.side == 'long' else -pct
                 # size_usd is the NOTIONAL value — leverage affects margin, not PnL.
-                gross_pnl = size_usd * signed_pct
-                # Deduct round-trip taker fees (IOC crosses book on both open + close).
-                # Without this, displayed PnL overstates real returns by ~9 bps per trade.
-                fees_usd = size_usd * HL_TAKER_FEE_RATE * 2
-                pnl_usd = gross_pnl - fees_usd
+                # For a staged-de-risk trade only the REMAINING notional is
+                # closed here; earlier partial reduces already realised their
+                # slice into realized_partial_pnl_usd. Legacy/non-partial rows
+                # have remaining_fraction=1.0 + realized_partial=0.0 → identical
+                # math to before.
+                rem_frac = trade.remaining_fraction if trade.remaining_fraction is not None else 1.0
+                prior_pnl = trade.realized_partial_pnl_usd or 0.0
+                remaining_size = size_usd * rem_frac
+                gross_pnl = remaining_size * signed_pct
+                # Round-trip taker fees: the OPEN fee was paid on the full
+                # original notional; partial reduces already charged their
+                # close-side fee. Here charge the open-share for the remaining
+                # slice + this final close fee.
+                fees_usd = remaining_size * HL_TAKER_FEE_RATE * 2
+                pnl_usd = gross_pnl - fees_usd + prior_pnl
 
                 trade.exit_price = exit_price
+                trade.remaining_fraction = 0.0
                 trade.pnl_usd = round(pnl_usd, 2)
                 trade.hold_seconds = hold_secs
                 # closed_at already set by the atomic UPDATE
@@ -357,6 +932,24 @@ async def close_and_finalize(
                     gross_pnl, fees_usd, pnl_usd, reason,
                 )
 
+                # ── Circuit breaker check (P1.1) ───────────────────────────
+                # Run after PnL is written. If daily DD or consecutive-loss
+                # threshold hit, trip the breaker — that nulls manual_window
+                # and blocks new trades for CB_LOCKOUT_HOURS.
+                # Infer which system this trade belonged to from its trigger
+                # post's source, so the right (independent) breaker is checked.
+                from app.services.circuit_breaker import check_and_trip
+                from app.services.signal_categories import SYSTEM_2_SOURCES
+                _src_row = await db.execute(
+                    select(Post.source).where(Post.id == trade.trigger_post_id)
+                )
+                _src = (_src_row.scalar_one_or_none() or "").lower()
+                _sys = "sys2" if _src in SYSTEM_2_SOURCES else "sys1"
+                cb_reason = await check_and_trip(wallet, db, _sys)
+                if cb_reason:
+                    logger.warning("Circuit breaker [%s] tripped for %s: %s",
+                                    _sys, wallet, cb_reason)
+
             except Exception as e:
                 logger.error("Failed to close trade %d: %s", trade_id, e)
                 # Rollback closed_at so recovery can retry this trade on next restart.
@@ -377,7 +970,58 @@ async def close_and_finalize(
 
 
 async def _close_after_hold(
-    trade_id: int, api_key: str, leverage: int, asset: str, wallet: str
+    trade_id: int, api_key: str, leverage: int, asset: str, wallet: str,
+    max_hold_seconds: int,
 ) -> None:
-    await asyncio.sleep(MAX_HOLD_SECONDS)
+    """Per-user max-hold backstop. Forces close after the configured duration.
+
+    The trailing-stop monitor will usually close the position long before this
+    fires, but for trades that drift sideways the cap prevents indefinite
+    funding-rate bleed.
+    """
+    await asyncio.sleep(max_hold_seconds)
     await close_and_finalize(trade_id, api_key, leverage, asset, wallet, reason="max_hold")
+
+
+# Flat-zone band: a position whose |unrealised| is within this % at the
+# time-stop checkpoint is considered "not working" and closed early.
+_TIME_STOP_FLAT_BAND_PCT = 2.0
+
+
+async def _time_stop_check(
+    trade_id: int, api_key: str, leverage: int, asset: str, wallet: str,
+    delay_seconds: int,
+) -> None:
+    """System-2 time-stop. After `delay_seconds`, if the trade is still open
+    AND roughly flat (|unrealised| < band), the thesis is slow/dead — close
+    it so capital isn't parked for the full multi-week max-hold on a dud.
+
+    A trade that's already moved (win or stopped) will be closed/gone by
+    now; the conditional UPDATE in close_and_finalize makes a late fire a
+    harmless no-op.
+    """
+    await asyncio.sleep(delay_seconds)
+
+    from sqlalchemy import select
+    async with AsyncSessionLocal() as db:
+        row = await db.execute(select(BotTrade).where(BotTrade.id == trade_id))
+        trade = row.scalar_one_or_none()
+        if trade is None or trade.closed_at is not None:
+            return  # already closed by trail / SL / invalidation
+        entry = trade.entry_price
+
+    from app.services.price_store import price_store
+    cur = price_store.latest_price(asset)
+    if not cur or not entry:
+        return  # no price → don't act blindly; max-hold will catch it
+
+    raw = (cur - entry) / entry
+    signed_pct = (raw if trade.side == "long" else -raw) * 100
+    if abs(signed_pct) < _TIME_STOP_FLAT_BAND_PCT:
+        logger.info(
+            "Time-stop firing trade %d (%s %s): flat at %.2f%% after %dh",
+            trade_id, trade.side, asset, signed_pct, delay_seconds // 3600,
+        )
+        await close_and_finalize(
+            trade_id, api_key, leverage, asset, wallet, reason="time_stop",
+        )