import pandas as pd import matplotlib.pyplot as plt import math # ========================= CONFIGURACIÓN ========================= # Paths 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 # Mapeo de desplazamiento por riesgo DESPLAZAMIENTOS = { "cons": 0, "inte": -5, "agre": -10, "ultr": -15 } # Parámetros ************************************************************** desde = "2025-10-01" hasta = "2026-02-13" symbol = "XSP" timeHour = "1245" riesgo = "agre" # cons | inte | agre | ultr # TP / SL USE_TP_SL = True TP = 100 # dolares SL = 100 # dolares # ************************************************************** desplazamiento = DESPLAZAMIENTOS.get(riesgo, 0) estrategia = "Vertical" print("\n" + "═"*60) print("📊 BACKTEST CONFIGURATION".center(60)) print("═"*60) print(f"{'Fecha Desde:':20} {desde}") print(f"{'Fecha Hasta:':20} {hasta}") print(f"{'Símbolo:':20} {symbol}") print(f"{'Hora Entrada:':20} {timeHour}") print(f"{'Estrategia:':20} {estrategia}") print(f"{'Riesgo:':20} {riesgo}") print("─"*60) print("💰 TAKE PROFIT / STOP LOSS") print(f"{'Activado:':20} {USE_TP_SL}") print(f"{'Take Profit:':20} ${TP}") print(f"{'Stop Loss:':20} ${SL}") print("═"*60 + "\n") print("Iniciando backtest...") elArchivo = f"/var/www/html/backtestingmarket/predictor_data/makekos/{symbol}/{symbol}_{estrategia}_strikes_{timeHour}.csv" # ========================= FUNCIONES ========================= def calcular_credito_y_underlying_desde_archivo_csv( archivo_option_chain, strike1, strike2, option_type, timeHour ): df = pd.read_csv(archivo_option_chain, usecols=range(9)) df['timestamp'] = pd.to_datetime(df['timestamp']) df = df[df['timestamp'] >= timeHour] if df.empty: return None, None s1 = df[df['strike'] == strike1].iloc[0] s2 = df[df['strike'] == strike2].iloc[0] if option_type == "CALL": p1 = (s1['bid_call'] + s1['ask_call']) / 2 p2 = (s2['bid_call'] + s2['ask_call']) / 2 else: p1 = (s1['bid_put'] + s1['ask_put']) / 2 p2 = (s2['bid_put'] + s2['ask_put']) / 2 return p1 - p2, s1['underlying_price'] 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 simular_tp_sl( archivo_option_chain, strike1, strike2, option_type, entry_time, credit, TP, SL ): 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, None for ts, snap in df.groupby('timestamp'): s1 = snap[snap['strike'] == strike1] s2 = snap[snap['strike'] == strike2] if s1.empty or s2.empty: 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 spread_price = p1 - p2 pnl = (credit - spread_price) * 100 if pnl >= TP: return TP, "TP", ts if pnl <= -SL: return -SL, "SL", ts return None, None, None # ========================= BACKTEST PRINCIPAL ========================= def calcular_precios_verticales(df_strikes, path_base, desplazamiento, desde, hasta): resultados = [] symbol = df_strikes['Symbol'].iloc[0] df_strikes['Day'] = pd.to_datetime(df_strikes['Day']) df_strikes = df_strikes[(df_strikes['Day'] >= desde) & (df_strikes['Day'] <= hasta)].copy() df_strikes['Strike1_Desplazado'] = df_strikes.apply( lambda r: r['Strike1'] + desplazamiento if r['Option_Type'] == 'CALL' else r['Strike1'] - desplazamiento, axis=1 ) df_strikes['Strike2_Desplazado'] = df_strikes.apply( lambda r: r['Strike2'] + desplazamiento if r['Option_Type'] == 'CALL' else r['Strike2'] - desplazamiento, axis=1 ) for _, r in df_strikes.iterrows(): fecha = r['Day'].strftime("%Y-%m-%d") hora = r['Hour'] strike1 = r['Strike1_Desplazado'] strike2 = r['Strike2_Desplazado'] option_type = r['Option_Type'] entry_time = pd.to_datetime(f"{r['Day']} {hora}") if symbol in ["SPX", "RUT", "XSP"]: archivo = f"{path_base}chains/optionChain_${symbol}_{fecha}.csv" else: archivo = f"{path_base}chains/optionChain_{symbol}_{fecha}.csv" credit, price_entry = calcular_credito_y_underlying_desde_archivo_csv( archivo, strike1, strike2, option_type, entry_time ) close_price = obtener_ultimo_underlying_price(archivo) exit_reason = "CLOSE" exit_time = None pnl = None if USE_TP_SL and credit is not None: pnl, exit_reason, exit_time = simular_tp_sl( archivo, strike1, strike2, option_type, entry_time, credit, TP, SL ) if pnl is None: spread = abs(strike1 - strike2) if option_type == "CALL": if close_price > max(strike1, strike2): pnl = (credit - spread) * 100 elif close_price < min(strike1, strike2): pnl = credit * 100 else: pnl = credit * 100 - (close_price - min(strike1, strike2)) * 100 else: if close_price < min(strike1, strike2): pnl = (credit - spread) * 100 elif close_price > max(strike1, strike2): pnl = credit * 100 else: pnl = credit * 100 - (max(strike1, strike2) - close_price) * 100 resultados.append({ "Day": fecha, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "Credit": round(credit * 100, 2) if credit is not None else None, "EntryPrice": price_entry, "ClosePrice": close_price, "P/L": round(pnl, 2), "Exit": exit_reason, "ExitTime": exit_time }) df = pd.DataFrame(resultados) return df, df['P/L'].sum(), (df['P/L'] > 0).sum(), (df['P/L'] < 0).sum() # ========================= EJECUCIÓN ========================= df_strikes = pd.read_csv(elArchivo) df, profit_total, wins, losses = calcular_precios_verticales( df_strikes, PATH_UBUNTU, desplazamiento, desde, hasta ) df_console = df.copy() df_console['P/L'] = df_console['P/L'].apply(colorize) print(df_console.to_string(index=False)) win_rate = (wins / (wins + losses) * 100) if (wins + losses) > 0 else 0 print(f"\nProfit Total: ${profit_total:.2f}") print(f"Wins: {wins} | Losses: {losses}") print(f"Win Rate: {win_rate:.2f}%") print(f"TP: {TP} | SL: {SL}") # ========================= OUTPUTS ========================= # Crear copia para exportar df_export = df.copy() # Línea separadora separator = pd.DataFrame([[""] * len(df_export.columns)], columns=df_export.columns) # Crear filas resumen usando la primera y última columna dinámicamente summary_data = [ ("===== SUMMARY =====", ""), ("Profit Total", round(profit_total, 2)), ("Wins", wins), ("Losses", losses), ("Win Rate (%)", round(win_rate, 2)), ("TP", TP), ("SL", SL), ("Riesgo", riesgo), ] summary_rows = [] for label, value in summary_data: row = {col: "" for col in df_export.columns} row[df_export.columns[0]] = label row[df_export.columns[-2]] = value summary_rows.append(row) df_summary = pd.DataFrame(summary_rows) df_final = pd.concat([df_export, separator, df_summary], ignore_index=True) df_final.to_csv( f"resultados/P&L_{estrategia}_{symbol}_{timeHour}_{desde.replace('-','')}_{hasta.replace('-','')}_{riesgo}_TPSL.csv", index=False )