import pandas as pd import os import time from datetime import datetime import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import matplotlib.ticker as mticker PATH_UBUNTU = "/var/www/html/flask_project/" # ========================= COLORES CONSOLA ========================= RED = "\033[91m" GREEN = "\033[92m" RESET = "\033[0m" def colorize(value): if isinstance(value, (int, float)): color = GREEN if value >= 0 else RED return f"{color}{value}{RESET}" return value # ========================= CONFIGURACION GENERAL ========================= DESPLAZAMIENTOS = { "cons": 0, "inte": -5, "agre": -10, "ultr": -15 } desde = "2025-05-01" hasta = "2026-03-13" symbol = "XSP" riesgo = "ultr" # cons | inte | agre | ultr # ---- Elegir estrategia: "Vertical" o "IronCondor" ---- estrategia = "IronCondor" CREDITO_TARGET = 0 # dolares COMISION_POR_PATA = 0.51 # dolares por pata (solo entrada, se deja expirar) ARCHIVO_TOP_HORARIOS = f"top3_horarios_{symbol}_{estrategia}_{riesgo}.csv" # ========================= HORA FIJA O DINAMICA ========================= # HORA_FIJA = None HORA_FIJA = "1020" # =========================================================================== # Validacion de estrategia if estrategia not in ("Vertical", "IronCondor"): raise ValueError(f"Estrategia no valida: '{estrategia}'. Usar 'Vertical' o 'IronCondor'.") factor = 5 if symbol in ["SPY", "QQQ", "XSP"] else 1 desplazamiento = DESPLAZAMIENTOS.get(riesgo, 0) / factor modo_hora = HORA_FIJA if HORA_FIJA else "DINAMICA (Top 1 semana siguiente)" comision_x_trade = COMISION_POR_PATA * (2 if estrategia == "Vertical" else 4) print("\n" + "="*70) print(f"BTM {estrategia.upper()} CREDIT TARGET BACKTEST".center(70)) print("="*70) print(f"{'Fecha Desde:':25} {desde}") print(f"{'Fecha Hasta:':25} {hasta}") print(f"{'Simbolo:':25} {symbol}") print(f"{'Estrategia:':25} {estrategia}") print(f"{'Hora Entrada:':25} {modo_hora}") print(f"{'Riesgo:':25} {riesgo}") print(f"{'Credito Target ($):':25} {CREDITO_TARGET}") print(f"{'Comision x Pata ($):':25} {COMISION_POR_PATA}") print(f"{'Comision x Trade ($):':25} {comision_x_trade}") print("="*70 + "\n") print("Iniciando backtest...\n") # ========================= CARGA TOP HORARIOS ========================= def cargar_ventanas_top_horarios(archivo_top, rank_objetivo=1): df_top = pd.read_csv(archivo_top) df_top = df_top[df_top['rank'] == rank_objetivo].copy() df_top['end_date'] = pd.to_datetime(df_top['end_date']) df_top['hora_str'] = df_top['hora'].apply(lambda h: str(int(h)).zfill(4)) df_top = df_top.sort_values('end_date').reset_index(drop=True) ventanas = [] for i, row in df_top.iterrows(): apply_from = row['end_date'] + pd.Timedelta(days=1) apply_until = df_top.loc[i + 1, 'end_date'] if i + 1 < len(df_top) else pd.Timestamp("2099-12-31") ventanas.append((apply_from, apply_until, row['hora_str'])) return ventanas def obtener_hora_para_fecha(fecha_dt, ventanas): for apply_from, apply_until, hora_str in ventanas: if apply_from <= fecha_dt <= apply_until: return hora_str return None # ========================= FUNCIONES VERTICAL ========================= def buscar_entrada_vertical(archivo_option_chain, strike1, strike2, option_type, entry_time, credito_target): try: df = pd.read_csv(archivo_option_chain) df['timestamp'] = pd.to_datetime(df['timestamp']) df = df[df['timestamp'] >= entry_time] except: return None, None es_primera_hora = True confirmaciones = 0 ultimo_credit = None ultimo_ts = None for ts, snap in df.groupby('timestamp'): s1 = snap[snap['strike'] == strike1] s2 = snap[snap['strike'] == strike2] if s1.empty or s2.empty: confirmaciones = 0 es_primera_hora = False continue if option_type == "CALL": p1 = (s1['bid_call'].iloc[0] + s1['ask_call'].iloc[0]) / 2 p2 = (s2['bid_call'].iloc[0] + s2['ask_call'].iloc[0]) / 2 else: p1 = (s1['bid_put'].iloc[0] + s1['ask_put'].iloc[0]) / 2 p2 = (s2['bid_put'].iloc[0] + s2['ask_put'].iloc[0]) / 2 credit_dollars = (p1 - p2) * 100 if credit_dollars >= credito_target: confirmaciones += 1 ultimo_credit = p1 - p2 ultimo_ts = ts if confirmaciones >= 3: if es_primera_hora: return ultimo_credit, ultimo_ts else: return credito_target / 100, ultimo_ts else: confirmaciones = 0 es_primera_hora = False return None, None def calcular_pnl_vertical(credit_entry, close_price, strike1, strike2, option_type): spread = abs(strike1 - strike2) if option_type == "CALL": strike_venta = min(strike1, strike2) strike_compra = max(strike1, strike2) if close_price > strike_compra: pnl = (credit_entry - spread) * 100 elif close_price < strike_venta: pnl = credit_entry * 100 else: pnl = credit_entry * 100 - (close_price - strike_venta) * 100 else: strike_venta = max(strike1, strike2) strike_compra = min(strike1, strike2) if close_price < strike_compra: pnl = (credit_entry - spread) * 100 elif close_price > strike_venta: pnl = credit_entry * 100 else: pnl = credit_entry * 100 - (strike_venta - close_price) * 100 return round(pnl, 2) # ========================= FUNCIONES IRON CONDOR ========================= def buscar_entrada_ic(archivo_option_chain, cs1, cs2, ps1, ps2, entry_time, credito_target): try: df = pd.read_csv(archivo_option_chain) df['timestamp'] = pd.to_datetime(df['timestamp']) df = df[df['timestamp'] >= entry_time] except: return None, None es_primera_hora = True confirmaciones = 0 ultimo_credit_call = None ultimo_credit_put = None ultimo_ts = None for ts, snap in df.groupby('timestamp'): c1 = snap[snap['strike'] == cs1] c2 = snap[snap['strike'] == cs2] p1 = snap[snap['strike'] == ps1] p2 = snap[snap['strike'] == ps2] if c1.empty or c2.empty or p1.empty or p2.empty: confirmaciones = 0 es_primera_hora = False continue call_mid1 = (c1['bid_call'].iloc[0] + c1['ask_call'].iloc[0]) / 2 call_mid2 = (c2['bid_call'].iloc[0] + c2['ask_call'].iloc[0]) / 2 put_mid1 = (p1['bid_put'].iloc[0] + p1['ask_put'].iloc[0]) / 2 put_mid2 = (p2['bid_put'].iloc[0] + p2['ask_put'].iloc[0]) / 2 credit_call = call_mid1 - call_mid2 credit_put = put_mid1 - put_mid2 credit_total = (credit_call + credit_put) * 100 if credit_total >= credito_target: confirmaciones += 1 ultimo_credit_call = credit_call ultimo_credit_put = credit_put ultimo_ts = ts if confirmaciones >= 3: if es_primera_hora: return (ultimo_credit_call + ultimo_credit_put), ultimo_ts else: return credito_target / 100, ultimo_ts else: confirmaciones = 0 es_primera_hora = False return None, None def calcular_pnl_ic(credit_entry, close_price, cs1, cs2, ps1, ps2): call_spread = abs(cs2 - cs1) put_spread = abs(ps1 - ps2) pnl_call = 0 if close_price <= cs1 else (-call_spread if close_price >= cs2 else -(close_price - cs1)) pnl_put = 0 if close_price >= ps1 else (-put_spread if close_price <= ps2 else -(ps1 - close_price)) return round((credit_entry + pnl_call + pnl_put) * 100, 2) # ========================= UTILIDAD COMUN ========================= def obtener_ultimo_underlying_price(archivo_option_chain): try: df = pd.read_csv(archivo_option_chain) df['timestamp'] = pd.to_datetime(df['timestamp']) return df.sort_values('timestamp')['underlying_price'].iloc[-1] except: return None def cargar_strikes(hora, symbol, estrategia, desplazamiento): path_makekos = "/var/www/html/backtestingmarket/predictor_data/makekos/" archivo_strikes = f"{path_makekos}{symbol}/{symbol}_{estrategia}_strikes_{hora}.csv" try: df_s = pd.read_csv(archivo_strikes) df_s['Day'] = pd.to_datetime(df_s['Day']) if estrategia == "Vertical": df_s['Strike1_Desplazado'] = df_s.apply( lambda r: r['Strike1'] + desplazamiento if r['Option_Type'] == 'CALL' else r['Strike1'] - desplazamiento, axis=1 ).astype(int) df_s['Strike2_Desplazado'] = df_s.apply( lambda r: r['Strike2'] + desplazamiento if r['Option_Type'] == 'CALL' else r['Strike2'] - desplazamiento, axis=1 ).astype(int) else: # IronCondor df_s['CS1'] = (df_s['Call_Strike1'] + desplazamiento).astype(int) df_s['CS2'] = (df_s['Call_Strike2'] + desplazamiento).astype(int) df_s['PS1'] = (df_s['Put_Strike1'] - desplazamiento).astype(int) df_s['PS2'] = (df_s['Put_Strike2'] - desplazamiento).astype(int) print(f"OK Strikes cargados para hora {hora}") return df_s except FileNotFoundError: print(f"WARN Archivo no encontrado para hora {hora}: {archivo_strikes}") return None def obtener_archivo_chain(symbol, fecha_str): if symbol in ["SPX", "RUT", "XSP"]: return f"{PATH_UBUNTU}chains/optionChain_${symbol}_{fecha_str}.csv" return f"{PATH_UBUNTU}chains/optionChain_{symbol}_{fecha_str}.csv" # ========================= BARRA DE PROGRESO ========================= def imprimir_progreso(idx, total, tiempo_inicio, fecha_str): elapsed = time.time() - tiempo_inicio avg_seg = elapsed / idx restantes = total - idx eta_seg = avg_seg * restantes pct = idx / total * 100 bloques = int(pct / 5) barra = "#" * bloques + "-" * (20 - bloques) elapsed_fmt = time.strftime("%H:%M:%S", time.gmtime(elapsed)) eta_fmt = time.strftime("%H:%M:%S", time.gmtime(eta_seg)) print( f"\r [{barra}] {pct:5.1f}% " f"{idx}/{total} " f"Elapsed: {elapsed_fmt} " f"ETA: {eta_fmt} " f"Fecha: {fecha_str} ", end="", flush=True ) # ========================= BACKTEST PRINCIPAL ========================= def backtest_unificado(ventanas, desde, hasta, symbol, estrategia, desplazamiento, comision_x_trade): resultados = [] horas_unicas = [HORA_FIJA] if HORA_FIJA else list(set(v[2] for v in ventanas)) strikes_por_hora = {} for hora in horas_unicas: strikes_por_hora[hora] = cargar_strikes(hora, symbol, estrategia, desplazamiento) desde_dt = pd.to_datetime(desde) hasta_dt = pd.to_datetime(hasta) todas_fechas = set() for df_s in strikes_por_hora.values(): if df_s is not None: fechas = df_s[(df_s['Day'] >= desde_dt) & (df_s['Day'] <= hasta_dt)]['Day'] todas_fechas.update(fechas.dt.date.tolist()) total_fechas = len(todas_fechas) print(f"\nTotal fechas de trading encontradas: {total_fechas}") print("Procesando...\n") tiempo_inicio = time.time() for idx, fecha_date in enumerate(sorted(todas_fechas), start=1): fecha_dt = pd.Timestamp(fecha_date) fecha_str = fecha_date.strftime("%Y-%m-%d") # ---- PROGRESO ---- imprimir_progreso(idx, total_fechas, tiempo_inicio, fecha_str) # ------------------ if HORA_FIJA: hora = HORA_FIJA else: hora = obtener_hora_para_fecha(fecha_dt, ventanas) if hora is None: print(f"\nWARN Sin hora Top 1 para {fecha_str}, saltando...") continue df_s = strikes_por_hora.get(hora) if df_s is None: continue fila = df_s[df_s['Day'].dt.date == fecha_date] if fila.empty: print(f"\nWARN Sin strike para {fecha_str} en hora {hora}, saltando...") continue r = fila.iloc[0] hora_fmt = f"{hora[:2]}:{hora[2:]}" entry_time = pd.to_datetime(f"{fecha_str} {hora_fmt}") + pd.Timedelta(seconds=30) archivo = obtener_archivo_chain(symbol, fecha_str) close_price = obtener_ultimo_underlying_price(archivo) if estrategia == "Vertical": option_type = r['Option_Type'] strike1 = r['Strike1_Desplazado'] strike2 = r['Strike2_Desplazado'] credit_entry, entry_ts = buscar_entrada_vertical( archivo, strike1, strike2, option_type, entry_time, CREDITO_TARGET ) if credit_entry is None: resultados.append({ "Day": fecha_str, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "EntryTime": None, "EntryCredit": None, "ClosePrice": close_price, "Exit": "NO_ENTRY", "Comision": 0.0, "P/L": 0.0 }) continue pnl_bruto = calcular_pnl_vertical(credit_entry, close_price, strike1, strike2, option_type) pnl_neto = round(pnl_bruto - comision_x_trade, 2) resultados.append({ "Day": fecha_str, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "EntryTime": entry_ts, "EntryCredit": round(credit_entry * 100, 2), "ClosePrice": close_price, "Exit": "CLOSE", "Comision": comision_x_trade, "P/L": pnl_neto }) else: # IronCondor cs1, cs2 = r['CS1'], r['CS2'] ps1, ps2 = r['PS1'], r['PS2'] # 🚫 Validar estructura IC: short put debe ser < short call if ps1 > cs1: continue credit_entry, entry_ts = buscar_entrada_ic( archivo, cs1, cs2, ps1, ps2, entry_time, CREDITO_TARGET ) if credit_entry is None: resultados.append({ "Day": fecha_str, "Time": hora, "Call_Strike1": cs1, "Call_Strike2": cs2, "Put_Strike1": ps1, "Put_Strike2": ps2, "EntryTime": None, "EntryCredit": None, "ClosePrice": close_price, "Exit": "NO_ENTRY", "Comision": 0.0, "P/L": 0.0 }) continue pnl_bruto = calcular_pnl_ic(credit_entry, close_price, cs1, cs2, ps1, ps2) pnl_neto = round(pnl_bruto - comision_x_trade, 2) resultados.append({ "Day": fecha_str, "Time": hora, "Call_Strike1": cs1, "Call_Strike2": cs2, "Put_Strike1": ps1, "Put_Strike2": ps2, "EntryTime": entry_ts, "EntryCredit": round(credit_entry * 100, 2), "ClosePrice": close_price, "Exit": "CLOSE", "Comision": comision_x_trade, "P/L": pnl_neto }) # Cerrar linea de progreso y mostrar tiempo total tiempo_total = time.time() - tiempo_inicio print() print(f"\n Backtest completado en {time.strftime('%H:%M:%S', time.gmtime(tiempo_total))}\n") if not resultados: if estrategia == "Vertical": cols = ["Day", "Time", "Strike1", "Strike2", "Option", "EntryTime", "EntryCredit", "ClosePrice", "Exit", "Comision", "P/L"] else: cols = ["Day", "Time", "Call_Strike1", "Call_Strike2", "Put_Strike1", "Put_Strike2", "EntryTime", "EntryCredit", "ClosePrice", "Exit", "Comision", "P/L"] return pd.DataFrame(columns=cols) return pd.DataFrame(resultados) # ========================= GRAFICO MENSUAL ========================= MESES_ES = { 1: "Ene", 2: "Feb", 3: "Mar", 4: "Abr", 5: "May", 6: "Jun", 7: "Jul", 8: "Ago", 9: "Sep", 10: "Oct", 11: "Nov", 12: "Dic" } def generar_grafico_mensual(df, symbol, estrategia, riesgo, modo_hora, credito_target, comision_x_trade, output_file): # Solo trades reales (excluir NO_ENTRY) df_trades = df[df['Exit'] == 'CLOSE'].copy() df_trades['Day'] = pd.to_datetime(df_trades['Day']) df_trades['Mes'] = df_trades['Day'].dt.to_period('M') # Agrupar por mes monthly = df_trades.groupby('Mes').agg( PnL_Total=('P/L', 'sum'), Num_Trades=('P/L', 'count') ).reset_index() monthly['Mes_Label'] = monthly['Mes'].apply( lambda p: f"{MESES_ES[p.month]} {p.year}" ) monthly['PnL_Total'] = monthly['PnL_Total'].round(2) colores = ['#2ecc71' if v >= 0 else '#e74c3c' for v in monthly['PnL_Total']] fig, ax1 = plt.subplots(figsize=(max(10, len(monthly) * 1.1), 6)) fig.patch.set_facecolor('#1a1a2e') ax1.set_facecolor('#16213e') # Barras bars = ax1.bar( monthly['Mes_Label'], monthly['PnL_Total'], color=colores, width=0.6, zorder=3, edgecolor='#0f3460', linewidth=0.8 ) # Linea en cero ax1.axhline(0, color='#aaaaaa', linewidth=0.8, linestyle='--', zorder=2) # Anotaciones sobre cada barra: valor P/L y numero de trades for bar, pnl, n_trades in zip(bars, monthly['PnL_Total'], monthly['Num_Trades']): altura = bar.get_height() rango = monthly['PnL_Total'].abs().max() or 1 offset_pnl = rango * 0.03 offset_trades = rango * 0.10 if altura >= 0: y_pnl = altura + offset_pnl y_trades = altura + offset_trades va = 'bottom' else: y_pnl = altura - offset_pnl y_trades = altura - offset_trades va = 'top' ax1.text( bar.get_x() + bar.get_width() / 2, y_pnl, f"${pnl:,.0f}", ha='center', va=va, fontsize=8.5, fontweight='bold', color='white', zorder=5 ) ax1.text( bar.get_x() + bar.get_width() / 2, y_trades, f"{n_trades} trades", ha='center', va=va, fontsize=7.5, color='#cccccc', zorder=5 ) # Formato ejes ax1.set_xlabel("Mes", color='#cccccc', fontsize=10) ax1.set_ylabel("P/L Neto ($)", color='#cccccc', fontsize=10) ax1.tick_params(axis='x', colors='#cccccc', rotation=35, labelsize=8.5) ax1.tick_params(axis='y', colors='#cccccc', labelsize=9) ax1.yaxis.set_major_formatter(mticker.FuncFormatter(lambda x, _: f"${x:,.0f}")) for spine in ax1.spines.values(): spine.set_edgecolor('#0f3460') ax1.grid(axis='y', color='#0f3460', linewidth=0.6, zorder=1) # Ampliar ylim para que las anotaciones no queden cortadas ymin, ymax = ax1.get_ylim() ax1.set_ylim(ymin * 1.22, ymax * 1.22) # Titulo profit_total = monthly['PnL_Total'].sum() total_trades = monthly['Num_Trades'].sum() titulo = ( f"{symbol} | {estrategia} | {riesgo.upper()} | Hora: {modo_hora}\n" f"Rendimiento Mensual (neto comisiones) " f"Profit Total: ${profit_total:,.2f} " f"Trades: {total_trades} " f"CT: ${credito_target} " f"Com/trade: ${comision_x_trade:.2f}" ) ax1.set_title(titulo, color='white', fontsize=10, pad=14) plt.tight_layout() img_file = output_file.replace(".csv", "_rendimiento_mensual.jpg") plt.savefig(img_file, dpi=150, bbox_inches='tight', facecolor=fig.get_facecolor(), format='jpg') plt.close() print(f"Grafico guardado en: {img_file}\n") # ========================= EJECUCION ========================= if HORA_FIJA: ventanas = [] print(f"Modo HORA FIJA activado: {HORA_FIJA}") print(f"(Se ignora el archivo: {ARCHIVO_TOP_HORARIOS})\n") else: ventanas = cargar_ventanas_top_horarios(ARCHIVO_TOP_HORARIOS, rank_objetivo=1) print("Ventanas de aplicacion (ultimas 5):") for v in ventanas[-5:]: print(f" Desde {v[0].date()} hasta {v[1].date()} -> hora {v[2]}") print() df_resultados = backtest_unificado( ventanas, desde, hasta, symbol, estrategia, desplazamiento, comision_x_trade ) if df_resultados.empty: print("\n" + "="*70) print(" SIN DATOS: No se encontraron fechas de trading en el rango indicado.") print(f" Verificar que existan archivos de strikes para '{symbol}' / hora '{modo_hora}'") print(f" y que el rango {desde} - {hasta} tenga datos disponibles.") print("="*70 + "\n") exit(0) profit_total = df_resultados['P/L'].sum() comisiones_totales = df_resultados['Comision'].sum() wins = (df_resultados['P/L'] > 0).sum() losses = (df_resultados['P/L'] < 0).sum() no_trades = (df_resultados['Exit'] == "NO_ENTRY").sum() win_rate = (wins / (wins + losses) * 100) if (wins + losses) > 0 else 0 df_console = df_resultados.copy() df_console['P/L'] = df_console['P/L'].apply(colorize) print(df_console.to_string(index=False)) print("\n================ RESUMEN =================") print(f"Estrategia: {estrategia}") print(f"Modo Hora: {modo_hora}") print(f"Profit Total (neto): ${profit_total:.2f}") print(f"Comisiones Totales: ${comisiones_totales:.2f}") print(f"Wins: {wins} | Losses: {losses}") print(f"No Trade Days: {no_trades}") print(f"Win Rate: {win_rate:.2f}%") print(f"Credito Target: ${CREDITO_TARGET}") print(f"Comision x Trade: ${comision_x_trade:.2f} ({2 if estrategia == 'Vertical' else 4} patas x ${COMISION_POR_PATA})") # ========================= GUARDAR CSV ========================= os.makedirs("resultados", exist_ok=True) separator = pd.DataFrame([[""] * len(df_resultados.columns)], columns=df_resultados.columns) summary_data = [ ("===== SUMMARY =====", ""), ("Estrategia", estrategia), ("Riesgo", riesgo), ("Hora", modo_hora), ("Profit Total (neto)", round(profit_total, 2)), ("Comisiones Totales", round(comisiones_totales, 2)), ("Comision x Trade", comision_x_trade), ("Comision x Pata", COMISION_POR_PATA), ("Wins", wins), ("Losses", losses), ("No Trade Days", no_trades), ("Win Rate (%)", round(win_rate, 2)), ("Credito Target", CREDITO_TARGET), ("Generado", datetime.now().strftime("%Y-%m-%d %H:%M:%S")) ] summary_rows = [] for label, value in summary_data: row = {col: "" for col in df_resultados.columns} row[df_resultados.columns[0]] = label row[df_resultados.columns[-1]] = value summary_rows.append(row) df_summary = pd.DataFrame(summary_rows) df_final = pd.concat([df_resultados, separator, df_summary], ignore_index=True) hora_label = HORA_FIJA if HORA_FIJA else "DINAMICA" output_file = ( f"resultados/P&L_{estrategia}_{symbol}_HORA_{hora_label}_" f"{desde.replace('-','')}_{hasta.replace('-','')}_{riesgo}_CT_{CREDITO_TARGET}.csv" ) df_final.to_csv(output_file, index=False) print(f"\nCSV guardado en: {output_file}") # ========================= GRAFICO ========================= generar_grafico_mensual( df_resultados, symbol, estrategia, riesgo, modo_hora, CREDITO_TARGET, comision_x_trade, output_file )