From 3e82cfa128b7d94a46b37c8e415f7216a2b948d5 Mon Sep 17 00:00:00 2001 From: smypmsa Date: Tue, 5 Aug 2025 14:51:48 +0000 Subject: [PATCH] fix(core): replace os with pathlib --- src/config_loader.py | 12 +++++++----- src/trading/universal_trader.py | 26 ++++++++++++++------------ src/utils/idl_manager.py | 13 ++++++------- 3 files changed, 27 insertions(+), 24 deletions(-) diff --git a/src/config_loader.py b/src/config_loader.py index 1930a5a..26b37d5 100644 --- a/src/config_loader.py +++ b/src/config_loader.py @@ -2,8 +2,8 @@ Updated configuration validation with comprehensive platform support. """ -import glob import os +from pathlib import Path from typing import Any import yaml @@ -52,13 +52,15 @@ PLATFORM_LISTENER_COMPATIBILITY = { def load_bot_config(path: str) -> dict: """Load and validate a bot configuration from a YAML file.""" - with open(path) as f: + config_path = Path(path) + with config_path.open() as f: config = yaml.safe_load(f) env_file = config.get("env_file") if env_file: - env_path = os.path.join(os.path.dirname(path), env_file) - if os.path.exists(env_path): + config_path = Path(path) + env_path = config_path.parent / env_file + if env_path.exists(): load_dotenv(env_path, override=True) else: load_dotenv(env_file, override=True) @@ -295,7 +297,7 @@ def validate_all_platform_configs(config_dir: str = "bots") -> dict[str, Any]: "listener_distribution": {}, } - config_files = glob.glob(os.path.join(config_dir, "*.yaml")) + config_files = list(Path(config_dir).glob("*.yaml")) for config_file in config_files: try: diff --git a/src/trading/universal_trader.py b/src/trading/universal_trader.py index 40c4004..9cbe827 100644 --- a/src/trading/universal_trader.py +++ b/src/trading/universal_trader.py @@ -5,8 +5,8 @@ Cleaned up to remove all platform-specific hardcoding. import asyncio import json -import os from datetime import datetime +from pathlib import Path from time import monotonic import uvloop @@ -546,8 +546,9 @@ class UniversalTrader: async def _save_token_info(self, token_info: TokenInfo) -> None: """Save token information to a file.""" try: - os.makedirs("trades", exist_ok=True) - file_name = os.path.join("trades", f"{token_info.mint}.txt") + trades_dir = Path("trades") + trades_dir.mkdir(exist_ok=True) + file_path = trades_dir / f"{token_info.mint}.txt" # Convert to dictionary for saving - platform-agnostic token_dict = { @@ -575,17 +576,17 @@ class UniversalTrader: if field_value is not None: token_dict[field_name] = str(field_value) - with open(file_name, "w") as file: - file.write(json.dumps(token_dict, indent=2)) + file_path.write_text(json.dumps(token_dict, indent=2)) - logger.info(f"Token information saved to {file_name}") - except Exception as e: - logger.error(f"Failed to save token information: {e!s}") + logger.info(f"Token information saved to {file_path}") + except OSError: + logger.exception("Failed to save token information") def _log_trade(self, action: str, token_info: TokenInfo, price: float, amount: float, tx_hash: str | None) -> None: """Log trade information.""" try: - os.makedirs("trades", exist_ok=True) + trades_dir = Path("trades") + trades_dir.mkdir(exist_ok=True) log_entry = { "timestamp": datetime.utcnow().isoformat(), @@ -598,10 +599,11 @@ class UniversalTrader: "tx_hash": str(tx_hash) if tx_hash else None, } - with open("trades/trades.log", "a") as log_file: + log_file_path = trades_dir / "trades.log" + with log_file_path.open("a", encoding="utf-8") as log_file: log_file.write(json.dumps(log_entry) + "\n") - except Exception as e: - logger.error(f"Failed to log trade information: {e!s}") + except OSError: + logger.exception("Failed to log trade information") # Backward compatibility alias diff --git a/src/utils/idl_manager.py b/src/utils/idl_manager.py index ec82af3..576a7b2 100644 --- a/src/utils/idl_manager.py +++ b/src/utils/idl_manager.py @@ -5,7 +5,7 @@ This module provides a single point of IDL loading and management to avoid duplicate loading across multiple platform implementation classes. """ -import os +from pathlib import Path from typing import Any from interfaces.core import Platform @@ -27,14 +27,13 @@ class IDLManager: def _setup_platform_idl_paths(self) -> None: """Setup IDL file paths for each platform.""" # Get the project root directory (3 levels up from this file) - current_dir = os.path.dirname(os.path.abspath(__file__)) - project_root = os.path.join(current_dir, "..", "..") - project_root = os.path.normpath(project_root) + current_file = Path(__file__) + project_root = current_file.parent.parent.parent # Define IDL paths for each platform self._idl_paths = { - Platform.LETS_BONK: os.path.join(project_root, "idl", "raydium_launchlab_idl.json"), - Platform.PUMP_FUN: os.path.join(project_root, "idl", "pump_fun_idl.json"), + Platform.LETS_BONK: project_root / "idl" / "raydium_launchlab_idl.json", + Platform.PUMP_FUN: project_root / "idl" / "pump_fun_idl.json", } def get_parser(self, platform: Platform, verbose: bool = False) -> IDLParser: @@ -61,7 +60,7 @@ class IDLManager: idl_path = self._idl_paths[platform] # Verify IDL file exists - if not os.path.exists(idl_path): + if not idl_path.exists(): raise FileNotFoundError(f"IDL file not found for {platform.value} at {idl_path}") # Load and cache the parser