wehub-resource-sync
69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
from types import ModuleType
|
|
from collections.abc import Callable
|
|
from importlib import import_module
|
|
|
|
from .object import HistoryRequest, TickData, BarData
|
|
from .setting import SETTINGS
|
|
from .locale import _
|
|
|
|
|
|
class BaseDatafeed:
|
|
"""
|
|
Abstract datafeed class for connecting to different datafeed.
|
|
"""
|
|
|
|
def init(self, output: Callable = print) -> bool:
|
|
"""
|
|
Initialize datafeed service connection.
|
|
"""
|
|
return False
|
|
|
|
def query_bar_history(self, req: HistoryRequest, output: Callable = print) -> list[BarData]:
|
|
"""
|
|
Query history bar data.
|
|
"""
|
|
output(_("查询K线数据失败：没有正确配置数据服务"))
|
|
return []
|
|
|
|
def query_tick_history(self, req: HistoryRequest, output: Callable = print) -> list[TickData]:
|
|
"""
|
|
Query history tick data.
|
|
"""
|
|
output(_("查询Tick数据失败：没有正确配置数据服务"))
|
|
return []
|
|
|
|
|
|
datafeed: BaseDatafeed | None = None
|
|
|
|
|
|
def get_datafeed() -> BaseDatafeed:
|
|
""""""
|
|
# Return datafeed object if already inited
|
|
global datafeed
|
|
if datafeed:
|
|
return datafeed
|
|
|
|
# Read datafeed related global setting
|
|
datafeed_name: str = SETTINGS["datafeed.name"]
|
|
|
|
if not datafeed_name:
|
|
datafeed = BaseDatafeed()
|
|
|
|
print(_("没有配置要使用的数据服务，请修改全局配置中的datafeed相关内容"))
|
|
else:
|
|
module_name: str = f"vnpy_{datafeed_name}"
|
|
|
|
# Try to import datafeed module
|
|
try:
|
|
module: ModuleType = import_module(module_name)
|
|
|
|
# Create datafeed object from module
|
|
datafeed = module.Datafeed()
|
|
# Use base class if failed
|
|
except ModuleNotFoundError:
|
|
datafeed = BaseDatafeed()
|
|
|
|
print(_("无法加载数据服务模块，请运行 pip install {} 尝试安装").format(module_name))
|
|
|
|
return datafeed
|