import os import csv import pytz from datetime import datetime import datetime from pathFiles import PATH_UBUNTU import sys import time import pprint from modules import terminal class archivosCSVClase: def __init__(self): self.listaTemporal1 = [] self.listaTemporal2 = [] self.listaTemporal3 = [] self.listaTemporal4 = [] self.listaTemporal5 = [] self.listaTemporal6 = [] self.listaTemporal15 = [] def guardarEnLista1(self, archivo, diccionario, SPXLastPrice, VIXLastPrice): hora_actual = time.strftime("%H:%M:%S") ic_price1 = self.obtener_icPriceCharles(diccionario) self.listaTemporal1.append(datetime.datetime.now( pytz.timezone('America/New_York')).strftime('%m/%d/%Y %H:%M:%S')) self.listaTemporal1.append(ic_price1) self.listaTemporal1.append(format(SPXLastPrice, ".3f")) self.listaTemporal1.append(VIXLastPrice) if len(self.listaTemporal1) >= 8: # detecta 2 entradas grupos = [self.listaTemporal1[i:i+4] for i in range(0, len(self.listaTemporal1), 4)] # Abrir el archivo en modo escritura with open(archivo, 'a', newline='') as archivo_csv: # Crear el escritor CSV escritor_csv = csv.writer(archivo_csv) # Escribir los grupos en el archivo CSV escritor_csv.writerows(grupos) self.listaTemporal1.clear() terminal.colorPrint.quote(str(hora_actual) + " IB 09:40 40 puntos: " + ic_price1) def guardarEnLista2(self, archivo, diccionario, SPXLastPrice, VIXLastPrice): hora_actual = time.strftime("%H:%M:%S") ic_price1 = self.obtener_icPriceCharles(diccionario) self.listaTemporal2.append(datetime.datetime.now( pytz.timezone('America/New_York')).strftime('%m/%d/%Y %H:%M:%S')) self.listaTemporal2.append(ic_price1) self.listaTemporal2.append(format(SPXLastPrice, ".3f")) self.listaTemporal2.append(VIXLastPrice) if len(self.listaTemporal2) >= 8: # detecta 2 entradas grupos = [self.listaTemporal2[i:i+4] for i in range(0, len(self.listaTemporal2), 4)] # Abrir el archivo en modo escritura with open(archivo, 'a', newline='') as archivo_csv: # Crear el escritor CSV escritor_csv = csv.writer(archivo_csv) # Escribir los grupos en el archivo CSV escritor_csv.writerows(grupos) self.listaTemporal2.clear() terminal.colorPrint.quote(str(hora_actual) + " IB 10:30 40 puntos: " + ic_price1) def guardarEnLista3(self, archivo, diccionario, SPXLastPrice, VIXLastPrice): hora_actual = time.strftime("%H:%M:%S") ic_price1 = self.obtener_icPriceCharles(diccionario) self.listaTemporal3.append(datetime.datetime.now( pytz.timezone('America/New_York')).strftime('%m/%d/%Y %H:%M:%S')) self.listaTemporal3.append(ic_price1) self.listaTemporal3.append(format(SPXLastPrice, ".3f")) self.listaTemporal3.append(VIXLastPrice) if len(self.listaTemporal3) >= 8: # detecta 2 entradas grupos = [self.listaTemporal3[i:i+4] for i in range(0, len(self.listaTemporal3), 4)] # Abrir el archivo en modo escritura with open(archivo, 'a', newline='') as archivo_csv: # Crear el escritor CSV escritor_csv = csv.writer(archivo_csv) # Escribir los grupos en el archivo CSV escritor_csv.writerows(grupos) self.listaTemporal3.clear() terminal.colorPrint.quote(str(hora_actual) + " IB 11:00 40 puntos: " + ic_price1) def guardarEnLista4(self, archivo, diccionario, SPXLastPrice, VIXLastPrice): hora_actual = time.strftime("%H:%M:%S") ic_price1 = self.obtener_icPriceCharles(diccionario) self.listaTemporal4.append(datetime.datetime.now( pytz.timezone('America/New_York')).strftime('%m/%d/%Y %H:%M:%S')) self.listaTemporal4.append(ic_price1) self.listaTemporal4.append(format(SPXLastPrice, ".3f")) self.listaTemporal4.append(VIXLastPrice) if len(self.listaTemporal4) >= 8: # detecta 2 entradas grupos = [self.listaTemporal4[i:i+4] for i in range(0, len(self.listaTemporal4), 4)] # Abrir el archivo en modo escritura with open(archivo, 'a', newline='') as archivo_csv: # Crear el escritor CSV escritor_csv = csv.writer(archivo_csv) # Escribir los grupos en el archivo CSV escritor_csv.writerows(grupos) self.listaTemporal4.clear() terminal.colorPrint.quote(str(hora_actual) + " IB 11:30 40 puntos: " + ic_price1) def guardarEnLista5(self, archivo, diccionario, SPXLastPrice, VIXLastPrice): hora_actual = time.strftime("%H:%M:%S") ic_price1 = self.obtener_icPriceCharles(diccionario) self.listaTemporal5.append(datetime.datetime.now( pytz.timezone('America/New_York')).strftime('%m/%d/%Y %H:%M:%S')) self.listaTemporal5.append(ic_price1) self.listaTemporal5.append(format(SPXLastPrice, ".3f")) self.listaTemporal5.append(VIXLastPrice) if len(self.listaTemporal5) >= 8: # detecta 2 entradas grupos = [self.listaTemporal5[i:i+4] for i in range(0, len(self.listaTemporal5), 4)] # Abrir el archivo en modo escritura with open(archivo, 'a', newline='') as archivo_csv: # Crear el escritor CSV escritor_csv = csv.writer(archivo_csv) # Escribir los grupos en el archivo CSV escritor_csv.writerows(grupos) self.listaTemporal5.clear() terminal.colorPrint.quote(str(hora_actual) + " IB 09:30 30 puntos: " + ic_price1) def guardarEnLista6(self, archivo, diccionario, SPXLastPrice, VIXLastPrice): hora_actual = time.strftime("%H:%M:%S") ic_price1 = self.obtener_icPriceCharles(diccionario) self.listaTemporal6.append(datetime.datetime.now( pytz.timezone('America/New_York')).strftime('%m/%d/%Y %H:%M:%S')) self.listaTemporal6.append(ic_price1) self.listaTemporal6.append(format(SPXLastPrice, ".3f")) self.listaTemporal6.append(VIXLastPrice) if len(self.listaTemporal6) >= 8: # detecta 2 entradas grupos = [self.listaTemporal6[i:i+4] for i in range(0, len(self.listaTemporal6), 4)] # Abrir el archivo en modo escritura with open(archivo, 'a', newline='') as archivo_csv: # Crear el escritor CSV escritor_csv = csv.writer(archivo_csv) # Escribir los grupos en el archivo CSV escritor_csv.writerows(grupos) self.listaTemporal6.clear() terminal.colorPrint.quote(str(hora_actual) + " IB 09:30 50 puntos: " + ic_price1) def obtenerSPX(self, archivo): # Verificar si el archivo CSV ya existe o no if os.path.isfile(archivo): print("\nThe CSV file exists. Getting SPX Strike based on the file.") with open(archivo, 'r') as archivo_csv: lector_csv = csv.reader(archivo_csv) next(lector_csv) # Saltar la primera fila segunda_fila = next(lector_csv) # Leer la segunda fila spxStrike = segunda_fila[2] return int(float(spxStrike)) else: # El archivo no existe retorna -1 para que luego busque el precio actual del SPX with open(archivo, 'w', newline='') as archivo_csv: escritor_csv = csv.writer(archivo_csv) escritor_csv.writerow(["CREADO"]) # Dar permisos de lectura y escritura para todos los usuarios os.chmod(archivo, 0o664) return -1 def creacionListasIB(self, symbol, fechaActual, spxStrike, width): listaIB = [] PreSymbol = symbol + ' ' + \ fechaActual.strftime("%y") + fechaActual.strftime("%m") + fechaActual.strftime("%d") listaIB.append(PreSymbol + "C0" + str(spxStrike + width) + "000") listaIB.append(PreSymbol + "C0" + str(spxStrike) + "000") listaIB.append(PreSymbol + "P0" + str(spxStrike)+ "000") listaIB.append(PreSymbol + "P0" + str(spxStrike - width)+ "000") return listaIB def obtener_icPriceCharles(self, diccionario): ic = [] if diccionario is not None: for dicc in diccionario: askprice = diccionario[dicc]['quote']['askPrice'] bidprice = diccionario[dicc]['quote']['bidPrice'] mediaprice = ((askprice+bidprice)/2)*100 ic.append(mediaprice) if len(ic) == 4: ic_price = format(-ic[0]+ic[1]+ic[2]-ic[3], ".2F") else: print("\nEl ic_price no pudo obtenerse!!!") ic_price = '0' return ic_price else: return '0', '0' def chequearArchivo(self, archivo): # Verificar si el archivo CSV ya existe o no if os.path.isfile(archivo): print("\nThe CSV file exists. Getting Data.") with open(archivo, 'r') as archivo_csv: lector_csv = csv.reader(archivo_csv) primera_fila = next(lector_csv) spxStrike = primera_fila[2] print ('aaaaaaaaaaaaabbbbbbbbbbbbbbb') time.sleep(40) return int(float(spxStrike)) else: # El archivo no existe retorna -1 para que luego busque los deltas with open(archivo, 'w'): pass # Crear un archivo vacío # Dar permisos de lectura y escritura para todos los usuarios os.chmod(archivo, 0o664) return -1 def obtener_delta(opcion_chain): # Función para obtener el delta de las opciones calls_delta = [opcion['delta'] for strikes in opcion_chain['callExpDateMap'].values() for strike in strikes.values() for opcion in strike] puts_delta = [opcion['delta'] for strikes in opcion_chain['putExpDateMap'].values() for strike in strikes.values() for opcion in strike] return calls_delta, puts_delta def filtrar_opciones(opciones_chain, delta_limite_call, delta_limite_put): calls = opciones_chain['callExpDateMap'] puts = opciones_chain['putExpDateMap'] call_cercano = None put_cercano = None print ("bbbbbbbbbbbbbbbbbb") for exp_date, strikes in calls.items(): for strike, detalles in strikes.items(): delta = detalles[0]['delta'] if abs(delta - delta_limite_call) < 0.05: if call_cercano is None or abs(call_cercano[1] - delta_limite_call) > abs(delta - delta_limite_call): call_cercano = (strike, delta) for exp_date, strikes in puts.items(): for strike, detalles in strikes.items(): delta = detalles[0]['delta'] if abs(delta - delta_limite_put) < 0.05: if put_cercano is None or abs(put_cercano[1] - delta_limite_put) > abs(delta - delta_limite_put): put_cercano = (strike, delta) print (call_cercano) return call_cercano, put_cercano def creacionListasIC(self, symbol, fechaActual, stoCall, stoPut, btoCall, btoPut): listaIB = [] PreSymbol = symbol + ' ' + \ fechaActual.strftime("%y") + fechaActual.strftime("%m") + fechaActual.strftime("%d") listaIB.append(PreSymbol + "C0" + str(stoCall) + "000") listaIB.append(PreSymbol + "C0" + str(btoCall) + "000") listaIB.append(PreSymbol + "P0" + str(stoPut)+ "000") listaIB.append(PreSymbol + "P0" + str(btoCall)+ "000") print (listaIB) return listaIB