mirror of
https://gitee.com/vnpy/vnpy.git
synced 2025-12-06 11:29:14 +08:00
68 lines
1.9 KiB
Python
68 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 |