d34da789ec
- Introduced `config_utils.py` for loading and managing strategy configurations, including functions for loading templates, generating dropdown options, and retrieving parameter schemas and default values. - Added JSON templates for EMA Crossover, MACD, and RSI strategies, defining their parameters and validation rules to enhance modularity and maintainability. - Implemented `StrategyManager` in `manager.py` for managing user-defined strategies with file-based storage, supporting easy sharing and portability. - Updated `__init__.py` to include new components and ensure proper module exports. - Enhanced error handling and logging practices across the new modules for improved reliability. These changes establish a robust foundation for strategy management and configuration, aligning with project goals for modularity, performance, and maintainability.
407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""
|
|
Strategy Management System
|
|
|
|
This module provides functionality to manage user-defined strategies with
|
|
file-based storage. Each strategy is saved as a separate JSON file for
|
|
portability and easy sharing.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Any, Tuple
|
|
from dataclasses import dataclass, asdict
|
|
from enum import Enum
|
|
import importlib
|
|
|
|
from utils.logger import get_logger
|
|
|
|
# Initialize logger
|
|
logger = get_logger()
|
|
|
|
# Base directory for strategies
|
|
STRATEGIES_DIR = Path("config/strategies")
|
|
USER_STRATEGIES_DIR = STRATEGIES_DIR / "user_strategies"
|
|
TEMPLATES_DIR = STRATEGIES_DIR / "templates"
|
|
|
|
|
|
class StrategyType(str, Enum):
|
|
"""Supported strategy types."""
|
|
EMA_CROSSOVER = "ema_crossover"
|
|
RSI = "rsi"
|
|
MACD = "macd"
|
|
|
|
|
|
class StrategyCategory(str, Enum):
|
|
"""Strategy categories."""
|
|
TREND_FOLLOWING = "trend_following"
|
|
MOMENTUM = "momentum"
|
|
MEAN_REVERSION = "mean_reversion"
|
|
SCALPING = "scalping"
|
|
SWING_TRADING = "swing_trading"
|
|
|
|
|
|
@dataclass
|
|
class StrategyConfig:
|
|
"""Strategy configuration data."""
|
|
id: str
|
|
name: str
|
|
description: str
|
|
strategy_type: str # StrategyType
|
|
category: str # StrategyCategory
|
|
parameters: Dict[str, Any]
|
|
timeframes: List[str]
|
|
enabled: bool = True
|
|
created_date: str = ""
|
|
modified_date: str = ""
|
|
|
|
def __post_init__(self):
|
|
"""Initialize timestamps if not provided."""
|
|
current_time = datetime.now(timezone.utc).isoformat()
|
|
if not self.created_date:
|
|
self.created_date = current_time
|
|
if not self.modified_date:
|
|
self.modified_date = current_time
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary for JSON serialization."""
|
|
return {
|
|
'id': self.id,
|
|
'name': self.name,
|
|
'description': self.description,
|
|
'strategy_type': self.strategy_type,
|
|
'category': self.category,
|
|
'parameters': self.parameters,
|
|
'timeframes': self.timeframes,
|
|
'enabled': self.enabled,
|
|
'created_date': self.created_date,
|
|
'modified_date': self.modified_date
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'StrategyConfig':
|
|
"""Create StrategyConfig from dictionary."""
|
|
return cls(
|
|
id=data['id'],
|
|
name=data['name'],
|
|
description=data.get('description', ''),
|
|
strategy_type=data['strategy_type'],
|
|
category=data.get('category', 'trend_following'),
|
|
parameters=data.get('parameters', {}),
|
|
timeframes=data.get('timeframes', []),
|
|
enabled=data.get('enabled', True),
|
|
created_date=data.get('created_date', ''),
|
|
modified_date=data.get('modified_date', '')
|
|
)
|
|
|
|
|
|
class StrategyManager:
|
|
"""Manager for user-defined strategies with file-based storage."""
|
|
|
|
def __init__(self):
|
|
"""Initialize the strategy manager."""
|
|
self.logger = logger
|
|
self._ensure_directories()
|
|
|
|
def _ensure_directories(self):
|
|
"""Ensure strategy directories exist."""
|
|
try:
|
|
USER_STRATEGIES_DIR.mkdir(parents=True, exist_ok=True)
|
|
TEMPLATES_DIR.mkdir(parents=True, exist_ok=True)
|
|
self.logger.debug("Strategy manager: Strategy directories created/verified")
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error creating strategy directories: {e}")
|
|
|
|
def _get_strategy_file_path(self, strategy_id: str) -> Path:
|
|
"""Get file path for a strategy."""
|
|
return USER_STRATEGIES_DIR / f"{strategy_id}.json"
|
|
|
|
def _get_template_file_path(self, strategy_type: str) -> Path:
|
|
"""Get file path for a strategy template."""
|
|
return TEMPLATES_DIR / f"{strategy_type}_template.json"
|
|
|
|
def save_strategy(self, strategy: StrategyConfig) -> bool:
|
|
"""
|
|
Save a strategy to file.
|
|
|
|
Args:
|
|
strategy: StrategyConfig instance to save
|
|
|
|
Returns:
|
|
True if saved successfully, False otherwise
|
|
"""
|
|
try:
|
|
# Update modified date
|
|
strategy.modified_date = datetime.now(timezone.utc).isoformat()
|
|
|
|
file_path = self._get_strategy_file_path(strategy.id)
|
|
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(strategy.to_dict(), f, indent=2, ensure_ascii=False)
|
|
|
|
self.logger.info(f"Strategy manager: Saved strategy: {strategy.name} ({strategy.id})")
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error saving strategy {strategy.id}: {e}")
|
|
return False
|
|
|
|
def load_strategy(self, strategy_id: str) -> Optional[StrategyConfig]:
|
|
"""
|
|
Load a strategy from file.
|
|
|
|
Args:
|
|
strategy_id: ID of the strategy to load
|
|
|
|
Returns:
|
|
StrategyConfig instance or None if not found/error
|
|
"""
|
|
try:
|
|
file_path = self._get_strategy_file_path(strategy_id)
|
|
|
|
if not file_path.exists():
|
|
self.logger.warning(f"Strategy manager: Strategy file not found: {strategy_id}")
|
|
return None
|
|
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
strategy = StrategyConfig.from_dict(data)
|
|
self.logger.debug(f"Strategy manager: Loaded strategy: {strategy.name} ({strategy.id})")
|
|
return strategy
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error loading strategy {strategy_id}: {e}")
|
|
return None
|
|
|
|
def list_strategies(self, enabled_only: bool = False) -> List[StrategyConfig]:
|
|
"""
|
|
List all user strategies.
|
|
|
|
Args:
|
|
enabled_only: If True, only return enabled strategies
|
|
|
|
Returns:
|
|
List of StrategyConfig instances
|
|
"""
|
|
strategies = []
|
|
|
|
try:
|
|
if not USER_STRATEGIES_DIR.exists():
|
|
return strategies
|
|
|
|
for file_path in USER_STRATEGIES_DIR.glob("*.json"):
|
|
strategy = self.load_strategy(file_path.stem)
|
|
if strategy:
|
|
if not enabled_only or strategy.enabled:
|
|
strategies.append(strategy)
|
|
|
|
# Sort by name
|
|
strategies.sort(key=lambda s: s.name.lower())
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error listing strategies: {e}")
|
|
|
|
return strategies
|
|
|
|
def delete_strategy(self, strategy_id: str) -> bool:
|
|
"""
|
|
Delete a strategy file.
|
|
|
|
Args:
|
|
strategy_id: ID of the strategy to delete
|
|
|
|
Returns:
|
|
True if deleted successfully, False otherwise
|
|
"""
|
|
try:
|
|
file_path = self._get_strategy_file_path(strategy_id)
|
|
|
|
if file_path.exists():
|
|
file_path.unlink()
|
|
self.logger.info(f"Strategy manager: Deleted strategy: {strategy_id}")
|
|
return True
|
|
else:
|
|
self.logger.warning(f"Strategy manager: Strategy file not found for deletion: {strategy_id}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error deleting strategy {strategy_id}: {e}")
|
|
return False
|
|
|
|
def create_strategy(self, name: str, strategy_type: str, parameters: Dict[str, Any],
|
|
description: str = "", category: str = None,
|
|
timeframes: List[str] = None) -> Optional[StrategyConfig]:
|
|
"""
|
|
Create a new strategy with validation.
|
|
|
|
Args:
|
|
name: Strategy name
|
|
strategy_type: Type of strategy (must be valid StrategyType)
|
|
parameters: Strategy parameters
|
|
description: Optional description
|
|
category: Strategy category (defaults based on type)
|
|
timeframes: Supported timeframes (defaults to common ones)
|
|
|
|
Returns:
|
|
StrategyConfig instance if created successfully, None otherwise
|
|
"""
|
|
try:
|
|
# Validate strategy type
|
|
if strategy_type not in [t.value for t in StrategyType]:
|
|
self.logger.error(f"Strategy manager: Invalid strategy type: {strategy_type}")
|
|
return None
|
|
|
|
# Validate parameters against template
|
|
if not self._validate_parameters(strategy_type, parameters):
|
|
self.logger.error(f"Strategy manager: Invalid parameters for strategy type: {strategy_type}")
|
|
return None
|
|
|
|
# Set defaults
|
|
if category is None:
|
|
category = self._get_default_category(strategy_type)
|
|
|
|
if timeframes is None:
|
|
timeframes = self._get_default_timeframes(strategy_type)
|
|
|
|
# Create strategy
|
|
strategy = StrategyConfig(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
description=description,
|
|
strategy_type=strategy_type,
|
|
category=category,
|
|
parameters=parameters,
|
|
timeframes=timeframes,
|
|
enabled=True
|
|
)
|
|
|
|
# Save strategy
|
|
if self.save_strategy(strategy):
|
|
self.logger.info(f"Strategy manager: Created strategy: {name}")
|
|
return strategy
|
|
else:
|
|
return None
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error creating strategy: {e}")
|
|
return None
|
|
|
|
def update_strategy(self, strategy_id: str, **updates) -> bool:
|
|
"""
|
|
Update an existing strategy.
|
|
|
|
Args:
|
|
strategy_id: ID of strategy to update
|
|
**updates: Fields to update
|
|
|
|
Returns:
|
|
True if updated successfully, False otherwise
|
|
"""
|
|
try:
|
|
strategy = self.load_strategy(strategy_id)
|
|
if not strategy:
|
|
return False
|
|
|
|
# Update fields
|
|
for field, value in updates.items():
|
|
if hasattr(strategy, field):
|
|
setattr(strategy, field, value)
|
|
|
|
# Validate parameters if they were updated
|
|
if 'parameters' in updates:
|
|
if not self._validate_parameters(strategy.strategy_type, strategy.parameters):
|
|
self.logger.error(f"Strategy manager: Invalid parameters for update")
|
|
return False
|
|
|
|
return self.save_strategy(strategy)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error updating strategy {strategy_id}: {e}")
|
|
return False
|
|
|
|
def get_strategies_by_category(self, category: str) -> List[StrategyConfig]:
|
|
"""Get strategies filtered by category."""
|
|
return [s for s in self.list_strategies() if s.category == category]
|
|
|
|
def get_available_strategy_types(self) -> List[str]:
|
|
"""Get list of available strategy types."""
|
|
return [t.value for t in StrategyType]
|
|
|
|
def _get_default_category(self, strategy_type: str) -> str:
|
|
"""Get default category for a strategy type."""
|
|
category_mapping = {
|
|
StrategyType.EMA_CROSSOVER.value: StrategyCategory.TREND_FOLLOWING.value,
|
|
StrategyType.RSI.value: StrategyCategory.MOMENTUM.value,
|
|
StrategyType.MACD.value: StrategyCategory.TREND_FOLLOWING.value,
|
|
}
|
|
return category_mapping.get(strategy_type, StrategyCategory.TREND_FOLLOWING.value)
|
|
|
|
def _get_default_timeframes(self, strategy_type: str) -> List[str]:
|
|
"""Get default timeframes for a strategy type."""
|
|
timeframe_mapping = {
|
|
StrategyType.EMA_CROSSOVER.value: ["1h", "4h", "1d"],
|
|
StrategyType.RSI.value: ["15m", "1h", "4h", "1d"],
|
|
StrategyType.MACD.value: ["1h", "4h", "1d"],
|
|
}
|
|
return timeframe_mapping.get(strategy_type, ["1h", "4h", "1d"])
|
|
|
|
def _validate_parameters(self, strategy_type: str, parameters: Dict[str, Any]) -> bool:
|
|
"""Validate strategy parameters against template."""
|
|
try:
|
|
# Import here to avoid circular dependency
|
|
from config.strategies.config_utils import validate_strategy_parameters
|
|
|
|
is_valid, errors = validate_strategy_parameters(strategy_type, parameters)
|
|
if not is_valid:
|
|
for error in errors:
|
|
self.logger.error(f"Strategy manager: Parameter validation error: {error}")
|
|
|
|
return is_valid
|
|
|
|
except ImportError:
|
|
self.logger.warning("Strategy manager: Could not import validation function, skipping parameter validation")
|
|
return True
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error validating parameters: {e}")
|
|
return False
|
|
|
|
def get_template(self, strategy_type: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Load strategy template for the given type.
|
|
|
|
Args:
|
|
strategy_type: Strategy type to get template for
|
|
|
|
Returns:
|
|
Template dictionary or None if not found
|
|
"""
|
|
try:
|
|
file_path = self._get_template_file_path(strategy_type)
|
|
|
|
if not file_path.exists():
|
|
self.logger.warning(f"Strategy manager: Template not found: {strategy_type}")
|
|
return None
|
|
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
template = json.load(f)
|
|
|
|
return template
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Strategy manager: Error loading template {strategy_type}: {e}")
|
|
return None
|
|
|
|
|
|
# Global strategy manager instance
|
|
_strategy_manager = None
|
|
|
|
|
|
def get_strategy_manager() -> StrategyManager:
|
|
"""Get global strategy manager instance (singleton pattern)."""
|
|
global _strategy_manager
|
|
if _strategy_manager is None:
|
|
_strategy_manager = StrategyManager()
|
|
return _strategy_manager |