Utilities & Configuration

class Config:
    """
    Configuration management class with dict-like interface.
    
    Provides hierarchical configuration with validation, serialization,
    and dynamic updates. Supports nested configurations and type checking.
    """
    
    def __init__(self, config=None, **kwargs):
        """
        Initialize configuration.
        
        Parameters:
        - config: dict, initial configuration values
        - kwargs: additional configuration parameters
        """
    
    def __getitem__(self, key):
        """Get configuration value."""
    
    def __setitem__(self, key, value):
        """Set configuration value."""
    
    def get(self, key, default=None):
        """Get configuration value with default."""
    
    def update(self, other):
        """Update configuration from another config or dict."""
    
    def copy(self):
        """Create a copy of the configuration."""
    
    def reset(self):
        """Reset to default configuration."""

class Configured:
    """
    Base class for configurable objects.
    
    Provides automatic configuration inheritance and parameter management
    for all VectorBT components.
    """
    
    def __init__(self, **kwargs):
        """Initialize with configuration parameters."""
    
    def configure(self, **kwargs):
        """Update configuration."""
    
    def get_config(self):
        """Get current configuration."""

class AtomicConfig:
    """
    Thread-safe configuration class.
    
    Provides atomic operations for configuration updates in
    multi-threaded environments.
    """
    
    def __init__(self, config=None):
        """Initialize atomic configuration."""
    
    def atomic_update(self, updates):
        """Atomically update multiple configuration values."""

# Global settings instance
vbt.settings: Config

class CacheCondition:
    """
    Cache condition management for intelligent caching.
    
    Determines when to cache results based on input parameters,
    memory constraints, and computation complexity.
    """
    
    def __init__(self, condition_func=None):
        """
        Initialize cache condition.
        
        Parameters:
        - condition_func: callable, function determining cache eligibility
        """
    
    def should_cache(self, *args, **kwargs):
        """Determine if result should be cached."""

def cached_property(func):
    """
    Property caching decorator.
    
    Caches property values on first access with automatic invalidation
    when underlying data changes.
    
    Parameters:
    - func: callable, property getter function
    
    Returns:
    property: Cached property descriptor
    """

def cached_method(func):
    """
    Method caching decorator.
    
    Caches method results based on input parameters with intelligent
    cache key generation and memory management.
    
    Parameters:
    - func: callable, method to cache
    
    Returns:
    callable: Cache-enabled method
    """

def atomic_dict(*args, **kwargs):
    """
    Create atomic dictionary for thread-safe operations.
    
    Parameters:
    - args: positional arguments for dict initialization
    - kwargs: keyword arguments for dict initialization
    
    Returns:
    dict: Thread-safe atomic dictionary
    """

def merge_dicts(*dicts, **kwargs):
    """
    Merge multiple dictionaries with conflict resolution.
    
    Parameters:
    - dicts: sequence of dictionaries to merge
    - strategy: str, merge strategy ('override', 'keep_first', 'deep_merge')
    - in_place: bool, modify first dictionary in place
    
    Returns:
    dict: Merged dictionary
    """

class Sub:
    """
    Template substitution class.
    
    Provides framework for template-based code generation with
    parameter validation and nested substitutions.
    """
    
    def __init__(self, template, **kwargs):
        """
        Initialize substitution template.
        
        Parameters:
        - template: str, template string with placeholders
        - kwargs: substitution parameters
        """
    
    def substitute(self, **kwargs):
        """Perform template substitution."""

class Rep:
    """
    Template replacement class.
    
    Advanced replacement system supporting conditional logic,
    loops, and complex transformations.
    """
    
    def __init__(self, pattern, replacement):
        """
        Initialize replacement rule.
        
        Parameters:
        - pattern: str or regex, pattern to match
        - replacement: str or callable, replacement value
        """
    
    def apply(self, text):
        """Apply replacement to text."""

class RepEval:
    """
    Evaluation-based replacement class.
    
    Performs replacements based on evaluated expressions,
    enabling dynamic template generation.
    """
    
    def __init__(self, pattern, eval_expr):
        """
        Initialize evaluation-based replacement.
        
        Parameters:
        - pattern: str, pattern to match
        - eval_expr: str, expression to evaluate for replacement
        """

class RepFunc:
    """
    Function-based replacement class.
    
    Uses custom functions for complex replacement logic
    with full access to match context.
    """
    
    def __init__(self, pattern, func):
        """
        Initialize function-based replacement.
        
        Parameters:
        - pattern: str or regex, pattern to match
        - func: callable, replacement function
        """

def deep_substitute(obj, substitutions, **kwargs):
    """
    Perform deep template substitution on nested objects.
    
    Parameters:
    - obj: object, nested structure to process
    - substitutions: dict, substitution mappings
    - max_depth: int, maximum recursion depth
    
    Returns:
    object: Object with substitutions applied
    """

class AsyncJob:
    """
    Asynchronous job management.
    
    Provides framework for long-running computations with progress tracking,
    cancellation support, and result retrieval.
    """
    
    def __init__(self, func, *args, **kwargs):
        """
        Initialize async job.
        
        Parameters:
        - func: callable, function to execute asynchronously
        - args: positional arguments for function
        - kwargs: keyword arguments for function
        """
    
    def start(self):
        """Start job execution."""
    
    def cancel(self):
        """Cancel job execution."""
    
    def is_running(self):
        """Check if job is running."""
    
    def get_result(self, timeout=None):
        """Get job result with optional timeout."""
    
    def get_progress(self):
        """Get job progress information."""

class AsyncScheduler:
    """
    Job scheduling system.
    
    Manages multiple asynchronous jobs with priority queues,
    resource allocation, and dependency management.
    """
    
    def __init__(self, max_workers=None):
        """
        Initialize scheduler.
        
        Parameters:
        - max_workers: int, maximum concurrent workers
        """
    
    def submit(self, func, *args, priority=0, **kwargs):
        """
        Submit job for execution.
        
        Parameters:
        - func: callable, function to execute
        - args: function arguments
        - priority: int, job priority (higher = more urgent)
        - kwargs: function keyword arguments
        
        Returns:
        AsyncJob: Job handle
        """
    
    def shutdown(self, wait=True):
        """Shutdown scheduler."""

class ScheduleManager:
    """
    Schedule management utilities.
    
    Provides cron-like scheduling capabilities for recurring tasks
    with timezone support and failure handling.
    """
    
    def __init__(self):
        """Initialize schedule manager."""
    
    def schedule(self, func, schedule_str, **kwargs):
        """
        Schedule recurring task.
        
        Parameters:
        - func: callable, function to schedule
        - schedule_str: str, schedule specification (cron-like)
        - timezone: str, timezone for scheduling
        - max_retries: int, maximum retry attempts
        
        Returns:
        str: Schedule ID
        """
    
    def cancel(self, schedule_id):
        """Cancel scheduled task."""
    
    def list_schedules(self):
        """List all active schedules."""

class Figure:
    """
    Enhanced Plotly figure with VectorBT customizations.
    
    Provides additional functionality for financial plotting including
    candlestick charts, volume profiles, and indicator overlays.
    """
    
    def __init__(self, *args, **kwargs):
        """Initialize enhanced figure."""
    
    def add_ohlc(self, ohlc_data, **kwargs):
        """Add OHLC candlestick chart."""
    
    def add_volume(self, volume_data, **kwargs):
        """Add volume bars."""
    
    def add_indicator(self, indicator_data, **kwargs):
        """Add technical indicator."""
    
    def show(self, **kwargs):
        """Display figure."""

class FigureWidget:
    """
    Enhanced Plotly figure widget for Jupyter notebooks.
    
    Interactive plotting widget with real-time updates and
    callback support for dynamic visualizations.
    """
    
    def __init__(self, *args, **kwargs):
        """Initialize figure widget."""
    
    def update(self, **kwargs):
        """Update widget data."""

def make_figure(**kwargs):
    """
    Create enhanced figure with VectorBT defaults.
    
    Parameters:
    - kwargs: figure configuration parameters
    
    Returns:
    Figure: Enhanced Plotly figure
    """

def make_subplots(rows=1, cols=1, **kwargs):
    """
    Create subplot layout with VectorBT styling.
    
    Parameters:
    - rows: int, number of subplot rows
    - cols: int, number of subplot columns
    - kwargs: subplot configuration
    
    Returns:
    Figure: Figure with subplot layout
    """

def save_animation(fig, filename, **kwargs):
    """
    Save figure animation to file.
    
    Parameters:
    - fig: Figure, figure to animate
    - filename: str, output filename
    - fps: int, frames per second
    - duration: float, animation duration
    
    Returns:
    str: Path to saved animation
    """

def set_seed(seed):
    """
    Set random seed for reproducible results.
    
    Parameters:
    - seed: int, random seed value
    """

class CancelledError(Exception):
    """
    Exception raised when async operation is cancelled.
    
    Provides information about cancellation reason and timing.
    """
    
    def __init__(self, message="Operation was cancelled"):
        """Initialize cancellation error."""

import vectorbt as vbt

# Global settings access
print(vbt.settings['plotting']['width'])
print(vbt.settings['caching']['enabled'])

# Update global settings  
vbt.settings.update({
    'plotting': {
        'width': 1000,
        'height': 600,
        'theme': 'plotly_dark'
    },
    'caching': {
        'enabled': True,
        'max_size': '1GB'
    }
})

# Create custom configuration
custom_config = vbt.Config({
    'indicators': {
        'rsi_window': 14,  
        'ma_windows': [20, 50, 200]
    },
    'portfolio': {
        'initial_cash': 100000,
        'fees': 0.001
    }
})

# Use configuration in components
class CustomStrategy(vbt.Configured):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.rsi_window = self.config['rsi_window']
        self.ma_windows = self.config['ma_windows']

strategy = CustomStrategy(config=custom_config['indicators'])

# Method caching
class DataProcessor:
    @vbt.cached_method
    def expensive_calculation(self, data, window):
        # Expensive computation that benefits from caching
        return data.rolling(window).apply(lambda x: x.sum() ** 2)

processor = DataProcessor()

# First call - computed and cached
result1 = processor.expensive_calculation(price_data, 20)

# Second call with same parameters - retrieved from cache  
result2 = processor.expensive_calculation(price_data, 20)

# Property caching
class Portfolio:
    @vbt.cached_property
    def total_return(self):
        # Expensive calculation cached until data changes
        return self.value.iloc[-1] / self.value.iloc[0] - 1

portfolio = Portfolio()
return1 = portfolio.total_return  # Computed
return2 = portfolio.total_return  # Retrieved from cache

# Basic template substitution
template = vbt.Sub(
    "Strategy: {strategy_name}, Window: {window}, Threshold: {threshold}"
)

strategy_desc = template.substitute(
    strategy_name="RSI Oversold",
    window=14,
    threshold=30
)

# Advanced replacement system
code_template = """
def {func_name}(data, window={window}):
    {indicator} = vbt.{indicator_class}.run(data, {window})
    return {indicator}.{output_attr}
"""

replacements = {
    'func_name': 'calculate_rsi',
    'window': 14,
    'indicator': 'rsi_result',  
    'indicator_class': 'RSI',
    'output_attr': 'rsi'
}

generated_code = vbt.deep_substitute(code_template, replacements)

# Async job execution
def long_backtest(symbols, strategy_params):
    results = {}
    for symbol in symbols:
        # Long-running backtest
        data = vbt.YFData.download(symbol, period="5y")
        portfolio = run_strategy(data, strategy_params)
        results[symbol] = portfolio.total_return()
    return results

# Submit async job
job = vbt.AsyncJob(
    long_backtest,
    symbols=["AAPL", "GOOGL", "MSFT", "TSLA"],
    strategy_params={'rsi_window': 14, 'threshold': 30}
)

job.start()

# Check progress
while job.is_running():
    progress = job.get_progress()
    print(f"Progress: {progress}%")
    time.sleep(1)

# Get results
results = job.get_result()

# Job scheduling
scheduler = vbt.AsyncScheduler(max_workers=4)

# Submit multiple jobs
jobs = []
for symbol in ["AAPL", "GOOGL", "MSFT"]:
    job = scheduler.submit(
        run_backtest,
        symbol=symbol,
        priority=1
    )
    jobs.append(job)

# Wait for all jobs
results = [job.get_result() for job in jobs]
scheduler.shutdown()

# Create enhanced figure
fig = vbt.make_figure(
    width=1200,
    height=800,
    title="Portfolio Analysis"
)

# Add OHLC data
fig.add_ohlc(
    ohlc_data=data.get(['Open', 'High', 'Low', 'Close']),
    name="Price"
)

# Add volume
fig.add_volume(
    volume_data=data.get('Volume'),
    secondary_y=True
)

# Add indicators
fig.add_indicator(
    indicator_data=rsi_values,
    name="RSI",
    secondary_y=True
)

fig.show()

# Subplot layout
fig = vbt.make_subplots(
    rows=3, cols=1,
    shared_xaxes=True,
    subplot_titles=['Price', 'Volume', 'RSI']
)

# Animation
animated_fig = create_price_animation(price_data)
vbt.save_animation(
    animated_fig,
    "price_animation.gif",
    fps=10,
    duration=5
)

# Thread-safe dictionary
atomic_cache = vbt.atomic_dict()

# Safe concurrent access
def worker_function(data, cache_key):
    if cache_key not in atomic_cache:
        result = expensive_computation(data)
        atomic_cache[cache_key] = result
    return atomic_cache[cache_key]

# Multiple threads can safely access the cache
import threading

threads = []
for i in range(10):
    t = threading.Thread(
        target=worker_function,
        args=(data, f"key_{i}")
    )
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Merge configurations
base_config = {
    'portfolio': {'initial_cash': 10000},
    'indicators': {'rsi_window': 14}
}

user_config = {
    'portfolio': {'fees': 0.001},
    'plotting': {'width': 1000}
}

merged_config = vbt.merge_dicts(
    base_config,
    user_config,
    strategy='deep_merge'
)
# Result: Complete merged configuration with all keys