VeighNa量化社区
你的开源社区量化交易平台 | vn.py | vnpy
Member
avatar
加入于:
帖子: 3
声望: 0

`from vnpy.event import EventEngine
from vnpy.trader.engine import MainEngine
from vnpy.trader.ui import MainWindow, create_qapp

from vnpy_ib import IbGateway
from vnpy_datamanager import DataManagerApp

def main():
"""主入口函数"""
qapp = create_qapp()

event_engine = EventEngine()
main_engine = MainEngine(event_engine)
main_engine.add_gateway(IbGateway)
main_engine.add_app(DataManagerApp)
main_engine.connect(setting={
    'TWS地址': '127.0.0.1',
    'TWS端口': 7497,
    '客户号':1,
    '交易账户':'test'
},gateway_name='IB')



main_window = MainWindow(main_engine, event_engine)
main_window.showMaximized()

qapp.exec()


if name == "main":
main()
`
代码可以跑起来,窗口也有,但大多都是基于UI操作的,但是如果我想通过代码来下载历史数据应该怎么写呢?

Super Moderator
avatar
加入于:
帖子: 100
声望: 10

导入必要的模块

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()

© 2015-2022 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号

沪公网安备 31011502017034号

【用户协议】
【隐私政策】
【免责条款】