import os import pandas as pd from flask import Blueprint, jsonify, request from datetime import datetime import math import gc from functools import lru_cache import pyarrow as pa import pyarrow.parquet as pq import sys from services.backtesting_service import run_backtesting backtestingIdeaV2_bp = Blueprint('backtestingIdeaV2', __name__, template_folder='templates') # Ruta base para archivos PATH_UBUNTU_CHAINS = "/var/www/html/flask_project/" def _load_chain(symbol: str, fecha: str, base: str) -> pd.DataFrame: folder = os.path.join(base, "chains") prefix = "$" if symbol in ["SPX", "RUT", "XSP"] else "" pq_path = os.path.join(folder, f"optionChain_{prefix}{symbol}_{fecha}.parquet") # print(f"[DEBUG] Intentando cargar {symbol} para {fecha}") if os.path.exists(pq_path): # print(f"[DEBUG] Leyendo PARQUET desde {pq_path}") df = pq.read_table(pq_path).to_pandas() else: # print(f"[ERROR] No existe PARQUET para {fecha} (ni CSV)") return pd.DataFrame() df["avg_call"] = (df["bid_call"] + df["ask_call"]) / 2 df["avg_put"] = (df["bid_put"] + df["ask_put"]) / 2 df.set_index(["timestamp", "strike"], inplace=True) return df # ============================ FUNCIONES VERTICALES ============================ def calcular_credito_y_underlying_desde_archivo_csv(archivo_option_chain, strike1, strike2, option_type, timeHour): try: 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 0, 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"]] strike1_data = strikes_data[strikes_data['strike'] == strike1].iloc[0] strike2_data = strikes_data[strikes_data['strike'] == strike2].iloc[0] 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 except: return 0, None def calcular_precios_verticales(vertical_strikes_df, path_csv_base, desplazamiento, desde, hasta): import gc resultados = [] symbol = vertical_strikes_df['Symbol'].iloc[0] vertical_strikes_df['Day'] = pd.to_datetime(vertical_strikes_df['Day']) # Filtrar rango de fechas vertical_strikes_df = vertical_strikes_df[ (vertical_strikes_df['Day'] >= pd.to_datetime(desde)) & (vertical_strikes_df['Day'] <= pd.to_datetime(hasta)) ].copy() # Aplicar desplazamientos vertical_strikes_df['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['Strike2_Desplazado'] = vertical_strikes_df.apply( lambda row: row['Strike2'] + desplazamiento if row['Option_Type'] == 'CALL' else row['Strike2'] - desplazamiento, axis=1) df_cache = {} # Cache optionChain for _, row in vertical_strikes_df.iterrows(): fecha = pd.to_datetime(row['Day']).strftime('%Y-%m-%d') hora = row['Hour'] specific_time = pd.to_datetime(f"{row['Day']} {hora}") strike1 = row['Strike1_Desplazado'] strike2 = row['Strike2_Desplazado'] option_type = row['Option_Type'] # Load optionChain if fecha not in df_cache: try: df_cache[fecha] = _load_chain(symbol, fecha, path_csv_base) except: df_cache[fecha] = pd.DataFrame() df_chain = df_cache[fecha] credit_result, underlying_price_specific_time = calcular_credito_y_underlying_desde_df( df_chain, strike1, strike2, option_type, specific_time) # ✅ Validar ambos: strikes faltantes o underlying vacío if credit_result is None or underlying_price_specific_time is None: print(f"[SKIP] {fecha} {hora} - Strikes faltantes o underlying vacío para {strike1}/{strike2}") continue underlying_price_16 = obtener_ultimo_underlying_price_desde_df(df_chain) spreadDiff = abs(strike1 - strike2) # Calcular P/L if underlying_price_16 is not None: if option_type == 'CALL': if underlying_price_16 > max(strike1, strike2): ganancia_dia = (credit_result - spreadDiff) * 100 elif underlying_price_16 < min(strike1, strike2): ganancia_dia = credit_result * 100 else: valor_itm = (underlying_price_16 - min(strike1, strike2)) * 100 ganancia_dia = credit_result * 100 - valor_itm else: # PUT if underlying_price_16 < min(strike1, strike2): ganancia_dia = (credit_result - spreadDiff) * 100 elif underlying_price_16 > max(strike1, strike2): ganancia_dia = credit_result * 100 else: valor_itm = (max(strike1, strike2) - underlying_price_16) * 100 ganancia_dia = credit_result * 100 - valor_itm else: ganancia_dia = None # ITM/OTM status if underlying_price_16 is not None: if option_type == 'CALL': if underlying_price_16 > max(strike1, strike2): itmOtm = "ITM" elif underlying_price_16 < min(strike1, strike2): itmOtm = "OTM" else: itmOtm = "ITM P" else: if underlying_price_16 < min(strike1, strike2): itmOtm = "ITM" elif underlying_price_16 > max(strike1, strike2): itmOtm = "OTM" else: itmOtm = "ITM P" else: itmOtm = "ERROR" resultados.append({ 'Day': fecha, 'Time': hora, 'Strikes': f"{strike1}/{strike2}", 'Option': option_type, 'Credit': round(credit_result * 100, 2), 'Price': underlying_price_specific_time, 'Close': underlying_price_16, 'P/L': round(ganancia_dia, 2) if ganancia_dia is not None else None, 'Diff': round(underlying_price_16 - strike1, 2) if underlying_price_16 else None, 'itmOtm': itmOtm }) # Liberar memoria si es necesario if fecha in df_cache: del df_cache[fecha] gc.collect() df_resultados = pd.DataFrame(resultados) # ✅ Filtro de respaldo por si algo se cuela df_resultados = df_resultados[ df_resultados['Credit'].notnull() & df_resultados['Price'].notnull() ].copy() print(f"[INFO] Ideas válidas: {len(df_resultados)} / Descartadas: {len(resultados) - len(df_resultados)}") profit_total = df_resultados['P/L'].sum(skipna=True) positivos = df_resultados[df_resultados['P/L'] >= 0].shape[0] negativos = df_resultados[df_resultados['P/L'] < 0].shape[0] return df_resultados, profit_total, positivos, negativos def calcular_credito_y_underlying_desde_df(df, strike1, strike2, option_type, timeHour): try: # Asegura segundos en 00 si es necesario timeHour = timeHour.replace(second=15, microsecond=0) df_debug = df.reset_index() # en parquet ya tiene index = ["timestamp", "strike"] filtered_chain = df_debug[df_debug['timestamp'] >= timeHour] # print(f"[DEBUG] → Filtrados por timestamp >= {timeHour}, total: {len(filtered_chain)}") strikes_data = filtered_chain[filtered_chain['strike'].isin([strike1, strike2])] if strikes_data.empty or len(strikes_data['strike'].unique()) < 2: print("[WARNING] → No se encontraron ambos strikes.") return 0, None strike1_data = strikes_data[strikes_data['strike'] == strike1].iloc[0] strike2_data = strikes_data[strikes_data['strike'] == strike2].iloc[0] # MOSTRAR EL TIMESTAMP REAL DE LAS FILAS USADAS # print(f"[DEBUG] → Timestamp usado strike1: {strike1_data['timestamp']}") # print(f"[DEBUG] → Timestamp usado strike2: {strike2_data['timestamp']}") avg1 = strike1_data[f'avg_{option_type.lower()}'] avg2 = strike2_data[f'avg_{option_type.lower()}'] credit_received = avg1 - avg2 underlying_price = strike1_data['underlying_price'] return credit_received, underlying_price except Exception as e: print(f"[ERROR] calcular_credito_y_underlying_desde_df(): {e}") return None, None def obtener_ultimo_underlying_price_desde_df(df): try: return df.sort_values(by='timestamp')['underlying_price'].iloc[-1] except: return None # ============================ FUNCIONES IRON CONDOR ============================ def calcular_ganancia_perdida_dia_iron_condor(credit_result, underlying_price_16, strikes): if underlying_price_16 is None or credit_result is None: return None diff_put = strikes['Put_Strike1'] - strikes['Put_Strike2'] diff_call = strikes['Call_Strike2'] - strikes['Call_Strike1'] if underlying_price_16 < strikes['Put_Strike2'] or underlying_price_16 > strikes['Call_Strike2']: return (credit_result - max(diff_put, diff_call)) * 100 if strikes['Put_Strike1'] <= underlying_price_16 <= strikes['Call_Strike1']: return credit_result * 100 if strikes['Put_Strike2'] <= underlying_price_16 < strikes['Put_Strike1']: return (credit_result - min(strikes['Put_Strike1'] - underlying_price_16, diff_put)) * 100 if strikes['Call_Strike1'] < underlying_price_16 <= strikes['Call_Strike2']: return (credit_result - min(underlying_price_16 - strikes['Call_Strike1'], diff_call)) * 100 return None def calcular_precios_iron_condor(df, path_csv_base, desplazamiento, desde, hasta): import gc resultados = [] symbol = df['Symbol'].iloc[0] df['Day'] = pd.to_datetime(df['Day']) df = df[(df['Day'] >= pd.to_datetime(desde)) & (df['Day'] <= pd.to_datetime(hasta))].copy() df_cache = {} for _, row in df.iterrows(): fecha = row['Day'].strftime('%Y-%m-%d') hora = row['Hour'] specific_time = pd.to_datetime(f"{fecha} {hora}") 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 } if fecha not in df_cache: try: df_cache[fecha] = _load_chain(symbol, fecha, path_csv_base) except: df_cache[fecha] = pd.DataFrame() df_chain = df_cache[fecha] credit_result, underlying_price, _, _ = calcular_credito_y_underlying_iron_condor_df( df_chain, strikes, specific_time) # ✅ Validar ambos: si no hay strikes o underlying → SKIP if credit_result is None or underlying_price is None: print(f"[SKIP] {fecha} {hora} - Strikes faltantes o underlying vacío.") continue underlying_price_16 = obtener_ultimo_underlying_price_desde_df(df_chain) ganancia_dia = calcular_ganancia_perdida_dia_iron_condor( credit_result, underlying_price_16, strikes ) if underlying_price_16 is not None: if underlying_price_16 < strikes['Put_Strike2'] or underlying_price_16 > strikes['Call_Strike2']: itmOtm = "ITM" elif strikes['Put_Strike1'] <= underlying_price_16 <= strikes['Call_Strike1']: itmOtm = "OTM" else: itmOtm = "ITM P" else: itmOtm = "N/A" dif_call = f"{(underlying_price_16 - strikes['Call_Strike1']):.2f}" if underlying_price_16 else None dif_put = f"{(underlying_price_16 - strikes['Put_Strike1']):.2f}" if underlying_price_16 else None resultados.append({ 'Day': fecha, 'Time': hora, 'Strikes': f"{strikes['Put_Strike2']}/{strikes['Put_Strike1']} {strikes['Call_Strike1']}/{strikes['Call_Strike2']}", 'Price': underlying_price, 'Close': underlying_price_16, 'Credit': round(credit_result * 100, 2), 'P/L': round(ganancia_dia, 2) if isinstance(ganancia_dia, (int, float)) else None, 'Dif': f"{dif_put}(P) / {dif_call}(C)", 'itmOtm': itmOtm }) del df_cache[fecha] gc.collect() df_resultados = pd.DataFrame(resultados) # ✅ Filtro de respaldo: nada de Credit o Price nulos df_resultados = df_resultados[ df_resultados['Credit'].notnull() & df_resultados['Price'].notnull() ].copy() print(f"[INFO] Ideas válidas: {len(df_resultados)} / Descartadas: {len(resultados) - len(df_resultados)}") profit_total = df_resultados['P/L'].sum(skipna=True) positivos = df_resultados[df_resultados['P/L'] >= 0].shape[0] negativos = df_resultados[df_resultados['P/L'] < 0].shape[0] return df_resultados, profit_total, positivos, negativos def calcular_credito_y_underlying_iron_condor_df(df, strikes, timeHour): try: df_debug = df.reset_index() # porque el parquet ya está indexado por ["timestamp", "strike"] filtered_chain = df_debug[df_debug['timestamp'] >= timeHour] if filtered_chain.empty: print("[WARNING] No hay datos después del timeHour") return None, None, None, None strikes_data = filtered_chain[filtered_chain['strike'].isin([ strikes['Call_Strike1'], strikes['Call_Strike2'], strikes['Put_Strike1'], strikes['Put_Strike2'] ])] strike_values = {} for key, strike in strikes.items(): row = strikes_data[strikes_data['strike'] == strike] if row.empty: strike_values[key] = None elif key.startswith("Call"): strike_values[key] = row["avg_call"].iloc[0] else: strike_values[key] = row["avg_put"].iloc[0] if any(v is None for v in strike_values.values()): print("[WARNING] Faltan strikes necesarios para calcular crédito.") return None, None, None, None 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 except Exception as e: print(f"[ERROR] calcular_credito_y_underlying_iron_condor_df(): {e}") return None, None, None, None # ============================ RUTA PRINCIPAL ============================ @backtestingIdeaV2_bp.route('/get_backtesting_ideaV2', methods=['GET']) def get_backtesting_ideaV2(): # 1) Leer y normalizar parámetros desde = request.args.get('desde') hasta = request.args.get('hasta') symbol = request.args.get('symbol', 'SPX') estrategia = request.args.get('estrategia', 'Vertical') # "Vertical" | "IronCondor" timeHour = request.args.get('hora', '1340') risk = request.args.get('risk', 'intermedio').lower() # 2) Validaciones rápidas if not desde or not hasta: return jsonify({'error': 'Parámetros de fecha inválidos (desde/hasta)'}), 400 if estrategia.lower() not in ('vertical', 'ironcondor'): return jsonify({'error': 'Estrategia no reconocida'}), 400 # 3) Delegar toda la lógica de backtest al servicio try: df_result, total, wins, losses = run_backtesting( symbol=symbol, estrategia=estrategia, timeHour=timeHour, risk=risk, desde=desde, hasta=hasta ) except FileNotFoundError as e: # p.ej. si falta el CSV de strikes return jsonify({'error': str(e)}), 404 except Exception as e: # fallback controlado return jsonify({'error': f'Error interno: {e}'}), 500 # 4) Responder en el formato que el cliente ya espera return jsonify({ 'data': df_result.astype(object).to_dict(orient='records'), 'profit_total': float(round(total, 2)), 'wins': int(wins), 'losses': int(losses) }) def obtener_ultimo_underlying_price(archivo_option_chain): try: option_chain_df = pd.read_csv(archivo_option_chain) option_chain_df['timestamp'] = pd.to_datetime(option_chain_df['timestamp']) option_chain_df = option_chain_df.sort_values(by='timestamp') return option_chain_df['underlying_price'].iloc[-1] if 'underlying_price' in option_chain_df.columns else None except: return None @backtestingIdeaV2_bp.route("/procesar_fila_vertical", methods=["POST"]) def procesar_fila_vertical(): data = request.get_json() try: symbol = data.get("symbol") fecha = data.get("Day") hora = data.get("Time") tipo = data.get("Option") # "CALL" o "PUT" # print("📌 [DEBUG] Day recibido del cliente:", fecha) # print("📌 [DEBUG] Time recibido del cliente:", hora) strikes_raw = data.get("Strikes", "") strike1, strike2 = [float(s) for s in strikes_raw.split("/")] if tipo == "PUT" and strike1 < strike2: strike1, strike2 = strike2, strike1 if tipo == "CALL" and strike1 > strike2: strike1, strike2 = strike2, strike1 df_chain = _load_chain(symbol, fecha, PATH_UBUNTU_CHAINS) if df_chain.empty: return jsonify({"status": "error", "message": "No se pudo cargar option chain"}), 400 time_obj = pd.to_datetime(f"{fecha} {hora}") credit_inicial, _ = calcular_credito_y_underlying_desde_df(df_chain, strike1, strike2, tipo, time_obj) if not isinstance(credit_inicial, (int, float)): return jsonify({"status": "error", "message": "No se pudo calcular el crédito inicial"}), 400 evolution = [] df_filtered = df_chain.reset_index() df_filtered = df_filtered[df_filtered["timestamp"] >= time_obj] for ts, group in df_filtered.groupby("timestamp"): try: avg1 = group[group["strike"] == strike1][f"avg_{tipo.lower()}"].iloc[0] avg2 = group[group["strike"] == strike2][f"avg_{tipo.lower()}"].iloc[0] underlying_price = group["underlying_price"].iloc[0] credit_actual = avg1 - avg2 profit_loss = (credit_inicial - credit_actual) * 100 evolution.append({ "timestamp": ts.strftime("%Y-%m-%d %H:%M:%S"), "credit": float(round(credit_actual * 100, 2)), "profit_loss": float(round(profit_loss, 2)), "underlying_price": float(round(underlying_price, 2)) }) except: continue return jsonify({ "status": "success", "symbol": symbol, "fecha": fecha, "strikes": f"{strike1}/{strike2}", "tipo": tipo, "initial_credit": round(credit_inicial * 100, 2), "evolution": evolution, }) except Exception as e: return jsonify({"status": "error", "message": str(e)}), 500 @backtestingIdeaV2_bp.route("/procesar_fila_ironcondor", methods=["POST"]) def procesar_fila_ironcondor(): data = request.get_json() try: symbol = data.get("symbol") fecha = data.get("Day") hora = data.get("Time") strikes_raw = data.get("Strikes", "").strip() # ✅ Separar bloques PUT y CALL try: put_block, call_block = strikes_raw.split(" ") put1, put2 = [float(s) for s in put_block.split("/")] call1, call2 = [float(s) for s in call_block.split("/")] sell_put, buy_put = max(put1, put2), min(put1, put2) sell_call, buy_call = min(call1, call2), max(call1, call2) except Exception: return jsonify({"status": "error", "message": f"Strikes inválidos: {strikes_raw}"}), 400 df_chain = _load_chain(symbol, fecha, PATH_UBUNTU_CHAINS) if df_chain.empty: return jsonify({"status": "error", "message": "No se pudo cargar option chain"}), 400 time_obj = pd.to_datetime(f"{fecha} {hora}") df_filtered = df_chain.reset_index() df_filtered = df_filtered[df_filtered["timestamp"] >= time_obj] # ✅ Calcular crédito inicial def obtener_credito(df, ts): try: df_momento = df[df["timestamp"] >= ts] ts_primero = df_momento["timestamp"].min() df_ts = df[df["timestamp"] == ts_primero] c1 = df_ts[df_ts["strike"] == sell_call]["avg_call"].iloc[0] c2 = df_ts[df_ts["strike"] == buy_call]["avg_call"].iloc[0] p1 = df_ts[df_ts["strike"] == sell_put]["avg_put"].iloc[0] p2 = df_ts[df_ts["strike"] == buy_put]["avg_put"].iloc[0] return (c1 - c2) + (p1 - p2), ts_primero except Exception as e: print(f"[ERROR crédito inicial]: {e}") return None, None credit_inicial, timestamp_inicial = obtener_credito(df_filtered, time_obj) if credit_inicial is None: return jsonify({"status": "error", "message": "No se pudo calcular el crédito inicial"}), 400 # ✅ Generar evolución evolution = [] for ts, group in df_filtered.groupby("timestamp"): try: c1 = group[group["strike"] == sell_call]["avg_call"].iloc[0] c2 = group[group["strike"] == buy_call]["avg_call"].iloc[0] p1 = group[group["strike"] == sell_put]["avg_put"].iloc[0] p2 = group[group["strike"] == buy_put]["avg_put"].iloc[0] underlying_price = group["underlying_price"].iloc[0] credit_actual = (c1 - c2) + (p1 - p2) profit_loss = (credit_inicial - credit_actual) * 100 evolution.append({ "timestamp": ts.strftime("%Y-%m-%d %H:%M:%S"), "credit": float(round(credit_actual * 100, 2)), "profit_loss": float(round(profit_loss, 2)), "underlying_price": float(round(underlying_price, 2)) }) except: continue return jsonify({ "status": "success", "symbol": symbol, "fecha": fecha, "strikes": f"{sell_put}/{buy_put} {sell_call}/{buy_call}", "initial_credit": round(credit_inicial * 100, 2), "evolution": evolution, }) except Exception as e: return jsonify({"status": "error", "message": str(e)}), 500