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)Processa o histórico de candlesticks e os 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
}
# ... processa e retorna os candlesticks agrupados
get_color(candle)Determina a cor de um candlestick baseado em seus preços de abertura e fechamento.
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)Processa ticks individuais e os incorpora aos candlesticks existentes.
def process_tick(tick, candles, period=60):
pair, timestamp, price, direction = tick
# ... processa o tick e atualiza os candlesticks
get_timestamp()Obtém o timestamp atual em UTC.
date_to_timestamp(dt)Converte um objeto datetime para timestamp.
timestamp_to_date(timestamp)Converte um timestamp para objeto datetime.
get_timestamp_days_ago(days)Calcula o timestamp de X dias atrás.
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 o tempo de expiração para uma operação no 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 o próximo timeframe baseado em parâmetros específicos.
def get_next_timeframe(timestamp, time_zone, timeframe: int, open_time: str = None) -> str:
# ... calcula e retorna o próximo timeframe
truncate(f, n)Trunca um número decimal para n decimais.
def truncate(f, n):
return math.floor(f * 10 ** n) / 10 ** n
group_by_period(data, period)Agrupa dados 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
O projeto utiliza o módulo logging do Python para o tratamento de erros e 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 rejeitado, realizando reconexão automática.")
global_value.check_rejected_connection = 1
if global_value.websocket_error_reason == "not_money":
self.api.account_balance = {"liveBalance": 0}
O sistema implementa um mecanismo de reconexão automática quando erros são detectados:
async def connect(attempts=5):
check, reason = await client.connect()
if not check:
attempt = 0
while attempt <= attempts:
# ... lógica de reconexão
await asyncio.sleep(5)