Utilities and Helpers Documentation - PyQuotex

Index

• Candle Processing
• Timestamp Handling
• Expiration Calculation
• Data Formatting
• Error Handling

Candle Processing

Main Functions

process_candles(history, period)

Processes candle history and groups them by period.

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
    }
    # ... processes and returns grouped candles

get_color(candle)

Determines candle color based on open and close prices.

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)

Processes individual ticks and incorporates them into existing candles.

def process_tick(tick, candles, period=60):
    pair, timestamp, price, direction = tick
    # ... processes tick and updates candles

Timestamp Handling

Time Functions

get_timestamp()

Gets current UTC timestamp.

date_to_timestamp(dt)

Converts datetime object to timestamp.

timestamp_to_date(timestamp)

Converts timestamp to datetime object.

get_timestamp_days_ago(days)

Calculates timestamp from X days ago.

def get_timestamp_days_ago(days):
    current_time = int(time.time())
    seconds_in_day = 86400
    return current_time - (days * seconds_in_day)

Expiration Calculation

Expiration Functions

get_expiration_time_quotex(timestamp, duration)

Calculates expiration time for a Quotex operation.

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)

Calculates next timeframe based on specific parameters.

def get_next_timeframe(timestamp, time_zone, timeframe: int, open_time: str = None) -> str:
    # ... calculates and returns next timeframe

Data Formatting

Formatting Functions

truncate(f, n)

Truncates a decimal number to n decimals.

def truncate(f, n):
    return math.floor(f * 10 ** n) / 10 ** n

group_by_period(data, period)

Groups data by specific period.

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

Error Handling

Logging System

The project uses Python’s logging module for error handling and debugging.

import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

Main Error Types Handled

1. WebSocket Connection Errors

   def on_error(self, wss, error):
    logger.error(error)
    global_value.websocket_error_reason = str(error)
    global_value.check_websocket_if_error = True
   

2. Authentication Errors

   if "authorization/reject" in str(message):
    logger.info("Token rejected, performing automatic reconnection.")
    global_value.check_rejected_connection = 1
   

3. Trading Operation Errors

   if global_value.websocket_error_reason == "not_money":
    self.api.account_balance = {"liveBalance": 0}
   

Best Practices

1. Async Error Handling
   • Use try/except in async operations
   • Implement appropriate timeouts
   • Handle network errors specifically
2. Logging
   • Use different logging levels (DEBUG, INFO, ERROR)
   • Include contextual information in logs
   • Maintain error logs for analysis
3. Data Validation
   • Verify data before processing
   • Validate input parameters
   • Handle edge cases specifically