import os import re from datetime import datetime import pandas as pd import matplotlib.pyplot as plt from matplotlib.animation import FuncAnimation # ================= CONFIGURACIÓN ================= TICKERS = ["SPX", "RUT", "XSP", "SPY", "QQQ"] ESTRATEGIAS = ["Vertical", "IronCondor"] RIESGOS = ["cons", "inte", "agre", "ultr"] BASE_DATA_DIR = "/var/www/html/backtestingmarket/backtestingSemanal" BASE_OUTPUT_DIR = "/var/www/html/backtestingmarket/backtestingSemanal" # Subcarpetas de salida ANIM_DIR = os.path.join(BASE_OUTPUT_DIR, "animaciones") TOP3_DIR = os.path.join(BASE_OUTPUT_DIR, "top3") COL_HORA = "hora" COL_PROFIT = "profit" # ================= UTILIDADES ================= def ordenar_horas(horas_set): try: return sorted(horas_set, key=lambda h: int(str(h).replace(":", ""))) except Exception: return sorted(horas_set) def patron_filename(riesgo, ticker, estrategia): return re.compile( rf"resultados_backtesting_{riesgo}_{ticker}_(\d{{4}}-\d{{2}}-\d{{2}})_(\d{{4}}-\d{{2}}-\d{{2}})_{estrategia}\.csv$" ) def asegurar_dirs(): for d in [ANIM_DIR, TOP3_DIR]: os.makedirs(d, exist_ok=True) # ================= 1. LISTAR ARCHIVOS ================= def listar_archivos_ordenados(data_dir, regex): archivos_info = [] if not os.path.isdir(data_dir): return archivos_info for fname in os.listdir(data_dir): match = regex.match(fname) if not match: continue start_str, end_str = match.groups() archivos_info.append({ "fname": fname, "path": os.path.join(data_dir, fname), "start_date": datetime.strptime(start_str, "%Y-%m-%d"), "end_date": datetime.strptime(end_str, "%Y-%m-%d"), }) archivos_info.sort(key=lambda x: (x["end_date"], x["start_date"])) return archivos_info # ================= 2. PREPARAR DATOS ================= def preparar_datos(archivos_info, ticker, estrategia, riesgo, top3_csv): all_hours = set() global_min, global_max = None, None frames_data = [] top3_rows = [] for info in archivos_info: df = pd.read_csv(info["path"]) all_hours.update(df[COL_HORA].astype(str)) profits = df[COL_PROFIT].astype(float) global_min = profits.min() if global_min is None else min(global_min, profits.min()) global_max = profits.max() if global_max is None else max(global_max, profits.max()) all_hours_sorted = ordenar_horas(all_hours) margen = 1.0 if global_max == global_min else 0.05 * (global_max - global_min) y_min = global_min - margen y_max = global_max + margen for info in archivos_info: df = pd.read_csv(info["path"]) df[COL_PROFIT] = df[COL_PROFIT].astype(float) mapa_profit = dict(zip(df[COL_HORA].astype(str), df[COL_PROFIT])) heights = [mapa_profit.get(h, 0.0) for h in all_hours_sorted] top3_df = ( df[[COL_HORA, COL_PROFIT]] .sort_values(by=COL_PROFIT, ascending=False) .head(3) .reset_index(drop=True) ) for idx, row in top3_df.iterrows(): top3_rows.append({ "ticker": ticker, "estrategia": estrategia, "riesgo": riesgo, "start_date": info["start_date"].date(), "end_date": info["end_date"].date(), "rank": idx + 1, "hora": row[COL_HORA], "profit": float(row[COL_PROFIT]), }) frames_data.append({ "heights": heights, "title": ( f"{ticker} {riesgo.upper()} - {estrategia}\n" f"{info['start_date'].date()} → {info['end_date'].date()}" ), "top3_profit_sum": float(top3_df[COL_PROFIT].sum()) }) if top3_rows: pd.DataFrame(top3_rows).to_csv(top3_csv, index=False) print(f" ✔ TOP3 CSV: {top3_csv}") return all_hours_sorted, y_min, y_max, frames_data # ================= 3. ANIMACIÓN ================= def crear_animacion(all_hours_sorted, y_min, y_max, frames_data, output_video): fig, ax = plt.subplots(figsize=(10, 5)) x_pos = range(len(all_hours_sorted)) data0 = frames_data[0] heights0 = data0["heights"] top3_idx = set(sorted(range(len(heights0)), key=lambda i: heights0[i], reverse=True)[:3]) colors = ["blue" if i in top3_idx else ("red" if h < 0 else "green") for i, h in enumerate(heights0)] bars = ax.bar(x_pos, heights0, color=colors) ax.set_xticks(list(x_pos)) ax.set_xticklabels(all_hours_sorted, rotation=90) ax.set_ylim(y_min, y_max) ax.set_xlabel("Horario (hhmm)") ax.set_ylabel("Profit") ax.set_title(data0["title"]) top3_text = ax.text( 0.98, 0.95, f"Top 3 profit: {data0['top3_profit_sum']:,.0f}", transform=ax.transAxes, ha="right", va="top", bbox=dict(boxstyle="round", fc="white", alpha=0.7) ) plt.tight_layout() def update(frame_idx): data = frames_data[frame_idx] heights = data["heights"] top3 = set(sorted(range(len(heights)), key=lambda i: heights[i], reverse=True)[:3]) for i, (bar, h) in enumerate(zip(bars, heights)): bar.set_height(h) bar.set_color("blue" if i in top3 else ("red" if h < 0 else "green")) ax.set_title(data["title"]) top3_text.set_text(f"Top 3 profit: {data['top3_profit_sum']:,.0f}") return list(bars) + [top3_text] anim = FuncAnimation(fig, update, frames=len(frames_data), interval=1000, repeat=True) anim.save(output_video, fps=1, dpi=150) plt.close(fig) print(f" ✔ ANIMACIÓN: {output_video}") # ================= MAIN ================= def main(): asegurar_dirs() total = len(TICKERS) * len(ESTRATEGIAS) * len(RIESGOS) procesados = 0 saltados = 0 for ticker in TICKERS: for estrategia in ESTRATEGIAS: for riesgo in RIESGOS: etiqueta = f"{ticker} | {estrategia} | {riesgo}" data_dir = os.path.join(BASE_DATA_DIR, ticker, estrategia) regex = patron_filename(riesgo, ticker, estrategia) archivos_info = listar_archivos_ordenados(data_dir, regex) if not archivos_info: print(f"[SKIP] {etiqueta} — sin archivos válidos") saltados += 1 continue print(f"\n[{procesados + saltados + 1}/{total}] Procesando: {etiqueta} ({len(archivos_info)} archivos)") # Rutas de salida sufijo = f"{ticker}_{estrategia}_{riesgo}" output_video = os.path.join(ANIM_DIR, f"anim_{sufijo}.mp4") top3_csv = os.path.join(TOP3_DIR, f"top3_horarios_{sufijo}.csv") all_hours_sorted, y_min, y_max, frames_data = preparar_datos( archivos_info, ticker, estrategia, riesgo, top3_csv ) crear_animacion(all_hours_sorted, y_min, y_max, frames_data, output_video) procesados += 1 print(f"\n{'='*50}") print(f"Finalizado. Procesados: {procesados} | Saltados (sin datos): {saltados}") print(f" Animaciones → {ANIM_DIR}") print(f" TOP3 → {TOP3_DIR}") if __name__ == "__main__": main()