gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-08-06 12:07:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

wip(core): platform aware trading

Browse Source
This commit is contained in:
smypmsa
2025-08-02 13:27:07 +00:00
parent 9ec9eeb793
commit 227ef0b3bd
9 changed files with 699 additions and 506 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config_loader.py
+174 -143
View File
@@ -1,8 +1,5 @@
"""
Updated configuration validation with platform support.
This extends the existing config_loader.py to support platform selection
and validation while maintaining backward compatibility.
Updated configuration validation with comprehensive platform support.
"""
import os
@@ -27,77 +24,33 @@ REQUIRED_FIELDS = [
]
CONFIG_VALIDATION_RULES = [
# (path, type, min_value, max_value, error_message)
(
"trade.buy_amount",
(int, float),
0,
float("inf"),
"trade.buy_amount must be a positive number",
),
("trade.buy_amount", (int, float), 0, float("inf"), "trade.buy_amount must be a positive number"),
("trade.buy_slippage", float, 0, 1, "trade.buy_slippage must be between 0 and 1"),
("trade.sell_slippage", float, 0, 1, "trade.sell_slippage must be between 0 and 1"),
(
"priority_fees.fixed_amount",
int,
0,
float("inf"),
"priority_fees.fixed_amount must be a non-negative integer",
),
(
"priority_fees.extra_percentage",
float,
0,
1,
"priority_fees.extra_percentage must be between 0 and 1",
),
(
"priority_fees.hard_cap",
int,
0,
float("inf"),
"priority_fees.hard_cap must be a non-negative integer",
),
(
"retries.max_attempts",
int,
0,
100,
"retries.max_attempts must be between 0 and 100",
),
(
"filters.max_token_age",
(int, float),
0,
float("inf"),
"filters.max_token_age must be a non-negative number",
),
("priority_fees.fixed_amount", int, 0, float("inf"), "priority_fees.fixed_amount must be a non-negative integer"),
("priority_fees.extra_percentage", float, 0, 1, "priority_fees.extra_percentage must be between 0 and 1"),
("priority_fees.hard_cap", int, 0, float("inf"), "priority_fees.hard_cap must be a non-negative integer"),
("retries.max_attempts", int, 0, 100, "retries.max_attempts must be between 0 and 100"),
("filters.max_token_age", (int, float), 0, float("inf"), "filters.max_token_age must be a non-negative number"),
]
# Valid values for enum-like fields (extended with platform support)
# Valid values for enum-like fields
VALID_VALUES = {
"filters.listener_type": ["logs", "blocks", "geyser", "pumpportal"],
"cleanup.mode": ["disabled", "on_fail", "after_sell", "post_session"],
"trade.exit_strategy": ["time_based", "tp_sl", "manual"],
"platform": ["pump_fun", "lets_bonk"], # Platform validation
"platform": ["pump_fun", "lets_bonk"],
}
# Platform-specific listener compatibility
PLATFORM_LISTENER_COMPATIBILITY = {
Platform.PUMP_FUN: ["logs", "blocks", "geyser", "pumpportal"],
Platform.LETS_BONK: ["logs", "blocks", "geyser"], # PumpPortal is pump.fun only
}
def load_bot_config(path: str) -> dict:
"""
Load and validate a bot configuration from a YAML file.
Extended to support platform selection with backward compatibility.
Args:
path: Path to the YAML configuration file (relative or absolute)
Returns:
Validated configuration dictionary
Raises:
FileNotFoundError: If the configuration file doesn't exist
ValueError: If the configuration is invalid
"""
"""Load and validate a bot configuration from a YAML file."""
with open(path) as f:
config = yaml.safe_load(f)
@@ -107,7 +60,6 @@ def load_bot_config(path: str) -> dict:
if os.path.exists(env_path):
load_dotenv(env_path, override=True)
else:
# If not found relative to config, try relative to current working directory
load_dotenv(env_file, override=True)
resolve_env_vars(config)
@@ -117,18 +69,11 @@ def load_bot_config(path: str) -> dict:
config["platform"] = "pump_fun"
validate_config(config)
return config
def resolve_env_vars(config: dict) -> None:
"""
Recursively resolve environment variables in the configuration.
Args:
config: Configuration dictionary to process
"""
"""Recursively resolve environment variables in the configuration."""
def resolve_env(value):
if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
env_var = value[2:-1]
@@ -149,19 +94,7 @@ def resolve_env_vars(config: dict) -> None:
def get_nested_value(config: dict, path: str) -> Any:
"""
Get a nested value from the configuration using dot notation.
Args:
config: Configuration dictionary
path: Path to the value using dot notation (e.g., "trade.buy_amount")
Returns:
The value at the specified path
Raises:
ValueError: If the path doesn't exist in the configuration
"""
"""Get a nested value from the configuration using dot notation."""
keys = path.split(".")
value = config
for key in keys:
@@ -172,16 +105,7 @@ def get_nested_value(config: dict, path: str) -> Any:
def validate_config(config: dict) -> None:
"""
Validate the configuration against defined rules.
Extended to include platform-specific validation.
Args:
config: Configuration dictionary to validate
Raises:
ValueError: If the configuration is invalid
"""
"""Validate the configuration against defined rules with platform support."""
# Validate required fields
for field in REQUIRED_FIELDS:
get_nested_value(config, field)
@@ -198,10 +122,8 @@ def validate_config(config: dict) -> None:
raise ValueError(f"Range error: {error_msg}")
except ValueError as e:
# Re-raise if it's our own error
if str(e).startswith(("Type error:", "Range error:")):
raise
# Otherwise, the field might be missing
continue
# Validate enum-like fields
@@ -211,7 +133,6 @@ def validate_config(config: dict) -> None:
if value not in valid_values:
raise ValueError(f"{path} must be one of {valid_values}")
except ValueError:
# Skip if the field is missing
continue
# Cannot enable both dynamic and fixed priority fees
@@ -219,11 +140,8 @@ def validate_config(config: dict) -> None:
dynamic = get_nested_value(config, "priority_fees.enable_dynamic")
fixed = get_nested_value(config, "priority_fees.enable_fixed")
if dynamic and fixed:
raise ValueError(
"Cannot enable both dynamic and fixed priority fees simultaneously"
)
raise ValueError("Cannot enable both dynamic and fixed priority fees simultaneously")
except ValueError:
# Skip if one of the fields is missing
pass
# Platform-specific validation
@@ -238,16 +156,30 @@ def validate_config(config: dict) -> None:
def validate_platform_config(config: dict, platform: Platform) -> None:
"""
Validate platform-specific configuration requirements.
"""Validate platform-specific configuration requirements."""
# Check if platform is supported
try:
from platforms import platform_factory
if not platform_factory.registry.is_platform_supported(platform):
raise ValueError(f"Platform {platform.value} is not supported. Available platforms: {[p.value for p in platform_factory.get_supported_platforms()]}")
except ImportError:
# If platform factory not available, just validate enum
pass
Args:
config: Configuration dictionary
platform: Platform enum value
# Validate listener compatibility with platform
try:
listener_type = get_nested_value(config, "filters.listener_type")
compatible_listeners = PLATFORM_LISTENER_COMPATIBILITY.get(platform, [])
if listener_type not in compatible_listeners:
raise ValueError(
f"Listener type '{listener_type}' is not compatible with platform '{platform.value}'. "
f"Compatible listeners: {compatible_listeners}"
)
except ValueError:
pass
Raises:
ValueError: If platform-specific config is invalid
"""
# Platform-specific configuration validation
if platform == Platform.PUMP_FUN:
# pump.fun doesn't require additional config beyond base requirements
pass
@@ -259,18 +191,7 @@ def validate_platform_config(config: dict, platform: Platform) -> None:
def get_platform_from_config(config: dict) -> Platform:
"""
Extract platform enum from configuration.
Args:
config: Configuration dictionary
Returns:
Platform enum value
Raises:
ValueError: If platform is invalid
"""
"""Extract platform enum from configuration."""
platform_str = config.get("platform", "pump_fun")
try:
return Platform(platform_str)
@@ -278,46 +199,156 @@ def get_platform_from_config(config: dict) -> Platform:
raise ValueError(f"Invalid platform '{platform_str}'. Must be one of: {[p.value for p in Platform]}")
def print_config_summary(config: dict) -> None:
"""
Print a summary of the loaded configuration.
Extended to show platform information.
def validate_platform_listener_combination(platform: Platform, listener_type: str) -> bool:
"""Check if a platform and listener type are compatible.
Args:
config: Configuration dictionary
platform: Platform enum
listener_type: Listener type string
Returns:
True if combination is valid
"""
compatible_listeners = PLATFORM_LISTENER_COMPATIBILITY.get(platform, [])
return listener_type in compatible_listeners
def get_supported_listeners_for_platform(platform: Platform) -> list[str]:
"""Get list of supported listener types for a platform.
Args:
platform: Platform enum
Returns:
List of supported listener types
"""
return PLATFORM_LISTENER_COMPATIBILITY.get(platform, [])
def get_platform_specific_required_config(platform: Platform) -> list[str]:
"""Get platform-specific required configuration paths.
Args:
platform: Platform enum
Returns:
List of additional required config paths for the platform
"""
if platform == Platform.PUMP_FUN:
return [] # No additional requirements
elif platform == Platform.LETS_BONK:
return [] # No additional requirements yet
else:
return []
def print_config_summary(config: dict) -> None:
"""Print a summary of the loaded configuration with platform info."""
platform_str = config.get("platform", "pump_fun")
print(f"Bot name: {config.get('name', 'unnamed')}")
print(f"Platform: {platform_str}")
print(
f"Listener type: {config.get('filters', {}).get('listener_type', 'not configured')}"
)
print(f"Listener type: {config.get('filters', {}).get('listener_type', 'not configured')}")
# Validate platform-listener combination
try:
platform = Platform(platform_str)
listener_type = config.get('filters', {}).get('listener_type')
if listener_type and not validate_platform_listener_combination(platform, listener_type):
print(f"WARNING: Listener '{listener_type}' may not be compatible with platform '{platform_str}'")
except ValueError:
print(f"WARNING: Invalid platform '{platform_str}'")
trade = config.get("trade", {})
print("Trade settings:")
print(f" - Buy amount: {trade.get('buy_amount', 'not configured')} SOL")
print(f" - Buy slippage: {trade.get('buy_slippage', 'not configured') * 100}%")
print(
f" - Extreme fast mode: {'enabled' if trade.get('extreme_fast_mode') else 'disabled'}"
)
print(f" - Extreme fast mode: {'enabled' if trade.get('extreme_fast_mode') else 'disabled'}")
fees = config.get("priority_fees", {})
print("Priority fees:")
if fees.get("enable_dynamic"):
print(" - Dynamic fees enabled")
elif fees.get("enable_fixed"):
print(
f" - Fixed fee: {fees.get('fixed_amount', 'not configured')} microlamports"
)
print(f" - Fixed fee: {fees.get('fixed_amount', 'not configured')} microlamports")
print("Configuration loaded successfully!")
if __name__ == "__main__":
# Example usage with platform configuration
config = load_bot_config("bots/bot-sniper.yaml")
print_config_summary(config)
def validate_all_platform_configs(config_dir: str = "bots") -> dict[str, Any]:
"""Validate all bot configurations in a directory.
platform = get_platform_from_config(config)
print(f"Detected platform: {platform}")
Args:
config_dir: Directory containing bot config files
Returns:
Dictionary with validation results
"""
import glob
import os
results = {
"valid_configs": [],
"invalid_configs": [],
"platform_distribution": {},
"listener_distribution": {},
}
config_files = glob.glob(os.path.join(config_dir, "*.yaml"))
for config_file in config_files:
try:
config = load_bot_config(config_file)
platform = get_platform_from_config(config)
listener_type = config.get('filters', {}).get('listener_type', 'unknown')
results["valid_configs"].append({
"file": config_file,
"name": config.get("name"),
"platform": platform.value,
"listener": listener_type,
"enabled": config.get("enabled", True)
})
# Track distributions
platform_key = platform.value
results["platform_distribution"][platform_key] = results["platform_distribution"].get(platform_key, 0) + 1
results["listener_distribution"][listener_type] = results["listener_distribution"].get(listener_type, 0) + 1
except Exception as e:
results["invalid_configs"].append({
"file": config_file,
"error": str(e)
})
return results
if __name__ == "__main__":
# Example usage with platform configuration validation
import sys
if len(sys.argv) > 1:
config_path = sys.argv[1]
try:
config = load_bot_config(config_path)
print_config_summary(config)
platform = get_platform_from_config(config)
print(f"Detected platform: {platform}")
print(f"Supported listeners for this platform: {get_supported_listeners_for_platform(platform)}")
except Exception as e:
print(f"Configuration error: {e}")
else:
# Validate all configs in bots directory
results = validate_all_platform_configs()
print("Configuration validation results:")
print(f"Valid configs: {len(results['valid_configs'])}")
print(f"Invalid configs: {len(results['invalid_configs'])}")
print(f"Platform distribution: {results['platform_distribution']}")
print(f"Listener distribution: {results['listener_distribution']}")
if results['invalid_configs']:
print("\nInvalid configurations:")
for invalid in results['invalid_configs']:
print(f" {invalid['file']}: {invalid['error']}")