TIANHE
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""
|
|
数据源工厂
|
|
根据市场类型返回对应的数据源
|
|
"""
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
from app.data_sources.base import BaseDataSource
|
|
from app.utils.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class DataSourceFactory:
|
|
"""数据源工厂"""
|
|
|
|
_sources: Dict[str, BaseDataSource] = {}
|
|
|
|
@classmethod
|
|
def get_source(cls, market: str) -> BaseDataSource:
|
|
"""
|
|
获取指定市场的数据源
|
|
|
|
Args:
|
|
market: 市场类型 (Crypto, USStock, AShare, HShare)
|
|
|
|
Returns:
|
|
数据源实例
|
|
"""
|
|
if market not in cls._sources:
|
|
cls._sources[market] = cls._create_source(market)
|
|
return cls._sources[market]
|
|
|
|
@classmethod
|
|
def get_data_source(cls, name: str) -> BaseDataSource:
|
|
"""
|
|
Backward compatible alias used by older code paths.
|
|
|
|
Some modules historically called `get_data_source("binance")` to fetch a crypto data source.
|
|
In the localized Python backend we primarily use `get_source("Crypto")`.
|
|
"""
|
|
key = (name or "").strip().lower()
|
|
if key in ("crypto", "binance", "okx", "bybit", "bitget", "kucoin", "gate", "mexc", "kraken", "coinbase"):
|
|
return cls.get_source("Crypto")
|
|
if key in ("futures",):
|
|
return cls.get_source("Futures")
|
|
# Default to Crypto for safety (most callers want a ticker for crypto pairs).
|
|
return cls.get_source("Crypto")
|
|
|
|
@classmethod
|
|
def _create_source(cls, market: str) -> BaseDataSource:
|
|
"""创建数据源实例"""
|
|
if market == 'Crypto':
|
|
from app.data_sources.crypto import CryptoDataSource
|
|
return CryptoDataSource()
|
|
elif market == 'USStock':
|
|
from app.data_sources.us_stock import USStockDataSource
|
|
return USStockDataSource()
|
|
elif market == 'AShare':
|
|
from app.data_sources.cn_stock import AShareDataSource
|
|
return AShareDataSource()
|
|
elif market == 'HShare':
|
|
from app.data_sources.cn_stock import HShareDataSource
|
|
return HShareDataSource()
|
|
elif market == 'Forex':
|
|
from app.data_sources.forex import ForexDataSource
|
|
return ForexDataSource()
|
|
elif market == 'Futures':
|
|
from app.data_sources.futures import FuturesDataSource
|
|
return FuturesDataSource()
|
|
else:
|
|
raise ValueError(f"不支持的市场类型: {market}")
|
|
|
|
@classmethod
|
|
def get_kline(
|
|
cls,
|
|
market: str,
|
|
symbol: str,
|
|
timeframe: str,
|
|
limit: int,
|
|
before_time: Optional[int] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
获取K线数据的便捷方法
|
|
|
|
Args:
|
|
market: 市场类型
|
|
symbol: 交易对/股票代码
|
|
timeframe: 时间周期
|
|
limit: 数据条数
|
|
before_time: 获取此时间之前的数据
|
|
|
|
Returns:
|
|
K线数据列表
|
|
"""
|
|
try:
|
|
source = cls.get_source(market)
|
|
klines = source.get_kline(symbol, timeframe, limit, before_time)
|
|
|
|
# 确保数据按时间排序
|
|
klines.sort(key=lambda x: x['time'])
|
|
|
|
return klines
|
|
except Exception as e:
|
|
logger.error(f"Failed to fetch K-lines {market}:{symbol} - {str(e)}")
|
|
return []
|
|
|