mirror of
https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git
synced 2026-08-15 08:18:04 +00:00
fix(core): replace os with pathlib
This commit is contained in:
@@ -2,8 +2,8 @@
|
|||||||
Updated configuration validation with comprehensive platform support.
|
Updated configuration validation with comprehensive platform support.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import glob
|
|
||||||
import os
|
import os
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
@@ -52,13 +52,15 @@ PLATFORM_LISTENER_COMPATIBILITY = {
|
|||||||
|
|
||||||
def load_bot_config(path: str) -> dict:
|
def load_bot_config(path: str) -> dict:
|
||||||
"""Load and validate a bot configuration from a YAML file."""
|
"""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)
|
config = yaml.safe_load(f)
|
||||||
|
|
||||||
env_file = config.get("env_file")
|
env_file = config.get("env_file")
|
||||||
if env_file:
|
if env_file:
|
||||||
env_path = os.path.join(os.path.dirname(path), env_file)
|
config_path = Path(path)
|
||||||
if os.path.exists(env_path):
|
env_path = config_path.parent / env_file
|
||||||
|
if env_path.exists():
|
||||||
load_dotenv(env_path, override=True)
|
load_dotenv(env_path, override=True)
|
||||||
else:
|
else:
|
||||||
load_dotenv(env_file, override=True)
|
load_dotenv(env_file, override=True)
|
||||||
@@ -295,7 +297,7 @@ def validate_all_platform_configs(config_dir: str = "bots") -> dict[str, Any]:
|
|||||||
"listener_distribution": {},
|
"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:
|
for config_file in config_files:
|
||||||
try:
|
try:
|
||||||
|
|||||||
@@ -5,8 +5,8 @@ Cleaned up to remove all platform-specific hardcoding.
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import json
|
import json
|
||||||
import os
|
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from pathlib import Path
|
||||||
from time import monotonic
|
from time import monotonic
|
||||||
|
|
||||||
import uvloop
|
import uvloop
|
||||||
@@ -546,8 +546,9 @@ class UniversalTrader:
|
|||||||
async def _save_token_info(self, token_info: TokenInfo) -> None:
|
async def _save_token_info(self, token_info: TokenInfo) -> None:
|
||||||
"""Save token information to a file."""
|
"""Save token information to a file."""
|
||||||
try:
|
try:
|
||||||
os.makedirs("trades", exist_ok=True)
|
trades_dir = Path("trades")
|
||||||
file_name = os.path.join("trades", f"{token_info.mint}.txt")
|
trades_dir.mkdir(exist_ok=True)
|
||||||
|
file_path = trades_dir / f"{token_info.mint}.txt"
|
||||||
|
|
||||||
# Convert to dictionary for saving - platform-agnostic
|
# Convert to dictionary for saving - platform-agnostic
|
||||||
token_dict = {
|
token_dict = {
|
||||||
@@ -575,17 +576,17 @@ class UniversalTrader:
|
|||||||
if field_value is not None:
|
if field_value is not None:
|
||||||
token_dict[field_name] = str(field_value)
|
token_dict[field_name] = str(field_value)
|
||||||
|
|
||||||
with open(file_name, "w") as file:
|
file_path.write_text(json.dumps(token_dict, indent=2))
|
||||||
file.write(json.dumps(token_dict, indent=2))
|
|
||||||
|
|
||||||
logger.info(f"Token information saved to {file_name}")
|
logger.info(f"Token information saved to {file_path}")
|
||||||
except Exception as e:
|
except OSError:
|
||||||
logger.error(f"Failed to save token information: {e!s}")
|
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:
|
def _log_trade(self, action: str, token_info: TokenInfo, price: float, amount: float, tx_hash: str | None) -> None:
|
||||||
"""Log trade information."""
|
"""Log trade information."""
|
||||||
try:
|
try:
|
||||||
os.makedirs("trades", exist_ok=True)
|
trades_dir = Path("trades")
|
||||||
|
trades_dir.mkdir(exist_ok=True)
|
||||||
|
|
||||||
log_entry = {
|
log_entry = {
|
||||||
"timestamp": datetime.utcnow().isoformat(),
|
"timestamp": datetime.utcnow().isoformat(),
|
||||||
@@ -598,10 +599,11 @@ class UniversalTrader:
|
|||||||
"tx_hash": str(tx_hash) if tx_hash else None,
|
"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")
|
log_file.write(json.dumps(log_entry) + "\n")
|
||||||
except Exception as e:
|
except OSError:
|
||||||
logger.error(f"Failed to log trade information: {e!s}")
|
logger.exception("Failed to log trade information")
|
||||||
|
|
||||||
|
|
||||||
# Backward compatibility alias
|
# Backward compatibility alias
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ This module provides a single point of IDL loading and management to avoid
|
|||||||
duplicate loading across multiple platform implementation classes.
|
duplicate loading across multiple platform implementation classes.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import os
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from interfaces.core import Platform
|
from interfaces.core import Platform
|
||||||
@@ -27,14 +27,13 @@ class IDLManager:
|
|||||||
def _setup_platform_idl_paths(self) -> None:
|
def _setup_platform_idl_paths(self) -> None:
|
||||||
"""Setup IDL file paths for each platform."""
|
"""Setup IDL file paths for each platform."""
|
||||||
# Get the project root directory (3 levels up from this file)
|
# Get the project root directory (3 levels up from this file)
|
||||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
current_file = Path(__file__)
|
||||||
project_root = os.path.join(current_dir, "..", "..")
|
project_root = current_file.parent.parent.parent
|
||||||
project_root = os.path.normpath(project_root)
|
|
||||||
|
|
||||||
# Define IDL paths for each platform
|
# Define IDL paths for each platform
|
||||||
self._idl_paths = {
|
self._idl_paths = {
|
||||||
Platform.LETS_BONK: os.path.join(project_root, "idl", "raydium_launchlab_idl.json"),
|
Platform.LETS_BONK: project_root / "idl" / "raydium_launchlab_idl.json",
|
||||||
Platform.PUMP_FUN: os.path.join(project_root, "idl", "pump_fun_idl.json"),
|
Platform.PUMP_FUN: project_root / "idl" / "pump_fun_idl.json",
|
||||||
}
|
}
|
||||||
|
|
||||||
def get_parser(self, platform: Platform, verbose: bool = False) -> IDLParser:
|
def get_parser(self, platform: Platform, verbose: bool = False) -> IDLParser:
|
||||||
@@ -61,7 +60,7 @@ class IDLManager:
|
|||||||
idl_path = self._idl_paths[platform]
|
idl_path = self._idl_paths[platform]
|
||||||
|
|
||||||
# Verify IDL file exists
|
# 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}")
|
raise FileNotFoundError(f"IDL file not found for {platform.value} at {idl_path}")
|
||||||
|
|
||||||
# Load and cache the parser
|
# Load and cache the parser
|
||||||
|
|||||||
Reference in New Issue
Block a user