This is page 8 of 29. Use http://codebase.md/wshobson/maverick-mcp?lines=false&page={x} to view the full context.
# Directory Structure
```
├── .dockerignore
├── .env.example
├── .github
│ ├── dependabot.yml
│ ├── FUNDING.yml
│ ├── ISSUE_TEMPLATE
│ │ ├── bug_report.md
│ │ ├── config.yml
│ │ ├── feature_request.md
│ │ ├── question.md
│ │ └── security_report.md
│ ├── pull_request_template.md
│ └── workflows
│ ├── claude-code-review.yml
│ └── claude.yml
├── .gitignore
├── .python-version
├── .vscode
│ ├── launch.json
│ └── settings.json
├── alembic
│ ├── env.py
│ ├── script.py.mako
│ └── versions
│ ├── 001_initial_schema.py
│ ├── 003_add_performance_indexes.py
│ ├── 006_rename_metadata_columns.py
│ ├── 008_performance_optimization_indexes.py
│ ├── 009_rename_to_supply_demand.py
│ ├── 010_self_contained_schema.py
│ ├── 011_remove_proprietary_terms.py
│ ├── 013_add_backtest_persistence_models.py
│ ├── 014_add_portfolio_models.py
│ ├── 08e3945a0c93_merge_heads.py
│ ├── 9374a5c9b679_merge_heads_for_testing.py
│ ├── abf9b9afb134_merge_multiple_heads.py
│ ├── adda6d3fd84b_merge_proprietary_terms_removal_with_.py
│ ├── e0c75b0bdadb_fix_financial_data_precision_only.py
│ ├── f0696e2cac15_add_essential_performance_indexes.py
│ └── fix_database_integrity_issues.py
├── alembic.ini
├── CLAUDE.md
├── CODE_OF_CONDUCT.md
├── CONTRIBUTING.md
├── DATABASE_SETUP.md
├── docker-compose.override.yml.example
├── docker-compose.yml
├── Dockerfile
├── docs
│ ├── api
│ │ └── backtesting.md
│ ├── BACKTESTING.md
│ ├── COST_BASIS_SPECIFICATION.md
│ ├── deep_research_agent.md
│ ├── exa_research_testing_strategy.md
│ ├── PORTFOLIO_PERSONALIZATION_PLAN.md
│ ├── PORTFOLIO.md
│ ├── SETUP_SELF_CONTAINED.md
│ └── speed_testing_framework.md
├── examples
│ ├── complete_speed_validation.py
│ ├── deep_research_integration.py
│ ├── llm_optimization_example.py
│ ├── llm_speed_demo.py
│ ├── monitoring_example.py
│ ├── parallel_research_example.py
│ ├── speed_optimization_demo.py
│ └── timeout_fix_demonstration.py
├── LICENSE
├── Makefile
├── MANIFEST.in
├── maverick_mcp
│ ├── __init__.py
│ ├── agents
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── circuit_breaker.py
│ │ ├── deep_research.py
│ │ ├── market_analysis.py
│ │ ├── optimized_research.py
│ │ ├── supervisor.py
│ │ └── technical_analysis.py
│ ├── api
│ │ ├── __init__.py
│ │ ├── api_server.py
│ │ ├── connection_manager.py
│ │ ├── dependencies
│ │ │ ├── __init__.py
│ │ │ ├── stock_analysis.py
│ │ │ └── technical_analysis.py
│ │ ├── error_handling.py
│ │ ├── inspector_compatible_sse.py
│ │ ├── inspector_sse.py
│ │ ├── middleware
│ │ │ ├── error_handling.py
│ │ │ ├── mcp_logging.py
│ │ │ ├── rate_limiting_enhanced.py
│ │ │ └── security.py
│ │ ├── openapi_config.py
│ │ ├── routers
│ │ │ ├── __init__.py
│ │ │ ├── agents.py
│ │ │ ├── backtesting.py
│ │ │ ├── data_enhanced.py
│ │ │ ├── data.py
│ │ │ ├── health_enhanced.py
│ │ │ ├── health_tools.py
│ │ │ ├── health.py
│ │ │ ├── intelligent_backtesting.py
│ │ │ ├── introspection.py
│ │ │ ├── mcp_prompts.py
│ │ │ ├── monitoring.py
│ │ │ ├── news_sentiment_enhanced.py
│ │ │ ├── performance.py
│ │ │ ├── portfolio.py
│ │ │ ├── research.py
│ │ │ ├── screening_ddd.py
│ │ │ ├── screening_parallel.py
│ │ │ ├── screening.py
│ │ │ ├── technical_ddd.py
│ │ │ ├── technical_enhanced.py
│ │ │ ├── technical.py
│ │ │ └── tool_registry.py
│ │ ├── server.py
│ │ ├── services
│ │ │ ├── __init__.py
│ │ │ ├── base_service.py
│ │ │ ├── market_service.py
│ │ │ ├── portfolio_service.py
│ │ │ ├── prompt_service.py
│ │ │ └── resource_service.py
│ │ ├── simple_sse.py
│ │ └── utils
│ │ ├── __init__.py
│ │ ├── insomnia_export.py
│ │ └── postman_export.py
│ ├── application
│ │ ├── __init__.py
│ │ ├── commands
│ │ │ └── __init__.py
│ │ ├── dto
│ │ │ ├── __init__.py
│ │ │ └── technical_analysis_dto.py
│ │ ├── queries
│ │ │ ├── __init__.py
│ │ │ └── get_technical_analysis.py
│ │ └── screening
│ │ ├── __init__.py
│ │ ├── dtos.py
│ │ └── queries.py
│ ├── backtesting
│ │ ├── __init__.py
│ │ ├── ab_testing.py
│ │ ├── analysis.py
│ │ ├── batch_processing_stub.py
│ │ ├── batch_processing.py
│ │ ├── model_manager.py
│ │ ├── optimization.py
│ │ ├── persistence.py
│ │ ├── retraining_pipeline.py
│ │ ├── strategies
│ │ │ ├── __init__.py
│ │ │ ├── base.py
│ │ │ ├── ml
│ │ │ │ ├── __init__.py
│ │ │ │ ├── adaptive.py
│ │ │ │ ├── ensemble.py
│ │ │ │ ├── feature_engineering.py
│ │ │ │ └── regime_aware.py
│ │ │ ├── ml_strategies.py
│ │ │ ├── parser.py
│ │ │ └── templates.py
│ │ ├── strategy_executor.py
│ │ ├── vectorbt_engine.py
│ │ └── visualization.py
│ ├── config
│ │ ├── __init__.py
│ │ ├── constants.py
│ │ ├── database_self_contained.py
│ │ ├── database.py
│ │ ├── llm_optimization_config.py
│ │ ├── logging_settings.py
│ │ ├── plotly_config.py
│ │ ├── security_utils.py
│ │ ├── security.py
│ │ ├── settings.py
│ │ ├── technical_constants.py
│ │ ├── tool_estimation.py
│ │ └── validation.py
│ ├── core
│ │ ├── __init__.py
│ │ ├── technical_analysis.py
│ │ └── visualization.py
│ ├── data
│ │ ├── __init__.py
│ │ ├── cache_manager.py
│ │ ├── cache.py
│ │ ├── django_adapter.py
│ │ ├── health.py
│ │ ├── models.py
│ │ ├── performance.py
│ │ ├── session_management.py
│ │ └── validation.py
│ ├── database
│ │ ├── __init__.py
│ │ ├── base.py
│ │ └── optimization.py
│ ├── dependencies.py
│ ├── domain
│ │ ├── __init__.py
│ │ ├── entities
│ │ │ ├── __init__.py
│ │ │ └── stock_analysis.py
│ │ ├── events
│ │ │ └── __init__.py
│ │ ├── portfolio.py
│ │ ├── screening
│ │ │ ├── __init__.py
│ │ │ ├── entities.py
│ │ │ ├── services.py
│ │ │ └── value_objects.py
│ │ ├── services
│ │ │ ├── __init__.py
│ │ │ └── technical_analysis_service.py
│ │ ├── stock_analysis
│ │ │ ├── __init__.py
│ │ │ └── stock_analysis_service.py
│ │ └── value_objects
│ │ ├── __init__.py
│ │ └── technical_indicators.py
│ ├── exceptions.py
│ ├── infrastructure
│ │ ├── __init__.py
│ │ ├── cache
│ │ │ └── __init__.py
│ │ ├── caching
│ │ │ ├── __init__.py
│ │ │ └── cache_management_service.py
│ │ ├── connection_manager.py
│ │ ├── data_fetching
│ │ │ ├── __init__.py
│ │ │ └── stock_data_service.py
│ │ ├── health
│ │ │ ├── __init__.py
│ │ │ └── health_checker.py
│ │ ├── persistence
│ │ │ ├── __init__.py
│ │ │ └── stock_repository.py
│ │ ├── providers
│ │ │ └── __init__.py
│ │ ├── screening
│ │ │ ├── __init__.py
│ │ │ └── repositories.py
│ │ └── sse_optimizer.py
│ ├── langchain_tools
│ │ ├── __init__.py
│ │ ├── adapters.py
│ │ └── registry.py
│ ├── logging_config.py
│ ├── memory
│ │ ├── __init__.py
│ │ └── stores.py
│ ├── monitoring
│ │ ├── __init__.py
│ │ ├── health_check.py
│ │ ├── health_monitor.py
│ │ ├── integration_example.py
│ │ ├── metrics.py
│ │ ├── middleware.py
│ │ └── status_dashboard.py
│ ├── providers
│ │ ├── __init__.py
│ │ ├── dependencies.py
│ │ ├── factories
│ │ │ ├── __init__.py
│ │ │ ├── config_factory.py
│ │ │ └── provider_factory.py
│ │ ├── implementations
│ │ │ ├── __init__.py
│ │ │ ├── cache_adapter.py
│ │ │ ├── macro_data_adapter.py
│ │ │ ├── market_data_adapter.py
│ │ │ ├── persistence_adapter.py
│ │ │ └── stock_data_adapter.py
│ │ ├── interfaces
│ │ │ ├── __init__.py
│ │ │ ├── cache.py
│ │ │ ├── config.py
│ │ │ ├── macro_data.py
│ │ │ ├── market_data.py
│ │ │ ├── persistence.py
│ │ │ └── stock_data.py
│ │ ├── llm_factory.py
│ │ ├── macro_data.py
│ │ ├── market_data.py
│ │ ├── mocks
│ │ │ ├── __init__.py
│ │ │ ├── mock_cache.py
│ │ │ ├── mock_config.py
│ │ │ ├── mock_macro_data.py
│ │ │ ├── mock_market_data.py
│ │ │ ├── mock_persistence.py
│ │ │ └── mock_stock_data.py
│ │ ├── openrouter_provider.py
│ │ ├── optimized_screening.py
│ │ ├── optimized_stock_data.py
│ │ └── stock_data.py
│ ├── README.md
│ ├── tests
│ │ ├── __init__.py
│ │ ├── README_INMEMORY_TESTS.md
│ │ ├── test_cache_debug.py
│ │ ├── test_fixes_validation.py
│ │ ├── test_in_memory_routers.py
│ │ ├── test_in_memory_server.py
│ │ ├── test_macro_data_provider.py
│ │ ├── test_mailgun_email.py
│ │ ├── test_market_calendar_caching.py
│ │ ├── test_mcp_tool_fixes_pytest.py
│ │ ├── test_mcp_tool_fixes.py
│ │ ├── test_mcp_tools.py
│ │ ├── test_models_functional.py
│ │ ├── test_server.py
│ │ ├── test_stock_data_enhanced.py
│ │ ├── test_stock_data_provider.py
│ │ └── test_technical_analysis.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── performance_monitoring.py
│ │ ├── portfolio_manager.py
│ │ ├── risk_management.py
│ │ └── sentiment_analysis.py
│ ├── utils
│ │ ├── __init__.py
│ │ ├── agent_errors.py
│ │ ├── batch_processing.py
│ │ ├── cache_warmer.py
│ │ ├── circuit_breaker_decorators.py
│ │ ├── circuit_breaker_services.py
│ │ ├── circuit_breaker.py
│ │ ├── data_chunking.py
│ │ ├── database_monitoring.py
│ │ ├── debug_utils.py
│ │ ├── fallback_strategies.py
│ │ ├── llm_optimization.py
│ │ ├── logging_example.py
│ │ ├── logging_init.py
│ │ ├── logging.py
│ │ ├── mcp_logging.py
│ │ ├── memory_profiler.py
│ │ ├── monitoring_middleware.py
│ │ ├── monitoring.py
│ │ ├── orchestration_logging.py
│ │ ├── parallel_research.py
│ │ ├── parallel_screening.py
│ │ ├── quick_cache.py
│ │ ├── resource_manager.py
│ │ ├── shutdown.py
│ │ ├── stock_helpers.py
│ │ ├── structured_logger.py
│ │ ├── tool_monitoring.py
│ │ ├── tracing.py
│ │ └── yfinance_pool.py
│ ├── validation
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── data.py
│ │ ├── middleware.py
│ │ ├── portfolio.py
│ │ ├── responses.py
│ │ ├── screening.py
│ │ └── technical.py
│ └── workflows
│ ├── __init__.py
│ ├── agents
│ │ ├── __init__.py
│ │ ├── market_analyzer.py
│ │ ├── optimizer_agent.py
│ │ ├── strategy_selector.py
│ │ └── validator_agent.py
│ ├── backtesting_workflow.py
│ └── state.py
├── PLANS.md
├── pyproject.toml
├── pyrightconfig.json
├── README.md
├── scripts
│ ├── dev.sh
│ ├── INSTALLATION_GUIDE.md
│ ├── load_example.py
│ ├── load_market_data.py
│ ├── load_tiingo_data.py
│ ├── migrate_db.py
│ ├── README_TIINGO_LOADER.md
│ ├── requirements_tiingo.txt
│ ├── run_stock_screening.py
│ ├── run-migrations.sh
│ ├── seed_db.py
│ ├── seed_sp500.py
│ ├── setup_database.sh
│ ├── setup_self_contained.py
│ ├── setup_sp500_database.sh
│ ├── test_seeded_data.py
│ ├── test_tiingo_loader.py
│ ├── tiingo_config.py
│ └── validate_setup.py
├── SECURITY.md
├── server.json
├── setup.py
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── core
│ │ └── test_technical_analysis.py
│ ├── data
│ │ └── test_portfolio_models.py
│ ├── domain
│ │ ├── conftest.py
│ │ ├── test_portfolio_entities.py
│ │ └── test_technical_analysis_service.py
│ ├── fixtures
│ │ └── orchestration_fixtures.py
│ ├── integration
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── README.md
│ │ ├── run_integration_tests.sh
│ │ ├── test_api_technical.py
│ │ ├── test_chaos_engineering.py
│ │ ├── test_config_management.py
│ │ ├── test_full_backtest_workflow_advanced.py
│ │ ├── test_full_backtest_workflow.py
│ │ ├── test_high_volume.py
│ │ ├── test_mcp_tools.py
│ │ ├── test_orchestration_complete.py
│ │ ├── test_portfolio_persistence.py
│ │ ├── test_redis_cache.py
│ │ ├── test_security_integration.py.disabled
│ │ └── vcr_setup.py
│ ├── performance
│ │ ├── __init__.py
│ │ ├── test_benchmarks.py
│ │ ├── test_load.py
│ │ ├── test_profiling.py
│ │ └── test_stress.py
│ ├── providers
│ │ └── test_stock_data_simple.py
│ ├── README.md
│ ├── test_agents_router_mcp.py
│ ├── test_backtest_persistence.py
│ ├── test_cache_management_service.py
│ ├── test_cache_serialization.py
│ ├── test_circuit_breaker.py
│ ├── test_database_pool_config_simple.py
│ ├── test_database_pool_config.py
│ ├── test_deep_research_functional.py
│ ├── test_deep_research_integration.py
│ ├── test_deep_research_parallel_execution.py
│ ├── test_error_handling.py
│ ├── test_event_loop_integrity.py
│ ├── test_exa_research_integration.py
│ ├── test_exception_hierarchy.py
│ ├── test_financial_search.py
│ ├── test_graceful_shutdown.py
│ ├── test_integration_simple.py
│ ├── test_langgraph_workflow.py
│ ├── test_market_data_async.py
│ ├── test_market_data_simple.py
│ ├── test_mcp_orchestration_functional.py
│ ├── test_ml_strategies.py
│ ├── test_optimized_research_agent.py
│ ├── test_orchestration_integration.py
│ ├── test_orchestration_logging.py
│ ├── test_orchestration_tools_simple.py
│ ├── test_parallel_research_integration.py
│ ├── test_parallel_research_orchestrator.py
│ ├── test_parallel_research_performance.py
│ ├── test_performance_optimizations.py
│ ├── test_production_validation.py
│ ├── test_provider_architecture.py
│ ├── test_rate_limiting_enhanced.py
│ ├── test_runner_validation.py
│ ├── test_security_comprehensive.py.disabled
│ ├── test_security_cors.py
│ ├── test_security_enhancements.py.disabled
│ ├── test_security_headers.py
│ ├── test_security_penetration.py
│ ├── test_session_management.py
│ ├── test_speed_optimization_validation.py
│ ├── test_stock_analysis_dependencies.py
│ ├── test_stock_analysis_service.py
│ ├── test_stock_data_fetching_service.py
│ ├── test_supervisor_agent.py
│ ├── test_supervisor_functional.py
│ ├── test_tool_estimation_config.py
│ ├── test_visualization.py
│ └── utils
│ ├── test_agent_errors.py
│ ├── test_logging.py
│ ├── test_parallel_screening.py
│ └── test_quick_cache.py
├── tools
│ ├── check_orchestration_config.py
│ ├── experiments
│ │ ├── validation_examples.py
│ │ └── validation_fixed.py
│ ├── fast_dev.sh
│ ├── hot_reload.py
│ ├── quick_test.py
│ └── templates
│ ├── new_router_template.py
│ ├── new_tool_template.py
│ ├── screening_strategy_template.py
│ └── test_template.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/maverick_mcp/utils/batch_processing.py:
--------------------------------------------------------------------------------
```python
"""
Batch processing utilities for efficient multi-symbol operations.
This module provides utilities for processing multiple stock symbols
efficiently using concurrent execution and batching strategies.
"""
import asyncio
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, TypeVar
import pandas as pd
import yfinance as yf
from maverick_mcp.providers.stock_data import EnhancedStockDataProvider
from maverick_mcp.utils.logging import get_logger
logger = get_logger(__name__)
T = TypeVar("T")
class BatchProcessor:
"""
Utility class for efficient batch processing of stock operations.
Provides methods for processing multiple symbols concurrently
with proper error handling and resource management.
"""
def __init__(self, max_workers: int = 10, batch_size: int = 50):
"""
Initialize batch processor.
Args:
max_workers: Maximum number of concurrent workers
batch_size: Maximum number of symbols per batch
"""
self.max_workers = max_workers
self.batch_size = batch_size
self.executor = ThreadPoolExecutor(max_workers=max_workers)
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit with cleanup."""
self.executor.shutdown(wait=True)
def process_symbols_concurrent(
self,
symbols: list[str],
processor_func: Callable[[str], T],
error_default: T | None = None,
) -> dict[str, T]:
"""
Process multiple symbols concurrently using ThreadPoolExecutor.
Args:
symbols: List of stock symbols to process
processor_func: Function to apply to each symbol
error_default: Default value to return on error
Returns:
Dictionary mapping symbols to their processed results
"""
results = {}
# Submit all tasks
future_to_symbol = {
self.executor.submit(processor_func, symbol): symbol for symbol in symbols
}
# Collect results as they complete
for future in as_completed(future_to_symbol):
symbol = future_to_symbol[future]
try:
result = future.result()
results[symbol] = result
except Exception as e:
logger.warning(f"Error processing {symbol}: {e}")
if error_default is not None:
results[symbol] = error_default
return results
async def process_symbols_async(
self,
symbols: list[str],
async_processor_func: Callable[[str], Any],
max_concurrent: int | None = None,
) -> dict[str, Any]:
"""
Process multiple symbols asynchronously with concurrency limit.
Args:
symbols: List of stock symbols to process
async_processor_func: Async function to apply to each symbol
max_concurrent: Maximum concurrent operations (defaults to max_workers)
Returns:
Dictionary mapping symbols to their processed results
"""
if max_concurrent is None:
max_concurrent = self.max_workers
semaphore = asyncio.Semaphore(max_concurrent)
async def process_with_semaphore(symbol: str):
async with semaphore:
try:
return symbol, await async_processor_func(symbol)
except Exception as e:
logger.warning(f"Error processing {symbol}: {e}")
return symbol, None
# Process all symbols concurrently
tasks = [process_with_semaphore(symbol) for symbol in symbols]
results = await asyncio.gather(*tasks)
# Convert to dictionary, filtering out None results
return {symbol: result for symbol, result in results if result is not None}
def process_in_batches(
self,
symbols: list[str],
batch_processor_func: Callable[[list[str]], dict[str, T]],
) -> dict[str, T]:
"""
Process symbols in batches for improved efficiency.
Args:
symbols: List of stock symbols to process
batch_processor_func: Function that processes a batch of symbols
Returns:
Dictionary mapping symbols to their processed results
"""
results = {}
# Process symbols in batches
for i in range(0, len(symbols), self.batch_size):
batch = symbols[i : i + self.batch_size]
try:
batch_results = batch_processor_func(batch)
results.update(batch_results)
except Exception as e:
logger.error(f"Error processing batch {i // self.batch_size + 1}: {e}")
# Process individual symbols as fallback
for symbol in batch:
try:
individual_result = batch_processor_func([symbol])
results.update(individual_result)
except Exception as e2:
logger.warning(
f"Error processing individual symbol {symbol}: {e2}"
)
return results
class StockDataBatchProcessor:
"""Specialized batch processor for stock data operations."""
def __init__(self, provider: EnhancedStockDataProvider | None = None):
"""Initialize with optional stock data provider."""
self.provider = provider or EnhancedStockDataProvider()
def get_batch_stock_data(
self, symbols: list[str], start_date: str, end_date: str
) -> dict[str, pd.DataFrame]:
"""
Fetch stock data for multiple symbols efficiently using yfinance batch download.
Args:
symbols: List of stock symbols
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
Returns:
Dictionary mapping symbols to their DataFrames
"""
try:
# Use yfinance batch download for efficiency
tickers_str = " ".join(symbols)
data = yf.download(
tickers_str,
start=start_date,
end=end_date,
group_by="ticker",
threads=True,
progress=False,
)
results = {}
if len(symbols) == 1:
# Single symbol case
symbol = symbols[0]
if not data.empty:
# Standardize column names
df = data.copy()
if "Close" in df.columns:
df.columns = df.columns.str.title()
results[symbol] = df
else:
# Multiple symbols case
for symbol in symbols:
try:
if symbol in data.columns.get_level_values(0):
symbol_data = data[symbol].copy()
# Remove any NaN-only rows
symbol_data = symbol_data.dropna(how="all")
if not symbol_data.empty:
results[symbol] = symbol_data
except Exception as e:
logger.warning(f"Error extracting data for {symbol}: {e}")
return results
except Exception as e:
logger.error(f"Error in batch stock data download: {e}")
# Fallback to individual downloads
return self._fallback_individual_downloads(symbols, start_date, end_date)
def _fallback_individual_downloads(
self, symbols: list[str], start_date: str, end_date: str
) -> dict[str, pd.DataFrame]:
"""Fallback to individual downloads if batch fails."""
results = {}
with BatchProcessor(max_workers=5) as processor:
def download_single(symbol: str) -> pd.DataFrame:
try:
return self.provider.get_stock_data(symbol, start_date, end_date)
except Exception as e:
logger.warning(f"Error downloading {symbol}: {e}")
return pd.DataFrame()
symbol_results = processor.process_symbols_concurrent(
symbols, download_single, pd.DataFrame()
)
# Filter out empty DataFrames
results = {
symbol: df for symbol, df in symbol_results.items() if not df.empty
}
return results
async def get_batch_stock_data_async(
self, symbols: list[str], start_date: str, end_date: str
) -> dict[str, pd.DataFrame]:
"""
Async version of batch stock data fetching.
Args:
symbols: List of stock symbols
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
Returns:
Dictionary mapping symbols to their DataFrames
"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self.get_batch_stock_data, symbols, start_date, end_date
)
def get_batch_stock_info(self, symbols: list[str]) -> dict[str, dict[str, Any]]:
"""
Get stock info for multiple symbols efficiently.
Args:
symbols: List of stock symbols
Returns:
Dictionary mapping symbols to their info dictionaries
"""
with BatchProcessor(max_workers=10) as processor:
def get_info(symbol: str) -> dict[str, Any]:
try:
ticker = yf.Ticker(symbol)
return ticker.info
except Exception as e:
logger.warning(f"Error getting info for {symbol}: {e}")
return {}
return processor.process_symbols_concurrent(symbols, get_info, {})
def get_batch_technical_analysis(
self, symbols: list[str], days: int = 365
) -> dict[str, pd.DataFrame]:
"""
Get technical analysis for multiple symbols efficiently.
Args:
symbols: List of stock symbols
days: Number of days of data
Returns:
Dictionary mapping symbols to DataFrames with technical indicators
"""
from maverick_mcp.utils.stock_helpers import get_stock_dataframe
with BatchProcessor(max_workers=8) as processor:
def get_analysis(symbol: str) -> pd.DataFrame:
try:
return get_stock_dataframe(symbol, days)
except Exception as e:
logger.warning(
f"Error getting technical analysis for {symbol}: {e}"
)
return pd.DataFrame()
results = processor.process_symbols_concurrent(
symbols, get_analysis, pd.DataFrame()
)
# Filter out empty DataFrames
return {symbol: df for symbol, df in results.items() if not df.empty}
# Convenience functions for common batch operations
def batch_download_stock_data(
symbols: list[str], start_date: str, end_date: str
) -> dict[str, pd.DataFrame]:
"""
Convenience function for batch downloading stock data.
Args:
symbols: List of stock symbols
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
Returns:
Dictionary mapping symbols to their DataFrames
"""
processor = StockDataBatchProcessor()
return processor.get_batch_stock_data(symbols, start_date, end_date)
async def batch_download_stock_data_async(
symbols: list[str], start_date: str, end_date: str
) -> dict[str, pd.DataFrame]:
"""
Convenience function for async batch downloading stock data.
Args:
symbols: List of stock symbols
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
Returns:
Dictionary mapping symbols to their DataFrames
"""
processor = StockDataBatchProcessor()
return await processor.get_batch_stock_data_async(symbols, start_date, end_date)
def batch_get_stock_info(symbols: list[str]) -> dict[str, dict[str, Any]]:
"""
Convenience function for batch getting stock info.
Args:
symbols: List of stock symbols
Returns:
Dictionary mapping symbols to their info dictionaries
"""
processor = StockDataBatchProcessor()
return processor.get_batch_stock_info(symbols)
def batch_get_technical_analysis(
symbols: list[str], days: int = 365
) -> dict[str, pd.DataFrame]:
"""
Convenience function for batch technical analysis.
Args:
symbols: List of stock symbols
days: Number of days of data
Returns:
Dictionary mapping symbols to DataFrames with technical indicators
"""
processor = StockDataBatchProcessor()
return processor.get_batch_technical_analysis(symbols, days)
```
--------------------------------------------------------------------------------
/tests/test_provider_architecture.py:
--------------------------------------------------------------------------------
```python
"""
Test cases for the new provider architecture.
This module demonstrates how to use the new interface-based architecture
for testing and validates that the abstractions work correctly.
"""
import pandas as pd
import pytest
from maverick_mcp.providers.dependencies import (
DependencyOverride,
create_test_dependencies,
get_dependencies_for_testing,
)
from maverick_mcp.providers.factories.config_factory import ConfigurationFactory
from maverick_mcp.providers.factories.provider_factory import ProviderFactory
from maverick_mcp.providers.mocks.mock_cache import MockCacheManager
from maverick_mcp.providers.mocks.mock_config import MockConfigurationProvider
from maverick_mcp.providers.mocks.mock_macro_data import MockMacroDataProvider
from maverick_mcp.providers.mocks.mock_market_data import MockMarketDataProvider
from maverick_mcp.providers.mocks.mock_stock_data import (
MockStockDataFetcher,
MockStockScreener,
)
class TestProviderInterfaces:
"""Test the provider interfaces work correctly."""
@pytest.mark.asyncio
async def test_mock_cache_manager(self):
"""Test the mock cache manager implementation."""
cache = MockCacheManager()
# Test basic operations
assert await cache.get("nonexistent") is None
assert await cache.set("test_key", "test_value", 60) is True
assert await cache.get("test_key") == "test_value"
assert await cache.exists("test_key") is True
assert await cache.delete("test_key") is True
assert await cache.exists("test_key") is False
# Test batch operations
items = [("key1", "value1", 60), ("key2", "value2", 60)]
assert await cache.set_many(items) == 2
results = await cache.get_many(["key1", "key2", "key3"])
assert results == {"key1": "value1", "key2": "value2"}
# Test call logging
call_log = cache.get_call_log()
assert len(call_log) > 0
assert call_log[0]["method"] == "get"
@pytest.mark.asyncio
async def test_mock_stock_data_fetcher(self):
"""Test the mock stock data fetcher implementation."""
fetcher = MockStockDataFetcher()
# Test stock data retrieval
data = await fetcher.get_stock_data("AAPL", "2024-01-01", "2024-01-31")
assert isinstance(data, pd.DataFrame)
assert not data.empty
assert "Close" in data.columns
# Test real-time data
realtime = await fetcher.get_realtime_data("AAPL")
assert realtime is not None
assert "symbol" in realtime
assert realtime["symbol"] == "AAPL"
# Test stock info
info = await fetcher.get_stock_info("AAPL")
assert "symbol" in info
assert info["symbol"] == "AAPL"
# Test market status
is_open = await fetcher.is_market_open()
assert isinstance(is_open, bool)
# Test call logging
call_log = fetcher.get_call_log()
assert len(call_log) > 0
@pytest.mark.asyncio
async def test_mock_stock_screener(self):
"""Test the mock stock screener implementation."""
screener = MockStockScreener()
# Test maverick recommendations
maverick = await screener.get_maverick_recommendations(limit=5)
assert isinstance(maverick, list)
assert len(maverick) <= 5
# Test bear recommendations
bear = await screener.get_maverick_bear_recommendations(limit=3)
assert isinstance(bear, list)
assert len(bear) <= 3
# Test trending recommendations
trending = await screener.get_trending_recommendations(limit=2)
assert isinstance(trending, list)
assert len(trending) <= 2
# Test all recommendations
all_recs = await screener.get_all_screening_recommendations()
assert "maverick_stocks" in all_recs
assert "maverick_bear_stocks" in all_recs
assert "trending_stocks" in all_recs
@pytest.mark.asyncio
async def test_mock_market_data_provider(self):
"""Test the mock market data provider implementation."""
provider = MockMarketDataProvider()
# Test market summary
summary = await provider.get_market_summary()
assert isinstance(summary, dict)
assert "^GSPC" in summary
# Test top gainers
gainers = await provider.get_top_gainers(5)
assert isinstance(gainers, list)
assert len(gainers) <= 5
# Test sector performance
sectors = await provider.get_sector_performance()
assert isinstance(sectors, dict)
assert "Technology" in sectors
@pytest.mark.asyncio
async def test_mock_macro_data_provider(self):
"""Test the mock macro data provider implementation."""
provider = MockMacroDataProvider()
# Test individual indicators
gdp = await provider.get_gdp_growth_rate()
assert "current" in gdp
assert "previous" in gdp
unemployment = await provider.get_unemployment_rate()
assert "current" in unemployment
vix = await provider.get_vix()
assert isinstance(vix, int | float) or vix is None
# Test comprehensive statistics
stats = await provider.get_macro_statistics()
assert "sentiment_score" in stats
assert "gdp_growth_rate" in stats
def test_mock_configuration_provider(self):
"""Test the mock configuration provider implementation."""
config = MockConfigurationProvider()
# Test default values
assert config.get_database_url() == "sqlite:///:memory:"
assert config.is_cache_enabled() is False
assert config.is_development_mode() is True
# Test overrides
config.set_override("CACHE_ENABLED", True)
assert config.is_cache_enabled() is True
# Test helper methods
config.enable_cache()
assert config.is_cache_enabled() is True
config.disable_cache()
assert config.is_cache_enabled() is False
class TestProviderFactory:
"""Test the provider factory functionality."""
def test_provider_factory_creation(self):
"""Test creating providers through the factory."""
config = ConfigurationFactory.create_test_config()
factory = ProviderFactory(config)
# Test provider creation
cache_manager = factory.get_cache_manager()
assert cache_manager is not None
persistence = factory.get_persistence()
assert persistence is not None
stock_fetcher = factory.get_stock_data_fetcher()
assert stock_fetcher is not None
# Test singleton behavior
cache_manager2 = factory.get_cache_manager()
assert cache_manager is cache_manager2
def test_provider_factory_validation(self):
"""Test provider factory configuration validation."""
config = ConfigurationFactory.create_test_config()
factory = ProviderFactory(config)
errors = factory.validate_configuration()
assert isinstance(errors, list)
# Test config should have no errors
assert len(errors) == 0
def test_provider_factory_reset(self):
"""Test provider factory cache reset."""
config = ConfigurationFactory.create_test_config()
factory = ProviderFactory(config)
# Create providers
cache1 = factory.get_cache_manager()
# Reset factory
factory.reset_cache()
# Get provider again
cache2 = factory.get_cache_manager()
# Should be different instances
assert cache1 is not cache2
class TestDependencyInjection:
"""Test the dependency injection system."""
def test_dependency_override_context(self):
"""Test dependency override context manager."""
mock_cache = MockCacheManager()
with DependencyOverride(cache_manager=mock_cache):
# Inside the context, dependencies should be overridden
# This would be tested with actual dependency resolution
pass
# Outside the context, dependencies should be restored
assert True # Placeholder assertion
def test_create_test_dependencies(self):
"""Test creating test dependencies."""
mock_cache = MockCacheManager()
deps = create_test_dependencies(cache_manager=mock_cache)
assert "cache_manager" in deps
assert deps["cache_manager"] is mock_cache
assert "stock_data_fetcher" in deps
assert "configuration" in deps
def test_get_dependencies_for_testing(self):
"""Test getting dependencies configured for testing."""
deps = get_dependencies_for_testing()
assert isinstance(deps, dict)
assert "cache_manager" in deps
assert "stock_data_fetcher" in deps
class TestIntegrationScenarios:
"""Test integration scenarios using the new architecture."""
@pytest.mark.asyncio
async def test_stock_data_with_caching(self):
"""Test stock data fetching with caching integration."""
# Create mock dependencies
cache = MockCacheManager()
fetcher = MockStockDataFetcher()
config = MockConfigurationProvider()
config.enable_cache()
# Set up test data
test_data = pd.DataFrame(
{
"Open": [100.0, 101.0],
"High": [102.0, 103.0],
"Low": [99.0, 100.0],
"Close": [101.0, 102.0],
"Volume": [1000000, 1100000],
},
index=pd.date_range("2024-01-01", periods=2),
)
fetcher.set_test_data("AAPL", test_data)
# Test the integration
cache_key = "stock_data:AAPL:2024-01-01:2024-01-02"
# First call should fetch from provider
data = await fetcher.get_stock_data("AAPL", "2024-01-01", "2024-01-02")
assert not data.empty
# Cache the result
await cache.set(cache_key, data.to_dict(), ttl=300)
# Verify cache hit
cached_result = await cache.get(cache_key)
assert cached_result is not None
@pytest.mark.asyncio
async def test_screening_workflow(self):
"""Test a complete screening workflow."""
screener = MockStockScreener()
# Set up test recommendations
test_maverick = [
{"symbol": "TEST1", "combined_score": 95, "momentum_score": 90},
{"symbol": "TEST2", "combined_score": 85, "momentum_score": 85},
]
screener.set_test_recommendations("maverick", test_maverick)
# Test the workflow
results = await screener.get_maverick_recommendations(limit=10, min_score=80)
assert len(results) == 2
# Test filtering
filtered_results = await screener.get_maverick_recommendations(
limit=10, min_score=90
)
assert len(filtered_results) == 1
assert filtered_results[0]["symbol"] == "TEST1"
def test_configuration_scenarios(self):
"""Test different configuration scenarios."""
# Test development config
dev_config = ConfigurationFactory.create_development_config()
assert dev_config.is_development_mode()
# Test with overrides
test_config = ConfigurationFactory.create_test_config(
{
"CACHE_ENABLED": "true",
"AUTH_ENABLED": "true",
}
)
assert test_config.is_cache_enabled()
assert test_config.is_auth_enabled()
def test_mock_behavior_verification(self):
"""Test that mocks properly track behavior for verification."""
cache = MockCacheManager()
# Perform some operations
import asyncio
async def perform_operations():
await cache.set("key1", "value1")
await cache.get("key1")
await cache.delete("key1")
asyncio.run(perform_operations())
# Verify call log
call_log = cache.get_call_log()
assert len(call_log) == 3
assert call_log[0]["method"] == "set"
assert call_log[1]["method"] == "get"
assert call_log[2]["method"] == "delete"
if __name__ == "__main__":
# Run a simple smoke test
import asyncio
async def smoke_test():
"""Run a simple smoke test of the architecture."""
print("Running provider architecture smoke test...")
# Test mock implementations
cache = MockCacheManager()
await cache.set("test", "value")
result = await cache.get("test")
assert result == "value"
print("✓ Cache manager working")
fetcher = MockStockDataFetcher()
data = await fetcher.get_stock_data("AAPL")
assert not data.empty
print("✓ Stock data fetcher working")
screener = MockStockScreener()
recommendations = await screener.get_maverick_recommendations()
assert len(recommendations) > 0
print("✓ Stock screener working")
# Test factory
config = ConfigurationFactory.create_test_config()
factory = ProviderFactory(config)
errors = factory.validate_configuration()
assert len(errors) == 0
print("✓ Provider factory working")
print("All tests passed! 🎉")
asyncio.run(smoke_test())
```
--------------------------------------------------------------------------------
/maverick_mcp/api/openapi_config.py:
--------------------------------------------------------------------------------
```python
"""
Custom OpenAPI configuration for MaverickMCP API.
This module provides enhanced OpenAPI schema generation with:
- Comprehensive API metadata tailored for the open-source build
- Standardized tags and descriptions
- Custom examples and documentation
- Export functionality for Postman/Insomnia
"""
from typing import Any
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from fastapi.responses import JSONResponse, Response
def custom_openapi(app: FastAPI) -> dict[str, Any]:
"""
Generate custom OpenAPI schema with enhanced documentation.
Args:
app: FastAPI application instance
Returns:
OpenAPI schema dictionary
"""
if app.openapi_schema:
return app.openapi_schema
description = """
# MaverickMCP Personal Research API
MaverickMCP is an open-source Model Context Protocol (MCP) server focused on
independent research, portfolio experimentation, and desktop analytics. It runs
entirely without billing, subscription tracking, or usage credits.
## Highlights
- 📊 **Historical & Intraday Market Data** — request equities data across
flexible ranges with caching for fast iteration.
- 📈 **Advanced Technical Analysis** — generate RSI, MACD, Bollinger Bands, and
other indicator overlays for deeper insight.
- 🧪 **Backtesting & Scenario Tools** — evaluate trading ideas with the
VectorBT-powered engine and inspect saved results locally.
- 🧠 **Research Agents & Screeners** — launch summarization and screening tools
that operate with zero payment integration.
- 🛡️ **Secure Defaults** — observability hooks, CSP headers, and rate limiting
are enabled without requiring extra configuration.
## Access Model
- No authentication or API keys are required in this distribution.
- There is no purchase flow, billing portal, or credit ledger.
- All stateful data remains on the machine that hosts the server.
## Error Handling
Every error response follows this JSON envelope:
```json
{
"success": false,
"error": {
"code": "ERROR_CODE",
"message": "Human readable explanation"
},
"status_code": 400,
"trace_id": "uuid-for-debugging"
}
```
## Support
- Documentation: https://github.com/wshobson/maverick-mcp#readme
- GitHub Issues: https://github.com/wshobson/maverick-mcp/issues
- Discussions: https://github.com/wshobson/maverick-mcp/discussions
"""
tags = [
{
"name": "Technical Analysis",
"description": """
Stock technical indicators and analytics for personal research.
Generate RSI, MACD, Bollinger Bands, and multi-indicator overlays
without authentication or billing requirements.
""",
},
{
"name": "Market Data",
"description": """
Historical and intraday market data endpoints.
Fetch quotes, price history, and metadata with smart caching to keep
local research responsive.
""",
},
{
"name": "Stock Screening",
"description": """
Automated screeners and discovery workflows.
Run Maverick and custom screening strategies to surface candidates
for deeper analysis.
""",
},
{
"name": "Research Agents",
"description": """
AI-assisted research personas and orchestration helpers.
Summarize market structure, compile reports, and investigate trends
entirely within your self-hosted environment.
""",
},
{
"name": "Backtesting",
"description": """
Strategy evaluation and performance inspection tools.
Execute parameterized backtests with VectorBT and review results
without uploading data to third-party services.
""",
},
{
"name": "Portfolio",
"description": """
Personal portfolio calculators and scenario planners.
Model allocations, rebalance strategies, and track watchlists with
zero dependency on payment providers.
""",
},
{
"name": "Monitoring",
"description": """
Operational monitoring and diagnostics endpoints.
Inspect Prometheus metrics, runtime health, and background task
status for self-hosted deployments.
""",
},
{
"name": "Health",
"description": """
Lightweight readiness and liveness checks.
Ideal for Docker, Kubernetes, or local supervisor probes.
""",
},
]
servers = [
{
"url": "http://localhost:8000",
"description": "Local HTTP development server",
},
{
"url": "http://0.0.0.0:8003",
"description": "Default SSE transport endpoint",
},
]
openapi_schema = get_openapi(
title="MaverickMCP API",
version="1.0.0",
description=description,
routes=app.routes,
tags=tags,
servers=servers,
contact={
"name": "MaverickMCP Maintainers",
"url": "https://github.com/wshobson/maverick-mcp",
},
license_info={
"name": "MIT License",
"url": "https://github.com/wshobson/maverick-mcp/blob/main/LICENSE",
},
)
# Add external docs
openapi_schema["externalDocs"] = {
"description": "Project documentation",
"url": "https://github.com/wshobson/maverick-mcp#readme",
}
# The open-source build intentionally has no authentication schemes.
openapi_schema.setdefault("components", {})
openapi_schema["components"]["securitySchemes"] = {}
openapi_schema["security"] = []
# Add common response schemas
if "components" not in openapi_schema:
openapi_schema["components"] = {}
openapi_schema["components"]["responses"] = {
"UnauthorizedError": {
"description": "Authentication required or invalid credentials",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"},
"example": {
"success": False,
"error": {
"code": "UNAUTHORIZED",
"message": "Authentication required",
},
"status_code": 401,
},
}
},
},
"ForbiddenError": {
"description": "Insufficient permissions",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"},
"example": {
"success": False,
"error": {
"code": "FORBIDDEN",
"message": "Insufficient permissions for this operation",
},
"status_code": 403,
},
}
},
},
"NotFoundError": {
"description": "Resource not found",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"},
"example": {
"success": False,
"error": {
"code": "NOT_FOUND",
"message": "The requested resource was not found",
},
"status_code": 404,
},
}
},
},
"ValidationError": {
"description": "Request validation failed",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ValidationErrorResponse"},
"example": {
"success": False,
"error": {
"code": "VALIDATION_ERROR",
"message": "Validation failed",
},
"errors": [
{
"code": "INVALID_FORMAT",
"field": "email",
"message": "Invalid email format",
}
],
"status_code": 422,
},
}
},
},
"RateLimitError": {
"description": "Rate limit exceeded",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/RateLimitResponse"},
"example": {
"success": False,
"error": {
"code": "RATE_LIMIT_EXCEEDED",
"message": "Too many requests",
},
"rate_limit": {
"limit": 100,
"remaining": 0,
"reset": "2024-01-15T12:00:00Z",
"retry_after": 42,
},
"status_code": 429,
},
}
},
},
"ServerError": {
"description": "Internal server error",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ErrorResponse"},
"example": {
"success": False,
"error": {
"code": "INTERNAL_ERROR",
"message": "An unexpected error occurred",
},
"status_code": 500,
"trace_id": "550e8400-e29b-41d4-a716-446655440000",
},
}
},
},
}
# Cache the schema
app.openapi_schema = openapi_schema
return app.openapi_schema
def configure_openapi(app: FastAPI) -> None:
"""
Configure OpenAPI for the FastAPI application.
Args:
app: FastAPI application instance
"""
# Override the default OpenAPI function
app.openapi = lambda: custom_openapi(app) # type: ignore[assignment]
# Add export endpoints
@app.get("/api/openapi.json", include_in_schema=False)
async def get_openapi_json():
"""Get raw OpenAPI JSON specification."""
return JSONResponse(content=custom_openapi(app))
@app.get("/api/openapi.yaml", include_in_schema=False)
async def get_openapi_yaml():
"""Get OpenAPI specification in YAML format."""
import yaml
openapi_dict = custom_openapi(app)
yaml_content = yaml.dump(openapi_dict, sort_keys=False, allow_unicode=True)
return Response(
content=yaml_content,
media_type="application/x-yaml",
headers={"Content-Disposition": "attachment; filename=openapi.yaml"},
)
# Add Postman collection export
@app.get("/api/postman.json", include_in_schema=False)
async def get_postman_collection():
"""Export API as Postman collection."""
from maverick_mcp.api.utils.postman_export import convert_to_postman
openapi_dict = custom_openapi(app)
postman_collection = convert_to_postman(openapi_dict)
return JSONResponse(
content=postman_collection,
headers={
"Content-Disposition": "attachment; filename=maverickmcp-api.postman_collection.json"
},
)
# Add Insomnia collection export
@app.get("/api/insomnia.json", include_in_schema=False)
async def get_insomnia_collection():
"""Export API as Insomnia collection."""
from maverick_mcp.api.utils.insomnia_export import convert_to_insomnia
openapi_dict = custom_openapi(app)
insomnia_collection = convert_to_insomnia(openapi_dict)
return JSONResponse(
content=insomnia_collection,
headers={
"Content-Disposition": "attachment; filename=maverickmcp-api.insomnia_collection.json"
},
)
# ReDoc configuration
REDOC_CONFIG = {
"spec_url": "/api/openapi.json",
"title": "MaverickMCP API Documentation",
"favicon_url": "https://maverickmcp.com/favicon.ico",
"logo": {"url": "https://maverickmcp.com/logo.png", "altText": "MaverickMCP Logo"},
"theme": {
"colors": {
"primary": {
"main": "#2563eb" # Blue-600
}
},
"typography": {"fontSize": "14px", "code": {"fontSize": "13px"}},
},
"hideDownloadButton": False,
"disableSearch": False,
"showExtensions": True,
"expandResponses": "200,201",
"requiredPropsFirst": True,
"sortPropsAlphabetically": False,
"payloadSampleIdx": 0,
"hideHostname": False,
"noAutoAuth": False,
}
```
--------------------------------------------------------------------------------
/maverick_mcp/backtesting/optimization.py:
--------------------------------------------------------------------------------
```python
"""Strategy optimization utilities for VectorBT."""
from typing import Any
import numpy as np
import pandas as pd
class StrategyOptimizer:
"""Optimizer for trading strategy parameters."""
def __init__(self, engine):
"""Initialize optimizer with VectorBT engine.
Args:
engine: VectorBTEngine instance
"""
self.engine = engine
def generate_param_grid(
self, strategy_type: str, optimization_level: str = "medium"
) -> dict[str, list]:
"""Generate parameter grid based on strategy and optimization level.
Args:
strategy_type: Type of strategy
optimization_level: Level of optimization (coarse, medium, fine)
Returns:
Parameter grid for optimization
"""
if strategy_type == "sma_cross":
return self._sma_param_grid(optimization_level)
elif strategy_type == "rsi":
return self._rsi_param_grid(optimization_level)
elif strategy_type == "macd":
return self._macd_param_grid(optimization_level)
elif strategy_type == "bollinger":
return self._bollinger_param_grid(optimization_level)
elif strategy_type == "momentum":
return self._momentum_param_grid(optimization_level)
else:
raise ValueError(f"Unknown strategy type: {strategy_type}")
def _sma_param_grid(self, level: str) -> dict[str, list]:
"""Generate SMA crossover parameter grid."""
if level == "coarse":
return {
"fast_period": [5, 10, 20],
"slow_period": [20, 50, 100],
}
elif level == "fine":
return {
"fast_period": list(range(5, 25, 2)),
"slow_period": list(range(20, 101, 5)),
}
else: # medium
return {
"fast_period": [5, 10, 15, 20],
"slow_period": [20, 30, 50, 100],
}
def _rsi_param_grid(self, level: str) -> dict[str, list]:
"""Generate RSI parameter grid."""
if level == "coarse":
return {
"period": [7, 14, 21],
"oversold": [20, 30],
"overbought": [70, 80],
}
elif level == "fine":
return {
"period": list(range(7, 22, 2)),
"oversold": list(range(20, 41, 5)),
"overbought": list(range(60, 81, 5)),
}
else: # medium
return {
"period": [7, 14, 21],
"oversold": [20, 25, 30, 35],
"overbought": [65, 70, 75, 80],
}
def _macd_param_grid(self, level: str) -> dict[str, list]:
"""Generate MACD parameter grid."""
if level == "coarse":
return {
"fast_period": [8, 12],
"slow_period": [21, 26],
"signal_period": [9],
}
elif level == "fine":
return {
"fast_period": list(range(8, 15)),
"slow_period": list(range(20, 31)),
"signal_period": list(range(7, 12)),
}
else: # medium
return {
"fast_period": [8, 10, 12, 14],
"slow_period": [21, 24, 26, 30],
"signal_period": [7, 9, 11],
}
def _bollinger_param_grid(self, level: str) -> dict[str, list]:
"""Generate Bollinger Bands parameter grid."""
if level == "coarse":
return {
"period": [10, 20],
"std_dev": [1.5, 2.0, 2.5],
}
elif level == "fine":
return {
"period": list(range(10, 31, 2)),
"std_dev": np.arange(1.0, 3.1, 0.25).tolist(),
}
else: # medium
return {
"period": [10, 15, 20, 25],
"std_dev": [1.5, 2.0, 2.5, 3.0],
}
def _momentum_param_grid(self, level: str) -> dict[str, list]:
"""Generate momentum parameter grid."""
if level == "coarse":
return {
"lookback": [10, 20, 30],
"threshold": [0.03, 0.05, 0.10],
}
elif level == "fine":
return {
"lookback": list(range(10, 41, 2)),
"threshold": np.arange(0.02, 0.11, 0.01).tolist(),
}
else: # medium
return {
"lookback": [10, 15, 20, 25, 30],
"threshold": [0.02, 0.03, 0.05, 0.07, 0.10],
}
async def walk_forward_analysis(
self,
symbol: str,
strategy_type: str,
parameters: dict[str, Any],
start_date: str,
end_date: str,
window_size: int = 252, # Trading days in a year
step_size: int = 63, # Trading days in a quarter
optimization_window: int = 504, # 2 years for optimization
) -> dict[str, Any]:
"""Perform walk-forward analysis.
Args:
symbol: Stock symbol
strategy_type: Strategy type
parameters: Initial parameters
start_date: Start date
end_date: End date
window_size: Test window size in days
step_size: Step size for rolling window
optimization_window: Optimization window size
Returns:
Walk-forward analysis results
"""
results = []
# Convert dates to pandas datetime
start = pd.to_datetime(start_date)
end = pd.to_datetime(end_date)
current = start + pd.Timedelta(days=optimization_window)
while current <= end:
# Optimization period
opt_start = current - pd.Timedelta(days=optimization_window)
opt_end = current
# Test period
test_start = current
test_end = min(current + pd.Timedelta(days=window_size), end)
# Optimize on training data
param_grid = self.generate_param_grid(strategy_type, "coarse")
optimization = await self.engine.optimize_parameters(
symbol=symbol,
strategy_type=strategy_type,
param_grid=param_grid,
start_date=opt_start.strftime("%Y-%m-%d"),
end_date=opt_end.strftime("%Y-%m-%d"),
top_n=1,
)
best_params = optimization["best_parameters"]
# Test on out-of-sample data
if test_start < test_end:
test_result = await self.engine.run_backtest(
symbol=symbol,
strategy_type=strategy_type,
parameters=best_params,
start_date=test_start.strftime("%Y-%m-%d"),
end_date=test_end.strftime("%Y-%m-%d"),
)
results.append(
{
"period": f"{test_start.strftime('%Y-%m-%d')} to {test_end.strftime('%Y-%m-%d')}",
"parameters": best_params,
"in_sample_sharpe": optimization["best_metric_value"],
"out_sample_return": test_result["metrics"]["total_return"],
"out_sample_sharpe": test_result["metrics"]["sharpe_ratio"],
"out_sample_drawdown": test_result["metrics"]["max_drawdown"],
}
)
# Move window forward
current += pd.Timedelta(days=step_size)
# Calculate aggregate metrics
if results:
avg_return = np.mean([r["out_sample_return"] for r in results])
avg_sharpe = np.mean([r["out_sample_sharpe"] for r in results])
avg_drawdown = np.mean([r["out_sample_drawdown"] for r in results])
consistency = sum(1 for r in results if r["out_sample_return"] > 0) / len(
results
)
else:
avg_return = avg_sharpe = avg_drawdown = consistency = 0
return {
"symbol": symbol,
"strategy": strategy_type,
"periods_tested": len(results),
"average_return": avg_return,
"average_sharpe": avg_sharpe,
"average_drawdown": avg_drawdown,
"consistency": consistency,
"walk_forward_results": results,
"summary": self._generate_wf_summary(avg_return, avg_sharpe, consistency),
}
def _generate_wf_summary(
self, avg_return: float, avg_sharpe: float, consistency: float
) -> str:
"""Generate walk-forward analysis summary."""
summary = f"Walk-forward analysis shows {avg_return * 100:.1f}% average return "
summary += f"with Sharpe ratio of {avg_sharpe:.2f}. "
summary += f"Strategy was profitable in {consistency * 100:.0f}% of periods. "
if avg_sharpe >= 1.0 and consistency >= 0.7:
summary += "Results indicate robust performance across different market conditions."
elif avg_sharpe >= 0.5 and consistency >= 0.5:
summary += "Results show moderate robustness with room for improvement."
else:
summary += "Results suggest the strategy may not be robust to changing market conditions."
return summary
async def monte_carlo_simulation(
self,
backtest_results: dict[str, Any],
num_simulations: int = 1000,
confidence_levels: list[float] | None = None,
) -> dict[str, Any]:
"""Run Monte Carlo simulation on backtest results.
Args:
backtest_results: Results from run_backtest
num_simulations: Number of simulations to run
confidence_levels: Confidence levels for percentiles
Returns:
Monte Carlo simulation results
"""
if confidence_levels is None:
confidence_levels = [0.05, 0.25, 0.50, 0.75, 0.95]
trades = backtest_results.get("trades", [])
if not trades:
return {"error": "No trades to simulate"}
# Extract returns from trades
trade_returns = [t["return"] for t in trades]
# Run simulations
simulated_returns = []
simulated_drawdowns = []
for _ in range(num_simulations):
# Bootstrap sample with replacement
sampled_returns = np.random.choice(
trade_returns, size=len(trade_returns), replace=True
)
# Calculate cumulative return
cumulative = np.cumprod(1 + np.array(sampled_returns))
total_return = cumulative[-1] - 1
# Calculate max drawdown
running_max = np.maximum.accumulate(cumulative)
drawdown = (cumulative - running_max) / running_max
max_drawdown = np.min(drawdown)
simulated_returns.append(total_return)
simulated_drawdowns.append(max_drawdown)
# Calculate percentiles
return_percentiles = np.percentile(
simulated_returns, np.array(confidence_levels) * 100
)
drawdown_percentiles = np.percentile(
simulated_drawdowns, np.array(confidence_levels) * 100
)
return {
"num_simulations": num_simulations,
"expected_return": np.mean(simulated_returns),
"return_std": np.std(simulated_returns),
"return_percentiles": dict(
zip(
[f"p{int(cl * 100)}" for cl in confidence_levels],
return_percentiles.tolist(),
strict=False,
)
),
"expected_drawdown": np.mean(simulated_drawdowns),
"drawdown_std": np.std(simulated_drawdowns),
"drawdown_percentiles": dict(
zip(
[f"p{int(cl * 100)}" for cl in confidence_levels],
drawdown_percentiles.tolist(),
strict=False,
)
),
"probability_profit": sum(1 for r in simulated_returns if r > 0)
/ num_simulations,
"var_95": return_percentiles[0], # Value at Risk at 95% confidence
"summary": self._generate_mc_summary(
np.mean(simulated_returns),
return_percentiles[0],
sum(1 for r in simulated_returns if r > 0) / num_simulations,
),
}
def _generate_mc_summary(
self, expected_return: float, var_95: float, prob_profit: float
) -> str:
"""Generate Monte Carlo simulation summary."""
summary = f"Monte Carlo simulation shows {expected_return * 100:.1f}% expected return "
summary += f"with {prob_profit * 100:.1f}% probability of profit. "
summary += f"95% Value at Risk is {abs(var_95) * 100:.1f}%. "
if prob_profit >= 0.8 and expected_return > 0.10:
summary += "Strategy shows strong probabilistic edge."
elif prob_profit >= 0.6 and expected_return > 0:
summary += "Strategy shows positive expectancy with moderate confidence."
else:
summary += "Strategy may not have sufficient edge for live trading."
return summary
```
--------------------------------------------------------------------------------
/maverick_mcp/api/services/resource_service.py:
--------------------------------------------------------------------------------
```python
"""
Resource service for MaverickMCP API.
Handles MCP resources including health endpoints and stock data resources.
Extracted from server.py to improve code organization and maintainability.
"""
from typing import Any
from .base_service import BaseService
class ResourceService(BaseService):
"""
Service class for MCP resource operations.
Provides health endpoints, stock data resources, and other MCP resources.
"""
def register_tools(self):
"""Register resource endpoints with MCP."""
@self.mcp.resource("health://")
def health_resource() -> dict[str, Any]:
"""
Comprehensive health check endpoint.
Returns system health status including database, Redis, and external services.
"""
return self._health_check()
@self.mcp.resource("stock://{ticker}")
def stock_resource(ticker: str) -> Any:
"""
Get stock data resource for a specific ticker.
Args:
ticker: Stock ticker symbol
Returns:
Stock data resource
"""
return self._get_stock_resource(ticker)
@self.mcp.resource("stock://{ticker}/{start_date}/{end_date}")
def stock_resource_with_dates(
ticker: str, start_date: str, end_date: str
) -> Any:
"""
Get stock data resource for a specific ticker and date range.
Args:
ticker: Stock ticker symbol
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
Returns:
Stock data resource for the specified date range
"""
return self._get_stock_resource_with_dates(ticker, start_date, end_date)
@self.mcp.resource("stock_info://{ticker}")
def stock_info_resource(ticker: str) -> dict[str, Any]:
"""
Get stock information resource for a specific ticker.
Args:
ticker: Stock ticker symbol
Returns:
Stock information resource
"""
return self._get_stock_info_resource(ticker)
def _health_check(self) -> dict[str, Any]:
"""Comprehensive health check implementation."""
from maverick_mcp.config.validation import get_validation_status
from maverick_mcp.data.cache import get_redis_client
from maverick_mcp.data.health import get_database_health
health_status = {
"status": "healthy",
"timestamp": self._get_current_timestamp(),
"version": "1.0.0",
"environment": "production" if not self.is_debug_mode() else "development",
"services": {},
"configuration": {},
}
# Check database health
try:
db_health = get_database_health()
health_status["services"]["database"] = {
"status": "healthy" if db_health.get("connected") else "unhealthy",
"details": db_health,
}
except Exception as e:
health_status["services"]["database"] = {
"status": "unhealthy",
"error": str(e),
}
health_status["status"] = "degraded"
# Check Redis health
try:
redis_client = get_redis_client()
if redis_client:
redis_client.ping()
health_status["services"]["redis"] = {
"status": "healthy",
"cache_enabled": True,
}
else:
health_status["services"]["redis"] = {
"status": "unavailable",
"cache_enabled": False,
"fallback": "in-memory cache",
}
except Exception as e:
health_status["services"]["redis"] = {
"status": "unhealthy",
"error": str(e),
"fallback": "in-memory cache",
}
if health_status["status"] == "healthy":
health_status["status"] = "degraded"
# Check authentication service
health_status["services"]["authentication"] = {
"status": "healthy" if self.is_auth_enabled() else "disabled",
"enabled": self.is_auth_enabled(),
}
# Configuration validation
try:
validation_status = get_validation_status()
health_status["configuration"] = {
"status": "valid" if validation_status.get("valid") else "invalid",
"details": validation_status,
}
except Exception as e:
health_status["configuration"] = {
"status": "error",
"error": str(e),
}
health_status["status"] = "unhealthy"
# External services check (stock data providers)
health_status["services"]["stock_data"] = self._check_stock_data_providers()
self.log_tool_usage("health_check", status=health_status["status"])
return health_status
def _check_stock_data_providers(self) -> dict[str, Any]:
"""Check health of stock data providers."""
try:
from maverick_mcp.providers.stock_data import StockDataProvider
provider = StockDataProvider()
# Test with a simple request
test_data = provider.get_stock_data("AAPL", days=1)
if not test_data.empty:
return {
"status": "healthy",
"provider": "yfinance",
"last_test": self._get_current_timestamp(),
}
else:
return {
"status": "degraded",
"provider": "yfinance",
"issue": "Empty data returned",
}
except Exception as e:
return {
"status": "unhealthy",
"provider": "yfinance",
"error": str(e),
}
def _get_stock_resource(self, ticker: str) -> Any:
"""Get stock resource implementation."""
try:
from maverick_mcp.providers.stock_data import StockDataProvider
provider = StockDataProvider()
# Get recent stock data (30 days)
df = provider.get_stock_data(ticker.upper(), days=30)
if df.empty:
return {
"error": f"No data available for ticker {ticker}",
"ticker": ticker.upper(),
}
# Convert DataFrame to resource format
resource_data = {
"ticker": ticker.upper(),
"data_points": len(df),
"date_range": {
"start": df.index[0].isoformat(),
"end": df.index[-1].isoformat(),
},
"latest_price": float(df["Close"].iloc[-1]),
"price_change": float(df["Close"].iloc[-1] - df["Close"].iloc[-2])
if len(df) > 1
else 0,
"volume": int(df["Volume"].iloc[-1]),
"high_52w": float(df["High"].max()),
"low_52w": float(df["Low"].min()),
"data": df.to_dict(orient="records"),
}
self.log_tool_usage("stock_resource", ticker=ticker)
return resource_data
except Exception as e:
self.logger.error(f"Failed to get stock resource for {ticker}: {e}")
return {
"error": f"Failed to fetch stock data: {str(e)}",
"ticker": ticker.upper(),
}
def _get_stock_resource_with_dates(
self, ticker: str, start_date: str, end_date: str
) -> Any:
"""Get stock resource with date range implementation."""
try:
from datetime import datetime
from maverick_mcp.providers.stock_data import StockDataProvider
# Validate date format
try:
datetime.strptime(start_date, "%Y-%m-%d")
datetime.strptime(end_date, "%Y-%m-%d")
except ValueError:
return {
"error": "Invalid date format. Use YYYY-MM-DD format.",
"start_date": start_date,
"end_date": end_date,
}
provider = StockDataProvider()
# Get stock data for specified date range
df = provider.get_stock_data(ticker.upper(), start_date, end_date)
if df.empty:
return {
"error": f"No data available for ticker {ticker} in date range {start_date} to {end_date}",
"ticker": ticker.upper(),
"start_date": start_date,
"end_date": end_date,
}
# Convert DataFrame to resource format
resource_data = {
"ticker": ticker.upper(),
"start_date": start_date,
"end_date": end_date,
"data_points": len(df),
"actual_date_range": {
"start": df.index[0].isoformat(),
"end": df.index[-1].isoformat(),
},
"price_summary": {
"open": float(df["Open"].iloc[0]),
"close": float(df["Close"].iloc[-1]),
"high": float(df["High"].max()),
"low": float(df["Low"].min()),
"change": float(df["Close"].iloc[-1] - df["Open"].iloc[0]),
"change_pct": float(
(
(df["Close"].iloc[-1] - df["Open"].iloc[0])
/ df["Open"].iloc[0]
)
* 100
),
},
"volume_summary": {
"total": int(df["Volume"].sum()),
"average": int(df["Volume"].mean()),
"max": int(df["Volume"].max()),
},
"data": df.to_dict(orient="records"),
}
self.log_tool_usage(
"stock_resource_with_dates",
ticker=ticker,
start_date=start_date,
end_date=end_date,
)
return resource_data
except Exception as e:
self.logger.error(
f"Failed to get stock resource for {ticker} ({start_date} to {end_date}): {e}"
)
return {
"error": f"Failed to fetch stock data: {str(e)}",
"ticker": ticker.upper(),
"start_date": start_date,
"end_date": end_date,
}
def _get_stock_info_resource(self, ticker: str) -> dict[str, Any]:
"""Get stock info resource implementation."""
try:
from maverick_mcp.providers.stock_data import StockDataProvider
provider = StockDataProvider()
# Get stock information
stock_info = provider.get_stock_info(ticker.upper())
if not stock_info:
return {
"error": f"No information available for ticker {ticker}",
"ticker": ticker.upper(),
}
# Format stock info resource
resource_data = {
"ticker": ticker.upper(),
"company_name": stock_info.get(
"longName", stock_info.get("shortName", "N/A")
),
"sector": stock_info.get("sector", "N/A"),
"industry": stock_info.get("industry", "N/A"),
"market_cap": stock_info.get("marketCap"),
"enterprise_value": stock_info.get("enterpriseValue"),
"pe_ratio": stock_info.get("trailingPE"),
"forward_pe": stock_info.get("forwardPE"),
"price_to_book": stock_info.get("priceToBook"),
"dividend_yield": stock_info.get("dividendYield"),
"beta": stock_info.get("beta"),
"52_week_high": stock_info.get("fiftyTwoWeekHigh"),
"52_week_low": stock_info.get("fiftyTwoWeekLow"),
"average_volume": stock_info.get("averageVolume"),
"shares_outstanding": stock_info.get("sharesOutstanding"),
"float_shares": stock_info.get("floatShares"),
"business_summary": stock_info.get("longBusinessSummary", "N/A"),
"website": stock_info.get("website"),
"employees": stock_info.get("fullTimeEmployees"),
"last_updated": self._get_current_timestamp(),
}
self.log_tool_usage("stock_info_resource", ticker=ticker)
return resource_data
except Exception as e:
self.logger.error(f"Failed to get stock info resource for {ticker}: {e}")
return {
"error": f"Failed to fetch stock information: {str(e)}",
"ticker": ticker.upper(),
}
def _get_current_timestamp(self) -> str:
"""Get current timestamp in ISO format."""
from datetime import UTC, datetime
return datetime.now(UTC).isoformat()
```
--------------------------------------------------------------------------------
/scripts/README_TIINGO_LOADER.md:
--------------------------------------------------------------------------------
```markdown
# Tiingo Data Loader for Maverick-MCP
A comprehensive, production-ready data loader for fetching market data from Tiingo API and storing it in the Maverick-MCP database with technical indicators and screening algorithms.
## Features
### 🚀 Core Capabilities
- **Async Operations**: High-performance async data fetching with configurable concurrency
- **Rate Limiting**: Built-in rate limiting to respect Tiingo's 2400 requests/hour limit
- **Progress Tracking**: Resume capability with checkpoint files for interrupted loads
- **Error Handling**: Exponential backoff retry logic with comprehensive error handling
- **Batch Processing**: Efficient batch processing with configurable batch sizes
### 📊 Data Processing
- **Technical Indicators**: 50+ technical indicators using pandas-ta
- **Screening Algorithms**: Built-in Maverick, Bear Market, and Supply/Demand screens
- **Database Optimization**: Bulk inserts with connection pooling
- **Data Validation**: Comprehensive data validation and cleaning
### 🎛️ Flexible Configuration
- **Multiple Symbol Sources**: S&P 500, custom files, individual symbols, or all Tiingo-supported tickers
- **Date Range Control**: Configurable date ranges with year-based shortcuts
- **Processing Options**: Enable/disable technical indicators and screening
- **Performance Tuning**: Adjustable concurrency, batch sizes, and timeouts
## Installation
### Prerequisites
1. **Tiingo API Token**: Sign up at [tiingo.com](https://www.tiingo.com) and get your API token
2. **Database**: Ensure Maverick-MCP database is set up and accessible
3. **Python Dependencies**: pandas-ta, aiohttp, SQLAlchemy, and other requirements
### Setup
```bash
# Set your Tiingo API token
export TIINGO_API_TOKEN=your_token_here
# Set database URL (if different from default)
export DATABASE_URL=postgresql://user:pass@localhost/maverick_mcp
# Make scripts executable
chmod +x scripts/load_tiingo_data.py
chmod +x scripts/load_example.py
```
## Usage Examples
### 1. Load S&P 500 Stocks (Top 100)
```bash
# Load 2 years of data with technical indicators
python scripts/load_tiingo_data.py --sp500 --years 2 --calculate-indicators
# Load with screening algorithms
python scripts/load_tiingo_data.py --sp500 --years 1 --run-screening
```
### 2. Load Specific Symbols
```bash
# Load individual stocks
python scripts/load_tiingo_data.py --symbols AAPL,MSFT,GOOGL,AMZN,TSLA --years 3
# Load with custom date range
python scripts/load_tiingo_data.py --symbols AAPL,MSFT --start-date 2020-01-01 --end-date 2023-12-31
```
### 3. Load from File
```bash
# Create a symbol file
echo -e "AAPL\nMSFT\nGOOGL\nAMZN\nTSLA" > my_symbols.txt
# Load from file
python scripts/load_tiingo_data.py --file my_symbols.txt --calculate-indicators --run-screening
```
### 4. Resume Interrupted Load
```bash
# If a load was interrupted, resume from checkpoint
python scripts/load_tiingo_data.py --resume --checkpoint-file load_progress.json
```
### 5. Performance-Optimized Load
```bash
# High-performance loading with larger batches and more concurrency
python scripts/load_tiingo_data.py --sp500 --batch-size 100 --max-concurrent 10 --no-checkpoint
```
### 6. All Supported Tickers
```bash
# Load all Tiingo-supported symbols (this will take a while!)
python scripts/load_tiingo_data.py --supported --batch-size 50 --max-concurrent 8
```
## Command Line Options
### Symbol Selection (Required - choose one)
- `--symbols AAPL,MSFT,GOOGL` - Comma-separated list of symbols
- `--file symbols.txt` - Load symbols from file (one per line or comma-separated)
- `--sp500` - Load S&P 500 symbols (top 100 most liquid)
- `--sp500-full` - Load full S&P 500 (500 symbols)
- `--supported` - Load all Tiingo-supported symbols
- `--resume` - Resume from checkpoint file
### Date Range Options
- `--years 2` - Number of years of historical data (default: 2)
- `--start-date 2020-01-01` - Custom start date (YYYY-MM-DD)
- `--end-date 2023-12-31` - Custom end date (YYYY-MM-DD, default: today)
### Processing Options
- `--calculate-indicators` - Calculate technical indicators (default: True)
- `--no-indicators` - Skip technical indicator calculations
- `--run-screening` - Run screening algorithms after data loading
### Performance Options
- `--batch-size 50` - Batch size for processing (default: 50)
- `--max-concurrent 5` - Maximum concurrent requests (default: 5)
### Database Options
- `--create-tables` - Create database tables if they don't exist
- `--database-url` - Override database URL
### Progress Tracking
- `--checkpoint-file load_progress.json` - Checkpoint file location
- `--no-checkpoint` - Disable checkpoint saving
## Configuration
The loader can be customized through the `tiingo_config.py` file:
```python
from scripts.tiingo_config import TiingoConfig, get_config_for_environment
# Get environment-specific config
config = get_config_for_environment('production')
# Customize settings
config.max_concurrent_requests = 10
config.default_batch_size = 100
config.maverick_min_momentum_score = 80.0
```
### Available Configurations
- **Rate Limiting**: Requests per hour, retry settings
- **Technical Indicators**: Periods for RSI, SMA, EMA, MACD, etc.
- **Screening Criteria**: Minimum momentum scores, volume thresholds
- **Database Settings**: Batch sizes, connection pooling
- **Symbol Lists**: Predefined lists for different strategies
## Technical Indicators Calculated
The loader calculates 50+ technical indicators including:
### Trend Indicators
- **SMA**: 20, 50, 150, 200-period Simple Moving Averages
- **EMA**: 21-period Exponential Moving Average
- **ADX**: Average Directional Index (14-period)
### Momentum Indicators
- **RSI**: Relative Strength Index (14-period)
- **MACD**: Moving Average Convergence Divergence (12,26,9)
- **Stochastic**: Stochastic Oscillator (14,3,3)
- **Momentum Score**: Relative Strength vs Market
### Volatility Indicators
- **ATR**: Average True Range (14-period)
- **Bollinger Bands**: 20-period with 2 standard deviations
- **ADR**: Average Daily Range percentage
### Volume Indicators
- **Volume SMA**: 30-period volume average
- **Volume Ratio**: Current vs average volume
- **VWAP**: Volume Weighted Average Price
### Custom Indicators
- **Momentum**: 10 and 20-period price momentum
- **BB Squeeze**: Bollinger Band squeeze detection
- **Price Position**: Position relative to moving averages
## Screening Algorithms
### Maverick Momentum Screen
Identifies stocks with strong upward momentum:
- Price above 21-day EMA
- EMA-21 above SMA-50
- SMA-50 above SMA-200
- Relative Strength Rating > 70
- Minimum volume thresholds
### Bear Market Screen
Identifies stocks in downtrends:
- Price below 21-day EMA
- EMA-21 below SMA-50
- Relative Strength Rating < 30
- High volume on down moves
### Supply/Demand Breakout Screen
Identifies accumulation patterns:
- Price above SMA-50 and SMA-200
- Strong relative strength (RS > 60)
- Institutional accumulation signals
- Volume dry-up followed by expansion
## Progress Tracking & Resume
The loader automatically saves progress to a checkpoint file:
```json
{
"timestamp": "2024-01-15T10:30:00",
"total_symbols": 100,
"processed_symbols": 75,
"successful_symbols": 73,
"completed_symbols": ["AAPL", "MSFT", ...],
"failed_symbols": ["BADTICKER", "ANOTHERBAD"],
"errors": [...],
"elapsed_time": 3600
}
```
To resume an interrupted load:
```bash
python scripts/load_tiingo_data.py --resume --checkpoint-file load_progress.json
```
## Error Handling
### Automatic Retry Logic
- **Exponential Backoff**: 1s, 2s, 4s delays between retries
- **Rate Limit Handling**: Automatic delays when rate limited
- **Connection Errors**: Automatic retry with timeout handling
### Error Reporting
- **Detailed Logging**: All errors logged with context
- **Error Tracking**: Failed symbols tracked in checkpoint
- **Graceful Degradation**: Continue processing other symbols on individual failures
## Performance Optimization
### Database Optimizations
- **Bulk Inserts**: Use PostgreSQL's UPSERT for efficiency
- **Connection Pooling**: Reuse database connections
- **Batch Processing**: Process multiple symbols together
- **Index Creation**: Automatically create performance indexes
### Memory Management
- **Streaming Processing**: Process data in chunks to minimize memory usage
- **Garbage Collection**: Explicit cleanup of large DataFrames
- **Connection Limits**: Prevent connection exhaustion
### Monitoring
```bash
# Monitor progress in real-time
tail -f tiingo_data_loader.log
# Check database stats
python scripts/load_tiingo_data.py --database-stats
# Monitor system resources
htop # CPU and memory usage
iotop # Disk I/O usage
```
## Examples and Testing
### Interactive Examples
```bash
# Run interactive examples
python scripts/load_example.py
```
The example script provides:
1. Load sample stocks (5 symbols)
2. Load sector stocks (technology)
3. Resume interrupted load demonstration
4. Database statistics viewer
### Testing Different Configurations
```bash
# Test with small dataset
python scripts/load_tiingo_data.py --symbols AAPL,MSFT --years 1 --batch-size 10
# Test screening only (no new data)
python scripts/load_tiingo_data.py --symbols AAPL --years 0.1 --run-screening
# Test resume functionality
python scripts/load_tiingo_data.py --symbols AAPL,MSFT,GOOGL,AMZN,TSLA --batch-size 2
# Interrupt with Ctrl+C, then resume:
python scripts/load_tiingo_data.py --resume
```
## Troubleshooting
### Common Issues
#### 1. API Token Issues
```bash
# Check if token is set
echo $TIINGO_API_TOKEN
# Test API access
curl -H "Authorization: Token $TIINGO_API_TOKEN" \
"https://api.tiingo.com/tiingo/daily/AAPL"
```
#### 2. Database Connection Issues
```bash
# Check database URL
echo $DATABASE_URL
# Test database connection
python -c "from maverick_mcp.data.models import SessionLocal; print('DB OK')"
```
#### 3. Rate Limiting
If you're getting rate limited frequently:
- Reduce `--max-concurrent` (default: 5)
- Increase `--batch-size` to reduce total requests
- Consider upgrading to Tiingo's paid plan
#### 4. Memory Issues
For large loads:
- Reduce `--batch-size`
- Reduce `--max-concurrent`
- Monitor memory usage with `htop`
#### 5. Checkpoint Corruption
```bash
# Remove corrupted checkpoint
rm load_progress.json
# Start fresh
python scripts/load_tiingo_data.py --symbols AAPL,MSFT --no-checkpoint
```
### Performance Benchmarks
Typical performance on modern hardware:
- **Small Load (10 symbols, 1 year)**: 2-3 minutes
- **Medium Load (100 symbols, 2 years)**: 15-20 minutes
- **Large Load (500 symbols, 2 years)**: 1-2 hours
- **Full Load (3000+ symbols, 2 years)**: 6-12 hours
## Integration with Maverick-MCP
The loaded data integrates seamlessly with Maverick-MCP:
### API Endpoints
The data is immediately available through Maverick-MCP's API endpoints:
- `/api/v1/stocks` - Stock information
- `/api/v1/prices/{symbol}` - Price data
- `/api/v1/technical/{symbol}` - Technical indicators
- `/api/v1/screening/maverick` - Maverick stock screen results
### MCP Tools
Use the loaded data in MCP tools:
- `get_stock_analysis` - Comprehensive stock analysis
- `run_screening` - Run custom screens
- `get_technical_indicators` - Retrieve calculated indicators
- `portfolio_analysis` - Analyze portfolio performance
## Advanced Usage
### Custom Symbol Lists
Create sector-specific or strategy-specific symbol files:
```bash
# Create growth stock list
cat > growth_stocks.txt << EOF
TSLA
NVDA
AMZN
GOOGL
META
NFLX
CRM
ADBE
EOF
# Load growth stocks
python scripts/load_tiingo_data.py --file growth_stocks.txt --run-screening
```
### Automated Scheduling
Set up daily data updates with cron:
```bash
# Add to crontab (crontab -e)
# Daily update at 6 PM EST (after market close)
0 18 * * 1-5 cd /path/to/maverick-mcp && python scripts/load_tiingo_data.py --sp500 --years 0.1 --run-screening >> /var/log/tiingo_updates.log 2>&1
# Weekly full reload on weekends
0 2 * * 6 cd /path/to/maverick-mcp && python scripts/load_tiingo_data.py --sp500 --years 2 --calculate-indicators --run-screening >> /var/log/tiingo_weekly.log 2>&1
```
### Integration with CI/CD
```yaml
# GitHub Actions workflow
name: Update Market Data
on:
schedule:
- cron: '0 18 * * 1-5' # 6 PM EST weekdays
jobs:
update-data:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Setup Python
uses: actions/setup-python@v2
with:
python-version: '3.12'
- name: Install dependencies
run: pip install -r requirements.txt
- name: Update market data
env:
TIINGO_API_TOKEN: ${{ secrets.TIINGO_API_TOKEN }}
DATABASE_URL: ${{ secrets.DATABASE_URL }}
run: |
python scripts/load_tiingo_data.py --sp500 --years 0.1 --run-screening
```
## Support and Contributing
### Getting Help
- **Documentation**: Check this README and inline code comments
- **Logging**: Enable debug logging for detailed troubleshooting
- **Examples**: Use the example script to understand usage patterns
### Contributing
To add new features or fix bugs:
1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Submit a pull request
### Feature Requests
Common requested features:
- Additional data providers (Alpha Vantage, Yahoo Finance)
- More technical indicators
- Custom screening algorithms
- Real-time data streaming
- Portfolio backtesting integration
```
--------------------------------------------------------------------------------
/maverick_mcp/tests/test_in_memory_server.py:
--------------------------------------------------------------------------------
```python
"""
In-memory tests for Maverick-MCP server using FastMCP patterns.
These tests demonstrate how to test the server without external processes
or network calls, using FastMCP's in-memory transport capabilities.
"""
import asyncio
from datetime import datetime, timedelta
from typing import Any
from unittest.mock import Mock, patch
import pytest
from fastmcp import Client
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from maverick_mcp.api.server import mcp
from maverick_mcp.data.models import Base, PriceCache, Stock
@pytest.fixture
def mock_redis():
"""Mock Redis client for testing."""
with patch("maverick_mcp.data.cache._get_redis_client") as mock_redis:
# Mock Redis client
redis_instance = Mock()
redis_instance.get.return_value = None
redis_instance.set.return_value = True
redis_instance.delete.return_value = True
redis_instance.ping.return_value = True
mock_redis.return_value = redis_instance
yield redis_instance
@pytest.fixture
def test_db():
"""Create an in-memory SQLite database for testing."""
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
# Add some test data
with Session(engine) as session:
# Add test stocks
aapl = Stock(
ticker_symbol="AAPL",
company_name="Apple Inc.",
sector="Technology",
industry="Consumer Electronics",
)
msft = Stock(
ticker_symbol="MSFT",
company_name="Microsoft Corp.",
sector="Technology",
industry="Software",
)
session.add_all([aapl, msft])
session.commit()
# Add test price data
base_date = datetime.now() - timedelta(days=30)
for i in range(30):
date = base_date + timedelta(days=i)
session.add(
PriceCache(
stock_id=aapl.stock_id,
date=date,
open_price=150.0 + i,
high_price=152.0 + i,
low_price=149.0 + i,
close_price=151.0 + i,
volume=1000000 + i * 10000,
)
)
session.commit()
# Patch the database connection
with patch("maverick_mcp.data.models.engine", engine):
with patch("maverick_mcp.data.models.SessionLocal", lambda: Session(engine)):
yield engine
class TestInMemoryServer:
"""Test suite for in-memory server operations."""
@pytest.mark.asyncio
async def test_server_health(self, test_db, mock_redis):
"""Test the health endpoint returns correct status."""
async with Client(mcp) as client:
result = await client.read_resource("health://")
# Result is a list of content items
assert len(result) > 0
assert result[0].text is not None
health_data = eval(result[0].text) # Convert string representation to dict
# In testing environment, status might be degraded due to mocked services
assert health_data["status"] in ["ok", "degraded"]
assert "version" in health_data
assert "components" in health_data
# Check available components
components = health_data["components"]
# Redis should be healthy (mocked)
if "redis" in components:
assert components["redis"]["status"] == "healthy"
# Database status can be error in test environment due to SQLite pool differences
if "database" in components:
assert components["database"]["status"] in [
"healthy",
"degraded",
"unhealthy",
"error",
]
@pytest.mark.asyncio
async def test_fetch_stock_data(self, test_db, mock_redis):
"""Test fetching stock data from the database."""
async with Client(mcp) as client:
result = await client.call_tool(
"/data_fetch_stock_data",
{
"request": {
"ticker": "AAPL",
"start_date": "2024-01-01",
"end_date": "2024-01-31",
}
},
)
assert len(result) > 0
assert result[0].text is not None
# Result should contain stock data
data = eval(result[0].text)
assert data["ticker"] == "AAPL"
assert "columns" in data
assert "Open" in data["columns"]
assert "Close" in data["columns"]
assert data["record_count"] > 0
@pytest.mark.asyncio
async def test_rsi_analysis(self, test_db, mock_redis):
"""Test RSI technical analysis calculation."""
async with Client(mcp) as client:
result = await client.call_tool(
"/technical_get_rsi_analysis", {"ticker": "AAPL", "period": 14}
)
assert len(result) > 0
assert result[0].text is not None
# Should contain RSI data
data = eval(result[0].text)
assert "analysis" in data
assert "ticker" in data
assert data["ticker"] == "AAPL"
@pytest.mark.asyncio
async def test_batch_stock_data(self, test_db, mock_redis):
"""Test batch fetching of multiple stocks."""
async with Client(mcp) as client:
result = await client.call_tool(
"/data_fetch_stock_data_batch",
{
"request": {
"tickers": ["AAPL", "MSFT"],
"start_date": "2024-01-01",
"end_date": "2024-01-31",
}
},
)
assert len(result) > 0
assert result[0].text is not None
data = eval(result[0].text)
assert "results" in data
assert "AAPL" in data["results"]
assert "MSFT" in data["results"]
assert data["success_count"] == 2
@pytest.mark.asyncio
async def test_invalid_ticker(self, test_db, mock_redis):
"""Test handling of invalid ticker symbols."""
async with Client(mcp) as client:
# Invalid ticker should return an error, not raise an exception
result = await client.call_tool(
"/data_fetch_stock_data",
{
"request": {
"ticker": "INVALID123", # Invalid format
"start_date": "2024-01-01",
"end_date": "2024-01-31",
}
},
)
# Should return empty data for invalid ticker
assert len(result) > 0
assert result[0].text is not None
data = eval(result[0].text)
# Invalid ticker returns empty data
assert data["record_count"] == 0
assert len(data["data"]) == 0
@pytest.mark.asyncio
async def test_date_validation(self, test_db, mock_redis):
"""Test date range validation."""
async with Client(mcp) as client:
with pytest.raises(Exception) as exc_info:
await client.call_tool(
"/data_fetch_stock_data",
{
"request": {
"ticker": "AAPL",
"start_date": "2024-01-31",
"end_date": "2024-01-01", # End before start
}
},
)
# Should fail with validation error
assert (
"error" in str(exc_info.value).lower()
or "validation" in str(exc_info.value).lower()
)
@pytest.mark.asyncio
async def test_concurrent_requests(self, test_db, mock_redis):
"""Test handling multiple concurrent requests."""
async with Client(mcp) as client:
# Create multiple concurrent tasks
tasks = [
client.call_tool(
"/data_fetch_stock_data",
{
"request": {
"ticker": "AAPL",
"start_date": "2024-01-01",
"end_date": "2024-01-31",
}
},
)
for _ in range(5)
]
# All should complete successfully
results = await asyncio.gather(*tasks)
assert len(results) == 5
for result in results:
assert len(result) > 0
assert result[0].text is not None
data = eval(result[0].text)
assert data["ticker"] == "AAPL"
class TestResourceManagement:
"""Test resource management and cleanup."""
@pytest.mark.asyncio
async def test_list_resources(self, test_db, mock_redis):
"""Test listing available resources."""
async with Client(mcp) as client:
resources = await client.list_resources()
# In the current implementation, resources may be empty or have different URIs
# Just check that the call succeeds
assert isinstance(resources, list)
@pytest.mark.asyncio
async def test_read_resource(self, test_db, mock_redis):
"""Test reading a specific resource."""
async with Client(mcp) as client:
result = await client.read_resource("health://")
assert len(result) > 0
assert result[0].text is not None
# Should contain cache status information
assert (
"redis" in result[0].text.lower() or "memory" in result[0].text.lower()
)
class TestErrorHandling:
"""Test error handling and edge cases."""
@pytest.mark.asyncio
async def test_database_error_handling(self, mock_redis):
"""Test graceful handling of database errors."""
# No test_db fixture, so database should fail
with patch(
"maverick_mcp.data.models.SessionLocal", side_effect=Exception("DB Error")
):
async with Client(mcp) as client:
result = await client.read_resource("health://")
assert len(result) > 0
health_data = eval(result[0].text)
# Database should show an error
assert health_data["status"] in ["degraded", "unhealthy"]
assert "components" in health_data
@pytest.mark.asyncio
async def test_cache_fallback(self, test_db):
"""Test fallback to in-memory cache when Redis is unavailable."""
# No mock_redis fixture, should fall back to memory
with patch(
"maverick_mcp.data.cache.redis.Redis", side_effect=Exception("Redis Error")
):
async with Client(mcp) as client:
result = await client.read_resource("health://")
assert len(result) > 0
health_data = eval(result[0].text)
# Cache should fall back to memory
assert "components" in health_data
if "cache" in health_data["components"]:
assert health_data["components"]["cache"]["type"] == "memory"
class TestPerformanceMetrics:
"""Test performance monitoring and metrics."""
@pytest.mark.asyncio
async def test_query_performance_tracking(self, test_db, mock_redis):
"""Test that query performance is tracked."""
# Skip this test as health_monitor is not available
pytest.skip("health_monitor not available in current implementation")
# Utility functions for testing
def create_test_stock_data(symbol: str, days: int = 30) -> dict[str, Any]:
"""Create test stock data for a given symbol."""
data: dict[str, Any] = {"symbol": symbol, "prices": []}
base_date = datetime.now() - timedelta(days=days)
base_price = 100.0
for i in range(days):
date = base_date + timedelta(days=i)
price = base_price + (i * 0.5) # Gradual increase
data["prices"].append(
{
"date": date.isoformat(),
"open": price,
"high": price + 1,
"low": price - 1,
"close": price + 0.5,
"volume": 1000000,
}
)
return data
@pytest.mark.asyncio
async def test_with_mock_data_provider(test_db, mock_redis):
"""Test with mocked external data provider."""
test_data = create_test_stock_data("TSLA", 30)
with patch("yfinance.download") as mock_yf:
# Mock yfinance response
mock_df = Mock()
mock_df.empty = False
mock_df.to_dict.return_value = test_data["prices"]
mock_yf.return_value = mock_df
async with Client(mcp) as client:
result = await client.call_tool(
"/data_fetch_stock_data",
{
"request": {
"ticker": "TSLA",
"start_date": "2024-01-01",
"end_date": "2024-01-31",
}
},
)
assert len(result) > 0
assert result[0].text is not None
assert "TSLA" in result[0].text
if __name__ == "__main__":
# Run tests directly
pytest.main([__file__, "-v"])
```
--------------------------------------------------------------------------------
/maverick_mcp/utils/cache_warmer.py:
--------------------------------------------------------------------------------
```python
"""
Cache warming utilities for pre-loading commonly used data.
Improves performance by pre-fetching and caching frequently accessed data.
"""
import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from typing import Any
from maverick_mcp.data.cache import (
CacheManager,
ensure_timezone_naive,
generate_cache_key,
get_cache_stats,
)
from maverick_mcp.providers.stock_data import EnhancedStockDataProvider
from maverick_mcp.utils.yfinance_pool import get_yfinance_pool
logger = logging.getLogger(__name__)
class CacheWarmer:
"""Pre-loads frequently accessed data into cache for improved performance."""
def __init__(
self,
data_provider: EnhancedStockDataProvider | None = None,
cache_manager: CacheManager | None = None,
max_workers: int = 5,
):
"""Initialize cache warmer.
Args:
data_provider: Stock data provider instance
cache_manager: Cache manager instance
max_workers: Maximum number of parallel workers
"""
self.data_provider = data_provider or EnhancedStockDataProvider()
self.cache = cache_manager or CacheManager()
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self._yf_pool = get_yfinance_pool()
# Common symbols to warm up
self.popular_symbols = [
"SPY",
"QQQ",
"AAPL",
"MSFT",
"GOOGL",
"AMZN",
"NVDA",
"META",
"TSLA",
"BRK-B",
"JPM",
"V",
"JNJ",
"WMT",
"PG",
"UNH",
"HD",
"MA",
"DIS",
"BAC",
"XOM",
"PFE",
"ABBV",
"KO",
"CVX",
"PEP",
"TMO",
"AVGO",
"COST",
"MRK",
"VZ",
"ADBE",
"CMCSA",
"NKE",
]
# Common date ranges
self.common_periods = [
("1d", 1), # Yesterday
("5d", 5), # Last week
("1mo", 30), # Last month
("3mo", 90), # Last 3 months
("1y", 365), # Last year
]
async def warm_popular_stocks(self, symbols: list[str] | None = None):
"""Pre-load data for popular stocks.
Args:
symbols: List of symbols to warm up (uses default popular list if None)
"""
symbols = symbols or self.popular_symbols
logger.info(f"Warming cache for {len(symbols)} popular stocks")
# Warm up in parallel batches
batch_size = 5
for i in range(0, len(symbols), batch_size):
batch = symbols[i : i + batch_size]
await self._warm_batch(batch)
logger.info("Popular stocks cache warming completed")
async def _warm_batch(self, symbols: list[str]):
"""Warm cache for a batch of symbols."""
tasks = []
for symbol in symbols:
# Warm different time periods
for period_name, days in self.common_periods:
task = asyncio.create_task(
self._warm_symbol_period(symbol, period_name, days)
)
tasks.append(task)
# Wait for all tasks with timeout
try:
await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True), timeout=30
)
except TimeoutError:
logger.warning(f"Timeout warming batch: {symbols}")
async def _warm_symbol_period(self, symbol: str, period: str, days: int):
"""Warm cache for a specific symbol and period."""
try:
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
# Generate versioned cache key
cache_key = generate_cache_key(
"backtest_data",
symbol=symbol,
start_date=start_date,
end_date=end_date,
interval="1d",
)
# Check if already cached
if await self.cache.exists(cache_key):
logger.debug(f"Cache already warm for {symbol} ({period})")
return
# Fetch data using the data provider
data = await asyncio.get_event_loop().run_in_executor(
self.executor,
self.data_provider.get_stock_data,
symbol,
start_date,
end_date,
None, # period
"1d", # interval
)
if data is not None and not data.empty:
# Normalize column names and ensure timezone-naive
data.columns = [col.lower() for col in data.columns]
data = ensure_timezone_naive(data)
# Cache with adaptive TTL based on data age
ttl = 86400 if days > 7 else 3600 # 24h for older data, 1h for recent
await self.cache.set(cache_key, data, ttl=ttl)
logger.debug(f"Warmed cache for {symbol} ({period}) - {len(data)} rows")
except Exception as e:
logger.warning(f"Failed to warm cache for {symbol} ({period}): {e}")
async def warm_screening_data(self):
"""Pre-load screening recommendations."""
logger.info("Warming screening data cache")
try:
# Warm maverick recommendations
await asyncio.get_event_loop().run_in_executor(
self.executor,
self.data_provider.get_maverick_recommendations,
20, # limit
None, # min_score
)
# Warm bear recommendations
await asyncio.get_event_loop().run_in_executor(
self.executor,
self.data_provider.get_maverick_bear_recommendations,
20,
None,
)
# Warm supply/demand breakouts
await asyncio.get_event_loop().run_in_executor(
self.executor,
self.data_provider.get_supply_demand_breakout_recommendations,
20,
None,
)
logger.info("Screening data cache warming completed")
except Exception as e:
logger.error(f"Failed to warm screening cache: {e}")
async def warm_technical_indicators(self, symbols: list[str] | None = None):
"""Pre-calculate and cache technical indicators for symbols.
Args:
symbols: List of symbols (uses top 10 popular if None)
"""
symbols = symbols or self.popular_symbols[:10]
logger.info(f"Warming technical indicators for {len(symbols)} stocks")
tasks = []
for symbol in symbols:
task = asyncio.create_task(self._warm_symbol_technicals(symbol))
tasks.append(task)
try:
await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True), timeout=60
)
except TimeoutError:
logger.warning("Timeout warming technical indicators")
logger.info("Technical indicators cache warming completed")
async def _warm_symbol_technicals(self, symbol: str):
"""Warm technical indicator cache for a symbol."""
try:
# Get recent data
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=100)).strftime("%Y-%m-%d")
# Common technical indicator cache keys
indicators = [
("sma", [20, 50, 200]),
("ema", [12, 26]),
("rsi", [14]),
("macd", [12, 26, 9]),
("bb", [20, 2]),
]
for indicator, params in indicators:
for param in params:
cache_key = f"technical:{symbol}:{indicator}:{param}:{start_date}:{end_date}"
if await self.cache.exists(cache_key):
continue
# Note: Actual technical calculation would go here
# For now, we're just warming the stock data cache
logger.debug(
f"Would warm {indicator} for {symbol} with param {param}"
)
except Exception as e:
logger.warning(f"Failed to warm technicals for {symbol}: {e}")
async def run_full_warmup(self, report_stats: bool = True):
"""Run complete cache warming routine."""
logger.info("Starting full cache warmup")
# Get initial cache stats
initial_stats = get_cache_stats() if report_stats else None
start_time = asyncio.get_event_loop().time()
# Run all warming tasks
results = await asyncio.gather(
self.warm_popular_stocks(),
self.warm_screening_data(),
self.warm_technical_indicators(),
return_exceptions=True,
)
end_time = asyncio.get_event_loop().time()
# Report results and performance
successful_tasks = sum(1 for r in results if not isinstance(r, Exception))
failed_tasks = len(results) - successful_tasks
logger.info(
f"Full cache warmup completed in {end_time - start_time:.2f}s - "
f"{successful_tasks} successful, {failed_tasks} failed"
)
if report_stats and initial_stats:
final_stats = get_cache_stats()
new_items = final_stats["sets"] - initial_stats["sets"]
hit_rate_change = (
final_stats["hit_rate_percent"] - initial_stats["hit_rate_percent"]
)
logger.info(
f"Cache warmup results: +{new_items} items cached, "
f"hit rate change: {hit_rate_change:+.1f}%"
)
async def schedule_periodic_warmup(self, interval_minutes: int = 30):
"""Schedule periodic cache warming.
Args:
interval_minutes: Minutes between warmup runs
"""
logger.info(f"Starting periodic cache warmup every {interval_minutes} minutes")
while True:
try:
await self.run_full_warmup()
except Exception as e:
logger.error(f"Error in periodic warmup: {e}")
# Wait for next cycle
await asyncio.sleep(interval_minutes * 60)
async def benchmark_cache_performance(
self, symbols: list[str] | None = None
) -> dict[str, Any]:
"""Benchmark cache performance for analysis.
Args:
symbols: List of symbols to test (uses top 5 if None)
Returns:
Dictionary with benchmark results
"""
symbols = symbols or self.popular_symbols[:5]
logger.info(f"Benchmarking cache performance with {len(symbols)} symbols")
# Test data retrieval performance
import time
start_time = time.time()
cache_hits = 0
cache_misses = 0
for symbol in symbols:
for _period_name, days in self.common_periods:
end_date = datetime.now().strftime("%Y-%m-%d")
start_date = (datetime.now() - timedelta(days=days)).strftime(
"%Y-%m-%d"
)
cache_key = generate_cache_key(
"backtest_data",
symbol=symbol,
start_date=start_date,
end_date=end_date,
interval="1d",
)
cached_data = await self.cache.get(cache_key)
if cached_data is not None:
cache_hits += 1
else:
cache_misses += 1
end_time = time.time()
# Calculate metrics
total_requests = cache_hits + cache_misses
hit_rate = (cache_hits / total_requests * 100) if total_requests > 0 else 0
avg_request_time = (
(end_time - start_time) / total_requests if total_requests > 0 else 0
)
# Get current cache stats
cache_stats = get_cache_stats()
benchmark_results = {
"symbols_tested": len(symbols),
"total_requests": total_requests,
"cache_hits": cache_hits,
"cache_misses": cache_misses,
"hit_rate_percent": round(hit_rate, 2),
"avg_request_time_ms": round(avg_request_time * 1000, 2),
"total_time_seconds": round(end_time - start_time, 2),
"cache_stats": cache_stats,
}
logger.info(
f"Benchmark completed: {hit_rate:.1f}% hit rate, "
f"{avg_request_time * 1000:.1f}ms avg request time"
)
return benchmark_results
def shutdown(self):
"""Clean up resources."""
self.executor.shutdown(wait=False)
logger.info("Cache warmer shutdown")
async def warm_cache_on_startup():
"""Convenience function to warm cache on application startup."""
warmer = CacheWarmer()
try:
# Only warm the most critical data on startup
await warmer.warm_popular_stocks(
["SPY", "QQQ", "AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "TSLA"]
)
await warmer.warm_screening_data()
finally:
warmer.shutdown()
if __name__ == "__main__":
# Example usage
async def main():
warmer = CacheWarmer()
try:
await warmer.run_full_warmup()
finally:
warmer.shutdown()
asyncio.run(main())
```
--------------------------------------------------------------------------------
/docs/exa_research_testing_strategy.md:
--------------------------------------------------------------------------------
```markdown
# ExaSearch Research Integration Testing Strategy
This document outlines the comprehensive testing strategy for validating the ExaSearch integration with the MaverickMCP research agent architecture.
## Overview
The testing strategy covers all aspects of the research system with ExaSearch provider:
- **DeepResearchAgent** orchestration with ExaSearch integration
- **Specialized Subagents** (Fundamental, Technical, Sentiment, Competitive)
- **Parallel Research Orchestration** and task distribution
- **Timeout Handling** and circuit breaker patterns
- **MCP Tool Integration** via research router endpoints
- **Performance Benchmarking** across research depths and configurations
## Test Architecture
### Test Categories
1. **Unit Tests** (`pytest -m unit`)
- Individual component testing in isolation
- Mock external dependencies
- Fast execution (< 30 seconds total)
- No external API calls
2. **Integration Tests** (`pytest -m integration`)
- End-to-end workflow testing
- Real ExaSearch API integration
- Multi-component interaction validation
- Requires `EXA_API_KEY` environment variable
3. **Performance Tests** (`pytest -m slow`)
- Benchmark different research depths
- Parallel vs sequential execution comparison
- Memory usage and timeout resilience
- Longer execution times (2-5 minutes)
4. **Benchmark Suite** (`scripts/benchmark_exa_research.py`)
- Comprehensive performance analysis
- Cross-configuration comparison
- Detailed metrics and reporting
- Production-ready performance validation
## Test Files and Structure
```
tests/
├── test_exa_research_integration.py # Main comprehensive test suite
└── conftest.py # Shared fixtures and configuration
scripts/
├── run_exa_tests.py # Test runner utility
└── benchmark_exa_research.py # Performance benchmark suite
docs/
└── exa_research_testing_strategy.md # This document
```
## Key Test Components
### 1. ExaSearchProvider Tests
**Coverage:**
- Provider initialization with/without API key
- Adaptive timeout calculation for different query complexities
- Failure recording and health status management
- Successful search execution with realistic mock responses
- Timeout handling and error recovery
- Circuit breaker integration
**Key Test Methods:**
```python
test_exa_provider_initialization()
test_timeout_calculation()
test_failure_recording_and_health_status()
test_exa_search_success()
test_exa_search_timeout()
test_exa_search_unhealthy_provider()
```
### 2. DeepResearchAgent Tests
**Coverage:**
- Agent initialization with ExaSearch provider
- Research execution with different depths (basic, standard, comprehensive, exhaustive)
- Timeout budget allocation and management
- Error handling when no providers are available
- Complete research workflow from query to results
**Key Test Methods:**
```python
test_agent_initialization_with_exa()
test_research_comprehensive_success()
test_research_comprehensive_no_providers()
test_research_depth_levels()
```
### 3. Specialized Subagent Tests
**Coverage:**
- All 4 subagent types: Fundamental, Technical, Sentiment, Competitive
- Query generation for each specialization
- Results processing and analysis
- Focus area validation
- Cross-subagent consistency
**Key Test Methods:**
```python
test_fundamental_research_agent()
test_technical_research_agent()
test_sentiment_research_agent()
test_competitive_research_agent()
```
### 4. Parallel Research Orchestration Tests
**Coverage:**
- ParallelResearchOrchestrator initialization and configuration
- Task preparation and prioritization
- Successful parallel execution with multiple tasks
- Failure handling and partial success scenarios
- Circuit breaker integration
- Performance efficiency measurement
**Key Test Methods:**
```python
test_orchestrator_initialization()
test_parallel_execution_success()
test_parallel_execution_with_failures()
test_circuit_breaker_integration()
```
### 5. Task Distribution Engine Tests
**Coverage:**
- Topic relevance analysis for different task types
- Intelligent task distribution based on query content
- Priority assignment based on relevance scores
- Fallback mechanisms when no relevant tasks found
**Key Test Methods:**
```python
test_topic_relevance_analysis()
test_task_distribution_basic()
test_task_distribution_fallback()
test_task_priority_assignment()
```
### 6. Timeout and Circuit Breaker Tests
**Coverage:**
- Timeout budget allocation across research phases
- Provider health monitoring and recovery
- Research behavior during provider failures
- Graceful degradation strategies
**Key Test Methods:**
```python
test_timeout_budget_allocation()
test_provider_health_monitoring()
test_research_with_provider_failures()
```
### 7. Performance Benchmark Tests
**Coverage:**
- Cross-depth performance comparison (basic → exhaustive)
- Parallel vs sequential execution efficiency
- Memory usage monitoring during parallel execution
- Scalability under load
**Key Test Methods:**
```python
test_research_depth_performance()
test_parallel_vs_sequential_performance()
test_memory_usage_monitoring()
```
### 8. MCP Integration Tests
**Coverage:**
- MCP tool endpoint validation
- Research router integration
- Request/response model validation
- Error handling in MCP context
**Key Test Methods:**
```python
test_comprehensive_research_mcp_tool()
test_research_without_exa_key()
test_research_request_validation()
test_get_research_agent_optimization()
```
### 9. Content Analysis Tests
**Coverage:**
- AI-powered content analysis functionality
- Fallback mechanisms when LLM analysis fails
- Batch content processing
- Sentiment and insight extraction
**Key Test Methods:**
```python
test_content_analysis_success()
test_content_analysis_fallback()
test_batch_content_analysis()
```
### 10. Error Handling and Edge Cases
**Coverage:**
- Empty search results handling
- Malformed API responses
- Network timeout recovery
- Concurrent request limits
- Memory constraints
**Key Test Methods:**
```python
test_empty_search_results()
test_malformed_search_response()
test_network_timeout_recovery()
test_concurrent_request_limits()
```
## Test Data and Fixtures
### Mock Data Factories
The test suite includes comprehensive mock data factories:
- **`mock_llm`**: Realistic LLM responses for different research phases
- **`mock_exa_client`**: ExaSearch API client with query-specific responses
- **`sample_research_tasks`**: Representative research tasks for parallel execution
- **`mock_settings`**: Configuration with ExaSearch integration enabled
### Realistic Test Scenarios
Test scenarios cover real-world usage patterns:
```python
test_queries = [
"AAPL stock financial analysis and investment outlook",
"Tesla market sentiment and competitive position",
"Microsoft earnings performance and growth prospects",
"tech sector analysis and market trends",
"artificial intelligence investment opportunities",
]
research_depths = ["basic", "standard", "comprehensive", "exhaustive"]
focus_areas = {
"fundamentals": ["earnings", "valuation", "financial_health"],
"technicals": ["chart_patterns", "technical_indicators", "price_action"],
"sentiment": ["market_sentiment", "analyst_ratings", "news_sentiment"],
"competitive": ["competitive_position", "market_share", "industry_trends"],
}
```
## Running Tests
### Quick Start
```bash
# Install dependencies
uv sync
# Set environment variable (for integration tests)
export EXA_API_KEY=your_exa_api_key
# Run unit tests (fast, no external dependencies)
python scripts/run_exa_tests.py --unit
# Run integration tests (requires EXA_API_KEY)
python scripts/run_exa_tests.py --integration
# Run all tests
python scripts/run_exa_tests.py --all
# Run quick test suite
python scripts/run_exa_tests.py --quick
# Run with coverage reporting
python scripts/run_exa_tests.py --coverage
```
### Direct pytest Commands
```bash
# Unit tests only
pytest tests/test_exa_research_integration.py -m unit -v
# Integration tests (requires API key)
pytest tests/test_exa_research_integration.py -m integration -v
# Performance tests
pytest tests/test_exa_research_integration.py -m slow -v
# All tests
pytest tests/test_exa_research_integration.py -v
```
### Performance Benchmarks
```bash
# Comprehensive benchmarks
python scripts/benchmark_exa_research.py
# Quick benchmarks (reduced test matrix)
python scripts/benchmark_exa_research.py --quick
# Specific depth testing
python scripts/benchmark_exa_research.py --depth basic --focus fundamentals
# Parallel execution analysis only
python scripts/benchmark_exa_research.py --depth standard --parallel --no-timeout
```
## Test Environment Setup
### Prerequisites
1. **Python 3.12+**: Core runtime requirement
2. **uv or pip**: Package management
3. **ExaSearch API Key**: For integration tests
```bash
export EXA_API_KEY=your_api_key_here
```
### Optional Dependencies
- **Redis**: For caching layer tests (optional)
- **PostgreSQL**: For database integration tests (optional)
- **psutil**: For memory usage monitoring in performance tests
### Environment Validation
```bash
# Validate environment setup
python scripts/run_exa_tests.py --validate
```
## Expected Test Results
### Performance Benchmarks
**Research Depth Performance Expectations:**
- **Basic**: < 15 seconds execution time
- **Standard**: 15-30 seconds execution time
- **Comprehensive**: 30-45 seconds execution time
- **Exhaustive**: 45-60 seconds execution time
**Parallel Execution Efficiency:**
- **Speedup**: 2-4x faster than sequential for 3+ subagents
- **Memory Usage**: < 100MB additional during parallel execution
- **Error Rate**: < 5% for timeout-related failures
### Success Criteria
**Unit Tests:**
- ✅ 100% pass rate expected
- ⚡ Complete in < 30 seconds
- 🔄 No external dependencies
**Integration Tests:**
- ✅ 95%+ pass rate (allowing for API variability)
- ⏱️ Complete in < 5 minutes
- 🔑 Requires valid EXA_API_KEY
**Performance Tests:**
- ✅ 90%+ pass rate (allowing for performance variability)
- ⏱️ Complete in < 10 minutes
- 📊 Generate detailed performance metrics
## Debugging and Troubleshooting
### Common Issues
1. **Missing EXA_API_KEY**
```
Error: Research functionality unavailable - Exa search provider not configured
Solution: Set EXA_API_KEY environment variable
```
2. **Import Errors**
```
ImportError: No module named 'exa_py'
Solution: Install dependencies with `uv sync` or `pip install -e .`
```
3. **Timeout Failures**
```
Error: Research operation timed out
Solution: Check network connection or reduce research scope
```
4. **Memory Issues**
```
Error: Memory usage exceeded limits
Solution: Reduce parallel agents or test data size
```
### Debug Mode
Enable detailed logging for debugging:
```bash
export PYTHONPATH=/path/to/maverick-mcp
export LOG_LEVEL=DEBUG
python scripts/run_exa_tests.py --unit --verbose
```
### Test Output Analysis
**Successful Test Run Example:**
```
🧪 Running ExaSearch Unit Tests
============================
test_exa_provider_initialization PASSED [ 5%]
test_timeout_calculation PASSED [ 10%]
test_failure_recording_and_health_status PASSED [ 15%]
...
✅ All tests completed successfully!
```
**Benchmark Report Example:**
```
📊 BENCHMARK SUMMARY REPORT
============================
📋 Total Tests: 25
✅ Successful: 23
❌ Failed: 2
⏱️ Total Time: 127.3s
📈 Performance Metrics:
Avg Execution Time: 18.45s
Min/Max Time: 8.21s / 45.67s
Avg Confidence Score: 0.78
Avg Sources Analyzed: 8.2
```
## Continuous Integration
### CI/CD Integration
The test suite is designed for CI/CD integration:
```yaml
# Example GitHub Actions workflow
- name: Run ExaSearch Tests
env:
EXA_API_KEY: ${{ secrets.EXA_API_KEY }}
run: |
python scripts/run_exa_tests.py --unit
python scripts/run_exa_tests.py --integration
python scripts/benchmark_exa_research.py --quick
```
### Test Markers for CI
Use pytest markers for selective testing:
```bash
# Fast tests only (for PR validation)
pytest -m "not slow and not external"
# Full test suite (for main branch)
pytest -m "not external" --maxfail=5
# External API tests (nightly/weekly)
pytest -m external
```
## Maintenance and Updates
### Adding New Tests
1. **Extend existing test classes** for related functionality
2. **Follow naming conventions**: `test_[component]_[scenario]`
3. **Use appropriate markers**: `@pytest.mark.unit`, `@pytest.mark.integration`
4. **Mock external dependencies** in unit tests
5. **Include error scenarios** and edge cases
### Updating Test Data
1. **Mock responses** should reflect real ExaSearch API responses
2. **Test queries** should cover different complexity levels
3. **Performance baselines** should be updated as system improves
4. **Error scenarios** should match actual failure modes
### Performance Regression Detection
1. **Baseline metrics** stored in benchmark results
2. **Automated comparison** against previous runs
3. **Alert thresholds** for performance degradation
4. **Regular benchmark execution** in CI/CD
## Conclusion
This comprehensive testing strategy ensures the ExaSearch integration is thoroughly validated across all dimensions:
- ✅ **Functional Correctness**: All components work as designed
- ⚡ **Performance Characteristics**: System meets timing requirements
- 🛡️ **Error Resilience**: Graceful handling of failures and edge cases
- 🔗 **Integration Quality**: Seamless operation across component boundaries
- 📊 **Monitoring Capability**: Detailed metrics and reporting for ongoing maintenance
The test suite provides confidence in the ExaSearch integration's reliability and performance for production deployment.
```
--------------------------------------------------------------------------------
/maverick_mcp/monitoring/health_monitor.py:
--------------------------------------------------------------------------------
```python
"""
Background health monitoring system.
This module provides background tasks for continuous health monitoring,
alerting, and automatic recovery actions for the backtesting system.
"""
import asyncio
import logging
import time
from datetime import UTC, datetime, timedelta
from typing import Any
from maverick_mcp.config.settings import get_settings
from maverick_mcp.monitoring.status_dashboard import get_status_dashboard
from maverick_mcp.utils.circuit_breaker import get_circuit_breaker_manager
logger = logging.getLogger(__name__)
settings = get_settings()
# Monitoring intervals (seconds)
HEALTH_CHECK_INTERVAL = 30
CIRCUIT_BREAKER_CHECK_INTERVAL = 60
RESOURCE_CHECK_INTERVAL = 45
ALERT_CHECK_INTERVAL = 120
# Alert thresholds
ALERT_THRESHOLDS = {
"consecutive_failures": 5,
"high_cpu_duration": 300, # 5 minutes
"high_memory_duration": 300, # 5 minutes
"circuit_breaker_open_duration": 180, # 3 minutes
}
class HealthMonitor:
"""Background health monitoring system."""
def __init__(self):
self.running = False
self.tasks = []
self.alerts_sent = {}
self.start_time = time.time()
self.health_history = []
self.dashboard = get_status_dashboard()
self.circuit_breaker_manager = get_circuit_breaker_manager()
async def start(self):
"""Start all background monitoring tasks."""
if self.running:
logger.warning("Health monitor is already running")
return
self.running = True
logger.info("Starting health monitoring system...")
# Initialize circuit breakers
self.circuit_breaker_manager.initialize()
# Start monitoring tasks
self.tasks = [
asyncio.create_task(self._health_check_loop()),
asyncio.create_task(self._circuit_breaker_monitor_loop()),
asyncio.create_task(self._resource_monitor_loop()),
asyncio.create_task(self._alert_processor_loop()),
]
logger.info(f"Started {len(self.tasks)} monitoring tasks")
async def stop(self):
"""Stop all monitoring tasks."""
if not self.running:
return
self.running = False
logger.info("Stopping health monitoring system...")
# Cancel all tasks
for task in self.tasks:
task.cancel()
# Wait for tasks to complete
if self.tasks:
await asyncio.gather(*self.tasks, return_exceptions=True)
self.tasks.clear()
logger.info("Health monitoring system stopped")
async def _health_check_loop(self):
"""Background loop for general health checks."""
logger.info("Started health check monitoring loop")
while self.running:
try:
await self._perform_health_check()
await asyncio.sleep(HEALTH_CHECK_INTERVAL)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error in health check loop: {e}")
await asyncio.sleep(HEALTH_CHECK_INTERVAL)
logger.info("Health check monitoring loop stopped")
async def _circuit_breaker_monitor_loop(self):
"""Background loop for circuit breaker monitoring."""
logger.info("Started circuit breaker monitoring loop")
while self.running:
try:
await self._check_circuit_breakers()
await asyncio.sleep(CIRCUIT_BREAKER_CHECK_INTERVAL)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error in circuit breaker monitoring loop: {e}")
await asyncio.sleep(CIRCUIT_BREAKER_CHECK_INTERVAL)
logger.info("Circuit breaker monitoring loop stopped")
async def _resource_monitor_loop(self):
"""Background loop for resource monitoring."""
logger.info("Started resource monitoring loop")
while self.running:
try:
await self._check_resource_usage()
await asyncio.sleep(RESOURCE_CHECK_INTERVAL)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error in resource monitoring loop: {e}")
await asyncio.sleep(RESOURCE_CHECK_INTERVAL)
logger.info("Resource monitoring loop stopped")
async def _alert_processor_loop(self):
"""Background loop for alert processing."""
logger.info("Started alert processing loop")
while self.running:
try:
await self._process_alerts()
await asyncio.sleep(ALERT_CHECK_INTERVAL)
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error in alert processing loop: {e}")
await asyncio.sleep(ALERT_CHECK_INTERVAL)
logger.info("Alert processing loop stopped")
async def _perform_health_check(self):
"""Perform comprehensive health check."""
try:
from maverick_mcp.api.routers.health_enhanced import (
_get_detailed_health_status,
)
health_status = await _get_detailed_health_status()
# Log health status
logger.debug(f"Health check: {health_status['status']}")
# Record health data
self._record_health_data(health_status)
# Check for issues requiring attention
await self._analyze_health_trends(health_status)
except Exception as e:
logger.error(f"Failed to perform health check: {e}")
async def _check_circuit_breakers(self):
"""Monitor circuit breaker states and perform recovery actions."""
try:
cb_status = self.circuit_breaker_manager.get_health_status()
for name, status in cb_status.items():
state = status.get("state", "unknown")
# Check for stuck open circuit breakers
if state == "open":
await self._handle_open_circuit_breaker(name, status)
# Check for high failure rates
metrics = status.get("metrics", {})
failure_rate = metrics.get("failure_rate", 0)
if failure_rate > 0.5: # 50% failure rate
await self._handle_high_failure_rate(name, failure_rate)
except Exception as e:
logger.error(f"Failed to check circuit breakers: {e}")
async def _check_resource_usage(self):
"""Monitor system resource usage."""
try:
from maverick_mcp.api.routers.health_enhanced import _get_resource_usage
resource_usage = _get_resource_usage()
# Check CPU usage
if resource_usage.cpu_percent > 80:
await self._handle_high_cpu_usage(resource_usage.cpu_percent)
# Check memory usage
if resource_usage.memory_percent > 85:
await self._handle_high_memory_usage(resource_usage.memory_percent)
# Check disk usage
if resource_usage.disk_percent > 90:
await self._handle_high_disk_usage(resource_usage.disk_percent)
except Exception as e:
logger.error(f"Failed to check resource usage: {e}")
async def _process_alerts(self):
"""Process and manage alerts."""
try:
dashboard_data = await self.dashboard.get_dashboard_data()
alerts = dashboard_data.get("alerts", [])
for alert in alerts:
await self._handle_alert(alert)
except Exception as e:
logger.error(f"Failed to process alerts: {e}")
def _record_health_data(self, health_status: dict[str, Any]):
"""Record health data for trend analysis."""
timestamp = datetime.now(UTC)
health_record = {
"timestamp": timestamp.isoformat(),
"overall_status": health_status.get("status", "unknown"),
"components_healthy": len(
[
c
for c in health_status.get("components", {}).values()
if c.status == "healthy"
]
),
"components_total": len(health_status.get("components", {})),
"resource_usage": health_status.get("resource_usage", {}),
}
self.health_history.append(health_record)
# Keep only last 24 hours of data
cutoff_time = timestamp - timedelta(hours=24)
self.health_history = [
record
for record in self.health_history
if datetime.fromisoformat(record["timestamp"].replace("Z", "+00:00"))
> cutoff_time
]
async def _analyze_health_trends(self, current_status: dict[str, Any]):
"""Analyze health trends and predict issues."""
if len(self.health_history) < 10:
return # Not enough data for trend analysis
# Analyze degradation trends
recent_records = self.health_history[-10:] # Last 10 records
unhealthy_trend = sum(
1
for record in recent_records
if record["overall_status"] in ["degraded", "unhealthy"]
)
if unhealthy_trend >= 7: # 70% of recent checks are problematic
logger.warning(
"Detected concerning health trend - system may need attention"
)
await self._trigger_maintenance_alert()
async def _handle_open_circuit_breaker(self, name: str, status: dict[str, Any]):
"""Handle circuit breaker that's been open too long."""
# Check if we've already alerted for this breaker recently
alert_key = f"cb_open_{name}"
last_alert = self.alerts_sent.get(alert_key)
if last_alert and (time.time() - last_alert) < 300: # 5 minutes
return
logger.warning(f"Circuit breaker '{name}' has been open - investigating")
# Record alert
self.alerts_sent[alert_key] = time.time()
# Could implement automatic recovery attempts here
# For now, just log the issue
async def _handle_high_failure_rate(self, name: str, failure_rate: float):
"""Handle high failure rate in circuit breaker."""
logger.warning(f"High failure rate detected for {name}: {failure_rate:.1%}")
async def _handle_high_cpu_usage(self, cpu_percent: float):
"""Handle sustained high CPU usage."""
alert_key = "high_cpu"
last_alert = self.alerts_sent.get(alert_key)
if last_alert and (time.time() - last_alert) < 600: # 10 minutes
return
logger.warning(f"High CPU usage detected: {cpu_percent:.1f}%")
self.alerts_sent[alert_key] = time.time()
async def _handle_high_memory_usage(self, memory_percent: float):
"""Handle sustained high memory usage."""
alert_key = "high_memory"
last_alert = self.alerts_sent.get(alert_key)
if last_alert and (time.time() - last_alert) < 600: # 10 minutes
return
logger.warning(f"High memory usage detected: {memory_percent:.1f}%")
self.alerts_sent[alert_key] = time.time()
async def _handle_high_disk_usage(self, disk_percent: float):
"""Handle high disk usage."""
alert_key = "high_disk"
last_alert = self.alerts_sent.get(alert_key)
if last_alert and (time.time() - last_alert) < 1800: # 30 minutes
return
logger.error(f"Critical disk usage detected: {disk_percent:.1f}%")
self.alerts_sent[alert_key] = time.time()
async def _handle_alert(self, alert: dict[str, Any]):
"""Handle individual alert."""
severity = alert.get("severity", "info")
# Log alert based on severity
if severity == "critical":
logger.error(f"Critical alert: {alert.get('title', 'Unknown')}")
elif severity == "warning":
logger.warning(f"Warning alert: {alert.get('title', 'Unknown')}")
else:
logger.info(f"Info alert: {alert.get('title', 'Unknown')}")
async def _trigger_maintenance_alert(self):
"""Trigger alert for system maintenance needed."""
alert_key = "maintenance_needed"
last_alert = self.alerts_sent.get(alert_key)
if last_alert and (time.time() - last_alert) < 3600: # 1 hour
return
logger.error("System health trends indicate maintenance may be needed")
self.alerts_sent[alert_key] = time.time()
def get_monitoring_status(self) -> dict[str, Any]:
"""Get current monitoring system status."""
return {
"running": self.running,
"uptime_seconds": time.time() - self.start_time,
"active_tasks": len([t for t in self.tasks if not t.done()]),
"total_tasks": len(self.tasks),
"health_records": len(self.health_history),
"alerts_sent_count": len(self.alerts_sent),
"last_health_check": max(
[record["timestamp"] for record in self.health_history], default=None
),
}
# Global health monitor instance
_health_monitor = HealthMonitor()
def get_health_monitor() -> HealthMonitor:
"""Get the global health monitor instance."""
return _health_monitor
async def start_health_monitoring():
"""Start health monitoring system (convenience function)."""
await _health_monitor.start()
async def stop_health_monitoring():
"""Stop health monitoring system (convenience function)."""
await _health_monitor.stop()
def get_monitoring_status() -> dict[str, Any]:
"""Get monitoring status (convenience function)."""
return _health_monitor.get_monitoring_status()
```
--------------------------------------------------------------------------------
/maverick_mcp/tests/test_stock_data_provider.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the StockDataProvider class.
"""
import unittest
from datetime import datetime
from unittest.mock import MagicMock, PropertyMock, patch
import pandas as pd
from maverick_mcp.providers.stock_data import StockDataProvider
class TestStockDataProvider(unittest.TestCase):
"""Test suite for StockDataProvider."""
def setUp(self):
"""Set up test fixtures."""
self.provider = StockDataProvider()
# Create sample data
self.sample_data = pd.DataFrame(
{
"Open": [100, 101, 102, 103, 104],
"High": [101, 102, 103, 104, 105],
"Low": [99, 100, 101, 102, 103],
"Close": [100.5, 101.5, 102.5, 103.5, 104.5],
"Volume": [1000000, 1100000, 1200000, 1300000, 1400000],
},
index=pd.date_range(end=datetime.now(), periods=5, freq="D"),
)
@patch("yfinance.Ticker")
def test_get_stock_data_with_period(self, mock_ticker_class):
"""Test fetching stock data with period parameter."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.history.return_value = self.sample_data
result = self.provider.get_stock_data("AAPL", period="5d")
self.assertIsInstance(result, pd.DataFrame)
self.assertEqual(len(result), 5)
mock_ticker.history.assert_called_once_with(period="5d", interval="1d")
@patch("yfinance.Ticker")
def test_get_stock_data_with_dates(self, mock_ticker_class):
"""Test fetching stock data with date range."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.history.return_value = self.sample_data
start_date = "2024-01-01"
end_date = "2024-01-05"
# Disable cache to avoid database connection
result = self.provider.get_stock_data(
"AAPL", start_date, end_date, use_cache=False
)
self.assertIsInstance(result, pd.DataFrame)
mock_ticker.history.assert_called_once_with(
start=start_date, end=end_date, interval="1d"
)
@patch("yfinance.Ticker")
def test_get_stock_data_empty_response(self, mock_ticker_class):
"""Test handling of empty data response."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.history.return_value = pd.DataFrame()
result = self.provider.get_stock_data("INVALID")
self.assertIsInstance(result, pd.DataFrame)
self.assertTrue(result.empty)
self.assertListEqual(
list(result.columns), ["Open", "High", "Low", "Close", "Volume"]
)
@patch("yfinance.Ticker")
def test_get_stock_data_missing_columns(self, mock_ticker_class):
"""Test handling of missing columns in data."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
# Return data missing Volume column
incomplete_data = self.sample_data[["Open", "High", "Low", "Close"]].copy()
mock_ticker.history.return_value = incomplete_data
# Disable cache to ensure we get mocked data
result = self.provider.get_stock_data("AAPL", use_cache=False)
self.assertIsInstance(result, pd.DataFrame)
self.assertIn("Volume", result.columns)
# Volume should be 0 when missing (not NaN)
self.assertTrue((result["Volume"] == 0).all())
@patch("yfinance.Ticker")
def test_get_stock_data_with_retry(self, mock_ticker_class):
"""Test retry mechanism on timeout."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
# First call times out, second succeeds
import requests
mock_ticker.history.side_effect = [
requests.Timeout("Read timeout"),
self.sample_data,
]
# Disable cache to avoid database connection
result = self.provider.get_stock_data("AAPL", use_cache=False)
self.assertIsInstance(result, pd.DataFrame)
self.assertEqual(mock_ticker.history.call_count, 2)
@patch("yfinance.Ticker")
def test_get_stock_info(self, mock_ticker_class):
"""Test fetching stock info."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.info = {
"symbol": "AAPL",
"longName": "Apple Inc.",
"marketCap": 3000000000000,
"sector": "Technology",
}
result = self.provider.get_stock_info("AAPL")
self.assertIsInstance(result, dict)
self.assertEqual(result["symbol"], "AAPL")
self.assertEqual(result["longName"], "Apple Inc.")
@patch("yfinance.Ticker")
def test_get_stock_info_error(self, mock_ticker_class):
"""Test error handling in stock info fetching."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
# Simulate an exception when accessing info
type(mock_ticker).info = PropertyMock(side_effect=Exception("API Error"))
result = self.provider.get_stock_info("INVALID")
self.assertIsInstance(result, dict)
self.assertEqual(result, {})
@patch("yfinance.Ticker")
def test_get_realtime_data(self, mock_ticker_class):
"""Test fetching real-time data."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
# Mock today's data
today_data = pd.DataFrame(
{"Close": [105.0], "Volume": [1500000]}, index=[datetime.now()]
)
mock_ticker.history.return_value = today_data
mock_ticker.info = {"previousClose": 104.0}
result = self.provider.get_realtime_data("AAPL")
self.assertIsInstance(result, dict)
self.assertIsNotNone(result)
assert result is not None # Type narrowing for pyright
self.assertEqual(result["symbol"], "AAPL")
self.assertEqual(result["price"], 105.0)
self.assertEqual(result["change"], 1.0)
self.assertAlmostEqual(result["change_percent"], 0.96, places=2)
@patch("yfinance.Ticker")
def test_get_all_realtime_data(self, mock_ticker_class):
"""Test fetching real-time data for multiple symbols."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
# Mock data for each symbol
mock_data = pd.DataFrame(
{"Close": [105.0], "Volume": [1500000]}, index=[datetime.now()]
)
mock_ticker.history.return_value = mock_data
mock_ticker.info = {"previousClose": 104.0}
symbols = ["AAPL", "GOOGL", "MSFT"]
result = self.provider.get_all_realtime_data(symbols)
self.assertIsInstance(result, dict)
for symbol in symbols:
self.assertIn(symbol, result)
self.assertEqual(result[symbol]["symbol"], symbol)
def test_is_market_open_weekday(self):
"""Test market open check on weekday."""
# Mock a weekday during market hours
with patch("maverick_mcp.providers.stock_data.datetime") as mock_datetime:
with patch("maverick_mcp.providers.stock_data.pytz") as mock_pytz:
# Create mock timezone
mock_tz = MagicMock()
mock_pytz.timezone.return_value = mock_tz
# Tuesday at 2 PM ET
mock_now = MagicMock()
mock_now.weekday.return_value = 1 # Tuesday
mock_now.hour = 14
mock_now.minute = 0
mock_now.__le__ = lambda self, other: True # For market_open <= now
mock_now.__ge__ = lambda self, other: False # For now <= market_close
# Mock replace to return different times for market open/close
def mock_replace(**kwargs):
if kwargs.get("hour") == 9: # market open
m = MagicMock()
m.__le__ = lambda self, other: True
return m
elif kwargs.get("hour") == 16: # market close
m = MagicMock()
m.__ge__ = lambda self, other: True
return m
return mock_now
mock_now.replace = mock_replace
mock_datetime.now.return_value = mock_now
result = self.provider.is_market_open()
self.assertTrue(result)
def test_is_market_open_weekend(self):
"""Test market open check on weekend."""
# Mock a Saturday
with patch("maverick_mcp.providers.stock_data.datetime") as mock_datetime:
with patch("maverick_mcp.providers.stock_data.pytz") as mock_pytz:
# Create mock timezone
mock_tz = MagicMock()
mock_pytz.timezone.return_value = mock_tz
# Saturday at 2 PM ET
mock_now = MagicMock()
mock_now.weekday.return_value = 5 # Saturday
mock_datetime.now.return_value = mock_now
result = self.provider.is_market_open()
self.assertFalse(result)
@patch("yfinance.Ticker")
def test_get_news(self, mock_ticker_class):
"""Test fetching news."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.news = [
{
"title": "Apple announces new product",
"publisher": "Reuters",
"link": "https://example.com/1",
"providerPublishTime": 1704150000,
"type": "STORY",
},
{
"title": "Apple stock rises",
"publisher": "Bloomberg",
"link": "https://example.com/2",
"providerPublishTime": 1704153600,
"type": "STORY",
},
]
result = self.provider.get_news("AAPL", limit=2)
self.assertIsInstance(result, pd.DataFrame)
self.assertEqual(len(result), 2)
self.assertIn("title", result.columns)
self.assertIn("providerPublishTime", result.columns)
# Check timestamp conversion
self.assertEqual(result["providerPublishTime"].dtype, "datetime64[ns]")
@patch("yfinance.Ticker")
def test_get_news_empty(self, mock_ticker_class):
"""Test fetching news with no results."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.news = []
result = self.provider.get_news("AAPL")
self.assertIsInstance(result, pd.DataFrame)
self.assertTrue(result.empty)
self.assertListEqual(
list(result.columns),
["title", "publisher", "link", "providerPublishTime", "type"],
)
@patch("yfinance.Ticker")
def test_get_earnings(self, mock_ticker_class):
"""Test fetching earnings data."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
# Mock earnings data
mock_earnings = pd.DataFrame({"Revenue": [100, 110], "Earnings": [10, 12]})
mock_ticker.earnings = mock_earnings
mock_ticker.earnings_dates = pd.DataFrame({"EPS Estimate": [1.5, 1.6]})
mock_ticker.earnings_trend = {"trend": [{"growth": 0.1}]}
result = self.provider.get_earnings("AAPL")
self.assertIsInstance(result, dict)
self.assertIn("earnings", result)
self.assertIn("earnings_dates", result)
self.assertIn("earnings_trend", result)
@patch("yfinance.Ticker")
def test_get_recommendations(self, mock_ticker_class):
"""Test fetching analyst recommendations."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_recommendations = pd.DataFrame(
{
"firm": ["Morgan Stanley", "Goldman Sachs"],
"toGrade": ["Buy", "Hold"],
"fromGrade": ["Hold", "Sell"],
"action": ["upgrade", "upgrade"],
}
)
mock_ticker.recommendations = mock_recommendations
result = self.provider.get_recommendations("AAPL")
self.assertIsInstance(result, pd.DataFrame)
self.assertEqual(len(result), 2)
self.assertIn("firm", result.columns)
self.assertIn("toGrade", result.columns)
@patch("yfinance.Ticker")
def test_is_etf_by_quote_type(self, mock_ticker_class):
"""Test ETF detection by quoteType."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.info = {"quoteType": "ETF"}
result = self.provider.is_etf("SPY")
self.assertTrue(result)
@patch("yfinance.Ticker")
def test_is_etf_by_symbol(self, mock_ticker_class):
"""Test ETF detection by known symbol."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.info = {} # No quoteType
result = self.provider.is_etf("SPY")
self.assertTrue(result) # SPY is in the known ETF list
@patch("yfinance.Ticker")
def test_is_etf_false(self, mock_ticker_class):
"""Test non-ETF detection."""
mock_ticker = MagicMock()
mock_ticker_class.return_value = mock_ticker
mock_ticker.info = {"quoteType": "EQUITY", "longName": "Apple Inc."}
result = self.provider.is_etf("AAPL")
self.assertFalse(result)
def test_singleton_pattern(self):
"""Test that StockDataProvider follows singleton pattern."""
provider1 = StockDataProvider()
provider2 = StockDataProvider()
self.assertIs(provider1, provider2)
if __name__ == "__main__":
unittest.main()
```
--------------------------------------------------------------------------------
/maverick_mcp/monitoring/integration_example.py:
--------------------------------------------------------------------------------
```python
"""
Example integration of monitoring metrics into backtesting code.
This module demonstrates how to integrate the monitoring system
into existing backtesting strategies and data providers.
"""
import asyncio
from typing import Any
import numpy as np
import pandas as pd
from maverick_mcp.monitoring.metrics import get_backtesting_metrics
from maverick_mcp.monitoring.middleware import (
MetricsCircuitBreaker,
get_metrics_middleware,
track_api_call,
track_resource_usage,
track_strategy_execution,
)
from maverick_mcp.utils.logging import get_logger
logger = get_logger(__name__)
class MonitoredStockDataProvider:
"""
Example stock data provider with integrated monitoring.
Shows how to add metrics tracking to data fetching operations.
"""
def __init__(self):
self.circuit_breaker = MetricsCircuitBreaker(
provider="tiingo",
endpoint="/daily",
failure_threshold=5,
recovery_timeout=60,
)
self.logger = get_logger(f"{__name__}.MonitoredStockDataProvider")
@track_api_call("tiingo", "/daily/{symbol}")
async def get_stock_data(
self, symbol: str, start_date: str = None, end_date: str = None
) -> pd.DataFrame:
"""
Fetch stock data with automatic API call tracking.
The @track_api_call decorator automatically tracks:
- Request duration
- Success/failure status
- Rate limiting metrics
"""
# Simulate API call delay
await asyncio.sleep(0.1)
# Simulate occasional API errors for demonstration
if np.random.random() < 0.05: # 5% error rate
raise Exception("API rate limit exceeded")
# Generate sample data
dates = pd.date_range(start="2023-01-01", end="2023-12-31", freq="D")
data = pd.DataFrame(
{
"Date": dates,
"Open": np.random.uniform(100, 200, len(dates)),
"High": np.random.uniform(150, 250, len(dates)),
"Low": np.random.uniform(50, 150, len(dates)),
"Close": np.random.uniform(100, 200, len(dates)),
"Volume": np.random.randint(1000000, 10000000, len(dates)),
}
)
# Track additional metrics
collector = get_backtesting_metrics()
collector.track_cache_operation(
cache_type="api_response",
operation="fetch",
hit=False, # Assume cache miss for this example
key_pattern=f"stock_data_{symbol}",
)
return data
async def get_stock_data_with_circuit_breaker(self, symbol: str) -> pd.DataFrame:
"""
Fetch stock data with circuit breaker protection.
Automatically tracks circuit breaker state changes and failures.
"""
return await self.circuit_breaker.call(self.get_stock_data, symbol=symbol)
class MonitoredTradingStrategy:
"""
Example trading strategy with comprehensive monitoring.
Shows how to add metrics tracking to strategy execution.
"""
def __init__(self, name: str):
self.name = name
self.data_provider = MonitoredStockDataProvider()
self.middleware = get_metrics_middleware()
self.logger = get_logger(f"{__name__}.MonitoredTradingStrategy")
@track_strategy_execution("RSI_Strategy", "AAPL", "1D")
async def run_backtest(
self, symbol: str, data: pd.DataFrame = None, data_points: int = None
) -> dict[str, Any]:
"""
Run backtest with automatic strategy execution tracking.
The @track_strategy_execution decorator automatically tracks:
- Execution duration
- Memory usage
- Success/failure status
- Data points processed
"""
if data is None:
data = await self.data_provider.get_stock_data(symbol)
# Track the actual data points being processed
actual_data_points = len(data)
# Simulate strategy calculations
await self._calculate_rsi_signals(data)
# Simulate backtest execution
performance_results = await self._simulate_trading(data, symbol)
# Add data points info to results for metrics tracking
performance_results["data_points_processed"] = actual_data_points
return performance_results
@track_resource_usage("rsi_calculation")
async def _calculate_rsi_signals(self, data: pd.DataFrame) -> pd.DataFrame:
"""
Calculate RSI signals with resource usage tracking.
The @track_resource_usage decorator tracks:
- Memory usage during calculation
- Computation time
- Data size category
"""
# Simulate RSI calculation (simplified)
await asyncio.sleep(0.05) # Simulate computation time
# Calculate RSI (simplified version)
data["rsi"] = np.random.uniform(20, 80, len(data))
data["signal"] = np.where(
data["rsi"] < 30, 1, np.where(data["rsi"] > 70, -1, 0)
)
return data
async def _simulate_trading(
self, data: pd.DataFrame, symbol: str
) -> dict[str, Any]:
"""
Simulate trading and calculate performance metrics.
Returns comprehensive performance metrics that will be
automatically tracked by the strategy execution decorator.
"""
# Simulate trading logic
signals = data["signal"]
# Calculate returns (simplified)
total_return = np.random.uniform(-10, 30) # Random return between -10% and 30%
sharpe_ratio = np.random.uniform(-0.5, 2.5) # Random Sharpe ratio
max_drawdown = np.random.uniform(5, 25) # Random drawdown 5-25%
win_rate = np.random.uniform(0.35, 0.75) # Random win rate 35-75%
# Count trades
position_changes = np.diff(signals)
total_trades = np.sum(np.abs(position_changes))
winning_trades = int(total_trades * win_rate)
# Track portfolio updates
collector = get_backtesting_metrics()
collector.update_portfolio_metrics(
portfolio_id="demo_portfolio",
portfolio_value_usd=100000 * (1 + total_return / 100),
daily_pnl_usd=total_return * 1000, # Simulated daily PnL
strategy=self.name,
positions=[{"symbol": symbol, "quantity": 100, "type": "long"}],
)
# Return performance metrics in expected format
return {
"strategy_name": self.name,
"symbol": symbol,
"total_return": total_return,
"returns": total_return, # Alternative key name
"sharpe_ratio": sharpe_ratio,
"max_drawdown": max_drawdown,
"max_dd": max_drawdown, # Alternative key name
"win_rate": win_rate * 100, # Convert to percentage
"win_ratio": win_rate, # Alternative key name
"total_trades": int(total_trades),
"num_trades": int(total_trades), # Alternative key name
"winning_trades": winning_trades,
"performance_summary": {
"profitable": total_return > 0,
"risk_adjusted_return": sharpe_ratio,
"maximum_loss": max_drawdown,
},
}
class MonitoredDatabaseRepository:
"""
Example database repository with monitoring integration.
Shows how to add database operation tracking.
"""
def __init__(self):
self.middleware = get_metrics_middleware()
self.logger = get_logger(f"{__name__}.MonitoredDatabaseRepository")
async def save_backtest_results(
self, strategy_name: str, symbol: str, results: dict[str, Any]
) -> bool:
"""
Save backtest results with database operation tracking.
"""
async with self.middleware.track_database_operation(
query_type="INSERT", table_name="backtest_results", operation="save_results"
):
# Simulate database save operation
await asyncio.sleep(0.02)
# Simulate occasional database errors
if np.random.random() < 0.01: # 1% error rate
raise Exception("Database connection timeout")
self.logger.info(
f"Saved backtest results for {strategy_name} on {symbol}",
extra={
"strategy": strategy_name,
"symbol": symbol,
"total_return": results.get("total_return", 0),
},
)
return True
async def get_historical_performance(
self, strategy_name: str, days: int = 30
) -> list[dict[str, Any]]:
"""
Retrieve historical performance with tracking.
"""
async with self.middleware.track_database_operation(
query_type="SELECT",
table_name="backtest_results",
operation="get_performance",
):
# Simulate database query
await asyncio.sleep(0.01)
# Generate sample historical data
historical_data = []
for i in range(days):
historical_data.append(
{
"date": f"2024-01-{i + 1:02d}",
"strategy": strategy_name,
"return": np.random.uniform(-2, 3),
"sharpe_ratio": np.random.uniform(0.5, 2.0),
}
)
return historical_data
# Demonstration function
async def demonstrate_monitoring_integration():
"""
Demonstrate comprehensive monitoring integration.
This function shows how all monitoring components work together
in a typical backtesting workflow.
"""
logger.info("Starting monitoring integration demonstration")
# Initialize components
strategy = MonitoredTradingStrategy("RSI_Momentum")
repository = MonitoredDatabaseRepository()
# List of symbols to test
symbols = ["AAPL", "GOOGL", "MSFT", "TSLA"]
for symbol in symbols:
try:
logger.info(f"Running backtest for {symbol}")
# Run backtest (automatically tracked)
results = await strategy.run_backtest(
symbol=symbol,
data_points=252, # One year of trading days
)
# Save results (automatically tracked)
await repository.save_backtest_results(
strategy_name=strategy.name, symbol=symbol, results=results
)
# Get historical performance (automatically tracked)
historical = await repository.get_historical_performance(
strategy_name=strategy.name, days=30
)
logger.info(
f"Completed backtest for {symbol}",
extra={
"symbol": symbol,
"total_return": results.get("total_return", 0),
"sharpe_ratio": results.get("sharpe_ratio", 0),
"historical_records": len(historical),
},
)
except Exception as e:
logger.error(f"Backtest failed for {symbol}: {e}")
# Manually track the anomaly
collector = get_backtesting_metrics()
collector.detect_anomaly(
anomaly_type="backtest_execution_failure",
severity="critical",
context={
"strategy_name": strategy.name,
"symbol": symbol,
"error": str(e),
},
)
logger.info("Monitoring integration demonstration completed")
# Alert checking function
async def check_and_report_anomalies():
"""
Example function to check for anomalies and generate alerts.
This would typically be run periodically by a scheduler.
"""
logger.info("Checking for performance anomalies")
collector = get_backtesting_metrics()
# Simulate checking various metrics and detecting anomalies
anomalies_detected = 0
# Example: Check if any strategy has poor recent performance
strategies = ["RSI_Momentum", "MACD_Trend", "Bollinger_Bands"]
for strategy in strategies:
# Simulate performance check
recent_sharpe = np.random.uniform(-1.0, 2.5)
recent_drawdown = np.random.uniform(5, 35)
if recent_sharpe < 0.5:
collector.detect_anomaly(
anomaly_type="low_sharpe_ratio",
severity="warning" if recent_sharpe > 0 else "critical",
context={
"strategy_name": strategy,
"sharpe_ratio": recent_sharpe,
"threshold": 0.5,
},
)
anomalies_detected += 1
if recent_drawdown > 25:
collector.detect_anomaly(
anomaly_type="high_drawdown",
severity="critical",
context={
"strategy_name": strategy,
"max_drawdown": recent_drawdown,
"threshold": 25,
},
)
anomalies_detected += 1
logger.info(f"Anomaly check completed. Detected {anomalies_detected} anomalies")
return anomalies_detected
if __name__ == "__main__":
# Run the demonstration
async def main():
await demonstrate_monitoring_integration()
await check_and_report_anomalies()
# Print metrics summary
collector = get_backtesting_metrics()
metrics_text = collector.get_metrics_text()
print("\n" + "=" * 50)
print("PROMETHEUS METRICS SAMPLE:")
print("=" * 50)
print(metrics_text[:1000] + "..." if len(metrics_text) > 1000 else metrics_text)
asyncio.run(main())
```
--------------------------------------------------------------------------------
/scripts/seed_db.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Database seeding script for MaverickMCP.
This script populates the database with sample stock data from Tiingo API,
including stocks, price data, and sample screening results.
"""
import logging
import os
import sys
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from pathlib import Path
# Add the project root to the Python path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# noqa: E402 - imports must come after sys.path modification
from sqlalchemy import create_engine # noqa: E402
from sqlalchemy.orm import sessionmaker # noqa: E402
from maverick_mcp.data.models import ( # noqa: E402
MaverickBearStocks,
MaverickStocks,
PriceCache,
Stock,
SupplyDemandBreakoutStocks,
TechnicalCache,
bulk_insert_price_data,
bulk_insert_screening_data,
)
from maverick_mcp.providers.stock_data import EnhancedStockDataProvider # noqa: E402
# Set up logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("maverick_mcp.seed")
# Sample stock tickers for different categories
SAMPLE_STOCKS = {
"large_cap": [
"AAPL",
"MSFT",
"GOOGL",
"AMZN",
"TSLA",
"NVDA",
"META",
"BRK-B",
"JNJ",
"V",
],
"growth": ["AMD", "CRM", "SHOP", "ROKU", "ZM", "DOCU", "SNOW", "PLTR", "RBLX", "U"],
"value": ["KO", "PFE", "XOM", "CVX", "JPM", "BAC", "WMT", "PG", "T", "VZ"],
"small_cap": [
"UPST",
"SOFI",
"OPEN",
"WISH",
"CLOV",
"SPCE",
"LCID",
"RIVN",
"BYND",
"PTON",
],
}
def get_database_url() -> str:
"""Get the database URL from environment or settings."""
return os.getenv("DATABASE_URL") or "sqlite:///maverick_mcp.db"
def setup_stock_provider(session) -> EnhancedStockDataProvider:
"""Set up Enhanced stock data provider."""
# The EnhancedStockDataProvider uses yfinance and doesn't require API keys
provider = EnhancedStockDataProvider(db_session=session)
logger.info("Enhanced stock data provider initialized")
return provider
def create_sample_stocks(session, stocks_list: list[str]) -> dict[str, Stock]:
"""Create sample stock records."""
logger.info(f"Creating {len(stocks_list)} sample stocks...")
created_stocks = {}
for ticker in stocks_list:
try:
# Create basic stock record
stock = Stock.get_or_create(
session,
ticker_symbol=ticker,
company_name=f"{ticker} Inc.", # Simple placeholder
sector="Technology", # Default sector
exchange="NASDAQ",
country="US",
currency="USD",
is_active=True,
)
created_stocks[ticker] = stock
logger.info(f"Created stock: {ticker}")
except Exception as e:
logger.error(f"Error creating stock {ticker}: {e}")
continue
logger.info(f"Successfully created {len(created_stocks)} stocks")
return created_stocks
def fetch_and_store_price_data(
session, stock_provider: EnhancedStockDataProvider, stocks: dict[str, Stock]
) -> None:
"""Fetch price data using EnhancedStockDataProvider and store in database."""
logger.info("Fetching price data using yfinance...")
# Get data for last 200 days
end_date = datetime.now(UTC)
start_date = end_date - timedelta(days=200)
success_count = 0
for ticker, _stock in stocks.items():
try:
logger.info(f"Fetching price data for {ticker}...")
# Get price data using the enhanced provider
data = stock_provider.get_stock_data(
ticker,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if data is not None and not data.empty:
# Store in database
inserted_count = bulk_insert_price_data(session, ticker, data)
logger.info(f"Inserted {inserted_count} price records for {ticker}")
success_count += 1
else:
logger.warning(f"No price data received for {ticker}")
except Exception as e:
logger.error(f"Error fetching price data for {ticker}: {e}")
continue
logger.info(
f"Successfully fetched price data for {success_count}/{len(stocks)} stocks"
)
def generate_sample_screening_data(stocks: dict[str, Stock]) -> list[dict]:
"""Generate sample screening data for testing."""
logger.info("Generating sample screening data...")
screening_data = []
for i, (ticker, _stock) in enumerate(stocks.items()):
# Generate realistic-looking screening data
base_score = 50 + (i % 40) + (hash(ticker) % 20) # Score between 50-110
data = {
"ticker": ticker,
"close": round(100 + (i * 10) + (hash(ticker) % 50), 2),
"volume": 1000000 + (i * 100000),
"momentum_score": round(base_score + (hash(ticker) % 30), 2),
"combined_score": min(100, base_score + (hash(ticker) % 25)),
"ema_21": round(95 + (i * 9), 2),
"sma_50": round(90 + (i * 8), 2),
"sma_150": round(85 + (i * 7), 2),
"sma_200": round(80 + (i * 6), 2),
"adr_pct": round(2 + (hash(ticker) % 8), 2),
"atr": round(3 + (hash(ticker) % 5), 2),
"pattern_type": ["Breakout", "Continuation", "Reversal", "Consolidation"][
i % 4
],
"squeeze_status": [
"No Squeeze",
"Low Squeeze",
"Mid Squeeze",
"High Squeeze",
][i % 4],
"consolidation_status": ["Base", "Flag", "Pennant", "Triangle"][i % 4],
"entry_signal": ["Buy", "Hold", "Watch", "Caution"][i % 4],
"compression_score": hash(ticker) % 10,
"pattern_detected": 1 if hash(ticker) % 3 == 0 else 0,
}
screening_data.append(data)
return screening_data
def create_sample_screening_results(session, stocks: dict[str, Stock]) -> None:
"""Create sample screening results for all categories."""
logger.info("Creating sample screening results...")
# Generate sample data
screening_data = generate_sample_screening_data(stocks)
# Split data for different screening types
total_stocks = len(screening_data)
# Top 60% for Maverick stocks (bullish)
maverick_data = sorted(
screening_data, key=lambda x: x["combined_score"], reverse=True
)[: int(total_stocks * 0.6)]
maverick_count = bulk_insert_screening_data(session, MaverickStocks, maverick_data)
logger.info(f"Created {maverick_count} Maverick screening results")
# Bottom 40% for Bear stocks
bear_data = sorted(screening_data, key=lambda x: x["combined_score"])[
: int(total_stocks * 0.4)
]
# Add bear-specific fields
for data in bear_data:
data["score"] = 100 - data["combined_score"] # Invert score for bear
data["rsi_14"] = 70 + (hash(data["ticker"]) % 20) # High RSI for bear
data["macd"] = -1 * (hash(data["ticker"]) % 5) / 10 # Negative MACD
data["macd_signal"] = -0.5 * (hash(data["ticker"]) % 3) / 10
data["macd_histogram"] = data["macd"] - data["macd_signal"]
data["dist_days_20"] = hash(data["ticker"]) % 15
data["atr_contraction"] = hash(data["ticker"]) % 2 == 0
data["big_down_vol"] = hash(data["ticker"]) % 3 == 0
bear_count = bulk_insert_screening_data(session, MaverickBearStocks, bear_data)
logger.info(f"Created {bear_count} Bear screening results")
# Top 40% for Supply/Demand breakouts
supply_demand_data = sorted(
screening_data, key=lambda x: x["momentum_score"], reverse=True
)[: int(total_stocks * 0.4)]
# Add supply/demand specific fields
for data in supply_demand_data:
data["accumulation_rating"] = (hash(data["ticker"]) % 8) + 2 # 2-9 rating
data["distribution_rating"] = 10 - data["accumulation_rating"]
data["breakout_strength"] = (hash(data["ticker"]) % 7) + 3 # 3-9 rating
data["avg_volume_30d"] = data["volume"] * 1.2 # 20% above current volume
supply_demand_count = bulk_insert_screening_data(
session, SupplyDemandBreakoutStocks, supply_demand_data
)
logger.info(f"Created {supply_demand_count} Supply/Demand breakout results")
def create_sample_technical_indicators(session, stocks: dict[str, Stock]) -> None:
"""Create sample technical indicator cache data."""
logger.info("Creating sample technical indicator cache...")
# Create sample technical data for the last 30 days
end_date = datetime.now(UTC).date()
indicator_count = 0
for days_ago in range(30):
date = end_date - timedelta(days=days_ago)
for ticker, stock in list(stocks.items())[
:10
]: # Limit to first 10 stocks for demo
try:
# RSI
rsi_value = 30 + (hash(f"{ticker}{date}") % 40) # RSI between 30-70
rsi_cache = TechnicalCache(
stock_id=stock.stock_id,
date=date,
indicator_type="RSI_14",
value=Decimal(str(rsi_value)),
period=14,
)
session.add(rsi_cache)
# SMA_20
sma_value = 100 + (hash(f"{ticker}{date}sma") % 50)
sma_cache = TechnicalCache(
stock_id=stock.stock_id,
date=date,
indicator_type="SMA_20",
value=Decimal(str(sma_value)),
period=20,
)
session.add(sma_cache)
indicator_count += 2
except Exception as e:
logger.error(f"Error creating technical indicators for {ticker}: {e}")
continue
session.commit()
logger.info(f"Created {indicator_count} technical indicator cache entries")
def verify_data(session) -> None:
"""Verify that data was seeded correctly."""
logger.info("Verifying seeded data...")
# Count records in each table
stock_count = session.query(Stock).count()
price_count = session.query(PriceCache).count()
maverick_count = session.query(MaverickStocks).count()
bear_count = session.query(MaverickBearStocks).count()
supply_demand_count = session.query(SupplyDemandBreakoutStocks).count()
technical_count = session.query(TechnicalCache).count()
logger.info("=== Data Seeding Summary ===")
logger.info(f"Stocks: {stock_count}")
logger.info(f"Price records: {price_count}")
logger.info(f"Maverick screening: {maverick_count}")
logger.info(f"Bear screening: {bear_count}")
logger.info(f"Supply/Demand screening: {supply_demand_count}")
logger.info(f"Technical indicators: {technical_count}")
logger.info("============================")
# Test a few queries
if maverick_count > 0:
top_maverick = (
session.query(MaverickStocks)
.order_by(MaverickStocks.combined_score.desc())
.first()
)
if top_maverick and top_maverick.stock:
logger.info(
f"Top Maverick stock: {top_maverick.stock.ticker_symbol} (Score: {top_maverick.combined_score})"
)
if bear_count > 0:
top_bear = (
session.query(MaverickBearStocks)
.order_by(MaverickBearStocks.score.desc())
.first()
)
if top_bear and top_bear.stock:
logger.info(
f"Top Bear stock: {top_bear.stock.ticker_symbol} (Score: {top_bear.score})"
)
def main():
"""Main seeding function."""
logger.info("Starting MaverickMCP database seeding...")
# No API key required for yfinance provider
logger.info("Using yfinance for stock data - no API key required")
# Set up database connection
database_url = get_database_url()
engine = create_engine(database_url, echo=False)
SessionLocal = sessionmaker(bind=engine)
# Set up stock data provider
stock_provider = None
with SessionLocal() as session:
try:
# Get all stock tickers
all_tickers = []
for _category, tickers in SAMPLE_STOCKS.items():
all_tickers.extend(tickers)
logger.info(
f"Seeding database with {len(all_tickers)} stocks across {len(SAMPLE_STOCKS)} categories"
)
# Set up provider with session
stock_provider = setup_stock_provider(session)
# Create stocks
stocks = create_sample_stocks(session, all_tickers)
if not stocks:
logger.error("No stocks created. Exiting.")
return False
# Fetch price data (this takes time, so we'll do a subset)
price_stocks = {
k: v for i, (k, v) in enumerate(stocks.items()) if i < 10
} # First 10 stocks
fetch_and_store_price_data(session, stock_provider, price_stocks)
# Create screening results
create_sample_screening_results(session, stocks)
# Create technical indicators
create_sample_technical_indicators(session, stocks)
# Verify data
verify_data(session)
logger.info("✅ Database seeding completed successfully!")
return True
except Exception as e:
logger.error(f"Database seeding failed: {e}")
session.rollback()
return False
if __name__ == "__main__":
success = main()
if not success:
sys.exit(1)
```
--------------------------------------------------------------------------------
/alembic/versions/010_self_contained_schema.py:
--------------------------------------------------------------------------------
```python
"""Create self-contained schema with mcp_ prefixed tables
Revision ID: 010_self_contained_schema
Revises: 009_rename_to_supply_demand
Create Date: 2025-01-31
This migration creates a complete self-contained schema for maverick-mcp
with all tables prefixed with 'mcp_' to avoid conflicts with external systems.
Tables created:
- mcp_stocks: Master stock information
- mcp_price_cache: Historical price data
- mcp_maverick_stocks: Maverick screening results
- mcp_maverick_bear_stocks: Bear market screening results
- mcp_supply_demand_breakouts: Supply/demand analysis
- mcp_technical_cache: Technical indicator cache
"""
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
# revision identifiers
revision = "010_self_contained_schema"
down_revision = "009_rename_to_supply_demand"
branch_labels = None
depends_on = None
def upgrade():
"""Create self-contained schema with all mcp_ prefixed tables."""
# Check if we're using PostgreSQL or SQLite
op.get_bind()
print("🚀 Creating self-contained maverick-mcp schema...")
# 1. Create mcp_stocks table (master stock data)
print("📊 Creating mcp_stocks table...")
op.create_table(
"mcp_stocks",
sa.Column("stock_id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"ticker_symbol", sa.String(10), nullable=False, unique=True, index=True
),
sa.Column("company_name", sa.String(255)),
sa.Column("description", sa.Text()),
sa.Column("sector", sa.String(100)),
sa.Column("industry", sa.String(100)),
sa.Column("exchange", sa.String(50)),
sa.Column("country", sa.String(50)),
sa.Column("currency", sa.String(3)),
sa.Column("isin", sa.String(12)),
sa.Column("market_cap", sa.BigInteger()),
sa.Column("shares_outstanding", sa.BigInteger()),
sa.Column("is_etf", sa.Boolean(), default=False),
sa.Column("is_active", sa.Boolean(), default=True, index=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
# Create indexes for mcp_stocks
op.create_index("mcp_stocks_ticker_idx", "mcp_stocks", ["ticker_symbol"])
op.create_index("mcp_stocks_sector_idx", "mcp_stocks", ["sector"])
op.create_index("mcp_stocks_exchange_idx", "mcp_stocks", ["exchange"])
# 2. Create mcp_price_cache table
print("💰 Creating mcp_price_cache table...")
op.create_table(
"mcp_price_cache",
sa.Column("price_cache_id", postgresql.UUID(as_uuid=True), primary_key=True),
sa.Column(
"stock_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("mcp_stocks.stock_id"),
nullable=False,
),
sa.Column("date", sa.Date(), nullable=False),
sa.Column("open_price", sa.Numeric(12, 4)),
sa.Column("high_price", sa.Numeric(12, 4)),
sa.Column("low_price", sa.Numeric(12, 4)),
sa.Column("close_price", sa.Numeric(12, 4)),
sa.Column("volume", sa.BigInteger()),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
# Create unique constraint and indexes for price cache
op.create_unique_constraint(
"mcp_price_cache_stock_date_unique", "mcp_price_cache", ["stock_id", "date"]
)
op.create_index(
"mcp_price_cache_stock_id_date_idx", "mcp_price_cache", ["stock_id", "date"]
)
op.create_index("mcp_price_cache_date_idx", "mcp_price_cache", ["date"])
# 3. Create mcp_maverick_stocks table
print("🎯 Creating mcp_maverick_stocks table...")
op.create_table(
"mcp_maverick_stocks",
sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
sa.Column(
"stock_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("mcp_stocks.stock_id"),
nullable=False,
index=True,
),
sa.Column("date_analyzed", sa.Date(), nullable=False),
# OHLCV Data
sa.Column("open_price", sa.Numeric(12, 4), default=0),
sa.Column("high_price", sa.Numeric(12, 4), default=0),
sa.Column("low_price", sa.Numeric(12, 4), default=0),
sa.Column("close_price", sa.Numeric(12, 4), default=0),
sa.Column("volume", sa.BigInteger(), default=0),
# Technical Indicators
sa.Column("ema_21", sa.Numeric(12, 4), default=0),
sa.Column("sma_50", sa.Numeric(12, 4), default=0),
sa.Column("sma_150", sa.Numeric(12, 4), default=0),
sa.Column("sma_200", sa.Numeric(12, 4), default=0),
sa.Column("rs_rating", sa.Numeric(5, 2), default=0),
sa.Column("avg_vol_30d", sa.Numeric(15, 2), default=0),
sa.Column("adr_pct", sa.Numeric(5, 2), default=0),
sa.Column("atr", sa.Numeric(12, 4), default=0),
# Pattern Analysis
sa.Column("pattern_type", sa.String(50)),
sa.Column("squeeze_status", sa.String(50)),
sa.Column("vcp_status", sa.String(50)),
sa.Column("entry_signal", sa.String(50)),
# Scoring
sa.Column("compression_score", sa.Integer(), default=0),
sa.Column("pattern_detected", sa.Integer(), default=0),
sa.Column("combined_score", sa.Integer(), default=0),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
# Create indexes for maverick stocks
op.create_index(
"mcp_maverick_stocks_combined_score_idx",
"mcp_maverick_stocks",
["combined_score"],
)
op.create_index(
"mcp_maverick_stocks_rs_rating_idx", "mcp_maverick_stocks", ["rs_rating"]
)
op.create_index(
"mcp_maverick_stocks_date_analyzed_idx",
"mcp_maverick_stocks",
["date_analyzed"],
)
op.create_index(
"mcp_maverick_stocks_stock_date_idx",
"mcp_maverick_stocks",
["stock_id", "date_analyzed"],
)
# 4. Create mcp_maverick_bear_stocks table
print("🐻 Creating mcp_maverick_bear_stocks table...")
op.create_table(
"mcp_maverick_bear_stocks",
sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
sa.Column(
"stock_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("mcp_stocks.stock_id"),
nullable=False,
index=True,
),
sa.Column("date_analyzed", sa.Date(), nullable=False),
# OHLCV Data
sa.Column("open_price", sa.Numeric(12, 4), default=0),
sa.Column("high_price", sa.Numeric(12, 4), default=0),
sa.Column("low_price", sa.Numeric(12, 4), default=0),
sa.Column("close_price", sa.Numeric(12, 4), default=0),
sa.Column("volume", sa.BigInteger(), default=0),
# Technical Indicators
sa.Column("rs_rating", sa.Numeric(5, 2), default=0),
sa.Column("ema_21", sa.Numeric(12, 4), default=0),
sa.Column("sma_50", sa.Numeric(12, 4), default=0),
sa.Column("sma_200", sa.Numeric(12, 4), default=0),
sa.Column("rsi_14", sa.Numeric(5, 2), default=0),
# MACD Indicators
sa.Column("macd", sa.Numeric(12, 6), default=0),
sa.Column("macd_signal", sa.Numeric(12, 6), default=0),
sa.Column("macd_histogram", sa.Numeric(12, 6), default=0),
# Bear Market Indicators
sa.Column("dist_days_20", sa.Integer(), default=0),
sa.Column("adr_pct", sa.Numeric(5, 2), default=0),
sa.Column("atr_contraction", sa.Boolean(), default=False),
sa.Column("atr", sa.Numeric(12, 4), default=0),
sa.Column("avg_vol_30d", sa.Numeric(15, 2), default=0),
sa.Column("big_down_vol", sa.Boolean(), default=False),
# Pattern Analysis
sa.Column("squeeze_status", sa.String(50)),
sa.Column("vcp_status", sa.String(50)),
# Scoring
sa.Column("score", sa.Integer(), default=0),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
# Create indexes for bear stocks
op.create_index(
"mcp_maverick_bear_stocks_score_idx", "mcp_maverick_bear_stocks", ["score"]
)
op.create_index(
"mcp_maverick_bear_stocks_rs_rating_idx",
"mcp_maverick_bear_stocks",
["rs_rating"],
)
op.create_index(
"mcp_maverick_bear_stocks_date_analyzed_idx",
"mcp_maverick_bear_stocks",
["date_analyzed"],
)
op.create_index(
"mcp_maverick_bear_stocks_stock_date_idx",
"mcp_maverick_bear_stocks",
["stock_id", "date_analyzed"],
)
# 5. Create mcp_supply_demand_breakouts table
print("📈 Creating mcp_supply_demand_breakouts table...")
op.create_table(
"mcp_supply_demand_breakouts",
sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
sa.Column(
"stock_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("mcp_stocks.stock_id"),
nullable=False,
index=True,
),
sa.Column("date_analyzed", sa.Date(), nullable=False),
# OHLCV Data
sa.Column("open_price", sa.Numeric(12, 4), default=0),
sa.Column("high_price", sa.Numeric(12, 4), default=0),
sa.Column("low_price", sa.Numeric(12, 4), default=0),
sa.Column("close_price", sa.Numeric(12, 4), default=0),
sa.Column("volume", sa.BigInteger(), default=0),
# Technical Indicators
sa.Column("ema_21", sa.Numeric(12, 4), default=0),
sa.Column("sma_50", sa.Numeric(12, 4), default=0),
sa.Column("sma_150", sa.Numeric(12, 4), default=0),
sa.Column("sma_200", sa.Numeric(12, 4), default=0),
sa.Column("rs_rating", sa.Numeric(5, 2), default=0),
sa.Column("avg_volume_30d", sa.Numeric(15, 2), default=0),
sa.Column("adr_pct", sa.Numeric(5, 2), default=0),
sa.Column("atr", sa.Numeric(12, 4), default=0),
# Pattern Analysis
sa.Column("pattern_type", sa.String(50)),
sa.Column("squeeze_status", sa.String(50)),
sa.Column("vcp_status", sa.String(50)),
sa.Column("entry_signal", sa.String(50)),
# Supply/Demand Analysis
sa.Column("accumulation_rating", sa.Numeric(5, 2), default=0),
sa.Column("distribution_rating", sa.Numeric(5, 2), default=0),
sa.Column("breakout_strength", sa.Numeric(5, 2), default=0),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
# Create indexes for supply/demand breakouts
op.create_index(
"mcp_supply_demand_breakouts_rs_rating_idx",
"mcp_supply_demand_breakouts",
["rs_rating"],
)
op.create_index(
"mcp_supply_demand_breakouts_date_analyzed_idx",
"mcp_supply_demand_breakouts",
["date_analyzed"],
)
op.create_index(
"mcp_supply_demand_breakouts_stock_date_idx",
"mcp_supply_demand_breakouts",
["stock_id", "date_analyzed"],
)
op.create_index(
"mcp_supply_demand_breakouts_ma_filter_idx",
"mcp_supply_demand_breakouts",
["close_price", "sma_50", "sma_150", "sma_200"],
)
# 6. Create mcp_technical_cache table
print("🔧 Creating mcp_technical_cache table...")
op.create_table(
"mcp_technical_cache",
sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
sa.Column(
"stock_id",
postgresql.UUID(as_uuid=True),
sa.ForeignKey("mcp_stocks.stock_id"),
nullable=False,
),
sa.Column("date", sa.Date(), nullable=False),
sa.Column("indicator_type", sa.String(50), nullable=False),
# Flexible indicator values
sa.Column("value", sa.Numeric(20, 8)),
sa.Column("value_2", sa.Numeric(20, 8)),
sa.Column("value_3", sa.Numeric(20, 8)),
# Metadata and parameters
sa.Column("metadata", sa.Text()),
sa.Column("period", sa.Integer()),
sa.Column("parameters", sa.Text()),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
)
# Create unique constraint and indexes for technical cache
op.create_unique_constraint(
"mcp_technical_cache_stock_date_indicator_unique",
"mcp_technical_cache",
["stock_id", "date", "indicator_type"],
)
op.create_index(
"mcp_technical_cache_stock_date_idx",
"mcp_technical_cache",
["stock_id", "date"],
)
op.create_index(
"mcp_technical_cache_indicator_idx", "mcp_technical_cache", ["indicator_type"]
)
op.create_index("mcp_technical_cache_date_idx", "mcp_technical_cache", ["date"])
print("✅ Self-contained schema created successfully!")
print("📋 Tables created:")
print(" - mcp_stocks (master stock data)")
print(" - mcp_price_cache (historical prices)")
print(" - mcp_maverick_stocks (maverick screening)")
print(" - mcp_maverick_bear_stocks (bear screening)")
print(" - mcp_supply_demand_breakouts (supply/demand analysis)")
print(" - mcp_technical_cache (technical indicators)")
print("🎯 Maverick-MCP is now completely self-contained!")
def downgrade():
"""Drop all self-contained tables."""
print("⚠️ Dropping self-contained maverick-mcp schema...")
# Drop tables in reverse order due to foreign key constraints
tables = [
"mcp_technical_cache",
"mcp_supply_demand_breakouts",
"mcp_maverick_bear_stocks",
"mcp_maverick_stocks",
"mcp_price_cache",
"mcp_stocks",
]
for table in tables:
print(f"🗑️ Dropping {table}...")
op.drop_table(table)
print("✅ Self-contained schema removed!")
```
--------------------------------------------------------------------------------
/tests/test_performance_optimizations.py:
--------------------------------------------------------------------------------
```python
"""
Tests for performance optimization systems.
This module tests the Redis connection pooling, request caching,
query optimization, and database index improvements.
"""
import asyncio
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from maverick_mcp.data.performance import (
QueryOptimizer,
RedisConnectionManager,
RequestCache,
cached,
get_performance_metrics,
initialize_performance_systems,
redis_manager,
request_cache,
)
from maverick_mcp.providers.optimized_stock_data import OptimizedStockDataProvider
from maverick_mcp.tools.performance_monitoring import (
analyze_database_indexes,
get_cache_performance_metrics,
get_comprehensive_performance_report,
get_redis_connection_health,
)
class TestRedisConnectionManager:
"""Test Redis connection pooling and management."""
@pytest.mark.asyncio
async def test_redis_manager_initialization(self):
"""Test Redis connection manager initialization."""
manager = RedisConnectionManager()
# Test initialization
with patch("redis.asyncio.ConnectionPool.from_url") as mock_pool:
mock_pool.return_value = MagicMock()
with patch("redis.asyncio.Redis") as mock_redis:
mock_client = AsyncMock()
mock_redis.return_value = mock_client
success = await manager.initialize()
assert success or not success # May fail in test environment
mock_pool.assert_called_once()
@pytest.mark.asyncio
async def test_redis_manager_health_check(self):
"""Test Redis health check functionality."""
manager = RedisConnectionManager()
with patch.object(manager, "_client") as mock_client:
mock_client.ping = AsyncMock()
# Test successful health check
result = await manager._health_check()
assert isinstance(result, bool)
@pytest.mark.asyncio
async def test_redis_manager_command_execution(self):
"""Test Redis command execution with error handling."""
manager = RedisConnectionManager()
with patch.object(manager, "get_client") as mock_get_client:
mock_client = AsyncMock()
mock_client.get.return_value = "test_value"
mock_get_client.return_value = mock_client
result = await manager.execute_command("get", "test_key")
assert result == "test_value"
mock_client.get.assert_called_once_with("test_key")
def test_redis_manager_metrics(self):
"""Test Redis connection metrics collection."""
manager = RedisConnectionManager()
metrics = manager.get_metrics()
assert isinstance(metrics, dict)
assert "healthy" in metrics
assert "initialized" in metrics
assert "connections_created" in metrics
class TestRequestCache:
"""Test request-level caching system."""
@pytest.mark.asyncio
async def test_cache_key_generation(self):
"""Test cache key generation from function arguments."""
cache = RequestCache()
key1 = cache._generate_cache_key("test", "arg1", "arg2", kwarg1="value1")
key2 = cache._generate_cache_key("test", "arg1", "arg2", kwarg1="value1")
key3 = cache._generate_cache_key("test", "arg1", "arg3", kwarg1="value1")
assert key1 == key2 # Same arguments should generate same key
assert key1 != key3 # Different arguments should generate different keys
assert key1.startswith("cache:test:")
@pytest.mark.asyncio
async def test_cache_ttl_configuration(self):
"""Test TTL configuration for different data types."""
cache = RequestCache()
assert cache._get_ttl("stock_data") == 3600
assert cache._get_ttl("technical_analysis") == 1800
assert cache._get_ttl("market_data") == 300
assert cache._get_ttl("unknown") == cache._default_ttls["default"]
@pytest.mark.asyncio
async def test_cache_operations(self):
"""Test basic cache operations."""
cache = RequestCache()
with patch.object(redis_manager, "get_client") as mock_get_client:
mock_client = AsyncMock()
mock_get_client.return_value = mock_client
# Test cache miss
mock_client.get.return_value = None
result = await cache.get("test_key")
assert result is None
# Test cache set
mock_client.setex.return_value = True
await cache.set("test_key", {"data": "value"}, ttl=60)
mock_client.setex.assert_called_once()
def test_cache_metrics(self):
"""Test cache metrics collection."""
cache = RequestCache()
cache._hit_count = 10
cache._miss_count = 5
metrics = cache.get_metrics()
assert metrics["hit_count"] == 10
assert metrics["miss_count"] == 5
assert metrics["total_requests"] == 15
assert metrics["hit_rate"] == 2 / 3
class TestQueryOptimizer:
"""Test database query optimization."""
def test_query_monitoring_decorator(self):
"""Test query monitoring decorator functionality."""
optimizer = QueryOptimizer()
@optimizer.monitor_query("test_query")
async def test_query():
await asyncio.sleep(0.1) # Simulate query time
return "result"
# This would need to be run to test properly
assert hasattr(test_query, "__name__")
def test_query_stats_collection(self):
"""Test query statistics collection."""
optimizer = QueryOptimizer()
# Simulate some query stats
optimizer._query_stats["test_query"] = {
"count": 5,
"total_time": 2.5,
"avg_time": 0.5,
"max_time": 1.0,
"min_time": 0.1,
}
stats = optimizer.get_query_stats()
assert "query_stats" in stats
assert "test_query" in stats["query_stats"]
assert stats["query_stats"]["test_query"]["avg_time"] == 0.5
class TestCachedDecorator:
"""Test the cached decorator functionality."""
@pytest.mark.asyncio
async def test_cached_decorator_basic(self):
"""Test basic cached decorator functionality."""
call_count = 0
@cached(data_type="test", ttl=60)
async def test_function(arg1, arg2="default"):
nonlocal call_count
call_count += 1
return f"result_{arg1}_{arg2}"
with patch.object(request_cache, "get") as mock_get:
with patch.object(request_cache, "set") as mock_set:
# First call - cache miss
mock_get.return_value = None
mock_set.return_value = True
result1 = await test_function("test", arg2="value")
assert result1 == "result_test_value"
assert call_count == 1
# Second call - cache hit
mock_get.return_value = "cached_result"
result2 = await test_function("test", arg2="value")
assert result2 == "cached_result"
assert call_count == 1 # Function not called again
class TestOptimizedStockDataProvider:
"""Test the optimized stock data provider."""
@pytest.fixture
def provider(self):
"""Create an optimized stock data provider instance."""
return OptimizedStockDataProvider()
@pytest.mark.asyncio
async def test_provider_caching_configuration(self, provider):
"""Test provider caching configuration."""
assert provider.cache_ttl_stock_data == 3600
assert provider.cache_ttl_screening == 7200
assert provider.cache_ttl_market_data == 300
@pytest.mark.asyncio
async def test_provider_performance_metrics(self, provider):
"""Test provider performance metrics collection."""
metrics = await provider.get_performance_metrics()
assert "cache_metrics" in metrics
assert "query_stats" in metrics
assert "cache_ttl_config" in metrics
class TestPerformanceMonitoring:
"""Test performance monitoring tools."""
@pytest.mark.asyncio
async def test_redis_health_check(self):
"""Test Redis health check functionality."""
with patch.object(redis_manager, "get_client") as mock_get_client:
mock_client = AsyncMock()
mock_client.set.return_value = True
mock_client.get.return_value = "test_value"
mock_client.delete.return_value = 1
mock_get_client.return_value = mock_client
health = await get_redis_connection_health()
assert "redis_health" in health
assert "connection_metrics" in health
assert "timestamp" in health
@pytest.mark.asyncio
async def test_cache_performance_metrics(self):
"""Test cache performance metrics collection."""
with patch.object(request_cache, "set") as mock_set:
with patch.object(request_cache, "get") as mock_get:
with patch.object(request_cache, "delete") as mock_delete:
mock_set.return_value = True
mock_get.return_value = {"test": "data", "timestamp": time.time()}
mock_delete.return_value = True
metrics = await get_cache_performance_metrics()
assert "cache_performance" in metrics
assert "performance_test" in metrics
@pytest.mark.asyncio
async def test_comprehensive_performance_report(self):
"""Test comprehensive performance report generation."""
with patch(
"maverick_mcp.tools.performance_monitoring.get_redis_connection_health"
) as mock_redis:
with patch(
"maverick_mcp.tools.performance_monitoring.get_cache_performance_metrics"
) as mock_cache:
with patch(
"maverick_mcp.tools.performance_monitoring.get_query_performance_metrics"
) as mock_query:
with patch(
"maverick_mcp.tools.performance_monitoring.analyze_database_indexes"
) as mock_indexes:
mock_redis.return_value = {
"redis_health": {"status": "healthy"}
}
mock_cache.return_value = {
"cache_performance": {"hit_rate": 0.85}
}
mock_query.return_value = {
"query_performance": {"query_stats": {}}
}
mock_indexes.return_value = {"poor_index_usage": []}
report = await get_comprehensive_performance_report()
assert "overall_health_score" in report
assert "component_scores" in report
assert "recommendations" in report
assert "detailed_metrics" in report
class TestPerformanceInitialization:
"""Test performance system initialization."""
@pytest.mark.asyncio
async def test_performance_systems_initialization(self):
"""Test initialization of all performance systems."""
with patch.object(redis_manager, "initialize") as mock_init:
mock_init.return_value = True
result = await initialize_performance_systems()
assert isinstance(result, dict)
assert "redis_manager" in result
assert "request_cache" in result
assert "query_optimizer" in result
@pytest.mark.asyncio
async def test_performance_metrics_collection(self):
"""Test comprehensive performance metrics collection."""
with patch.object(redis_manager, "get_metrics") as mock_redis_metrics:
with patch.object(request_cache, "get_metrics") as mock_cache_metrics:
mock_redis_metrics.return_value = {"healthy": True}
mock_cache_metrics.return_value = {"hit_rate": 0.8}
metrics = await get_performance_metrics()
assert "redis_manager" in metrics
assert "request_cache" in metrics
assert "query_optimizer" in metrics
assert "timestamp" in metrics
class TestDatabaseIndexAnalysis:
"""Test database index analysis functionality."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_database_index_analysis(self):
"""Test database index analysis (integration test)."""
try:
from maverick_mcp.data.models import get_db
session = next(get_db())
try:
# Test that the analysis doesn't crash
# Actual results depend on database state
recommendations = await analyze_database_indexes()
# Should return a dictionary structure
assert isinstance(recommendations, dict)
finally:
session.close()
except Exception as e:
# Database may not be available in test environment
pytest.skip(f"Database not available for integration test: {e}")
@pytest.mark.asyncio
async def test_performance_system_integration():
"""Test integration between all performance systems."""
# This is a basic integration test that ensures the systems can work together
try:
# Initialize systems
init_result = await initialize_performance_systems()
assert isinstance(init_result, dict)
# Get metrics
metrics = await get_performance_metrics()
assert isinstance(metrics, dict)
# Test caching
cache_result = await request_cache.set("test_key", "test_value", ttl=60)
await request_cache.get("test_key")
# Clean up
if cache_result:
await request_cache.delete("test_key")
except Exception as e:
# Some operations may fail in test environment
pytest.skip(f"Integration test skipped due to environment: {e}")
if __name__ == "__main__":
# Run basic tests
pytest.main([__file__, "-v"])
```