""" Agente Monitor de Cierres de Leonex. Para cada posicion abierta registrada en `open_trades`, compara el precio actual contra los parametros Triple Barrier (take-profit, stop-loss, timeout) calculados al abrir el trade. Cuando una barrera se cumple, envia la orden SELL correspondiente a Alpaca y marca el trade como cerrado. Es el "agente que sale del mercado" — el complemento del executor (que entra). Sin este, las posiciones abiertas se quedan abiertas para siempre. Modos: python agents/agente_close_monitor.py # solo lectura python agents/agente_close_monitor.py --dry-run # imprime que cerraria python agents/agente_close_monitor.py --execute --confirm # cierra de verdad Idealmente se ejecuta despues de agente_executor (orden de apertura) o por cron varias veces al dia mientras hay mercado. """ from __future__ import annotations import argparse import json import logging import sqlite3 import sys from dataclasses import asdict, dataclass, field from datetime import datetime try: from datetime import UTC # Python 3.11+ except ImportError: # Python 3.10 from datetime import timezone UTC = timezone.utc from pathlib import Path from typing import Optional sys.path.insert(0, str(Path(__file__).resolve().parent)) from alpaca_client import ( # noqa: E402 fetch_snapshot, get_latest_quote, submit_market_order, ) PROJECT_ROOT = Path(__file__).resolve().parents[1] DATA_DIR = PROJECT_ROOT / "data" LOGS_DIR = PROJECT_ROOT / "logs" DASHBOARD_DATA_DIR = PROJECT_ROOT / "dashboard" / "data" DB_PATH = DATA_DIR / "Leonex.sqlite" REPORT_PATH = DASHBOARD_DATA_DIR / "close_monitor_report.json" # Salida por REGIMEN: cierra una posicion EN BENEFICIO cuando el precio pierde # su media de tendencia (el regimen que justificaba la operacion se da la # vuelta). Mismo concepto que la 4a barrera del backtest, en vivo: se asegura # la ganancia en vez de esperar al TP y comerse la reversion. REGIME_SMA_WINDOW = 50 # Desviacion maxima tolerada entre el quote (bid/ask) y el current_price que # reporta la posicion de Alpaca. Por encima, el quote se considera rancio o # con spread absurdo (feed IEX gratuito) y se usa el precio de la posicion. # Evita "SL fantasma" por mids desviados. Override: LEONEX_QUOTE_SANITY_DEV. import os as _os try: QUOTE_SANITY_MAX_DEV = float( _os.environ.get("LEONEX_QUOTE_SANITY_DEV", "0.02")) except ValueError: QUOTE_SANITY_MAX_DEV = 0.02 # Minutos de vida minima de un trade antes de que el close_monitor pueda # cerrarlo por TP/SL/regimen. Evita el churn compra→venta en segundos dentro # del mismo ciclo del bridge. Override: LEONEX_CLOSE_GRACE_MIN (0 desactiva). try: CLOSE_GRACE_MIN = float(_os.environ.get("LEONEX_CLOSE_GRACE_MIN", "45")) except ValueError: CLOSE_GRACE_MIN = 45.0 # ───────────────────────────────────────────────────────────────────────────── # Trailing TP/SL dinamico (Kaufman/Carver: let winners run + ratchet up stops). # Solo se aplica a posiciones con POTENCIAL (filtro tecnico): # LONG : regimen alcista AND rsi0 AND ret>0 # SHORT: regimen bajista AND rsi>TRAILING_RSI_MIN_SHORT AND mom20<0 AND ret>0 # Cuando el filtro pasa: # - SL_new = max(SL_actual, current_price - TRAILING_SL_ATR_MULT * ATR) # Esto cierra parte del beneficio sin matar la operacion. Cuando current # ya esta muy por encima del entry, el SL queda BREAKEVEN o mejor. # - TP_new = max(TP_actual, current_price + TRAILING_TP_ATR_MULT * ATR) # SOLO si current >= TRAILING_TP_EXTENSION_THRESHOLD * TP_actual (es decir, # ya ha viajado >=70% del camino al TP). Esto extiende el objetivo en # winners fuertes en vez de cerrar prematuramente. # Solo extiende (max), nunca reduce. Cuando la posicion va MAL o el filtro # tecnico falla, los TP/SL originales se mantienen intactos. TRAILING_ENABLED = True TRAILING_RSI_MAX_LONG = 80.0 TRAILING_RSI_MIN_SHORT = 20.0 TRAILING_ATR_WINDOW = 14 TRAILING_SL_ATR_MULT = 1.0 TRAILING_TP_ATR_MULT = 2.0 TRAILING_TP_EXTENSION_THRESHOLD = 0.70 # current >= 70% del TP actual @dataclass class OpenTradeView: id: int ticker: str side: int qty: float entry_date: str entry_price: float sigma_entry: float pt_pct: float sl_pct: float vert_barrier_days: int current_price: float current_return_pct: float days_open: int distance_to_tp: float # cuanto falta (positivo) para tocar TP distance_to_sl: float # cuanto falta (positivo) para tocar SL days_to_timeout: int proposed_action: str # 'hold' | 'close_tp' | 'close_sl' | 'close_timeout' proposed_notes: str = "" # Campos trailing (solo se rellenan si TRAILING_ENABLED y la posicion # tiene potencial). pt_pct y sl_pct arriba YA reflejan los valores actuales # (que pueden haber sido ratchet up de runs anteriores); estos campos # capturan SI se ajusto en ESTE run y desde donde. trailing_potential: bool = False trailing_applied_now: bool = False trailing_prev_pt_pct: float = 0.0 trailing_prev_sl_pct: float = 0.0 trailing_note: str = "" @dataclass class CloseAction: ticker: str side: int qty: float reason: str # 'tp' | 'sl' | 'timeout' entry_price: float exit_price: float pnl_pct: float days_open: int open_trade_id: int status: str = "planned" # planned | submitted | rejected | skipped alpaca_sell_order_id: Optional[str] = None notes: str = "" @dataclass class CloseMonitorReport: generated_at: str mode: str # read_only | dry_run | execute n_open_trades: int n_proposed_closes: int open_trades_view: list[OpenTradeView] = field(default_factory=list) close_actions: list[CloseAction] = field(default_factory=list) def _parse_iso(s: str) -> Optional[datetime]: if not s: return None try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except Exception: return None def _now_utc() -> datetime: return datetime.now(UTC) def evaluate_open_trade(row: dict, current_price: float, trend_sma: Optional[float] = None) -> OpenTradeView: """Decide hold / close_tp / close_sl / close_regime / close_timeout para un trade abierto. `trend_sma` es la media de tendencia del activo; si la posicion esta en beneficio y el precio cruza esa media en contra, se cierra por REGIMEN (asegura la ganancia).""" entry = float(row["entry_price"]) side = int(row["side"]) pt = float(row["pt_pct"]) sl = float(row["sl_pct"]) vert = int(row["vert_barrier_days"]) entry_dt = _parse_iso(row["entry_date"]) or _now_utc() now = _now_utc() days_open = max(0, (now - entry_dt).days) days_to_timeout = max(0, vert - days_open) if entry <= 0 or current_price <= 0: ret = 0.0 else: # Retorno log direccional import math log_ret = math.log(current_price / entry) * (1 if side == 1 else -1) ret = log_ret # Distancia (positiva = aun no tocada) distance_to_tp = pt - ret distance_to_sl = sl + ret # cuando ret < -sl, distance_to_sl < 0 # Regimen girado: solo si la posicion esta EN BENEFICIO y el precio cruza # su media de tendencia en contra (long bajo la media / short sobre ella). regime_flipped = ( trend_sma is not None and trend_sma > 0 and ret > 0 and ((side == 1 and current_price < trend_sma) or (side == -1 and current_price > trend_sma))) action = "hold" note = "" if ret >= pt: action = "close_tp" note = f"return {ret*100:+.2f}% >= TP {pt*100:.2f}%" elif ret <= -sl: action = "close_sl" note = f"return {ret*100:+.2f}% <= -SL {-sl*100:.2f}%" elif regime_flipped: action = "close_regime" note = (f"en beneficio ({ret*100:+.2f}%) y el regimen giro: precio " f"{current_price:.2f} cruzo su media {trend_sma:.2f}") elif days_open >= vert: action = "close_timeout" note = f"dias abiertos={days_open} >= {vert}" return OpenTradeView( id=int(row["id"]), ticker=str(row["ticker"]), side=side, qty=float(row["qty"]), entry_date=str(row["entry_date"]), entry_price=entry, sigma_entry=float(row["sigma_entry"]), pt_pct=pt, sl_pct=sl, vert_barrier_days=vert, current_price=current_price, current_return_pct=round(ret, 6), days_open=days_open, distance_to_tp=round(distance_to_tp, 6), distance_to_sl=round(distance_to_sl, 6), days_to_timeout=days_to_timeout, proposed_action=action, proposed_notes=note, ) def fetch_recent_ohlc(db_path: Path, ticker: str, n: int = 30) -> list[tuple[float, float, float, float]]: """Devuelve los ultimos `n` (high, low, close, prev_close) en orden cronologico ascendente. Vacio si no hay datos suficientes.""" if not db_path.exists(): return [] try: with sqlite3.connect(db_path) as conn: rows = conn.execute( "SELECT date, high, low, close FROM prices " "WHERE ticker = ? ORDER BY date DESC LIMIT ?", (ticker, int(n) + 1), ).fetchall() except Exception: return [] if not rows or len(rows) < 2: return [] rows = list(reversed(rows)) # ascendente out = [] prev_close = float(rows[0][3]) for i in range(1, len(rows)): h = float(rows[i][1]) if rows[i][1] is not None else prev_close lo = float(rows[i][2]) if rows[i][2] is not None else prev_close cl = float(rows[i][3]) if rows[i][3] is not None else prev_close out.append((h, lo, cl, prev_close)) prev_close = cl return out def fetch_atr(db_path: Path, ticker: str, window: int = TRAILING_ATR_WINDOW) -> Optional[float]: """ATR(window) sobre datos diarios. Devuelve el ATR absoluto en $. Wilder: media de True Range donde TR = max(high-low, |high-prev_close|, |low-prev_close|). None si no hay suficiente historia.""" ohlc = fetch_recent_ohlc(db_path, ticker, n=window + 5) if len(ohlc) < window: return None trs = [] for h, lo, _cl, prev_close in ohlc[-window:]: tr = max(h - lo, abs(h - prev_close), abs(lo - prev_close)) trs.append(tr) if not trs: return None return sum(trs) / len(trs) def fetch_rsi_14(db_path: Path, ticker: str) -> Optional[float]: """RSI(14) sobre cierres diarios. Wilder smoothing. None si insuficiente.""" if not db_path.exists(): return None try: with sqlite3.connect(db_path) as conn: rows = conn.execute( "SELECT close FROM prices WHERE ticker = ? " "ORDER BY date DESC LIMIT 30", (ticker,), ).fetchall() except Exception: return None closes = [float(r[0]) for r in reversed(rows) if r and r[0] is not None] if len(closes) < 15: return None gains = [] losses = [] for i in range(1, len(closes)): diff = closes[i] - closes[i - 1] gains.append(max(diff, 0.0)) losses.append(max(-diff, 0.0)) avg_gain = sum(gains[:14]) / 14 avg_loss = sum(losses[:14]) / 14 for i in range(14, len(gains)): avg_gain = (avg_gain * 13 + gains[i]) / 14 avg_loss = (avg_loss * 13 + losses[i]) / 14 if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss return 100.0 - (100.0 / (1.0 + rs)) def fetch_momentum_20d(db_path: Path, ticker: str) -> Optional[float]: """Retorno simple de los ultimos 20 dias. None si insuficiente.""" if not db_path.exists(): return None try: with sqlite3.connect(db_path) as conn: rows = conn.execute( "SELECT close FROM prices WHERE ticker = ? " "ORDER BY date DESC LIMIT 22", (ticker,), ).fetchall() except Exception: return None closes = [float(r[0]) for r in reversed(rows) if r and r[0] is not None] if len(closes) < 21: return None return (closes[-1] / closes[-21]) - 1.0 def fetch_current_regime(db_path: Path, ticker: str) -> Optional[str]: """Ultimo regimen registrado para el ticker en la tabla regimes (poblada por agente_regimen.py). Valores: TENDENCIA_ALCISTA | TENDENCIA_BAJISTA | LATERAL | CRISIS | DESCONOCIDO.""" if not db_path.exists(): return None try: with sqlite3.connect(db_path) as conn: row = conn.execute( "SELECT regime FROM regimes WHERE ticker = ? " "ORDER BY date DESC LIMIT 1", (ticker,), ).fetchone() except Exception: return None return str(row[0]) if row and row[0] else None def has_potential(side: int, regime: Optional[str], rsi: Optional[float], momentum: Optional[float], current_ret: float) -> tuple[bool, str]: """Filtro tecnico para trailing. Devuelve (es_potencial, razon). LONG: regimen contiene 'ALCISTA' AND rsi<80 AND mom20>0 AND ret>0 SHORT: regimen contiene 'BAJISTA' AND rsi>20 AND mom20<0 AND ret>0 Si falta algun dato, devuelve False con la razon.""" if current_ret <= 0: return False, f"en perdida ({current_ret*100:+.2f}%)" if regime is None: return False, "regimen no disponible" if rsi is None: return False, "rsi no disponible" if momentum is None: return False, "momentum no disponible" regime_up = regime.upper() if side == 1: # LONG if "ALCISTA" not in regime_up: return False, f"regimen {regime} no alcista" if rsi >= TRAILING_RSI_MAX_LONG: return False, f"rsi {rsi:.1f} >= {TRAILING_RSI_MAX_LONG} (sobrecompra)" if momentum <= 0: return False, f"mom20 {momentum*100:+.2f}% no positivo" return True, (f"LONG sano: regimen {regime}, rsi {rsi:.1f}, " f"mom20 {momentum*100:+.2f}%, ret {current_ret*100:+.2f}%") elif side == -1: # SHORT if "BAJISTA" not in regime_up: return False, f"regimen {regime} no bajista" if rsi <= TRAILING_RSI_MIN_SHORT: return False, f"rsi {rsi:.1f} <= {TRAILING_RSI_MIN_SHORT} (sobreventa)" if momentum >= 0: return False, f"mom20 {momentum*100:+.2f}% no negativo" return True, (f"SHORT sano: regimen {regime}, rsi {rsi:.1f}, " f"mom20 {momentum*100:+.2f}%, ret {current_ret*100:+.2f}%") return False, f"side desconocido {side}" def compute_trailing_levels( side: int, entry: float, current_price: float, atr: float, pt_pct_current: float, sl_pct_current: float, current_ret: float, ) -> tuple[float, float, bool, str]: """Calcula nuevos pt_pct y sl_pct con trailing. Solo extiende; nunca reduce. Devuelve (new_pt_pct, new_sl_pct, was_adjusted, note). LONG: new_sl_price = max(entry*(1-sl_pct_current), current_price - 1*ATR) new_sl_pct = (entry - new_sl_price) / entry -- distancia % positiva (puede ser NEGATIVA si new_sl > entry -> SL por encima de entry = breakeven o mejor; lo dejamos negativo asi el evaluador trata sl_pct como nivel relativo y close_sl se dispara correctamente cuando ret <= -sl_pct). new_pt_pct = max(pt_pct_current, (current_price + 2*ATR)/entry - 1) (solo si current >= 70% del camino hacia TP_old, sino pt_pct_current se mantiene). SHORT: simetrico. """ if atr is None or atr <= 0 or entry <= 0 or current_price <= 0: return pt_pct_current, sl_pct_current, False, "atr/precios invalidos" if side == 1: # SL: precio absoluto del SL nuevo = current - 1*ATR. Pero solo si # eso es mas favorable que el SL actual. sl_price_current = entry * (1.0 - sl_pct_current) sl_price_trail = current_price - TRAILING_SL_ATR_MULT * atr sl_price_new = max(sl_price_current, sl_price_trail) new_sl_pct = (entry - sl_price_new) / entry # puede ser <0 si supera entry # TP: solo extender si current ya ha viajado >= threshold% del camino. tp_price_current = entry * (1.0 + pt_pct_current) progress_to_tp = (current_price - entry) / (tp_price_current - entry) \ if tp_price_current > entry else 0.0 if progress_to_tp >= TRAILING_TP_EXTENSION_THRESHOLD: tp_price_trail = current_price + TRAILING_TP_ATR_MULT * atr tp_price_new = max(tp_price_current, tp_price_trail) new_pt_pct = (tp_price_new / entry) - 1.0 else: new_pt_pct = pt_pct_current elif side == -1: sl_price_current = entry * (1.0 + sl_pct_current) sl_price_trail = current_price + TRAILING_SL_ATR_MULT * atr sl_price_new = min(sl_price_current, sl_price_trail) new_sl_pct = (sl_price_new - entry) / entry tp_price_current = entry * (1.0 - pt_pct_current) progress_to_tp = (entry - current_price) / (entry - tp_price_current) \ if entry > tp_price_current else 0.0 if progress_to_tp >= TRAILING_TP_EXTENSION_THRESHOLD: tp_price_trail = current_price - TRAILING_TP_ATR_MULT * atr tp_price_new = min(tp_price_current, tp_price_trail) new_pt_pct = 1.0 - (tp_price_new / entry) else: new_pt_pct = pt_pct_current else: return pt_pct_current, sl_pct_current, False, "side desconocido" # Solo aplicamos si HAY cambio material (>= 0.1% en cualquier nivel) EPS = 0.001 pt_changed = (new_pt_pct - pt_pct_current) > EPS sl_changed = (new_sl_pct - sl_pct_current) > EPS # mas pequeño sl_pct = menos perdida # Para SL, "menos perdida" = sl_pct mas pequeño (o negativo). Pero arriba # ya garantizamos sl_price_new >= sl_price_current. Entonces new_sl_pct # SIEMPRE <= sl_pct_current (por construccion). Si la diferencia es no # trivial, sl_changed: sl_changed = (sl_pct_current - new_sl_pct) > EPS was_adjusted = pt_changed or sl_changed if not was_adjusted: return pt_pct_current, sl_pct_current, False, "trailing sin cambio material" parts = [] if sl_changed: parts.append(f"SL {sl_pct_current*100:+.2f}% -> {new_sl_pct*100:+.2f}%") if pt_changed: parts.append(f"TP {pt_pct_current*100:+.2f}% -> {new_pt_pct*100:+.2f}%") note = "trailing aplicado: " + ", ".join(parts) return new_pt_pct, new_sl_pct, True, note def persist_trailing_levels(db_path: Path, trade_id: int, new_pt_pct: float, new_sl_pct: float, note: str) -> None: """UPDATE open_trades SET pt_pct, sl_pct + INSERT trailing_adjustments log.""" try: with sqlite3.connect(db_path) as conn: # Asegura tabla de log conn.execute( """ CREATE TABLE IF NOT EXISTS trailing_adjustments ( id INTEGER PRIMARY KEY AUTOINCREMENT, trade_id INTEGER NOT NULL, ts TEXT NOT NULL, new_pt_pct REAL, new_sl_pct REAL, note TEXT ) """ ) conn.execute( "UPDATE open_trades SET pt_pct = ?, sl_pct = ? WHERE id = ?", (float(new_pt_pct), float(new_sl_pct), int(trade_id)), ) conn.execute( "INSERT INTO trailing_adjustments " "(trade_id, ts, new_pt_pct, new_sl_pct, note) VALUES (?,?,?,?,?)", (int(trade_id), _now_utc().isoformat(), float(new_pt_pct), float(new_sl_pct), str(note)), ) conn.commit() except Exception: pass def fetch_open_trades(db_path: Path) -> list[dict]: if not db_path.exists(): return [] with sqlite3.connect(db_path) as conn: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT * FROM open_trades WHERE status = 'open' ORDER BY entry_date ASC" ).fetchall() return [dict(r) for r in rows] def fetch_trend_sma(db_path: Path, ticker: str, window: int = REGIME_SMA_WINDOW) -> Optional[float]: """Media de los ultimos `window` cierres diarios de un activo — su tendencia de fondo. Devuelve None si no hay historico suficiente.""" if not db_path.exists(): return None try: with sqlite3.connect(db_path) as conn: rows = conn.execute( "SELECT close FROM prices WHERE ticker = ? " "ORDER BY date DESC LIMIT ?", (ticker, int(window))).fetchall() except Exception: return None closes = [float(r[0]) for r in rows if r and r[0] is not None] if len(closes) < window: return None return sum(closes) / len(closes) def mark_trade_closed( db_path: Path, trade_id: int, reason: str, exit_price: float, pnl_pct: float, alpaca_sell_order_id: Optional[str], ) -> None: with sqlite3.connect(db_path) as conn: conn.execute( """ UPDATE open_trades SET status = ?, exit_date = ?, exit_price = ?, pnl_pct = ?, alpaca_sell_order_id = ?, closed_at = ? WHERE id = ? """, ( f"closed_{reason}", _now_utc().isoformat(), float(exit_price), float(pnl_pct), alpaca_sell_order_id, _now_utc().isoformat(), int(trade_id), ), ) conn.commit() class AgenteCloseMonitor: def __init__(self, db_path: Path = DB_PATH, report_path: Path = REPORT_PATH) -> None: self.db_path = db_path self.report_path = report_path self.logger = self._build_logger() DASHBOARD_DATA_DIR.mkdir(parents=True, exist_ok=True) def run(self, mode: str = "read_only") -> CloseMonitorReport: self.logger.info("Iniciando Monitor de Cierres (modo=%s)", mode) # Snapshot Alpaca para coherencia (no es estrictamente necesario, pero util) snap = fetch_snapshot() alpaca_positions_by_symbol = {p.symbol: p for p in snap.positions} if snap.positions else {} rows = fetch_open_trades(self.db_path) if not rows: self.logger.info("Sin posiciones abiertas registradas en open_trades.") return self._empty_report(mode) views: list[OpenTradeView] = [] actions: list[CloseAction] = [] for row in rows: ticker = row["ticker"] # ── Reconciliacion open_trades ↔ posiciones reales Alpaca ──── # Si la fila dice 'open' pero Alpaca ya NO tiene posicion en ese # ticker (cierre manual, orden externa, o fila vieja duplicada # por re-entradas), NUNCA hay que vender: vender sin posicion # abre un CORTO. En execute la marcamos closed_orphan para que # deje de evaluarse; en dry-run solo avisamos. if snap.configured and ticker not in alpaca_positions_by_symbol: if mode == "execute": try: mark_trade_closed(self.db_path, int(row["id"]), "orphan", 0.0, 0.0, None) self.logger.warning( "%s: fila open_trades #%s huerfana (sin posicion " "en Alpaca) — marcada closed_orphan, NO se vende", ticker, row["id"]) except Exception as exc: self.logger.warning( "%s: fila huerfana #%s no se pudo marcar: %s", ticker, row["id"], exc) else: self.logger.warning( "%s: fila open_trades #%s huerfana (sin posicion en " "Alpaca) — se limpiara en el proximo execute", ticker, row["id"]) continue quote = get_latest_quote(ticker) if snap.configured else None # Precio para evaluar TP/SL: # 1. BID del ultimo quote (para un LONG es el precio ejecutable # real de salida; el mid puede estar inflado/deprimido por un # spread ancho del feed IEX gratuito). Para SHORT usamos ASK. # 2. Sanity check contra el current_price que reporta la propia # posicion de Alpaca: si el quote se desvia mas de # QUOTE_SANITY_MAX_DEV, el quote esta rancio o el spread es # absurdo → usamos el precio de la posicion. Esto elimina los # "SL fantasma" (p.ej. DLTR: mid decia -1.9% y la venta real # se lleno un +2.7% por encima de ese mid). pos = alpaca_positions_by_symbol.get(ticker) pos_price = float(pos.current_price) if pos else 0.0 side_row = int(row["side"]) exec_quote = 0.0 if quote: if side_row == 1: exec_quote = float(quote.get("bid") or 0.0) else: exec_quote = float(quote.get("ask") or 0.0) if exec_quote <= 0: exec_quote = float(quote.get("mid") or 0.0) if exec_quote > 0 and pos_price > 0: dev = abs(exec_quote / pos_price - 1.0) if dev > QUOTE_SANITY_MAX_DEV: self.logger.warning( "%s: quote %.2f se desvia %.1f%% del precio de " "posicion %.2f — quote rancio/spread ancho, uso el " "precio de posicion", ticker, exec_quote, dev * 100, pos_price) current_price = pos_price else: current_price = exec_quote elif exec_quote > 0: current_price = exec_quote elif pos_price > 0: current_price = pos_price else: current_price = 0.0 if current_price <= 0: self.logger.warning("Sin precio actual para %s — skip", ticker) continue trend_sma = fetch_trend_sma(self.db_path, ticker, REGIME_SMA_WINDOW) # ─── Trailing TP/SL dinamico ──────────────────────────────────── # Si la posicion va en verde Y el filtro tecnico aprueba (regimen + # rsi + momentum), subimos SL y opcionalmente TP. Solo extiende. # Persistimos en open_trades antes de evaluar para que la decision # close_tp/close_sl se base en los nuevos niveles. trailing_potential = False trailing_applied_now = False trailing_prev_pt = float(row["pt_pct"]) trailing_prev_sl = float(row["sl_pct"]) trailing_note = "" if TRAILING_ENABLED: # ret instantaneo (mismo calculo que evaluate_open_trade) import math as _m entry_tmp = float(row["entry_price"]) side_tmp = int(row["side"]) if entry_tmp > 0 and current_price > 0: ret_tmp = (_m.log(current_price / entry_tmp) * (1 if side_tmp == 1 else -1)) else: ret_tmp = 0.0 regime_now = fetch_current_regime(self.db_path, ticker) rsi_now = fetch_rsi_14(self.db_path, ticker) mom_now = fetch_momentum_20d(self.db_path, ticker) potential, why = has_potential( side_tmp, regime_now, rsi_now, mom_now, ret_tmp) trailing_potential = potential trailing_note = why if potential: atr_abs = fetch_atr(self.db_path, ticker, TRAILING_ATR_WINDOW) new_pt, new_sl, adjusted, adj_note = compute_trailing_levels( side_tmp, entry_tmp, current_price, atr_abs or 0.0, trailing_prev_pt, trailing_prev_sl, ret_tmp) if adjusted: persist_trailing_levels( self.db_path, int(row["id"]), new_pt, new_sl, adj_note) # Reflejamos en el row para que evaluate_open_trade # use los nuevos thresholds en esta misma corrida. row["pt_pct"] = new_pt row["sl_pct"] = new_sl trailing_applied_now = True trailing_note = adj_note self.logger.info("Trailing %s: %s", ticker, adj_note) view = evaluate_open_trade(row, current_price, trend_sma) view.trailing_potential = trailing_potential view.trailing_applied_now = trailing_applied_now view.trailing_prev_pt_pct = trailing_prev_pt view.trailing_prev_sl_pct = trailing_prev_sl view.trailing_note = trailing_note # ── Grace period para trades recien abiertos ──────────────── # El close_monitor corre segundos despues del executor en el # mismo ciclo del bridge. Un trade recien comprado, evaluado al # bid, arranca ya "en rojo" por el spread; con barreras # estrechas eso disparaba un SL a los segundos de abrir (DVA: # compra 18:15:32, venta 18:15:48). Durante los primeros # CLOSE_GRACE_MIN minutos no se permite cerrar por TP/SL/regimen # (un trade tan joven tampoco puede estar en timeout). if view.proposed_action != "hold" and CLOSE_GRACE_MIN > 0: _entry_dt = _parse_iso(str(row["entry_date"])) if _entry_dt is not None: _age_min = (_now_utc() - _entry_dt).total_seconds() / 60.0 if 0 <= _age_min < CLOSE_GRACE_MIN: self.logger.info( "%s: '%s' suprimido por grace period " "(%.0f min < %.0f min de vida)", ticker, view.proposed_action, _age_min, CLOSE_GRACE_MIN) view.proposed_notes = ( f"grace_period({_age_min:.0f}m<" f"{CLOSE_GRACE_MIN:.0f}m): era " f"{view.proposed_action} | " + (view.proposed_notes or "")) view.proposed_action = "hold" views.append(view) if view.proposed_action == "hold": continue reason = view.proposed_action.replace("close_", "") # Nunca vender mas cantidad de la que Alpaca dice que tenemos: # con filas duplicadas/viejas en open_trades, qty de la fila # puede superar la posicion real y el exceso abriria un corto. sell_qty = float(view.qty) if pos is not None: try: _pos_qty = abs(float(pos.qty)) if 0 < _pos_qty < sell_qty: self.logger.warning( "%s: qty de la fila (%s) > posicion real (%s) — " "clamp a la posicion", ticker, sell_qty, _pos_qty) sell_qty = _pos_qty except (TypeError, ValueError): pass action = CloseAction( ticker=ticker, side=view.side, qty=sell_qty, reason=reason, entry_price=view.entry_price, exit_price=current_price, pnl_pct=view.current_return_pct, days_open=view.days_open, open_trade_id=view.id, status="planned", notes=view.proposed_notes, ) actions.append(action) self.logger.info("Open trades=%d | propuestos cierre=%d", len(views), len(actions)) # Ejecutar cierres si toca if mode == "execute" and actions: self._execute_closes(actions) elif mode == "dry_run" and actions: self.logger.info("DRY-RUN: %d cierres NO se envian", len(actions)) report = CloseMonitorReport( generated_at=_now_utc().isoformat(), mode=mode, n_open_trades=len(views), n_proposed_closes=len(actions), open_trades_view=views, close_actions=actions, ) self._export(report) return report def _execute_closes(self, actions: list[CloseAction]) -> None: for a in actions: sell_side = "sell" if a.side == 1 else "buy" # cerrar long = sell, cerrar short = buy self.logger.info("Cerrando %s qty=%s motivo=%s (PnL=%.2f%%)", a.ticker, a.qty, a.reason, a.pnl_pct * 100) result = submit_market_order(a.ticker, a.qty, side=sell_side, time_in_force="day") if result["ok"]: a.status = result["status"] or "submitted" a.alpaca_sell_order_id = result["order_id"] a.notes += f" | sent={a.alpaca_sell_order_id}" try: mark_trade_closed( self.db_path, a.open_trade_id, a.reason, a.exit_price, a.pnl_pct, a.alpaca_sell_order_id, ) except Exception as exc: self.logger.warning("No se pudo marcar trade %d como cerrado: %s", a.open_trade_id, exc) a.notes += f" | db_update_failed={exc!r}" else: a.status = "rejected" a.notes += f" | error={result['error']}" def _empty_report(self, mode: str) -> CloseMonitorReport: report = CloseMonitorReport( generated_at=_now_utc().isoformat(), mode=mode, n_open_trades=0, n_proposed_closes=0, ) self._export(report) return report def _export(self, report: CloseMonitorReport) -> None: payload = { **{k: v for k, v in asdict(report).items() if k not in ("open_trades_view", "close_actions")}, "open_trades_view": [asdict(v) for v in report.open_trades_view], "close_actions": [asdict(a) for a in report.close_actions], } self.report_path.write_text( json.dumps(payload, indent=2, ensure_ascii=False, default=str), encoding="utf-8", ) self.logger.info("Informe exportado → %s", self.report_path) def _build_logger(self) -> logging.Logger: LOGS_DIR.mkdir(parents=True, exist_ok=True) logger = logging.getLogger("agente_close_monitor") logger.setLevel(logging.INFO) logger.handlers.clear() fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s") fh = logging.FileHandler(LOGS_DIR / "agente_close_monitor.log", encoding="utf-8") fh.setFormatter(fmt) sh = logging.StreamHandler() sh.setFormatter(fmt) logger.addHandler(fh) logger.addHandler(sh) return logger def main() -> int: parser = argparse.ArgumentParser(description="Agente Monitor de Cierres de Leonex") parser.add_argument("--dry-run", action="store_true", help="Imprime que cerraria pero NO envia SELL.") parser.add_argument("--execute", action="store_true", help="Envia ordenes de cierre reales. Requiere --confirm.") parser.add_argument("--confirm", action="store_true", help="Confirmacion explicita junto a --execute.") args = parser.parse_args() if args.execute and not args.confirm: print("[!] --execute requiere ademas --confirm. Aborto.") return 2 if args.execute and args.confirm: mode = "execute" elif args.dry_run: mode = "dry_run" else: mode = "read_only" # ── LEONEX_DRY_RUN_MODE: modo observacion global ───────────────────── # Con el flag activo el monitor evalua y journalea igual (incluida la # limpieza que NO manda ordenes, como marcar filas huerfanas), pero no # envia ningun cierre a Alpaca aunque el invocador pida --execute. # Vive en el agente para que ningun camino (scheduler, endpoint, cron, # CLI) pueda saltarselo. if mode == "execute" and _os.environ.get( "LEONEX_DRY_RUN_MODE", "").strip().lower() in ("1", "true", "yes"): mode = "dry_run" print("[DRY-RUN GLOBAL] LEONEX_DRY_RUN_MODE=1 activo — cierres NO se envian.") if sys.stdout.encoding and sys.stdout.encoding.lower() != "utf-8": import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") agente = AgenteCloseMonitor() report = agente.run(mode=mode) sep = "=" * 64 print(f"\n{sep}") print(f"Leonex -- Monitor de Cierres (modo={mode})") print(sep) print(f"Posiciones abiertas registradas : {report.n_open_trades}") print(f"Cierres propuestos : {report.n_proposed_closes}") if report.open_trades_view: print() print(f"{'Ticker':<8} {'Entrada':>9} {'Actual':>9} {'Ret':>8} {'TP':>7} {'SL':>7} {'Dias':>5} {'Accion':<14}") for v in report.open_trades_view: print(f"{v.ticker:<8} ${v.entry_price:>7.2f} ${v.current_price:>7.2f} " f"{v.current_return_pct*100:>+6.2f}% {v.pt_pct*100:>5.2f}% {v.sl_pct*100:>5.2f}% " f"{v.days_open:>5} {v.proposed_action:<14}") if report.close_actions: print() print("Cierres:") for a in report.close_actions: tag = {"submitted": "ENVIADO", "rejected": "RECHAZADO", "planned": "PLAN"}.get(a.status, a.status.upper()) print(f" [{tag}] {a.ticker:<6} qty={a.qty} motivo={a.reason:<8} " f"PnL={a.pnl_pct*100:+.2f}% {a.notes}") print(sep) if mode == "read_only": print("Lectura. --dry-run para preview, --execute --confirm para cerrar.") elif mode == "dry_run": print("DRY-RUN: 0 ordenes enviadas. Anade --execute --confirm para cerrar.") else: print("EXECUTE: cierres enviados a Alpaca paper. Revisa dashboard.") return 0 if __name__ == "__main__": import pandas as pd # noqa: F401 (importado para mantener consistencia con otros agentes) raise SystemExit(main())