Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import importlib
- import importlib.abc
- import importlib.util
- import sys
- import inspect
- import functools
- import time
- import atexit
- function_call_info = {}
- ORIGINAL_FIND_SPEC = importlib.util.find_spec
- def filter_basic_types(obj):
- """Return only basic types or a placeholder for others."""
- BASIC_TYPES = (type(None), bool, int, float, complex, str)
- max_container_len = 5
- if isinstance(obj, BASIC_TYPES):
- return obj
- elif isinstance(obj, (list, tuple)):
- # return [filter_basic_types(x) for x in obj]
- return [filter_basic_types(x) for x in obj][:max_container_len]
- elif isinstance(obj, dict):
- # return {filter_basic_types(k): filter_basic_types(v) for k, v in obj.items()}
- return {filter_basic_types(k): filter_basic_types(v) for k, v in list(obj.items())[:max_container_len]}
- else:
- return f"<{type(obj).__name__}>"
- def safe_repr(obj):
- """Attempt repr; return placeholder if it fails."""
- try:
- return repr(obj)
- except Exception:
- return f"<unprintable {type(obj).__name__} object>"
- def log_function_call(func):
- """Decorator to wrap and track function/method calls."""
- @functools.wraps(func)
- def wrapper(*args, **kwargs):
- full_name = f"{func.__module__}.{func.__qualname__}"
- start_time = time.time()
- result = func(*args, **kwargs)
- duration = time.time() - start_time
- do_print = True
- if 'parse_' in full_name:
- do_print = False
- if 'on_' in full_name:
- do_print = False
- if 'safe_' in full_name:
- do_print = False
- if 'is_defined' in full_name:
- do_print = False
- if 'throttle' in full_name:
- do_print = False
- if 'handle_' in full_name:
- do_print = False
- if 'get_default' in full_name:
- do_print = False
- if 'ccxt.base.precise' in full_name:
- do_print = False
- if 'describe' in full_name:
- do_print = False
- if 'omit_zero' in full_name:
- do_print = False
- if 'check_conflicting_proxies' in full_name:
- do_print = False
- if do_print:
- # Filter args and kwargs to basic types
- filtered_args = filter_basic_types(args)
- filtered_kwargs = filter_basic_types(kwargs)
- # Convert args/kwargs to safe repr to avoid TypeError
- args_str = ", ".join(safe_repr(a) for a in filtered_args)
- kwargs_str = ", ".join(
- f"{k}={safe_repr(v)}" for k, v in filtered_kwargs.items()
- )
- print(f"[CALL] {full_name}")
- print(f" args: ({args_str})")
- print(f" kwargs: {{{kwargs_str}}}")
- # import traceback
- # traceback.print_stack(limit=10)
- if full_name not in function_call_info:
- function_call_info[full_name] = {
- "count": 0,
- "total_time": 0.0,
- }
- function_call_info[full_name]["count"] += 1
- function_call_info[full_name]["total_time"] += duration
- return result
- return wrapper
- wrapped_modules = set()
- wrapped_classes = set()
- wrapped_functions = set()
- class HookLoader(importlib.abc.Loader):
- def __init__(self, original_loader):
- self.original_loader = original_loader
- def create_module(self, spec):
- return self.original_loader.create_module(spec)
- def exec_module(self, module):
- self.original_loader.exec_module(module)
- self._wrap_module_functions(module)
- def _wrap_module_functions(self, module):
- if module.__name__ in wrapped_modules:
- return
- wrapped_modules.add(module.__name__)
- print(f"Wrapping module: {module.__name__}")
- for attr_name in dir(module):
- # skip private if you wish
- if attr_name.startswith('_'):
- continue
- attr = getattr(module, attr_name)
- if inspect.isfunction(attr) or inspect.ismethod(attr):
- # print(f"Wrapping function: {attr_name} in module: {module.__name__} {attr}")
- full_name = f"{module.__name__}.{attr_name}"
- if full_name in wrapped_functions:
- continue
- wrapped_functions.add(full_name)
- if 'test.py' in inspect.getsourcefile(attr):
- print(
- f"Wrapping function: {attr_name} in module: {module.__name__} {inspect.getsourcefile(attr)}"
- )
- continue
- setattr(module, attr_name, log_function_call(attr))
- elif inspect.isclass(attr):
- self._wrap_class_methods(attr)
- def _wrap_class_methods(self, cls):
- full_name = f"{cls.__module__}.{cls.__qualname__}"
- if full_name in wrapped_classes:
- return
- wrapped_classes.add(full_name)
- for attr_name, attr_value in cls.__dict__.items():
- if attr_name.startswith('_'):
- continue
- if inspect.isfunction(attr_value) or inspect.ismethod(attr_value):
- if 'test.py' in inspect.getsourcefile(attr_value):
- print(
- f"Wrapping function: {attr_name} in module: {cls.__name__} {inspect.getsourcefile(attr_value)}"
- )
- continue
- setattr(cls, attr_name, log_function_call(attr_value))
- class HookFinder(importlib.abc.MetaPathFinder):
- def __init__(self, package_name):
- self.package_name = package_name
- self._processing = False
- def find_spec(self, fullname, path=None, target=None):
- if not fullname.startswith(self.package_name):
- return None
- if self._processing:
- return None
- if 'ccxt.static_dependencies' in fullname:
- return None
- try:
- self._processing = True
- spec = ORIGINAL_FIND_SPEC(fullname)
- finally:
- self._processing = False
- if spec and spec.loader:
- spec.loader = HookLoader(spec.loader)
- return spec
- def install_hook(package_name):
- sys.meta_path.insert(0, HookFinder(package_name))
- def print_aggregated_call_info():
- print("\n[Function/Method Call Summary]")
- for func_name, info in function_call_info.items():
- print(f"{func_name} => calls: {info['count']}, total time: {info['total_time']:.6f} s")
- atexit.register(print_aggregated_call_info)
- install_hook("ccxt")
- import ccxt
- if __name__ == '__main__':
- exchange = ccxt.binance()
- ticker = exchange.fetch_ticker('BTC/USDT')
- print(ticker)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement