"""
统一的股票代码处理工具
整合股票分类、标准化、转换等功能
"""
import re
from typing import Dict, Optional, List, Tuple
from .stock_market_classifier import get_stock_classifier, MarketType, ExchangeType
class StockSymbolProcessor:
"""股票代码处理器 - 统一处理股票代码的分类、标准化和转换"""
def __init__(self):
self.classifier = get_stock_classifier()
def process_symbol(self, symbol: str) -> Dict:
"""
全面处理股票代码，返回所有相关信息
Args:
symbol: 原始股票代码
Returns:
Dict: 包含分类、标准化后的各种格式
"""
# 基础分类
classification = self.classifier.classify_stock(symbol)
# 生成各种标准化格式
formats = self._generate_all_formats(symbol, classification)
# 数据源策略
data_sources = self._get_data_source_strategy(classification)
# 合并结果
result = {
**classification,
"formats": formats,
"market_simple_name": self.get_market_simple_name(symbol, classification),
"data_sources": data_sources,
"original": symbol,
}
return result
def _generate_all_formats(self, symbol: str, classification: Dict) -> Dict:
"""生成所有需要的代码格式"""
return {
"tushare": self.get_tushare_format(symbol, classification),
"akshare": self.get_akshare_format(symbol, classification),
"yfinance": self.get_yfinance_format(symbol, classification),
"news_api": self.get_news_api_format(symbol, classification),
"cache_key": self.get_cache_key(symbol, classification),
"display": self.get_display_format(symbol, classification),
}
def get_tushare_format(self, symbol: str, classification: Dict = None) -> str:
"""获取Tushare API格式的代码"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
if classification["is_china"]:
# A股：确保有交易所后缀
clean_code = self._extract_base_code(symbol)
if "." in symbol and symbol.count(".") == 1:
return symbol
if clean_code.isdigit() and len(clean_code) == 6:
if clean_code.startswith(("60", "68")):
return f"{clean_code}.SH"
elif clean_code.startswith(("00", "30")):
return f"{clean_code}.SZ"
elif clean_code.startswith("8"):
return f"{clean_code}.BJ"
return symbol
elif classification["is_hk"]:
# 港股：Tushare港股格式
clean_code = self._extract_base_code(symbol)
if clean_code.isdigit():
return f"{clean_code.zfill(5)}.HK"
return symbol
else:
# 美股：Tushare不支持，返回原始代码
return self._extract_base_code(symbol).upper()
def get_akshare_format(self, symbol: str, classification: Dict = None) -> str:
"""获取AKShare API格式的代码"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
if classification["is_china"]:
# A股：纯数字代码
return self._extract_base_code(symbol)
elif classification["is_hk"]:
# 港股：5位数字代码
clean_code = self._extract_base_code(symbol)
if clean_code.isdigit():
return clean_code.zfill(5)
return clean_code
else:
# 美股：去除后缀的大写代码
return self._extract_base_code(symbol).upper()
def get_yfinance_format(self, symbol: str, classification: Dict = None) -> str:
"""获取YFinance API格式的代码"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
if classification["is_china"]:
# A股：添加Yahoo Finance后缀
clean_code = self._extract_base_code(symbol)
if clean_code.startswith(("60", "68")):
return f"{clean_code}.SS" # 上交所
else:
return f"{clean_code}.SZ" # 深交所
elif classification["is_hk"]:
# 港股：添加.HK后缀
clean_code = self._extract_base_code(symbol)
if clean_code.isdigit():
# yfinance 对港股代码格式的最终规律:
# 1. 先将代码转为整数，去除所有前导零 (e.g., '00700' -> 700)
numeric_code = int(clean_code)
# 2. 根据原始代码的位数决定格式
# - 如果原始代码不足4位 (e.g., 5, 700), 则补零到4位
# - 如果原始代码是4位或5位 (e.g., 9988, 89888), 则保持原样
if numeric_code < 1000: # 适用于 1, 5, 700 等
return f"{numeric_code:04d}.HK"
else: # 适用于 9988, 89888 等
return f"{numeric_code}.HK"
return f"{clean_code.upper()}.HK"
else:
# 美股：纯代码，去除后缀
return self._extract_base_code(symbol).upper()
def get_news_api_format(self, symbol: str, classification: Dict = None) -> str:
"""获取新闻API格式的代码"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
if classification["is_china"]:
# A股新闻：纯数字代码
return self._extract_base_code(symbol)
elif classification["is_hk"]:
# 港股新闻：标准5位.HK格式
clean_code = self._extract_base_code(symbol)
if clean_code.isdigit():
return f"{clean_code.zfill(5)}.HK"
elif clean_code.upper().endswith(".HK"):
return symbol.upper()
else:
return f"{clean_code}.HK"
else:
# 美股新闻：纯代码，去除所有后缀
clean_code = self._extract_base_code(symbol).upper()
# 移除常见美股后缀
us_suffixes = [".US", ".NASDAQ", ".NYSE", ".NMS"]
for suffix in us_suffixes:
if clean_code.endswith(suffix):
clean_code = clean_code.replace(suffix, "")
break
return clean_code
def get_cache_key(self, symbol: str, classification: Dict = None) -> str:
"""获取缓存键格式的代码"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
clean_code = self._extract_base_code(symbol)
if classification["is_china"]:
# A股缓存：6位数字
return clean_code
elif classification["is_hk"]:
# 港股缓存：5位数字
if clean_code.isdigit():
return clean_code.zfill(5)
return clean_code
else:
# 美股缓存：大写字母代码
return clean_code.upper()
def get_display_format(self, symbol: str, classification: Dict = None) -> str:
"""获取显示格式的代码"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
if classification["is_china"]:
# A股显示：代码 + 交易所
clean_code = self._extract_base_code(symbol)
if clean_code.startswith(("60", "68")):
return f"{clean_code}(SH)"
elif clean_code.startswith(("00", "30")):
return f"{clean_code}(SZ)"
elif clean_code.startswith("8"):
return f"{clean_code}(BJ)"
return clean_code
elif classification["is_hk"]:
# 港股显示：5位代码.HK
clean_code = self._extract_base_code(symbol)
if clean_code.isdigit():
return f"{clean_code.zfill(5)}.HK"
return symbol
else:
# 美股显示：纯代码
return self._extract_base_code(symbol).upper()
def _extract_base_code(self, symbol: str) -> str:
"""提取基础股票代码，去除所有后缀"""
if not symbol:
return ""
# 去除常见后缀
suffixes = [
".SH",
".SZ",
".BJ",
".SS",
".XSHE",
".XSHG", # A股后缀
".HK",
".hk", # 港股后缀
".US",
".NASDAQ",
".NYSE",
".NMS", # 美股后缀
]
clean_symbol = symbol.strip().upper()
for suffix in suffixes:
if clean_symbol.endswith(suffix.upper()):
clean_symbol = clean_symbol[: -len(suffix)]
break
return clean_symbol
def _get_data_source_strategy(self, classification: Dict) -> Dict:
"""根据市场类型获取数据源策略"""
if classification["is_china"]:
return {
"fundamentals": ["tushare", "akshare"],
"market_data": ["tushare", "akshare"],
"news": ["akshare", "eastmoney", "sina"],
"priority": "tushare",
}
elif classification["is_hk"]:
return {
"fundamentals": ["tushare", "akshare", "yfinance"],
"market_data": ["tushare", "akshare", "yfinance"],
"news": ["akshare", "yfinance", "rss"],
"priority": "tushare",
}
else: # US market
return {
"fundamentals": ["yfinance", "akshare"],
"market_data": ["yfinance", "akshare"],
"news": ["yfinance", "finnhub", "alpha_vantage", "newsapi"],
"priority": "yfinance",
}
def get_market_simple_name(self, symbol: str, classification: Dict = None) -> str:
"""获取简化的市场名称"""
if classification is None:
classification = self.classifier.classify_stock(symbol)
if classification["is_china"]:
return "china"
elif classification["is_hk"]:
return "hk"
elif classification["is_us"]:
return "us"
else:
return "unknown"
def batch_process_symbols(self, symbols: List[str]) -> Dict[str, Dict]:
"""批量处理股票代码"""
results = {}
for symbol in symbols:
try:
results[symbol] = self.process_symbol(symbol)
except Exception as e:
results[symbol] = {
"error": str(e),
"original": symbol,
"formats": {},
"data_sources": {},
}
return results
def validate_symbol_format(self, symbol: str, expected_market: str = None) -> Dict:
"""验证股票代码格式"""
result = {"is_valid": False, "market": None, "errors": [], "suggestions": []}
if not symbol or not symbol.strip():
result["errors"].append("股票代码不能为空")
return result
classification = self.classifier.classify_stock(symbol)
if classification["market"] == "未知":
result["errors"].append("无法识别的股票代码格式")
result["suggestions"].append("请检查股票代码是否正确")
else:
result["is_valid"] = True
result["market"] = classification["market"]
# 如果指定了期望市场，进行验证
if expected_market:
expected_map = {"china": "is_china", "hk": "is_hk", "us": "is_us"}
if expected_market in expected_map and not classification.get(
expected_map[expected_market], False
):
result["is_valid"] = False
result["errors"].append(f"股票代码不属于{expected_market}市场")
return result
# 全局处理器实例
_processor = None
def get_symbol_processor() -> StockSymbolProcessor:
"""获取股票代码处理器实例（单例模式）"""
global _processor
if _processor is None:
_processor = StockSymbolProcessor()
return _processor
# 便利函数
def process_symbol(symbol: str) -> Dict:
"""处理股票代码的便利函数"""
return get_symbol_processor().process_symbol(symbol)
def get_tushare_format(symbol: str) -> str:
"""获取Tushare格式代码的便利函数"""
return get_symbol_processor().get_tushare_format(symbol)
def get_akshare_format(symbol: str) -> str:
"""获取AKShare格式代码的便利函数"""
return get_symbol_processor().get_akshare_format(symbol)
def get_yfinance_format(symbol: str) -> str:
"""获取YFinance格式代码的便利函数"""
return get_symbol_processor().get_yfinance_format(symbol)
def get_news_api_format(symbol: str) -> str:
"""获取新闻API格式代码的便利函数"""
return get_symbol_processor().get_news_api_format(symbol)
def get_cache_key(symbol: str) -> str:
"""获取缓存键的便利函数"""
return get_symbol_processor().get_cache_key(symbol)
def get_market_simple_name(symbol: str) -> str:
"""获取市场简化名称的便利函数"""
return get_symbol_processor().get_market_simple_name(symbol)
def get_data_source_strategy(symbol: str) -> Dict:
"""获取数据源策略的便利函数"""
processor = get_symbol_processor()
classification = processor.classifier.classify_stock(symbol)
return processor._get_data_source_strategy(classification)
