导入必要的模块
from vnpy.event import EventEngine
from vnpy.trader.engine import MainEngine
from vnpy.trader.setting import SETTINGS
from vnpy_ib import IbGateway
from vnpy_datamanager import DataManagerApp
from vnpy.trader.database import get_database
from vnpy.trader.object import HistoryRequest, Interval, Exchange
from datetime import datetime
初始化事件引擎和主引擎
event_engine = EventEngine()
main_engine = MainEngine(event_engine)
添加 IB Gateway
main_engine.add_gateway(IbGateway)
连接 IB Gateway
main_engine.connect(setting={
'TWS地址': '127.0.0.1', # IB Gateway 的地址
'TWS端口': 7497, # IB Gateway 的端口
'客户号': 1, # 客户号
'交易账户': 'test' # 交易账户
}, gateway_name='IB')
添加 DataManagerApp
data_manager = main_engine.add_app(DataManagerApp)
获取数据库实例
database = get_database()
定义下载历史数据的函数
def download_and_save_data(symbol, exchange, interval, start, end):
"""
下载历史数据并保存到数据库
:param symbol: 合约代码 (如 "AAPL")
:param exchange: 交易所 (如 "SMART")
:param interval: 数据周期 (如 Interval.DAILY)
:param start: 开始日期 (如 "20230101")
:param end: 结束日期 (如 "20231231")
"""
# 创建历史数据请求
req = HistoryRequest(
symbol=symbol,
exchange=Exchange(exchange),
interval=interval,
start=datetime.strptime(start, "%Y%m%d"),
end=datetime.strptime(end, "%Y%m%d")
)
# 下载历史数据
data = data_manager.download_data(req)
# 保存数据到数据库
if data:
database.save_bar_data(data)
print(f"数据下载成功并保存到数据库:{symbol}.{exchange}")
else:
print(f"数据下载失败:{symbol}.{exchange}")
示例:下载 AAPL 的日线数据
download_and_save_data(
symbol="AAPL",
exchange="SMART",
interval=Interval.DAILY,
start="20230101",
end="20231231"
)
关闭主引擎
main_engine.close()
基于RAG技术开发的VeighNa AI助手,你的个人专属 Desk Quant