trumpsignal-backend/app/services/recovery.py

"""
Startup rehydration: re-register TP/SL + max-hold for every open trade in DB.

Called once from lifespan() after tables are ensured. Without this, any deploy
or crash silently drops TP/SL protection on live positions.
"""
import asyncio
import logging
from datetime import datetime, timezone

from sqlalchemy import select

from app.database import AsyncSessionLocal
from app.models import BotTrade, Post, Subscription

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 close_and_finalize
    from app.services.crypto import decrypt_api_key
    from app.services.signal_categories import (
        get_stop_ladder as _get_stop_ladder,
        sys2_derisk_ladder as _sys2_derisk_ladder,
        sys2_addon_ladder as _sys2_addon_ladder,
        sys2_peak_trail as _sys2_peak_trail,
    )
    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)))
        open_trades = result.scalars().all()

        if not open_trades:
            logger.info("Rehydration: no open trades.")
            return

        now = datetime.now(timezone.utc)
        for t in open_trades:
            sub_res = await db.execute(
                select(Subscription).where(Subscription.wallet_address == t.wallet_address)
            )
            sub = sub_res.scalar_one_or_none()
            if sub is None or not sub.hl_api_key:
                logger.warning(
                    "Open trade %d has no subscription/key — marking as closed(abandoned)", t.id
                )
                t.closed_at = now.replace(tzinfo=None)
                continue

            try:
                api_key = decrypt_api_key(sub.hl_api_key)
            except Exception as exc:
                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 from the trade's FROZEN exit profile (eff_* columns),
            # NOT the live Subscription. The trade may be a 90-day System-2
            # reversal; using sub.* would rehydrate it with the user's Trump
            # stop (1.5%) and 7-day max-hold — silently rewriting its risk on
            # every restart. eff_* is stamped at open and never changes.
            #
            # Legacy rows (opened before this migration) have NULL eff_* —
            # fall back to the Subscription so they still get *some* watcher.
            eff_sl   = t.eff_stop_loss_pct         if t.eff_stop_loss_pct is not None else sub.stop_loss_pct
            eff_tp   = t.eff_take_profit_pct        # may legitimately be None (sys2 pure-trail)
            eff_tr   = t.eff_trailing_stop_pct      if t.eff_trailing_stop_pct is not None else sub.trailing_stop_pct
            eff_tra  = t.eff_trailing_activate_pct  if t.eff_trailing_activate_pct is not None else sub.trailing_activate_at_pct
            eff_mh   = t.eff_max_hold_hours         if t.eff_max_hold_hours is not None else (sub.max_hold_hours or 168)
            # NOTE: peak_gain_pct is now RESTORED from the row (throttled
            # persistence) so a pyramided / in-profit System-2 trade keeps its
            # regime across restarts. min_hold still resets on rehydrate, but
            # it only matters in the first 30 min of a Trump trade; a restart
            # inside that window is rare and the downside (TP can fire early)
            # is bounded.
            post_res = await db.execute(
                select(Post).where(Post.id == t.trigger_post_id)
            )
            trigger_post = post_res.scalar_one_or_none()
            register_trade(
                trade_id=t.id,
                wallet=t.wallet_address,
                api_key=api_key,
                leverage=trade_leverage,
                asset=t.asset,
                side=t.side,
                entry_price=t.entry_price,
                take_profit_pct=eff_tp,
                stop_loss_pct=eff_sl,
                trailing_stop_pct=eff_tr,
                trailing_activate_at_pct=eff_tra,
                invalidation=t.eff_invalidation,
                invalidation_price=(
                    t.eff_invalidation_price
                    if t.eff_invalidation_price is not None
                    else (trigger_post.invalidation_price if trigger_post else None)
                ),
                min_hold_until_ts=t.eff_min_hold_until_ts,
                stop_ladder=_get_stop_ladder(
                    trigger_post.category if trigger_post else None
                ),
                # Restore staged de-risk: rebuild the ladder from the trade's
                # FROZEN leverage and skip steps already executed before the
                # restart (derisk_steps_done is persisted on the row).
                # Restore staged de-risk/pyramid/peak-trail with the trade's
                # FROZEN risk mode + leverage, skipping steps already executed
                # before the restart (persisted on the row).
                derisk_ladder=(
                    _sys2_derisk_ladder(t.leverage or 1, t.sys2_mode)
                    if _get_stop_ladder(trigger_post.category if trigger_post else None)
                    else None
                ),
                derisk_done=(t.derisk_steps_done or 0),
                addon_ladder=(
                    _sys2_addon_ladder(t.sys2_mode)
                    if _get_stop_ladder(trigger_post.category if trigger_post else None)
                    else None
                ),
                addon_done=(t.addon_steps_done or 0),
                # Restore the monotonic peak so a pyramided / in-profit trade
                # doesn't fall back to the underwater de-risk regime on restart.
                initial_peak=(t.peak_gain_pct or 0.0),
                peak_trail=(
                    _sys2_peak_trail(t.sys2_mode)
                    if _get_stop_ladder(trigger_post.category if trigger_post else None)
                    else None
                ),
                grow_mode=bool(getattr(t, "grow_mode", False)),
            )

            # Remaining hold from the trade's FROZEN max_hold (e.g. 2160h for
            # an sma_reclaim), not the user's Trump setting.
            opened_aware = t.opened_at.replace(tzinfo=timezone.utc)
            elapsed = (now - opened_aware).total_seconds()
            max_hold_seconds = int(eff_mh) * 3600
            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)
                task = asyncio.create_task(
                    close_and_finalize(
                        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=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",
                    )
                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))