52 KiB
52 KiB
AI 助手 数据管道实施指南
完整代码框架和集成步骤
预计投入: 110小时（3周） 难度: 中等 优先级: 🔴 高（直接影响产品竞争力）
目录
Phase 1: 基本面数据（第1周）
步骤1: 创建基本面数据获取模块
文件: /root/work/stock/backend/app/modules/fundamental_data.py
"""
基本面数据获取模块
获取财务基本面数据用于AI助手的深度分析
"""
import tushare as ts
import pandas as pd
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
class FundamentalDataFetcher:
"""获取股票基本面数据的主类"""
def __init__(self, tushare_token: str):
"""
初始化获取器
Args:
tushare_token: TuShare API token
"""
self.ts = ts.pro_api(tushare_token)
self.logger = logger
def get_stock_fundamentals(self, symbol: str, date: Optional[str] = None) -> Dict:
"""
获取单只股票的完整基本面数据
Args:
symbol: 股票代码 (e.g., "600000")
date: 数据日期，默认最新
Returns:
Dict包含所有基本面指标
"""
try:
# 转换格式（6位编号 → 股票代码）
ts_code = f"{symbol}.SH" if symbol.startswith("6") else f"{symbol}.SZ"
fundamentals = {
'symbol': symbol,
'ts_code': ts_code,
'timestamp': datetime.now().isoformat(),
}
# 1. 获取基本面快照数据
logger.info(f"获取 {symbol} 的基本面快照数据...")
daily_basic = self._fetch_daily_basic(ts_code, date)
if daily_basic is not None:
fundamentals.update(daily_basic)
# 2. 获取财务数据
logger.info(f"获取 {symbol} 的财务数据...")
income_data = self._fetch_income(ts_code)
if income_data is not None:
fundamentals.update(income_data)
# 3. 获取资产负债数据
logger.info(f"获取 {symbol} 的资产负债数据...")
balance_data = self._fetch_balance(ts_code)
if balance_data is not None:
fundamentals.update(balance_data)
# 4. 计算衍生指标
logger.info(f"计算 {symbol} 的衍生指标...")
derived_metrics = self._calculate_derived_metrics(fundamentals)
fundamentals.update(derived_metrics)
# 5. 获取行业对标数据
logger.info(f"获取 {symbol} 的行业对标数据...")
industry_bench = self._fetch_industry_benchmarks(ts_code, fundamentals)
if industry_bench is not None:
fundamentals.update(industry_bench)
return fundamentals
except Exception as e:
self.logger.error(f"获取 {symbol} 基本面数据失败: {str(e)}")
return {}
def _fetch_daily_basic(self, ts_code: str, date: Optional[str] = None) -> Optional[Dict]:
"""
获取日基本面数据（PE, PB, 市值等）
"""
try:
# 使用TuShare的daily_basic接口
df = self.ts.daily_basic(
ts_code=ts_code,
trade_date=date,
fields='ts_code,trade_date,pe,pe_ttm,pb,dv_ratio,dv_ttm,total_mv,circ_mv'
)
if df.empty:
self.logger.warning(f"{ts_code} 没有基本面数据")
return None
row = df.iloc[0]
return {
'pe': float(row['pe']) if pd.notna(row['pe']) else None,
'pe_ttm': float(row['pe_ttm']) if pd.notna(row['pe_ttm']) else None,
'pb': float(row['pb']) if pd.notna(row['pb']) else None,
'dv_ratio': float(row['dv_ratio']) if pd.notna(row['dv_ratio']) else None,
'total_market_cap': float(row['total_mv']) if pd.notna(row['total_mv']) else None,
'circulation_market_cap': float(row['circ_mv']) if pd.notna(row['circ_mv']) else None,
}
except Exception as e:
self.logger.error(f"获取 {ts_code} daily_basic 失败: {str(e)}")
return None
def _fetch_income(self, ts_code: str, periods: int = 4) -> Optional[Dict]:
"""
获取财务报表数据（净利润、营收等）
Args:
ts_code: 股票代码
periods: 获取最近N期数据
"""
try:
# 获取最近期数的财务数据
df = self.ts.income(
ts_code=ts_code,
start_date=None,
end_date=None,
fields='ts_code,ann_date,f_ann_date,end_date,report_type,comp_type,'
'revenue,operate_profit,n_income,n_income_attr_p,n_income_attr_s'
)
if df.empty:
self.logger.warning(f"{ts_code} 没有收入数据")
return None
# 按公告日期排序，取最新的
df['f_ann_date'] = pd.to_datetime(df['f_ann_date'])
df = df.sort_values('f_ann_date', ascending=False)
current_period = df.iloc[0]
prev_period = df.iloc[1] if len(df) > 1 else None
# 计算增速
n_income_current = float(current_period['n_income']) if pd.notna(current_period['n_income']) else 0
n_income_prev = float(prev_period['n_income']) if prev_period is not None and pd.notna(prev_period['n_income']) else None
growth_rate = None
if n_income_prev and n_income_prev > 0:
growth_rate = (n_income_current / n_income_prev - 1) * 100
return {
'revenue': float(current_period['revenue']) if pd.notna(current_period['revenue']) else None,
'operate_profit': float(current_period['operate_profit']) if pd.notna(current_period['operate_profit']) else None,
'net_profit': n_income_current,
'net_profit_attr_parent': float(current_period['n_income_attr_p']) if pd.notna(current_period['n_income_attr_p']) else None,
'net_profit_growth_rate': growth_rate, # 同比增长率 %
'report_type': str(current_period['report_type']),
}
except Exception as e:
self.logger.error(f"获取 {ts_code} 收入数据失败: {str(e)}")
return None
def _fetch_balance(self, ts_code: str) -> Optional[Dict]:
"""
获取资产负债表数据（总资产、负债等）
"""
try:
df = self.ts.balancesheet(
ts_code=ts_code,
start_date=None,
end_date=None,
fields='ts_code,ann_date,f_ann_date,total_assets,total_liab,'
'total_hldr_eqy_exc_min_int,total_hldr_eqy_inc_min_int'
)
if df.empty:
return None
# 取最新期
df['f_ann_date'] = pd.to_datetime(df['f_ann_date'])
row = df.sort_values('f_ann_date', ascending=False).iloc[0]
return {
'total_assets': float(row['total_assets']) if pd.notna(row['total_assets']) else None,
'total_liabilities': float(row['total_liab']) if pd.notna(row['total_liab']) else None,
'total_shareholders_equity': float(row['total_hldr_eqy_exc_min_int']) if pd.notna(row['total_hldr_eqy_exc_min_int']) else None,
}
except Exception as e:
self.logger.error(f"获取 {ts_code} 资产负债表失败: {str(e)}")
return None
def _calculate_derived_metrics(self, fundamentals: Dict) -> Dict:
"""
基于原始数据计算衍生指标
Returns:
包含新衍生指标的字典
"""
derived = {}
# ROE (净资产收益率) = 净利润 / 总股东权益
if fundamentals.get('net_profit') and fundamentals.get('total_shareholders_equity'):
if fundamentals['total_shareholders_equity'] > 0:
roe = (fundamentals['net_profit'] / fundamentals['total_shareholders_equity']) * 100
derived['roe'] = roe
# 净债务率 = (负债 - 现金) / 总资产
if fundamentals.get('total_liabilities') and fundamentals.get('total_assets'):
if fundamentals['total_assets'] > 0:
net_debt_ratio = (fundamentals['total_liabilities'] / fundamentals['total_assets']) * 100
derived['net_debt_ratio'] = net_debt_ratio
# PEG = PE / 增长率 (如果增长率>0)
if fundamentals.get('pe_ttm') and fundamentals.get('net_profit_growth_rate'):
if fundamentals['net_profit_growth_rate'] > 0:
peg = fundamentals['pe_ttm'] / fundamentals['net_profit_growth_rate']
derived['peg'] = peg
# 毛利率 = (营收 - 营业利润) / 营收
if fundamentals.get('revenue') and fundamentals.get('operate_profit'):
if fundamentals['revenue'] > 0:
gross_margin = ((fundamentals['revenue'] - fundamentals['operate_profit']) / fundamentals['revenue']) * 100
derived['gross_margin'] = gross_margin
return derived
def _fetch_industry_benchmarks(self, ts_code: str, fundamentals: Dict) -> Optional[Dict]:
"""
获取行业对标数据，方便对比
Returns:
包含行业平均值和排名的字典
"""
try:
# 从fundamentals中获取行业信息
# 注：实际需要从其他数据源获取行业信息
# 这里是简化示意，实际可以调用TuShare的行业接口
# 获取行业日行情（行业平均PE等）
df = self.ts.gn_daily(
ts_code=ts_code,
start_date=None,
end_date=None,
fields='ts_code,trade_date,gn_code,gn_name,pe_avg,pb_avg,roe_avg'
)
if df.empty:
return None
row = df.iloc[0]
return {
'industry_name': str(row['gn_name']),
'industry_code': str(row['gn_code']),
'industry_pe_avg': float(row['pe_avg']) if pd.notna(row['pe_avg']) else None,
'industry_pb_avg': float(row['pb_avg']) if pd.notna(row['pb_avg']) else None,
'industry_roe_avg': float(row['roe_avg']) if pd.notna(row['roe_avg']) else None,
}
except Exception as e:
self.logger.warning(f"获取 {ts_code} 行业对标数据失败: {str(e)}")
return None
def get_multiple_stocks_fundamentals(self, symbols: List[str]) -> Dict[str, Dict]:
"""
批量获取多只股票的基本面数据
Args:
symbols: 股票代码列表
Returns:
{symbol: fundamentals_dict}
"""
results = {}
for symbol in symbols:
self.logger.info(f"正在处理 {symbol}...")
results[symbol] = self.get_stock_fundamentals(symbol)
return results
# 使用示例
if __name__ == "__main__":
# 配置日志
logging.basicConfig(level=logging.INFO)
# 初始化
fetcher = FundamentalDataFetcher(
tushare_token="your_tushare_token"
)
# 获取单只股票
fundamentals = fetcher.get_stock_fundamentals("600000")
print("600000 基本面数据:")
print(fundamentals)
# 批量获取
symbols = ["600000", "600031", "601988", "601398"]
all_fundamentals = fetcher.get_multiple_stocks_fundamentals(symbols)
print("\n批量获取结果:")
for symbol, data in all_fundamentals.items():
print(f"{symbol}: {data}")
步骤2: 创建数据库表
文件: /root/work/stock/backend/app/database_migrations.py
"""
数据库迁移脚本 - 为AI助手添加新表
"""
from sqlalchemy import create_engine, Column, String, Float, Integer, DateTime, Text, BigInteger
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
Base = declarative_base()
class StockFundamentals(Base):
"""股票基本面数据表"""
__tablename__ = "stock_fundamentals"
id = Column(Integer, primary_key=True)
symbol = Column(String(20), nullable=False, index=True)
ts_code = Column(String(20), nullable=False)
date = Column(DateTime, nullable=False, index=True)
# 快照数据
pe = Column(Float)
pe_ttm = Column(Float)
pb = Column(Float)
total_market_cap = Column(Float) # 总市值
circulation_market_cap = Column(Float) # 流通市值
# 财务数据
revenue = Column(BigInteger) # 营收
operate_profit = Column(BigInteger) # 营业利润
net_profit = Column(BigInteger) # 净利润
net_profit_growth_rate = Column(Float) # 净利润增速 %
# 资产负债
total_assets = Column(BigInteger) # 总资产
total_liabilities = Column(BigInteger) # 总负债
total_shareholders_equity = Column(BigInteger) # 总股东权益
# 衍生指标
roe = Column(Float) # 净资产收益率 %
net_debt_ratio = Column(Float) # 净债务率 %
peg = Column(Float) # PE/Growth
gross_margin = Column(Float) # 毛利率 %
# 行业对标
industry_name = Column(String(100))
industry_code = Column(String(20))
industry_pe_avg = Column(Float)
industry_pb_avg = Column(Float)
industry_roe_avg = Column(Float)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
{"schema": "stock_data"}, # 指定schema
)
class MarketSentiment(Base):
"""市场情绪数据表"""
__tablename__ = "market_sentiment"
id = Column(Integer, primary_key=True)
symbol = Column(String(20), nullable=False, index=True)
timestamp = Column(DateTime, nullable=False, index=True)
# 资金面
main_net_inflow = Column(BigInteger) # 主力净流入
retail_net_inflow = Column(BigInteger) # 散户净流入
institutional_holding_ratio = Column(Float) # 机构持仓占比 %
# 融资融券
margin_balance = Column(BigInteger) # 融资余额
short_balance = Column(BigInteger) # 融券余额
# 成交数据
volume = Column(Integer) # 成交量
turnover_ratio = Column(Float) # 换手率 %
daily_change = Column(Float) # 日涨幅 %
# 情绪评分
sentiment_score = Column(Integer) # -100 ~ 100
created_at = Column(DateTime, default=datetime.utcnow)
__table_args__ = (
{"schema": "stock_data"},
)
class AIKnowledgeCache(Base):
"""AI知识库缓存表"""
__tablename__ = "ai_knowledge_cache"
id = Column(Integer, primary_key=True)
symbol = Column(String(20), nullable=False, index=True)
cache_key = Column(String(200), nullable=False)
knowledge_type = Column(String(100)) # "fundamentals_summary", "sentiment_analysis", etc.
content = Column(Text) # 知识内容
relevance_score = Column(Float) # 相关性评分 0-1
created_at = Column(DateTime, default=datetime.utcnow)
expires_at = Column(DateTime) # 过期时间
__table_args__ = (
{"schema": "stock_data"},
)
def create_tables(database_url: str):
"""创建新表"""
engine = create_engine(database_url)
# 创建schema
with engine.connect() as conn:
conn.execute("CREATE SCHEMA IF NOT EXISTS stock_data")
conn.commit()
# 创建表
Base.metadata.create_all(engine)
logger.info("数据库表创建成功")
def migrate_existing_data(session):
"""迁移现有数据（如果有的话）"""
pass
# 使用示例
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
database_url = "postgresql://user:password@localhost:5432/stock_db"
create_tables(database_url)
步骤3: 修改AI助手路由，使用新数据
文件: /root/work/stock/backend/app/routes/ai_assistant.py (修改)
"""
增强的AI助手路由 - 集成新的基本面数据
"""
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
import logging
from datetime import datetime
from app.config import get_settings
from app.modules.fundamental_data import FundamentalDataFetcher
from app.modules.cache_manager import cache
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/ai", tags=["AI Assistant"])
settings = get_settings()
class ChatMessage(BaseModel):
role: str # "user" or "assistant"
content: str
class ChatRequest(BaseModel):
messages: List[ChatMessage] = []
query: str
symbol: Optional[str] = None # 如果问的是某只股票
class ChatResponse(BaseModel):
response: str
thinking: str = ""
data_sources: List[str] = [] # 使用的数据源
def get_fundamental_fetcher():
"""依赖注入 - 获取基本面数据获取器"""
return FundamentalDataFetcher(settings.tushare_token)
@router.post("/chat")
async def chat(
request: ChatRequest,
fetcher: FundamentalDataFetcher = Depends(get_fundamental_fetcher)
) -> ChatResponse:
"""
增强的AI助手 - 支持基本面分析
"""
try:
query_lower = request.query.lower()
data_sources = ["Contextual Knowledge"]
# 如果问题涉及具体股票，获取其基本面数据
fundamentals_context = None
if request.symbol:
cache_key = f"ai:fundamentals:{request.symbol}"
fundamentals_context = cache.get(cache_key)
if not fundamentals_context:
logger.info(f"从TuShare获取 {request.symbol} 的基本面数据...")
fundamentals_context = fetcher.get_stock_fundamentals(request.symbol)
if fundamentals_context:
# 缓存24小时
cache.set(cache_key, fundamentals_context, ex=86400)
data_sources.append("TuShare")
# 生成增强的回答
response_text = generate_enhanced_response(
query=request.query,
fundamentals=fundamentals_context,
symbols_in_query=extract_symbols(query_lower)
)
return ChatResponse(
response=response_text,
thinking="",
data_sources=data_sources
)
except Exception as e:
logger.error(f"AI助手错误: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"处理请求失败: {str(e)}"
)
def extract_symbols(query: str) -> List[str]:
"""从查询中提取股票代码"""
import re
# 匹配6位数字（A股代码）
symbols = re.findall(r'\b[036]\d{5}\b', query)
return symbols
def generate_enhanced_response(
query: str,
fundamentals: Optional[dict],
symbols_in_query: List[str]
) -> str:
"""
生成增强的回答，包含基本面分析
"""
query_lower = query.lower()
# PE相关问题
if "pe" in query_lower or "市盈率" in query_lower:
if fundamentals:
pe = fundamentals.get('pe_ttm')
industry_pe = fundamentals.get('industry_pe_avg')
industry_name = fundamentals.get('industry_name', '该行业')
if pe and industry_pe:
if pe < industry_pe:
return f"""这只股票的PE比率为 {pe:.2f}，低于{industry_name}平均的 {industry_pe:.2f}。
**分析：**
- 相对估值**偏低**，可能被市场低估
- 但需要考虑为何低估：
✓ 可能是基本面向好的机会
✗ 也可能反映市场对其前景的担忧
**建议：**
检查净利润增速和ROE，判断是价值陷阱还是真正的机会。"""
return f"""PE比率（市盈率）= 股价 / 每股收益
**当前数据不足，无法对比。请提供股票代码。**"""
# ROE相关问题
if "roe" in query_lower or "净资产收益率" in query_lower:
if fundamentals and fundamentals.get('roe'):
roe = fundamentals['roe']
return f"""这只股票的ROE（净资产收益率）为 {roe:.2f}%。
**ROE分析：**
- ROE > 15%：优秀，公司盈利能力强
- ROE 10-15%：良好，合理范围
- ROE < 10%：一般，需要关注
当前公司的ROE属于{"优秀" if roe > 15 else "良好" if roe > 10 else "需要改进"}水平。"""
return "无法获取ROE数据，请提供股票代码。"
# 增长性问题
if "增速" in query_lower or "增长" in query_lower:
if fundamentals and fundamentals.get('net_profit_growth_rate'):
growth = fundamentals['net_profit_growth_rate']
return f"""这只股票的净利润增速为 {growth:.2f}%（同比）。
**增长性评估：**
- 增速 > 30%：高增长，市场看好，可能有估值溢价
- 增速 15-30%：良好增长，合理的增长期望
- 增速 0-15%：缓慢增长，需要关注转机
- 增速 < 0%：负增长，风险信号
**投资建议：**
- 结合PE和PEG判断是否有投资价值
- 关注未来增长前景"""
return "无法获取增长数据，请提供股票代码。"
# 默认回答
return f"""关于您的问题："{query}"
我现在可以帮助您分析以下方面：
- **PE比率** - 相对估值分析
- **ROE** - 盈利能力评估
- **增速** - 增长潜力分析
- **风险** - 投资风险管理
如果您想深度分析某只股票，请提供股票代码（如"600000"）。"""
@router.get("/stock-analysis/{symbol}")
async def get_stock_analysis(
symbol: str,
fetcher: FundamentalDataFetcher = Depends(get_fundamental_fetcher)
) -> dict:
"""
获取某只股票的详细分析数据（给AI助手使用）
"""
try:
cache_key = f"stock_analysis:{symbol}"
cached = cache.get(cache_key)
if cached:
return cached
# 获取基本面数据
fundamentals = fetcher.get_stock_fundamentals(symbol)
analysis = {
"symbol": symbol,
"timestamp": datetime.now().isoformat(),
"fundamentals": fundamentals,
"ready_for_ai": bool(fundamentals),
}
# 缓存12小时
cache.set(cache_key, analysis, ex=43200)
return analysis
except Exception as e:
logger.error(f"获取 {symbol} 分析数据失败: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
步骤4: 集成到主应用
文件: /root/work/stock/backend/main.py (修改)
在应用启动时初始化基本面数据获取器：
# 在应用启动事件中添加
from app.modules.fundamental_data import FundamentalDataFetcher
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# Startup
logger.info("🚀 启动股票推荐系统...")
# 初始化基本面数据获取器
try:
fundamental_fetcher = FundamentalDataFetcher(settings.tushare_token)
logger.info("✅ 基本面数据获取器初始化成功")
app.state.fundamental_fetcher = fundamental_fetcher
except Exception as e:
logger.warning(f"⚠️ 基本面数据获取器初始化失败: {e}")
yield
# Shutdown
logger.info("🛑 关闭股票推荐系统...")
测试Phase 1
# 1. 测试基本面数据获取
curl -X GET "http://localhost:8001/api/ai/stock-analysis/600000"
# 2. 测试AI助手与基本面数据的交互
curl -X POST "http://localhost:8001/api/ai/chat" \
-H "Content-Type: application/json" \
-d '{
"messages": [],
"query": "600000的PE比率如何",
"symbol": "600000"
}'
# 3. 检查数据库
psql -U stock_user -d stock_db -c "SELECT * FROM stock_data.stock_fundamentals LIMIT 1;"
Phase 2: 情绪和消息数据（第2周）
步骤1: 创建市场情绪爬虫
文件: /root/work/stock/backend/app/modules/sentiment_fetcher.py
"""
市场情绪数据获取模块
实时获取资金面、成交量等市场情绪指标
"""
import aiohttp
import asyncio
from typing import Dict, Optional
from datetime import datetime
import logging
import re
import json
logger = logging.getLogger(__name__)
class MarketSentimentFetcher:
"""获取市场情绪的爬虫类"""
def __init__(self):
self.sina_url_base = "https://hq.sinajs.cn/list="
self.eastmoney_url = "https://emweb.securities.eastmoney.com/api/qt/stock/get"
async def get_sentiment(self, symbol: str, update_frequency: str = "5min") -> Dict:
"""
获取股票的实时市场情绪
Args:
symbol: 股票代码（6位数）
update_frequency: 更新频率
Returns:
包含情绪评分的字典
"""
try:
# 并行获取新浪和东方财富数据
sina_task = self._fetch_sina_data(symbol)
eastmoney_task = self._fetch_eastmoney_data(symbol)
sina_data, eastmoney_data = await asyncio.gather(
sina_task,
eastmoney_task,
return_exceptions=True
)
# 合并数据
sentiment = {
'symbol': symbol,
'timestamp': datetime.now().isoformat(),
'sina': sina_data if isinstance(sina_data, dict) else {},
'eastmoney': eastmoney_data if isinstance(eastmoney_data, dict) else {},
}
# 计算综合情绪评分
sentiment['sentiment_score'] = self._calculate_sentiment_score(
sina_data if isinstance(sina_data, dict) else {},
eastmoney_data if isinstance(eastmoney_data, dict) else {}
)
return sentiment
except Exception as e:
logger.error(f"获取 {symbol} 情绪数据失败: {str(e)}")
return {}
async def _fetch_sina_data(self, symbol: str) -> Optional[Dict]:
"""
从新浪财经获取实时行情数据
"""
try:
# 转换格式
sina_code = f"sh{symbol[:-3]}" if symbol.startswith("6") else f"sz{symbol[:-3]}"
url = f"{self.sina_url_base}{sina_code}"
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
if resp.status != 200:
logger.warning(f"新浪获取失败: {resp.status}")
return None
text = await resp.text()
# 解析新浪数据格式
# var hq_str_sh000001="上证指数,3217.32,..."
match = re.search(r'"([^"]+)"', text)
if not match:
return None
parts = match.group(1).split(',')
if len(parts) < 9:
return None
return {
'name': parts[0],
'price': float(parts[1]),
'change': float(parts[2]),
'change_pct': float(parts[3]),
'volume': int(float(parts[8])), # 成交量
'amount': float(parts[9]) if len(parts) > 9 else 0, # 成交金额
}
except asyncio.TimeoutError:
logger.warning(f"新浪获取 {symbol} 超时")
return None
except Exception as e:
logger.error(f"新浪数据解析失败: {str(e)}")
return None
async def _fetch_eastmoney_data(self, symbol: str) -> Optional[Dict]:
"""
从东方财富获取资金面数据
"""
try:
# 转换格式
secid = f"1.{symbol[:-3]}" if symbol.startswith("6") else f"0.{symbol[:-3]}"
params = {
'secid': secid,
'ut': 'fa5fd1943c7b86f57f711e9e99cf21af',
'fields': 'f57,f58,f107,f116,f152,f161,f163,f164,f174,f167,f168,f169'
}
async with aiohttp.ClientSession() as session:
async with session.get(
self.eastmoney_url,
params=params,
timeout=aiohttp.ClientTimeout(total=5)
) as resp:
if resp.status != 200:
logger.warning(f"东方财富获取失败: {resp.status}")
return None
data = await resp.json()
if data.get('code') != 0 or not data.get('data'):
return None
quote = data['data']['quote']
return {
'turnover_ratio': float(quote.get('f168', 0)) / 100, # 换手率
'volume_ratio': float(quote.get('f169', 0)), # 量比
'pe': float(quote.get('f107', 0)),
'pb': float(quote.get('f116', 0)),
'market_cap': float(quote.get('f161', 0)), # 总市值
}
except asyncio.TimeoutError:
logger.warning(f"东方财富获取 {symbol} 超时")
return None
except Exception as e:
logger.error(f"东方财富数据解析失败: {str(e)}")
return None
def _calculate_sentiment_score(self, sina_data: Dict, eastmoney_data: Dict) -> int:
"""
基于多个指标计算情绪评分
Returns:
-100 ~ 100 的情绪评分
"""
score = 0
# 1. 价格动量 (-50 ~ 50)
if sina_data.get('change_pct'):
pct = sina_data['change_pct']
if pct > 5:
score += 40
elif pct > 2:
score += 20
elif pct < -5:
score -= 40
elif pct < -2:
score -= 20
# 2. 换手率 (-20 ~ 20)
if eastmoney_data.get('turnover_ratio'):
ratio = eastmoney_data['turnover_ratio']
if ratio > 0.05: # 换手率 > 5%
score += 15
elif ratio > 0.02:
score += 8
elif ratio < 0.01:
score -= 10
# 3. 量比 (-20 ~ 20)
if eastmoney_data.get('volume_ratio'):
vr = eastmoney_data['volume_ratio']
if vr > 1.5:
score += 10
elif vr < 0.5:
score -= 10
return max(-100, min(100, score))
def interpret_sentiment_score(self, score: int) -> str:
"""将情绪评分转换为文字描述"""
if score > 60:
return "极度看涨（市场极度乐观）"
elif score > 30:
return "看涨（市场情绪良好）"
elif score > -30:
return "中性（市场情绪平稳）"
elif score > -60:
return "看跌（市场情绪转弱）"
else:
return "极度看跌（市场极度悲观）"
# 使用示例
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
async def test():
fetcher = MarketSentimentFetcher()
sentiment = await fetcher.get_sentiment("600000")
print(f"600000 情绪数据: {json.dumps(sentiment, indent=2, ensure_ascii=False)}")
print(f"情绪评分: {sentiment['sentiment_score']}")
print(f"解释: {fetcher.interpret_sentiment_score(sentiment['sentiment_score'])}")
asyncio.run(test())
步骤2: 添加公告爬虫
文件: /root/work/stock/backend/app/modules/announcement_crawler.py
"""
公告爬虫 - 从巨潮资讯、新浪等获取上市公司公告
"""
import aiohttp
import asyncio
from typing import List, Dict, Optional
from datetime import datetime
import logging
import re
from bs4 import BeautifulSoup
logger = logging.getLogger(__name__)
class AnnouncementCrawler:
"""获取上市公司公告的爬虫"""
def __init__(self):
self.cninfo_base = "http://www.cninfo.com.cn"
self.sina_base = "http://finance.sina.com.cn"
async def get_announcements(self, symbol: str, days: int = 7) -> List[Dict]:
"""
获取最近N天的公告
Args:
symbol: 股票代码
days: 天数
Returns:
公告列表，每条包含类型、标题、内容摘要等
"""
try:
# 从新浪财经获取（更快）
announcements = await self._fetch_from_sina(symbol, days)
return announcements
except Exception as e:
logger.error(f"获取 {symbol} 公告失败: {str(e)}")
return []
async def _fetch_from_sina(self, symbol: str, days: int) -> List[Dict]:
"""从新浪财经获取公告"""
try:
# 新浪财经公告接口
sina_code = f"sh{symbol[:-3]}" if symbol.startswith("6") else f"sz{symbol[:-3]}"
url = f"http://vip.stock.finance.sina.com.cn/q/go.php?symbol={sina_code}"
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
if resp.status != 200:
return []
html = await resp.text()
# 解析HTML获取公告链接
soup = BeautifulSoup(html, 'html.parser')
announcements = []
# 查找公告表格
tables = soup.find_all('table', class_='infoTable')
for table in tables[:5]: # 只取前5条
rows = table.find_all('tr')
for row in rows[1:]: # 跳过表头
cols = row.find_all('td')
if len(cols) >= 2:
link = cols[0].find('a')
if link:
announcements.append({
'title': link.text.strip(),
'url': link.get('href'),
'date': cols[1].text.strip() if len(cols) > 1 else None,
'type': self._classify_announcement(link.text),
'importance': self._assess_importance(link.text),
})
return announcements
except Exception as e:
logger.error(f"从新浪获取公告失败: {str(e)}")
return []
def _classify_announcement(self, title: str) -> str:
"""分类公告类型"""
if '业绩预��' in title or '年报' in title or '半年报' in title:
return '业绩'
elif '重组' in title or '并购' in title:
return '重组'
elif '增持' in title or '减持' in title:
return '股东'
elif '诉讼' in title or '纠纷' in title:
return '诉讼'
elif '风险提示' in title:
return '风险'
else:
return '其他'
def _assess_importance(self, title: str) -> int:
"""评估公告重要程度（1-5）"""
keywords_5 = ['重大事项', '停牌', '退市', '暂停']
keywords_4 = ['业绩预告', '年报', '重组']
keywords_3 = ['增持', '减持', '融资']
for kw in keywords_5:
if kw in title:
return 5
for kw in keywords_4:
if kw in title:
return 4
for kw in keywords_3:
if kw in title:
return 3
return 2
# 使用示例
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
async def test():
crawler = AnnouncementCrawler()
announcements = await crawler.get_announcements("600000", days=7)
print(f"获取到 {len(announcements)} 条公告:")
for ann in announcements[:3]:
print(f"- {ann['title']} ({ann['type']}, 重要度: {ann['importance']})")
asyncio.run(test())
Phase 3: 优化和集成（第3周）
步骤1: 缓存策略优化
文件: /root/work/stock/backend/app/modules/cache_strategy.py (新增)
"""
多层缓存策略 - 优化AI助手的响应速度
"""
from app.modules.cache_manager import cache
from typing import Dict, Optional
from datetime import datetime, timedelta
import json
import logging
logger = logging.getLogger(__name__)
class AIAssistantCacheManager:
"""AI助手的多层缓存管理"""
# 缓存键前缀
PREFIX = "ai_cache"
# 缓存过期时间
TTL = {
'fundamentals': 86400, # 基本面: 24小时
'sentiment': 300, # 情绪: 5分钟
'announcements': 3600, # 公告: 1小时
'knowledge': 3600, # 知识: 1小时
'session': 1800, # 会话: 30分钟
}
@staticmethod
def get_fundamentals_cache(symbol: str) -> Optional[Dict]:
"""获取缓存的基本面数据"""
key = f"{AIAssistantCacheManager.PREFIX}:fundamentals:{symbol}"
return cache.get(key)
@staticmethod
def set_fundamentals_cache(symbol: str, data: Dict) -> None:
"""缓存基本面数据"""
key = f"{AIAssistantCacheManager.PREFIX}:fundamentals:{symbol}"
cache.set(key, data, ex=AIAssistantCacheManager.TTL['fundamentals'])
@staticmethod
def get_sentiment_cache(symbol: str) -> Optional[Dict]:
"""获取缓存的情绪数据"""
key = f"{AIAssistantCacheManager.PREFIX}:sentiment:{symbol}"
return cache.get(key)
@staticmethod
def set_sentiment_cache(symbol: str, data: Dict) -> None:
"""缓存情绪数据"""
key = f"{AIAssistantCacheManager.PREFIX}:sentiment:{symbol}"
cache.set(key, data, ex=AIAssistantCacheManager.TTL['sentiment'])
@staticmethod
def get_knowledge_summary(symbol: str, knowledge_type: str) -> Optional[str]:
"""获取AI知识总结"""
key = f"{AIAssistantCacheManager.PREFIX}:knowledge:{symbol}:{knowledge_type}"
return cache.get(key)
@staticmethod
def set_knowledge_summary(symbol: str, knowledge_type: str, content: str) -> None:
"""缓存AI知识总结"""
key = f"{AIAssistantCacheManager.PREFIX}:knowledge:{symbol}:{knowledge_type}"
cache.set(key, content, ex=AIAssistantCacheManager.TTL['knowledge'])
@staticmethod
def invalidate_symbol_cache(symbol: str) -> None:
"""清除某只股票的所有缓存（当有新公告时）"""
patterns = [
f"{AIAssistantCacheManager.PREFIX}:*:{symbol}",
]
# 使用Redis的模式删除
for pattern in patterns:
keys = cache.keys(pattern)
if keys:
cache.delete(*keys)
logger.info(f"已清除 {symbol} 的缓存")
# 使用示例
if __name__ == "__main__":
# 缓存基本面数据
fundamentals = {"pe": 12.5, "roe": 18.3}
AIAssistantCacheManager.set_fundamentals_cache("600000", fundamentals)
# 检索缓存
cached = AIAssistantCacheManager.get_fundamentals_cache("600000")
print(f"缓存数据: {cached}")
步骤2: 集成所有模块到AI路由
文件: /root/work/stock/backend/app/routes/ai_assistant.py (完整版)
"""
完整的AI助手路由 - 整合所有数据源
"""
from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel
from typing import List, Optional, Dict
import logging
from datetime import datetime
import asyncio
from app.config import get_settings
from app.modules.fundamental_data import FundamentalDataFetcher
from app.modules.sentiment_fetcher import MarketSentimentFetcher
from app.modules.announcement_crawler import AnnouncementCrawler
from app.modules.cache_manager import cache
from app.modules.cache_strategy import AIAssistantCacheManager
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/ai", tags=["AI Assistant"])
settings = get_settings()
class ChatMessage(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
messages: List[ChatMessage] = []
query: str
symbol: Optional[str] = None
class ChatResponse(BaseModel):
response: str
thinking: str = ""
data_sources: List[str] = []
symbol: Optional[str] = None
def get_fetchers():
"""依赖注入 - 获取所有数据获取器"""
return {
'fundamental': FundamentalDataFetcher(settings.tushare_token),
'sentiment': MarketSentimentFetcher(),
'announcements': AnnouncementCrawler(),
}
@router.post("/chat")
async def enhanced_chat(
request: ChatRequest,
background_tasks: BackgroundTasks,
fetchers: Dict = Depends(get_fetchers)
) -> ChatResponse:
"""
增强的AI助手 - 整合所有数据源
"""
try:
query_lower = request.query.lower()
data_sources = []
# 如果指定了股票代码，获取完整数据
context_data = {}
if request.symbol:
context_data = await _gather_stock_context(
request.symbol,
fetchers,
background_tasks
)
data_sources = context_data.get('sources', ['Contextual Knowledge'])
# 生成AI回答
response_text = _generate_response(
query=request.query,
context=context_data,
messages=request.messages
)
return ChatResponse(
response=response_text,
symbol=request.symbol,
data_sources=data_sources
)
except Exception as e:
logger.error(f"AI助手错误: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
async def _gather_stock_context(
symbol: str,
fetchers: Dict,
background_tasks: BackgroundTasks
) -> Dict:
"""
汇总股票的完整上下文（并行获取）
"""
context = {
'symbol': symbol,
'timestamp': datetime.now().isoformat(),
'sources': ['Contextual Knowledge'],
}
try:
# 并行获取所有数据
tasks = []
# 1. 基本面数据 (优先级高，先从缓存查)
fundamental_cache = AIAssistantCacheManager.get_fundamentals_cache(symbol)
if fundamental_cache:
context['fundamentals'] = fundamental_cache
context['sources'].append('Cache')
else:
tasks.append(_fetch_fundamentals(symbol, fetchers['fundamental']))
# 2. 市场情绪 (实时获取)
sentiment_cache = AIAssistantCacheManager.get_sentiment_cache(symbol)
if sentiment_cache:
context['sentiment'] = sentiment_cache
else:
tasks.append(_fetch_sentiment(symbol, fetchers['sentiment']))
# 3. 公告 (后台获取，不阻塞响应)
background_tasks.add_task(
_fetch_announcements,
symbol,
fetchers['announcements'],
context
)
# 等待并行任务完成
if tasks:
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, dict):
if 'net_profit' in result: # 基本面数据
context['fundamentals'] = result
context['sources'].append('TuShare')
# 缓存
background_tasks.add_task(
AIAssistantCacheManager.set_fundamentals_cache,
symbol, result
)
elif 'sentiment_score' in result: # 情绪数据
context['sentiment'] = result
context['sources'].append('RealTime')
# 缓存
background_tasks.add_task(
AIAssistantCacheManager.set_sentiment_cache,
symbol, result
)
return context
except Exception as e:
logger.error(f"汇总 {symbol} 上下文失败: {str(e)}")
return context
async def _fetch_fundamentals(symbol: str, fetcher: FundamentalDataFetcher) -> Dict:
"""异步获取基本面"""
return fetcher.get_stock_fundamentals(symbol)
async def _fetch_sentiment(symbol: str, fetcher: MarketSentimentFetcher) -> Dict:
"""异步获取情绪"""
return await fetcher.get_sentiment(symbol)
async def _fetch_announcements(
symbol: str,
crawler: AnnouncementCrawler,
context: Dict
) -> None:
"""后台获取公告"""
try:
announcements = await crawler.get_announcements(symbol, days=7)
context['announcements'] = announcements
if announcements:
context['sources'].append('Announcements')
except Exception as e:
logger.error(f"获取 {symbol} 公告失败: {str(e)}")
def _generate_response(
query: str,
context: Dict,
messages: List[ChatMessage]
) -> str:
"""
基于上下文和查询生成AI回答
"""
query_lower = query.lower()
# 从context中提取数据
fundamentals = context.get('fundamentals', {})
sentiment = context.get('sentiment', {})
announcements = context.get('announcements', [])
# PE相关
if "pe" in query_lower or "市盈率" in query_lower:
if fundamentals.get('pe_ttm'):
pe = fundamentals['pe_ttm']
industry_pe = fundamentals.get('industry_pe_avg')
response = f"**PE比率分析**\n\n"
response += f"当前PE（TTM）: **{pe:.2f}**\n"
if industry_pe:
diff_pct = ((pe - industry_pe) / industry_pe) * 100
response += f"行业平均PE: {industry_pe:.2f}\n"
response += f"相对行业: {'↑ 高' if diff_pct > 0 else '↓ 低'} {abs(diff_pct):.1f}%\n\n"
if pe < industry_pe * 0.8:
response += "📊 **相对低估**，可能存在投资机会\n"
elif pe > industry_pe * 1.5:
response += "⚠️ **相对高估**，需要谨慎\n"
else:
response += "➡️ **合理估值**，价格稳定\n"
# 结合增长情况
if fundamentals.get('peg'):
response += f"\n**PEG指数**: {fundamentals['peg']:.2f}\n"
if fundamentals['peg'] < 1:
response += "PEG < 1，增长期公司，可能有投资价值\n"
return response
# ROE相关
if "roe" in query_lower or "净资产收益率" in query_lower:
if fundamentals.get('roe'):
roe = fundamentals['roe']
response = f"**ROE（净资产收益率）: {roe:.2f}%**\n\n"
if roe > 20:
response += "🌟 **优秀** - 公司创造价值的能力很强\n"
elif roe > 15:
response += "✅ **良好** - 公司创造价值的能力不错\n"
elif roe > 10:
response += "➡️ **一般** - 公司创造价值的能力中等\n"
else:
response += "⚠️ **较弱** - 公司创造价值的能力需要提升\n"
if fundamentals.get('industry_roe_avg'):
industry_roe = fundamentals['industry_roe_avg']
response += f"\n行业平均ROE: {industry_roe:.2f}%\n"
if roe > industry_roe:
response += "相对行业: 👆 更优秀"
else:
response += "相对行业: 👇 需要改进"
return response
# 增长相关
if "增速" in query_lower or "增长" in query_lower:
if fundamentals.get('net_profit_growth_rate') is not None:
growth = fundamentals['net_profit_growth_rate']
response = f"**净利润增速（同比）: {growth:.2f}%**\n\n"
if growth > 30:
response += "🚀 **高增长** - 市场看好，可能有估值溢价\n"
elif growth > 15:
response += "📈 **良好增长** - 合理的增长期望\n"
elif growth > 0:
response += "➡️ **缓慢增长** - 需要关注转机\n"
else:
response += "📉 **负增长** - 需要重点关注\n"
# 结合PE
if fundamentals.get('peg'):
response += f"\nPEG指数: {fundamentals['peg']:.2f}\n"
if fundamentals['peg'] < 1:
response += "相对PE较低，增长期公司\n"
return response
# 市场情绪相关
if "情绪" in query_lower or "资金" in query_lower:
if sentiment.get('sentiment_score') is not None:
score = sentiment['sentiment_score']
response = f"**市场情绪评分: {score}/100**\n\n"
if score > 50:
response += "🟢 **看涨** - 市场乐观，资金持续流入\n"
elif score > 0:
response += "🟡 **中性偏好** - 市场情绪稳定\n"
elif score > -50:
response += "🟠 **中性偏差** - 市场有所谨慎\n"
else:
response += "🔴 **看跌** - 市场悲观，资金流出\n"
if sentiment.get('sina'):
sina = sentiment['sina']
response += f"\n成交变化: {sina.get('change_pct', 0):.2f}%\n"
response += f"换手率: {sentiment.get('eastmoney', {}).get('turnover_ratio', 0)*100:.2f}%\n"
return response
# 最近公告
if "公告" in query_lower or "news" in query_lower.lower():
if announcements:
response = "**最近公告:**\n\n"
for ann in announcements[:5]:
response += f"- [{ann['type']}] {ann['title']}\n"
response += f" 日期: {ann.get('date', 'N/A')}\n"
return response
# 综合分析
response = "**综合分析**\n\n"
if fundamentals:
response += f"📊 PE: {fundamentals.get('pe_ttm', 'N/A')}\n"
response += f"💰 ROE: {fundamentals.get('roe', 'N/A')}%\n"
response += f"📈 增速: {fundamentals.get('net_profit_growth_rate', 'N/A')}%\n"
if sentiment:
response += f"📡 情绪评分: {sentiment.get('sentiment_score', 'N/A')}\n"
return response if response != "**综合分析**\n\n" else "请提供更多信息进行分析"
@router.get("/health")
async def health_check() -> Dict:
"""AI助手健康检查"""
return {
'status': 'healthy',
'modules': {
'fundamental_fetcher': 'ready',
'sentiment_fetcher': 'ready',
'announcement_crawler': 'ready',
},
'timestamp': datetime.now().isoformat()
}
测试和部署
完整测试清单
#!/bin/bash
echo "=== AI 助手数据管道测试 ==="
# 1. 基本面数据测试
echo "1. 测试基本面数据获取..."
curl -X GET "http://localhost:8001/api/ai/stock-analysis/600000"
# 2. AI回答PE问题
echo -e "\n2. 测试PE分析..."
curl -X POST "http://localhost:8001/api/ai/chat" \
-H "Content-Type: application/json" \
-d '{
"messages": [],
"query": "600000的PE比率怎么样",
"symbol": "600000"
}'
# 3. AI回答ROE问题
echo -e "\n3. 测试ROE分析..."
curl -X POST "http://localhost:8001/api/ai/chat" \
-H "Content-Type: application/json" \
-d '{
"messages": [],
"query": "这只股票的ROE如何",
"symbol": "600000"
}'
# 4. 数据库验证
echo -e "\n4. 验证数据库..."
psql -U stock_user -d stock_db -c "
SELECT symbol, pe, roe, net_profit_growth_rate
FROM stock_data.stock_fundamentals
LIMIT 5;
"
# 5. 缓存验证
echo -e "\n5. 验证缓存..."
redis-cli --rdb
echo -e "\n=== 测试完成 ==="
部署步骤
# 1. 更新依赖
pip install beautifulsoup4 lxml sentence-transformers
# 2. 数据库初始化
python -c "from app.database_migrations import create_tables; create_tables('postgresql://...')"
# 3. 重建后端
docker compose build backend --no-cache
# 4. 重启系统
docker compose restart backend
# 5. 运行测试
bash test_ai_pipeline.sh
# 6. 监控日志
docker compose logs -f backend | grep "AI Assistant"
预计完成时间: 110小时（3周） 下一阶段: 向量化知识库 + 多轮对话支持