import pandas as pd import os from datetime import datetime 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 # ========================= CONFIGURACIÓN GENERAL ========================= DESPLAZAMIENTOS = { "cons": 0, "inte": -5, "agre": -10, "ultr": -15 } desde = "2025-03-01" hasta = "2026-02-17" symbol = "XSP" timeHour = "1025" estrategia = "IronCondor" riesgo = "agre" CREDITO_TARGET = 50 # dólares (suma de call spread + put spread) # Si es símbolo pequeño, divide por 5 factor = 5 if symbol in ["SPY", "QQQ", "XSP"] else 1 desplazamiento = DESPLAZAMIENTOS.get(riesgo, 0) / factor elArchivo = f"/var/www/html/backtestingmarket/predictor_data/makekos/{symbol}/{symbol}_{estrategia}_strikes_{timeHour}.csv" # ========================= PRINT PARÁMETROS ========================= print("\n" + "═"*70) print("📊 BTM IC CREDIT TARGET BACKTEST".center(70)) print("═"*70) print(f"{'Fecha Desde:':25} {desde}") print(f"{'Fecha Hasta:':25} {hasta}") print(f"{'Símbolo:':25} {symbol}") print(f"{'Estrategia:':25} {estrategia}") print(f"{'Hora Entrada:':25} {timeHour}") print(f"{'Riesgo:':25} {riesgo}") print(f"{'Credito Target ($):':25} {CREDITO_TARGET}") print("═"*70 + "\n") print("Iniciando backtest...\n") # ========================= FUNCIONES ========================= def buscar_entrada_ic_por_credito( archivo_option_chain, call_strike1, call_strike2, put_strike1, put_strike2, 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 # ← flag for ts, snap in df.groupby('timestamp'): cs1 = snap[snap['strike'] == call_strike1] cs2 = snap[snap['strike'] == call_strike2] ps1 = snap[snap['strike'] == put_strike1] ps2 = snap[snap['strike'] == put_strike2] if cs1.empty or cs2.empty or ps1.empty or ps2.empty: continue call_mid1 = (cs1['bid_call'].iloc[0] + cs1['ask_call'].iloc[0]) / 2 call_mid2 = (cs2['bid_call'].iloc[0] + cs2['ask_call'].iloc[0]) / 2 put_mid1 = (ps1['bid_put'].iloc[0] + ps1['ask_put'].iloc[0]) / 2 put_mid2 = (ps2['bid_put'].iloc[0] + ps2['ask_put'].iloc[0]) / 2 credit_call = call_mid1 - call_mid2 credit_put = put_mid1 - put_mid2 credit_total_dollars = (credit_call + credit_put) * 100 if credit_total_dollars >= credito_target: if es_primera_hora: return (credit_call + credit_put), ts # ← crédito real del mercado else: return credito_target / 100, ts # ← exactamente el target es_primera_hora = False # ← a partir del segundo timestamp, ya no es la primera return None, None 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 calcular_pnl_ic(credit_entry, close_price, call_strike1, call_strike2, put_strike1, put_strike2): """ Calcula el P/L del Iron Condor al cierre. Call spread (bear call): vendido call_strike1, comprado call_strike2 - Máx ganancia si close < call_strike1 - Máx pérdida si close > call_strike2 - Pérdida parcial entre los dos strikes Put spread (bull put): vendido put_strike1, comprado put_strike2 - Máx ganancia si close > put_strike1 - Máx pérdida si close < put_strike2 - Pérdida parcial entre los dos strikes credit_entry = crédito total por contrato (en puntos, no dólares) """ call_spread = abs(call_strike2 - call_strike1) put_spread = abs(put_strike1 - put_strike2) # --- CALL SPREAD P/L --- if close_price <= call_strike1: pnl_call = 0 # máx ganancia (se cuenta en crédito) elif close_price >= call_strike2: pnl_call = -call_spread # pérdida total del call spread else: pnl_call = -(close_price - call_strike1) # pérdida parcial # --- PUT SPREAD P/L --- if close_price >= put_strike1: pnl_put = 0 # máx ganancia elif close_price <= put_strike2: pnl_put = -put_spread # pérdida total del put spread else: pnl_put = -(put_strike1 - close_price) # pérdida parcial # P/L total en dólares pnl_total = (credit_entry + pnl_call + pnl_put) * 100 return round(pnl_total, 2) # ========================= BACKTEST PRINCIPAL ========================= def backtest_ic_credito_target(df_strikes): 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() # Aplicar desplazamiento simétrico # Calls se desplazan hacia ARRIBA (- desplazamiento porque desplazamiento es negativo para agre) # Puts se desplazan hacia ABAJO (+ desplazamiento) df_strikes['CS1'] = (df_strikes['Call_Strike1'] + desplazamiento).astype(int) df_strikes['CS2'] = (df_strikes['Call_Strike2'] + desplazamiento).astype(int) df_strikes['PS1'] = (df_strikes['Put_Strike1'] - desplazamiento).astype(int) df_strikes['PS2'] = (df_strikes['Put_Strike2'] - desplazamiento).astype(int) for _, r in df_strikes.iterrows(): fecha = r['Day'].strftime("%Y-%m-%d") hora = r['Hour'] cs1 = r['CS1'] cs2 = r['CS2'] ps1 = r['PS1'] ps2 = r['PS2'] entry_time = pd.to_datetime(f"{r['Day']} {hora}") if symbol in ["SPX", "RUT", "XSP"]: archivo = f"{PATH_UBUNTU}chains/optionChain_${symbol}_{fecha}.csv" else: archivo = f"{PATH_UBUNTU}chains/optionChain_{symbol}_{fecha}.csv" credit_entry, entry_ts = buscar_entrada_ic_por_credito( archivo, cs1, cs2, ps1, ps2, entry_time, CREDITO_TARGET ) close_price = obtener_ultimo_underlying_price(archivo) if credit_entry is None: resultados.append({ "Day": fecha, "Time": hora, "Call_Strike1": cs1, "Call_Strike2": cs2, "Put_Strike1": ps1, "Put_Strike2": ps2, "EntryTime": None, "EntryCredit": None, "ClosePrice": close_price, "Exit": "NO_ENTRY", "P/L": 0.0 }) continue pnl = calcular_pnl_ic(credit_entry, close_price, cs1, cs2, ps1, ps2) resultados.append({ "Day": fecha, "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", "P/L": pnl }) return pd.DataFrame(resultados) # ========================= EJECUCIÓN ========================= df_strikes = pd.read_csv(elArchivo) df_resultados = backtest_ic_credito_target(df_strikes) profit_total = df_resultados['P/L'].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"Profit Total: ${profit_total:.2f}") print(f"Wins: {wins} | Losses: {losses}") print(f"No Trade Days: {no_trades}") print(f"Win Rate: {win_rate:.2f}%") # ========================= GUARDAR CSV ========================= os.makedirs("resultados", exist_ok=True) separator = pd.DataFrame([[""] * len(df_resultados.columns)], columns=df_resultados.columns) summary_data = [ ("===== SUMMARY =====", ""), ("Riesgo", riesgo), ("Profit Total", round(profit_total, 2)), ("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) df_final.to_csv( f"resultados/P&L_{estrategia}_{symbol}_{timeHour}_{desde.replace('-','')}_{hasta.replace('-','')}_{riesgo}_CT_{CREDITO_TARGET}.csv", index=False ) print("\nCSV guardado correctamente.\n")