File size: 5,338 Bytes
00e2392 13f5f18 00e2392 13f5f18 00e2392 13f5f18 00e2392 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
"""
MCP Trader Server using FastMCP
"""
import sys
import os
# Add src to pythonpath so imports work
current_dir = os.path.dirname(os.path.abspath(__file__))
src_dir = os.path.dirname(os.path.dirname(current_dir))
if src_dir not in sys.path:
sys.path.append(src_dir)
from mcp.server.fastmcp import FastMCP
from typing import List, Dict, Any
from core.mcp_telemetry import log_usage, log_trace, log_metric
import uuid
import time
# Local imports (assuming src/mcp-trader is a package or run from src)
try:
from .data.market_data import get_market_data, get_current_price
from .data.fundamentals import get_fundamental_data
from .strategies.momentum import analyze_momentum
from .strategies.mean_reversion import analyze_mean_reversion
from .strategies.value import analyze_value
from .strategies.golden_cross import analyze_golden_cross
from .strategies.macd_crossover import analyze_macd_crossover
from .strategies.bollinger_squeeze import analyze_bollinger_squeeze
from .indicators.technical import calculate_sma, calculate_rsi, calculate_macd
except ImportError:
# Fallback if run directly and relative imports fail
from data.market_data import get_market_data, get_current_price
from data.fundamentals import get_fundamental_data
from strategies.momentum import analyze_momentum
from strategies.mean_reversion import analyze_mean_reversion
from strategies.value import analyze_value
from strategies.golden_cross import analyze_golden_cross
from strategies.macd_crossover import analyze_macd_crossover
from strategies.bollinger_squeeze import analyze_bollinger_squeeze
from indicators.technical import calculate_sma, calculate_rsi, calculate_macd
# Initialize FastMCP Server
mcp = FastMCP("MCP Trader", host="0.0.0.0")
@mcp.tool()
def get_stock_price(symbol: str) -> float:
"""Get the current price for a stock symbol."""
start_time = time.time()
trace_id = str(uuid.uuid4())
span_id = str(uuid.uuid4())
log_usage("mcp-trader", "get_stock_price")
try:
price = get_current_price(symbol)
duration = (time.time() - start_time) * 1000
# Metric
log_metric("mcp-trader", "stock_price", price, {"symbol": symbol})
log_trace("mcp-trader", trace_id, span_id, "get_stock_price", duration, "ok")
return price
except Exception as e:
duration = (time.time() - start_time) * 1000
log_trace("mcp-trader", trace_id, span_id, "get_stock_price", duration, "error")
raise e
@mcp.tool()
def get_stock_fundamentals(symbol: str) -> Dict[str, Any]:
"""Get fundamental data (PE, Market Cap, Sector) for a stock."""
log_usage("mcp-trader", "get_stock_fundamentals")
return get_fundamental_data(symbol)
@mcp.tool()
def get_momentum_strategy(symbol: str) -> Dict[str, Any]:
"""
Run Momentum Strategy analysis on a stock.
Returns Buy/Sell/Hold recommendation based on RSI, MACD, and Price Trend.
"""
log_usage("mcp-trader", "get_momentum_strategy")
return analyze_momentum(symbol)
@mcp.tool()
def get_mean_reversion_strategy(symbol: str) -> Dict[str, Any]:
"""
Run Mean Reversion Strategy analysis on a stock.
Returns Buy/Sell/Hold recommendation based on Bollinger Bands and RSI.
"""
return analyze_mean_reversion(symbol)
@mcp.tool()
def get_value_strategy(symbol: str) -> Dict[str, Any]:
"""
Run Value Strategy analysis on a stock.
Returns Buy/Sell/Hold recommendation based on fundamentals (PE, Dividend Yield).
"""
return analyze_value(symbol)
@mcp.tool()
def get_golden_cross_strategy(symbol: str) -> Dict[str, Any]:
"""
Run Golden Cross Strategy (Trend Following).
Detects SMA 50 crossing above/below SMA 200.
"""
return analyze_golden_cross(symbol)
@mcp.tool()
def get_macd_crossover_strategy(symbol: str) -> Dict[str, Any]:
"""
Run MACD Crossover Strategy (Momentum).
Detects MACD line crossing Signal line.
"""
return analyze_macd_crossover(symbol)
@mcp.tool()
def get_bollinger_squeeze_strategy(symbol: str) -> Dict[str, Any]:
"""
Run Bollinger Squeeze Strategy (Volatility).
Detects low volatility periods followed by potential breakouts.
"""
return analyze_bollinger_squeeze(symbol)
@mcp.tool()
def get_technical_summary(symbol: str) -> Dict[str, Any]:
"""
Get a summary of technical indicators for a stock (RSI, MACD, SMA).
"""
raw_data = get_market_data(symbol, period="3mo")
if not raw_data:
return {"error": "No data found"}
import pandas as pd
df = pd.DataFrame(raw_data)
return {
"symbol": symbol,
"price": df['close'].iloc[-1],
"rsi_14": calculate_rsi(df),
"sma_20": calculate_sma(df, 20),
"sma_50": calculate_sma(df, 50),
"macd": calculate_macd(df)
}
if __name__ == "__main__":
# Run the MCP server
import os
if os.environ.get("MCP_TRANSPORT") == "sse":
import uvicorn
port = int(os.environ.get("PORT", 7860))
uvicorn.run(mcp.sse_app(), host="0.0.0.0", port=port)
else:
mcp.run()
|