# import pandas as pd # import matplotlib.pyplot as plt # 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 GENERAL ========================= # PATH_UBUNTU = "/var/www/html/flask_project/" # desde = "2025-11-01" # hasta = "2026-02-13" # symbol = "SPX" # timeHour = "1020" # desplazamiento = -15 # estrategia = "Vertical" # # ===== NUEVA VARIABLE CLAVE ===== # CREDITO_TARGET = 180 # dólares # elArchivo = f"/var/www/html/backtestingmarket/predictor_data/makekos/{symbol}/{symbol}_{estrategia}_strikes_{timeHour}.csv" # # ========================= FUNCIONES ========================= # def buscar_entrada_por_credito( # archivo_option_chain, # strike1, # strike2, # option_type, # entry_time, # credito_target # ): # """ # Busca el PRIMER snapshot >= entry_time # donde el crédito del spread >= credito_target. # Devuelve: # credit_entry (float), # entry_timestamp (datetime) # o (None, None) si nunca se alcanza. # """ # 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 # 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 # credit_dollars = (p1 - p2) * 100 # if credit_dollars >= credito_target: # return p1 - p2, ts # 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 # # ========================= BACKTEST PRINCIPAL ========================= # def backtest_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() # 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) # # ===== SI NO HUBO ENTRADA ===== # 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 # # ===== SI HUBO ENTRADA ===== # 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: # PUT # 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 # 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) # }) # df = pd.DataFrame(resultados) # return df # # ========================= EJECUCIÓN ========================= # df_strikes = pd.read_csv(elArchivo) # df_resultados = backtest_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() # 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}") # # ========================= GUARDAR CSV ========================= # df_resultados.to_csv( # f"resultados/P&L_{estrategia}_{symbol}_{timeHour}_{desde.replace('-','')}_{hasta.replace('-','')}_D{desplazamiento}_CT_{CREDITO_TARGET}.csv", # index=False # ) import pandas as pd import matplotlib.pyplot as plt import math 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 ========================= # ===== MAPEO RIESGO ===== DESPLAZAMIENTOS = { "cons": 0, "inte": -5, "agre": -10, "ultr": -15 } desde = "2025-03-01" hasta = "2026-02-17" symbol = "XSP" timeHour = "1245" estrategia = "Vertical" riesgo = "agre" CREDITO_TARGET = 50 # dólares # 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 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_por_credito( 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 # ← flag para saber si es el timestamp exacto de entrada 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 credit_dollars = (p1 - p2) * 100 if credit_dollars >= credito_target: if es_primera_hora: return p1 - p2, ts # ← hora exacta: crédito real del mercado else: return credito_target / 100, ts # ← hora posterior: 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 # ========================= BACKTEST PRINCIPAL ========================= def backtest_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() df_strikes['Strike1_Desplazado'] = df_strikes.apply( lambda r: r['Strike1'] + desplazamiento if r['Option_Type'] == 'CALL' else r['Strike1'] - desplazamiento, axis=1 ).astype(int) df_strikes['Strike2_Desplazado'] = df_strikes.apply( lambda r: r['Strike2'] + desplazamiento if r['Option_Type'] == 'CALL' else r['Strike2'] - desplazamiento, axis=1 ).astype(int) 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, "ClosePrice": close_price, # ← NUEVA COLUMNA "Exit": "NO_ENTRY", "P/L": 0.0 }) continue 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 resultados.append({ "Day": fecha, "Time": hora, "Strike1": strike1, "Strike2": strike2, "Option": option_type, "EntryTime": entry_ts, "EntryCredit": round(credit_entry * 100, 2), "ClosePrice": close_price, # ← NUEVA COLUMNA "Exit": "CLOSE", "P/L": round(pnl, 2) }) return pd.DataFrame(resultados) # ========================= EJECUCIÓN ========================= df_strikes = pd.read_csv(elArchivo) df_resultados = backtest_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")