import pandas as pd import math 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 ========================= PATH_UBUNTU = "/var/www/html/flask_project/" desde = "2025-10-01" hasta = "2025-12-29" symbol = "SPX" timeHour = "1245" desplazamiento = -15 estrategia = "Vertical" CREDITO_TARGET = 170 # dólares TP = 150 # dólares USE_TP = True elArchivo = f"/var/www/html/backtestingmarket/predictor_data/makekos/{symbol}/{symbol}_{estrategia}_strikes_{timeHour}.csv" # ========================= FUNCIONES ========================= def buscar_entrada_por_credito(archivo, strike1, strike2, option_type, entry_time, credito_target): try: df = pd.read_csv(archivo) df['timestamp'] = pd.to_datetime(df['timestamp']) df = df[df['timestamp'] >= entry_time] except: return 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 if (p1 - p2) * 100 >= credito_target: return p1 - p2, ts return None, None def evaluar_salida_tp( archivo, strike1, strike2, option_type, entry_ts, credit_entry, TP ): try: df = pd.read_csv(archivo) df['timestamp'] = pd.to_datetime(df['timestamp']) df = df[df['timestamp'] >= entry_ts] except: return 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 pnl = (credit_entry - (p1 - p2)) * 100 if pnl >= TP: return TP, "TP", ts return None, None, None def obtener_ultimo_underlying_price(archivo): try: df = pd.read_csv(archivo) df['timestamp'] = pd.to_datetime(df['timestamp']) return df.sort_values('timestamp')['underlying_price'].iloc[-1] except: return None # ========================= BACKTEST ========================= def backtest_credito_target_tp(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() 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_UBUNTU}chains/optionChain_${symbol}_{fecha}.csv" else: archivo = f"{PATH_UBUNTU}chains/optionChain_{symbol}_{fecha}.csv" credit_entry, entry_ts = buscar_entrada_por_credito( archivo, strike1, strike2, option_type, entry_time, CREDITO_TARGET ) close_price = obtener_ultimo_underlying_price(archivo) if credit_entry is None: resultados.append({ "Day": fecha, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "EntryTime": None, "EntryCredit": None, "Exit": "NO_ENTRY", "P/L": 0.0 }) continue if USE_TP: pnl_tp, exit_reason, exit_ts = evaluar_salida_tp( archivo, strike1, strike2, option_type, entry_ts, credit_entry, TP ) if pnl_tp is not None: resultados.append({ "Day": fecha, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "EntryTime": entry_ts, "EntryCredit": round(credit_entry * 100, 2), "Exit": "TP", "P/L": pnl_tp }) continue spread = abs(strike1 - strike2) if option_type == "CALL": strike_venta, strike_compra = min(strike1, strike2), 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, strike_compra = max(strike1, strike2), 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 resultados.append({ "Day": fecha, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "EntryTime": entry_ts, "EntryCredit": round(credit_entry * 100, 2), "Exit": "CLOSE", "P/L": round(pnl, 2) }) return pd.DataFrame(resultados) # ========================= EJECUCIÓN ========================= df_strikes = pd.read_csv(elArchivo) df_resultados = backtest_credito_target_tp(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() 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"Credito Target: ${CREDITO_TARGET} | TP: ${TP}") df_resultados.to_csv( f"resultados/P&L_{estrategia}_{symbol}_{timeHour}_{desde.replace('-','')}_{hasta.replace('-','')}_D{desplazamiento}_CREDTARGET_TP.csv", index=False )