Implementación propia frente a TA-Lib: validar resultados.

import numpy as np
import pandas as pd

bruto = pd.read_csv("nova_diario_sin_ajustar.csv",
                    parse_dates=["fecha"], index_col="fecha")
eventos = pd.read_csv("nova_eventos.csv", parse_dates=["fecha_ex"])

def factores(bruto, eventos):
    cierre_ant = bruto["cierre"].shift(1)
    f_precio = pd.Series(1.0, index=bruto.index)
    f_volumen = pd.Series(1.0, index=bruto.index)
    for e in eventos.itertuples():
        antes = bruto.index < e.fecha_ex
        if e.tipo == "split":
            f_precio[antes] /= e.ratio
            f_volumen[antes] *= e.ratio
        else:
            f_precio[antes] *= 1 - e.importe / cierre_ant[e.fecha_ex]
    return f_precio, f_volumen

f_precio, f_volumen = factores(bruto, eventos)
px = bruto[["apertura", "maximo", "minimo", "cierre"]].mul(
    f_precio, axis=0)
px.columns = ["o", "h", "l", "c"]
px["v"] = (bruto["volumen"] * f_volumen).astype("int64")
o, h, l, c, v = (px[k] for k in ["o", "h", "l", "c", "v"])
print(len(px), "sesiones,", px.index[0].date(), "a",
      px.index[-1].date())

def sma(x, n):
    return x.rolling(n).mean()

def ema(x, n):
    return x.ewm(span=n, adjust=False, min_periods=n).mean()

def wilder(x, n):
    return x.ewm(alpha=1 / n, adjust=False, min_periods=n).mean()

def rsi(x, n=14):
    d = x.diff()
    sube = wilder(d.clip(lower=0), n)
    baja = wilder((-d).clip(lower=0), n)
    return 100 - 100 / (1 + sube / baja)

def macd(x, rapida=12, lenta=26, senal=9):
    linea = ema(x, rapida) - ema(x, lenta)
    s = ema(linea, senal)
    return pd.DataFrame({"macd": linea, "senal": s,
                         "hist": linea - s})

def rango_verdadero(h, l, c):
    c_ant = c.shift(1)
    tramos = pd.concat([h - l, (h - c_ant).abs(),
                        (l - c_ant).abs()], axis=1)
    return tramos.max(axis=1, skipna=False)

def atr(h, l, c, n=14):
    return wilder(rango_verdadero(h, l, c), n)

def bollinger(x, n=20, k=2, ddof=0):
    media = sma(x, n)
    sd = x.rolling(n).std(ddof=ddof)
    return pd.DataFrame({"media": media, "sup": media + k * sd,
                         "inf": media - k * sd})

def estocastico(c, h, l, n=14, k=3, d=3):
    minimo, maximo = l.rolling(n).min(), h.rolling(n).max()
    k_rapido = 100 * (c - minimo) / (maximo - minimo)
    k_lento = k_rapido.rolling(k).mean()
    return pd.DataFrame({"k": k_lento, "d": k_lento.rolling(d).mean()})

def adx(h, l, c, n=14):
    sube, baja = h.diff(), -l.diff()
    dm_mas = sube.where((sube > baja) & (sube > 0), 0.0)
    dm_menos = baja.where((baja > sube) & (baja > 0), 0.0)
    a = atr(h, l, c, n)
    di_mas = 100 * wilder(dm_mas, n) / a
    di_menos = 100 * wilder(dm_menos, n) / a
    dx = 100 * (di_mas - di_menos).abs() / (di_mas + di_menos)
    return wilder(dx, n)

def obv(c, v):
    return (np.sign(c.diff()).fillna(0) * v).cumsum()

def kama(x, n=10, rapida=2, lenta=30):
    er = (x - x.shift(n)).abs() / x.diff().abs().rolling(n).sum()
    a_r, a_l = 2 / (rapida + 1), 2 / (lenta + 1)
    s = ((er * (a_r - a_l) + a_l) ** 2).to_numpy()
    p = x.to_numpy()
    out = np.full(len(p), np.nan)
    out[n] = p[n]
    for i in range(n + 1, len(p)):
        out[i] = out[i - 1] + s[i] * (p[i] - out[i - 1])
    return pd.Series(out, index=x.index)

import talib

print("TA-Lib", talib.__version__)
grupos = talib.get_function_groups()
print(sum(len(f) for f in grupos.values()), "funciones")
pd.Series({g: len(f) for g, f in grupos.items()}).sort_values()

talib.SMA(c.to_numpy().astype("float32"), timeperiod=20)

import pandas_ta as ta  # no instala con numpy >= 2.3 ni Python 3.14

rsi_pta = ta.rsi(px["c"], length=14)            # Series RSI_14
bandas_pta = ta.bbands(px["c"], length=20, std=2)  # DataFrame
atr_pta = ta.atr(px["h"], px["l"], px["c"], length=14)

C, H, L = c.to_numpy(), h.to_numpy(), l.to_numpy()

def comparar(propio, ref, tol=1e-8):
    a, b = np.asarray(propio, float), np.asarray(ref, float)
    d = np.abs(a - b)
    malas = np.flatnonzero(~(d <= tol) & ~np.isnan(b))
    return {"nan_propio": int(np.isnan(a).sum()),
            "nan_talib": int(np.isnan(b).sum()),
            "dif_max": np.nanmax(d), "dif_final": d[-1],
            "coincide_desde": malas[-1] + 1 if len(malas) else 0}

def resumen(casos):
    t = pd.DataFrame({k: comparar(*par) for k, par in casos.items()}).T
    enteros = ["nan_propio", "nan_talib", "coincide_desde"]
    return t.astype({k: int for k in enteros})

macd_ta, senal_ta, _ = talib.MACD(C)
casos = {
    "SMA 20": (sma(c, 20), talib.SMA(C, 20)),
    "Bollinger sup.": (bollinger(c)["sup"], talib.BBANDS(C, 20)[0]),
    "Estocástico %K": (estocastico(c, h, l)["k"],
                       talib.STOCH(H, L, C, 14, 3, 0, 3, 0)[0]),
    "EMA 20": (ema(c, 20), talib.EMA(C, 20)),
    "RSI 14": (rsi(c), talib.RSI(C, 14)),
    "ATR 14": (atr(h, l, c), talib.ATR(H, L, C, 14)),
    "ADX 14": (adx(h, l, c), talib.ADX(H, L, C, 14)),
    "KAMA 10": (kama(c), talib.KAMA(C, 10)),
    "MACD": (macd(c)["macd"], macd_ta),
    "Bollinger ddof=1": (bollinger(c, ddof=1)["sup"],
                         talib.BBANDS(C, 20)[0]),
    "OBV": (obv(c, v), talib.OBV(C, v.to_numpy().astype(float))),
}
resumen(casos)

def sembrar(x, n, inicio=0):
    """Media simple de x[inicio:inicio+n] en inicio+n-1; NaN antes."""
    s = x.astype(float).copy()
    s.iloc[: inicio + n - 1] = np.nan
    s.iloc[inicio + n - 1] = x.iloc[inicio:inicio + n].mean()
    return s

def ema_ta(x, n):
    return sembrar(x, n).ewm(span=n, adjust=False).mean()

def wilder_ta(x, n):
    # el primer cambio (o rango verdadero) está en la posición 1
    return sembrar(x, n, inicio=1).ewm(alpha=1 / n, adjust=False).mean()

def rsi_ta(x, n=14):
    d = x.diff()
    sube = wilder_ta(d.clip(lower=0), n)
    baja = wilder_ta((-d).clip(lower=0), n)
    return 100 - 100 / (1 + sube / baja)

def atr_ta(h, l, c, n=14):
    return wilder_ta(rango_verdadero(h, l, c), n)

exactos = {
    "EMA 20": (ema_ta(c, 20), talib.EMA(C, 20)),
    "RSI 14": (rsi_ta(c), talib.RSI(C, 14)),
    "ATR 14": (atr_ta(h, l, c), talib.ATR(H, L, C, 14)),
}
resumen(exactos)

def ema_desde(x, n, fin):
    """EMA sembrada con la media simple de x[fin-n+1 .. fin]."""
    s = x.astype(float).copy()
    s.iloc[:fin] = np.nan
    s.iloc[fin] = x.iloc[fin - n + 1: fin + 1].mean()
    return s.ewm(span=n, adjust=False).mean()

ingenuo = ema_ta(c, 12) - ema_ta(c, 26)
alineado = ema_desde(c, 12, 25) - ema_desde(c, 26, 25)
senal = ema_desde(alineado, 9, 33)
resumen({
    "MACD ingenuo": (ingenuo, macd_ta),
    "MACD alineado": (alineado.where(senal.notna()), macd_ta),
    "Señal alineada": (senal, senal_ta),
})

ancho_0 = bollinger(c)["sup"] - bollinger(c)["media"]
ancho_1 = bollinger(c, ddof=1)["sup"] - bollinger(c)["media"]
cociente = (ancho_1 / ancho_0).dropna()
print(f"Ancho ddof=1 / ddof=0: {cociente.min():.6f} a "
      f"{cociente.max():.6f}; raíz de 20/19 = {np.sqrt(20 / 19):.6f}")

dif_obv = talib.OBV(C, v.to_numpy().astype(float)) - obv(c, v)
print(f"OBV TA-Lib - propio: de {dif_obv.min():,.0f}"
      f" a {dif_obv.max():,.0f}")
print(f"Volumen de la primera sesión: {v.iloc[0]:,}")

completo = talib.RSI(C, 14)[500:]
tardio = talib.RSI(C[500:], 14)
dif = np.abs(tardio - completo)
for k in [14, 50, 100, 200]:
    print(f"{k:>3} sesiones después del inicio: {dif[k]:.6f} puntos")

normal = talib.RSI(C, 14)
talib.set_unstable_period("RSI", 100)
con_inestable = talib.RSI(C, 14)
talib.set_unstable_period("RSI", 0)  # el ajuste es global: volver a 0

print("Vacíos:", np.isnan(normal).sum(), "->",
      np.isnan(con_inestable).sum())
print("Diferencia máxima donde hay valor:",
      np.nanmax(np.abs(con_inestable - normal)))

from talib import abstract
inestables = [f for f in talib.get_functions()
              if "Function has an unstable period"
              in abstract.Function(f).info["function_flags"]]
print(len(inestables), "funciones inestables:", ", ".join(inestables))

funciones = {
    "SMA 20": lambda C, H, L: talib.SMA(C, 20),
    "Bollinger 20": lambda C, H, L: talib.BBANDS(C, 20)[0],
    "EMA 20": lambda C, H, L: talib.EMA(C, 20),
    "KAMA 10": lambda C, H, L: talib.KAMA(C, 10),
    "ATR 14": lambda C, H, L: talib.ATR(H, L, C, 14),
    "RSI 14": lambda C, H, L: talib.RSI(C, 14),
    "MACD": lambda C, H, L: talib.MACD(C)[0],
    "ADX 14": lambda C, H, L: talib.ADX(H, L, C, 14),
}
filas = {}
for nombre, f in funciones.items():
    completo = f(C, H, L)[500:]
    tardio = f(C[500:], H[500:], L[500:])
    escala = np.nanmean(np.abs(completo))
    rel = np.abs(tardio - completo) / escala
    malas = np.flatnonzero(~(rel < 1e-4))
    filas[nombre] = {"lookback": int(np.isnan(tardio).sum()),
                     "coincide_desde": int(malas[-1] + 1)}
calentamiento = pd.DataFrame(filas).T
calentamiento["inestable"] = (calentamiento["coincide_desde"]
                              - calentamiento["lookback"])
calentamiento

from talib import stream

s = stream.RSI(C[:-5], timeperiod=14)  # historia sin las 5 últimas
en_vivo = [s.update(x) for x in C[-5:]]  # 5 barras cerradas, una a una
print("Stream:  ", np.round(en_vivo, 4))
print("Función: ", np.round(talib.RSI(C, 14)[-5:], 4))
print("peek con una barra a medias:", round(s.peek(C[-1] * 1.01), 4),
      "| valor guardado:", round(s.value, 4))

CALENTAMIENTO = 300

def tras_calentamiento(x):
    return np.asarray(x, float)[CALENTAMIENTO:]

def test_igual_que_talib_tras_calentamiento():
    for propio, ref in [(ema(c, 20), talib.EMA(C, 20)),
                        (rsi(c), talib.RSI(C, 14)),
                        (atr(h, l, c), talib.ATR(H, L, C, 14))]:
        np.testing.assert_allclose(tras_calentamiento(propio),
                                   tras_calentamiento(ref),
                                   rtol=0, atol=1e-6)

def test_semilla_exacta_y_mismos_vacios():
    propio, ref = rsi_ta(c).to_numpy(), talib.RSI(C, 14)
    assert (np.isnan(propio) == np.isnan(ref)).all()
    np.testing.assert_allclose(propio, ref, rtol=0, atol=1e-10)

def test_rangos_validos():
    r = rsi(c).dropna()
    assert r.between(0, 100).all()
    assert (atr(h, l, c).dropna() > 0).all()
    b = bollinger(c).dropna()
    assert (b["sup"] >= b["media"]).all()

def test_sin_anticipacion():
    corte = "2022-12-30"
    for f in (rsi, lambda x: ema(x, 20), lambda x: sma(x, 20)):
        completo, truncado = f(c).loc[:corte], f(c.loc[:corte])
        np.testing.assert_allclose(completo, truncado, equal_nan=True)

tests = [test_igual_que_talib_tras_calentamiento,
         test_semilla_exacta_y_mismos_vacios,
         test_rangos_validos, test_sin_anticipacion]
for t in tests:
    t()
    print("PASSED", t.__name__)

import numpy as np
import pytest
import talib

from mi_sistema.indicadores import atr, ema, rsi
from mi_sistema.datos import nova_ajustada

PX = nova_ajustada()
C, H, L = (PX[k].to_numpy() for k in ["c", "h", "l"])

@pytest.mark.parametrize("propio, ref", [
    (lambda: ema(PX["c"], 20), lambda: talib.EMA(C, 20)),
    (lambda: rsi(PX["c"]), lambda: talib.RSI(C, 14)),
    (lambda: atr(PX["h"], PX["l"], PX["c"]),
     lambda: talib.ATR(H, L, C, 14)),
])
def test_igual_que_talib(propio, ref):
    np.testing.assert_allclose(propio().to_numpy()[300:], ref()[300:],
                               rtol=0, atol=1e-6)