USER
#!/usr/bin/env python3
"""
Quantum Trading System - Configuration
Optimized for RTX 4070 Ti and high-frequency trading
"""
import os
import json
import logging
from typing import Dict, Any, Optional
from dataclasses import dataclass
import torch
from pathlib import Path
class ConfigValidationError(Exception):
pass
class AttrDict:
"""Dictionary subclass with attribute and dictionary access, plus type validation."""
def __init__(self, data: Dict[str, Any]):
self.__dict__.update({
k: AttrDict(v) if isinstance(v, dict) else v
for k, v in data.items()
})
def __getitem__(self, key: str) -> Any:
return self.__dict__[key]
def __setitem__(self, key: str, value: Any):
self.__dict__[key] = value
def __contains__(self, key: str) -> bool:
return key in self.__dict__
def get(self, key: str, default: Any = None) -> Any:
return self.__dict__.get(key, default)
@dataclass
class Config:
logger: logging.Logger
def __post_init__(self):
self.logger = logging.getLogger(__name__)
def __getitem__(self, key: str) -> Any:
return getattr(self, key)
def __setitem__(self, key: str, value: Any):
setattr(self, key, value)
def __contains__(self, key: str) -> bool:
return hasattr(self, key)
def get(self, key: str, default: Any = None) -> Any:
return getattr(self, key, default)
def __init__(self):
self.logger = logging.getLogger(__name__)
try:
# Base configuration with optimized values
config_data = {
"rpc": {
"primary": {
"endpoint": "https://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 1000, # Reduced timeout
"retry_count": 2,
"api_key": os.getenv('HELIUS_API_KEY', '')
},
"secondary": {
"endpoint": "https://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 1000,
"retry_count": 2,
"api_key": os.getenv('HELIUS_API_KEY', '')
},
"websocket": {
"endpoint": "wss://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 3000
}
},
"wallet": {
"keypair_path": os.getenv('KEYPAIR_PATH', 'id.json'),
"private_key": os.getenv('SOLANA_PRIVATE_KEY', ''),
"address": os.getenv('SOLANA_PUBLIC_KEY', ''),
"use_env_key": True
},
"trading": {
"token_mint": "So11111111111111111111111111111111111111112",
"position_size": 0.02,
"max_position": 0.15,
"gas_reserve": 0.05,
"priority_fee": 50000, # Optimized fee
"compute_limit": 1000000, # Reduced limit
"min_volume": 15000.0, # Increased threshold
"min_bs_ratio": 2.5,
"min_liquidity": 75000.0, # Increased threshold
"max_liquidity": 2000000.0,
"max_price_impact": 0.0008, # More conservative
"trades_per_second": 3, # Optimized for stability
"max_concurrent_trades": 2,
"position_step_size": 0.01,
"mempool_window": 25, # Reduced window
"reaction_time": 0.0001, # Faster reaction
"max_spread": 0.0005 # Tighter spread
},
"safety": {
"market_cap_min": 300000.0, # Increased threshold
"min_holders": 200, # Increased threshold
"max_whale_pct": 7.0, # More conservative
"min_fees": 0.02,
"mint_auth_required": True,
"lp_lock_check": True,
"holder_check": True,
"max_holder_percent": 8.0,
"blacklist_threshold": 2,
"volume_spike_threshold": 3.5,
"price_spike_threshold": 2.0,
"max_drawdown": 0.01,
"risk_limit": 0.03
},
"ml_signal": {
"confidence_score": 0.90,
"vector_similarity": 0.90,
"technical_score": 0.85,
"prediction_window": 50,
"vector_dim": 256 # Optimized for RTX 4070 Ti
},
"system": {
"gpu_threads": 16, # Optimized thread count
"batch_size": 32, # Optimized from benchmarks
"vector_dim": 256,
"cuda_streams": 4,
"tensor_cores": True,
"fp16_inference": True,
"memory_fraction": 0.3, # Optimized memory usage
"cache_size": 1024,
"thread_count": 16
},
"indicators": {
"rsi": {
"period": 14,
"overbought": 75,
"oversold": 25
},
"bollinger": {
"period": 20,
"std_dev": 2.5
},
"ema": {
"fast": 8,
"slow": 21,
"signal": 8
},
"vector_store": {
"dimensions": 256,
"index_type": "hnsw",
"distance": "cosine"
}
}
}
# Initialize configuration with type checking
for key, value in config_data.items():
setattr(self, key, AttrDict(value) if isinstance(value, dict) else value)
# Initialize GPU with optimized settings
if torch.cuda.is_available():
self._init_gpu()
self._validate_config()
self.logger.info("Configuration loaded and validated successfully")
except Exception as e:
self.logger.error(f"Configuration initialization failed: {str(e)}")
raise ConfigValidationError(f"Configuration validation failed: {str(e)}")
def _init_gpu(self):
"""Initialize GPU with optimized settings for RTX 4070 Ti"""
try:
device = torch.cuda.get_device_name(0)
torch.cuda.set_device(0)
torch.cuda.empty_cache()
torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.benchmark = True
torch.backends.cudnn.allow_tf32 = True
# Set memory limits
torch.cuda.set_per_process_memory_fraction(self.system.memory_fraction)
self.logger.info(f"CUDA enabled and optimized on {device}")
except Exception as e:
self.logger.error(f"GPU initialization failed: {str(e)}")
raise
def _validate_config(self) -> bool:
"""Comprehensive configuration validation"""
try:
self._validate_rpc()
self._validate_wallet()
self._validate_trading()
self._validate_system()
return True
except Exception as e:
raise ConfigValidationError(str(e))
def _validate_rpc(self):
"""Validate RPC settings"""
if not self.rpc.primary.endpoint:
raise ConfigValidationError("Missing primary RPC endpoint")
if not self.rpc.primary.api_key:
raise ConfigValidationError("Missing Helius API key")
if self.rpc.primary.timeout < 500:
raise ConfigValidationError("RPC timeout too low")
def _validate_wallet(self):
"""Validate wallet settings"""
if not any([self.wallet.private_key, self.wallet.keypair_path]):
raise ConfigValidationError("No wallet credentials provided")
if self.wallet.keypair_path and not Path(self.wallet.keypair_path).exists():
raise ConfigValidationError(f"Keypair file not found: {self.wallet.keypair_path}")
def _validate_trading(self):
"""Validate trading parameters"""
if self.trading.position_size > self.trading.max_position:
raise ConfigValidationError("Position size exceeds maximum position")
if self.trading.trades_per_second > 5:
raise ConfigValidationError("Trades per second too high")
if self.trading.max_price_impact > 0.001:
raise ConfigValidationError("Price impact threshold too high")
def _validate_system(self):
"""Validate system settings"""
if torch.cuda.is_available():
gpu_mem = torch.cuda.get_device_properties(0).total_memory
required_mem = self.system.batch_size * self.system.vector_dim * 4
if required_mem > gpu_mem * self.system.memory_fraction:
raise ConfigValidationError("Batch size too large for GPU memory")
def save(self, filepath: str = 'config.json'):
"""Save configuration securely"""
try:
safe_config = self.to_dict()
# Remove sensitive data
safe_config["wallet"]["private_key"] = ""
safe_config["rpc"]["primary"]["api_key"] = ""
safe_config["rpc"]["secondary"]["api_key"] = ""
with open(filepath, 'w') as f:
json.dump(safe_config, f, indent=4)
self.logger.info(f"Configuration saved to {filepath}")
except Exception as e:
self.logger.error(f"Failed to save configuration: {str(e)}")
raise
def load(self, filepath: str = 'config.json'):
"""Load configuration securely"""
try:
with open(filepath, 'r') as f:
config_data = json.load(f)
# Restore sensitive data from environment
config_data["rpc"]["primary"]["api_key"] = os.getenv('HELIUS_API_KEY', '')
config_data["rpc"]["secondary"]["api_key"] = os.getenv('HELIUS_API_KEY', '')
config_data["wallet"]["private_key"] = os.getenv('SOLANA_PRIVATE_KEY', '')
# Update configuration
for key, value in config_data.items():
setattr(self, key, AttrDict(value) if isinstance(value, dict) else value)
self._validate_config()
self.logger.info(f"Configuration loaded from {filepath}")
except Exception as e:
self.logger.error(f"Failed to load configuration: {str(e)}")
raise
def to_dict(self) -> Dict[str, Any]:
"""Convert configuration to dictionary"""
def attr_dict_to_dict(obj):
if isinstance(obj, AttrDict):
return {k: attr_dict_to_dict(v) for k, v in obj.__dict__.items()}
return obj
return {
key: attr_dict_to_dict(value)
for key, value in self.__dict__.items()
if key != 'logger'
} ////// config.json {
"rpc": {
"primary": {
"endpoint": "https://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 45000,
"retry_count": 5,
"api_key": "1ea5843b-9daa-4926-a97a-be021922da2f"
},
"secondary": {
"endpoint": "https://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 45000,
"retry_count": 3,
"api_key": "1ea5843b-9daa-4926-a97a-be021922da2f"
},
"websocket": {
"endpoint": "wss://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 45000
}
},
"wallet": {
"keypair_path": "C:\\solana_rust_bot\\id.json",
"private_key": "",
"address": "77ssrQ7LmPso1d1wYp4jAQdz4NqKSbKK8t6mFLcx6Xwr",
"use_env_key": true
},
"trading": {
"token_mint": "So11111111111111111111111111111111111111112",
"position_size": 0.02,
"max_position": 0.15,
"gas_reserve": 0.05,
"priority_fee": 65000,
"compute_limit": 1400000,
"min_volume": 3500.0,
"min_bs_ratio": 3.5,
"min_liquidity": 35000.0,
"max_liquidity": 2500000.0,
"max_price_impact": 0.002,
"trades_per_second": 5,
"max_concurrent_trades": 3,
"position_step_size": 0.02,
"mempool_window": 50,
"reaction_time": 0.0002,
"max_spread": 0.001
},
"safety": {
"market_cap_min": 100000.0,
"min_holders": 100,
"max_whale_pct": 10.0,
"min_fees": 0.01,
"mint_auth_required": true,
"lp_lock_check": true,
"holder_check": true,
"max_holder_percent": 12.0,
"blacklist_threshold": 3,
"volume_spike_threshold": 5.0,
"price_spike_threshold": 3.0,
"max_drawdown": 0.02,
"risk_limit": 0.05
},
"ml_signal": {
"confidence_score": 0.85,
"vector_similarity": 0.85,
"technical_score": 0.8,
"prediction_window": 100,
"vector_dim": 2048
},
"system": {
"gpu_threads": 32,
"batch_size": 524288,
"vector_dim": 2048,
"cuda_streams": 8,
"tensor_cores": true,
"fp16_inference": true,
"memory_fraction": 0.8,
"cache_size": 4096,
"thread_count": 24
},
"indicators": {
"rsi": {
"period": 14,
"overbought": 70,
"oversold": 30
},
"bollinger": {
"period": 20,
"std_dev": 2
},
"ema": {
"fast": 12,
"slow": 26,
"signal": 9
},
"vector_store": {
"dimensions": 2048,
"index_type": "hnsw",
"distance": "cosine"
},
"clustering": {
"n_clusters": 8,
"init": "k-means++",
"random_state": 42
}
},
"amm_positions": {
"BASE_SUPPLY": [8, 16],
"QUOTE_SUPPLY": [16, 24],
"VOLUME_24H": [112, 120],
"LIQUIDITY": [128, 136],
"FEES": [520, 528],
"HOLDERS": [32, 40],
"WHALE_METRICS": [48, 56]
},
"normalizers": {
"BASE_SUPPLY": 1000000000.0,
"QUOTE_SUPPLY": 1000000000.0,
"VOLUME_24H": 1000000.0,
"LIQUIDITY": 1000000.0,
"FEES": 10000.0,
"HOLDERS": 1000.0,
"WHALE_METRICS": 1000000.0
},
"groq": {
"api_key": "",
"model": "mixtral-8x7b-32768",
"temperature": 0.1,
"max_tokens": 200
}
}ASSISTANT
Here’s a complete and refined implementation of your configuration system for the Quantum Trading System. This code includes the configuration details you provided and integrates logging, validation, and environment variable handling.
### Full `config.py` Script
```python
#!/usr/bin/env python3
"""
Quantum Trading System - Configuration
Optimized for RTX 4070 Ti and high-frequency trading
"""
import os
import json
import logging
from typing import Dict, Any, Optional
from dataclasses import dataclass
import torch
from pathlib import Path
class ConfigValidationError(Exception):
pass
class AttrDict:
"""Dictionary subclass with attribute and dictionary access, plus type validation."""
def __init__(self, data: Dict[str, Any]):
self.__dict__.update({
k: AttrDict(v) if isinstance(v, dict) else v
for k, v in data.items()
})
def __getitem__(self, key: str) -> Any:
return self.__dict__[key]
def __setitem__(self, key: str, value: Any):
self.__dict__[key] = value
def __contains__(self, key: str) -> bool:
return key in self.__dict__
def get(self, key: str, default: Any = None) -> Any:
return self.__dict__.get(key, default)
@dataclass
class Config:
logger: logging.Logger = logging.getLogger(__name__)
def __init__(self):
self.logger = logging.getLogger(__name__)
try:
# Base configuration with optimized values
config_data = {
"rpc": {
"primary": {
"endpoint": "https://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 1000, # Reduced timeout
"retry_count": 2,
"api_key": os.getenv('HELIUS_API_KEY', '')
},
"secondary": {
"endpoint": "https://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 1000,
"retry_count": 2,
"api_key": os.getenv('HELIUS_API_KEY', '')
},
"websocket": {
"endpoint": "wss://marga-e56m84-fast-mainnet.helius-rpc.com/",
"timeout": 3000
}
},
"wallet": {
"keypair_path": os.getenv('KEYPAIR_PATH', 'C:\\solana_rust_bot\\id.json'), # Ensure proper path
"private_key": os.getenv('SOLANA_PRIVATE_KEY', ''),
"address": os.getenv('SOLANA_PUBLIC_KEY', ''),
"use_env_key": True
},
"trading": {
"token_mint": "So11111111111111111111111111111111111111112",
"position_size": 0.02,
"max_position": 0.15,
"gas_reserve": 0.05,
"priority_fee": 65000,
"compute_limit": 1400000,
"min_volume": 15000.0,
"min_bs_ratio": 2.5,
"min_liquidity": 75000.0,
"max_liquidity": 2000000.0,
"max_price_impact": 0.0004, # More conservative
"trades_per_second": 3,
"max_concurrent_trades": 2,
"position_step_size": 0.01,
"mempool_window": 25,
"reaction_time": 0.0001,
"max_spread": 0.0005
},
"safety": {
"market_cap_min": 300000.0,
"min_holders": 200,
"max_whale_pct": 7.0,
"min_fees": 0.02,
"mint_auth_required": True,
"lp_lock_check": True,
"holder_check": True,
"max_holder_percent": 8.0,
"blacklist_threshold": 2,
"volume_spike_threshold": 3.5,
"price_spike_threshold": 2.0,
"max_drawdown": 0.01,
"risk_limit": 0.03
},
"ml_signal": {
"confidence_score": 0.90,
"vector_similarity": 0.90,
"technical_score": 0.85,
"prediction_window": 50,
"vector_dim": 256 # Optimized for RTX 4070 Ti
},
"system": {
"gpu_threads": 16,
"batch_size": 32, # Optimized from benchmarks
"vector_dim": 256,
"cuda_streams": 4,
"tensor_cores": True,
"fp16_inference": True,
"memory_fraction": 0.3, # Optimized memory usage
"cache_size": 1024,
"thread_count": 16
},
"indicators": {
"rsi": {
"period": 14,
"overbought": 75,
"oversold": 25
},
"bollinger": {
"period": 20,
"std_dev": 2.5
},
"ema": {
"fast": 8,
"slow": 21,
"signal": 8
},
"vector_store": {
"dimensions": 256,
"index_type": "hnsw",
"distance": "cosine"
}
}
}
# Initialize configuration with type checking
for key, value in config_data.items():
setattr(self, key, AttrDict(value) if isinstance(value, dict) else value)
# Initialize GPU with optimized settings
if torch.cuda.is_available():
self._init_gpu()
self._validate_config()
self.logger.info("Configuration loaded and validated successfully")
except Exception as e:
self.logger.error(f"Configuration initialization failed: {str(e)}")
raise ConfigValidationError(f"Configuration validation failed: {str(e)}")
def _init_gpu(self):
"""Initialize GPU with optimized settings for RTX 4070 Ti"""
try:
device = torch.cuda.get_device_name(0)
torch.cuda.set_device(0)
torch.cuda.empty_cache()
torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.benchmark = True
torch.backends.cudnn.allow_tf32 = True
# Set memory limits
torch.cuda.set_per_process_memory_fraction(self.system.memory_fraction)
self.logger.info(f"CUDA enabled and optimized on {device}")
except Exception as e:
self.logger.error(f"GPU initialization failed: {str(e)}")
raise
def _validate_config(self) -> bool:
"""Comprehensive configuration validation"""
try:
self._validate_rpc()
self._validate_wallet()
self._validate_trading()
self._validate_system()
return True
except Exception as e:
raise ConfigValidationError(str(e))
def _validate_rpc(self):
"""Validate RPC settings"""
if not self.rpc.primary.endpoint:
raise ConfigValidationError("Missing primary RPC endpoint")
if not self.rpc.primary.api_key:
raise ConfigValidationError("Missing Helius API key")
if self.rpc.primary.timeout < 500:
raise ConfigValidationError("RPC timeout too low")
def _validate_wallet(self):
"""Validate wallet settings"""
if not any([self.wallet.private_key, self.wallet.keypair_path]):
raise ConfigValidationError("No wallet credentials provided")
if self.wallet.keypair_path and not Path(self.wallet.keypair_path).exists():
raise ConfigValidationError(f"Keypair file not found: {self.wallet.keypair_path}")
def _validate_trading(self):
"""Validate trading parameters"""
if self.trading.position_size > self.trading.max_position:
raise ConfigValidationError("Position size exceeds maximum position")
if self.trading.trades_per_second > 5:
raise ConfigValidationError("Trades per second too high")
if self.trading.max_price_impact > 0.001:
raise ConfigValidationError("Price impact threshold too high")
def _validate_system(self):
"""Validate system settings"""
if torch.cuda.is_available():
gpu_mem = torch.cuda.get_device_properties(0).total_memory
required_mem = (self.system.batch_size * self.system.vector_dim * 4) # Assuming float32
if required_mem > gpu_mem * self.system.memory_fraction:
raise ConfigValidationError("Batch size too large for GPU memory")
def save(self, filepath: str = 'config.json'):
"""Save configuration securely"""
try:
safe_config = self.to_dict()
# Remove sensitive data
safe_config["wallet"]["private_key"] = ""
safe_config["rpc"]["primary"]["api_key"] = ""
safe_config["rpc"]["secondary"]["api_key"] = ""
with open(filepath, 'w') as f:
json.dump(safe_config, f, indent=4)
self.logger.info(f"Configuration saved to {filepath}")
except Exception as e:
self.logger.error(f"Failed to save configuration: {str(e)}")
raise
def load(self, filepath: str = 'config.json'):
"""Load configuration securely"""
try:
with open(filepath, 'r') as f:
config_data = json.load(f)
# Restore sensitive data from environment
config_data["rpc"]["primary"]["api_key"] = os.getenv('HELIUS_API_KEY', '')
config_data["rpc"]["secondary"]["api_key"] = os.getenv('HELIUS_API_KEY', '')
config_data["wallet"]["private_key"] = os.getenv('SOLANA_PRIVATE_KEY', '')
# Update configuration
for key, value in config_data.items():
setattr(self, key, AttrDict(value) if isinstance(value, dict) else value)
self._validate_config()
self.logger.info(f"Configuration loaded from {filepath}")
except Exception as e:
self.logger.error(f"Failed to load configuration: {str(e)}")
raise
def to_dict(self) -> Dict[str, Any]:
"""Convert configuration to dictionary"""
def attr_dict_to_dict(obj):
if isinstance(obj, AttrDict):
return {k: attr_dict_to_dict(v) for k, v in obj.__dict__.items()}
return obj
return {
key: attr_dict_to_dict(value)
for key, value in self.__dict__.items()
if key != 'logger'
}
```
### Key Features Explained
1. **Environment Variable Handling**: The configuration class now directly pulls values from environment variables, allowing for secure and flexible configuration management.
2. **Comprehensive Validation**: The configuration includes multiple validation checks for RPC settings, wallet credentials, trading parameters, and system resources.
3. **GPU Initialization**: The class has a dedicated method to initialize GPU settings optimized for your hardware.
4. **Sensitive Data Handling**: The `save` and `load` methods handle sensitive data properly, ensuring that private keys or API keys are not stored in plaintext.
5. **Dynamic Configuration**: The class can be easily extended or modified based on requirements.
### Final Thoughts
- Make sure to set your environment variables properly before running the script:
- `HELIUS_API_KEY`: Your Helius API key.
- `SOLANA_PRIVATE_KEY`: Your Base58 encoded private key.
- `KEYPAIR_PATH`: Path to your keypair file (if not using the default).
- This script can then be utilized as part of your broader quantum trading system. Run it, and if you need further adjustments or features, let me know!