import pandas as pd import matplotlib.pyplot as plt import math # pd.set_option('display.max_rows', None) # Definir códigos ANSI para colores RED = "\033[91m" GREEN = "\033[92m" RESET = "\033[0m" # Función para aplicar color def colorize(value): if isinstance(value, (int, float)): # Asegurarse de que es un número color = GREEN if value >= 0 else RED return f"{color}{value}{RESET}" return value # Variables de entorno PATH_UBUNTU = "/var/www/html/flask_project/" # Variables modificables desde = "2025-02-01" # Fecha inicial en formato 'YYYY-MM-DD' hasta = "2025-05-02" # Fecha final en formato 'YYYY-MM-DD' symbol = "SPX" # Símbolo del activo timeHour = '1340' desplazamiento = -10 estrategia = "Vertical" elArchivo = '/var/www/html/backtestingmarket/predictor_data/makekos/' + symbol + '/' + symbol + '_' + estrategia + '_strikes_' + timeHour + '.csv' # Función para calcular el crédito recibido y el precio subyacente desde un archivo CSV def calcular_credito_y_underlying_desde_archivo_csv(archivo_option_chain, strike1, strike2, option_type, timeHour): option_chain_df = pd.read_csv(archivo_option_chain, usecols=range(9)) option_chain_df['timestamp'] = pd.to_datetime(option_chain_df['timestamp']) filtered_chain = option_chain_df[option_chain_df['timestamp'] >= timeHour] if filtered_chain.empty: return "No data found after the specified time", None strikes_data = filtered_chain[(filtered_chain['strike'].isin([strike1, strike2]))] if option_type == 'PUT': strikes_data = strikes_data[["strike", "bid_put", "ask_put", "underlying_price"]] elif option_type == 'CALL': strikes_data = strikes_data[["strike", "bid_call", "ask_call", "underlying_price"]] if strikes_data.empty: return "Strikes not found in the filtered data", None try: strike1_data = strikes_data[strikes_data['strike'] == strike1].iloc[0] strike2_data = strikes_data[strikes_data['strike'] == strike2].iloc[0] except: return 0, None if option_type == 'PUT': strike1_avg = (strike1_data['bid_put'] + strike1_data['ask_put']) / 2 strike2_avg = (strike2_data['bid_put'] + strike2_data['ask_put']) / 2 elif option_type == 'CALL': strike1_avg = (strike1_data['bid_call'] + strike1_data['ask_call']) / 2 strike2_avg = (strike2_data['bid_call'] + strike2_data['ask_call']) / 2 credit_received = strike1_avg - strike2_avg underlying_price = strike1_data['underlying_price'] return credit_received, underlying_price def obtener_ultimo_underlying_price(archivo_option_chain): try: # Leer el archivo CSV option_chain_df = pd.read_csv(archivo_option_chain) # Asegurarse de que 'timestamp' sea de tipo datetime option_chain_df['timestamp'] = pd.to_datetime(option_chain_df['timestamp']) # Ordenar por timestamp para garantizar el último registro option_chain_df = option_chain_df.sort_values(by='timestamp') # Obtener el último precio del subyacente if 'underlying_price' in option_chain_df.columns: ultimo_underlying_price = option_chain_df['underlying_price'].iloc[-1] return ultimo_underlying_price else: return None except Exception as e: return None # Función para iterar sobre los días y calcular el crédito para cada vertical def calcular_precios_verticales(vertical_strikes_df, path_csv_base, desplazamiento, desde, hasta): resultados = [] symbol = vertical_strikes_df['Symbol'].iloc[0] # Filtrar el DataFrame de strikes para incluir solo las fechas dentro del rango vertical_strikes_df['Day'] = pd.to_datetime(vertical_strikes_df['Day']) vertical_strikes_df = vertical_strikes_df[(vertical_strikes_df['Day'] >= pd.to_datetime(desde)) & (vertical_strikes_df['Day'] <= pd.to_datetime(hasta))].copy() # Aplicar el desplazamiento a los strikes según el tipo de opción usando .loc vertical_strikes_df.loc[:, 'Strike1_Desplazado'] = vertical_strikes_df.apply( lambda row: row['Strike1'] + desplazamiento if row['Option_Type'] == 'CALL' else row['Strike1'] - desplazamiento, axis=1 ) vertical_strikes_df.loc[:, 'Strike2_Desplazado'] = vertical_strikes_df.apply( lambda row: row['Strike2'] + desplazamiento if row['Option_Type'] == 'CALL' else row['Strike2'] - desplazamiento, axis=1 ) for index, row in vertical_strikes_df.iterrows(): fecha = pd.to_datetime(row['Day']).strftime('%Y-%m-%d') hora = row['Hour'] strike1 = row['Strike1_Desplazado'] strike2 = row['Strike2_Desplazado'] option_type = row['Option_Type'] if option_type == 'CALL': strike_venta = min(strike1, strike2) strike_compra = max(strike1, strike2) elif option_type == 'PUT': strike_venta = max(strike1, strike2) strike_compra = min(strike1, strike2) specific_time = pd.to_datetime(f"{row['Day']} {hora}") if symbol == 'SPX' or symbol == 'RUT' or symbol == 'XSP': archivo_option_chain = path_csv_base + "chains/optionChain_$" + symbol + '_' + fecha + ".csv" else: archivo_option_chain = path_csv_base + "chains/optionChain_" + symbol + '_' + fecha + ".csv" try: credit_result, underlying_price_specific_time = calcular_credito_y_underlying_desde_archivo_csv(archivo_option_chain, strike1, strike2, option_type, specific_time) except FileNotFoundError: credit_result = 'File not found' underlying_price_specific_time = None try: underlying_price_16 = obtener_ultimo_underlying_price(archivo_option_chain) except: underlying_price_16 = None spreadDiff = abs(strike1-strike2) if underlying_price_16 is not None and underlying_price_specific_time is not None: if option_type == 'CALL': if underlying_price_16 > strike_compra: ganancia_dia = (credit_result - spreadDiff) * 100 elif underlying_price_16 < strike_venta: ganancia_dia = credit_result * 100 else: valor_itm = (underlying_price_16 - strike_venta) * 100 ganancia_dia = credit_result * 100 - valor_itm elif option_type == 'PUT': if underlying_price_16 < strike_compra: ganancia_dia = (credit_result - spreadDiff) * 100 elif underlying_price_16 > strike_venta: ganancia_dia = credit_result * 100 else: valor_itm = (strike_venta - underlying_price_16) * 100 ganancia_dia = credit_result * 100 - valor_itm else: ganancia_dia = 'No calculable' itmOtm = "" # Inicialización # Verifica si los valores son válidos antes de operar if underlying_price_specific_time is not None and strike_venta is not None: if option_type == 'CALL': itmOtm = "ITM" if (underlying_price_specific_time - strike_venta) > 0 else "OTM" else: itmOtm = "OTM" if (underlying_price_specific_time - strike_venta) > 0 else "ITM" else: itmOtm = "ERROR" # O maneja el caso según lo que necesites resultados.append({ 'Day': fecha, 'Time': hora, 'Strike1': strike1, 'Strike2': strike2, 'Option': option_type, 'Credit': f"$ {0.00:.2f}" if not isinstance(credit_result, (int, float)) or math.isnan(credit_result) else f"$ {credit_result * 100:.2f}", 'Price': underlying_price_specific_time, # Precio en specific_time 'Close': underlying_price_16, 'P/L': ganancia_dia, 'Dif': abs(underlying_price_specific_time - strike_venta) if underlying_price_specific_time is not None else None, 'itmOtm': itmOtm }) resultados_df = pd.DataFrame(resultados) resultados_df['P/L'] = pd.to_numeric(resultados_df['P/L'], errors='coerce') resultados_df['P/L'] = resultados_df['P/L'].map(lambda x: round(x, 2) if pd.notnull(x) else x) profit_total = resultados_df['P/L'].sum() dias_profit_positivo = resultados_df[resultados_df['P/L'] > 0].shape[0] dias_profit_negativo = resultados_df[resultados_df['P/L'] < 0].shape[0] return resultados_df, profit_total, dias_profit_positivo, dias_profit_negativo # ========================== CARGA DE DATOS ========================== # Leer el archivo CSV con los strikes para la estrategia vertical_strikes_df = pd.read_csv(elArchivo) # Calcular precios de la estrategia en el rango de fechas seleccionado resultados_verticales, profit_total, dias_profit_positivo, dias_profit_negativo = calcular_precios_verticales( vertical_strikes_df, PATH_UBUNTU, desplazamiento, desde, hasta ) # ========================== VISUALIZACIÓN EN CONSOLA ========================== # Crear una copia para imprimir en consola con color resultados_consola = resultados_verticales.copy() resultados_consola['P/L'] = resultados_consola['P/L'].apply(colorize) # Mostrar los resultados en consola con formato print(resultados_consola.to_string(index=False)) print(f"\nSímbolo: {symbol}") print(f"Profit Total: $ {profit_total:.2f}") print(f"Días con profit positivo: {dias_profit_positivo}") print(f"Días con profit negativo: {dias_profit_negativo}") # Calcular y mostrar el Win Rate en porcentaje total_dias = dias_profit_positivo + dias_profit_negativo win_rate = (dias_profit_positivo / total_dias * 100) if total_dias > 0 else None print(f"Win Rate: {win_rate:.2f}%" if win_rate is not None else "Win Rate: No se puede calcular (sin días registrados)") print(f"Desplazamiento: {desplazamiento}") print(f"Estrategia: {estrategia}") # Guardar los resultados en un archivo CSV output_csv = f"resultados/P&L_{estrategia}_{symbol}_{timeHour}_{desde.replace('-', '')}_{hasta.replace('-', '')}_D{desplazamiento}.csv" resultados_verticales.to_csv(output_csv, index=False) # ========================== PROCESAMIENTO DE DATOS PARA GRÁFICOS ========================== # Asegurar que la columna 'Day' es de tipo datetime resultados_verticales['Day'] = pd.to_datetime(resultados_verticales['Day'], errors='coerce') # Manejar valores nulos en 'Day' if resultados_verticales['Day'].isnull().any(): print("Advertencia: Se encontraron valores no válidos en 'Day'. Se eliminarán las filas afectadas.") resultados_verticales = resultados_verticales.dropna(subset=['Day']) # Convertir 'P/L' a numérico resultados_verticales['P/L'] = pd.to_numeric(resultados_verticales['P/L'], errors='coerce') # ========================== GRÁFICO 1: GANANCIAS MENSUALES ========================== # Agrupar y sumar ganancias por mes resultados_verticales['Month'] = resultados_verticales['Day'].dt.to_period('M') ganancias_mensuales = resultados_verticales.groupby('Month')['P/L'].sum() # Crear gráfico de barras plt.figure(figsize=(10, 6)) plt.bar(ganancias_mensuales.index.astype(str), ganancias_mensuales.values) plt.title(f"Profit & Loss - {estrategia} {symbol}\n{desde} a {hasta}, Time {timeHour}, D{desplazamiento}") plt.xlabel('Mes') plt.ylabel('Ganancia Total ($)') plt.xticks(rotation=45) plt.grid(axis='y', linestyle='--', alpha=0.7) plt.legend([f"Profit Total: ${profit_total:.2f}"], loc='upper left', fontsize=10, frameon=True) plt.tight_layout() # Guardar gráfico output_filename = f"graficos/P&L Mensual_{estrategia}_{symbol}_{timeHour}_{desde.replace('-', '')}_{hasta.replace('-', '')}_D{desplazamiento}.png" plt.savefig(output_filename) # ========================== GRÁFICO 2: GANANCIAS ACUMULADAS ========================== # Calcular la ganancia acumulada resultados_verticales['Ganancia_Acumulada'] = resultados_verticales['P/L'].cumsum() # Crear gráfico de líneas plt.figure(figsize=(10, 6)) plt.plot(resultados_verticales['Day'], resultados_verticales['Ganancia_Acumulada'], linestyle='-') plt.title(f"Profit & Loss Acumulado - {estrategia} {symbol}\n{desde} a {hasta}, Time {timeHour}, D{desplazamiento}") plt.xlabel('Fecha') plt.ylabel('Ganancia Acumulada ($)') plt.xticks(rotation=45) plt.grid(axis='y', linestyle='--', alpha=0.7) plt.legend([f"Profit Total: ${profit_total:.2f}"], loc='upper left', fontsize=10, frameon=True) # Guardar gráfico output_filename_line = f"graficos/P&L Acumulado_{estrategia}_{symbol}_{timeHour}_{desde.replace('-', '')}_{hasta.replace('-', '')}_D{desplazamiento}.png" plt.savefig(output_filename_line)