import pandas as pd import os PATH_UBUNTU = "/var/www/html/flask_project/" pd.set_option("display.max_rows", None) symbol = "SPX" estrategia = "Sonar" DESPLAZAMIENTO = -5 desde = "2025-04-28" # Fecha inicial hasta = "2025-05-01" # Fecha final # Archivo base elArchivo = f"bola8/makekos/{symbol}/{symbol}_{estrategia}_strikes_0000.csv" 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 ############################################################ # 1) Generar lista de archivos con incrementos de 5 min ############################################################ def generar_lista_archivos(base_archivo, start_hour=10, start_minute=5, end_hour=15): archivos = [] prefix, _ = base_archivo.rsplit("_", 1) for hour in range(start_hour, end_hour + 1): for minute in range(0, 60, 5): if hour == start_hour and minute < start_minute: continue if hour == end_hour and minute > 0: break time_str = f"{hour:02}{minute:02}" archivo = f"{prefix}_{time_str}.csv" archivos.append(archivo) return archivos lista_archivos = generar_lista_archivos(elArchivo) ############################################################ # 2) Funciones para Iron Condor ############################################################ def calcular_credito_y_underlying_iron_condor(archivo_option_chain, strikes, time): """ Calcula el crédito recibido y el precio subyacente de un Iron Condor en 'archivo_option_chain' a partir de 'strikes'. strikes: { 'Call_Strike1', 'Call_Strike2', 'Put_Strike1', 'Put_Strike2' } """ option_chain_df = pd.read_csv(archivo_option_chain) option_chain_df["timestamp"] = pd.to_datetime(option_chain_df["timestamp"]) filtered_chain = option_chain_df[option_chain_df["timestamp"] >= time] if filtered_chain.empty: return "No data found after the specified time", None, None, None # Filtrar datos para los 4 strikes relevant_strikes = [ strikes["Call_Strike1"], strikes["Call_Strike2"], strikes["Put_Strike1"], strikes["Put_Strike2"], ] strikes_data = filtered_chain[filtered_chain["strike"].isin(relevant_strikes)] if strikes_data.empty: return "Strikes not found in the filtered data", None, None, None # Calcular valores promedio strike_values = {} for key, strike in strikes.items(): sub = strikes_data[strikes_data["strike"] == strike] if sub.empty: strike_values[key] = None continue # Calcular el mid if "Call" in key: strike_values[key] = (sub["bid_call"].iloc[0] + sub["ask_call"].iloc[0]) / 2 else: # "Put" strike_values[key] = (sub["bid_put"].iloc[0] + sub["ask_put"].iloc[0]) / 2 credit_calls = strike_values["Call_Strike1"] - strike_values["Call_Strike2"] credit_puts = strike_values["Put_Strike1"] - strike_values["Put_Strike2"] credit_received = credit_calls + credit_puts underlying_price = strikes_data["underlying_price"].iloc[0] return credit_received, underlying_price, credit_calls, credit_puts def calcular_ganancia_perdida_dia(credit_result, underlying_price_16, strikes): """ Calcula la ganancia/pérdida diaria de un Iron Condor en base a 'credit_result' y 'underlying_price_16'. """ if underlying_price_16 is None or credit_result is None: return None # Distancias diferencia_strikes_put = strikes["Put_Strike1"] - strikes["Put_Strike2"] diferencia_strikes_call = strikes["Call_Strike2"] - strikes["Call_Strike1"] # Caso 1: Precio < Put_Strike2 o > Call_Strike2 => pérdida max if underlying_price_16 < strikes["Put_Strike2"] or underlying_price_16 > strikes["Call_Strike2"]: return (credit_result - max(diferencia_strikes_put, diferencia_strikes_call)) * 100 # Caso 2: Precio dentro de [Put_Strike1, Call_Strike1] => ganancia max if strikes["Put_Strike1"] <= underlying_price_16 <= strikes["Call_Strike1"]: return credit_result * 100 # Caso 3: Precio ITM en PUT if strikes["Put_Strike2"] <= underlying_price_16 < strikes["Put_Strike1"]: perdida_put = strikes["Put_Strike1"] - underlying_price_16 # Limitar a lo que corresponde al spread return (credit_result - min(perdida_put, diferencia_strikes_put)) * 100 # Caso 4: Precio ITM en CALL if strikes["Call_Strike1"] < underlying_price_16 <= strikes["Call_Strike2"]: perdida_call = underlying_price_16 - strikes["Call_Strike1"] return (credit_result - min(perdida_call, diferencia_strikes_call)) * 100 return None def calcular_precios_iron_condor(vertical_strikes_df, path_csv_base, desplazamiento): """ Devuelve (df_resultados, profit_total, dias_profit_positivo, dias_profit_negativo). df_resultados tiene al menos: [Day, Hour, Ganancia_Dia]. """ if vertical_strikes_df.empty: return pd.DataFrame(), 0, 0, 0 symbol = vertical_strikes_df["Symbol"].iloc[0] registros = [] for _, row in vertical_strikes_df.iterrows(): fecha = pd.to_datetime(row["Day"]).strftime("%Y-%m-%d") hora = row["Hour"] # Ajustar strikes con el desplazamiento strikes = { "Call_Strike1": row["Call_Strike1"] + desplazamiento, "Call_Strike2": row["Call_Strike2"] + desplazamiento, "Put_Strike1": row["Put_Strike1"] - desplazamiento, "Put_Strike2": row["Put_Strike2"] - desplazamiento, } # Formatear textual (opcional) strikes_format = ( f"{strikes['Call_Strike1']}/{strikes['Call_Strike2']} " f"{strikes['Put_Strike1']}/{strikes['Put_Strike2']}" ) # Nombrar archivo if symbol in ["RUT", "XSP", "SPX"]: archivo_option_chain = path_csv_base + f"chains/optionChain_${symbol}_{fecha}.csv" else: archivo_option_chain = path_csv_base + f"chains/optionChain_{symbol}_{fecha}.csv" specific_time = pd.to_datetime(f"{row['Day']} {hora}") try: credit_result, _, credit_calls, credit_puts = calcular_credito_y_underlying_iron_condor( archivo_option_chain, strikes, specific_time ) except FileNotFoundError: credit_result = None credit_calls, credit_puts = None, None try: # credit_result_16, underlying_price_16, _, _ = calcular_credito_y_underlying_iron_condor(archivo_option_chain, strikes, specific_time_16) underlying_price_16 = obtener_ultimo_underlying_price(archivo_option_chain) except FileNotFoundError: underlying_price_16 = None # # Cierre a las 16:00 # specific_time_16 = pd.to_datetime(f"{row['Day']} 16:00:00") # try: # # Volvemos a usar la misma funcion para ver precio final # credit_result_16, underlying_price_16, _, _ = calcular_credito_y_underlying_iron_condor( # archivo_option_chain, strikes, specific_time_16 # ) # except FileNotFoundError: # credit_result_16 = None # underlying_price_16 = None # Calcular ganancia/pérdida ganancia_dia = calcular_ganancia_perdida_dia(credit_result, underlying_price_16, strikes) registros.append({ "Day": fecha, "Hour": hora, "Strikes": strikes_format, "Underlying_Price_16": underlying_price_16, "Credit_Received": credit_result, "Credit_Calls": credit_calls, "Credit_Puts": credit_puts, "Ganancia_Dia": ganancia_dia, }) df_out = pd.DataFrame(registros) if df_out.empty: return df_out, 0, 0, 0 df_out["Ganancia_Dia"] = pd.to_numeric(df_out["Ganancia_Dia"], errors="coerce") df_out["Ganancia_Dia"] = df_out["Ganancia_Dia"].round(2) profit_total = df_out["Ganancia_Dia"].sum(skipna=True) dias_profit_positivo = df_out[df_out["Ganancia_Dia"] > 0].shape[0] dias_profit_negativo = df_out[df_out["Ganancia_Dia"] < 0].shape[0] return df_out, profit_total, dias_profit_positivo, dias_profit_negativo # Lista de feriados en formato 'YYYY-MM-DD' feriados = ["2025-02-17"] ############################################################ # 4) Lógica principal (similar a verticales con rachas) ############################################################ output_csv = f"{symbol}_{estrategia}_{desde}_{hasta}_Desp{DESPLAZAMIENTO}_Rachas.csv" if os.path.exists(output_csv): os.remove(output_csv) lista_archivos = generar_lista_archivos(elArchivo, start_hour=10, start_minute=5, end_hour=15) all_results = pd.DataFrame(columns=["Day", "Hour", "Ganancia_Dia"]) for archivo in lista_archivos: print(f"Procesando archivo: {archivo} para rango {desde} a {hasta}") path_completo = os.path.join(PATH_UBUNTU, archivo) if not os.path.exists(path_completo): print(f"No existe: {path_completo}") continue df_strikes = pd.read_csv(path_completo) df_strikes["Day"] = pd.to_datetime(df_strikes["Day"]) # Filtrar rango de fechas df_strikes = df_strikes[ (df_strikes["Day"] >= desde) & (df_strikes["Day"] <= hasta) ] if df_strikes.empty: continue # Calculamos Ganancia_Dia en df_gan df_gan, _, _, _ = calcular_precios_iron_condor( df_strikes, PATH_UBUNTU, DESPLAZAMIENTO ) # Limpiar DataFrame df_gan.dropna(axis="columns", how="all", inplace=True) df_gan.dropna(axis="rows", how="all", inplace=True) # Unimos a all_results if not df_gan.empty: # reindex para que coincidan las columnas df_gan_tmp = df_gan.reindex(columns=all_results.columns, fill_value=pd.NA) all_results = pd.concat([all_results, df_gan_tmp], ignore_index=True) if all_results.empty: print(f"Sin datos para {desde} a {hasta}") exit() # Ordenar desc all_results["Day"] = pd.to_datetime(all_results["Day"]) # Filtrar sábados, domingos y feriados all_results = all_results[ (all_results["Day"].dt.dayofweek < 5) & # 0=Lunes, 4=Viernes (excluye sábado y domingo) (~all_results["Day"].dt.strftime("%Y-%m-%d").isin(feriados)) # Excluye feriados ] all_results.sort_values(by="Day", ascending=False, inplace=True) # Generar rachas negativas negative_streak = {} horas_unicas = sorted(all_results["Hour"].unique()) # Crear la lista de días en orden asc, excluyendo sábados, domingos y feriados all_days_in_range = pd.date_range(start=desde, end=hasta) day_strs = [ d.strftime("%Y-%m-%d") for d in all_days_in_range if d.weekday() < 5 and d.strftime("%Y-%m-%d") not in feriados ] for hour in horas_unicas: df_hour = all_results[all_results["Hour"] == hour].sort_values( by="Day", ascending=False ) streak = 0 stop_racha = False day_profits = {ds: 0 for ds in day_strs} for _, row in df_hour.iterrows(): d = row["Day"].strftime("%Y-%m-%d") val = row["Ganancia_Dia"] if pd.notnull(val): day_profits[d] = val if not stop_racha: if val < 0: streak += 1 else: stop_racha = True else: day_profits[d] = 0 negative_streak[hour] = {"streak": streak, "profits_by_day": day_profits} cols = ["Hour", "Streak"] + day_strs[::-1] # Invertimos la lista de días en las columnas with open(output_csv, "w") as f: f.write(",".join(cols) + "\n") for hour in sorted(negative_streak.keys()): row = [hour, str(negative_streak[hour]["streak"])] for ds in day_strs[::-1]: # Iteramos sobre los días en orden inverso row.append(str(negative_streak[hour]["profits_by_day"].get(ds, 0))) f.write(",".join(row) + "\n") print(f"Rachas negativas guardadas en '{output_csv}'") # ############################################################ # # 4) Lógica principal (similar a verticales con rachas) # ############################################################ # output_csv = f"{symbol}_{estrategia}_{desde}_{hasta}_Desp{DESPLAZAMIENTO}_Rachas.csv" # if os.path.exists(output_csv): # os.remove(output_csv) # lista_archivos = generar_lista_archivos(elArchivo, start_hour=10, start_minute=5, end_hour=15) # all_results = pd.DataFrame(columns=["Day", "Hour", "Ganancia_Dia"]) # for archivo in lista_archivos: # print(f"Procesando archivo: {archivo} para rango {desde} a {hasta}") # path_completo = os.path.join(PATH_UBUNTU, archivo) # if not os.path.exists(path_completo): # print(f"No existe: {path_completo}") # continue # df_strikes = pd.read_csv(path_completo) # df_strikes["Day"] = pd.to_datetime(df_strikes["Day"]) # df_strikes = df_strikes[ # (df_strikes["Day"] >= desde) & (df_strikes["Day"] <= hasta) # ] # if df_strikes.empty: # continue # # Calculamos Ganancia_Dia en df_gan # df_gan, _, _, _ = calcular_precios_iron_condor( # df_strikes, PATH_UBUNTU, DESPLAZAMIENTO # ) # # Limpiar DataFrame # df_gan.dropna(axis="columns", how="all", inplace=True) # df_gan.dropna(axis="rows", how="all", inplace=True) # # Unimos a all_results # if not df_gan.empty: # # reindex para que coincidan las columnas # df_gan_tmp = df_gan.reindex(columns=all_results.columns, fill_value=pd.NA) # all_results = pd.concat([all_results, df_gan_tmp], ignore_index=True) # if all_results.empty: # print(f"Sin datos para {desde} a {hasta}") # exit() # # Ordenar desc # all_results["Day"] = pd.to_datetime(all_results["Day"]) # all_results.sort_values(by="Day", ascending=False, inplace=True) # # Generar rachas negativas # negative_streak = {} # horas_unicas = sorted(all_results["Hour"].unique()) # # Crear la lista de días en orden asc # all_days_in_range = pd.date_range(start=desde, end=hasta) # day_strs = [d.strftime("%Y-%m-%d") for d in all_days_in_range] # for hour in horas_unicas: # df_hour = all_results[all_results["Hour"] == hour].sort_values( # by="Day", ascending=False # ) # streak = 0 # stop_racha = False # day_profits = {ds: 0 for ds in day_strs} # for _, row in df_hour.iterrows(): # d = row["Day"].strftime("%Y-%m-%d") # val = row["Ganancia_Dia"] # if pd.notnull(val): # day_profits[d] = val # if not stop_racha: # if val < 0: # streak += 1 # else: # stop_racha = True # else: # day_profits[d] = 0 # negative_streak[hour] = {"streak": streak, "profits_by_day": day_profits} # cols = ["Hour", "Streak"] + day_strs[::-1] # Invertimos la lista de días en las columnas # with open(output_csv, "w") as f: # f.write(",".join(cols) + "\n") # for hour in sorted(negative_streak.keys()): # row = [hour, str(negative_streak[hour]["streak"])] # for ds in day_strs[::-1]: # Iteramos sobre los días en orden inverso # row.append(str(negative_streak[hour]["profits_by_day"].get(ds, 0))) # f.write(",".join(row) + "\n") # print(f"Rachas negativas guardadas en '{output_csv}'")