一、导入tick数据
使用vnpy的UI的数据导入界面，只有导入1m,1h的功能，没有发现导入tick数据的选项，所以只能自己写代码导入。
查看存储tick数据的数据库表格，字段如下：
id INTEGER 0 0 0 -1 0
symbol VARCHAR 255 0 0 0 0
exchange VARCHAR 255 0 0 0 0
datetime DATETIME 0 0 0 0 0
name VARCHAR 255 0 0 0 0
volume REAL 0 0 0 0 0
turnover REAL 0 0 0 0 0
open_interest REAL 0 0 0 0 0
last_price REAL 0 0 0 0 0
last_volume REAL 0 0 0 0 0
limit_up REAL 0 0 0 0 0
limit_down REAL 0 0 0 0 0
open_price REAL 0 0 0 0 0
high_price REAL 0 0 0 0 0
low_price REAL 0 0 0 0 0
pre_close REAL 0 0 0 0 0
bid_price_1 REAL 0 0 0 0 0
bid_price_2 REAL 0 0 -1 0 0
bid_price_3 REAL 0 0 -1 0 0
bid_price_4 REAL 0 0 -1 0 0
bid_price_5 REAL 0 0 -1 0 0
ask_price_1 REAL 0 0 0 0 0
ask_price_2 REAL 0 0 -1 0 0
ask_price_3 REAL 0 0 -1 0 0
ask_price_4 REAL 0 0 -1 0 0
ask_price_5 REAL 0 0 -1 0 0
bid_volume_1 REAL 0 0 0 0 0
bid_volume_2 REAL 0 0 -1 0 0
bid_volume_3 REAL 0 0 -1 0 0
bid_volume_4 REAL 0 0 -1 0 0
bid_volume_5 REAL 0 0 -1 0 0
ask_volume_1 REAL 0 0 0 0 0
ask_volume_2 REAL 0 0 -1 0 0
ask_volume_3 REAL 0 0 -1 0 0
ask_volume_4 REAL 0 0 -1 0 0
ask_volume_5 REAL 0 0 -1 0 0
localtime DATETIME 0 0 -1 0 0
然后将存储数据的csv文件与下面的代码放到一个文件夹下面，运行代码就可以实现导入了。
import os
import csv
from datetime import datetime, time
from vnpy.trader.constant import Exchange
from vnpy.trader.database import get_database # 替换为 get_database()
from vnpy.trader.object import TickData
def run_load_csv():
"""遍历同一文件夹内所有csv文件，并且载入到数据库中"""
for file in os.listdir("."):
if not file.endswith(".csv"):
continue
print("载入文件：", file)
csv_load(file)
def csv_load(file):
"""读取csv文件内容，并写入到数据库中"""
# 获取数据库实例（新版 VeighNa 方式）
database = get_database()
with open(file, "r") as f:
reader = csv.DictReader(f)
ticks = []
start = None
count = 0
for item in reader:
# 生成 datetime（修复原代码中的错误）
date = item["交易日"]
second = item["最后修改时间"]
millisecond = item["最后修改毫秒"]
standard_time = date + " " + second + "." + millisecond
dt = datetime.strptime(standard_time, "%Y%m%d %H:%M:%S.%f")
local_dt = datetime.strptime(date, "%Y%m%d") # 修复原代码中的 date.strptime 错误
# 过滤非交易时段（15:01-20:59）
if dt.time() > time(15, 1) and dt.time() < time(20, 59):
continue
tick = TickData(
symbol="IF88", # 注意：原代码中 symbol 未定义，建议从 CSV 读取或改为固定值
datetime=dt,
exchange=Exchange.SHFE,
name="IF88", # 原代码中 symbol 未定义，此处需修正
volume=float(item["数量"]),
turnover=float(item["成交金额"]),
open_interest=float(item["持仓量"]),
last_price=float(item["最新价"]),
limit_up=float(item["涨停板价"]),
limit_down=float(item["跌停板价"]),
open_price=float(item["今开盘"]),
high_price=float(item["最高价"]),
low_price=float(item["最低价"]),
pre_close=float(item["昨收盘"]),
bid_price_1=float(item["申买价一"]),
bid_price_2=float(item["申买价二"]),
bid_price_3=float(item["申买价三"]),
bid_price_4=float(item["申买价四"]),
bid_price_5=float(item["申买价五"]),
bid_volume_1=float(item["申买量一"]),
bid_volume_2=float(item["申买量二"]),
bid_volume_3=float(item["申买量三"]),
bid_volume_4=float(item["申买量四"]),
bid_volume_5=float(item["申买量五"]),
ask_price_1=float(item["申卖价一"]),
ask_price_2=float(item["申卖价二"]),
ask_price_3=float(item["申卖价三"]),
ask_price_4=float(item["申卖价四"]),
ask_price_5=float(item["申卖价五"]),
ask_volume_1=float(item["申卖量一"]),
ask_volume_2=float(item["申卖量二"]),
ask_volume_3=float(item["申卖量三"]),
ask_volume_4=float(item["申卖量四"]),
ask_volume_5=float(item["申卖量五"]),
localtime=local_dt,
gateway_name="DB",
)
ticks.append(tick)
# 统计信息
count += 1
if not start:
start = tick.datetime
end = tick.datetime
database.save_tick_data(ticks) # 使用 database.save_tick_data 替代原 database_manager
print("插入数据", start, "-", end, "总数量：", count)
if __name__ == "__main__":
run_load_csv()
二、回测代码：
需要导入BacktestingMode，以及设定回测模式为BacktestingMode。
from vnpy_ctastrategy.backtesting import BacktestingEngine
from vnpy_ctastrategy.base import BacktestingMode
from vnpy_ctastrategy.strategies.tick01 import tick01
from datetime import datetime
# 初始化回测引擎
engine = BacktestingEngine()
# 设置回测参数
engine.set_parameters(
vt_symbol="IF88.SHFE", # 股票/期货代码（需确保数据存在）
interval="1m", # 日线级别
start=datetime(2023, 4, 1),
end=datetime(2023, 4, 4), # 延长回测时间以提高可靠性
rate=0.0001, # 手续费率（万三）
slippage=1, # 滑点（股票0.01元/期货1跳）
size=10, # 每手合约乘数（股票为100）
pricetick=1, # 最小价格变动（股票0.01元）
capital=1000_000, # 初始资金
mode=BacktestingMode.TICK,
)
# 添加策略（可调整布林带参数）
engine.add_strategy(tick01, {})
# 加载数据 + 运行回测
engine.load_data()
engine.run_backtesting()
# 计算回测结果
df = engine.calculate_result() # 返回回测结果的DataFrame
engine.calculate_statistics() # 计算统计指标
# 显示图表（关键步骤！）
engine.show_chart()
# 如果图表未弹出，尝试强制显示（适用于Jupyter Notebook）
import matplotlib.pyplot as plt
plt.show()
参考：https://www.vnpy.com/forum/topic/1992-vn-pyshe-qu-jing-xuan-8-tickshu-ju-zai-ru-he-ce-lue-hui-ce
三、导入Bar数据
查看C:\veighna_studio\Lib\site-packages\vnpy\trader下面的database.py,可以发现下面的代码，那么如果要导入1分钟bar的数据，那么也可以参考导入tick数据就可以了。
class BaseDatabase(ABC):
"""
Abstract database class for connecting to different database.
"""
@abstractmethod
def save_bar_data(self, bars: List[BarData], stream: bool = False) -> bool:
"""
Save bar data into database.
"""
pass
@abstractmethod
def save_tick_data(self, ticks: List[TickData], stream: bool = False) -> bool:
"""
Save tick data into database.
"""
pass