pyquotex is a Python library designed to easily integrate with the Quotex API, enabling automated trading operations. Fully open-source and licensed under MIT, the library provides features like order execution, balance checking, real-time market data collection, and more. Perfect for traders and developers looking to build efficient and customized solutions.
process_candles(history, period)Procesa el historial de velas y las agrupa por período.
def process_candles(history, period):
candles = []
current_candle = {
'open': None,
'high': float('-inf'),
'low': float('inf'),
'close': None,
'start_time': None,
'end_time': None,
'ticks': 0
}
# ... procesa y retorna las velas agrupadas
get_color(candle)Determina el color de una vela basado en sus precios de apertura y cierre.
def get_color(candle):
if candle['open'] < candle['close']:
return 'green'
elif candle['open'] > candle['close']:
return 'red'
return 'gray'
process_tick(tick, candles, period=60)Procesa ticks individuales y los incorpora a las velas existentes.
def process_tick(tick, candles, period=60):
pair, timestamp, price, direction = tick
# ... procesa el tick y actualiza las velas
get_timestamp()Obtiene el timestamp actual en UTC.
date_to_timestamp(dt)Convierte un objeto datetime a timestamp.
timestamp_to_date(timestamp)Convierte un timestamp a objeto datetime.
get_timestamp_days_ago(days)Calcula el timestamp de hace X días.
def get_timestamp_days_ago(days):
current_time = int(time.time())
seconds_in_day = 86400
return current_time - (days * seconds_in_day)
get_expiration_time_quotex(timestamp, duration)Calcula el tiempo de expiración para una operación en Quotex.
def get_expiration_time_quotex(timestamp, duration):
now_date = datetime.fromtimestamp(timestamp)
shift = 0
if now_date.second >= 30:
shift = 1
exp_date = now_date.replace(second=0, microsecond=0)
exp_date = exp_date + timedelta(minutes=int(duration / 60) + shift)
return date_to_timestamp(exp_date)
get_next_timeframe(timestamp, time_zone, timeframe, open_time=None)Calcula el siguiente timeframe basado en parámetros específicos.
def get_next_timeframe(timestamp, time_zone, timeframe: int, open_time: str = None) -> str:
# ... calcula y retorna el siguiente timeframe
truncate(f, n)Trunca un número decimal a n decimales.
def truncate(f, n):
return math.floor(f * 10 ** n) / 10 ** n
group_by_period(data, period)Agrupa datos por período específico.
def group_by_period(data, period):
grouped = defaultdict(list)
for tick in data:
timestamp = tick[0]
timeframe = int(timestamp // period)
grouped[timeframe].append(tick)
return grouped
El proyecto utiliza el módulo logging de Python para el manejo de errores y debugging.
import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
def on_error(self, wss, error):
logger.error(error)
global_value.websocket_error_reason = str(error)
global_value.check_websocket_if_error = True
if "authorization/reject" in str(message):
logger.info("Token rechazado, realizando reconexión automática.")
global_value.check_rejected_connection = 1
if global_value.websocket_error_reason == "not_money":
self.api.account_balance = {"liveBalance": 0}
El sistema implementa un mecanismo de reconexión automática cuando se detectan errores:
async def connect(attempts=5):
check, reason = await client.connect()
if not check:
attempt = 0
while attempt <= attempts:
# ... lógica de reconexión
await asyncio.sleep(5)