import csv from datetime import datetime, timedelta from tasks_backtestingidea2 import _run_backtesting_logic import sys import os sys.path.insert(0, '/var/www/html/backtestingmarket') # πŸ“ Carpeta donde se guardan los detalles semanales DETALLE_DIR = "detalle_semanal" os.makedirs(DETALLE_DIR, exist_ok=True) # πŸ—“οΈ Todos los trimestres a analizar RANGOS = [ # ("2025-02-08", "2025-05-09"), # ("2025-02-15", "2025-05-16"), # ("2025-02-22", "2025-05-23"), # ("2025-03-01", "2025-05-30"), # ("2025-03-08", "2025-06-06"), # ("2025-03-15", "2025-06-13"), # ("2025-03-22", "2025-06-20"), # ("2025-03-29", "2025-06-27"), # ("2025-04-05", "2025-07-04"), # ("2025-04-12", "2025-07-11"), # ("2025-04-19", "2025-07-18"), # ("2025-04-26", "2025-07-25"), # ("2025-05-03", "2025-08-01"), # ("2025-05-10", "2025-08-08"), # ("2025-05-17", "2025-08-15"), # ("2025-05-24", "2025-08-22"), # ("2025-05-31", "2025-08-29"), # ("2025-06-07", "2025-09-05"), # ("2025-06-14", "2025-09-12"), # ("2025-06-21", "2025-09-19"), # ("2025-06-28", "2025-09-26"), # ("2025-07-05", "2025-10-03"), # ("2025-07-12", "2025-10-10"), # ("2025-07-19", "2025-10-17"), # ("2025-07-26", "2025-10-24"), # ("2025-08-02", "2025-10-31"), # ("2025-08-09", "2025-11-07"), # ("2025-08-16", "2025-11-14"), # ("2025-08-23", "2025-11-21"), # ("2025-08-30", "2025-11-28"), # ("2025-09-06", "2025-12-05"), # ("2025-09-13", "2025-12-12"), # ("2025-09-20", "2025-12-19"), # ("2025-09-27", "2025-12-26"), # ("2025-10-04", "2026-01-02"), # ("2025-10-11", "2026-01-09"), # ("2025-10-18", "2026-01-16"), # ("2025-10-25", "2026-01-23"), # ("2025-11-01", "2026-01-30"), # ("2025-11-08", "2026-02-06"), # ("2025-11-15", "2026-02-13"), # ("2025-11-22", "2026-02-20"), # ("2025-11-29", "2026-02-27"), # ("2025-12-06", "2026-03-06"), # ("2025-12-13", "2026-03-13"), # ("2025-12-20", "2026-03-20"), # ("2025-12-27", "2026-03-27"), # ("2026-01-03", "2026-04-03"), # ("2026-01-10", "2026-04-10"), ("2026-01-17", "2026-04-17"), ] # πŸ•’ Rango horario: 10:05 a 15:00 en pasos de 5 min HORA_INICIO_STR = "10:05" HORA_FIN_STR = "15:00" # πŸ“Š ConfiguraciΓ³n β€” comentar/descomentar segΓΊn necesidad CONFIGURACIONES = { "Vertical": { "SPX": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "RUT": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "SPY": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "QQQ": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "XSP": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], }, "IronCondor": { "SPX": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "RUT": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "SPY": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "QQQ": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], "XSP": ["ultra_agresivo", "agresivo", "intermedio", "conservador"], } } # ───────────────────────────────────────────── # πŸ”§ Helpers # ───────────────────────────────────────────── def generar_semanas(desde_str, hasta_str): semanas = [] inicio = datetime.strptime(desde_str, "%Y-%m-%d") hasta = datetime.strptime(hasta_str, "%Y-%m-%d") while inicio <= hasta: fin_semana = inicio + timedelta(days=6) if fin_semana > hasta: fin_semana = hasta semanas.append(( inicio.strftime("%Y-%m-%d"), fin_semana.strftime("%Y-%m-%d") )) inicio += timedelta(days=7) return semanas def nombre_detalle(estrategia, symbol, risk, desde, hasta): risk_short = risk[:4] return os.path.join( DETALLE_DIR, f"detalle_{risk_short}_{symbol}_{estrategia}_{desde}_{hasta}.csv" ) def semana_ya_guardada(estrategia, symbol, risk, desde, hasta): return os.path.exists(nombre_detalle(estrategia, symbol, risk, desde, hasta)) def guardar_detalle_semana(estrategia, symbol, risk, desde, hasta, resultados): path = nombre_detalle(estrategia, symbol, risk, desde, hasta) with open(path, "w", newline="") as csvfile: fieldnames = ["hora", "profit", "wins", "losses"] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for fila in resultados: writer.writerow(fila) print(f" πŸ“ Detalle guardado: {path}") def consultar_semana(estrategia, symbol, risk, desde, hasta): hora_inicio = datetime.strptime(HORA_INICIO_STR, "%H:%M") hora_fin = datetime.strptime(HORA_FIN_STR, "%H:%M") resultados = [] while hora_inicio <= hora_fin: hora_str = hora_inicio.strftime("%H%M") print(f" πŸ•’ {hora_str} ...", end=" ", flush=True) try: result = _run_backtesting_logic( symbol=symbol, estrategia=estrategia, timeHour=hora_str, risk=risk, desde=desde, hasta=hasta, modo='HORA_FIJA', ) profit = result.get("profit_total", 0) wins = result.get("wins", 0) losses = result.get("losses", 0) print(f"profit={profit} | wins={wins} | losses={losses}") resultados.append({"hora": hora_str, "profit": profit, "wins": wins, "losses": losses}) except FileNotFoundError: print(f"sin archivo") resultados.append({"hora": hora_str, "profit": 0, "wins": 0, "losses": 0}) except Exception as e: print(f"[ERROR] {e}") resultados.append({"hora": hora_str, "profit": 0, "wins": 0, "losses": 0}) hora_inicio += timedelta(minutes=5) return resultados def cargar_detalle(estrategia, symbol, risk, desde, hasta): path = nombre_detalle(estrategia, symbol, risk, desde, hasta) resultados = [] with open(path, newline="") as csvfile: reader = csv.DictReader(csvfile) for row in reader: resultados.append({ "hora": row["hora"], "profit": float(row["profit"]), "wins": int(row["wins"]), "losses": int(row["losses"]) }) return resultados def sumar_semanas(lista_de_resultados): totales = {} for semana in lista_de_resultados: for fila in semana: hora = fila["hora"] if hora not in totales: totales[hora] = {"hora": hora, "profit": 0, "wins": 0, "losses": 0} totales[hora]["profit"] += fila["profit"] totales[hora]["wins"] += fila["wins"] totales[hora]["losses"] += fila["losses"] return [totales[h] for h in sorted(totales.keys())] def guardar_resultado_final(estrategia, symbol, risk, desde, hasta, resultados): # πŸ“ Estructura: backtestingSemanal/// output_dir = os.path.join("backtestingSemanal", symbol, estrategia) os.makedirs(output_dir, exist_ok=True) filename = f"resultados_backtesting_{risk[:4]}_{symbol}_{desde}_{hasta}_{estrategia}.csv" filepath = os.path.join(output_dir, filename) with open(filepath, "w", newline="") as csvfile: fieldnames = ["hora", "profit", "wins", "losses"] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() for fila in resultados: writer.writerow(fila) print(f"\n πŸ“Š Resultado final guardado: {filepath}") mejor = max(resultados, key=lambda x: x["profit"], default=None) if mejor: print(f" πŸ† Mejor hora: {mejor['hora']} | Profit: {mejor['profit']} | W:{mejor['wins']} L:{mejor['losses']}") # ───────────────────────────────────────────── # πŸš€ Main # ───────────────────────────────────────────── start_time_global = datetime.now() print(f"\nπŸ“… Total de trimestres a procesar: {len(RANGOS)}") for desde_total, hasta_total in RANGOS: print(f"\n{'#'*70}") print(f"πŸ—“οΈ Trimestre: {desde_total} β†’ {hasta_total}") print(f"{'#'*70}") semanas = generar_semanas(desde_total, hasta_total) print(f" Semanas: {len(semanas)}") for estrategia, symbols_config in CONFIGURACIONES.items(): for symbol, risks in symbols_config.items(): for risk in risks: print(f"\n{'='*70}") print(f"πŸ“Š {symbol} | {estrategia} | {risk}") print(f"{'='*70}") semanas_data = [] for desde_sem, hasta_sem in semanas: if semana_ya_guardada(estrategia, symbol, risk, desde_sem, hasta_sem): print(f"\n ⏭️ Semana {desde_sem} β†’ {hasta_sem} ya existe, cargando...") datos = cargar_detalle(estrategia, symbol, risk, desde_sem, hasta_sem) else: print(f"\n πŸ” Consultando semana {desde_sem} β†’ {hasta_sem}...") datos = consultar_semana(estrategia, symbol, risk, desde_sem, hasta_sem) guardar_detalle_semana(estrategia, symbol, risk, desde_sem, hasta_sem, datos) semanas_data.append(datos) total = sumar_semanas(semanas_data) guardar_resultado_final(estrategia, symbol, risk, desde_total, hasta_total, total) end_time_global = datetime.now() elapsed = end_time_global - start_time_global elapsed_str = str(elapsed).split(".")[0] print(f"\n⏱️ Tiempo total transcurrido: {elapsed_str}\n")