feat: add daily budget, active window, trade snapshots, and price impact monitor

- New migrations for daily_budget, active_window, and bottrade snapshot
- Add trumpstruth scraper and price_impact_monitor service
- Expand bot_engine, hyperliquid, recovery, and tp_sl_monitor logic
- Update API/schemas/models for new features

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

app/services/analysis.py

@@ -207,7 +207,7 @@ async def analyze_post(text: str) -> dict:
         if asset not in ("BTC", "ETH", None):
             asset = None
 
-        reasoning = str(result.get("reasoning", ""))[:600]
+        reasoning = str(result.get("reasoning", ""))[:1200]
 
         return {
             "relevant": relevant,

app/services/binance.py

@@ -45,6 +45,10 @@ async def _process_message(raw: str):
         from app.services.tp_sl_monitor import on_price_tick
         on_price_tick(asset, candle["close"])
 
+        # Price-impact peak tracker (updates rolling max for open post windows)
+        from app.services.price_impact_monitor import on_price_tick as pi_tick
+        pi_tick(asset, candle["high"], candle["low"], candle["close"])
+
         # Broadcast live price tick
         await manager.broadcast({
             "type": "price",

app/services/bot_engine.py

@@ -22,15 +22,38 @@ from app.services.price_store import price_store  # noqa: F401 (used elsewhere)
 logger = logging.getLogger(__name__)
 
 # Platform-wide thresholds (per-user values in Subscription override where applicable)
-GLOBAL_MIN_CONFIDENCE = 80       # hard floor — user min_confidence must be >= this
+# 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)
 
+# Hyperliquid perp fees (mainnet, base tier).
+# IOC orders always cross the book → taker fee both on open and close.
+# Ref: https://hyperliquid.gitbook.io/hyperliquid-docs/trading/fees
+HL_TAKER_FEE_RATE = 0.00045       # 4.5 bps per side → 9 bps round-trip
+
 
 # Per-trade locks prevent TP/SL and max-hold tasks from both calling close_and_finalize
 # simultaneously on the same trade. Lock is process-local — with the atomic
 # conditional UPDATE below, multi-process is still safe (losers become no-ops).
 _close_locks: Dict[int, asyncio.Lock] = {}
 
+# Strong references to background close tasks. Python's GC can collect
+# unreferenced asyncio.Task objects before they finish — keeping them here
+# prevents silent task cancellation. Entries are removed when tasks complete.
+_background_tasks: set = set()
+
+# Per-wallet locks for the open-position critical section.
+# Prevents two concurrent signals from both passing the daily-budget check
+# before either trade is written to DB (TOCTOU race).
+_wallet_open_locks: Dict[str, asyncio.Lock] = {}
+
+
+def _wallet_lock(wallet: str) -> asyncio.Lock:
+    lock = _wallet_open_locks.get(wallet)
+    if lock is None:
+        lock = asyncio.Lock()
+        _wallet_open_locks[wallet] = lock
+    return lock
+
 
 def _lock_for(trade_id: int) -> asyncio.Lock:
     lock = _close_locks.get(trade_id)
@@ -47,14 +70,12 @@ async def process_post(post: Post, db: AsyncSession) -> None:
     """
     if not post.relevant:
         return
-    if (post.ai_confidence or 0) < GLOBAL_MIN_CONFIDENCE:
-        logger.info("Post %d skipped: confidence %d < %d", post.id, post.ai_confidence, GLOBAL_MIN_CONFIDENCE)
-        return
-    if post.signal not in ('buy', 'short'):
+    if post.signal not in ('buy', 'short', 'sell'):
         logger.info("Post %d skipped: signal=%s is not actionable", post.id, post.signal)
         return
 
     asset = post.price_impact_asset or 'BTC'
+    # "sell" is treated as a short signal (bearish directional trade)
     side = 'long' if post.signal == 'buy' else 'short'
 
     logger.info("Signal: post=%d asset=%s side=%s confidence=%d", post.id, asset, side, post.ai_confidence)
@@ -76,6 +97,9 @@ async def process_post(post: Post, db: AsyncSession) -> None:
             take_profit_pct=s.take_profit_pct,
             stop_loss_pct=s.stop_loss_pct,
             min_confidence=s.min_confidence,
+            daily_budget_usd=s.daily_budget_usd,
+            active_from=s.active_from,
+            active_until=s.active_until,
         )
         for s in subscribers
     ]
@@ -100,70 +124,119 @@ 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]
+    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"]:
         logger.info("Sub %s filters out post %d: conf %d < user min %d",
                     wallet, post_id, post_confidence, sub["min_confidence"])
         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
+
     try:
         api_key = decrypt_api_key(sub["hl_api_key"])
     except Exception as exc:
         logger.error("Cannot decrypt key for %s: %s", wallet, exc)
         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,
-            )
-
-            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)
+    # 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:
+            start_of_day = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=None)
+            async with AsyncSessionLocal() as budget_db:
+                spent_result = await budget_db.execute(
+                    select(BotTrade).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
+                )
+            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)
                 return
 
-            result = await trader.open_position(asset, side, sub["position_size_usd"])
-            entry_price = result.get('fill_price', 0.0)
+        # 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,
+                )
 
-            trade = BotTrade(
-                asset=asset,
-                side=side,
-                entry_price=entry_price,
-                wallet_address=wallet,
-                trigger_post_id=post_id,
-                opened_at=datetime.now(timezone.utc).replace(tzinfo=None),
-                hl_order_id=result.get('order_id'),
-            )
-            db.add(trade)
-            await db.commit()
-            await db.refresh(trade)
+                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
 
-            logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)",
-                        side, asset, wallet, entry_price, trade.id)
+                result = await trader.open_position(asset, side, sub["position_size_usd"])
+                entry_price = result.get('fill_price', 0.0)
 
-            from app.services.tp_sl_monitor import register_trade
-            register_trade(
-                trade_id=trade.id,
-                wallet=wallet,
-                api_key=api_key,
-                leverage=sub["leverage"],
-                asset=asset,
-                side=side,
-                entry_price=entry_price,
-                take_profit_pct=sub["take_profit_pct"],
-                stop_loss_pct=sub["stop_loss_pct"],
-            )
+                trade = BotTrade(
+                    asset=asset,
+                    side=side,
+                    entry_price=entry_price,
+                    wallet_address=wallet,
+                    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"],
+                    leverage=sub["leverage"],
+                )
+                db.add(trade)
+                await db.commit()
+                await db.refresh(trade)
 
-            asyncio.create_task(_close_after_hold(
-                trade.id, api_key, sub["leverage"], asset, wallet
-            ))
+                logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)",
+                            side, asset, wallet, entry_price, trade.id)
 
-        except Exception as e:
-            logger.error("Trade execution failed for %s: %s", wallet, e)
+                from app.services.tp_sl_monitor import register_trade
+                register_trade(
+                    trade_id=trade.id,
+                    wallet=wallet,
+                    api_key=api_key,
+                    leverage=sub["leverage"],
+                    asset=asset,
+                    side=side,
+                    entry_price=entry_price,
+                    take_profit_pct=sub["take_profit_pct"],
+                    stop_loss_pct=sub["stop_loss_pct"],
+                )
+
+                task = asyncio.create_task(_close_after_hold(
+                    trade.id, api_key, sub["leverage"], asset, wallet
+                ))
+                _background_tasks.add(task)
+                task.add_done_callback(_background_tasks.discard)
+
+            except Exception as e:
+                logger.error("Trade execution failed for %s: %s", wallet, e)
 
 
 async def close_and_finalize(
@@ -200,27 +273,61 @@ async def close_and_finalize(
                 result = await db.execute(select(BotTrade).where(BotTrade.id == trade_id))
                 trade = result.scalar_one()
 
-                sub_res = await db.execute(
-                    select(Subscription).where(Subscription.wallet_address == wallet)
-                )
-                sub = sub_res.scalar_one_or_none()
-                size_usd = sub.position_size_usd if sub else 20.0
+                # Prefer the historical snapshot stamped on the trade row;
+                # fall back to current Subscription or caller's leverage for
+                # legacy rows.
+                if trade.size_usd is not None:
+                    size_usd = trade.size_usd
+                else:
+                    sub_res = await db.execute(
+                        select(Subscription).where(Subscription.wallet_address == wallet)
+                    )
+                    sub = sub_res.scalar_one_or_none()
+                    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=leverage,
+                    leverage=trade_leverage,
                     mainnet=settings.hl_mainnet,
                 )
                 close_result = await trader.close_position(asset)
-                exit_price = close_result.get('fill_price', 0.0)
+
+                # Detect "position was already closed externally" before we even tried
+                if close_result.get("already_closed"):
+                    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.exit_price = None
+                    trade.pnl_usd = None
+                    trade.hold_seconds = int(
+                        (datetime.now(timezone.utc) -
+                         trade.opened_at.replace(tzinfo=timezone.utc)).total_seconds()
+                    )
+                    await db.commit()
+                    from app.services.tp_sl_monitor import unregister
+                    unregister(trade_id)
+                    _close_locks.pop(trade_id, None)
+                    return
+
+                exit_price = close_result.get("fill_price")
+                if not exit_price:
+                    raise ValueError(f"close_position returned no fill_price for trade {trade_id}")
 
                 now_aware = datetime.now(timezone.utc)
                 opened_aware = trade.opened_at.replace(tzinfo=timezone.utc)
                 hold_secs = int((now_aware - opened_aware).total_seconds())
                 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
-                pnl_usd = size_usd * signed_pct * leverage
+                # 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
 
                 trade.exit_price = exit_price
                 trade.pnl_usd = round(pnl_usd, 2)
@@ -233,11 +340,29 @@ async def close_and_finalize(
                 unregister(trade_id)
                 _close_locks.pop(trade_id, None)
 
-                logger.info("Closed %s %s for %s @ %.2f PnL=%.2f (reason=%s)",
-                            trade.side, asset, wallet, exit_price, pnl_usd, reason)
+                logger.info(
+                    "Closed %s %s for %s @ %.2f gross=%.2f fees=%.2f net=%.2f (reason=%s)",
+                    trade.side, asset, wallet, exit_price,
+                    gross_pnl, fees_usd, pnl_usd, 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.
+                # Without this, the trade is "closed" in DB but position still open on HL.
+                try:
+                    async with AsyncSessionLocal() as rb_db:
+                        await rb_db.execute(
+                            update(BotTrade)
+                            .where(BotTrade.id == trade_id)
+                            .values(closed_at=None)
+                        )
+                        await rb_db.commit()
+                        logger.info("Rolled back closed_at for trade %d — will retry on recovery", trade_id)
+                except Exception as rb_exc:
+                    logger.error("Failed to rollback closed_at for trade %d: %s", trade_id, rb_exc)
+                finally:
+                    _close_locks.pop(trade_id, None)
 
 
 async def _close_after_hold(

app/services/hyperliquid.py

@@ -111,7 +111,7 @@ class HyperliquidTrader:
             return positions
         except Exception as exc:
             logger.error("get_open_positions error: %s", exc)
-            return []
+            raise
 
     async def set_leverage(self, coin: str) -> None:
         """Set isolated leverage for the given coin."""
@@ -180,14 +180,30 @@ class HyperliquidTrader:
         )
 
         statuses = result.get("response", {}).get("data", {}).get("statuses", [{}])
-        filled = statuses[0] if statuses else {}
-        fill_price = float(
-            (filled.get("filled") or {}).get("avgPx", mid) or mid
-        )
-        order_id = str((filled.get("resting") or {}).get("oid", ""))
+        status = statuses[0] if statuses else {}
 
-        logger.info("Opened %s %s @ %.2f (order_id=%s)", side, coin, fill_price, order_id)
-        return {"order_id": order_id, "fill_price": fill_price, "size_coins": size_coins}
+        # "error" key present → HL rejected the order entirely
+        if "error" in status:
+            raise ValueError(f"HL order rejected: {status['error']}")
+
+        filled_info = status.get("filled") or {}
+        total_sz = float(filled_info.get("totalSz", 0) or 0)
+        if total_sz <= 0:
+            # IOC returned with zero fill — no position was opened
+            raise ValueError(
+                f"IOC order for {coin} returned 0 fill "
+                f"(status={status}). No position opened."
+            )
+
+        fill_price = float(filled_info.get("avgPx", mid) or mid)
+
+        # oid lives under "filled" for IOC; "resting" is fallback for unexpected GTC
+        oid_src = filled_info or status.get("resting") or {}
+        order_id = str(oid_src.get("oid", "")) if oid_src else ""
+
+        logger.info("Opened %s %s @ %.2f size=%.5f (order_id=%s)",
+                    side, coin, fill_price, total_sz, order_id)
+        return {"order_id": order_id, "fill_price": fill_price, "size_coins": total_sz}
 
     async def close_position(self, asset: str) -> dict:
         """
@@ -201,8 +217,9 @@ class HyperliquidTrader:
         positions = await self.get_open_positions()
         target = next((p for p in positions if p.get("coin") == coin), None)
         if target is None:
-            logger.warning("No open %s position to close", coin)
-            return {"fill_price": 0.0}
+            logger.warning("No open %s position found on HL — already closed externally?", coin)
+            # Return sentinel so callers can detect "nothing to close" vs "closed @ price"
+            return {"fill_price": None, "already_closed": True}
 
         szi = float(target["szi"])
         is_buy = szi < 0          # closing a short → buy back
@@ -235,10 +252,21 @@ class HyperliquidTrader:
         )
 
         statuses = result.get("response", {}).get("data", {}).get("statuses", [{}])
-        filled = statuses[0] if statuses else {}
-        fill_price = float(
-            (filled.get("filled") or {}).get("avgPx", mid) or mid
-        )
+        status = statuses[0] if statuses else {}
+        filled_info = status.get("filled") or {}
+        total_sz = float(filled_info.get("totalSz", 0) or 0)
 
-        logger.info("Closed %s position @ %.2f", coin, fill_price)
-        return {"fill_price": fill_price}
+        if total_sz <= 0:
+            # IOC close got 0 fill — position may have already been closed externally.
+            # Re-check to be sure before giving up.
+            positions_after = await self.get_open_positions()
+            still_open = next((p for p in positions_after if p.get("coin") == coin), None)
+            if still_open is None:
+                logger.info("Close IOC got 0 fill but position is gone — treated as closed", )
+                return {"fill_price": mid, "already_closed": False}
+            logger.error("Close IOC got 0 fill and position still open for %s", coin)
+            raise ValueError(f"Failed to close {coin} position: IOC returned 0 fill")
+
+        fill_price = float(filled_info.get("avgPx", mid) or mid)
+        logger.info("Closed %s position @ %.2f (size=%.5f)", coin, fill_price, total_sz)
+        return {"fill_price": fill_price, "already_closed": False}

app/services/price_impact_monitor.py

@@ -0,0 +1,204 @@
+"""
+Rolling price-impact tracker for Trump posts.
+
+For each newly-saved relevant post, we track the peak favorable move
+(max high for BUY signals, max low for SHORT signals) in three windows:
+5 m, 15 m, and 1 h.
+
+Rules:
+  - While the window is OPEN  → live rolling peak, updated on every price tick
+  - When the window CLOSES    → final peak is written to DB, window is sealed
+  - When all three windows close → post is unregistered to free memory
+
+The result is that:
+  • A brand-new post immediately shows the running peak since publication.
+  • A post 20 minutes old shows final m5, final m15, and live 1h peak.
+  • A post 2 hours old shows final m5 / m15 / m1h from DB.
+
+Integration points:
+  binance.py   → call on_price_tick(asset, high, low, close) each candle
+  truth_social.py → call register_post(...) after flush, before commit
+  posts.py     → call get_live_impact(post_id) to overlay live values
+"""
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+# Window durations in seconds
+WINDOWS = {"m5": 5 * 60, "m15": 15 * 60, "m1h": 60 * 60}
+
+
+@dataclass
+class TrackedPost:
+    post_id: int
+    asset: str
+    signal: Optional[str]     # "buy" | "short" | "sell" | "hold" | None
+    entry_price: float        # price at post time
+    published_at: datetime    # naive UTC
+
+    # Running peaks — signed % relative to entry_price
+    # For BUY:   positive = price went up   (we want max)
+    # For SHORT: positive = price went down (we want max of -pct)
+    peak_m5:  Optional[float] = None
+    peak_m15: Optional[float] = None
+    peak_m1h: Optional[float] = None
+
+    # True once the window has expired and the value is finalised in DB
+    done_m5:  bool = False
+    done_m15: bool = False
+    done_m1h: bool = False
+
+
+# post_id → TrackedPost
+_tracked: Dict[int, TrackedPost] = {}
+
+# Strong refs to DB-write tasks so GC doesn't collect them mid-flight
+_background_tasks: set = set()
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Public API
+# ─────────────────────────────────────────────────────────────────────────────
+
+def register_post(
+    post_id: int,
+    asset: str,
+    signal: Optional[str],
+    entry_price: float,
+    published_at: datetime,
+) -> None:
+    """Call immediately after a relevant post is flushed to DB."""
+    if entry_price <= 0:
+        return
+    if not asset:
+        return
+    _tracked[post_id] = TrackedPost(
+        post_id=post_id,
+        asset=asset.upper(),
+        signal=signal,
+        entry_price=entry_price,
+        published_at=published_at.replace(tzinfo=None),  # store as naive UTC
+    )
+    logger.info("PriceImpact: tracking post %d (%s %s @ %.2f)",
+                post_id, signal, asset, entry_price)
+
+
+def unregister(post_id: int) -> None:
+    _tracked.pop(post_id, None)
+
+
+def get_live_impact(post_id: int) -> Optional[dict]:
+    """
+    Return the current live peaks for a post if it is still being tracked.
+    Returns None if the post has never been registered or all windows are done.
+    The dict keys match the Post model columns:
+      price_impact_m5, price_impact_m15, price_impact_m1h
+    Only keys with open (not-yet-sealed) windows are included — callers
+    should overlay these on top of DB values.
+    """
+    tp = _tracked.get(post_id)
+    if tp is None:
+        return None
+    out: dict = {}
+    if not tp.done_m5:
+        out["price_impact_m5"] = tp.peak_m5
+    if not tp.done_m15:
+        out["price_impact_m15"] = tp.peak_m15
+    if not tp.done_m1h:
+        out["price_impact_m1h"] = tp.peak_m1h
+    return out if out else None
+
+
+def on_price_tick(asset: str, high: float, low: float, close: float) -> None:
+    """
+    Called by binance.py on every 1-minute candle close.
+    Updates running peaks and fires DB-write tasks for expired windows.
+    """
+    asset = asset.upper()
+    if not _tracked:
+        return
+
+    now_naive = datetime.now(timezone.utc).replace(tzinfo=None)
+
+    for post_id, tp in list(_tracked.items()):
+        if tp.asset != asset:
+            continue
+
+        age_s = (now_naive - tp.published_at).total_seconds()
+
+        # Pick the extreme price for this signal direction
+        # BUY  → we care about how high price went (use candle high)
+        # SHORT/SELL → we care about how low price went (use candle low)
+        is_long = tp.signal in ("buy",)
+        extreme = high if is_long else low
+
+        # signed % gain in the signal's direction
+        if tp.entry_price > 0:
+            raw_pct = (extreme - tp.entry_price) / tp.entry_price * 100
+            signed_pct = raw_pct if is_long else -raw_pct
+        else:
+            signed_pct = None
+
+        # Update each open window
+        for win, duration in WINDOWS.items():
+            done_attr = f"done_{win}"
+            peak_attr = f"peak_{win}"
+            if getattr(tp, done_attr):
+                continue  # already sealed
+
+            if signed_pct is not None:
+                cur = getattr(tp, peak_attr)
+                if cur is None or signed_pct > cur:
+                    setattr(tp, peak_attr, round(signed_pct, 4))
+
+            # Has the window expired?
+            if age_s >= duration:
+                setattr(tp, done_attr, True)
+                final_peak = getattr(tp, peak_attr)
+                t = asyncio.create_task(_write_window_to_db(post_id, win, final_peak))
+                _background_tasks.add(t)
+                t.add_done_callback(_background_tasks.discard)
+                logger.debug("PriceImpact: post %d window %s closed → peak=%.4f%%",
+                             post_id, win, final_peak or 0)
+
+        # All windows done → free memory
+        if tp.done_m5 and tp.done_m15 and tp.done_m1h:
+            unregister(post_id)
+            logger.info("PriceImpact: post %d fully tracked, unregistered", post_id)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# DB write helpers
+# ─────────────────────────────────────────────────────────────────────────────
+
+_WINDOW_COLUMN = {
+    "m5":  "price_impact_m5",
+    "m15": "price_impact_m15",
+    "m1h": "price_impact_m1h",
+}
+
+
+async def _write_window_to_db(post_id: int, window: str, value: Optional[float]) -> None:
+    """Write the final peak for a single window to the DB."""
+    from sqlalchemy import update
+    from app.database import AsyncSessionLocal
+    from app.models import Post
+
+    col = _WINDOW_COLUMN[window]
+    try:
+        async with AsyncSessionLocal() as db:
+            await db.execute(
+                update(Post)
+                .where(Post.id == post_id)
+                .values({col: value})
+            )
+            await db.commit()
+        logger.info("PriceImpact: wrote post %d %s=%.4f%% to DB",
+                    post_id, window, value or 0)
+    except Exception as exc:
+        logger.error("PriceImpact: failed to write post %d %s: %s", post_id, window, exc)

app/services/recovery.py

@@ -18,7 +18,7 @@ logger = logging.getLogger(__name__)
 
 async def rehydrate_open_trades() -> None:
     # Imported locally to avoid circular imports at module load
-    from app.services.bot_engine import MAX_HOLD_SECONDS, _close_after_hold, close_and_finalize
+    from app.services.bot_engine import MAX_HOLD_SECONDS, close_and_finalize
     from app.services.crypto import decrypt_api_key
     from app.services.tp_sl_monitor import register_trade
 
@@ -49,12 +49,16 @@ async def rehydrate_open_trades() -> None:
                 logger.error("Cannot decrypt key for trade %d: %s", t.id, exc)
                 continue
 
+            # Use the leverage snapshot from the trade row (stamped at open time).
+            # Fall back to current Subscription only for legacy rows (pre-migration 005).
+            trade_leverage = t.leverage if t.leverage is not None else sub.leverage
+
             # Re-register TP/SL watcher
             register_trade(
                 trade_id=t.id,
                 wallet=t.wallet_address,
                 api_key=api_key,
-                leverage=sub.leverage,
+                leverage=trade_leverage,
                 asset=t.asset,
                 side=t.side,
                 entry_price=t.entry_price,
@@ -67,23 +71,29 @@ async def rehydrate_open_trades() -> None:
             elapsed = (now - opened_aware).total_seconds()
             remaining = MAX_HOLD_SECONDS - elapsed
 
+            from app.services.bot_engine import _background_tasks
+
             if remaining <= 0:
                 logger.info("Trade %d past max-hold on startup — closing now", t.id)
-                asyncio.create_task(
+                task = asyncio.create_task(
                     close_and_finalize(
-                        trade_id=t.id, api_key=api_key, leverage=sub.leverage,
+                        trade_id=t.id, api_key=api_key, leverage=trade_leverage,
                         asset=t.asset, wallet=t.wallet_address, reason="max_hold_recovery",
                     )
                 )
+                _background_tasks.add(task)
+                task.add_done_callback(_background_tasks.discard)
             else:
-                async def _delayed_close(trade_id=t.id, key=api_key, lev=sub.leverage,
+                async def _delayed_close(trade_id=t.id, key=api_key, lev=trade_leverage,
                                          asset=t.asset, wallet=t.wallet_address, delay=remaining):
                     await asyncio.sleep(delay)
                     await close_and_finalize(
                         trade_id=trade_id, api_key=key, leverage=lev,
                         asset=asset, wallet=wallet, reason="max_hold",
                     )
-                asyncio.create_task(_delayed_close())
+                task = asyncio.create_task(_delayed_close())
+                _background_tasks.add(task)
+                task.add_done_callback(_background_tasks.discard)
 
         await db.commit()
         logger.info("Rehydrated %d open trades.", len(open_trades))

app/services/tp_sl_monitor.py

@@ -33,6 +33,9 @@ class WatchedTrade:
 # trade_id -> WatchedTrade
 _watched: Dict[int, WatchedTrade] = {}
 
+# Strong references to fire-close tasks to prevent GC before completion
+_background_tasks: set = set()
+
 
 def register_trade(
     trade_id: int,
@@ -79,7 +82,9 @@ def on_price_tick(asset: str, price: float) -> None:
 
     for wt, reason in triggered:
         unregister(wt.trade_id)
-        asyncio.create_task(_fire_close(wt, reason))
+        task = asyncio.create_task(_fire_close(wt, reason))
+        _background_tasks.add(task)
+        task.add_done_callback(_background_tasks.discard)
 
 
 async def _fire_close(wt: WatchedTrade, reason: str) -> None: