""" 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, timedelta 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) # No global confidence floor — users pick their own 0–100 threshold via settings. # 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. # 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) if lock is None: lock = asyncio.Lock() _close_locks[trade_id] = 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. `db` is used only to fetch the subscriber list; each subscriber gets its own session. """ if not post.relevant: return # v5: "sell" is no longer emitted by the AI. Old DB rows might still have # it; treat as hold (do not trade) — see analysis.py docstring for why # the previous "sell→short" treatment was a semantic bug. if post.signal not in ('buy', 'short'): logger.info("Post %d skipped: signal=%s is not actionable", post.id, post.signal) return # v5: route the trade to the AI-decided perp. target_asset can be any # Hyperliquid perp ticker (BTC/ETH/SOL/TRUMP/...) — analysis.py already # validated against HL_PERPS and resolved chain fallbacks before this # field was set, so by the time we get here it's safe to trade directly. # Fallbacks for older / null rows: legacy price_impact_asset, then BTC. asset = post.target_asset or post.price_impact_asset or 'BTC' side = 'long' if post.signal == 'buy' else 'short' logger.info( "Routing post %d → trade %s/%s (category=%s, expected_move=%.2f%%)", post.id, asset, side, post.category, post.expected_move_pct or 0, ) 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 — manual-open + auto-manage model. # # We do NOT auto-open System-2 positions any more. The strategy is # day-K level (AHR999 / 200WMA / Pi Cycle Bottom take weeks to play # out) so a 24h entry delay is irrelevant — but the auto-open path # carried real risk: leverage-clip math, daily-budget split, # concurrency caps, sys2-CB pre-open gate, paper branches, key # handling. All of that is execution surface for ~0 alpha vs giving # the user the alert + letting them open on HL themselves. # # The valuable part of System-2 — the multi-month exit management # (5-rung stop ladder + downside de-risk + pyramid + peak-trail) — # is preserved in tp_sl_monitor and runs against positions the user # adopts via /api/positions/adopt (or the Telegram /adopt command). # # Here we just record the signal (already done by signals.ingest → # Post row) and let the Telegram fan-out send the alert with the # "open on HL and /adopt" CTA. No subscriber iteration, no open. logger.info( "%s [%s/%s] signal recorded — manual-open + /adopt model " "(no auto-open). Conf=%d", system2_display_name(), post.source, post.category, post.ai_confidence or 0, ) return 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() 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, 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 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 # Required-setup guard. System 1 (Trump) needs take-profit and stop-loss. # System 2 (reversal) supplies its own stop + trailing from the category # profile, so only the two Trump exit fields are user-required. # daily_budget_usd is OPTIONAL — null means no cap, not misconfigured. if not sub.get("_is_system_2"): required = ("take_profit_pct", "stop_loss_pct") 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 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, confidence_floor) 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"]) except Exception as exc: 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. # 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, _P.source) .outerjoin(_P, _P.id == BotTrade.trigger_post_id) .where( BotTrade.wallet_address == wallet, BotTrade.opened_at >= start_of_day, ) ) # 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), ) ) 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: # ── 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 result = await trader.open_position(asset, side, sized_position_usd) entry_price = result.get('fill_price', 0.0) # HL may clip leverage below the requested value (e.g. meme # asset capped at 3× when the user asked for 10×). Trust HL's # actual value — anything downstream that uses leverage for # risk math (sys2 protective stop, de-risk ladder, BotTrade # row) MUST use this, not sub["leverage"]. # paper-mode branch has no HL leverage → fall back to req. effective_leverage = int(result.get('effective_leverage') or sub["leverage"]) if effective_leverage != sub["leverage"]: logger.warning( "Sub %s: HL clipped leverage %d → %d for %s — " "recomputing sys2 protective stop / derisk ladder", wallet, sub["leverage"], effective_leverage, asset, ) # Sys2: rebuild protective stop + derisk ladder against the # ACTUAL leverage (so the full-close rung stays inside the # real HL liquidation line). Sys1 doesn't depend on leverage # for its stop math, but we still record the true value so # BotTrade.leverage reflects what HL actually applied. sub["leverage"] = effective_leverage if sys2: from app.services.signal_categories import ( sys2_protective_stop_pct as _sps, ) sub["stop_loss_pct"] = _sps(effective_leverage) # 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 ) # sub["stop_loss_pct"] / sub["leverage"] were just rewritten # above to match the HL-clipped effective leverage when sys2, # so freezing them into eff here is correct. 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, ) # Resolve sys2 mode BEFORE constructing the BotTrade row. # Python made this a local-name; reading it inside the # constructor before the later assignment used to throw # UnboundLocalError on every sys2 fire, leaving the HL # position open with NO DB record and NO watchdog. Critical. _sys2_mode = sub.get("_sys2_mode", "standard") if sys2 else None 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=sized_position_usd, base_size_usd=sized_position_usd, # immutable; pyramiding grows size_usd sys2_mode=_sys2_mode, 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 @ %.4f size=%.2f (trade_id=%d)", side, asset, wallet, entry_price, sized_position_usd, trade.id) # Register with the watchdog using the FROZEN profile. # _sys2_mode was resolved before the BotTrade constructor above; # reuse the same value here so de-risk/add-on ladders match # the row that was just written. from app.services.tp_sl_monitor import register_trade _derisk = None _addon = None _peak_trail = None if sys2: _mode_for_ladders = _sys2_mode or "standard" from app.services.signal_categories import ( sys2_derisk_ladder, sys2_addon_ladder, sys2_peak_trail, ) _derisk = sys2_derisk_ladder(sub["leverage"], _mode_for_ladders) _addon = sys2_addon_ladder(_mode_for_ladders) _peak_trail = sys2_peak_trail(_mode_for_ladders) 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=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, 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.released_at is not None: # User took back manual control between tick capture and # execution — do NOT partially reduce a position they're # now driving themselves. return True 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.released_at is not None: # User took back manual control — never pyramid into a # position they're now managing themselves. 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, leverage: int, asset: str, wallet: str, reason: str = "max_hold", force: bool = False, ) -> 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. `force=True` overrides the released_at guard — only the explicit user close path (manual_close API) should pass it. Automated paths (tp/sl monitor, max-hold, time-stop, reconciler) must respect release. """ 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. For automated # callers, ALSO require released_at IS NULL — a price tick # captured BEFORE release_management ran would otherwise be # able to close an HL position the user just took back # control of. The manual user-close endpoint sets force=True # because the user explicitly wants the position gone # regardless of release state. claim_stmt = ( update(BotTrade) .where(BotTrade.id == trade_id) .where(BotTrade.closed_at.is_(None)) ) if not force: claim_stmt = claim_stmt.where(BotTrade.released_at.is_(None)) claim = await db.execute(claim_stmt.values(closed_at=now_naive)) await db.commit() if claim.rowcount == 0: # Either someone else closed it, or (automated path) the # user released it between the price-tick capture and # this close attempt. Either way: bail without touching HL. return # Reload the row we just claimed result = await db.execute(select(BotTrade).where(BotTrade.id == trade_id)) trade = result.scalar_one() # 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 # ── 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 closed; pnl=%s (banked de-risk only).", trade_id, asset, round(banked, 2) if banked else None, ) trade.exit_price = 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() ) 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 # size_usd is the NOTIONAL value — leverage affects margin, not PnL. # 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 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 gross=%.2f fees=%.2f net=%.2f (reason=%s)", trade.side, asset, wallet, exit_price, 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. # 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( trade_id: int, api_key: str, leverage: int, asset: str, wallet: str, max_hold_seconds: int, ) -> None: """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", )