import numpy as np import pandas as pd import talib.abstract as ta from datetime import datetime, timezone, timedelta from scipy.signal import argrelextrema def acciones_precio(dataframe: pd.DataFrame) -> pd.DataFrame: # --- PATRONES DE ACCIÓN DE PRECIO (TA-Lib) --- dataframe['hammer'] = np.nan dataframe['inverted_hammer'] = np.nan dataframe['bullish_engulfing'] = np.nan dataframe['piercing'] = np.nan dataframe['morning_star'] = np.nan dataframe['three_white_soldiers'] = np.nan dataframe['hanging_man'] = np.nan dataframe['shooting_star'] = np.nan dataframe['bearish_engulfing'] = np.nan dataframe['evening_star'] = np.nan dataframe['three_black_crows'] = np.nan dataframe['dark_cloud_cover'] = np.nan # Valores "raw" (TA-Lib devuelve 0 si no hay patrón, +100 para alcista, -100 para bajista) dataframe['cdl_hammer_raw'] = ta.CDLHAMMER(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_inv_hammer_raw'] = ta.CDLINVERTEDHAMMER(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_engulfing_raw'] = ta.CDLENGULFING(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_piercing_raw'] = ta.CDLPIERCING(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_morning_star_raw'] = ta.CDLMORNINGSTAR(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'], penetration=0.3) dataframe['cdl_three_white_soldiers_raw']= ta.CDL3WHITESOLDIERS(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_hanging_man_raw'] = ta.CDLHANGINGMAN(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_shooting_star_raw'] = ta.CDLSHOOTINGSTAR(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_evening_star_raw'] = ta.CDLEVENINGSTAR(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'], penetration=0.3) dataframe['cdl_three_black_crows_raw'] = ta.CDL3BLACKCROWS(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close']) dataframe['cdl_dark_cloud_cover_raw'] = ta.CDLDARKCLOUDCOVER(dataframe['open'], dataframe['high'], dataframe['low'], dataframe['close'], penetration=0.5) # Columnas binarias cómodas para reglas (1 si se detecta, 0 si no) # Alcistas dataframe['hammer'] = (dataframe['cdl_hammer_raw'] > 0).astype('int8') dataframe['inverted_hammer'] = (dataframe['cdl_inv_hammer_raw'] > 0).astype('int8') dataframe['bullish_engulfing'] = (dataframe['cdl_engulfing_raw'] > 0).astype('int8') dataframe['piercing'] = (dataframe['cdl_piercing_raw'] > 0).astype('int8') dataframe['morning_star'] = (dataframe['cdl_morning_star_raw'] > 0).astype('int8') dataframe['three_white_soldiers'] = (dataframe['cdl_three_white_soldiers_raw'] > 0).astype('int8') # Bajistas dataframe['hanging_man'] = (dataframe['cdl_hanging_man_raw'] < 0).astype('int8') dataframe['shooting_star'] = (dataframe['cdl_shooting_star_raw'] < 0).astype('int8') dataframe['bearish_engulfing'] = (dataframe['cdl_engulfing_raw'] < 0).astype('int8') dataframe['evening_star'] = (dataframe['cdl_evening_star_raw'] < 0).astype('int8') dataframe['three_black_crows'] = (dataframe['cdl_three_black_crows_raw'] < 0).astype('int8') dataframe['dark_cloud_cover'] = (dataframe['cdl_dark_cloud_cover_raw'] < 0).astype('int8') return dataframe def changes(dataframe: pd.DataFrame) -> pd.DataFrame: dataframe['change_1d'] = dataframe['close'] - dataframe['close'].shift(3600) dataframe['change_4h'] = dataframe['close'] - dataframe['close'].shift(240) dataframe['change_1h'] = dataframe['close'] - dataframe['close'].shift(60) dataframe['change_30m'] = dataframe['close'] - dataframe['close'].shift(30) dataframe['change_15m'] = dataframe['close'] - dataframe['close'].shift(15) dataframe['change_5m'] = dataframe['close'] - dataframe['close'].shift(5) dataframe['change_1m'] = dataframe['close'] - dataframe['close'].shift(1) # Cambios porcentuales en cada timeframe dataframe['pct_change_1m'] = (dataframe['close'] / dataframe['close'].shift(1) - 1)*100 dataframe['pct_change_5m'] = (dataframe['close'] / dataframe['close'].shift(5) - 1)*100 dataframe['pct_change_15m'] = (dataframe['close'] / dataframe['close'].shift(15) - 1)*100 dataframe['pct_change_30m'] = (dataframe['close'] / dataframe['close'].shift(30) - 1)*100 dataframe['pct_change_1h'] = (dataframe['close'] / dataframe['close'].shift(60) - 1)*100 dataframe['pct_change_4h'] = (dataframe['close'] / dataframe['close'].shift(240) - 1)*100 dataframe['pct_change_1d'] = (dataframe['close'] / dataframe['close'].shift(1440) - 1)*100 dataframe['pct_change_fwd_1m'] = (dataframe['close'].shift(-1) / dataframe['close'] - 1) * 100 dataframe['pct_change_fwd_5m'] = (dataframe['close'].shift(-5) / dataframe['close'] - 1) * 100 dataframe['pct_change_fwd_15m'] = (dataframe['close'].shift(-15) / dataframe['close'] - 1) * 100 dataframe['pct_change_fwd_30m'] = (dataframe['close'].shift(-30) / dataframe['close'] - 1) * 100 dataframe['pct_change_fwd_1h'] = (dataframe['close'].shift(-60) / dataframe['close'] - 1) * 100 dataframe['pct_change_fwd_4h'] = (dataframe['close'].shift(-240) / dataframe['close'] - 1) * 100 dataframe['pct_change_fwd_1d'] = (dataframe['close'].shift(-1440) / dataframe['close'] - 1) * 100 dataframe['velocity_1d'] = dataframe['close'].pct_change(3600)*100 dataframe['velocity_4h'] = dataframe['close'].pct_change(240)*100 dataframe['velocity_1h'] = dataframe['close'].pct_change(60)*100 dataframe['velocity_30m'] = dataframe['close'].pct_change(30)*100 dataframe['velocity_15m'] = dataframe['close'].pct_change(15)*100 dataframe['velocity_5m'] = dataframe['close'].pct_change(5)*100 dataframe['velocity_1m'] = dataframe['close'].pct_change(1)*100 return dataframe def confirmaciones(dataframe: pd.DataFrame, window: int = 10) -> pd.DataFrame: """ Añade columnas: - 'confirmaciones_alcistas': número de tipos de confirmaciones alcistas ocurridas al menos una vez en la ventana. - 'confirmaciones_bajistas': número de tipos de confirmaciones bajistas ocurridas al menos una vez en la ventana. """ # Flags alcistas ema_alcista = dataframe.get('cruce_ema_alcista', False).fillna(False).astype(bool) macd_alcista = dataframe.get('cruce_macd_alcista', False).fillna(False).astype(bool) vol_alcista = dataframe.get('vol_confirmacion_alcista', False).fillna(False).astype(bool) div_alcista_normal = dataframe.get('bullish_divergence', False).fillna(False).astype(bool) div_alcista_oculta = dataframe.get('bullish_divergence_hidden', False).fillna(False).astype(bool) # Combinamos ambas divergencias como una sola div_alcista = div_alcista_normal | div_alcista_oculta # Flags bajistas ema_bajista = dataframe.get('cruce_ema_bajista', False).fillna(False).astype(bool) macd_bajista = dataframe.get('cruce_macd_bajista', False).fillna(False).astype(bool) vol_bajista = dataframe.get('vol_confirmacion_bajista', False).fillna(False).astype(bool) div_bajista_normal = dataframe.get('bearish_divergence', False).fillna(False).astype(bool) div_bajista_oculta = dataframe.get('bearish_divergence_hidden', False).fillna(False).astype(bool) # Combinamos ambas divergencias como una sola div_bajista = div_bajista_normal | div_bajista_oculta # Para cada fila, mirar si en la ventana de N velas ocurrió al menos una vez ema_window_alcista = ema_alcista.rolling(window=window, min_periods=1).max() macd_window_alcista = macd_alcista.rolling(window=window, min_periods=1).max() vol_window_alcista = vol_alcista.rolling(window=window, min_periods=1).max() div_window_alcista = div_alcista.rolling(window=window, min_periods=1).max() ema_window_bajista = ema_bajista.rolling(window=window, min_periods=1).max() macd_window_bajista = macd_bajista.rolling(window=window, min_periods=1).max() vol_window_bajista = vol_bajista.rolling(window=window, min_periods=1).max() div_window_bajista = div_bajista.rolling(window=window, min_periods=1).max() # Suma de confirmaciones: cada tipo aporta 1 si ocurrió al menos una vez dataframe["confirmaciones_alcistas"] = ( ema_window_alcista + macd_window_alcista + vol_window_alcista + div_window_alcista ) dataframe["confirmaciones_bajistas"] = ( ema_window_bajista + macd_window_bajista + vol_window_bajista + div_window_bajista ) return dataframe def confirmaciones2(dataframe: pd.DataFrame, window: int = 10, max_ventana_confirmaciones: int = 10) -> pd.DataFrame: """ Añade columnas: - 'confirmaciones_alcistas': número de tipos de confirmaciones alcistas ocurridas al menos una vez en la ventana. - 'confirmaciones_bajistas': número de tipos de confirmaciones bajistas ocurridas al menos una vez en la ventana. - 'confirmacion_alcista_valida': True si tras una divergencia alcista se acumulan al menos 2 confirmaciones diferentes. - 'confirmacion_bajista_valida': True si tras una divergencia bajista se acumulan al menos 2 confirmaciones diferentes. """ # Flags alcistas ema_alcista = dataframe.get('cruce_ema_alcista', False).fillna(False).astype(bool) macd_alcista = dataframe.get('cruce_macd_alcista', False).fillna(False).astype(bool) vol_alcista = dataframe.get('vol_confirmacion_alcista', False).fillna(False).astype(bool) div_alcista_normal = dataframe.get('bullish_divergence', False).fillna(False).astype(bool) div_alcista_oculta = dataframe.get('bullish_divergence_hidden', False).fillna(False).astype(bool) div_alcista = div_alcista_normal | div_alcista_oculta # Flags bajistas ema_bajista = dataframe.get('cruce_ema_bajista', False).fillna(False).astype(bool) macd_bajista = dataframe.get('cruce_macd_bajista', False).fillna(False).astype(bool) vol_bajista = dataframe.get('vol_confirmacion_bajista', False).fillna(False).astype(bool) div_bajista_normal = dataframe.get('bearish_divergence', False).fillna(False).astype(bool) div_bajista_oculta = dataframe.get('bearish_divergence_hidden', False).fillna(False).astype(bool) div_bajista = div_bajista_normal | div_bajista_oculta # Confirmaciones por ventana (como antes) ema_window_alcista = ema_alcista.rolling(window=window, min_periods=1).max() macd_window_alcista = macd_alcista.rolling(window=window, min_periods=1).max() vol_window_alcista = vol_alcista.rolling(window=window, min_periods=1).max() div_window_alcista = div_alcista.rolling(window=window, min_periods=1).max() ema_window_bajista = ema_bajista.rolling(window=window, min_periods=1).max() macd_window_bajista = macd_bajista.rolling(window=window, min_periods=1).max() vol_window_bajista = vol_bajista.rolling(window=window, min_periods=1).max() div_window_bajista = div_bajista.rolling(window=window, min_periods=1).max() dataframe["confirmaciones_alcistas"] = ( ema_window_alcista + macd_window_alcista + vol_window_alcista + div_window_alcista ) dataframe["confirmaciones_bajistas"] = ( ema_window_bajista + macd_window_bajista + vol_window_bajista + div_window_bajista ) # Inicializamos columnas de salida dataframe["confirmacion_alcista_valida"] = False dataframe["confirmacion_bajista_valida"] = False n = len(dataframe) # Confirmaciones válidas después de divergencia (ALCISTA) for i in range(n): if div_alcista.iloc[i]: confirmadas = set() for j in range(i + 1, min(i + 1 + max_ventana_confirmaciones, n)): if ema_alcista.iloc[j]: confirmadas.add("ema") if macd_alcista.iloc[j]: confirmadas.add("macd") if vol_alcista.iloc[j]: confirmadas.add("vol") if len(confirmadas) >= 2 and (j - i) >= 3: dataframe.at[j, "confirmacion_alcista_valida"] = True break # Confirmaciones válidas después de divergencia (BAJISTA) for i in range(n): if div_bajista.iloc[i]: confirmadas = set() for j in range(i + 1, min(i + 1 + max_ventana_confirmaciones, n)): if ema_bajista.iloc[j]: confirmadas.add("ema") if macd_bajista.iloc[j]: confirmadas.add("macd") if vol_bajista.iloc[j]: confirmadas.add("vol") if len(confirmadas) >= 2 and (j - i) >= 3: dataframe.at[j, "confirmacion_bajista_valida"] = True break return dataframe def confirmacion_volumen(dataframe: pd.DataFrame, fast: int = 5, slow: int = 10) -> pd.DataFrame: """ Confirma fuerza de volumen si el volumen actual supera la SMA del volumen. """ dataframe[f'vol_sma_{fast}'] = ta.EMA(dataframe['volume'], timeperiod=fast) dataframe[f'vol_sma_{slow}'] = ta.EMA(dataframe['volume'], timeperiod=slow) dataframe["vol_confirmacion_alcista"] = False dataframe["vol_confirmacion_bajista"] = False # Confirmación alcista: volumen actual > SMA volumen y vela verde dataframe['vol_confirmacion_alcista'] = ( (dataframe['volume'] > dataframe[f'vol_sma_{fast}']) & (dataframe['volume'].shift(1) < dataframe[f'vol_sma_{fast}'].shift(1)) & (dataframe['close'] > dataframe['open']) ) # Confirmación bajista: volumen actual > SMA volumen y vela roja dataframe['vol_confirmacion_bajista'] = ( (dataframe['volume'] > dataframe[f'vol_sma_{fast}']) & (dataframe['volume'].shift(1) < dataframe[f'vol_sma_{fast}'].shift(1)) & (dataframe['close'] < dataframe['open']) ) return dataframe def cruce_medias_macd(dataframe: pd.DataFrame, fast: int = 12, slow: int = 26, period: int = 9) -> pd.DataFrame: """ Añade columnas de cruces alcistas y bajistas entre MACD y su señal. """ # Calcula MACD macd, macdsignal, macdhist = ta.MACD( dataframe["close"], fastperiod=fast, slowperiod=slow, signalperiod=period ) dataframe['macd'] = macd dataframe['macdsignal'] = macdsignal dataframe['macdhist'] = macdhist dataframe["cruce_macd_alcista"] = False dataframe["cruce_macd_bajista"] = False # Cruce alcista: MACD cruza por encima de la señal dataframe['cruce_macd_alcista'] = ( (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macd'].shift(1) <= dataframe['macdsignal'].shift(1)) ) # Cruce bajista: MACD cruza por debajo de la señal dataframe['cruce_macd_bajista'] = ( (dataframe['macd'] < dataframe['macdsignal']) & (dataframe['macd'].shift(1) >= dataframe['macdsignal'].shift(1)) ) return dataframe def cruce_medias_ema(dataframe: pd.DataFrame, fast: int = 20, slow: int = 50) -> pd.DataFrame: """ Añade columnas de cruces alcistas y bajistas entre EMA 'fast' y EMA 'slow'. """ # Calcula EMAs dataframe[f'ema{fast}'] = ta.EMA(dataframe['close'], timeperiod=fast) dataframe[f'ema{slow}'] = ta.EMA(dataframe['close'], timeperiod=slow) dataframe["cruce_ema_alcista"] = False dataframe["cruce_ema_bajista"] = False # Cruce EMA rápida sobre EMA lenta cruce_ema_ema_alcista = ( (dataframe[f'ema{fast}'] > dataframe[f'ema{slow}']) & (dataframe[f'ema{fast}'].shift(1) <= dataframe[f'ema{slow}'].shift(1)) ) cruce_ema_ema_bajista = ( (dataframe[f'ema{fast}'] < dataframe[f'ema{slow}']) & (dataframe[f'ema{fast}'].shift(1) >= dataframe[f'ema{slow}'].shift(1)) ) # Cruce precio sobre EMA rápida cruce_precio_ema_alcista = ( (dataframe["close"] > dataframe[f'ema{fast}']) & (dataframe["close"].shift(1) <= dataframe[f'ema{fast}'].shift(1)) ) cruce_precio_ema_bajista = ( (dataframe["close"] < dataframe[f'ema{fast}']) & (dataframe["close"].shift(1) >= dataframe[f'ema{fast}'].shift(1)) ) # Combinar: si ocurre uno u otro, activar el mismo flag dataframe["cruce_ema_alcista"] = cruce_ema_ema_alcista | cruce_precio_ema_alcista dataframe["cruce_ema_bajista"] = cruce_ema_ema_bajista | cruce_precio_ema_bajista return dataframe def trendlines_up(dataframe: pd.DataFrame, price_maxs_trend: np.ndarray , idx: int, i2: int, i1: int) -> pd.DataFrame: maximos_en_rango = [] max_busqueda = 200 count = 0 ultimo_max = None # Indices de price_maxs que están en el rango ampliado price_maxs_backwards = sorted( [idx for idx in price_maxs_trend if idx <= i2 and idx >= max(i1 - max_busqueda, 0)], reverse=True ) for idx in price_maxs_backwards: precio_actual = dataframe.loc[dataframe.index[idx], 'max_ohlc'] if count <= 4: # Siempre añade los dos primeros puntos maximos_en_rango.append(idx) ultimo_max = precio_actual else: if precio_actual >= ultimo_max: maximos_en_rango.append(idx) ultimo_max = precio_actual else: break # Corta solo después del segundo punto count += 1 maximos_en_rango = sorted(maximos_en_rango) for idx in maximos_en_rango: dataframe.loc[dataframe.index[idx], 'price_maxs_backwards'] = dataframe.loc[dataframe.index[idx], 'max_ohlc'] x_max = np.array(maximos_en_rango) y_max = np.array([dataframe.loc[dataframe.index[idx], 'close'] for idx in maximos_en_rango]) # Inicializa los pesos a 1 pesos = np.ones_like(x_max, dtype=float) if len(pesos) > 2: pesos[0] = 4 # Doble peso al primer punto pesos[-1] = 4 # Doble peso al último punto if len(x_max) > 1: pendiente_superior, ordenada_superior = np.polyfit(x_max, y_max, 1, w=pesos) else: pendiente_superior = 0 ordenada_superior = y_max[0] if pendiente_superior < 0: # BEGIN - Subimos la ordenada para que pase siempre por el máximo que hay en la divergencia # Busca el primer índice de máximo dentro de i1 a i2 primer_maximo_idx = next((idx for idx in range(i2, i1 - 1, -1) if idx in price_maxs_trend), None) ultimo_idx_maximos = maximos_en_rango[0] if primer_maximo_idx is not None: precio_maximo = dataframe.loc[dataframe.index[primer_maximo_idx], 'max_ohlc'] valor_linea_en_max = pendiente_superior * primer_maximo_idx + ordenada_superior desplazamiento = precio_maximo - valor_linea_en_max ordenada_superior_ajustada = ordenada_superior + desplazamiento else: # Si no hay máximo en el rango, puedes dejar la ordenada igual o manejarlo como prefieras ordenada_superior_ajustada = ordenada_superior # END - Subimos la ordenada para que pase siempre por el máximo que hay en la divergencia # Línea de divergencia (sobre mínimos) en el índice del máximo # valor_linea_en_max = pendiente_precio * idx_maximo + ordenada_precio # desplazamiento = precio_maximo - valor_linea_en_max # ordenada_superior = ordenada_precio + desplazamiento start_idx = max(i1 - 60, 0) end_idx = min(i2 + 200, len(dataframe) - 1) for idx in range(start_idx, end_idx + 1): if idx < i1 and not np.isnan(dataframe.loc[dataframe.index[idx], 'divergencia_rsi_alcista_precio']): break y_superior = pendiente_superior * idx + ordenada_superior_ajustada if np.isnan(dataframe.loc[dataframe.index[idx], 'line_trend_bullish']): if idx >= ultimo_idx_maximos: dataframe.loc[dataframe.index[idx], 'line_trend_bullish'] = y_superior else: break if idx > i2 and ((dataframe.loc[dataframe.index[idx], 'close'] > y_superior and dataframe.loc[dataframe.index[idx], 'open'] > y_superior) or dataframe.loc[dataframe.index[idx], 'open'] > y_superior): dataframe.loc[dataframe.index[idx], 'break_trend'] = True break return dataframe def group_levels(precios_idx, precios_val, tolerancia=0.01, min_touches=1): niveles = [] frecuencias = [] indices = [] for idx, p in zip(precios_idx, precios_val): encontrado = False for i, n in enumerate(niveles): if abs(p - n) <= tolerancia * n: frecuencias[i] += 1 indices[i].append(idx) encontrado = True break if not encontrado: niveles.append(p) frecuencias.append(1) indices.append([idx]) # Devolver solo los que cumplen con los toques mínimos return [(n, idxs) for n, f, idxs in zip(niveles, frecuencias, indices) if f >= min_touches] def divergences_rsi(dataframe: pd.DataFrame, window: int = 5, trendlines_up_enable: bool = False) -> pd.DataFrame: """ Calcula divergencias RSI alcistas/bajistas (clásicas y ocultas). :param dataframe: pd.DataFrame con columnas ['rsi', 'close', 'date', ...] :param window: Tamaño de ventana para detección de extremos locales. :return: dataframe con columnas de divergencia seleccionadas. """ rsi_mins = [] price_mins = [] rsi_maxs = [] price_maxs = [] dataframe["bullish_divergence"] = 0 dataframe["bullish_divergence_hidden"] = 0 dataframe["bearish_divergence"] = 0 dataframe["bearish_divergence_hidden"] = 0 dataframe["test"] = 0 # Si hay suficientes datos, calcular los extremos if len(dataframe) > window: # BEGIN - Detectar mínimos locales rsi_mins = argrelextrema(dataframe['rsi'].round(2).values, np.less_equal, order=window)[0] price_mins = argrelextrema(dataframe['close'].values, np.less_equal, order=window)[0] dataframe.loc[dataframe.index[rsi_mins], 'rsi_min'] = dataframe.loc[dataframe.index[rsi_mins], 'rsi'] dataframe.loc[dataframe.index[price_mins], 'price_min'] = dataframe.loc[dataframe.index[price_mins], 'close'] # END - Detectar mínimos locales # BEGIN - Detectar máximos locales rsi_maxs = argrelextrema(dataframe['rsi'].values, np.greater_equal, order=window)[0] price_maxs = argrelextrema(dataframe['close'].values, np.greater_equal, order=window)[0] if trendlines_up_enable: dataframe['max_ohlc'] = dataframe[['open', 'close']].max(axis=1) price_maxs_trend = argrelextrema(dataframe['max_ohlc'].values, np.greater_equal, order=2)[0] dataframe['line_trend_bullish'] = np.nan dataframe['break_trend'] = False dataframe['price_maxs_backwards'] = np.nan dataframe.loc[dataframe.index[rsi_maxs], 'rsi_max'] = dataframe.loc[dataframe.index[rsi_maxs], 'rsi'] dataframe.loc[dataframe.index[price_maxs], 'price_max'] = dataframe.loc[dataframe.index[price_maxs], 'close'] # END - Detectar máximos locales # Agrupar niveles similares price_min_indices = price_mins price_min_values = dataframe['close'].iloc[price_mins].values niveles_soporte = group_levels(price_min_indices, price_min_values, tolerancia=0.01, min_touches=2) price_max_indices = price_maxs price_max_values = dataframe['close'].iloc[price_maxs].values niveles_resistencia = group_levels(price_max_indices, price_max_values, tolerancia=0.01, min_touches=2) # Pintar solo en las filas correspondientes for i, (nivel, indices) in enumerate(niveles_soporte): col = f'soporte_{i}' dataframe[col] = np.nan dataframe.loc[dataframe.index[indices], col] = nivel for i, (nivel, indices) in enumerate(niveles_resistencia): col = f'resistencia_{i}' dataframe[col] = np.nan dataframe.loc[dataframe.index[indices], col] = nivel # ---------------------------- # 🟢 Divergencias Alcistas bullish_divs = [] bullish_divs_hidden = [] dataframe['divergencia_rsi_alcista_rsi'] = np.nan dataframe['divergencia_rsi_alcista_precio'] = np.nan dataframe['divergencia_rsi_bajista_rsi'] = np.nan dataframe['divergencia_rsi_bajista_precio'] = np.nan dataframe['divergencia_rsi_alcista_oculta_rsi'] = np.nan dataframe['divergencia_rsi_alcista_oculta_precio'] = np.nan dataframe['divergencia_rsi_bajista_oculta_rsi'] = np.nan dataframe['divergencia_rsi_bajista_oculta_precio'] = np.nan for i in range(1, len(rsi_mins)): i1 = rsi_mins[i - 1] i2 = rsi_mins[i] if i1 in price_mins and i2 in price_mins: rsi1, rsi2 = dataframe.loc[dataframe.index[[i1, i2]], 'rsi'] price1, price2 = dataframe.loc[dataframe.index[[i1, i2]], 'close'] date1, date2 = dataframe.loc[dataframe.index[[i1, i2]], 'date'] open_i2 = dataframe['open'].iloc[i2] close_i2 = dataframe['close'].iloc[i2] percent_change_i2 = ((close_i2 - open_i2) / open_i2) * 100 date1_datetime = date1 date2_datetime = date2 if isinstance(date1_datetime, str): date1_datetime = pd.to_datetime(date1_datetime) if isinstance(date2_datetime, str): date2_datetime = pd.to_datetime(date2_datetime) diff_minutes = (date2_datetime - date1_datetime).total_seconds() / 60 max_rsi_between = dataframe["rsi"].iloc[i1:i2+1].max() max_close_between_raw = dataframe["close"].iloc[i1:i2+1].max() min_price = min(price1, price2) diff = max_close_between_raw - min_price max_close_between = min_price + (diff * 0.55) if rsi2 < rsi1 and price2 > price1: bullish_divs_hidden.append(i2) # y = mx + b -> y = pendiente * x + ordenada pendiente_rsi = (rsi1 - rsi2) / (i1 - i2) # b = y - mx -> ordenada = y - pendiente * x ordenada_rsi = rsi2 - pendiente_rsi * i2 pendiente_precio = (price1 - price2) / (i1 - i2) ordenada_precio = price2 - pendiente_precio * i2 for idx in range(i1, i2 + 1): # y = mx + b -> y = pendiente * x + ordenada y_rsi = pendiente_rsi * idx + ordenada_rsi dataframe.loc[dataframe.index[idx], 'divergencia_rsi_alcista_oculta_rsi'] = y_rsi y_precio = pendiente_precio * idx + ordenada_precio dataframe.loc[dataframe.index[idx], 'divergencia_rsi_alcista_oculta_precio'] = y_precio if rsi2 > rsi1 and price2 < price1: bullish_divs.append(i2) # y = mx + b -> y = pendiente * x + ordenada pendiente_rsi = (rsi1 - rsi2) / (i1 - i2) # b = y - mx -> ordenada = y - pendiente * x ordenada_rsi = rsi2 - pendiente_rsi * i2 pendiente_precio = (price1 - price2) / (i1 - i2) ordenada_precio = price2 - pendiente_precio * i2 for idx in range(i1, i2 + 1): # y = mx + b -> y = pendiente * x + ordenada y_rsi = pendiente_rsi * idx + ordenada_rsi dataframe.loc[dataframe.index[idx], 'divergencia_rsi_alcista_rsi'] = y_rsi y_precio = pendiente_precio * idx + ordenada_precio dataframe.loc[dataframe.index[idx], 'divergencia_rsi_alcista_precio'] = y_precio now = datetime.now(timezone.utc) if isinstance(date2, str): date2 = pd.to_datetime(date2) if trendlines_up_enable: dataframe = trendlines_up(dataframe, price_maxs_trend, idx=idx, i2=i2, i1=i1) dataframe.loc[dataframe.index[bullish_divs], 'bullish_divergence'] = 1 dataframe.loc[dataframe.index[bullish_divs_hidden], 'bullish_divergence_hidden'] = 1 # 🔴 Divergencias Bajistas bearish_divs = [] bearish_divs_hidden = [] for i in range(1, len(rsi_maxs)): i1 = rsi_maxs[i - 1] i2 = rsi_maxs[i] if i1 in price_maxs and i2 in price_maxs: rsi1, rsi2 = dataframe.loc[dataframe.index[[i1, i2]], 'rsi'] price1, price2 = dataframe.loc[dataframe.index[[i1, i2]], 'close'] date1, date2 = dataframe.loc[dataframe.index[[i1, i2]], 'date'] date1_datetime = date1 date2_datetime = date2 if isinstance(date1_datetime, str): date1_datetime = pd.to_datetime(date1_datetime) if isinstance(date2_datetime, str): date2_datetime = pd.to_datetime(date2_datetime) diff_minutes = (date2_datetime - date1_datetime).total_seconds() / 60 if rsi2 < rsi1 and price2 > price1: bearish_divs.append(i2) # y = mx + b -> y = pendiente * x + ordenada pendiente_rsi = (rsi1 - rsi2) / (i1 - i2) # b = y - mx -> ordenada = y - pendiente * x ordenada_rsi = rsi2 - pendiente_rsi * i2 pendiente_precio = (price1 - price2) / (i1 - i2) ordenada_precio = price2 - pendiente_precio * i2 for idx in range(i1, i2 + 1): # y = mx + b -> y = pendiente * x + ordenada y_rsi = pendiente_rsi * idx + ordenada_rsi dataframe.loc[dataframe.index[idx], 'divergencia_rsi_bajista_rsi'] = y_rsi y_precio = pendiente_precio * idx + ordenada_precio dataframe.loc[dataframe.index[idx], 'divergencia_rsi_bajista_precio'] = y_precio if rsi2 > rsi1 and price2 < price1: bearish_divs_hidden.append(i2) # y = mx + b -> y = pendiente * x + ordenada pendiente_rsi = (rsi1 - rsi2) / (i1 - i2) # b = y - mx -> ordenada = y - pendiente * x ordenada_rsi = rsi2 - pendiente_rsi * i2 pendiente_precio = (price1 - price2) / (i1 - i2) ordenada_precio = price2 - pendiente_precio * i2 for idx in range(i1, i2 + 1): # y = mx + b -> y = pendiente * x + ordenada y_rsi = pendiente_rsi * idx + ordenada_rsi dataframe.loc[dataframe.index[idx], 'divergencia_rsi_bajista_oculta_rsi'] = y_rsi y_precio = pendiente_precio * idx + ordenada_precio dataframe.loc[dataframe.index[idx], 'divergencia_rsi_bajista_oculta_precio'] = y_precio now = datetime.now(timezone.utc) if isinstance(date2, str): date2 = pd.to_datetime(date2) dataframe.loc[dataframe.index[bearish_divs], 'bearish_divergence'] = 1 dataframe.loc[dataframe.index[bearish_divs_hidden], 'bearish_divergence_hidden'] = 1 return dataframe