import logging import os import traceback import six logger = logging.getLogger(__name__) logger.setLevel(logging.WARNING) OPT_URL = "https://trading-ig.readthedocs.io/en/latest/faq.html#optional-dependencies" try: import numpy as np # noqa import pandas as pd except ImportError: _HAS_PANDAS = False logger.warning(f"pandas is not present in the environment. See {OPT_URL}") else: _HAS_PANDAS = True try: from munch import munchify # noqa except ImportError: _HAS_MUNCH = False logger.warning(f"munch is not present in the environment. See {OPT_URL}") else: _HAS_MUNCH = True DATE_FORMATS = {1: "%Y:%m:%d-%H:%M:%S", 2: "%Y/%m/%d %H:%M:%S", 3: "%Y/%m/%d %H:%M:%S"} def conv_resol(resolution): """Returns a string for resolution (from a Pandas)""" if _HAS_PANDAS: from pandas.tseries.frequencies import to_offset d = { to_offset("1s"): "SECOND", to_offset("1Min"): "MINUTE", to_offset("2Min"): "MINUTE_2", to_offset("3Min"): "MINUTE_3", to_offset("5Min"): "MINUTE_5", to_offset("10Min"): "MINUTE_10", to_offset("15Min"): "MINUTE_15", to_offset("30Min"): "MINUTE_30", to_offset("1h"): "HOUR", to_offset("2h"): "HOUR_2", to_offset("3h"): "HOUR_3", to_offset("4h"): "HOUR_4", to_offset("D"): "DAY", to_offset("W"): "WEEK", to_offset("ME"): "MONTH", } # pandas 3.0 removes the 'H' and 'M' frequency aliases, in favour of # 'h' and 'ME' so normalise them before calling to_offset alias = resolution if isinstance(alias, str): if alias.endswith("H"): alias = alias[:-1] + "h" elif alias in ("M", "1M"): alias = "ME" offset = to_offset(alias) if offset in d: return d[offset] else: logger.error(traceback.format_exc()) logger.warning(f"conv_resol returns '{resolution}'") return resolution else: return resolution def conv_datetime(dt, version=2): """Converts dt to string like version 1 = 2014:12:15-00:00:00 version 2 = 2014/12/15 00:00:00 version 3 = 2014/12/15 00:00:00 """ try: if isinstance(dt, six.string_types) and _HAS_PANDAS: dt = pd.to_datetime(dt) fmt = DATE_FORMATS[int(version)] return dt.strftime(fmt) except (ValueError, TypeError): logger.warning(f"conv_datetime returns {dt}") return dt def conv_to_ms(td): """Converts td to integer number of milliseconds""" try: if isinstance(td, six.integer_types): return td else: return int(td.total_seconds() * 1000.0) except ValueError: logger.error(traceback.format_exc()) logger.warning(f"conv_to_ms returns '{td}'") return td def remove(cache): """Remove cache""" try: filename = f"{cache}.sqlite" print(f"remove {filename}") os.remove(filename) except Exception as ex: logger.error(f"Problem removing cache file: {ex}") def print_full(x): """ Prints out a full data frame, no column hiding """ pd.set_option("display.max_rows", None) pd.set_option("display.max_columns", None) pd.set_option("display.width", 2000) # pd.set_option('display.float_format', '{:20,.2f}'.format) pd.set_option("display.max_colwidth", None) print(x) pd.reset_option("display.max_rows") pd.reset_option("display.max_columns") pd.reset_option("display.width") pd.reset_option("display.float_format") pd.reset_option("display.max_colwidth") def api_limit_hit(response_text): # note we don't check for historical data allowance - it only gets reset # once a week return ( "exceeded-api-key-allowance" in response_text or "exceeded-account-allowance" in response_text or "exceeded-account-trading-allowance" in response_text ) def token_invalid(response_text): return ( "oauth-token-invalid" in response_text or "client-token-invalid" in response_text )