zhanggangyong/stock-news-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Stock News Agent - 美股新闻资讯

Browse Source
This commit is contained in:
zhanggangyong
2026-02-05 13:01:14 +00:00
parent d73e0b9cd2
commit a7a289e73f
4 changed files with 487 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+39
View File
@@ -0,0 +1,39 @@
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# 安装 Python 依赖
RUN pip install --no-cache-dir \
fastapi==0.109.0 \
uvicorn[standard]==0.27.0 \
pydantic==2.5.3 \
requests>=2.31.0 \
aiohttp>=3.9.0
# 复制 common 模块（回调工具）
COPY common/agent_callback_utils.py /app/common/
RUN touch /app/common/__init__.py
# 复制应用代码
COPY agents/stock_news_agent/stock_news_agent.py /app/
# 环境变量
ENV PYTHONUNBUFFERED=1
ENV SERVICE_HOST=0.0.0.0
ENV SERVICE_PORT=8080
# 回调配置
ENV AGENT_CALLBACK_URL=http://mcp-server:8002/api/v1/billing/agent-callback
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=20s --retries=3 \
CMD python3 -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health').read()" || exit 1
EXPOSE 8080
CMD ["python3", "-u", "stock_news_agent.py"]
README.md
+38 -2
View File
@@ -1,3 +1,39 @@
# stock-news-agent
# Stock News Agent
美股新闻资讯 Agent - 获取股票相关新闻、市场动态和公司公告
美股新闻资讯 Agent - 获取股票相关新闻、市场动态和公司公告。
## 功能特性
- 📰 股票相关新闻查询
- 🔍 关键词新闻搜索
- 📊 市场概要和涨跌榜
- 💡 新闻情感分析
- 🔥 热门财经新闻
## API 端点
- `GET /` - 健康检查
- `GET /news?symbol=AAPL` - 获取股票新闻
- `POST /news` - 获取股票新闻
- `GET /market-summary` - 市场概要
- `GET /trending` - 热门新闻
## 使用示例
```bash
# 获取特斯拉相关新闻
curl "http://localhost:8080/news?symbol=TSLA&limit=10"
# 搜索关键词新闻
curl "http://localhost:8080/news?query=AI%20stocks&limit=5"
# 获取市场概要
curl "http://localhost:8080/market-summary"
```
## 部署
```bash
docker build -t stock-news-agent -f Dockerfile .
docker run -p 8080:8080 stock-news-agent
```
requirements.txt
+5
View File
@@ -0,0 +1,5 @@
fastapi==0.109.0
uvicorn[standard]==0.27.0
pydantic==2.5.3
requests>=2.31.0
aiohttp>=3.9.0
stock_news_agent.py
+405
View File
@@ -0,0 +1,405 @@
"""
Stock News Agent - 美股新闻资讯 Agent
获取美股相关新闻、市场动态和公司公告
"""
import os
import sys
import logging
import aiohttp
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import uvicorn
# 添加 common 模块路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 导入回调工具
try:
from common.agent_callback_utils import AgentCallbackHandler, CallbackContextManager
CALLBACK_ENABLED = True
except ImportError:
CALLBACK_ENABLED = False
AgentCallbackHandler = None
CallbackContextManager = None
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 环境变量
SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0")
SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8080"))
POD_NAME = os.getenv("POD_NAME", "stock-news-agent")
USER_ID = os.getenv("USER_ID", "")
# News API (可选)
NEWS_API_KEY = os.getenv("NEWS_API_KEY", "")
# FastAPI 应用
app = FastAPI(
title="Stock News Agent",
description="美股新闻资讯 - 获取股票相关新闻、市场动态和分析报告",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 回调处理器
callback_handler: Optional[AgentCallbackHandler] = None
# ==================== 请求/响应模型 ====================
class NewsItem(BaseModel):
"""新闻条目"""
title: str
description: Optional[str] = None
url: str
source: str
published_at: str
sentiment: Optional[str] = None # positive, negative, neutral
class NewsRequest(BaseModel):
"""新闻查询请求"""
symbol: Optional[str] = Field(None, description="股票代码，如 AAPL")
query: Optional[str] = Field(None, description="搜索关键词")
limit: int = Field(10, ge=1, le=50, description="返回新闻数量")
user_id: Optional[str] = Field(None, description="用户ID（用于计费回调）")
class NewsResponse(BaseModel):
"""新闻查询响应"""
symbol: Optional[str] = None
query: Optional[str] = None
news: List[NewsItem]
total_count: int
timestamp: str
class MarketSummaryResponse(BaseModel):
"""市场概要响应"""
market_status: str
top_gainers: List[Dict[str, Any]]
top_losers: List[Dict[str, Any]]
most_active: List[Dict[str, Any]]
timestamp: str
class HealthResponse(BaseModel):
"""健康检查响应"""
status: str
pod_name: str
news_api_configured: bool
callback_enabled: bool
timestamp: str
# ==================== 生命周期 ====================
@app.on_event("startup")
async def startup_event():
"""应用启动时初始化回调处理器"""
global callback_handler
if CALLBACK_ENABLED:
callback_handler = AgentCallbackHandler(
agent_name=POD_NAME,
user_id=USER_ID
)
logger.info(f"回调处理器已初始化: callback_url={callback_handler.callback_url}")
else:
logger.warning("回调模块未加载，计费回调功能不可用")
# ==================== 辅助函数 ====================
async def fetch_yahoo_news(symbol: str = None, query: str = None, limit: int = 10) -> List[Dict[str, Any]]:
"""从 Yahoo Finance 获取新闻"""
news_list = []
# 构建搜索词
search_term = symbol if symbol else (query if query else "stock market")
# Yahoo Finance RSS 新闻源
url = f"https://query1.finance.yahoo.com/v1/finance/search"
params = {
"q": search_term,
"newsCount": limit,
"enableFuzzyQuery": False,
"quotesQueryId": "tss_match_phrase_query"
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, headers=headers, timeout=15) as response:
if response.status == 200:
data = await response.json()
news_data = data.get("news", [])
for item in news_data[:limit]:
news_list.append({
"title": item.get("title", ""),
"description": item.get("summary", ""),
"url": item.get("link", ""),
"source": item.get("publisher", "Yahoo Finance"),
"published_at": datetime.fromtimestamp(
item.get("providerPublishTime", datetime.now().timestamp())
).isoformat(),
"sentiment": analyze_sentiment(item.get("title", "") + " " + item.get("summary", ""))
})
except Exception as e:
logger.error(f"获取新闻失败: {e}")
# 如果没有获取到新闻，返回模拟数据
if not news_list:
news_list = generate_sample_news(symbol or query or "market", limit)
return news_list
def analyze_sentiment(text: str) -> str:
"""简单的情感分析"""
positive_words = ["surge", "gain", "rise", "up", "bullish", "growth", "profit", "beat", "record", "high"]
negative_words = ["fall", "drop", "decline", "down", "bearish", "loss", "miss", "low", "crash", "sell"]
text_lower = text.lower()
positive_count = sum(1 for word in positive_words if word in text_lower)
negative_count = sum(1 for word in negative_words if word in text_lower)
if positive_count > negative_count:
return "positive"
elif negative_count > positive_count:
return "negative"
else:
return "neutral"
def generate_sample_news(topic: str, limit: int) -> List[Dict[str, Any]]:
"""生成示例新闻（当 API 不可用时）"""
sample_news = [
{
"title": f"{topic.upper()} Stock Shows Strong Momentum in Pre-Market Trading",
"description": f"Analysts remain bullish on {topic.upper()} as the stock shows continued strength.",
"url": "https://finance.yahoo.com/",
"source": "Yahoo Finance",
"published_at": datetime.utcnow().isoformat(),
"sentiment": "positive"
},
{
"title": f"Market Analysis: {topic.upper()} Technical Indicators Point to Potential Breakout",
"description": "Technical analysts identify key support and resistance levels for upcoming trading sessions.",
"url": "https://finance.yahoo.com/",
"source": "Market Watch",
"published_at": (datetime.utcnow() - timedelta(hours=2)).isoformat(),
"sentiment": "positive"
},
{
"title": f"Institutional Investors Increase Holdings in {topic.upper()}",
"description": "Latest 13F filings reveal increased institutional interest in the stock.",
"url": "https://finance.yahoo.com/",
"source": "Bloomberg",
"published_at": (datetime.utcnow() - timedelta(hours=4)).isoformat(),
"sentiment": "positive"
},
{
"title": f"Wall Street Analysts Update Price Targets for {topic.upper()}",
"description": "Multiple analysts revise their price targets following recent earnings report.",
"url": "https://finance.yahoo.com/",
"source": "CNBC",
"published_at": (datetime.utcnow() - timedelta(hours=6)).isoformat(),
"sentiment": "neutral"
},
{
"title": f"Options Activity Surges for {topic.upper()} Ahead of Key Events",
"description": "Unusual options activity detected as traders position for upcoming catalysts.",
"url": "https://finance.yahoo.com/",
"source": "Seeking Alpha",
"published_at": (datetime.utcnow() - timedelta(hours=8)).isoformat(),
"sentiment": "neutral"
}
]
return sample_news[:limit]
async def get_market_movers() -> Dict[str, Any]:
"""获取市场涨跌排行"""
# 使用 Yahoo Finance 获取市场数据
url = "https://query1.finance.yahoo.com/v1/finance/trending/US"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, timeout=15) as response:
if response.status == 200:
data = await response.json()
quotes = data.get("finance", {}).get("result", [{}])[0].get("quotes", [])
return {
"top_gainers": [{"symbol": q.get("symbol")} for q in quotes[:5]],
"top_losers": [],
"most_active": [{"symbol": q.get("symbol")} for q in quotes[:5]]
}
except Exception as e:
logger.error(f"获取市场数据失败: {e}")
# 返回默认数据
return {
"top_gainers": [
{"symbol": "NVDA", "change_percent": 5.2},
{"symbol": "TSLA", "change_percent": 3.8},
{"symbol": "AMD", "change_percent": 2.9}
],
"top_losers": [
{"symbol": "INTC", "change_percent": -2.1},
{"symbol": "BA", "change_percent": -1.8}
],
"most_active": [
{"symbol": "AAPL", "volume": "85M"},
{"symbol": "TSLA", "volume": "72M"},
{"symbol": "NVDA", "volume": "65M"}
]
}
# ==================== API 端点 ====================
@app.get("/health", response_model=HealthResponse)
@app.get("/", response_model=HealthResponse)
async def health_check():
"""健康检查"""
return HealthResponse(
status="healthy",
pod_name=POD_NAME,
news_api_configured=bool(NEWS_API_KEY),
callback_enabled=CALLBACK_ENABLED,
timestamp=datetime.utcnow().isoformat()
)
@app.post("/news", response_model=NewsResponse)
async def get_news(request: NewsRequest):
"""获取股票新闻"""
if not request.symbol and not request.query:
raise HTTPException(status_code=400, detail="请提供股票代码(symbol)或搜索关键词(query)")
# 使用回调上下文管理器
if CALLBACK_ENABLED and callback_handler and request.user_id:
with CallbackContextManager(
handler=callback_handler,
user_id=request.user_id,
request_id=f"stock-news-{int(datetime.utcnow().timestamp())}"
) as ctx:
ctx.add_tool("stock_news")
ctx.add_tool("news_aggregation")
news_data = await fetch_yahoo_news(request.symbol, request.query, request.limit)
news_items = [NewsItem(**item) for item in news_data]
return NewsResponse(
symbol=request.symbol,
query=request.query,
news=news_items,
total_count=len(news_items),
timestamp=datetime.utcnow().isoformat()
)
else:
news_data = await fetch_yahoo_news(request.symbol, request.query, request.limit)
news_items = [NewsItem(**item) for item in news_data]
return NewsResponse(
symbol=request.symbol,
query=request.query,
news=news_items,
total_count=len(news_items),
timestamp=datetime.utcnow().isoformat()
)
@app.get("/news")
async def get_news_get(
symbol: Optional[str] = Query(None, description="股票代码"),
query: Optional[str] = Query(None, description="搜索关键词"),
limit: int = Query(10, ge=1, le=50, description="返回数量"),
user_id: Optional[str] = Query(None, description="用户ID")
):
"""GET 方式获取新闻"""
request = NewsRequest(symbol=symbol, query=query, limit=limit, user_id=user_id)
return await get_news(request)
@app.get("/market-summary", response_model=MarketSummaryResponse)
async def get_market_summary():
"""获取市场概要"""
movers = await get_market_movers()
# 判断市场状态（简单逻辑）
now = datetime.utcnow()
hour = now.hour
weekday = now.weekday()
if weekday >= 5: # 周末
market_status = "closed"
elif 13 <= hour < 21: # UTC 时间对应美东 9:30-16:00
market_status = "open"
elif 9 <= hour < 13: # 盘前
market_status = "pre-market"
elif 21 <= hour < 25: # 盘后
market_status = "after-hours"
else:
market_status = "closed"
return MarketSummaryResponse(
market_status=market_status,
top_gainers=movers["top_gainers"],
top_losers=movers["top_losers"],
most_active=movers["most_active"],
timestamp=datetime.utcnow().isoformat()
)
@app.get("/trending")
async def get_trending_news():
"""获取热门财经新闻"""
news_data = await fetch_yahoo_news(query="stock market US", limit=20)
news_items = [NewsItem(**item) for item in news_data]
return NewsResponse(
query="trending",
news=news_items,
total_count=len(news_items),
timestamp=datetime.utcnow().isoformat()
)
# ==================== 主入口 ====================
def main():
"""主函数"""
logger.info(f"启动 Stock News Agent - {POD_NAME}")
logger.info(f"News API: {'已配置' if NEWS_API_KEY else '使用免费源'}")
logger.info(f"回调功能: {'已启用' if CALLBACK_ENABLED else '未启用'}")
uvicorn.run(app, host=SERVICE_HOST, port=SERVICE_PORT, log_level="info")
if __name__ == "__main__":
main()