# -*- coding: utf-8 -*- """ motor.py — Motor de señales de PySeñales Lab (se ejecuta en el navegador con Pyodide). Todo el cálculo ocurre en el ordenador de quien visita la web: indicadores, estrategias, backtest y el árbol de decisión. Nada se envía a ningún servidor. Material EDUCATIVO. Las señales que produce no son recomendaciones de inversión. """ import json import math import numpy as np import pandas as pd DF = None # velas cargadas (DataFrame con time, open, high, low, close, volume) ULTIMO = {} # último resultado calculado (para el editor y la red neuronal) # --------------------------------------------------------------------------- # 1. Datos # --------------------------------------------------------------------------- def cargar(velas_json): """Recibe una lista JSON de velas {time, open, high, low, close, volume}.""" global DF df = pd.DataFrame(json.loads(velas_json)) if "volume" not in df: df["volume"] = 0.0 df = df[["time", "open", "high", "low", "close", "volume"]].astype(float) df = df.dropna().drop_duplicates("time").sort_values("time").reset_index(drop=True) df["time"] = df["time"].astype("int64") DF = df return len(df) # --------------------------------------------------------------------------- # 2. Indicadores técnicos (implementados a mano, como en los apuntes) # --------------------------------------------------------------------------- def sma(s, n): return s.rolling(n).mean() def ema(s, n): return s.ewm(span=n, adjust=False).mean() def rsi(close, n=14): """RSI de Wilder: media exponencial de subidas / bajadas.""" delta = close.diff() sube = delta.clip(lower=0) baja = -delta.clip(upper=0) media_sube = sube.ewm(alpha=1 / n, adjust=False, min_periods=n).mean() media_baja = baja.ewm(alpha=1 / n, adjust=False, min_periods=n).mean() rs = media_sube / media_baja.replace(0, np.nan) return (100 - 100 / (1 + rs)).fillna(100.0).where(media_sube.notna()) def bollinger(close, n=20, k=2.0): media = close.rolling(n).mean() desv = close.rolling(n).std(ddof=0) return media + k * desv, media, media - k * desv def macd(close, rapida=12, lenta=26, senal=9): linea = ema(close, rapida) - ema(close, lenta) linea_senal = ema(linea, senal) return linea, linea_senal, linea - linea_senal def atr(df, n=14): prev = df["close"].shift(1) rango = pd.concat([df["high"] - df["low"], (df["high"] - prev).abs(), (df["low"] - prev).abs()], axis=1).max(axis=1) return rango.ewm(alpha=1 / n, adjust=False, min_periods=n).mean() def calcular_indicadores(df, p): ind = pd.DataFrame(index=df.index) c = df["close"] ind["sma_r"] = sma(c, p["sma_r"]) ind["sma_l"] = sma(c, p["sma_l"]) ind["rsi"] = rsi(c, p["rsi_n"]) ind["bb_sup"], ind["bb_med"], ind["bb_inf"] = bollinger(c, p["bb_n"], p["bb_k"]) ind["macd"], ind["macd_senal"], ind["macd_hist"] = macd(c) ind["atr"] = atr(df) return ind # --------------------------------------------------------------------------- # 3. Estrategias: cada una devuelve una POSICIÓN por vela (+1 largo, -1 corto, 0 fuera) # --------------------------------------------------------------------------- def estrategia_cruce_medias(df, ind, p): pos = np.where(ind["sma_r"] > ind["sma_l"], 1, -1) return pd.Series(pos, index=df.index).where(ind["sma_l"].notna(), 0) def estrategia_rsi(df, ind, p): """Reversión: compra en sobreventa, vende en sobrecompra, sale al cruzar 50.""" r = ind["rsi"] senal = pd.Series(np.nan, index=df.index) senal[r < p["rsi_bajo"]] = 1 senal[r > p["rsi_alto"]] = -1 cruza_50 = (r - 50).apply(np.sign).diff().fillna(0) != 0 senal[cruza_50 & senal.isna()] = 0 return senal.ffill().fillna(0) def estrategia_bollinger(df, ind, p): """Reversión a la media: compra bajo la banda inferior, vende sobre la superior.""" c = df["close"] senal = pd.Series(np.nan, index=df.index) senal[c < ind["bb_inf"]] = 1 senal[c > ind["bb_sup"]] = -1 cruza_media = (c - ind["bb_med"]).apply(np.sign).diff().fillna(0) != 0 senal[cruza_media & senal.isna()] = 0 return senal.ffill().fillna(0) def estrategia_macd(df, ind, p): pos = np.where(ind["macd"] > ind["macd_senal"], 1, -1) return pd.Series(pos, index=df.index).where(ind["macd_senal"].notna(), 0) ESTRATEGIAS = { "cruce_medias": ("Cruce de medias", estrategia_cruce_medias), "rsi": ("RSI sobrecompra/sobreventa", estrategia_rsi), "bollinger": ("Bandas de Bollinger", estrategia_bollinger), "macd": ("MACD", estrategia_macd), } # --------------------------------------------------------------------------- # 4. Backtest vectorizado (la posición de hoy se aplica al rendimiento de mañana) # --------------------------------------------------------------------------- def barras_por_anio(df): paso = float(np.median(np.diff(df["time"].values))) if len(df) > 2 else 86400.0 if paso >= 86400 * 0.9: return 252.0 # sesiones diarias return 365.0 * 86400.0 / paso * (5 / 7) # intradía, aprox. días hábiles def backtest(df, pos, coste_pct=0.05, solo_largos=False, desde=0): pos = pos.astype(float).copy() if solo_largos: pos = pos.clip(lower=0) pos.iloc[:desde] = 0 ret = df["close"].pct_change().fillna(0) cambios = pos.diff().abs().fillna(pos.abs()) ret_estr = pos.shift(1).fillna(0) * ret - cambios * coste_pct / 100.0 equity = (1 + ret_estr).cumprod() base = df["close"] / df["close"].iloc[desde] bh = base.where(df.index >= desde, 1.0) activos = ret_estr.iloc[desde + 1:] n_bpa = barras_por_anio(df) sharpe = float(activos.mean() / activos.std() * math.sqrt(n_bpa)) if activos.std() > 0 else 0.0 dd = float((equity / equity.cummax() - 1).min()) # operaciones: tramos de posición constante distinta de 0 tramo = (pos != pos.shift(1)).cumsum() ops = [] for _, g in ret_estr.groupby(tramo): p0 = pos.loc[g.index[0]] if p0 != 0: ops.append(float((1 + ret_estr.loc[g.index[0] + 1: g.index[-1] + 1]).prod() - 1)) ganadoras = sum(1 for o in ops if o > 0) return { "equity": equity, "bh": bh, "metricas": { "rentabilidad": float(equity.iloc[-1] - 1), "buy_hold": float(bh.iloc[-1] - 1), "sharpe": sharpe, "max_dd": dd, "operaciones": len(ops), "aciertos": (ganadoras / len(ops)) if ops else 0.0, }, } def marcadores(df, pos): """Flechas de compra/venta en los cambios de posición.""" out = [] prev = 0 for t, p in zip(df["time"].values, pos.values): if p != prev: if p > prev: out.append({"time": int(t), "tipo": "compra" if p > 0 else "cierre"}) else: out.append({"time": int(t), "tipo": "venta" if p < 0 else "cierre"}) prev = p return out # --------------------------------------------------------------------------- # 5. Variables (features) para los modelos de aprendizaje automático # --------------------------------------------------------------------------- NOMBRES_FEATURES = ["ret_1", "ret_3", "ret_10", "rsi", "dist_sma_r", "dist_sma_l", "bb_pct", "macd_hist_n", "atr_pct", "vol_rel"] def construir_features(df, ind): c = df["close"] X = pd.DataFrame(index=df.index) X["ret_1"] = c.pct_change(1) X["ret_3"] = c.pct_change(3) X["ret_10"] = c.pct_change(10) X["rsi"] = ind["rsi"] / 100.0 X["dist_sma_r"] = c / ind["sma_r"] - 1 X["dist_sma_l"] = c / ind["sma_l"] - 1 ancho = (ind["bb_sup"] - ind["bb_inf"]).replace(0, np.nan) X["bb_pct"] = (c - ind["bb_inf"]) / ancho X["macd_hist_n"] = ind["macd_hist"] / c X["atr_pct"] = ind["atr"] / c vol = df["volume"] X["vol_rel"] = (vol / vol.rolling(20).mean()).replace([np.inf, -np.inf], np.nan).fillna(1.0) - 1 # objetivo: ¿cierra la SIGUIENTE vela por encima de la actual? y = (c.shift(-1) > c).astype(float) y.iloc[-1] = np.nan return X, y def datos_ml(fraccion_train=0.7): """Prepara X (normalizada con estadísticas SOLO del tramo de entrenamiento) para TF.js.""" df, ind = ULTIMO["df"], ULTIMO["ind"] X, y = construir_features(df, ind) validas = X.notna().all(axis=1) idx = np.where(validas.values)[0] idx_lbl = idx[y.iloc[idx].notna().values] corte = int(len(idx_lbl) * fraccion_train) tr = idx_lbl[:corte] media = X.iloc[tr].mean() desv = X.iloc[tr].std().replace(0, 1) Xn = ((X - media) / desv).clip(-5, 5) return { "nombres": NOMBRES_FEATURES, "X": Xn.iloc[idx].values.round(5).tolist(), "y": y.iloc[idx].fillna(-1).values.tolist(), "idx": idx.tolist(), "corte": int(corte), "n_train": int(len(tr)), "n_test": int(len(idx_lbl) - corte), } class ArbolNumpy: """ Árbol de decisión CART escrito a mano con numpy (índice de Gini). Hace lo mismo que DecisionTreeClassifier de scikit-learn, pero en ~60 líneas para que se vea por dentro y sin descargar scikit-learn (~70 MB) en el navegador. """ def __init__(self, max_depth=3, min_samples_leaf=20, n_cortes=32): self.max_depth = max_depth self.min_samples_leaf = min_samples_leaf self.n_cortes = n_cortes @staticmethod def _gini(y): if len(y) == 0: return 0.0 p = y.mean() return 2 * p * (1 - p) # 1 - p² - (1-p)² def _mejor_corte(self, X, y): mejor = (None, None, self._gini(y)) # (variable, umbral, impureza ponderada) n = len(y) for j in range(X.shape[1]): col = X[:, j] for u in np.unique(np.quantile(col, np.linspace(0.03, 0.97, self.n_cortes))): izq = col <= u n_izq = izq.sum() if n_izq < self.min_samples_leaf or n - n_izq < self.min_samples_leaf: continue imp = (n_izq * self._gini(y[izq]) + (n - n_izq) * self._gini(y[~izq])) / n if imp < mejor[2] - 1e-12: mejor = (j, float(u), imp) return mejor def _crecer(self, X, y, prof): nodo = {"n": len(y), "p": float(y.mean()) if len(y) else 0.5} if prof < self.max_depth and len(y) >= 2 * self.min_samples_leaf: j, u, imp = self._mejor_corte(X, y) if j is not None: ganancia = self._gini(y) - imp self.importancia[j] += ganancia * len(y) izq = X[:, j] <= u nodo.update({"var": j, "umbral": u, "izq": self._crecer(X[izq], y[izq], prof + 1), "der": self._crecer(X[~izq], y[~izq], prof + 1)}) return nodo def fit(self, X, y): X, y = np.asarray(X, float), np.asarray(y, float) self.importancia = np.zeros(X.shape[1]) self.raiz = self._crecer(X, y, 0) total = self.importancia.sum() self.feature_importances_ = self.importancia / total if total > 0 else self.importancia self.classes_ = np.array([0.0, 1.0]) return self def _prob_fila(self, fila): nodo = self.raiz while "var" in nodo: nodo = nodo["izq"] if fila[nodo["var"]] <= nodo["umbral"] else nodo["der"] return nodo["p"] def predict_proba(self, X): p = np.array([self._prob_fila(f) for f in np.asarray(X, float)]) return np.column_stack([1 - p, p]) def score(self, X, y): return float(((self.predict_proba(X)[:, 1] >= 0.5) == (np.asarray(y) == 1)).mean()) def reglas(self, nombres): lineas = [] def rec(nodo, nivel): sangria = "| " * nivel if "var" not in nodo: lineas.append(f"{sangria}|--- P(sube)={nodo['p']:.2f} (n={nodo['n']})") return v, u = nombres[nodo["var"]], nodo["umbral"] lineas.append(f"{sangria}|--- {v} <= {u:.4f}") rec(nodo["izq"], nivel + 1) lineas.append(f"{sangria}|--- {v} > {u:.4f}") rec(nodo["der"], nivel + 1) rec(self.raiz, 0) return "\n".join(lineas) def arbol_decision(profundidad=3, fraccion_train=0.7, min_hojas=20, usar_sklearn=False): """ Árbol de decisión. Se entrena con el 70 % más antiguo de las velas y se evalúa con el 30 % más reciente (nunca se mezclan: así no se «ve el futuro»). usar_sklearn=False -> ArbolNumpy (hecho a mano); True -> DecisionTreeClassifier. """ df, ind = ULTIMO["df"], ULTIMO["ind"] X, y = construir_features(df, ind) validas = X.notna().all(axis=1) & y.notna() Xv, yv = X[validas], y[validas] corte = int(len(Xv) * fraccion_train) if corte < 50 or len(Xv) - corte < 20: raise ValueError("Hacen falta más velas (mín. ~150) para entrenar y probar el árbol.") Xtr, ytr, Xte, yte = Xv.iloc[:corte], yv.iloc[:corte], Xv.iloc[corte:], yv.iloc[corte:] if usar_sklearn: from sklearn.tree import DecisionTreeClassifier, export_text modelo = DecisionTreeClassifier(max_depth=int(profundidad), min_samples_leaf=int(min_hojas), random_state=42) modelo.fit(Xtr, ytr) reglas = export_text(modelo, feature_names=NOMBRES_FEATURES, decimals=4) else: modelo = ArbolNumpy(max_depth=int(profundidad), min_samples_leaf=int(min_hojas)) modelo.fit(Xtr.values, ytr.values) reglas = modelo.reglas(NOMBRES_FEATURES) acierto_train = float(modelo.score(Xtr, ytr)) acierto_test = float(modelo.score(Xte, yte)) base = float(max(yte.mean(), 1 - yte.mean())) # acertar diciendo siempre lo más frecuente # probabilidad de subida en TODAS las velas con features (incluida la última, sin etiqueta) filas = X.notna().all(axis=1) prob = pd.Series(np.nan, index=df.index) prob[filas] = modelo.predict_proba(X[filas])[:, list(modelo.classes_).index(1.0)] \ if 1.0 in modelo.classes_ else 0.0 desde = int(Xte.index[0]) importancias = sorted(zip(NOMBRES_FEATURES, modelo.feature_importances_.round(4).tolist()), key=lambda t: -t[1]) _guardar_modelo("arbol", "Árbol de decisión (" + ("scikit-learn" if usar_sklearn else "Python · numpy") + ")", prob, desde, {"train": acierto_train, "test": acierto_test, "base": base}) return { "motor": "scikit-learn" if usar_sklearn else "numpy (hecho a mano)", "acierto_train": acierto_train, "acierto_test": acierto_test, "base": base, "n_train": len(Xtr), "n_test": len(Xte), "reglas": reglas, "importancias": importancias, "desde_time": int(df["time"].iloc[desde]), "prob_ultima": float(prob.iloc[-1]) if not pd.isna(prob.iloc[-1]) else 0.5, } class RedNumpy: """ Perceptrón multicapa escrito con numpy: entrada -> ReLU -> ReLU -> sigmoide. Es la misma red que en Keras sería: Sequential([Dense(n1, 'relu'), Dense(n2, 'relu'), Dense(1, 'sigmoid')]) compile(loss='binary_crossentropy', optimizer='adam') pero con la propagación hacia atrás y el optimizador Adam a la vista. """ def __init__(self, n_entradas, n1=12, n2=8, lr=0.001, semilla=42): rng = np.random.default_rng(semilla) he = lambda a, b: rng.normal(0, math.sqrt(2.0 / a), (a, b)) # inicialización de He self.W = [he(n_entradas, n1), he(n1, n2), rng.normal(0, math.sqrt(1.0 / n2), (n2, 1))] self.b = [np.zeros(n1), np.zeros(n2), np.zeros(1)] self.lr, self.paso = lr, 0 self.m = [np.zeros_like(w) for w in self.W + self.b] self.v = [np.zeros_like(w) for w in self.W + self.b] self.rng = rng @staticmethod def _sigmoide(z): return 1.0 / (1.0 + np.exp(-np.clip(z, -30, 30))) def _adelante(self, X): z1 = X @ self.W[0] + self.b[0]; a1 = np.maximum(z1, 0) z2 = a1 @ self.W[1] + self.b[1]; a2 = np.maximum(z2, 0) p = self._sigmoide(a2 @ self.W[2] + self.b[2]).ravel() return z1, a1, z2, a2, p @staticmethod def perdida(p, y): p = np.clip(p, 1e-7, 1 - 1e-7) return float(-np.mean(y * np.log(p) + (1 - y) * np.log(1 - p))) # binary cross-entropy def _paso_adam(self, X, y): n = len(y) z1, a1, z2, a2, p = self._adelante(X) d3 = (p - y).reshape(-1, 1) / n # derivada de BCE + sigmoide gW3 = a2.T @ d3; gb3 = d3.sum(0) d2 = (d3 @ self.W[2].T) * (z2 > 0) gW2 = a1.T @ d2; gb2 = d2.sum(0) d1 = (d2 @ self.W[1].T) * (z1 > 0) gW1 = X.T @ d1; gb1 = d1.sum(0) grads = [gW1, gW2, gW3, gb1, gb2, gb3] params = self.W + self.b self.paso += 1 b1, b2, eps = 0.9, 0.999, 1e-7 for i, (w, g) in enumerate(zip(params, grads)): self.m[i] = b1 * self.m[i] + (1 - b1) * g self.v[i] = b2 * self.v[i] + (1 - b2) * g * g mh = self.m[i] / (1 - b1 ** self.paso) vh = self.v[i] / (1 - b2 ** self.paso) w -= self.lr * mh / (np.sqrt(vh) + eps) # actualización en el sitio def fit(self, X, y, epocas=40, lote=32, validacion=0.15): X, y = np.asarray(X, float), np.asarray(y, float) corte = int(len(y) * (1 - validacion)) # validación = final del tramo (cronológico) Xt, yt, Xv, yv = X[:corte], y[:corte], X[corte:], y[corte:] historia = [] for _ in range(int(epocas)): orden = self.rng.permutation(len(yt)) for i in range(0, len(yt), lote): idx = orden[i:i + lote] self._paso_adam(Xt[idx], yt[idx]) historia.append({"loss": round(self.perdida(self.predict_proba(Xt), yt), 5), "val": round(self.perdida(self.predict_proba(Xv), yv), 5) if len(yv) else None}) return historia def predict_proba(self, X): return self._adelante(np.asarray(X, float))[4] def red_neuronal(n1=12, n2=8, epocas=40, fraccion_train=0.7): """Entrena RedNumpy con las mismas variables y el mismo corte temporal que el árbol.""" d = datos_ml(fraccion_train) X = np.array(d["X"], float) y = np.array(d["y"], float) tr = np.arange(d["corte"]) te = np.arange(d["corte"], len(y)) te = te[y[te] >= 0] if len(tr) < 100 or len(te) < 20: raise ValueError("Hacen falta más velas (mín. ~200) para entrenar la red.") red = RedNumpy(X.shape[1], int(n1), int(n2)) historia = red.fit(X[tr], y[tr], epocas=int(epocas)) p = red.predict_proba(X) acierto = lambda idx: float(((p[idx] >= 0.5) == (y[idx] == 1)).mean()) base = float(max(y[te].mean(), 1 - y[te].mean())) prob = pd.Series(np.nan, index=ULTIMO["df"].index) prob.iloc[d["idx"]] = p calidad = {"train": acierto(tr), "test": acierto(te), "base": base} _guardar_modelo("red", "Red neuronal (Python · numpy)", prob, d["idx"][d["corte"]], calidad) return {"motor": "numpy (hecho a mano)", "acierto_train": calidad["train"], "acierto_test": calidad["test"], "base": base, "n_train": int(len(tr)), "n_test": int(len(te)), "historia": historia, "prob_ultima": float(p[-1])} def fijar_prob_red(prob_json, desde, calidad_json, nombre="Red neuronal (TF.js)"): """La red de TensorFlow.js entrena en JavaScript y deja aquí sus probabilidades.""" prob = pd.Series(json.loads(prob_json), index=ULTIMO["df"].index, dtype=float) _guardar_modelo("red", nombre, prob, int(desde), json.loads(calidad_json)) def _guardar_modelo(clave, nombre, prob, desde, calidad): ULTIMO.setdefault("modelos", {})[clave] = {"nombre": nombre, "prob": prob, "desde": int(desde), "calidad": calidad} def posicion_desde_prob(prob, umbral=0.55): """Probabilidad de subida -> posición. Zona muerta entre (1-umbral) y umbral.""" pos = pd.Series(np.nan, index=prob.index) pos[prob >= umbral] = 1 pos[prob <= 1 - umbral] = -1 pos[(prob > 1 - umbral) & (prob < umbral)] = 0 return pos.fillna(0) # --------------------------------------------------------------------------- # 6. Semáforo: cada estrategia vota con su posición en la ÚLTIMA vela # --------------------------------------------------------------------------- def semaforo(votos): vals = [v["voto"] for v in votos] if not vals: return {"puntuacion": 0, "texto": "SIN DATOS", "nivel": 0} m = sum(vals) / len(vals) if m >= 0.6: texto, nivel = "COMPRA FUERTE", 2 elif m >= 0.2: texto, nivel = "COMPRA", 1 elif m <= -0.6: texto, nivel = "VENTA FUERTE", -2 elif m <= -0.2: texto, nivel = "VENTA", -1 else: texto, nivel = "NEUTRAL", 0 return {"puntuacion": round(m, 3), "texto": texto, "nivel": nivel} # --------------------------------------------------------------------------- # 7. Punto de entrada que llama la web # --------------------------------------------------------------------------- def _serie(df, s, dec=6): out = [] for t, v in zip(df["time"].values, s.values): if v is not None and not (isinstance(v, float) and math.isnan(v)): out.append({"time": int(t), "value": round(float(v), dec)}) return out def analizar(params_json, extra_json="{}"): """ params: sma_r, sma_l, rsi_n, rsi_bajo, rsi_alto, bb_n, bb_k, coste, solo_largos, ver, umbral, incluir_arbol, incluir_red, solo_si_supera Los modelos de IA entrenados (árbol y red) aparecen como una estrategia más: su probabilidad de subida se convierte en posición con el umbral y votan en el semáforo. """ p = json.loads(params_json) extra = json.loads(extra_json) df = DF if df is None or len(df) < 30: raise ValueError("Carga primero al menos 30 velas.") ind = calcular_indicadores(df, p) ULTIMO["df"], ULTIMO["ind"], ULTIMO["p"] = df, ind, p posiciones = {} nombres = {} desde = {} for clave, (nombre, fn) in ESTRATEGIAS.items(): posiciones[clave] = fn(df, ind, p) nombres[clave] = nombre desde[clave] = 0 meta = {} for clave, m in ULTIMO.get("modelos", {}).items(): if len(m["prob"]) != len(df): continue posiciones[clave] = posicion_desde_prob(m["prob"], p.get("umbral", 0.55)) nombres[clave] = m["nombre"] desde[clave] = m["desde"] ult = m["prob"].iloc[-1] meta[clave] = {"calidad": m["calidad"], "prob": None if pd.isna(ult) else round(float(ult), 4), "incluir": bool(p.get("incluir_" + clave, True))} if "usuario" in ULTIMO and len(ULTIMO["usuario"]) == len(df): posiciones["usuario"] = ULTIMO["usuario"].fillna(0).clip(-1, 1) nombres["usuario"] = ULTIMO.get("usuario_nombre", "Mi estrategia (editor)") desde["usuario"] = 0 resultados = {} votos = [] for clave, pos in posiciones.items(): bt = backtest(df, pos, p["coste"], p["solo_largos"], desde[clave]) pos_eff = pos.clip(lower=0) if p["solo_largos"] else pos voto = float(pos_eff.iloc[-1]) resultados[clave] = {"nombre": nombres[clave], "metricas": bt["metricas"], "voto": voto, "desde": desde[clave]} vota, nota, prob = True, "", None if clave in meta: m = meta[clave] prob = m["prob"] q = m["calidad"] or {} if not m["incluir"]: vota, nota = False, "no vota: desactivado" elif p.get("solo_si_supera", True) and q.get("test", 0) <= q.get("base", 0): vota, nota = False, "no vota: acierta {:.0%} en prueba y la referencia {:.0%}".format( q.get("test", 0), q.get("base", 0)) resultados[clave]["vota"] = vota resultados[clave]["nota"] = nota resultados[clave]["prob"] = prob votos.append({"clave": clave, "nombre": nombres[clave], "voto": voto, "vota": vota, "nota": nota, "prob": prob}) if clave == p.get("ver", "cruce_medias"): resultados[clave]["equity"] = _serie(df, bt["equity"], 5) resultados[clave]["bh"] = _serie(df, bt["bh"], 5) resultados[clave]["marcadores"] = marcadores(df, pos_eff) ULTIMO["posiciones"] = posiciones ultimo = df.iloc[-1] return json.dumps({ "n": len(df), "ultima": {"time": int(ultimo["time"]), "close": float(ultimo["close"]), "rsi": None if pd.isna(ind["rsi"].iloc[-1]) else round(float(ind["rsi"].iloc[-1]), 2), "atr": None if pd.isna(ind["atr"].iloc[-1]) else float(ind["atr"].iloc[-1])}, "indicadores": {k: _serie(df, ind[k]) for k in ["sma_r", "sma_l", "bb_sup", "bb_med", "bb_inf", "rsi", "macd", "macd_senal", "macd_hist"]}, "resultados": resultados, "votos": votos, "semaforo": semaforo([v for v in votos if v["vota"]]), }) def ejecutar_usuario(codigo): """Ejecuta el código del editor. Debe definir mi_estrategia(df, ind) -> Serie de posiciones.""" if "df" not in ULTIMO: raise ValueError("Primero carga datos y pulsa Analizar.") espacio = {"pd": pd, "np": np, "sma": sma, "ema": ema, "rsi": rsi, "bollinger": bollinger, "macd": macd, "atr": atr} exec(codigo, espacio) if "mi_estrategia" not in espacio: raise ValueError("El código debe definir la función mi_estrategia(df, ind).") df, ind = ULTIMO["df"].copy(), ULTIMO["ind"].copy() pos = espacio["mi_estrategia"](df, ind) pos = pd.Series(np.asarray(pos, dtype=float), index=df.index) ULTIMO["usuario"] = pos ULTIMO["usuario_nombre"] = espacio.get("NOMBRE", "Mi estrategia (editor)") return int((pos.fillna(0) != 0).sum())