DEVELOPMENT.mdโข10.3 kB
# ๐ง Stock MCP ๅผๅๆๆกฃ
> ๆถๆ่ฎพ่ฎกใไบๆฌกๅผๅๅๆฉๅฑๆๅ
---
## ๐ ้กน็ฎๆถๆ
### ๆดไฝๆถๆ
```
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ Client Layer โ
โ (Claude Desktop / API Client / Web Dashboard) โ
โโโโโโโโโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ
โโโโโโโโโโโโโโโดโโโโโโโโโโโโโโ
โ โ
โผ โผ
โโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโ
โ MCP Server โ โ FastAPI โ
โ (Port 9999) โ โ (Port 9998) โ
โโโโโโโโโฌโโโโโโโโ โโโโโโโโโโฌโโโโโโโโโ
โ โ
โโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโ
โ
โโโโโโโโโโโโโโโผโโโโโโโโโโโโโโ
โ Service Layer โ
โ โโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ Market Service โ โ
โ โ Quote Service โ โ
โ โ News Service โ โ
โ โ Fundamentals Service โ โ
โ โ Tavily Service โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโ โ
โโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโ
โ
โโโโโโโโโโโโโโโผโโโโโโโโโโโโโโ
โ Data Source Layer โ
โ โโโโโโโโโโโโโโโโโโโโโโโโ โ
โ โ AKShare โ โ
โ โ Tushare โ โ
โ โ yFinance โ โ
โ โ Finnhub โ โ
โ โโโโโโโโโโโโโโโโโโโโโโโโ โ
โโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโ
โ
โโโโโโโโโโโโโโโผโโโโโโโโโโโโโโ
โ Redis Cache Layer โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
```
### ็ฎๅฝ็ปๆ
```
src/
โโโ server/
โ โโโ app.py # FastAPI ๅบ็จๅ
ฅๅฃ
โ โโโ mcp_server.py # MCP ๆๅกๅจ
โ โโโ routes/ # API ่ทฏ็ฑ
โ โ โโโ api_routes.py # RESTful ๆฅๅฃ
โ โ โโโ sse_routes.py # SSE ๆฅๅฃ
โ โโโ services/ # ไธๅก้ป่พๅฑ
โ โ โโโ market_service.py # ๅธๅบๆฐๆฎ
โ โ โโโ quote_service.py # ่กๆ
ๆๅก
โ โ โโโ news_service.py # ๆฐ้ปๆๅก
โ โ โโโ ...
โ โโโ utils/ # ๅทฅๅ
ท็ฑป
โ โโโ redis_cache.py # ็ผๅญ็ฎก็
โ โโโ symbol_processor.py # ่ก็ฅจไปฃ็ ๅค็
โ โโโ ...
โโโ config/
โโโ settings.py # ้
็ฝฎ็ฎก็
```
---
## ๐ ๏ธ ๆฌๅฐๅผๅ
### ๅผๅ็ฏๅข่ฎพ็ฝฎ
```bash
# 1. ๅๅปบ่ๆ็ฏๅข
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 2. ๅฎ่ฃ
ไพ่ต
pip install -r requirements.txt
pip install -r requirements-dev.txt # ๅผๅไพ่ต
# 3. ้
็ฝฎ็ฏๅขๅ้
cp .env.example .env
vim .env
# 4. ๅฏๅจ Redis๏ผDocker๏ผ
docker run -d -p 6379:6379 redis:alpine
# 5. ๅฏๅจๅผๅๆๅกๅจ
python main.py --http-port 9998 --mcp-port 9999
```
### ็ญ้่ฝฝๅผๅ
```bash
# ไฝฟ็จ uvicorn ็ญ้่ฝฝ
uvicorn src.server.app:app --reload --port 9998
```
---
## ๐ ๆทปๅ ๆฐๆฐๆฎๆบ
### 1. ๅๅปบๆๅก็ฑป
```python
# src/server/services/new_data_source_service.py
from typing import Dict, Any, Optional
from ..utils.redis_cache import cache_result
class NewDataSourceService:
"""ๆฐๆฐๆฎๆบๆๅก"""
def __init__(self):
self.api_key = os.getenv("NEW_API_KEY")
@cache_result(ttl=3600)
async def get_data(self, symbol: str) -> Dict[str, Any]:
"""่ทๅๆฐๆฎ๏ผๅธฆ็ผๅญ๏ผ"""
# ๅฎ็ฐๆฐๆฎ่ทๅ้ป่พ
return {
"symbol": symbol,
"data": "..."
}
```
### 2. ๆณจๅๅฐ่ทฏ็ฑ
```python
# src/server/routes/api_routes.py
from ..services.new_data_source_service import NewDataSourceService
new_service = NewDataSourceService()
@router.get("/api/new-endpoint")
async def get_new_data(symbol: str):
"""ๆฐๆฅๅฃ"""
try:
data = await new_service.get_data(symbol)
return {"code": 0, "data": data}
except Exception as e:
return {"code": 1, "error": str(e)}
```
### 3. ๆทปๅ MCP ๅทฅๅ
ท
```python
# src/server/mcp_server.py
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
if name == "new_tool":
symbol = arguments.get("symbol")
data = await new_service.get_data(symbol)
return [TextContent(type="text", text=json.dumps(data))]
```
---
## ๐งช ๆต่ฏ
### ๅๅ
ๆต่ฏ
```bash
# ่ฟ่กๆๆๆต่ฏ
pytest
# ่ฟ่ก็นๅฎๆต่ฏ
pytest tests/test_news_service.py
# ็ๆ่ฆ็็ๆฅๅ
pytest --cov=src --cov-report=html
```
### ๆต่ฏ็คบไพ
```python
# tests/test_quote_service.py
import pytest
from src.server.services.quote_service import QuoteService
@pytest.mark.asyncio
async def test_get_quote():
service = QuoteService()
result = await service.get_quote("600519")
assert result is not None
assert result["symbol"] == "600519"
assert "price" in result
```
### API ๆต่ฏ
```bash
# ไฝฟ็จ httpie
http GET localhost:9998/api/quote symbol==600519
# ไฝฟ็จ curl
curl "http://localhost:9998/api/quote?symbol=600519"
```
---
## ๐ ๆง่ฝไผๅ
### ็ผๅญ็ญ็ฅ
```python
# ไธๅๆฐๆฎ็ฑปๅไฝฟ็จไธๅ็็ผๅญๆถ้ด
CACHE_TTL_CONFIG = {
"realtime_quote": 60, # ๅฎๆถ่กๆ
1ๅ้
"daily_data": 3600, # ๆฅ็บฟๆฐๆฎ 1ๅฐๆถ
"financial": 86400, # ่ดขๅกๆฐๆฎ 1ๅคฉ
"company_info": 604800, # ๅ
ฌๅธไฟกๆฏ 1ๅจ
}
@cache_result(ttl=CACHE_TTL_CONFIG["realtime_quote"])
async def get_realtime_quote(symbol: str):
# ...
```
### ๅนถๅๅค็
```python
import asyncio
async def batch_get_quotes(symbols: list):
"""ๆน้่ทๅ่กๆ
"""
tasks = [get_quote(symbol) for symbol in symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
```
### ่ฟๆฅๆฑ
```python
# ไฝฟ็จ aiohttp ่ฟๆฅๆฑ
import aiohttp
session = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit=100),
timeout=aiohttp.ClientTimeout(total=30)
)
```
---
## ๐ ๅฎๅ
จๆไฝณๅฎ่ทต
### ็ฏๅขๅ้
```python
# โ ไธ่ฆ็กฌ็ผ็
API_KEY = "sk-abc123..."
# โ
ไฝฟ็จ็ฏๅขๅ้
import os
API_KEY = os.getenv("API_KEY")
```
### API ้ๆต
```python
from fastapi import HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.get("/api/quote")
@limiter.limit("100/minute")
async def get_quote(request: Request, symbol: str):
# ...
```
### ่พๅ
ฅ้ช่ฏ
```python
from pydantic import BaseModel, validator
class QuoteRequest(BaseModel):
symbol: str
market: str = "CN"
@validator("symbol")
def validate_symbol(cls, v):
if not v or len(v) > 10:
raise ValueError("Invalid symbol")
return v.upper()
```
---
## ๐ ไปฃ็ ้ฃๆ ผ
### ๆ ผๅผๅ
```bash
# ไฝฟ็จ black ๆ ผๅผๅไปฃ็
black src/
# ไฝฟ็จ isort ๆๅบๅฏผๅ
ฅ
isort src/
# ไฝฟ็จ flake8 ๆฃๆฅ
flake8 src/
```
### ็ฑปๅๆ็คบ
```python
from typing import Dict, List, Optional
def get_quote(symbol: str, market: str = "CN") -> Dict[str, Any]:
"""่ทๅ่ก็ฅจ่กๆ
Args:
symbol: ่ก็ฅจไปฃ็
market: ๅธๅบไปฃ็
Returns:
ๅ
ๅซ่กๆ
ๆฐๆฎ็ๅญๅ
ธ
"""
# ...
```
---
## ๐ ้จ็ฝฒ
### Docker ๆๅปบ
```bash
# ๆๅปบ้ๅ
docker build -t stock-mcp:latest .
# ๅคๆถๆๆๅปบ
docker buildx build --platform linux/amd64,linux/arm64 -t stock-mcp:latest .
```
### ๅฅๅบทๆฃๆฅ
```python
@app.get("/health")
async def health_check():
"""ๅฅๅบทๆฃๆฅๆฅๅฃ"""
redis_ok = await check_redis()
return {
"status": "healthy" if redis_ok else "degraded",
"redis": redis_ok,
"timestamp": datetime.now().isoformat()
}
```
---
## ๐ ่ฐ่ฏๆๅทง
### ๆฅๅฟ้
็ฝฎ
```python
import logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('logs/app.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
logger.debug("Debug message")
```
### ๆง่ฝๅๆ
```python
import time
from functools import wraps
def timing_decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
start = time.time()
result = await func(*args, **kwargs)
duration = time.time() - start
logger.info(f"{func.__name__} took {duration:.2f}s")
return result
return wrapper
@timing_decorator
async def slow_function():
# ...
```
---
## ๐ ๅ่่ตๆบ
- [FastAPI ๆๆกฃ](https://fastapi.tiangolo.com/)
- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk)
- [Redis Python ๅฎขๆท็ซฏ](https://redis-py.readthedocs.io/)
- [AKShare ๆๆกฃ](https://akshare.akfamily.xyz/)
---
<div align="center">
**๐ก ๆ็้ฎ๏ผๆฅ็ [ๅฎๆดๆๅ](GUIDE.md) ๆๆไบค [Issue](https://github.com/your-repo/issues)**
</div>