import pandas as pd import os from datetime import datetime # Variables PATH_UBUNTU = "/var/www/html/flask_project/" desde = "2025-05-01" hasta = "2025-06-11" symbol = "SPX" timeHour = '1110' desplazamiento = -10 risk = 30 estrategia = "Vertical" elArchivo = f'bola8/makekos/{symbol}/{symbol}_{estrategia}_strikes_{timeHour}.csv' def detectar_stoploss_por_underlying(archivo_option_chain, strike_venta_real, strike_compra, strike_stop_trigger, option_type, credit_inicial, timeHour): try: df = pd.read_csv(archivo_option_chain) df['timestamp'] = pd.to_datetime(df['timestamp']) except Exception: return None, None # Filtrar desde la hora deseada hora_inicio = pd.to_datetime(df['timestamp'].dt.date.min().strftime('%Y-%m-%d') + f" {timeHour[:2]}:{timeHour[2:]}") df = df[df['timestamp'] >= hora_inicio] if df.empty: return None, None # Verificar que ambos strikes reales existen strikes_disponibles = df['strike'].unique() if strike_venta_real not in strikes_disponibles or strike_compra not in strikes_disponibles: return None, None # Filtrar por strikes reales df = df[df['strike'].isin([strike_venta_real, strike_compra])] if df.empty: return None, None # Pivot por strike try: df_wide = df.pivot(index='timestamp', columns='strike') tipo = "call" if option_type == "CALL" else "put" bid_venta = df_wide[(f"bid_{tipo}", strike_venta_real)] ask_venta = df_wide[(f"ask_{tipo}", strike_venta_real)] bid_compra = df_wide[(f"bid_{tipo}", strike_compra)] ask_compra = df_wide[(f"ask_{tipo}", strike_compra)] # Usamos underlying_price de cualquier strike (los dos tienen el mismo valor) underlying_series = df[df['strike'] == strike_venta_real].set_index('timestamp')['underlying_price'] except: return None, None # Iterar y evaluar condición de stop for ts in df_wide.index: try: precio_underlying = underlying_series.get(ts) if precio_underlying is None: continue stop_triggered = ( (option_type == 'CALL' and precio_underlying > strike_stop_trigger) or (option_type == 'PUT' and precio_underlying < strike_stop_trigger) ) if stop_triggered: precio_spread = ((bid_venta[ts] + ask_venta[ts]) / 2 - (bid_compra[ts] + ask_compra[ts]) / 2) pl = (credit_inicial - precio_spread) * 100 return round(pl, 2), ts.time() except: continue # Si nunca se rompe el stop, se gana todo el crédito return round(credit_inicial * 100, 2), None def calcular_credito_inicial(df_chain, strike1, strike2, option_type, time_cut): df_chain['timestamp'] = pd.to_datetime(df_chain['timestamp']) df_chain = df_chain[df_chain['timestamp'] >= time_cut] df_chain = df_chain[df_chain['strike'].isin([strike1, strike2])] if df_chain.empty: return None try: if option_type == 'CALL': bid_1 = df_chain[(df_chain['strike'] == strike1)].iloc[0]['bid_call'] ask_1 = df_chain[(df_chain['strike'] == strike1)].iloc[0]['ask_call'] bid_2 = df_chain[(df_chain['strike'] == strike2)].iloc[0]['bid_call'] ask_2 = df_chain[(df_chain['strike'] == strike2)].iloc[0]['ask_call'] else: bid_1 = df_chain[(df_chain['strike'] == strike1)].iloc[0]['bid_put'] ask_1 = df_chain[(df_chain['strike'] == strike1)].iloc[0]['ask_put'] bid_2 = df_chain[(df_chain['strike'] == strike2)].iloc[0]['bid_put'] ask_2 = df_chain[(df_chain['strike'] == strike2)].iloc[0]['ask_put'] except: return None avg1 = (bid_1 + ask_1) / 2 avg2 = (bid_2 + ask_2) / 2 return avg1 - avg2 def analizar_verticales(df_strikes, path_csv_base, desplazamiento, desde, hasta): df_strikes['Day'] = pd.to_datetime(df_strikes['Day']) df_strikes = df_strikes[(df_strikes['Day'] >= desde) & (df_strikes['Day'] <= hasta)] df_strikes.loc[:, 'Strike1_Desplazado'] = df_strikes.apply( lambda row: row['Strike1'] + desplazamiento if row['Option_Type'] == 'CALL' else row['Strike1'] - desplazamiento, axis=1 ) df_strikes.loc[:, 'Strike2_Desplazado'] = df_strikes.apply( lambda row: row['Strike2'] + desplazamiento if row['Option_Type'] == 'CALL' else row['Strike2'] - desplazamiento, axis=1 ) resultados = [] for _, row in df_strikes.iterrows(): fecha = row['Day'].strftime('%Y-%m-%d') hora = row['Hour'] option_type = row['Option_Type'] strike1 = row['Strike1_Desplazado'] strike2 = row['Strike2_Desplazado'] # Strike reales usados en la estrategia if option_type == 'CALL': strike_venta_real = min(strike1, strike2) strike_compra = max(strike1, strike2) strike_stop_trigger = strike_venta_real + risk else: strike_venta_real = max(strike1, strike2) strike_compra = min(strike1, strike2) strike_stop_trigger = strike_venta_real - risk # Archivo de opción chain if symbol in ['SPX', 'RUT', 'XSP']: archivo_chain = f"{path_csv_base}chains/optionChain_${symbol}_{fecha}.csv" else: archivo_chain = f"{path_csv_base}chains/optionChain_{symbol}_{fecha}.csv" if not os.path.exists(archivo_chain): resultados.append({ 'Day': fecha, 'Time': hora, 'Strike1': strike1, 'Strike2': strike2, 'Option': option_type, 'Credit': None, 'Strike_Stop': None, 'Stop_Hit_PL': None, 'Stop_Hit_Time': None }) continue try: df_chain = pd.read_csv(archivo_chain) except: continue time_cut = pd.to_datetime(f"{fecha} {hora}") credit_inicial = calcular_credito_inicial(df_chain, strike1, strike2, option_type, time_cut) if credit_inicial is None: resultados.append({ 'Day': fecha, 'Time': hora, 'Strike1': strike1, 'Strike2': strike2, 'Option': option_type, 'Credit': None, 'Strike_Stop': None, 'Stop_Hit_PL': None, 'Stop_Hit_Time': None }) continue # Ahora usamos strike_stop_trigger para detectar el cruce del subyacente perdida, hora_stop = detectar_stoploss_por_underlying( archivo_chain, strike_venta_real, # strike realmente vendido (existe en el option chain) strike_compra, # strike comprado (también existe) strike_stop_trigger, # el nivel ajustado con risk (puede no existir) option_type, credit_inicial, timeHour ) resultados.append({ 'Day': fecha, 'Time': hora, 'Strike1': strike1, 'Strike2': strike2, 'Option': option_type, 'Credit': round(credit_inicial * 100, 2), 'Strike_Stop': strike_stop_trigger, 'Stop_Hit_PL': perdida, 'Stop_Hit_Time': hora_stop }) return pd.DataFrame(resultados) # =================== EJECUCIÓN =================== vertical_strikes_df = pd.read_csv(PATH_UBUNTU + elArchivo) df_resultado = analizar_verticales(vertical_strikes_df, PATH_UBUNTU, desplazamiento, pd.to_datetime(desde), pd.to_datetime(hasta)) # Mostrar print(df_resultado.to_string(index=False)) # Eliminar columna 'Status' si existe if 'Status' in df_resultado.columns: df_resultado.drop(columns=['Status'], inplace=True) # Asegurar valores numéricos en P/L df_resultado['Stop_Hit_PL'] = pd.to_numeric(df_resultado['Stop_Hit_PL'], errors='coerce') # Filtrar solo los válidos validos = df_resultado.dropna(subset=['Stop_Hit_PL']) # Recuento positivos = validos[validos['Stop_Hit_Time'].isna()].shape[0] # sobrevivieron = ganancia completa negativos = validos[validos['Stop_Hit_Time'].notna()].shape[0] # rompieron el strike total = positivos + negativos winrate = (positivos / total) * 100 if total > 0 else 0 acumulado = validos['Stop_Hit_PL'].sum() # Mostrar resumen print("\nResumen de resultados:") print(f"✅ Días ganadores (sobrevivieron): {positivos}") print(f"❌ Días con stop activado: {negativos}") print(f"📊 Win Rate: {winrate:.2f}%") print(f"💰 Ganancia/Pérdida Total: ${acumulado:.2f}") # Exportar CSV output_csv = f"resultados/StopLoss_Subyacente_{estrategia}_{symbol}_{timeHour}_{desde.replace('-', '')}_{hasta.replace('-', '')}_D{desplazamiento}.csv" df_resultado.to_csv(output_csv, index=False)