""" server_B.py — Cerebro B — VibeEngine v10.0 (APEX Remodelación NUMPY+LLM) ========================================================================= v10.0 — Remodelación según Plan Maestro APEX-ASYMMETRIC SWARM (Fase 2): ARQUITECTURA LLM+MATH (replicando filosofía de Cerebro A): FASE MATH (<2ms) — TODO el procesamiento pesado con numpy puro: · Filtro de Ruido Gaussiano: limpia las velas OHLCV de ticks falsos antes de calcular indicadores. Convolution kernel [0.25, 0.50, 0.25] vectorizado. · Donchian Channels (numpy): highest/lowest de ventana N. Sin pandas. · SuperTrend simplificado (numpy): ATR × factor, direccion up/down. · ADX (numpy vectorizado): True Range, DM+/DM- suavizados, DI, DX, ADX. · CHoCH (Change of Character): rotura de estructura HH/HL/LH/LL. · Fractales de Williams: máx/mín de 5 velas (confirmados). · Fase de mercado: accumulation/markup/distribution/panic/ranging. · SuperTrend_confirmed: bool que indica si el precio está encima/debajo. FASE LLM (Visto Bueno Final) — invocada SIEMPRE para dar veredicto: · NO recibe texto crudo ni calcula nada. Solo lee el JSON de MATH. · Prompt: panel de control ultra-comprimido (~60 chars). · Output: {"bias": "BULL|BEAR|NEUTRAL", "tendencia_confirmada": "UP|DOWN|FLAT"} · Si LLM falla → MATH local es el fallback (no bloquea el pipeline). RESTRICCIÓN: NO SE USA PANDAS en ninguna parte del código. Motivo: pandas-ta fue retirado de PyPI para Python 3.11+. Implementación 100% con numpy + listas puras. OUTPUT (100% backward compatible con swarm_engine v16): { "trend": "up|down|side", "volatility": "high|low", "strength": float 0-1, "market_phase": "accumulation|markup|distribution|panic|ranging", "trend_strength": float 0-1, "choch": bool, "fractal_top": bool, "fractal_bot": bool, "adx": float, # Nuevos v10.0: "donchian_upper": float, "donchian_lower": float, "supertrend_dir": "up|down", "supertrend_confirmed": bool, "gaussian_applied": bool, "llm_bias": "BULL|BEAR|NEUTRAL", "_math_ms": float, "_llm_ms": float, "_total_ms": float, } TELEMETRÍA: [B/MATH] ms | [B/LLM] ms | [B/TOTAL] ms """ import os, json, re, time, threading, math, asyncio from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import httpx # BitNet v6.0 — ik_llama.cpp via HTTP # ── BitNet v6.0 — Config HTTP (ik_llama.cpp) ───────────────────────────────── BITNET_PORT = int(os.environ.get("BITNET_PORT", "8080")) BITNET_HOST = os.environ.get("BITNET_HOST", "127.0.0.1") BITNET_BASE = f"http://{BITNET_HOST}:{BITNET_PORT}" BITNET_TIMEOUT = float(os.environ.get("BITNET_TIMEOUT", "45.0")) LLM_MAX_TOKENS = int(os.environ.get("LLM_MAX_TOKENS", "128")) N_CTX = int(os.environ.get("N_CTX", "2048")) MODEL_PATH = os.environ.get("MODEL_PATH", "/models/ggml-model-i2_s.gguf") CACHE_TTL = float(os.environ.get("CACHE_TTL", "45.0")) _bitnet_client = None def _get_bitnet_client(): global _bitnet_client if _bitnet_client is None or _bitnet_client.is_closed: _bitnet_client = httpx.AsyncClient( base_url=BITNET_BASE, timeout=httpx.Timeout(BITNET_TIMEOUT), limits=httpx.Limits(max_connections=4, max_keepalive_connections=2), ) return _bitnet_client async def _bitnet_infer(prompt_text: str, max_tokens: int = 128) -> dict: client = _get_bitnet_client() payload = { "prompt": prompt_text, "n_predict": max_tokens, "temperature": 0.0, "stop": ["<|im_end|>", "<|end_of_text|>", "\n\n"], "cache_prompt": True, } try: resp = await client.post("/completion", json=payload) resp.raise_for_status() data = resp.json() return {"raw": data.get("content", ""), "ok": True} except Exception as e: return {"raw": "", "ok": False, "error": str(e)[:60]} # ───────────────────────────────────────────────────────────────────────────── # numpy obligatorio — implementación de indicadores vectorizada try: import numpy as np _NP_OK = True except ImportError: _NP_OK = False print("[B] ⚠️ numpy no disponible — usando cálculos puro-Python") app = FastAPI(title="Cerebro B — VibeEngine v10.0 NUMPY+LLM") # ── Config ──────────────────────────────────────────────────────────────────── CEREBRO_ID = "B" VERSION = "10.0" # ── Cache ────────────────────────────────────────────────────────────────────── _B_CACHE: dict = {} _B_LOCK = threading.Lock() def _cache_get(sym: str): with _B_LOCK: e = _B_CACHE.get(sym) if e and (time.time() - e["ts"]) < CACHE_TTL: r = dict(e["result"]); r["_cached"] = True print(f"[B/CACHE] {sym}: hit ({int(time.time()-e['ts'])}s)") return r return None def _cache_set(sym: str, result: dict): with _B_LOCK: _B_CACHE[sym] = {"result": result, "ts": time.time()} def _parse_symbol(prompt: str) -> str: m = re.search(r'([A-Z]{2,6}(?:/USD)?)\s', prompt) return m.group(1) if m else "?" # ══════════════════════════════════════════════════════════════════════════════ # PARSEO DE DATOS OHLCV DESDE EL PROMPT # ══════════════════════════════════════════════════════════════════════════════ def _parse_closes(prompt: str) -> list: """Extrae lista de closes del prompt del orquestador.""" try: m = re.search(r'closes:\s*\[([^\]]+)\]', prompt) if m: return [float(x) for x in m.group(1).split(",") if x.strip()] except Exception: pass # Fallback: extraer cualquier número con decimales nums = re.findall(r'\d+\.\d{2,8}', prompt) return [float(n) for n in nums[-20:] if 0.001 < float(n) < 200000] def _parse_ohlcv(prompt: str) -> tuple: """ Intenta extraer OHLCV completo si está disponible. Retorna (opens, highs, lows, closes, volumes) como listas. Si no hay OHLCV, usa closes para todo y volumen unitario. """ closes = _parse_closes(prompt) if not closes: return [], [], [], [], [] # Aproximar H/L/O desde closes si no hay datos completos opens = [closes[0]] + closes[:-1] # open[i] ≈ close[i-1] highs = [c * 1.003 for c in closes] # estimación conservadora lows = [c * 0.997 for c in closes] vols = [1.0] * len(closes) # volumen unitario si no disponible return opens, highs, lows, closes, vols # ══════════════════════════════════════════════════════════════════════════════ # FILTRO DE RUIDO GAUSSIANO (numpy vectorizado) # Limpia ticks falsos antes de calcular indicadores # ══════════════════════════════════════════════════════════════════════════════ def _gaussian_filter(data: list) -> list: """ Filtro gaussiano simple: convolution con kernel [0.25, 0.50, 0.25]. Elimina ruido de ticks falsos sin desplazar la señal (kernel simétrico). Si numpy no está disponible, usa promedio ponderado puro-Python. """ if len(data) < 3: return data kernel = [0.25, 0.50, 0.25] if _NP_OK: arr = np.array(data, dtype=float) k = np.array(kernel) # Modo 'same' con padding de borde para preservar longitud padded = np.pad(arr, 1, mode='edge') result = np.convolve(padded, k, mode='valid') return result.tolist() else: # Puro Python: aplicar kernel a cada punto interno filtered = list(data) for i in range(1, len(data) - 1): filtered[i] = kernel[0] * data[i-1] + kernel[1] * data[i] + kernel[2] * data[i+1] return filtered # ══════════════════════════════════════════════════════════════════════════════ # DONCHIAN CHANNELS (numpy — sin pandas) # ══════════════════════════════════════════════════════════════════════════════ def _donchian(highs: list, lows: list, period: int = 20) -> tuple: """ Canales de Donchian: upper = max(high, N), lower = min(low, N). Retorna (upper, lower, mid) del último período. """ if len(highs) < 2: return 0.0, 0.0, 0.0 n = min(period, len(highs)) if _NP_OK: h_arr = np.array(highs[-n:], dtype=float) l_arr = np.array(lows[-n:], dtype=float) upper = float(np.max(h_arr)) lower = float(np.min(l_arr)) else: upper = max(highs[-n:]) lower = min(lows[-n:]) mid = (upper + lower) / 2.0 return round(upper, 6), round(lower, 6), round(mid, 6) # ══════════════════════════════════════════════════════════════════════════════ # ADX — Average Directional Index (numpy vectorizado, sin pandas) # ══════════════════════════════════════════════════════════════════════════════ def _compute_adx_numpy(highs: list, lows: list, closes: list, period: int = 14) -> float: """ ADX completo con numpy: TR, DM+/DM-, DI+/DI-, DX, ADX suavizado (Wilder). Requiere al menos period × 2 velas para ser estadísticamente válido. Si los datos son insuficientes, retorna 25.0 (neutral). """ n = len(closes) if n < max(period + 1, 5): return 25.0 if _NP_OK: h = np.array(highs, dtype=float) l = np.array(lows, dtype=float) c = np.array(closes, dtype=float) # True Range tr1 = h[1:] - l[1:] tr2 = np.abs(h[1:] - c[:-1]) tr3 = np.abs(l[1:] - c[:-1]) tr = np.maximum(tr1, np.maximum(tr2, tr3)) # Directional Movement dm_plus = np.where((h[1:] - h[:-1]) > (l[:-1] - l[1:]), np.maximum(h[1:] - h[:-1], 0.0), 0.0) dm_minus = np.where((l[:-1] - l[1:]) > (h[1:] - h[:-1]), np.maximum(l[:-1] - l[1:], 0.0), 0.0) # Suavizado de Wilder (EWM con alpha=1/period) def _wilder_smooth(arr, p): result = np.zeros(len(arr)) result[0] = np.mean(arr[:p]) if len(arr) >= p else arr[0] alpha = 1.0 / p for i in range(1, len(arr)): result[i] = result[i-1] * (1 - alpha) + arr[i] * alpha return result tr_s = _wilder_smooth(tr, period) dmp_s = _wilder_smooth(dm_plus, period) dmm_s = _wilder_smooth(dm_minus, period) di_plus = 100.0 * dmp_s / np.where(tr_s > 0, tr_s, 1.0) di_minus = 100.0 * dmm_s / np.where(tr_s > 0, tr_s, 1.0) dx_denom = di_plus + di_minus dx = np.where(dx_denom > 0, 100.0 * np.abs(di_plus - di_minus) / dx_denom, 0.0) adx_s = _wilder_smooth(dx, period) return round(float(adx_s[-1]), 1) else: # Fallback puro-Python sin numpy diffs = [abs(closes[i] - closes[i-1]) for i in range(1, n)] ups = sum(1 for i in range(1, n) if closes[i] > closes[i-1]) directional = abs(ups / max(n - 1, 1) - 0.5) * 2 return round(directional * 100, 1) # ══════════════════════════════════════════════════════════════════════════════ # SUPERTREND (numpy — sin pandas) # ══════════════════════════════════════════════════════════════════════════════ def _compute_supertrend(highs: list, lows: list, closes: list, atr_period: int = 10, multiplier: float = 3.0) -> tuple: """ SuperTrend simplificado sin pandas. Retorna (direction: "up"|"down", confirmed: bool). confirmed=True: precio encima de supertrend (señal alcista) confirmed=False: precio debajo de supertrend (señal bajista) """ n = len(closes) if n < atr_period + 2: return "side", False # ATR simplificado (media de TR sobre atr_period) if _NP_OK: h = np.array(highs, dtype=float) l = np.array(lows, dtype=float) c = np.array(closes, dtype=float) tr1 = h[1:] - l[1:] tr2 = np.abs(h[1:] - c[:-1]) tr3 = np.abs(l[1:] - c[:-1]) tr = np.maximum(tr1, np.maximum(tr2, tr3)) atr = float(np.mean(tr[-atr_period:])) else: tr_list = [max(highs[i] - lows[i], abs(highs[i] - closes[i-1]), abs(lows[i] - closes[i-1])) for i in range(1, n)] atr = sum(tr_list[-atr_period:]) / atr_period if tr_list else 0.01 # Banda central = (H+L)/2 ± multiplier × ATR hl2 = (highs[-1] + lows[-1]) / 2.0 upper_band = hl2 + multiplier * atr lower_band = hl2 - multiplier * atr last_close = closes[-1] prev_close = closes[-2] if n >= 2 else last_close # Dirección: si precio > upper_band → bajista (rompió resistencia hacia arriba con trampa) # si precio > lower_band y tendencia alcista → alcista if last_close > lower_band and prev_close > lower_band: direction = "up" confirmed = True elif last_close < upper_band and prev_close < upper_band: direction = "down" confirmed = False else: direction = "side" confirmed = False return direction, confirmed # ══════════════════════════════════════════════════════════════════════════════ # CHoCH, FRACTALES, FASE DE MERCADO (sin cambios de lógica, añadido gaussian) # ══════════════════════════════════════════════════════════════════════════════ def _detect_choch(closes: list) -> bool: if len(closes) < 6: return False highs_idx, lows_idx = [], [] for i in range(1, len(closes) - 1): if closes[i] > closes[i-1] and closes[i] > closes[i+1]: highs_idx.append(closes[i]) if closes[i] < closes[i-1] and closes[i] < closes[i+1]: lows_idx.append(closes[i]) if len(highs_idx) < 2 or len(lows_idx) < 2: return False choch_bear = (highs_idx[-1] < highs_idx[-2]) and (lows_idx[-1] < lows_idx[-2]) choch_bull = (highs_idx[-1] > highs_idx[-2]) and (lows_idx[-1] > lows_idx[-2]) return choch_bear or choch_bull def _detect_fractals(closes: list) -> tuple: if len(closes) < 5: return False, False c = closes[-5:] ft = c[2] > c[0] and c[2] > c[1] and c[2] > c[3] and c[2] > c[4] fb = c[2] < c[0] and c[2] < c[1] and c[2] < c[3] and c[2] < c[4] return ft, fb def _detect_market_phase(closes: list, adx: float, choch: bool, ft: bool, fb: bool, supertrend_dir: str) -> str: if len(closes) < 5: return "ranging" delta = closes[-1] - closes[0] pct = delta / closes[0] if closes[0] != 0 else 0 avg = sum(closes) / len(closes) at_top = closes[-1] > avg * 1.005 at_bottom = closes[-1] < avg * 0.995 # SuperTrend añade confirmación adicional en v10.0 if adx > 30 and pct > 0.003 and supertrend_dir == "up" and not ft: return "markup" if adx > 30 and pct < -0.003 and choch and supertrend_dir == "down": return "panic" if at_top and ft and adx < 25: return "distribution" if at_bottom and fb and adx < 25: return "accumulation" return "ranging" # ══════════════════════════════════════════════════════════════════════════════ # ANÁLISIS TÉCNICO COMPLETO (FASE MATH) # ══════════════════════════════════════════════════════════════════════════════ def _math_full_analysis(prompt: str, sym: str) -> dict: """ FASE MATH completa en <2ms: 1. Parseo OHLCV desde prompt 2. Filtro Gaussiano sobre closes (elimina ticks falsos) 3. Donchian Channels (numpy) 4. SuperTrend (numpy ATR) 5. ADX (Wilder, numpy) 6. CHoCH + Fractales + Fase de Mercado """ t0 = time.perf_counter() opens, highs, lows, closes, vols = _parse_ohlcv(prompt) # Fallback si no hay datos if len(closes) < 2: return { "trend": "side", "volatility": "low", "strength": 0.3, "market_phase": "ranging", "trend_strength": 0.3, "choch": False, "fractal_top": False, "fractal_bot": False, "adx": 25.0, "donchian_upper": 0.0, "donchian_lower": 0.0, "supertrend_dir": "side", "supertrend_confirmed": False, "gaussian_applied": False, "_math_ms": (time.perf_counter() - t0) * 1000, } # ── 1. Filtro Gaussiano ────────────────────────────────────────────────── closes_g = _gaussian_filter(closes) highs_g = _gaussian_filter(highs) if highs else closes_g lows_g = _gaussian_filter(lows) if lows else closes_g gaussian_applied = (closes_g != closes) # ── 2. Indicadores sobre datos suavizados ─────────────────────────────── adx = _compute_adx_numpy(highs_g, lows_g, closes_g, period=ADX_PERIOD) supertrend_dir, supertrend_confirmed = _compute_supertrend( highs_g, lows_g, closes_g, atr_period=SUPERTREND_ATR, multiplier=SUPERTREND_MULT) dc_upper, dc_lower, dc_mid = _donchian(highs_g, lows_g, period=DONCHIAN_PERIOD) # ── 3. Estructura de precio ───────────────────────────────────────────── choch = _detect_choch(closes_g) ft, fb = _detect_fractals(closes_g) market_phase = _detect_market_phase(closes_g, adx, choch, ft, fb, supertrend_dir) # ── 4. Trend / Volatility / Strength ──────────────────────────────────── delta = closes_g[-1] - closes_g[0] pct = abs(delta) / closes_g[0] if closes_g[0] > 0 else 0 trend = "side" if pct < 0.003 else ("up" if delta > 0 else "down") strength = round(min(0.95, 0.3 + pct * 20 + adx / 200), 4) avg = sum(closes_g) / len(closes_g) vol_pct = max(abs(c - avg) / avg for c in closes_g) if avg > 0 else 0 volatility = "high" if vol_pct > 0.005 else "low" math_ms = (time.perf_counter() - t0) * 1000 return { # Heredados (backward compat) "trend": trend, "volatility": volatility, "strength": strength, "market_phase": market_phase, "trend_strength": strength, "choch": choch, "fractal_top": ft, "fractal_bot": fb, "adx": adx, # Nuevos v10.0 "donchian_upper": dc_upper, "donchian_lower": dc_lower, "supertrend_dir": supertrend_dir, "supertrend_confirmed": supertrend_confirmed, "gaussian_applied": gaussian_applied, "_math_ms": round(math_ms, 2), } # ══════════════════════════════════════════════════════════════════════════════ # FASE LLM — Visto Bueno Final con sesgo BULL/BEAR/NEUTRAL # ══════════════════════════════════════════════════════════════════════════════ async def _llm_bias_final(math_data: dict, sym: str) -> dict: """ El LLM actúa como Comité de Dirección Visual: Recibe el panel de control destilado por MATH y emite el sesgo final. Output: {"bias": "BULL|BEAR|NEUTRAL", "tendencia_confirmada": "UP|DOWN|FLAT"} """ t0_llm = time.perf_counter() # Panel de control comprimido (~60 chars) trend = math_data.get("trend", "side") adx = math_data.get("adx", 25.0) phase = math_data.get("market_phase", "ranging")[:4] # 4 chars choch = "Y" if math_data.get("choch") else "N" st_dir = math_data.get("supertrend_dir", "side")[0] # "u"/"d"/"s" st_ok = "Y" if math_data.get("supertrend_confirmed") else "N" llm_prompt = ( "<|im_start|>system\n" 'Price action bias. SOLO JSON: {"bias":"BULL|BEAR|NEUTRAL","tc":"UP|DOWN|FLAT"}.\n' "Sin pensar.\n" "<|im_end|>\n" "<|im_start|>user\n" f'{{"s":"{sym[:8]}","tr":"{trend}","adx":{adx:.0f},' f'"ph":"{phase}","choch":"{choch}","st":"{st_dir}{st_ok}"}}\n' "<|im_end|>\n" "<|im_start|>assistant\n{" ) try: result = await _bitnet_infer(llm_prompt) raw_out_text = result.get("raw", "") llm_ms = (time.perf_counter() - t0_llm) * 1000 raw = "{" + raw_out_text m = re.search(r'\{[^{}]*"bias"\s*:\s*"(BULL|BEAR|NEUTRAL)"[^{}]*\}', raw, re.IGNORECASE) if m: parsed = json.loads(m.group()) bias = parsed.get("bias", "NEUTRAL").upper() tc = parsed.get("tc", "FLAT").upper() if bias not in ("BULL", "BEAR", "NEUTRAL"): bias = "NEUTRAL" if tc not in ("UP", "DOWN", "FLAT"): tc = "FLAT" print(f"[B/LLM] {sym}: bias={bias} tc={tc} | {llm_ms:.1f}ms") return {"llm_bias": bias, "tc": tc, "_llm_ms": round(llm_ms, 1)} # Parseo parcial if "BULL" in raw.upper(): bias = "BULL" elif "BEAR" in raw.upper(): bias = "BEAR" else: bias = "NEUTRAL" print(f"[B/LLM] {sym}: partial={bias} | {llm_ms:.1f}ms") return {"llm_bias": bias, "tc": "FLAT", "_llm_ms": round(llm_ms, 1)} except Exception as e: llm_ms = (time.perf_counter() - t0_llm) * 1000 print(f"[B/LLM-ERR] {sym}: {type(e).__name__}: {str(e)[:50]} | {llm_ms:.1f}ms → math fallback") # Fallback: derivar bias desde MATH trend = math_data.get("trend", "side") bias = {"up": "BULL", "down": "BEAR"}.get(trend, "NEUTRAL") return {"llm_bias": bias, "tc": "FLAT", "_llm_ms": round(llm_ms, 1), "_fallback": True} # ══════════════════════════════════════════════════════════════════════════════ # GATE MATH-FIRST (herencia de v9.2, reforzada en v10.0) # El LLM se activa SIEMPRE en v10.0 (a diferencia de v9.2 que tenía gate). # La decisión de si llamar al LLM es del diseño arquitectónico, no del gate. # Solo se omite en modo cached o si el análisis es trivialmente claro. # ══════════════════════════════════════════════════════════════════════════════ async def _run_inference(agent: str, prompt: str) -> dict: t0_total = time.perf_counter() sym = _parse_symbol(prompt) # ── Cache ───────────────────────────────────────────────────────────────── cached = _cache_get(sym) if sym and sym != "?" else None if cached: return cached # ── FASE 1: MATH ────────────────────────────────────────────────────────── math_data = _math_full_analysis(prompt, sym) math_ms = math_data["_math_ms"] print(f"[B/MATH] {sym}: trend={math_data['trend']} adx={math_data['adx']:.1f} " f"phase={math_data['market_phase']} choch={math_data['choch']} " f"st={math_data['supertrend_dir']} | {math_ms:.1f}ms") # ── FASE 2: LLM Visto Bueno Final ───────────────────────────────────────── # Gate: Si la señal es trivialmente clara (alta ADX + CHoCH + SuperTrend alineado), # omitir LLM para máxima velocidad. LLM solo en zona ambigua. _signal_clear = ( (math_data["trend"] in ("up", "down") and math_data["strength"] >= 0.55) or math_data["choch"] or math_data["market_phase"] in ("markup", "panic") or (math_data["adx"] > 30 and math_data["supertrend_confirmed"]) ) if _signal_clear: # MATH es suficiente — derivar bias directamente trend = math_data["trend"] bias = {"up": "BULL", "down": "BEAR"}.get(trend, "NEUTRAL") llm_ms = 0.0 print(f"[B/MATH-CLEAR] {sym}: bias={bias} (signal claro, skip LLM) | {math_ms:.1f}ms") else: # Zona ambigua → invocar LLM llm_data = await _llm_bias_final(math_data, sym) bias = llm_data["llm_bias"] llm_ms = llm_data["_llm_ms"] total_ms = (time.perf_counter() - t0_total) * 1000 print(f"[B/TOTAL] {sym}: math={math_ms:.1f}ms llm={llm_ms:.1f}ms total={total_ms:.1f}ms") result = { **math_data, "llm_bias": bias, "_llm_ms": round(llm_ms, 1), "_total_ms": round(total_ms, 1), "cerebro": "B", } # Limpiar campo interno de math antes de cachear result.pop("_math_ms", None) result["_math_ms"] = round(math_ms, 1) if sym and sym != "?": _cache_set(sym, result) return result # ── Endpoints FastAPI ───────────────────────────────────────────────────────── FALLBACK = { "trend": "side", "volatility": "low", "strength": 0.3, "market_phase": "ranging", "trend_strength": 0.3, "choch": False, "fractal_top": False, "fractal_bot": False, "adx": 25.0, "donchian_upper": 0.0, "donchian_lower": 0.0, "supertrend_dir": "side", "supertrend_confirmed": False, "gaussian_applied": False, "llm_bias": "NEUTRAL", } @app.get("/") def root(): return { "status": "online", "cerebro": CEREBRO_ID, "version": VERSION, "model": "BitNet-b1.58-2B-4T-i2_s (ik_llama.cpp)", "bitnet_server": BITNET_BASE, "agents": ["VibeEngine"], "features": [ "Gaussian Noise Filter (numpy convolution)", "Donchian Channels (numpy, sin pandas)", "SuperTrend (ATR numpy, sin pandas)", "ADX Wilder (numpy vectorizado, sin pandas)", "CHoCH Detection", "Williams Fractals", "Market Phase (5 estados)", "LLM Visto Bueno Final (BULL|BEAR|NEUTRAL)", ], "no_pandas": True, } @app.get("/health") def health(): return { "ok": True, "cerebro": CEREBRO_ID, "version": VERSION, "numpy_ok": _NP_OK, "pandas": False, # confirmación explícita: NO usamos pandas } @app.post("/analyze") async def analyze(request: Request): try: data = await request.json() except Exception: return JSONResponse({"error": "JSON inválido"}, status_code=400) agent = str(data.get("agent", "VibeEngine")) prompt = str(data.get("prompt", "")).strip()[:600] if not prompt: return JSONResponse({"error": "prompt requerido"}, status_code=400) try: result = await _run_inference(agent, prompt) return {"response": json.dumps(result), "agent": agent, "cerebro": CEREBRO_ID} except Exception as e: print(f"[B/ERROR] {e}") return JSONResponse({ "response": json.dumps(FALLBACK), "agent": agent, "cerebro": CEREBRO_ID, "fallback": True, })