这玩意儿我觉得还是由官方来实现最好，一直盼着图形这块好用些。
这玩意儿我觉得还是由官方来实现最好，一直盼着图形这块好用些。
谢谢大神分享，再顶一下楼上，同盼
用Python的交易员 wrote:
给你加个精华
首先，数据质量的好坏决定策略实盘的可行性，例如下载仿真历史数据，出现某段时间缺失的情况，这就要求建模分析的第一步是对行情数据进行画图，通过人眼初步判断数据的好坏质量：即无缺失的行情部分、也无明显的“异常”数据点，建议官方考虑一下，增加CTA回测N分钟的 K线显示。
版本更新太快了，3.0及后面的版本底层改为了PySide6，函数调用有点变化，小白一个，搞了大半天才跑起来，输入1分钟都会报错，具体原因不知道，直接复制粘贴覆盖vnpy_ctabacktester\ui\widget.py文件即可，vnpy3.5版本的widget.py代码如下所示：
import csv
import subprocess
from datetime import datetime, timedelta
from copy import copy
from typing import List
import numpy as np
import pyqtgraph as pg
from pandas import DataFrame
from vnpy.trader.constant import Interval, Direction, Exchange, Offset
from vnpy.trader.engine import MainEngine, BaseEngine
from vnpy.trader.ui import QtCore, QtWidgets, QtGui
from vnpy.trader.ui.widget import BaseMonitor, BaseCell, DirectionCell, EnumCell
from vnpy.event import Event, EventEngine
from vnpy.chart import ChartWidget, CandleItem, VolumeItem
from vnpy.trader.utility import load_json, save_json
from vnpy.trader.object import BarData, TradeData, OrderData
from vnpy.trader.database import DB_TZ
from vnpy_ctastrategy.backtesting import DailyResult
from vnpy.usertools.chart_items import SmaItem,BollItem
from ..engine import (
APP_NAME,
EVENT_BACKTESTER_LOG,
EVENT_BACKTESTER_BACKTESTING_FINISHED,
EVENT_BACKTESTER_OPTIMIZATION_FINISHED,
OptimizationSetting
)
def ConvertBar(bars,show_minute):
newbars=[]
i=len(bars)//show_minute
if len(bars)>show_minute*i:
i=i+1
newbars=[x for x in range(i)]
i=0
while i<((len(bars)//show_minute)+1):
if len(bars)==show_minute*i:
break
datetime=bars[show_minute*i].datetime
symbol=bars[show_minute*i].symbol
exchange=bars[show_minute*i].exchange
interval=bars[show_minute*i].interval
volume=bars[show_minute*i].volume
open_interest=bars[show_minute*i].open_interest
open_price=bars[show_minute*i].open_price
close_price=bars[show_minute*i].close_price
high_price=bars[show_minute*i].high_price
low_price=bars[show_minute*i].low_price
j=1
while j <show_minute:
if (show_minute*i+j)==len(bars):
break
high_price=max(high_price,bars[show_minute*i+j].high_price)
low_price=min(low_price,bars[show_minute*i+j].low_price)
close_price=bars[show_minute*i+j].close_price
j=j+1
newbars[i] = BarData(
symbol=symbol,
exchange=Exchange(exchange),
datetime=datetime,
interval=Interval(interval),
volume=volume,
open_price=open_price,
high_price=high_price,
open_interest=open_interest,
low_price=low_price,
close_price=close_price,
gateway_name="sqlite"
)
i=i+1
return newbars
class BacktesterManager(QtWidgets.QWidget):
""""""
setting_filename: str = "cta_backtester_setting.json"
signal_log: QtCore.Signal = QtCore.Signal(Event)
signal_backtesting_finished: QtCore.Signal = QtCore.Signal(Event)
signal_optimization_finished: QtCore.Signal = QtCore.Signal(Event)
def __init__(self, main_engine: MainEngine, event_engine: EventEngine) -> None:
""""""
super().__init__()
self.main_engine: MainEngine = main_engine
self.event_engine: EventEngine = event_engine
self.backtester_engine: BaseEngine = main_engine.get_engine(APP_NAME)
self.class_names: list = []
self.settings: dict = {}
self.target_display: str = ""
self.init_ui()
self.register_event()
self.backtester_engine.init_engine()
self.init_strategy_settings()
self.load_backtesting_setting()
def init_strategy_settings(self) -> None:
""""""
self.class_names = self.backtester_engine.get_strategy_class_names()
self.class_names.sort()
for class_name in self.class_names:
setting: dict = self.backtester_engine.get_default_setting(class_name)
self.settings[class_name] = setting
self.class_combo.addItems(self.class_names)
def init_ui(self) -> None:
""""""
self.setWindowTitle("CTA回测")
# Setting Part
self.class_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
self.symbol_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit("IF88.CFFEX")
self.interval_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
for interval in Interval:
self.interval_combo.addItem(interval.value)
end_dt: datetime = datetime.now()
start_dt: datetime = end_dt - timedelta(days=3 * 365)
self.start_date_edit: QtWidgets.QDateEdit = QtWidgets.QDateEdit(
QtCore.QDate(
start_dt.year,
start_dt.month,
start_dt.day
)
)
self.end_date_edit: QtWidgets.QDateEdit = QtWidgets.QDateEdit(
QtCore.QDate.currentDate()
)
self.rate_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit("0.000025")
self.slippage_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit("0.2")
self.size_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit("300")
self.pricetick_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit("0.2")
self.capital_line: QtWidgets.QLineEdit = QtWidgets.QLineEdit("1000000")
backtesting_button: QtWidgets.QPushButton = QtWidgets.QPushButton("开始回测")
backtesting_button.clicked.connect(self.start_backtesting)
optimization_button: QtWidgets.QPushButton = QtWidgets.QPushButton("参数优化")
optimization_button.clicked.connect(self.start_optimization)
self.result_button: QtWidgets.QPushButton = QtWidgets.QPushButton("优化结果")
self.result_button.clicked.connect(self.show_optimization_result)
self.result_button.setEnabled(False)
downloading_button: QtWidgets.QPushButton = QtWidgets.QPushButton("下载数据")
downloading_button.clicked.connect(self.start_downloading)
self.order_button: QtWidgets.QPushButton = QtWidgets.QPushButton("委托记录")
self.order_button.clicked.connect(self.show_backtesting_orders)
self.order_button.setEnabled(False)
self.trade_button: QtWidgets.QPushButton = QtWidgets.QPushButton("成交记录")
self.trade_button.clicked.connect(self.show_backtesting_trades)
self.trade_button.setEnabled(False)
self.daily_button: QtWidgets.QPushButton = QtWidgets.QPushButton("每日盈亏")
self.daily_button.clicked.connect(self.show_daily_results)
self.daily_button.setEnabled(False)
self.candle_button: QtWidgets.QPushButton = QtWidgets.QPushButton("K线图表")
self.candle_button.clicked.connect(self.show_candle_chart)
self.candle_button.setEnabled(False)
edit_button: QtWidgets.QPushButton = QtWidgets.QPushButton("代码编辑")
edit_button.clicked.connect(self.edit_strategy_code)
reload_button: QtWidgets.QPushButton = QtWidgets.QPushButton("策略重载")
reload_button.clicked.connect(self.reload_strategy_class)
for button in [
backtesting_button,
optimization_button,
downloading_button,
self.result_button,
self.order_button,
self.trade_button,
self.daily_button,
self.candle_button,
edit_button,
reload_button
]:
button.setFixedHeight(button.sizeHint().height() * 2)
form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
form.addRow("交易策略", self.class_combo)
form.addRow("本地代码", self.symbol_line)
form.addRow("K线周期", self.interval_combo)
form.addRow("开始日期", self.start_date_edit)
form.addRow("结束日期", self.end_date_edit)
form.addRow("手续费率", self.rate_line)
form.addRow("交易滑点", self.slippage_line)
form.addRow("合约乘数", self.size_line)
form.addRow("价格跳动", self.pricetick_line)
form.addRow("回测资金", self.capital_line)
result_grid: QtWidgets.QGridLayout = QtWidgets.QGridLayout()
result_grid.addWidget(self.trade_button, 0, 0)
result_grid.addWidget(self.order_button, 0, 1)
result_grid.addWidget(self.daily_button, 1, 0)
result_grid.addWidget(self.candle_button, 1, 1)
left_vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
left_vbox.addLayout(form)
left_vbox.addWidget(backtesting_button)
left_vbox.addWidget(downloading_button)
left_vbox.addStretch()
left_vbox.addLayout(result_grid)
left_vbox.addStretch()
left_vbox.addWidget(optimization_button)
left_vbox.addWidget(self.result_button)
left_vbox.addStretch()
left_vbox.addWidget(edit_button)
left_vbox.addWidget(reload_button)
# Result part
self.statistics_monitor: StatisticsMonitor = StatisticsMonitor()
self.log_monitor: QtWidgets.QTextEdit = QtWidgets.QTextEdit()
self.chart: BacktesterChart = BacktesterChart()
chart: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
chart.addWidget(self.chart)
self.trade_dialog: BacktestingResultDialog = BacktestingResultDialog(
self.main_engine,
self.event_engine,
"回测成交记录",
BacktestingTradeMonitor
)
self.order_dialog: BacktestingResultDialog = BacktestingResultDialog(
self.main_engine,
self.event_engine,
"回测委托记录",
BacktestingOrderMonitor
)
self.daily_dialog: BacktestingResultDialog = BacktestingResultDialog(
self.main_engine,
self.event_engine,
"回测每日盈亏",
DailyResultMonitor
)
# Candle Chart
self.candle_dialog: CandleChartDialog = CandleChartDialog()
# Layout
middle_vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
middle_vbox.addWidget(self.statistics_monitor)
middle_vbox.addWidget(self.log_monitor)
left_hbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
left_hbox.addLayout(left_vbox)
left_hbox.addLayout(middle_vbox)
left_widget: QtWidgets.QWidget = QtWidgets.QWidget()
left_widget.setLayout(left_hbox)
right_vbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
right_vbox.addWidget(self.chart)
right_widget: QtWidgets.QWidget = QtWidgets.QWidget()
right_widget.setLayout(right_vbox)
hbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
hbox.addWidget(left_widget)
hbox.addWidget(right_widget)
self.setLayout(hbox)
def load_backtesting_setting(self) -> None:
""""""
setting: dict = load_json(self.setting_filename)
if not setting:
return
self.class_combo.setCurrentIndex(
self.class_combo.findText(setting["class_name"])
)
self.symbol_line.setText(setting["vt_symbol"])
self.interval_combo.setCurrentIndex(
self.interval_combo.findText(setting["interval"])
)
start_str: str = setting.get("start", "")
if start_str:
start_dt: QtCore.QDate = QtCore.QDate.fromString(start_str, "yyyy-MM-dd")
self.start_date_edit.setDate(start_dt)
self.rate_line.setText(str(setting["rate"]))
self.slippage_line.setText(str(setting["slippage"]))
self.size_line.setText(str(setting["size"]))
self.pricetick_line.setText(str(setting["pricetick"]))
self.capital_line.setText(str(setting["capital"]))
def register_event(self) -> None:
""""""
self.signal_log.connect(self.process_log_event)
self.signal_backtesting_finished.connect(
self.process_backtesting_finished_event)
self.signal_optimization_finished.connect(
self.process_optimization_finished_event)
self.event_engine.register(EVENT_BACKTESTER_LOG, self.signal_log.emit)
self.event_engine.register(
EVENT_BACKTESTER_BACKTESTING_FINISHED, self.signal_backtesting_finished.emit)
self.event_engine.register(
EVENT_BACKTESTER_OPTIMIZATION_FINISHED, self.signal_optimization_finished.emit)
def process_log_event(self, event: Event) -> None:
""""""
msg = event.data
self.write_log(msg)
def write_log(self, msg) -> None:
""""""
timestamp: str = datetime.now().strftime("%H:%M:%S")
msg: str = f"{timestamp}\t{msg}"
self.log_monitor.append(msg)
def process_backtesting_finished_event(self, event: Event) -> None:
""""""
statistics: dict = self.backtester_engine.get_result_statistics()
self.statistics_monitor.set_data(statistics)
df: DataFrame = self.backtester_engine.get_result_df()
self.chart.set_data(df)
self.trade_button.setEnabled(True)
self.order_button.setEnabled(True)
self.daily_button.setEnabled(True)
# Tick data can not be displayed using candle chart
interval: str = self.interval_combo.currentText()
if interval != Interval.TICK.value:
self.candle_button.setEnabled(True)
def process_optimization_finished_event(self, event: Event) -> None:
""""""
self.write_log("请点击[优化结果]按钮查看")
self.result_button.setEnabled(True)
def start_backtesting(self) -> None:
""""""
class_name: str = self.class_combo.currentText()
if not class_name:
self.write_log("请选择要回测的策略")
return
vt_symbol: str = self.symbol_line.text()
interval: str = self.interval_combo.currentText()
start: datetime = self.start_date_edit.dateTime().toPython()
end: datetime = self.end_date_edit.dateTime().toPython()
rate: float = float(self.rate_line.text())
slippage: float = float(self.slippage_line.text())
size: float = float(self.size_line.text())
pricetick: float = float(self.pricetick_line.text())
capital: float = float(self.capital_line.text())
# Check validity of vt_symbol
if "." not in vt_symbol:
self.write_log("本地代码缺失交易所后缀，请检查")
return
_, exchange_str = vt_symbol.split(".")
if exchange_str not in Exchange.__members__:
self.write_log("本地代码的交易所后缀不正确，请检查")
return
# Save backtesting parameters
backtesting_setting: dict = {
"class_name": class_name,
"vt_symbol": vt_symbol,
"interval": interval,
"start": start.strftime("%Y-%m-%d"),
"rate": rate,
"slippage": slippage,
"size": size,
"pricetick": pricetick,
"capital": capital
}
save_json(self.setting_filename, backtesting_setting)
# Get strategy setting
old_setting: dict = self.settings[class_name]
dialog: BacktestingSettingEditor = BacktestingSettingEditor(class_name, old_setting)
i: int = dialog.exec()
if i != dialog.Accepted:
return
new_setting: dict = dialog.get_setting()
self.settings[class_name] = new_setting
result: bool = self.backtester_engine.start_backtesting(
class_name,
vt_symbol,
interval,
start,
end,
rate,
slippage,
size,
pricetick,
capital,
new_setting
)
if result:
self.statistics_monitor.clear_data()
self.chart.clear_data()
self.trade_button.setEnabled(False)
self.order_button.setEnabled(False)
self.daily_button.setEnabled(False)
self.candle_button.setEnabled(False)
self.trade_dialog.clear_data()
self.order_dialog.clear_data()
self.daily_dialog.clear_data()
self.candle_dialog.clear_data()
def start_optimization(self) -> None:
""""""
class_name: str = self.class_combo.currentText()
vt_symbol: str = self.symbol_line.text()
interval: str = self.interval_combo.currentText()
start: object = self.start_date_edit.dateTime().toPython()
end: object = self.end_date_edit.dateTime().toPython()
rate: float = float(self.rate_line.text())
slippage: float = float(self.slippage_line.text())
size: float = float(self.size_line.text())
pricetick: float = float(self.pricetick_line.text())
capital: float = float(self.capital_line.text())
parameters: dict = self.settings[class_name]
dialog: OptimizationSettingEditor = OptimizationSettingEditor(class_name, parameters)
i: int = dialog.exec()
if i != dialog.Accepted:
return
optimization_setting, use_ga = dialog.get_setting()
self.target_display: str = dialog.target_display
self.backtester_engine.start_optimization(
class_name,
vt_symbol,
interval,
start,
end,
rate,
slippage,
size,
pricetick,
capital,
optimization_setting,
use_ga
)
self.result_button.setEnabled(False)
def start_downloading(self) -> None:
""""""
vt_symbol: str = self.symbol_line.text()
interval: str = self.interval_combo.currentText()
start_date: QtCore.QDate = self.start_date_edit.date()
end_date: QtCore.QDate = self.end_date_edit.date()
start: datetime = datetime(
start_date.year(),
start_date.month(),
start_date.day(),
)
start: datetime = start.replace(tzinfo=DB_TZ)
end: datetime = datetime(
end_date.year(),
end_date.month(),
end_date.day(),
23,
59,
59,
)
end: datetime = end.replace(tzinfo=DB_TZ)
self.backtester_engine.start_downloading(
vt_symbol,
interval,
start,
end
)
def show_optimization_result(self) -> None:
""""""
result_values: list = self.backtester_engine.get_result_values()
dialog: OptimizationResultMonitor = OptimizationResultMonitor(
result_values,
self.target_display
)
dialog.exec_()
def show_backtesting_trades(self) -> None:
""""""
if not self.trade_dialog.is_updated():
trades: List[TradeData] = self.backtester_engine.get_all_trades()
self.trade_dialog.update_data(trades)
self.trade_dialog.exec_()
def show_backtesting_orders(self) -> None:
""""""
if not self.order_dialog.is_updated():
orders: List[OrderData] = self.backtester_engine.get_all_orders()
self.order_dialog.update_data(orders)
self.order_dialog.exec_()
def show_daily_results(self) -> None:
""""""
if not self.daily_dialog.is_updated():
results: List[DailyResult] = self.backtester_engine.get_all_daily_results()
self.daily_dialog.update_data(results)
self.daily_dialog.exec_()
def show_candle_chart(self):
""""""
if not self.candle_dialog.is_updated():
show_min=1
i, okPressed = QtWidgets.QInputDialog.getInt(self, "k线显示周期","请输入(分钟数):", 1, 0, 1500, 1)
if okPressed:
show_min=i
history = self.backtester_engine.get_history_data()
for ix, bar in enumerate(history):
self.candle_dialog.dt_ix_map_min[bar.datetime] = ix
newhistory=ConvertBar(history,show_min)
self.candle_dialog.update_history(newhistory)
trades = self.backtester_engine.get_all_trades()
self.candle_dialog.update_trades(trades,show_min)
self.candle_dialog.exec_()
def edit_strategy_code(self) -> None:
""""""
class_name: str = self.class_combo.currentText()
if not class_name:
return
file_path: str = self.backtester_engine.get_strategy_class_file(class_name)
cmd: list = ["code", file_path]
p: subprocess.CompletedProcess = subprocess.run(cmd, shell=True)
if p.returncode:
QtWidgets.QMessageBox.warning(
self,
"启动代码编辑器失败",
"请检查是否安装了Visual Studio Code，并将其路径添加到了系统全局变量中！"
)
def reload_strategy_class(self) -> None:
""""""
self.backtester_engine.reload_strategy_class()
current_strategy_name: str = self.class_combo.currentText()
self.class_combo.clear()
self.init_strategy_settings()
ix: int = self.class_combo.findText(current_strategy_name)
self.class_combo.setCurrentIndex(ix)
def show(self) -> None:
""""""
self.showMaximized()
class StatisticsMonitor(QtWidgets.QTableWidget):
""""""
KEY_NAME_MAP: dict = {
"start_date": "首个交易日",
"end_date": "最后交易日",
"total_days": "总交易日",
"profit_days": "盈利交易日",
"loss_days": "亏损交易日",
"capital": "起始资金",
"end_balance": "结束资金",
"total_return": "总收益率",
"annual_return": "年化收益",
"max_drawdown": "最大回撤",
"max_ddpercent": "百分比最大回撤",
"total_net_pnl": "总盈亏",
"total_commission": "总手续费",
"total_slippage": "总滑点",
"total_turnover": "总成交额",
"total_trade_count": "总成交笔数",
"daily_net_pnl": "日均盈亏",
"daily_commission": "日均手续费",
"daily_slippage": "日均滑点",
"daily_turnover": "日均成交额",
"daily_trade_count": "日均成交笔数",
"daily_return": "日均收益率",
"return_std": "收益标准差",
"sharpe_ratio": "夏普比率",
"return_drawdown_ratio": "收益回撤比"
}
def __init__(self) -> None:
""""""
super().__init__()
self.cells: dict = {}
self.init_ui()
def init_ui(self) -> None:
""""""
self.setRowCount(len(self.KEY_NAME_MAP))
self.setVerticalHeaderLabels(list(self.KEY_NAME_MAP.values()))
self.setColumnCount(1)
self.horizontalHeader().setVisible(False)
self.horizontalHeader().setSectionResizeMode(
QtWidgets.QHeaderView.Stretch
)
self.setEditTriggers(self.NoEditTriggers)
for row, key in enumerate(self.KEY_NAME_MAP.keys()):
cell: QtWidgets.QTableWidgetItem = QtWidgets.QTableWidgetItem()
self.setItem(row, 0, cell)
self.cells[key] = cell
def clear_data(self) -> None:
""""""
for cell in self.cells.values():
cell.setText("")
def set_data(self, data: dict) -> None:
""""""
data["capital"] = f"{data['capital']:,.2f}"
data["end_balance"] = f"{data['end_balance']:,.2f}"
data["total_return"] = f"{data['total_return']:,.2f}%"
data["annual_return"] = f"{data['annual_return']:,.2f}%"
data["max_drawdown"] = f"{data['max_drawdown']:,.2f}"
data["max_ddpercent"] = f"{data['max_ddpercent']:,.2f}%"
data["total_net_pnl"] = f"{data['total_net_pnl']:,.2f}"
data["total_commission"] = f"{data['total_commission']:,.2f}"
data["total_slippage"] = f"{data['total_slippage']:,.2f}"
data["total_turnover"] = f"{data['total_turnover']:,.2f}"
data["daily_net_pnl"] = f"{data['daily_net_pnl']:,.2f}"
data["daily_commission"] = f"{data['daily_commission']:,.2f}"
data["daily_slippage"] = f"{data['daily_slippage']:,.2f}"
data["daily_turnover"] = f"{data['daily_turnover']:,.2f}"
data["daily_trade_count"] = f"{data['daily_trade_count']:,.2f}"
data["daily_return"] = f"{data['daily_return']:,.2f}%"
data["return_std"] = f"{data['return_std']:,.2f}%"
data["sharpe_ratio"] = f"{data['sharpe_ratio']:,.2f}"
data["return_drawdown_ratio"] = f"{data['return_drawdown_ratio']:,.2f}"
for key, cell in self.cells.items():
value = data.get(key, "")
cell.setText(str(value))
class BacktestingSettingEditor(QtWidgets.QDialog):
"""
For creating new strategy and editing strategy parameters.
"""
def __init__(
self, class_name: str, parameters: dict
) -> None:
""""""
super(BacktestingSettingEditor, self).__init__()
self.class_name: str = class_name
self.parameters: dict = parameters
self.edits: dict = {}
self.init_ui()
def init_ui(self) -> None:
""""""
form: QtWidgets.QFormLayout = QtWidgets.QFormLayout()
# Add vt_symbol and name edit if add new strategy
self.setWindowTitle(f"策略参数配置：{self.class_name}")
button_text: str = "确定"
parameters: dict = self.parameters
for name, value in parameters.items():
type_ = type(value)
edit: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(value))
if type_ is int:
validator: QtGui.QIntValidator = QtGui.QIntValidator()
edit.setValidator(validator)
elif type_ is float:
validator: QtGui.QDoubleValidator = QtGui.QDoubleValidator()
edit.setValidator(validator)
form.addRow(f"{name} {type_}", edit)
self.edits[name] = (edit, type_)
button: QtWidgets.QPushButton = QtWidgets.QPushButton(button_text)
button.clicked.connect(self.accept)
form.addRow(button)
widget: QtWidgets.QWidget = QtWidgets.QWidget()
widget.setLayout(form)
scroll: QtWidgets.QScrollArea = QtWidgets.QScrollArea()
scroll.setWidgetResizable(True)
scroll.setWidget(widget)
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
vbox.addWidget(scroll)
self.setLayout(vbox)
def get_setting(self) -> dict:
""""""
setting: dict = {}
for name, tp in self.edits.items():
edit, type_ = tp
value_text = edit.text()
if type_ == bool:
if value_text == "True":
value = True
else:
value = False
else:
value = type_(value_text)
setting[name] = value
return setting
class BacktesterChart(pg.GraphicsLayoutWidget):
""""""
def __init__(self) -> None:
""""""
super().__init__(title="Backtester Chart")
self.dates: dict = {}
self.init_ui()
def init_ui(self) -> None:
""""""
pg.setConfigOptions(antialias=True)
# Create plot widgets
self.balance_plot = self.addPlot(
title="账户净值",
axisItems={"bottom": DateAxis(self.dates, orientation="bottom")}
)
self.nextRow()
self.drawdown_plot = self.addPlot(
title="净值回撤",
axisItems={"bottom": DateAxis(self.dates, orientation="bottom")}
)
self.nextRow()
self.pnl_plot = self.addPlot(
title="每日盈亏",
axisItems={"bottom": DateAxis(self.dates, orientation="bottom")}
)
self.nextRow()
self.distribution_plot = self.addPlot(title="盈亏分布")
# Add curves and bars on plot widgets
self.balance_curve = self.balance_plot.plot(
pen=pg.mkPen("#ffc107", width=3)
)
dd_color: str = "#303f9f"
self.drawdown_curve = self.drawdown_plot.plot(
fillLevel=-0.3, brush=dd_color, pen=dd_color
)
profit_color: str = 'r'
loss_color: str = 'g'
self.profit_pnl_bar = pg.BarGraphItem(
x=[], height=[], width=0.3, brush=profit_color, pen=profit_color
)
self.loss_pnl_bar = pg.BarGraphItem(
x=[], height=[], width=0.3, brush=loss_color, pen=loss_color
)
self.pnl_plot.addItem(self.profit_pnl_bar)
self.pnl_plot.addItem(self.loss_pnl_bar)
distribution_color: str = "#6d4c41"
self.distribution_curve = self.distribution_plot.plot(
fillLevel=-0.3, brush=distribution_color, pen=distribution_color
)
def clear_data(self) -> None:
""""""
self.balance_curve.setData([], [])
self.drawdown_curve.setData([], [])
self.profit_pnl_bar.setOpts(x=[], height=[])
self.loss_pnl_bar.setOpts(x=[], height=[])
self.distribution_curve.setData([], [])
def set_data(self, df) -> None:
""""""
if df is None:
return
count: int = len(df)
self.dates.clear()
for n, date in enumerate(df.index):
self.dates[n] = date
# Set data for curve of balance and drawdown
self.balance_curve.setData(df["balance"])
self.drawdown_curve.setData(df["drawdown"])
# Set data for daily pnl bar
profit_pnl_x: list = []
profit_pnl_height: list = []
loss_pnl_x: list = []
loss_pnl_height: list = []
for count, pnl in enumerate(df["net_pnl"]):
if pnl >= 0:
profit_pnl_height.append(pnl)
profit_pnl_x.append(count)
else:
loss_pnl_height.append(pnl)
loss_pnl_x.append(count)
self.profit_pnl_bar.setOpts(x=profit_pnl_x, height=profit_pnl_height)
self.loss_pnl_bar.setOpts(x=loss_pnl_x, height=loss_pnl_height)
# Set data for pnl distribution
hist, x = np.histogram(df["net_pnl"], bins="auto")
x = x[:-1]
self.distribution_curve.setData(x, hist)
class DateAxis(pg.AxisItem):
"""Axis for showing date data"""
def __init__(self, dates: dict, *args, **kwargs) -> None:
""""""
super().__init__(*args, **kwargs)
self.dates: dict = dates
def tickStrings(self, values, scale, spacing) -> list:
""""""
strings: list = []
for v in values:
dt = self.dates.get(v, "")
strings.append(str(dt))
return strings
class OptimizationSettingEditor(QtWidgets.QDialog):
"""
For setting up parameters for optimization.
"""
DISPLAY_NAME_MAP: dict = {
"总收益率": "total_return",
"夏普比率": "sharpe_ratio",
"收益回撤比": "return_drawdown_ratio",
"日均盈亏": "daily_net_pnl"
}
def __init__(
self, class_name: str, parameters: dict
) -> None:
""""""
super().__init__()
self.class_name: str = class_name
self.parameters: dict = parameters
self.edits: dict = {}
self.optimization_setting: OptimizationSetting = None
self.use_ga: bool = False
self.init_ui()
def init_ui(self) -> None:
""""""
QLabel: QtWidgets.QLabel = QtWidgets.QLabel
self.target_combo: QtWidgets.QComboBox = QtWidgets.QComboBox()
self.target_combo.addItems(list(self.DISPLAY_NAME_MAP.keys()))
grid: QtWidgets.QGridLayout = QtWidgets.QGridLayout()
grid.addWidget(QLabel("目标"), 0, 0)
grid.addWidget(self.target_combo, 0, 1, 1, 3)
grid.addWidget(QLabel("参数"), 1, 0)
grid.addWidget(QLabel("开始"), 1, 1)
grid.addWidget(QLabel("步进"), 1, 2)
grid.addWidget(QLabel("结束"), 1, 3)
# Add vt_symbol and name edit if add new strategy
self.setWindowTitle(f"优化参数配置：{self.class_name}")
validator: QtGui.QDoubleValidator = QtGui.QDoubleValidator()
row: int = 2
for name, value in self.parameters.items():
type_ = type(value)
if type_ not in [int, float]:
continue
start_edit: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(value))
step_edit: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(1))
end_edit: QtWidgets.QLineEdit = QtWidgets.QLineEdit(str(value))
for edit in [start_edit, step_edit, end_edit]:
edit.setValidator(validator)
grid.addWidget(QLabel(name), row, 0)
grid.addWidget(start_edit, row, 1)
grid.addWidget(step_edit, row, 2)
grid.addWidget(end_edit, row, 3)
self.edits[name] = {
"type": type_,
"start": start_edit,
"step": step_edit,
"end": end_edit
}
row += 1
parallel_button: QtWidgets.QPushButton = QtWidgets.QPushButton("多进程优化")
parallel_button.clicked.connect(self.generate_parallel_setting)
grid.addWidget(parallel_button, row, 0, 1, 4)
row += 1
ga_button: QtWidgets.QPushButton = QtWidgets.QPushButton("遗传算法优化")
ga_button.clicked.connect(self.generate_ga_setting)
grid.addWidget(ga_button, row, 0, 1, 4)
widget: QtWidgets.QWidget = QtWidgets.QWidget()
widget.setLayout(grid)
scroll: QtWidgets.QScrollArea = QtWidgets.QScrollArea()
scroll.setWidgetResizable(True)
scroll.setWidget(widget)
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
vbox.addWidget(scroll)
self.setLayout(vbox)
def generate_ga_setting(self) -> None:
""""""
self.use_ga: bool = True
self.generate_setting()
def generate_parallel_setting(self) -> None:
""""""
self.use_ga: bool = False
self.generate_setting()
def generate_setting(self) -> None:
""""""
self.optimization_setting = OptimizationSetting()
self.target_display: str = self.target_combo.currentText()
target_name: str = self.DISPLAY_NAME_MAP[self.target_display]
self.optimization_setting.set_target(target_name)
for name, d in self.edits.items():
type_ = d["type"]
start_value = type_(d["start"].text())
step_value = type_(d["step"].text())
end_value = type_(d["end"].text())
if start_value == end_value:
self.optimization_setting.add_parameter(name, start_value)
else:
self.optimization_setting.add_parameter(
name,
start_value,
end_value,
step_value
)
self.accept()
def get_setting(self) -> None:
""""""
return self.optimization_setting, self.use_ga
class OptimizationResultMonitor(QtWidgets.QDialog):
"""
For viewing optimization result.
"""
def __init__(
self, result_values: list, target_display: str
) -> None:
""""""
super().__init__()
self.result_values: list = result_values
self.target_display: str = target_display
self.init_ui()
def init_ui(self) -> None:
""""""
self.setWindowTitle("参数优化结果")
self.resize(1100, 500)
# Creat table to show result
table: QtWidgets.QTableWidget = QtWidgets.QTableWidget()
table.setColumnCount(2)
table.setRowCount(len(self.result_values))
table.setHorizontalHeaderLabels(["参数", self.target_display])
table.setEditTriggers(table.NoEditTriggers)
table.verticalHeader().setVisible(False)
table.horizontalHeader().setSectionResizeMode(
0, QtWidgets.QHeaderView.ResizeToContents
)
table.horizontalHeader().setSectionResizeMode(
1, QtWidgets.QHeaderView.Stretch
)
for n, tp in enumerate(self.result_values):
setting, target_value, _ = tp
setting_cell: QtWidgets.QTableWidgetItem = QtWidgets.QTableWidgetItem(str(setting))
target_cell: QtWidgets.QTableWidgetItem = QtWidgets.QTableWidgetItem(f"{target_value:.2f}")
setting_cell.setTextAlignment(QtCore.Qt.AlignCenter)
target_cell.setTextAlignment(QtCore.Qt.AlignCenter)
table.setItem(n, 0, setting_cell)
table.setItem(n, 1, target_cell)
# Create layout
button: QtWidgets.QPushButton = QtWidgets.QPushButton("保存")
button.clicked.connect(self.save_csv)
hbox: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
hbox.addStretch()
hbox.addWidget(button)
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
vbox.addWidget(table)
vbox.addLayout(hbox)
self.setLayout(vbox)
def save_csv(self) -> None:
"""
Save table data into a csv file
"""
path, _ = QtWidgets.QFileDialog.getSaveFileName(
self, "保存数据", "", "CSV(*.csv)")
if not path:
return
with open(path, "w") as f:
writer = csv.writer(f, lineterminator="\n")
writer.writerow(["参数", self.target_display])
for tp in self.result_values:
setting, target_value, _ = tp
row_data: list = [str(setting), str(target_value)]
writer.writerow(row_data)
class BacktestingTradeMonitor(BaseMonitor):
"""
Monitor for backtesting trade data.
"""
headers: dict = {
"tradeid": {"display": "成交号 ", "cell": BaseCell, "update": False},
"orderid": {"display": "委托号", "cell": BaseCell, "update": False},
"symbol": {"display": "代码", "cell": BaseCell, "update": False},
"exchange": {"display": "交易所", "cell": EnumCell, "update": False},
"direction": {"display": "方向", "cell": DirectionCell, "update": False},
"offset": {"display": "开平", "cell": EnumCell, "update": False},
"price": {"display": "价格", "cell": BaseCell, "update": False},
"volume": {"display": "数量", "cell": BaseCell, "update": False},
"datetime": {"display": "时间", "cell": BaseCell, "update": False},
"gateway_name": {"display": "接口", "cell": BaseCell, "update": False},
}
class BacktestingOrderMonitor(BaseMonitor):
"""
Monitor for backtesting order data.
"""
headers: dict = {
"orderid": {"display": "委托号", "cell": BaseCell, "update": False},
"symbol": {"display": "代码", "cell": BaseCell, "update": False},
"exchange": {"display": "交易所", "cell": EnumCell, "update": False},
"type": {"display": "类型", "cell": EnumCell, "update": False},
"direction": {"display": "方向", "cell": DirectionCell, "update": False},
"offset": {"display": "开平", "cell": EnumCell, "update": False},
"price": {"display": "价格", "cell": BaseCell, "update": False},
"volume": {"display": "总数量", "cell": BaseCell, "update": False},
"traded": {"display": "已成交", "cell": BaseCell, "update": False},
"status": {"display": "状态", "cell": EnumCell, "update": False},
"datetime": {"display": "时间", "cell": BaseCell, "update": False},
"gateway_name": {"display": "接口", "cell": BaseCell, "update": False},
}
class FloatCell(BaseCell):
"""
Cell used for showing pnl data.
"""
def __init__(self, content, data) -> None:
""""""
content: str = f"{content:.2f}"
super().__init__(content, data)
class DailyResultMonitor(BaseMonitor):
"""
Monitor for backtesting daily result.
"""
headers: dict = {
"date": {"display": "日期", "cell": BaseCell, "update": False},
"trade_count": {"display": "成交笔数", "cell": BaseCell, "update": False},
"start_pos": {"display": "开盘持仓", "cell": BaseCell, "update": False},
"end_pos": {"display": "收盘持仓", "cell": BaseCell, "update": False},
"turnover": {"display": "成交额", "cell": FloatCell, "update": False},
"commission": {"display": "手续费", "cell": FloatCell, "update": False},
"slippage": {"display": "滑点", "cell": FloatCell, "update": False},
"trading_pnl": {"display": "交易盈亏", "cell": FloatCell, "update": False},
"holding_pnl": {"display": "持仓盈亏", "cell": FloatCell, "update": False},
"total_pnl": {"display": "总盈亏", "cell": FloatCell, "update": False},
"net_pnl": {"display": "净盈亏", "cell": FloatCell, "update": False},
}
class BacktestingResultDialog(QtWidgets.QDialog):
""""""
def __init__(
self,
main_engine: MainEngine,
event_engine: EventEngine,
title: str,
table_class: QtWidgets.QTableWidget
) -> None:
""""""
super().__init__()
self.main_engine: MainEngine = main_engine
self.event_engine: EventEngine = event_engine
self.title: str = title
self.table_class: QtWidgets.QTableWidget = table_class
self.updated: bool = False
self.init_ui()
def init_ui(self) -> None:
""""""
self.setWindowTitle(self.title)
self.resize(1100, 600)
self.table: QtWidgets.QTableWidget = self.table_class(self.main_engine, self.event_engine)
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
vbox.addWidget(self.table)
self.setLayout(vbox)
def clear_data(self) -> None:
""""""
self.updated = False
self.table.setRowCount(0)
def update_data(self, data: list) -> None:
""""""
self.updated = True
data.reverse()
for obj in data:
self.table.insert_new_row(obj)
def is_updated(self) -> bool:
""""""
return self.updated
class CandleChartDialog(QtWidgets.QDialog):
""""""
def __init__(self) -> None:
""""""
super().__init__()
self.updated: bool = False
self.dt_ix_map: dict = {}
self.ix_bar_map: dict = {}
self.dt_ix_map_min = {}
self.high_price = 0
self.low_price = 0
self.price_range = 0
self.items: list = []
self.init_ui()
def init_ui(self) -> None:
""""""
self.setWindowTitle("回测K线图表")
self.resize(1400, 800)
# Create chart widget
self.chart: ChartWidget = ChartWidget()
self.chart.add_plot("candle", hide_x_axis=True)
self.chart.add_plot("volume", maximum_height=200)
self.chart.add_item(CandleItem, "candle", "candle")
self.chart.add_item(VolumeItem, "volume", "volume")
self.chart.add_item(SmaItem, "sma", "candle")
self.chart.add_item(BollItem, "boll", "candle")
self.chart.add_cursor()
# Add scatter item for showing tradings
self.trade_scatter = pg.ScatterPlotItem()
candle_plot = self.chart.get_plot("candle")
candle_plot.addItem(self.trade_scatter)
# Create help widget
text1: str = "红色虚线 —— 盈利交易"
label1: QtWidgets.QLabel = QtWidgets.QLabel(text1)
label1.setStyleSheet("color:red")
text2: str = "绿色虚线 —— 亏损交易"
label2: QtWidgets.QLabel = QtWidgets.QLabel(text2)
label2.setStyleSheet("color:#00FF00")
text3: str = "黄色向上箭头 —— 买入开仓 Buy"
label3: QtWidgets.QLabel = QtWidgets.QLabel(text3)
label3.setStyleSheet("color:yellow")
text4: str = "黄色向下箭头 —— 卖出平仓 Sell"
label4: QtWidgets.QLabel = QtWidgets.QLabel(text4)
label4.setStyleSheet("color:yellow")
text5: str = "紫红向下箭头 —— 卖出开仓 Short"
label5: QtWidgets.QLabel = QtWidgets.QLabel(text5)
label5.setStyleSheet("color:magenta")
text6: str = "紫红向上箭头 —— 买入平仓 Cover"
label6: QtWidgets.QLabel = QtWidgets.QLabel(text6)
label6.setStyleSheet("color:magenta")
hbox1: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
hbox1.addStretch()
hbox1.addWidget(label1)
hbox1.addStretch()
hbox1.addWidget(label2)
hbox1.addStretch()
hbox2: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
hbox2.addStretch()
hbox2.addWidget(label3)
hbox2.addStretch()
hbox2.addWidget(label4)
hbox2.addStretch()
hbox3: QtWidgets.QHBoxLayout = QtWidgets.QHBoxLayout()
hbox3.addStretch()
hbox3.addWidget(label5)
hbox3.addStretch()
hbox3.addWidget(label6)
hbox3.addStretch()
# Set layout
vbox: QtWidgets.QVBoxLayout = QtWidgets.QVBoxLayout()
vbox.addWidget(self.chart)
vbox.addLayout(hbox1)
vbox.addLayout(hbox2)
vbox.addLayout(hbox3)
self.setLayout(vbox)
def update_history(self, history: list) -> None:
""""""
self.updated = True
self.chart.update_history(history)
for ix, bar in enumerate(history):
self.ix_bar_map[ix] = bar
self.dt_ix_map[bar.datetime] = ix
if not self.high_price:
self.high_price = bar.high_price
self.low_price = bar.low_price
else:
self.high_price = max(self.high_price, bar.high_price)
self.low_price = min(self.low_price, bar.low_price)
self.price_range = self.high_price - self.low_price
def update_trades(self, trades: list,show_min:int):
""""""
trade_data = []
for trade in trades:
ix = self.dt_ix_map_min[trade.datetime]
ix=ix//show_min
scatter = {
"pos": (ix, trade.price),
"data": 1,
"size": 14,
"pen": pg.mkPen((255, 255, 255))
}
if trade.direction == Direction.LONG:
scatter_symbol = "t1" # Up arrow
else:
scatter_symbol = "t" # Down arrow
if trade.offset == Offset.OPEN:
scatter_brush = pg.mkBrush((255, 255, 0)) # Yellow
else:
scatter_brush = pg.mkBrush((0, 0, 255)) # Blue
scatter["symbol"] = scatter_symbol
scatter["brush"] = scatter_brush
trade_data.append(scatter)
self.trade_scatter.setData(trade_data)
def clear_data(self) -> None:
""""""
self.updated = False
candle_plot: pg.PlotItem = self.chart.get_plot("candle")
for item in self.items:
candle_plot.removeItem(item)
self.items.clear()
self.chart.clear_all()
self.dt_ix_map.clear()
self.ix_bar_map.clear()
def is_updated(self) -> bool:
""""""
return self.updated
def generate_trade_pairs(trades: list) -> list:
""""""
long_trades: list = []
short_trades: list = []
trade_pairs: list = []
for trade in trades:
trade: TradeData = copy(trade)
if trade.direction == Direction.LONG:
same_direction: list = long_trades
opposite_direction: list = short_trades
else:
same_direction: list = short_trades
opposite_direction: list = long_trades
while trade.volume and opposite_direction:
open_trade: TradeData = opposite_direction[0]
close_volume = min(open_trade.volume, trade.volume)
d: dict = {
"open_dt": open_trade.datetime,
"open_price": open_trade.price,
"close_dt": trade.datetime,
"close_price": trade.price,
"direction": open_trade.direction,
"volume": close_volume,
}
trade_pairs.append(d)
open_trade.volume -= close_volume
if not open_trade.volume:
opposite_direction.pop(0)
trade.volume -= close_volume
if trade.volume:
same_direction.append(trade)
return trade_pairs版本：3.5
时间：2023-1-3 10:34:16
报错：packages\vnpy_ctabacktester\ui\widget.py", line 67, in ConvertBar
newbars[i] = BarData(
IndexError: list assignment index out of range
解决：67行缩进不对。本来，if len(bars)>show_minute*i:这个判断只是为向下取整后如果长度短了一个，就给他加上。但无论是不是加上,后面的代码，newbars=[x for x in range(i)]都应该被执行。
