""" Trading decision loop. Called by the Truth Social scraper after each new post is analyzed and saved. Iterates all active subscribers and executes trades on their behalf. """ import asyncio import logging from datetime import datetime, timezone from typing import Dict from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.database import AsyncSessionLocal from app.models import BotTrade, Post, Subscription from app.services.crypto import decrypt_api_key from app.services.hyperliquid import HyperliquidTrader 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 MAX_HOLD_SECONDS = 3600 # force-close after 1 hour (shared across users) # 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] = {} def _lock_for(trade_id: int) -> asyncio.Lock: lock = _close_locks.get(trade_id) if lock is None: lock = asyncio.Lock() _close_locks[trade_id] = lock return lock async def process_post(post: Post, db: AsyncSession) -> None: """ Entry point called by the scraper after a new post is saved. `db` is used only to fetch the subscriber list; each subscriber gets its own session. """ 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'): logger.info("Post %d skipped: signal=%s is not actionable", post.id, post.signal) return asset = post.price_impact_asset or 'BTC' 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) result = await db.execute(select(Subscription).where(Subscription.active == True)) # noqa: E712 subscribers = result.scalars().all() if not subscribers: logger.info("No active subscribers, skipping trade execution") return # Snapshot primitive fields so tasks don't share the ORM object / session. subs_snapshot = [ dict( wallet=s.wallet_address, hl_api_key=s.hl_api_key, leverage=s.leverage, position_size_usd=s.position_size_usd, take_profit_pct=s.take_profit_pct, stop_loss_pct=s.stop_loss_pct, min_confidence=s.min_confidence, ) for s in subscribers ] post_id = post.id post_confidence = post.ai_confidence or 0 tasks = [ _execute_for_subscriber(sub, post_id, post_confidence, asset, side) for sub in subs_snapshot ] await asyncio.gather(*tasks, return_exceptions=True) async def _execute_for_subscriber( sub: dict, post_id: int, post_confidence: int, asset: str, side: str, ) -> None: wallet = sub["wallet"] if not sub["hl_api_key"]: logger.warning("Subscriber %s has no HL API key, skipping", wallet) 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 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) return result = await trader.open_position(asset, side, sub["position_size_usd"]) entry_price = result.get('fill_price', 0.0) 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) logger.info("Opened %s %s for %s @ %.2f (trade_id=%d)", side, asset, wallet, entry_price, trade.id) 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"], ) asyncio.create_task(_close_after_hold( trade.id, api_key, sub["leverage"], asset, wallet )) except Exception as e: logger.error("Trade execution failed for %s: %s", wallet, e) async def close_and_finalize( trade_id: int, api_key: str, leverage: int, asset: str, wallet: str, reason: str = "max_hold", ) -> None: """ Close a live HL position and write exit_price/pnl to BotTrade. Race-safe: a conditional UPDATE `closed_at = now WHERE closed_at IS NULL` lets exactly one caller win; losers return silently. Also wraps in a per-trade asyncio.Lock to prevent us even *attempting* the HL API call twice. """ async with _lock_for(trade_id): async with AsyncSessionLocal() as db: try: now_naive = datetime.now(timezone.utc).replace(tzinfo=None) # Atomic claim: only one caller sets closed_at. claim = await db.execute( update(BotTrade) .where(BotTrade.id == trade_id) .where(BotTrade.closed_at.is_(None)) .values(closed_at=now_naive) ) await db.commit() if claim.rowcount == 0: return # someone else closed it # Reload the row we just claimed 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 trader = HyperliquidTrader( api_private_key=api_key, account_address=wallet, leverage=leverage, mainnet=settings.hl_mainnet, ) close_result = await trader.close_position(asset) exit_price = close_result.get('fill_price', 0.0) 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 trade.exit_price = exit_price trade.pnl_usd = round(pnl_usd, 2) trade.hold_seconds = hold_secs # closed_at already set by the atomic UPDATE await db.commit() # Clean up TP/SL + lock to avoid leaking memory from app.services.tp_sl_monitor import unregister 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) except Exception as e: logger.error("Failed to close trade %d: %s", trade_id, e) async def _close_after_hold( trade_id: int, api_key: str, leverage: int, asset: str, wallet: str ) -> None: await asyncio.sleep(MAX_HOLD_SECONDS) await close_and_finalize(trade_id, api_key, leverage, asset, wallet, reason="max_hold")