"""
网络爬虫数据源实现
src/stock_data_source.py
基于网络爬虫的数据源实现
"""
import logging
from typing import List, Optional, Dict, Any
from stock_mcp.data_source_interface import FinancialDataInterface
logger = logging.getLogger(__name__)
class WebCrawlerDataSource(FinancialDataInterface):
def __init__(self):
self.kline_spider = None
self.searcher = None
self.real_time_spider = None
self.fundamental_crawler = None
self.valuation_crawler = None
self.financial_analysis_crawler = None
self.market_spider = None
self.smart_review_crawler = None
def initialize(self) -> bool:
"""
初始化所有爬虫组件
Returns:
bool: 初始化成功返回True
Raises:
ImportError: 当无法导入必要的模块时
Exception: 当初始化过程中的其他错误发生时
"""
try:
from stock_mcp.crawler.basic_data import StockSearcher
from stock_mcp.crawler.technical_data import KlineSpider
from stock_mcp.crawler.real_time_data import RealTimeDataSpider
from stock_mcp.crawler.fundamental_data import FundamentalDataCrawler
from stock_mcp.crawler.valuation_data import ValuationDataCrawler
from stock_mcp.crawler.financial_analysis import FinancialAnalysisCrawler
from stock_mcp.crawler.market import MarketSpider
from stock_mcp.crawler.smart_review import SmartReviewCrawler
self.kline_spider = KlineSpider()
self.searcher = StockSearcher()
self.real_time_spider = RealTimeDataSpider()
self.fundamental_crawler = FundamentalDataCrawler()
self.valuation_crawler = ValuationDataCrawler()
self.financial_analysis_crawler = FinancialAnalysisCrawler()
self.market_spider = MarketSpider()
self.smart_review_crawler = SmartReviewCrawler()
# 验证关键组件是否初始化成功
if not all([self.kline_spider, self.searcher, self.real_time_spider,
self.fundamental_crawler, self.valuation_crawler,
self.financial_analysis_crawler, self.market_spider,
self.smart_review_crawler]):
raise RuntimeError("一个或多个爬虫组件初始化失败")
return True
except ImportError as e:
raise ImportError(f"无法导入必要的爬虫模块: {e}") from e
except Exception as e:
raise Exception(f"初始化爬虫组件失败: {e}") from e
def cleanup(self):
self.kline_spider = None
self.searcher = None
self.real_time_spider = None
self.fundamental_crawler = None
self.valuation_crawler = None
self.financial_analysis_crawler = None
self.market_spider = None
self.smart_review_crawler = None
def get_historical_k_data(
self,
stock_code: str,
start_date: str,
end_date: str,
frequency: str = "d",
) -> List[Dict]:
beg = start_date.replace("-", "")
end = end_date.replace("-", "")
frequency_map = {
"5": 5,
"15": 15,
"30": 30,
"60": 60,
"d": 101,
"w": 102,
"m": 103
}
klt = frequency_map.get(frequency, 101)
return self.kline_spider.get_klines(
stock_code=stock_code,
beg=beg,
end=end,
klt=klt,
fqt=1
)
def get_stock_search(
self,
keyword: str
) -> Optional[List[Dict]]:
return self.searcher.search(keyword)
def get_technical_indicators(
self,
stock_code: str,
page_size: int = 30
) -> List[Dict]:
return self.kline_spider.get_technical_indicators(stock_code, page_size)
def get_last_trading_day(self) -> Optional[Dict]:
return self.searcher.last_trading_day()
def get_real_time_data(self, symbol: str) -> Dict:
return self.real_time_spider.get_real_time_data(symbol)
def get_main_business(self, stock_code: str, report_date: Optional[str] = None) -> Optional[List[Dict[Any, Any]]]:
return self.fundamental_crawler.get_main_business(stock_code, report_date)
def get_report_dates(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.fundamental_crawler.get_report_dates(stock_code)
def get_business_scope(self, stock_code: str) -> Optional[Dict[Any, Any]]:
return self.fundamental_crawler.get_business_scope(stock_code)
def get_business_review(self, stock_code: str) -> Optional[Dict[Any, Any]]:
return self.fundamental_crawler.get_business_review(stock_code)
def get_valuation_analysis(self, stock_code: str, date_type: int = 3) -> Optional[List[Dict[Any, Any]]]:
return self.valuation_crawler.get_valuation_analysis(stock_code, date_type)
def get_institutional_rating(self, stock_code: str, begin_time: str, end_time: str) -> Optional[List[Dict[Any, Any]]]:
return self.valuation_crawler.get_institutional_rating(stock_code, begin_time, end_time)
def get_main_financial_data(self, stock_code: str) -> Optional[Dict[Any, Any]]:
return self.fundamental_crawler.get_main_financial_data(stock_code)
def get_financial_summary(self, stock_code: str, date_type_code: str = "004") -> Optional[List[Dict[Any, Any]]]:
return self.financial_analysis_crawler.get_financial_summary(stock_code, date_type_code)
def get_holder_number(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.financial_analysis_crawler.get_holder_number(stock_code)
def get_industry_profit_comparison(self, stock_code: str, report_date: str = None) -> Optional[List[Dict[Any, Any]]]:
return self.financial_analysis_crawler.get_industry_profit_comparison(stock_code, report_date)
def get_financial_ratios(self, stock_code: str, report_dates: List[str] = None) -> Optional[List[Dict[Any, Any]]]:
return self.financial_analysis_crawler.get_financial_ratios(stock_code, report_dates)
def get_plate_quotation(self, plate_type: int = 2, page_size: int = 10) -> List[Dict]:
return self.market_spider.get_plate_quotation(plate_type, page_size)
def get_plate_fund_flow(self, plate_type: int = 2, page_size: int = 10) -> List[Dict]:
return self.market_spider.get_plate_fund_flow(plate_type, page_size)
def get_historical_fund_flow(self, stock_code: str, limit: int = 10) -> Optional[Dict]:
return self.market_spider.get_historical_fund_flow(stock_code, limit)
def get_billboard_data(self, trade_date: str, page_size: int = 10) -> List[Dict]:
return self.market_spider.get_billboard_data(trade_date, page_size)
def get_stock_billboard_data(self, stock_code: str, limit: int = 10) -> List[Dict]:
return self.market_spider.get_stock_billboard_data(stock_code, limit)
def get_growth_comparison(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.valuation_crawler.get_growth_comparison(stock_code)
def get_dupont_analysis_comparison(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.valuation_crawler.get_dupont_analysis_comparison(stock_code)
def get_valuation_comparison(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.valuation_crawler.get_valuation_comparison(stock_code)
def get_market_performance(self, secucode: str) -> Optional[List[Dict[Any, Any]]]:
return self.market_spider.get_market_performance(secucode)
def get_current_plate_changes(self, page_size: int = 10) -> Optional[List[Dict[Any, Any]]]:
return self.market_spider.get_current_plate_changes(page_size)
def get_current_count_changes(self) -> Optional[List[Dict[Any, Any]]]:
return self.market_spider.get_current_count_changes()
def get_macroeconomic_research(self, begin_time: str,
end_time: str) -> Optional[List[Dict[Any, Any]]]:
return self.market_spider.get_macroeconomic_research(begin_time, end_time)
def get_real_time_market_indices(self) -> List[Dict]:
return self.real_time_spider.get_real_time_market_indices()
def get_smart_score(self, stock_code: str) -> Optional[Dict[Any, Any]]:
return self.smart_review_crawler.get_smart_score(stock_code)
def get_smart_score_rank(self, stock_code: str) -> Optional[Dict[Any, Any]]:
return self.smart_review_crawler.get_smart_score_rank(stock_code)
def get_top_rated_stocks(self, page_size: int = 10) -> Optional[List[Dict[Any, Any]]]:
return self.smart_review_crawler.get_top_rated_stocks(page_size)
def get_main_force_control(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.smart_review_crawler.get_main_force_control(stock_code)
def get_participation_wish(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.smart_review_crawler.get_participation_wish(stock_code)
def get_intraday_changes(self, stock_code: str) -> Optional[List[Dict[Any, Any]]]:
return self.kline_spider.get_intraday_changes(stock_code)
