#!/usr/bin/env python3 """ SDR1 Analysis + Iron Condor Simulation - SPX --------------------------------------------- Lógica: - 9:30:20 → precio base del underlying (parquet) - 9:35:00 → se calculan SDR1 y strikes del IC (fijos para todo el día) - Desde 9:35 se recorren timestamps buscando crédito >= TARGET_CREDIT - Si se encuentra → IC abierto en ese timestamp - Si no se encuentra en todo el día → no se opera P&L: crédito cobrado vs cómo terminaron los strikes al cierre del underlying """ import os import glob import math import pandas as pd from datetime import datetime, date # ── Rutas ───────────────────────────────────────────────────────────────────── CHAINS_DIR = "/var/www/html/flask_project/chains" PREDICTOR_DIR = "/var/www/html/backtestingmarket/predictor_data/data/SPX" PREDICTOR_PATTERN = "prediction_$SPX_{date}.csv" # ── Parámetros ───────────────────────────────────────────────────────────────── DATE_FROM = date(2025, 5, 12) DATE_TO = date(2026, 4, 16) TARGET_CREDIT = 0 # <-- en USD, ajustar según necesidad MULTIPLIER = 100 # ────────────────────────────────────────────────────────────────────────────── def get_available_dates(): chain_files = glob.glob(os.path.join(CHAINS_DIR, "optionChain_$SPX_*.parquet")) dates = [] for cf in sorted(chain_files): basename = os.path.basename(cf) date_str = basename.replace("optionChain_$SPX_", "").replace(".parquet", "") try: d = datetime.strptime(date_str, "%Y-%m-%d").date() except ValueError: continue if not (DATE_FROM <= d <= DATE_TO): continue pred_file = os.path.join(PREDICTOR_DIR, PREDICTOR_PATTERN.replace("{date}", date_str)) if not os.path.exists(pred_file): continue dates.append((d, date_str, cf, pred_file)) return dates def get_chain_df(chain_parquet): df = pd.read_parquet(chain_parquet, columns=[ "timestamp", "underlying_price", "strike", "bid_call", "ask_call", "bid_put", "ask_put" ]) df["timestamp"] = pd.to_datetime(df["timestamp"]) return df.sort_values("timestamp") def get_base_price(df, date_str): """Primer underlying_price >= 09:30:20.""" cutoff = pd.Timestamp(f"{date_str} 09:30:20") df_after = df[df["timestamp"] >= cutoff] if df_after.empty: return None first_ts = df_after["timestamp"].iloc[0] return df_after[df_after["timestamp"] == first_ts]["underlying_price"].iloc[0] def get_close_price(df): """Underlying_price del último snapshot del día.""" last_ts = df["timestamp"].iloc[-1] return df[df["timestamp"] == last_ts]["underlying_price"].iloc[0] def get_mov_esperado(pred_csv): """Primer movimiento_esperado del predictor (~9:35).""" df = pd.read_csv(pred_csv, usecols=["timestamp", "movimiento_esperado"]) df["timestamp"] = pd.to_datetime(df["timestamp"]) df = df.sort_values("timestamp") if df.empty: return None return df["movimiento_esperado"].iloc[0] def get_mid(df_snap, strike, side): """Mid price de un strike en un snapshot dado.""" rows = df_snap[df_snap["strike"] == float(strike)] if rows.empty: return None row = rows.iloc[0] bid = row["bid_call"] if side == "call" else row["bid_put"] ask = row["ask_call"] if side == "call" else row["ask_put"] if pd.isna(bid) or pd.isna(ask) or bid <= 0 or ask <= 0: return None return round((bid + ask) / 2, 2) def calc_pnl(credit, short_call, long_call, short_put, long_put, close_price): call_intrinsic = max(0, close_price - short_call) - max(0, close_price - long_call) put_intrinsic = max(0, short_put - close_price) - max(0, long_put - close_price) return round(credit - call_intrinsic - put_intrinsic, 2) def find_entry(df, date_str, short_call, long_call, short_put, long_put): """ Recorre snapshots desde >= 9:35:00 buscando crédito >= TARGET_CREDIT. Retorna (timestamp_entry, credit_points, credit_usd) o None si no encuentra. """ cutoff = pd.Timestamp(f"{date_str} 09:35:00") df_after = df[df["timestamp"] >= cutoff] if df_after.empty: return None for ts, df_snap in df_after.groupby("timestamp", sort=True): mid_sc = get_mid(df_snap, short_call, "call") mid_lc = get_mid(df_snap, long_call, "call") mid_sp = get_mid(df_snap, short_put, "put") mid_lp = get_mid(df_snap, long_put, "put") if any(m is None for m in [mid_sc, mid_lc, mid_sp, mid_lp]): continue credit_pts = round((mid_sc - mid_lc) + (mid_sp - mid_lp), 2) credit_usd = round(credit_pts * MULTIPLIER, 2) if credit_usd >= TARGET_CREDIT: return ts, credit_pts, credit_usd return None def analyze(): dates = get_available_dates() if not dates: print(f"No se encontraron archivos para el rango {DATE_FROM} — {DATE_TO}") return print(f"\nTarget crédito: ${TARGET_CREDIT} | Rango: {DATE_FROM} → {DATE_TO}\n") print( f"{'Fecha':<12} {'Base':>8} {'ME':>7} {'SDR1 Low':>10} {'SDR1 High':>10} {'Cierre':>8} " f"{'SDR1':^10} " f"{'SC':>6} {'LC':>6} {'SP':>6} {'LP':>6} " f"{'Entry':>8} {'Crédito':>8} {'P&L':>8} Resultado IC" ) print("─" * 135) results = [] skipped = 0 for d, date_str, chain_file, pred_file in dates: df = get_chain_df(chain_file) base_price = get_base_price(df, date_str) if base_price is None: print(f"{date_str:<12} ⚠️ Sin precio base >= 9:30:20") continue close_price = get_close_price(df) mov_esp = get_mov_esperado(pred_file) if mov_esp is None: print(f"{date_str:<12} ⚠️ Sin datos de predictor") continue sdr1_high = base_price + mov_esp sdr1_low = base_price - mov_esp en_sdr1 = sdr1_low <= close_price <= sdr1_high sdr1_str = "✅ DENTRO" if en_sdr1 else "❌ FUERA" # Strikes fijos desde 9:35 short_call = math.floor(sdr1_high / 5) * 5 long_call = short_call + 5 short_put = math.ceil(sdr1_low / 5) * 5 long_put = short_put - 5 # Buscar entry con crédito >= TARGET entry = find_entry(df, date_str, short_call, long_call, short_put, long_put) if entry is None: print(f"{date_str:<12} ⏭️ Sin entry (crédito < ${TARGET_CREDIT} en todo el día)") skipped += 1 continue entry_ts, credit_pts, credit_usd = entry entry_time = entry_ts.strftime("%H:%M") pnl = calc_pnl(credit_pts, short_call, long_call, short_put, long_put, close_price) pnl_usd = round(pnl * MULTIPLIER, 2) ic_str = "✅ WIN" if pnl > 0 else ("🟡 BREAK" if pnl == 0 else "❌ LOSS") print( f"{date_str:<12} {base_price:>8.2f} {mov_esp:>7.2f} {sdr1_low:>10.2f} {sdr1_high:>10.2f} {close_price:>8.2f} " f"{sdr1_str:<10} " f"{short_call:>6} {long_call:>6} {short_put:>6} {long_put:>6} " f"{entry_time:>8} {credit_usd:>8.2f} {pnl:>8.2f} {ic_str} (${pnl_usd})" ) results.append({ "fecha": date_str, "precio_base": base_price, "mov_esperado": mov_esp, "sdr1_low": round(sdr1_low, 2), "sdr1_high": round(sdr1_high, 2), "precio_cierre": close_price, "dentro_sdr1": en_sdr1, "short_call": short_call, "long_call": long_call, "short_put": short_put, "long_put": long_put, "entry_time": entry_time, "credito_usd": credit_usd, "pnl_puntos": pnl, "pnl_usd": pnl_usd, "ic_resultado": ic_str, }) if not results: return total = len(results) cnt_sdr1 = sum(1 for r in results if r["dentro_sdr1"]) cnt_win = sum(1 for r in results if r["pnl_puntos"] > 0) cnt_loss = sum(1 for r in results if r["pnl_puntos"] < 0) total_pnl = round(sum(r["pnl_usd"] for r in results), 2) avg_credit = round(sum(r["credito_usd"] for r in results) / total, 2) print("─" * 135) print(f"\nRESUMEN") print(f" Total días analizados : {total + skipped}") print(f" ⏭️ Sin entry : {skipped:>3} (crédito < ${TARGET_CREDIT})") print(f" 📈 Días operados : {total:>3}") print(f" ✅ Dentro del SDR1 : {cnt_sdr1:>3} ({cnt_sdr1/total*100:.1f}%)") print(f" ── Iron Condor ────────────────────────────") print(f" ✅ WIN : {cnt_win:>3} ({cnt_win/total*100:.1f}%)") print(f" ❌ LOSS : {cnt_loss:>3} ({cnt_loss/total*100:.1f}%)") print(f" 📊 Crédito promedio : ${avg_credit}") print(f" 💰 P&L Total : ${total_pnl}") out_path = f"/var/www/html/backtestingmarket/sdr_ic_results_{DATE_FROM}_{DATE_TO}.csv" pd.DataFrame(results).to_csv(out_path, index=False) print(f"\n Resultados guardados en: {out_path}\n") if __name__ == "__main__": analyze()