Utils

A bunch of util functions

Cache

Bases: Struct

Cache class for handling file-based caching with support for asynchronous data loading.

Attributes:

Name	Type	Description
cache_path	str	The path where the cache file is stored.
method	Optional[Callable]	A callable, typically an async method, to fetch data when the cache doesn't exist or needs to be reloaded.
date_as_suffix	bool	If True, appends today's date to the cache file name as a suffix.

Methods:

Name	Description
get_async	Asynchronously retrieves data from the cache. If the cache file doesn't exist, it will fetch data using the provided method and store it in the cache.
clean_cache	Deletes the cache file from disk.

Usage:

async def my_function():
    await asyncio.sleep(5)
    return []

cache = Cache(
    cache_path=f"./cache/markets_{ex.exchange_name}.json", method=ex.get_contracts
)
data = await cache.get_async()

# Use this if you don't have any other async.io task, otherwise the program will end before `my_function` runs
await cache.wait_till_complete()

cache_path

cache_path: str

Where you want the file to be stored

date_as_suffix

date_as_suffix: bool = False

If true, will append today's date to the cache file.

dump_format

dump_format: FileFormat = JSON

Format of the file to dump to. It might be None if your method already dumps the file.

file_format

file_format: FileFormat = JSON

Format of the file to read.

method

method: Optional[Callable] = None

A lambda method to run in case the file doesn't exist

clean_cache

clean_cache()

Removes the cache file.

Source code in src/tradingtoolbox/utils/cache.py

def clean_cache(self):
    """
    Removes the cache file.
    """
    if os.path.exists(self.cache_path):
        os.remove(self.cache_path)

get_async

get_async(reload=True, **method_kwargs)

Gets the data from the cache. If it doesn't exist, it will reload it.

Parameters:

Name	Type	Description	Default
reload	bool	Whether to reload the data. Defaults to True.	True
**method_kwargs		Keyword arguments to pass to the method.	{}

Returns:

Name	Type	Description
dict		The data from the cache.

Source code in src/tradingtoolbox/utils/cache.py

async def get_async(self, reload=True, **method_kwargs):
    """
    Gets the data from the cache. If it doesn't exist, it will reload it.

    Args:
        reload (bool, optional): Whether to reload the data. Defaults to True.
        **method_kwargs: Keyword arguments to pass to the method.

    Returns:
        dict: The data from the cache.
    """
    data = self._read_file()
    if data is not None:
        if reload:
            self._task = asyncio.create_task(self._reload_async(**method_kwargs))
    else:
        await self._reload_async(**method_kwargs)
        data = self._read_file()
    return data  # type: ignore

wait_till_all_tasks_complete

wait_till_all_tasks_complete()

Waits for all tasks to complete. Use this if you don't have any other async.io task to avoid the program ending before the method is ran

Source code in src/tradingtoolbox/utils/cache.py

@staticmethod
async def wait_till_all_tasks_complete():
    """
    Waits for all tasks to complete.
    Use this if you don't have any other async.io task to avoid the program ending before the method is ran
    """
    while True:
        pending_tasks = []
        for task in asyncio.all_tasks():
            if not task.done():
                pending_tasks.append(task)
        if len(pending_tasks) == 1:
            break
        await asyncio.sleep(1)

wait_till_complete

wait_till_complete()

Waits for the task to complete. Use this if you don't have any other async.io task to avoid the program ending before the method is ran

Source code in src/tradingtoolbox/utils/cache.py

async def wait_till_complete(self):
    """
    Waits for the task to complete.
    Use this if you don't have any other async.io task to avoid the program ending before the method is ran
    """
    if self._task:
        while True:
            if self._task.done():
                break
            await asyncio.sleep(1)  # Check every second

Logger

Logger(suppressed_modules: list[str] = SUPPRESSED_MODULES, log_dir: str = './logs')

Custom Logger class utilizing Rich and Loguru for advanced logging.

This class sets up a logging system that uses both the Rich library for enhanced output formatting in the console and Loguru for handling log files and more sophisticated logging features. The logger is configured to display colored and detailed logs in the terminal, while also saving structured logs to a file for debugging purposes.

Key Features:

• Rich Tracebacks: Automatically installs Rich traceback for more readable error messages in the console, highlighting key information such as line numbers and functions.
• Log File Handling: Logs are saved in a specified directory with detailed information in JSON-like format, serialized for easier parsing.
• Log Levels: Configured to handle different log levels, focusing on INFO messages for the console and DEBUG level messages for log files.

Usage Example

from tradingtoolbox.utils.logger import logger, Logger

# [Optional] Create a custom logger
custom_logger = Logger(supressed_modules=["talib"], log_dir="./my_logs")

try:
    # Code that might fail
    print(a)
except Exception as e:
    logger.error()

logger.warning("This is a warning message")
logger.info({"key": "value"})
logger.print("This replaces the standard print")

Notes:

• The logger's console output is colorized using Rich, and it includes rich tracebacks for easier debugging.
• Log files are stored in the log_dir directory, defaulting to ./logs.

Initializes the custom logger instance.

Parameters:

Name	Type	Description	Default
suppressed_modules	list[str]	A list of modules to suppress from rich traceback (default is SUPPRESSED_MODULES).	SUPPRESSED_MODULES
log_dir	str	The directory where log files will be saved (default is "./logs").	'./logs'

Source code in src/tradingtoolbox/utils/logger.py

def __init__(
    self,
    suppressed_modules: list[str] = SUPPRESSED_MODULES,
    log_dir: str = "./logs",
):
    """
    Initializes the custom logger instance.

    Parameters:
        suppressed_modules:
            A list of modules to suppress from rich traceback (default is SUPPRESSED_MODULES).
        log_dir:
            The directory where log files will be saved (default is "./logs").
    """

    self._create_logs_dir(log_dir)

    # This will install rich to traceback, which is quite handy
    rich.traceback.install(
        show_locals=False,
        suppress=[__import__(name) for name in suppressed_modules],
    )

    config = {
        "handlers": [
            {
                "sink": RichHandler(
                    show_level=False,
                    show_time=True,
                    rich_tracebacks=True,
                    markup=True,
                    omit_repeated_times=False,
                ),
                # "sink": sys.stdout,
                # This will force us to only use the rich handler on normal levels
                "filter": lambda record: record["level"].name == "INFO",
                "format": "{message}",
            },
            # {
            #     "sink": sys.stdout,
            #     "colorize": True,
            #     "backtrace": True,
            #     "diagnose": True,
            #     "enqueue": False,
            #     "format": "<cyan>❯ {module}:{function} ({line})</cyan> | <green>{time:YYYY-MM-DD at HH:mm:ss.sss}</green>",
            #     "filter": lambda record: record["level"].name == "INFO",
            # },
            {
                "sink": "./logs/logs.log",
                "level": "DEBUG",
                "serialize": True,
                "enqueue": True,
                "colorize": True,
                "format": "<light-cyan>❯ {module}:{function} ({line})</light-cyan> | <light-black>{time:YYYY-MM-DD at HH:mm:ss.sss}</light-black>\n{message}",
            },
        ],
    }

    _logger.configure(**config)  # type: ignore
    self.logger = _logger.patch(patching)

error

error()

Logs the most recent traceback error in a readable format, useful for. Uses the ERROR level

Source code in src/tradingtoolbox/utils/logger.py

def error(self):
    """
    Logs the most recent traceback error in a readable format, useful for. Uses the ERROR level
    """
    console.print(Traceback())
    recent_traceback = traceback.format_exc(limit=10)
    self.logger.error(recent_traceback)

info

info(*obj)

Logs an informational message, replacing the standard print function. Uses the INFO level

Source code in src/tradingtoolbox/utils/logger.py

def info(self, *obj):
    """
    Logs an informational message, replacing the standard print function. Uses the INFO level
    """
    for item in obj:
        self.logger.opt(depth=2).info(pretty_repr(item))

warning

warning(obj)

Logs a warning message with the option to pretty-print an object. Uses the WARNING level

Source code in src/tradingtoolbox/utils/logger.py

def warning(self, obj):
    """
    Logs a warning message with the option to pretty-print an object. Uses the WARNING level
    """
    self.logger.opt(depth=2).warning(pretty_repr(obj))

print

print(*msg: Any) -> None

Logs the provided object using an advanced logging mechanism.

This method overrides the default print function to utilize a logger for output. It ensures that all output is captured through the logging system rather than standard output.

Parameters:

Name	Type	Description	Default
msg	Any	The object to be logged. It can be of any type that the logger can handle, including strings, numbers, or custom objects.	()

Usage Example

from tradingtoolbox.utils.logger import print

print("Hello world")

Source code in src/tradingtoolbox/utils/logger.py

def print(*msg: Any) -> None:
    """
    Logs the provided object using an advanced logging mechanism.

    This method overrides the default `print` function to utilize a
    logger for output. It ensures that all output is captured
    through the logging system rather than standard output.

    Parameters:
        msg: The object to be logged. It can be of any type that the
            logger can handle, including strings, numbers, or custom objects.


    **Usage Example**
    ```python
    from tradingtoolbox.utils.logger import print

    print("Hello world")
    ```
    """
    logger.info(*msg)