k
4ffcb442fe
- 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>
100 lines
4.0 KiB
Python
100 lines
4.0 KiB
Python
"""
|
|
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, 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 MAX_HOLD_SECONDS, close_and_finalize
|
|
from app.services.crypto import decrypt_api_key
|
|
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 TP/SL watcher
|
|
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=sub.take_profit_pct,
|
|
stop_loss_pct=sub.stop_loss_pct,
|
|
)
|
|
|
|
# Compute remaining hold; if already exceeded, close immediately
|
|
opened_aware = t.opened_at.replace(tzinfo=timezone.utc)
|
|
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)
|
|
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))
|