import os import sys import time import json import pytz from dotenv import load_dotenv from datetime import datetime, timedelta import requests import csv import pandas as pd # ps aux | grep python | grep "/var/www/html" # nohup /bin/python3 -u /var/www/html/process/gettingChains_outliers4.py > logs/gettingChains_outliers4.log 2>&1 & # Cargar variables de entorno load_dotenv() # Configuración global PATH_UBUNTU = "/var/www/html/flask_project/chains/" MAX_RETRIES = 3 RETRY_DELAY_SECONDS = 30 TOTAL_STRIKES = 120 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", "01/01/2025", "01/20/2025", "02/17/2025", "04/18/2025", "05/26/2025", "06/19/2025", "07/04/2025", "09/01/2025", "11/27/2025", "12/25/2025" ] # 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) por cada strike asegurando valores únicos. Devuelve también el valor total de GEX (versión simplificada). """ 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) gex = ((gamma_call * open_interest_call) + (gamma_put * open_interest_put)) * 100 row['gex_per_strike'] = gex if (gamma_call or gamma_put) else 0 total_gex += row['gex_per_strike'] return total_gex, data def calculate_gex_precise(data, underlying_price): """ Calcula el Gamma Exposure (GEX) preciso por cada strike y el total para SPX. """ total_gex = 0 multiplier = 100 percent_move = 0.01 for row in data: gamma_call = row.get('gamma_call', 0) gamma_put = row.get('gamma_put', 0) oi_call = row.get('open_interest_call', 0) oi_put = row.get('open_interest_put', 0) gex = ( (gamma_call * oi_call - gamma_put * oi_put) * multiplier * underlying_price * percent_move ) row['gex_precise'] = gex if (gamma_call or gamma_put) else 0 total_gex += row['gex_precise'] return total_gex, data def save_chain_to_cvs(data, symbol): """ Guarda la cadena de opciones procesada en un archivo CSV y Parquet, incluyendo GEX (simple y preciso). Filtra SOLO en SPY/QQQ las filas con delta_put < -1000 o gamma_put < -1000. """ # --- Filtro simple de outliers (solo SPY/QQQ) --- sym = str(symbol).upper().replace('$', '').strip() if sym in {'SPY', 'QQQ'} and data: filtered, dropped = [], 0 for r in data: # Parse seguros (None o no convertible -> no se considera outlier) try: dp = float(r.get('delta_put')) if r.get('delta_put') is not None else None except Exception: dp = None try: gp = float(r.get('gamma_put')) if r.get('gamma_put') is not None else None except Exception: gp = None if (dp is not None and dp < -1000) or (gp is not None and gp < -1000): dropped += 1 continue filtered.append(r) if dropped: print(f"[{sym}] {dropped} filas descartadas (delta_put/gamma_put < -1000).") data = filtered if not data: print(f"[{sym}] Snapshot vacío tras filtrar; no se guarda.") return # nada que guardar # --- Tu lógica original desde aquí --- fecha_actual = datetime.now().strftime("%Y-%m-%d") filename_csv = PATH_UBUNTU + f"optionChain_{symbol}_{fecha_actual}.csv" filename_parquet = PATH_UBUNTU + f"optionChain_{symbol}_{fecha_actual}.parquet" 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', 'gex_precise', 'gex_total_precise' ] file_exists = os.path.isfile(filename_csv) # GEX simple total_gex, data = calculate_gex(data) for row in data: row['gex_total'] = total_gex # GEX preciso underlying_price = data[0].get('underlying_price', 0) if data else 0 total_gex_precise, data = calculate_gex_precise(data, underlying_price) for row in data: row['gex_total_precise'] = total_gex_precise try: # Guardar CSV with open(filename_csv, mode='a', newline='') as file: writer = csv.DictWriter(file, fieldnames=header) if not file_exists: writer.writeheader() os.chmod(filename_csv, 0o664) writer.writerows(data) # Guardar PARQUET (append) df = pd.DataFrame(data) # Conversión explícita de tipos df["timestamp"] = pd.to_datetime(df["timestamp"], errors="coerce") df["underlying_price"] = pd.to_numeric(df["underlying_price"], errors="coerce") df["strike"] = pd.to_numeric(df["strike"], errors="coerce") float_cols = [ "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", "gex_precise", "gex_total_precise" ] for col in float_cols: df[col] = pd.to_numeric(df[col], errors="coerce") if os.path.exists(filename_parquet): df_old = pd.read_parquet(filename_parquet) df = pd.concat([df_old, df], ignore_index=True) df.to_parquet(filename_parquet, index=False, compression='zstd') except Exception as e: print(f"❌ Error al guardar CSV o PARQUET 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 esperar_hasta(destino_dt): """ Espera activa con logs espaciados para no inundar el log. • Si faltan > 10 min: log cada 60s • 10 min a 60s: log cada 10s • < 60s: log cada 1s """ while True: ahora = datetime.now() if ahora >= destino_dt: break resto = destino_dt - ahora total_seg = int(resto.total_seconds()) if total_seg > 600: step = 60 elif total_seg > 60: step = 10 else: step = 1 h, rem = divmod(total_seg, 3600) m, s = divmod(rem, 60) print(f"⏳ Esperando próxima apertura en {h:02d}:{m:02d}:{s:02d} (destino: {destino_dt.strftime('%Y-%m-%d %H:%M:%S')})", end='\r') time.sleep(step) print() # salto de línea tras la espera # --- helpers de calendario --- def es_habil(fecha, holidays): # Lunes=0 ... Domingo=6 -> excluir sábado(5) y domingo(6) if fecha.weekday() in (5, 6): return False return fecha.strftime("%m/%d/%Y") not in holidays def proximo_habil(fecha, holidays): d = fecha while not es_habil(d, holidays): d += timedelta(days=1) return d def limites_sesion(hora_actual, hora_market_open, hora_market_close, holidays): """ Devuelve (estado, apertura, cierre) donde: estado ∈ {"esperar", "en_mercado"} apertura y cierre son datetimes para la sesión relevante (de HOY si corresponde). Lógica: - Si hoy no es hábil => esperar a próxima apertura hábil. - Si ahora < apertura_hoy => esperar a apertura_hoy. - Si apertura_hoy <= ahora < cierre_hoy => en_mercado (hasta cierre_hoy). - Si ahora >= cierre_hoy => esperar a apertura del próximo hábil. """ hoy = hora_actual.date() if not es_habil(hoy, holidays): d = proximo_habil(hoy, holidays) apertura = datetime.combine(d, hora_market_open) cierre = datetime.combine(d, hora_market_close) return "esperar", apertura, cierre apertura_hoy = datetime.combine(hoy, hora_market_open) cierre_hoy = datetime.combine(hoy, hora_market_close) if hora_actual < apertura_hoy: return "esperar", apertura_hoy, cierre_hoy elif hora_actual < cierre_hoy: return "en_mercado", apertura_hoy, cierre_hoy else: d = proximo_habil(hoy + timedelta(days=1), holidays) apertura = datetime.combine(d, hora_market_open) cierre = datetime.combine(d, hora_market_close) return "esperar", apertura, cierre def main(): """ Daemon que NO termina: - 'esperar' hasta apertura de hoy (si corresponde) o del próximo hábil - 'en_mercado' procesa hasta el CIERRE de HOY """ print("Daemon iniciado. Controlando aperturas/cierres del mercado...") hora_market_open = datetime.strptime(TIME_OPEN, "%H:%M:%S").time() hora_market_close = datetime.strptime(TIME_CLOSE, "%H:%M:%S").time() while True: ahora = datetime.now() estado, apertura, cierre = limites_sesion(ahora, hora_market_open, hora_market_close, HOLIDAYS) # Debug útil # print(f"[DEBUG] ahora={ahora} estado={estado} apertura={apertura} cierre={cierre}") if estado == "esperar": if ahora < apertura: print(f"🔒 Mercado cerrado. Próxima apertura: {apertura.strftime('%Y-%m-%d %H:%M:%S')}") esperar_hasta(apertura) print("🔓 Mercado abrió. Comenzando streaming...") # vuelve al while: recalcula y ahora caerá en 'en_mercado' continue # Salvaguarda: si por timing ya es >= apertura, seguimos abajo a 'en_mercado' # Aquí estamos dentro de la sesión de HOY print("***** Mercado abierto. Iniciando streaming *****") try: # Log de fecha de procesamiento print(f"***** Fecha actual de procesamiento: {datetime.now().strftime('%Y-%m-%d')} *****") while datetime.now() < cierre: symbols = ["$RUT", "$XSP", "SPY", "QQQ", "$SPX"] try: procesar_simbolos(symbols) except Exception as e: print(f"\nError inesperado en ciclo de mercado: {e}") time.sleep(10) finally: print("\n***** Mercado cerrado (fin de jornada). Pasando a modo espera *****") # El while continúa y, al recalcular límites, esperará a la próxima apertura if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\nEjecución interrumpida por el usuario.") except Exception as e: print(f"Error crítico: {e}")