"""TMX Helpers Module."""
# pylint: disable=too-many-lines
# pylint: disable=unused-argument
# pylint: disable=simplifiable-if-expression
import asyncio
import json
from datetime import (
date as dateType,
datetime,
time,
timedelta,
)
from io import StringIO
from typing import Any, Dict, List, Literal, Optional, Union
import exchange_calendars as xcals
import pandas as pd
import pytz
from aiohttp_client_cache import SQLiteBackend
from aiohttp_client_cache.session import CachedSession
from dateutil import rrule
from openbb_core.app.utils import get_user_cache_directory
from openbb_core.provider.utils.helpers import amake_request, to_snake_case
from openbb_tmx.utils import gql
from pandas.tseries.holiday import next_workday
from random_user_agent.user_agent import UserAgent
cache_dir = get_user_cache_directory()
def get_random_agent() -> str:
"""Get a random user agent."""
user_agent_rotator = UserAgent(limit=100)
user_agent = user_agent_rotator.get_random_user_agent()
return user_agent
# Only used for obtaining the directory of all valid company tickers.
tmx_companies_backend = SQLiteBackend(
f"{cache_dir}/http/tmx_companies", expire_after=timedelta(days=2)
)
# Only used for obtaining the directory of all valid indices.
tmx_indices_backend = SQLiteBackend(
f"{cache_dir}/http/tmx_indices", expire_after=timedelta(days=1)
)
# Only used for obtaining the all ETFs JSON file.
tmx_etfs_backend = SQLiteBackend(
f"{cache_dir}/http/tmx_etfs", expire_after=timedelta(hours=4)
)
tmx_bonds_backend = SQLiteBackend(
f"{cache_dir}/http/tmx_bonds", expire_after=timedelta(days=1)
)
# Column map for ETFs.
COLUMNS_DICT = {
"symbol": "symbol",
"shortname": "short_name",
"longname": "name",
"fundfamily": "fund_family",
"regions": "regions",
"sectors": "sectors",
"currency": "currency",
"inceptiondate": "inception_date",
"unitprice": "unit_price",
"prevClose": "prev_close",
"close": "close",