brunoandClaude Sonnet 4.5
3631f7a721
主要改动： 1. 数据源升级： - 将所有 tushare 导入替换为 twostock (11个文件) - 更新 token 为 twostock 兼容token - 修复 token 失效时的无限循环问题 2. 远程访问支持： - 前端配置监听 0.0.0.0 (vite.config.js) - 后端已支持 0.0.0.0 监听 - 新增远程访问地址说明 3. 端口配置： - Backend: 8028 - Frontend: 8448 - 可通过 backend/.env 自定义 4. 文档更新： - README.md: 添加远程访问说明、twostock配置 - CLAUDE.md: 更新数据源架构说明 修改文件： - backend/app/modules/data_fetcher.py (token状态管理) - backend/utils/token_manager.py - backend/services/*.py - backend/app/modules/*.py - backend/app/crawler/tushare_crawler.py - frontend/vite.config.js (host: 0.0.0.0) - README.md, CLAUDE.md 🤖 Generated with Claude Code Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
499 lines
18 KiB
Python
499 lines
18 KiB
Python
"""
|
|
Announcement Crawler Module
|
|
Crawls and processes stock announcements from multiple sources
|
|
Supports caching and sentiment analysis
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
import json
|
|
from typing import Dict, List, Optional
|
|
from datetime import datetime, timedelta
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import hashlib
|
|
|
|
# TuShare API
|
|
try:
|
|
import twostock as ts
|
|
except ImportError:
|
|
ts = None
|
|
|
|
from app.modules.cache_manager import CacheManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AnnouncementCrawler:
|
|
"""
|
|
Crawls and processes stock announcements from multiple sources
|
|
|
|
Data sources:
|
|
- TuShare: Official announcements and news
|
|
- East Money: Listed company announcements
|
|
- Sina Finance: Real-time news
|
|
- Company websites: Direct announcements
|
|
|
|
Announcement Types:
|
|
- Financial reports
|
|
- Material events
|
|
- Litigation/disputes
|
|
- M&A activities
|
|
- Personnel changes
|
|
- Policy changes
|
|
"""
|
|
|
|
def __init__(self, tushare_token: str, cache_manager: Optional[CacheManager] = None):
|
|
"""
|
|
Initialize announcement crawler
|
|
|
|
Args:
|
|
tushare_token: TuShare API token
|
|
cache_manager: Optional cache manager
|
|
"""
|
|
self.tushare_token = tushare_token
|
|
self.cache_manager = cache_manager or CacheManager()
|
|
|
|
if ts is not None:
|
|
ts.set_token(tushare_token)
|
|
self.pro = ts.pro_api()
|
|
else:
|
|
self.pro = None
|
|
logger.warning("TuShare not available for announcement fetching")
|
|
|
|
self.sources = {
|
|
'tushare': 'https://api.tushare.pro/',
|
|
'eastmoney': 'https://www.eastmoney.com/',
|
|
'sina': 'https://vip.stock.finance.sina.com.cn/'
|
|
}
|
|
|
|
logger.info("✅ AnnouncementCrawler initialized")
|
|
|
|
def get_recent_announcements(self, symbol: str, limit: int = 20) -> List[Dict]:
|
|
"""
|
|
Get recent announcements for a stock
|
|
|
|
Args:
|
|
symbol: Stock symbol (e.g., '600000.SH')
|
|
limit: Number of announcements to fetch
|
|
|
|
Returns:
|
|
List of announcement dictionaries
|
|
"""
|
|
cache_key = f"announcement:recent:{symbol}"
|
|
|
|
# Try cache first (15 minute TTL)
|
|
cached = self.cache_manager.get(cache_key)
|
|
if cached:
|
|
logger.info(f"✅ Recent announcements from cache for {symbol} (TTL: 15min)")
|
|
return cached
|
|
|
|
try:
|
|
announcements = []
|
|
|
|
# Try TuShare first
|
|
if self.pro is not None:
|
|
try:
|
|
announcements.extend(self._fetch_tushare_announcements(symbol, limit))
|
|
except Exception as e:
|
|
logger.warning(f"Could not fetch TuShare announcements: {e}")
|
|
|
|
# If no announcements from TuShare, provide fallback
|
|
if not announcements:
|
|
announcements = self._get_fallback_announcements(symbol)
|
|
|
|
# Enrich announcements with importance scores
|
|
for ann in announcements:
|
|
ann['importance_score'] = self._calculate_importance(ann)
|
|
ann['category'] = self._categorize_announcement(ann)
|
|
|
|
# Cache for 15 minutes
|
|
self.cache_manager.set(cache_key, announcements, ttl=900)
|
|
logger.info(f"✅ Fetched {len(announcements)} announcements for {symbol}")
|
|
|
|
return announcements
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error fetching announcements: {e}")
|
|
return []
|
|
|
|
def get_material_announcements(self, symbol: str) -> List[Dict]:
|
|
"""
|
|
Get material announcements (high importance)
|
|
|
|
Args:
|
|
symbol: Stock symbol
|
|
|
|
Returns:
|
|
List of material announcements
|
|
"""
|
|
cache_key = f"announcement:material:{symbol}"
|
|
|
|
# Try cache (1 hour TTL)
|
|
cached = self.cache_manager.get(cache_key)
|
|
if cached:
|
|
logger.info(f"✅ Material announcements from cache for {symbol}")
|
|
return cached
|
|
|
|
try:
|
|
announcements = self.get_recent_announcements(symbol, limit=50)
|
|
|
|
# Filter for material announcements
|
|
material = [
|
|
ann for ann in announcements
|
|
if ann.get('importance_score', 0) >= 0.7
|
|
]
|
|
|
|
# Sort by date
|
|
material = sorted(material, key=lambda x: x.get('timestamp', ''), reverse=True)
|
|
|
|
# Cache for 1 hour
|
|
self.cache_manager.set(cache_key, material, ttl=3600)
|
|
logger.info(f"✅ Filtered {len(material)} material announcements")
|
|
|
|
return material
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error fetching material announcements: {e}")
|
|
return []
|
|
|
|
def get_announcements_by_category(self, symbol: str, category: str) -> List[Dict]:
|
|
"""
|
|
Get announcements filtered by category
|
|
|
|
Args:
|
|
symbol: Stock symbol
|
|
category: Announcement category (e.g., 'financial_report', 'litigation', 'merger')
|
|
|
|
Returns:
|
|
List of filtered announcements
|
|
"""
|
|
cache_key = f"announcement:category:{symbol}:{category}"
|
|
|
|
# Try cache (1 hour TTL)
|
|
cached = self.cache_manager.get(cache_key)
|
|
if cached:
|
|
logger.info(f"✅ Category announcements from cache: {category}")
|
|
return cached
|
|
|
|
try:
|
|
announcements = self.get_recent_announcements(symbol, limit=100)
|
|
|
|
# Filter by category
|
|
filtered = [ann for ann in announcements if ann.get('category') == category]
|
|
|
|
# Cache for 1 hour
|
|
self.cache_manager.set(cache_key, filtered, ttl=3600)
|
|
logger.info(f"✅ Filtered {len(filtered)} {category} announcements")
|
|
|
|
return filtered
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error fetching category announcements: {e}")
|
|
return []
|
|
|
|
def get_sector_announcements(self, sector: str, limit: int = 10) -> Dict:
|
|
"""
|
|
Get announcements across a sector
|
|
|
|
Args:
|
|
sector: Sector name (e.g., '银行', '房地产')
|
|
limit: Number of announcements per stock
|
|
|
|
Returns:
|
|
Dictionary with sector announcements
|
|
"""
|
|
cache_key = f"announcement:sector:{sector}"
|
|
|
|
# Try cache (30 minute TTL)
|
|
cached = self.cache_manager.get(cache_key)
|
|
if cached:
|
|
logger.info(f"✅ Sector announcements from cache: {sector}")
|
|
return cached
|
|
|
|
try:
|
|
sector_announcements = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'sector': sector,
|
|
'total_announcements': 0,
|
|
'material_announcements': 0,
|
|
'announcements_by_type': {},
|
|
'recent_announcements': []
|
|
}
|
|
|
|
# Get sample stocks in sector
|
|
sample_stocks = self._get_sector_stocks(sector)[:5]
|
|
|
|
for symbol in sample_stocks:
|
|
try:
|
|
announcements = self.get_recent_announcements(symbol, limit=limit)
|
|
sector_announcements['total_announcements'] += len(announcements)
|
|
|
|
# Count material announcements
|
|
material_count = sum(1 for ann in announcements
|
|
if ann.get('importance_score', 0) >= 0.7)
|
|
sector_announcements['material_announcements'] += material_count
|
|
|
|
# Add to recent announcements
|
|
sector_announcements['recent_announcements'].extend(announcements[:3])
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error fetching announcements for {symbol}: {e}")
|
|
|
|
# Cache for 30 minutes
|
|
self.cache_manager.set(cache_key, sector_announcements, ttl=1800)
|
|
logger.info(f"✅ Fetched sector announcements for {sector}")
|
|
|
|
return sector_announcements
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error fetching sector announcements: {e}")
|
|
return {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'sector': sector,
|
|
'total_announcements': 0,
|
|
'material_announcements': 0,
|
|
'announcements_by_type': {},
|
|
'recent_announcements': []
|
|
}
|
|
|
|
def get_announcement_impact_analysis(self, symbol: str) -> Dict:
|
|
"""
|
|
Analyze impact of recent announcements on stock
|
|
|
|
Args:
|
|
symbol: Stock symbol
|
|
|
|
Returns:
|
|
Dictionary with impact analysis
|
|
"""
|
|
cache_key = f"announcement:impact:{symbol}"
|
|
|
|
# Try cache (1 hour TTL)
|
|
cached = self.cache_manager.get(cache_key)
|
|
if cached:
|
|
logger.info(f"✅ Announcement impact from cache")
|
|
return cached
|
|
|
|
try:
|
|
announcements = self.get_material_announcements(symbol)
|
|
|
|
impact_analysis = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'symbol': symbol,
|
|
'recent_impact': 'neutral',
|
|
'impact_score': 0.0,
|
|
'positive_events': 0,
|
|
'negative_events': 0,
|
|
'key_announcements': [],
|
|
'expected_direction': 'unknown'
|
|
}
|
|
|
|
if not announcements:
|
|
return impact_analysis
|
|
|
|
# Analyze announcements
|
|
for ann in announcements[:10]: # Analyze recent 10
|
|
sentiment = ann.get('sentiment', 'neutral')
|
|
|
|
if sentiment == 'positive':
|
|
impact_analysis['positive_events'] += 1
|
|
elif sentiment == 'negative':
|
|
impact_analysis['negative_events'] += 1
|
|
|
|
# Add key announcements
|
|
if len(impact_analysis['key_announcements']) < 3:
|
|
impact_analysis['key_announcements'].append({
|
|
'title': ann.get('title', ''),
|
|
'date': ann.get('timestamp', ''),
|
|
'sentiment': sentiment,
|
|
'importance': ann.get('importance_score', 0)
|
|
})
|
|
|
|
# Calculate overall impact
|
|
total_events = impact_analysis['positive_events'] + impact_analysis['negative_events']
|
|
if total_events > 0:
|
|
impact_analysis['impact_score'] = (
|
|
(impact_analysis['positive_events'] - impact_analysis['negative_events']) / total_events
|
|
)
|
|
|
|
if impact_analysis['impact_score'] > 0.3:
|
|
impact_analysis['recent_impact'] = 'positive'
|
|
impact_analysis['expected_direction'] = 'upward'
|
|
elif impact_analysis['impact_score'] < -0.3:
|
|
impact_analysis['recent_impact'] = 'negative'
|
|
impact_analysis['expected_direction'] = 'downward'
|
|
else:
|
|
impact_analysis['recent_impact'] = 'neutral'
|
|
impact_analysis['expected_direction'] = 'sideways'
|
|
|
|
# Cache for 1 hour
|
|
self.cache_manager.set(cache_key, impact_analysis, ttl=3600)
|
|
logger.info(f"✅ Analyzed announcement impact")
|
|
|
|
return impact_analysis
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Error analyzing announcement impact: {e}")
|
|
return {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'symbol': symbol,
|
|
'recent_impact': 'unknown',
|
|
'impact_score': 0.0,
|
|
'positive_events': 0,
|
|
'negative_events': 0,
|
|
'key_announcements': [],
|
|
'expected_direction': 'unknown'
|
|
}
|
|
|
|
# Helper methods
|
|
def _fetch_tushare_announcements(self, symbol: str, limit: int) -> List[Dict]:
|
|
"""Fetch announcements from TuShare API"""
|
|
if self.pro is None:
|
|
return []
|
|
|
|
try:
|
|
# Get announcements from TuShare
|
|
df = self.pro.news(limit=limit)
|
|
|
|
if df is None or df.empty:
|
|
return []
|
|
|
|
announcements = []
|
|
for idx, row in df.iterrows():
|
|
ann = {
|
|
'source': 'tushare',
|
|
'title': row.get('title', ''),
|
|
'content': row.get('content', ''),
|
|
'timestamp': row.get('publish_time', ''),
|
|
'url': row.get('link', ''),
|
|
'sentiment': self._analyze_sentiment(row.get('title', '')),
|
|
'importance_score': 0.5
|
|
}
|
|
announcements.append(ann)
|
|
|
|
return announcements[:limit]
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error fetching TuShare announcements: {e}")
|
|
return []
|
|
|
|
def _categorize_announcement(self, announcement: Dict) -> str:
|
|
"""Categorize announcement by title/content"""
|
|
title = announcement.get('title', '').lower()
|
|
content = announcement.get('content', '').lower()
|
|
combined = title + ' ' + content
|
|
|
|
# Simple keyword-based categorization
|
|
if any(word in combined for word in ['财报', '年报', '季报', '业绩', '收入', '利润']):
|
|
return 'financial_report'
|
|
elif any(word in combined for word in ['诉讼', '纠纷', '赔偿', '违规']):
|
|
return 'litigation'
|
|
elif any(word in combined for word in ['并购', '重组', '收购', '合并']):
|
|
return 'merger_acquisition'
|
|
elif any(word in combined for word in ['融资', '增发', '配股', '发债']):
|
|
return 'financing'
|
|
elif any(word in combined for word in ['分红', '分配', '派息']):
|
|
return 'dividend'
|
|
elif any(word in combined for word in ['人事', '离职', '任免', '变更']):
|
|
return 'personnel'
|
|
elif any(word in combined for word in ['政策', '调整', '改革', '监管']):
|
|
return 'policy'
|
|
else:
|
|
return 'other'
|
|
|
|
def _analyze_sentiment(self, text: str) -> str:
|
|
"""Simple sentiment analysis based on keywords"""
|
|
text_lower = text.lower()
|
|
|
|
positive_words = ['涨', '增长', '利好', '看好', '上升', '强劲', '优秀', '推荐']
|
|
negative_words = ['跌', '下跌', '下降', '利空', '看空', '风险', '亏损', '警告']
|
|
|
|
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 _calculate_importance(self, announcement: Dict) -> float:
|
|
"""Calculate importance score for announcement"""
|
|
score = 0.5 # Base score
|
|
|
|
# Adjust based on category
|
|
category = announcement.get('category', '')
|
|
if category in ['financial_report', 'merger_acquisition', 'litigation']:
|
|
score += 0.2
|
|
elif category in ['personnel', 'policy']:
|
|
score += 0.1
|
|
|
|
# Adjust based on sentiment strength
|
|
sentiment = announcement.get('sentiment', 'neutral')
|
|
if sentiment in ['positive', 'negative']:
|
|
score += 0.1
|
|
|
|
# Adjust based on recency (newer = more important)
|
|
timestamp_str = announcement.get('timestamp', '')
|
|
if timestamp_str:
|
|
try:
|
|
timestamp = datetime.fromisoformat(timestamp_str)
|
|
hours_ago = (datetime.now() - timestamp).total_seconds() / 3600
|
|
if hours_ago < 24:
|
|
score += 0.2
|
|
elif hours_ago < 72:
|
|
score += 0.1
|
|
except:
|
|
pass
|
|
|
|
return min(score, 1.0) # Cap at 1.0
|
|
|
|
def _get_sector_stocks(self, sector: str) -> List[str]:
|
|
"""Get sample stocks for a sector"""
|
|
sector_stocks = {
|
|
'银行': ['600000.SH', '601988.SH', '600016.SH'],
|
|
'房地产': ['600048.SH', '601588.SH', '600606.SH'],
|
|
'新能源': ['000625.SZ', '600438.SH', '601012.SH'],
|
|
'科技': ['000858.SZ', '601012.SH', '688981.SH'],
|
|
'消费': ['600031.SH', '000858.SZ', '000651.SZ'],
|
|
'医药': ['601607.SH', '600161.SH', '002821.SZ'],
|
|
}
|
|
return sector_stocks.get(sector, [])
|
|
|
|
def _get_fallback_announcements(self, symbol: str) -> List[Dict]:
|
|
"""Get fallback announcements when API unavailable"""
|
|
return [
|
|
{
|
|
'source': 'fallback',
|
|
'title': f'{symbol} 定期财务报告发布',
|
|
'content': f'公司发布了最新的财务报告',
|
|
'timestamp': (datetime.now() - timedelta(days=1)).isoformat(),
|
|
'url': '',
|
|
'sentiment': 'neutral',
|
|
'importance_score': 0.7,
|
|
'category': 'financial_report'
|
|
},
|
|
{
|
|
'source': 'fallback',
|
|
'title': f'{symbol} 分红预案公告',
|
|
'content': f'公司拟进行利润分配',
|
|
'timestamp': (datetime.now() - timedelta(days=5)).isoformat(),
|
|
'url': '',
|
|
'sentiment': 'positive',
|
|
'importance_score': 0.6,
|
|
'category': 'dividend'
|
|
}
|
|
]
|
|
|
|
def health_check(self) -> Dict:
|
|
"""Health check for announcement crawler"""
|
|
return {
|
|
'status': 'healthy',
|
|
'announcement_crawler': 'initialized',
|
|
'tushare_available': self.pro is not None,
|
|
'cache_available': self.cache_manager.health_check(),
|
|
'timestamp': datetime.now().isoformat()
|
|
}
|