import os import sys import time import csv import json import pytz from dotenv import load_dotenv from datetime import datetime, timedelta import requests import pprint # Cargar variables de entorno load_dotenv() # Configuración global PATH_UBUNTU = "/var/www/html/flask_project/modelo/" MAX_RETRIES = 3 RETRY_DELAY_SECONDS = 30 TOTAL_STRIKES = 50 TIME_OPEN = "09:29:00" TIME_CLOSE = "16:02:00" HOLIDAYS = [ "01/01/2024", "01/15/2024", "02/19/2024", "05/27/2024", "06/19/2024", "07/04/2024", "09/02/2024", "11/28/2024", "12/25/2024" ] # Cliente Schwab try: import schwabdev client = schwabdev.Client( os.getenv('appKey'), os.getenv('appSecret'), os.getenv('callbackUrl'), verbose=True ) except ImportError: print("Error: No se pudo importar 'schwabdev'.") sys.exit(1) def obtener_cadena_con_reintento(symbol, from_date, to_date, total_strikes): """ Obtiene la cadena de opciones del cliente Schwab con reintentos. """ for intento in range(MAX_RETRIES): try: response = client.option_chains( symbol, fromDate=from_date, toDate=to_date, strikeCount=total_strikes ) return response.json() except (json.JSONDecodeError, requests.RequestException) as e: time.sleep(RETRY_DELAY_SECONDS) sys.exit(1) def extract_data(option_data): """ Extrae los datos relevantes de la cadena de opciones, incluyendo Open Interest, Gamma y Volumen. """ extracted_data = {} timestamp = datetime.now(pytz.timezone('America/New_York')).strftime('%Y-%m-%d %H:%M:%S') underlying_price = option_data.get('underlyingPrice', '') underlying_price = round(float(underlying_price), 3) if underlying_price else None for option_type in ['callExpDateMap', 'putExpDateMap']: if option_type in option_data: for date_key, strikes in option_data[option_type].items(): for strike, details in strikes.items(): for detail in details: strike_price = detail.get('strikePrice', None) if strike_price is not None: if strike_price not in extracted_data: extracted_data[strike_price] = { 'timestamp': timestamp, 'underlying_price': underlying_price, 'strike': strike_price, 'bid_call': None, 'ask_call': None, 'delta_call': None, 'gamma_call': None, 'open_interest_call': None, 'volume_call': None, 'bid_put': None, 'ask_put': None, 'delta_put': None, 'gamma_put': None, 'open_interest_put': None, 'volume_put': None, } if option_type == 'callExpDateMap': extracted_data[strike_price].update({ 'bid_call': detail.get('bid', None), 'ask_call': detail.get('ask', None), 'delta_call': detail.get('delta', None), 'gamma_call': detail.get('gamma', None), 'open_interest_call': detail.get('openInterest', None), 'volume_call': detail.get('totalVolume', None), }) elif option_type == 'putExpDateMap': extracted_data[strike_price].update({ 'bid_put': detail.get('bid', None), 'ask_put': detail.get('ask', None), 'delta_put': detail.get('delta', None), 'gamma_put': detail.get('gamma', None), 'open_interest_put': detail.get('openInterest', None), 'volume_put': detail.get('totalVolume', None), }) return list(extracted_data.values()) # def calculate_gex(data): # """ # Calcula el Gamma Exposure (GEX) sumando Gamma * Open Interest * 100. # """ # gex = sum((row['gamma_call'] or 0 + row['gamma_put'] or 0) * (row['open_interest_call'] or 0 + row['open_interest_put'] or 0) * 100 for row in data) # return gex # def calculate_gex(data): # """ # Calcula el Gamma Exposure (GEX) sumando (Gamma * Open Interest * 100). # """ # gex = sum( # ((row.get('gamma_call', 0) * row.get('open_interest_call', 0)) + # (row.get('gamma_put', 0) * row.get('open_interest_put', 0))) * 100 # for row in data # ) # return gex def calculate_gex(data): """ Calcula el Gamma Exposure (GEX) por cada strike asegurando valores únicos. Devuelve también el valor total de GEX. """ total_gex = 0 for row in data: gamma_call = row.get('gamma_call', 0) gamma_put = row.get('gamma_put', 0) open_interest_call = row.get('open_interest_call', 0) open_interest_put = row.get('open_interest_put', 0) row['gex_per_strike'] = ( (gamma_call * open_interest_call) + (gamma_put * open_interest_put) ) * 100 if (gamma_call or gamma_put) else 0 total_gex += row['gex_per_strike'] return total_gex, data def save_chain_to_cvs(data, symbol): """ Guarda la cadena de opciones procesada en un archivo CSV específico para el símbolo, incluyendo GEX por strike. """ fecha_actual = datetime.now().strftime("%Y-%m-%d") filename = PATH_UBUNTU + f"optionChain_{symbol}_{fecha_actual}__.csv" header = ['timestamp', 'underlying_price', 'strike', 'bid_call', 'ask_call', 'bid_put', 'ask_put', 'delta_call', 'delta_put', 'gamma_call', 'open_interest_call', 'volume_call', 'gamma_put', 'open_interest_put', 'volume_put', 'gex_per_strike', 'gex_total'] file_exists = os.path.isfile(filename) total_gex, data = calculate_gex(data) # Calcular GEX por strike y obtener el total for row in data: row['gex_total'] = total_gex # Añadir el total de GEX a cada fila try: with open(filename, mode='a', newline='') as file: writer = csv.DictWriter(file, fieldnames=header) if not file_exists: writer.writeheader() os.chmod(filename, 0o664) writer.writerows(data) except Exception as e: print(f"Error al guardar los datos en el archivo CSV para {symbol}: {e}") # def save_chain_to_cvs(data, symbol): # """ # Guarda la cadena de opciones procesada en un archivo CSV específico para el símbolo, incluyendo GEX. # """ # fecha_actual = datetime.now().strftime("%Y-%m-%d") # filename = PATH_UBUNTU + f"optionChain_{symbol}_{fecha_actual}__.csv" # header = ['timestamp', 'underlying_price', 'strike', 'bid_call', 'ask_call', 'bid_put', 'ask_put','delta_call', 'delta_put', 'gamma_call', 'open_interest_call', 'volume_call', # 'gamma_put', 'open_interest_put', 'volume_put', 'gex'] # file_exists = os.path.isfile(filename) # gex = calculate_gex(data) # Calcular GEX # for row in data: # row['gex'] = gex # try: # with open(filename, mode='a', newline='') as file: # writer = csv.DictWriter(file, fieldnames=header) # if not file_exists: # writer.writeheader() # os.chmod(filename, 0o664) # writer.writerows(data) # except Exception as e: # print(f"Error al guardar los datos en el archivo CSV para {symbol}: {e}") def mostrar_precio_en_consola(precios_por_simbolo): """ Muestra en consola la hora actual y los precios subyacentes por símbolo, generando una nueva línea en cada actualización. """ hora_actual = datetime.now().strftime('%H:%M:%S') salida = f"{hora_actual} | " + " | ".join([f"{simbolo}: {precio:.3f}" for simbolo, precio in precios_por_simbolo.items()]) print(salida) # Cada llamada a print crea una nueva línea def procesar_simbolos(symbols): """ Procesa múltiples símbolos y muestra los precios subyacentes en consola. """ fecha_actual = datetime.now().strftime("%Y-%m-%d") from_date = to_date = fecha_actual precios_por_simbolo = {} for symbol in symbols: try: cadena_opciones = obtener_cadena_con_reintento(symbol, from_date, to_date, TOTAL_STRIKES) datos_procesados = extract_data(cadena_opciones) precios_por_simbolo[symbol] = datos_procesados[0]['underlying_price'] save_chain_to_cvs(datos_procesados, symbol) except Exception as e: print(f"\nError al procesar {symbol}: {e}") mostrar_precio_en_consola(precios_por_simbolo) def calcular_hora_apertura(hora_actual, hora_market_open, holidays): """ Calcula la próxima hora de apertura del mercado considerando feriados y fines de semana. """ dias_a_sumar = 0 while True: nueva_fecha = hora_actual.date() + timedelta(days=dias_a_sumar) if nueva_fecha.weekday() not in [5, 6] and nueva_fecha.strftime("%m/%d/%Y") not in holidays: return datetime.combine(nueva_fecha, hora_market_open) dias_a_sumar += 1 def calcular_hora_cierre(hora_actual, hora_market_close): """ Calcula el horario de cierre para el día actual. """ return datetime.combine(hora_actual.date(), hora_market_close) def main(): """ Función principal que ejecuta el flujo del programa. """ print("***** Mercado abierto. Iniciando streaming *****") while True: hora_actual = datetime.now() hora_cierre = calcular_hora_cierre(hora_actual, datetime.strptime(TIME_CLOSE, "%H:%M:%S").time()) if hora_actual >= hora_cierre: print("\n***** Mercado cerrado. Finalizando *****") break try: symbols = ["$RUT", "$XSP", "SPY", "QQQ", "$SPX"] procesar_simbolos(symbols) except Exception as e: print(f"\nError inesperado: {e}") time.sleep(10) if __name__ == "__main__": try: hora_actual = datetime.now() hora_market_open = datetime.strptime(TIME_OPEN, "%H:%M:%S").time() hora_market_close = datetime.strptime(TIME_CLOSE, "%H:%M:%S").time() hora_apertura = calcular_hora_apertura(hora_actual, hora_market_open, HOLIDAYS) hora_cierre = calcular_hora_cierre(hora_actual, hora_market_close) if hora_apertura <= hora_actual < hora_cierre: print("Estamos dentro del horario de mercado. Ejecutando inmediatamente.") main() else: print(f"El mercado está cerrado. Próxima apertura: {hora_apertura.strftime('%Y-%m-%d %H:%M:%S')}") # Mostrar cuenta regresiva en formato hh:mm:ss while datetime.now() < hora_apertura: tiempo_restante = hora_apertura - datetime.now() horas, resto = divmod(tiempo_restante.seconds, 3600) minutos, segundos = divmod(resto, 60) print(f"Tiempo restante para la apertura del mercado: {horas:02}:{minutos:02}:{segundos:02}", end='\r') time.sleep(1) print("\nEl mercado ha abierto. Comenzando el flujo principal...") main() except KeyboardInterrupt: print("\nEjecución interrumpida por el usuario.") except Exception as e: print(f"Error crítico: {e}")