VNPY3.5版本多进程优化，一旦数据量大了和参数多了几个就总是报错，特别是使用64核CPU,128GB内存时，如图所示，遗传与多进程都试过，然后使用2.6版本的dolphindb数据库，又使用vnpy2.6版本，结果又报错，遗传算法多进程优化，感觉还不之前1个进程遗传算法优化来的稳定
老师您好，按照您的提示操作了，回测还是没有显示胜率这些，小白一个，widget里面不太会，使用的是vnpy3.5版本，可以分享一下代码吗？万分感激
用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
用Python的交易员 wrote:
mengrong wrote:
这个问题现在应该不存在了吧，直接对单个策略做不同合约的实例化就可以吧。
现在去实现数据还会相互干扰吗？
现在好像没有这个问题，是因为源代码做了修订吗？只要不在策略类下定义共享的可变数据对象（比如dict/list/set），而是把这些对象定义放到类的init函数下，就不会有影响。
社区中就不断出现实时K线图表的需求，加上前段时间的文华事件，开发一套100%完全开源的高性能图表工具，可能也变成了挺有价值的工作。
后续会根据社区用户的反馈情况，来决定是否要进一步增加图表功能，例如：
多周期K线图表
任意时间范围
技术指标
画线分析
等等（图上能玩的太多了~）
如上图所示，vnpy3.5版本utility里面aroon函数与vnpy2.4版本utility里面aroon函数相反，是否是写错了？望回复，感谢
2.4版本：aroon_up, aroon_down = talib.AROON(self.high, self.low, n)
3.5版本：aroon_down, aroon_up = talib.AROON(self.high, self.low, n)
巴黎小矿工 wrote:
新建NewBarGenerator类，继承BarGenerator类，然后修改 update_bar_minute_window 函数。写cta策略时，使用NewBarGenerator可以定义一些奇奇怪怪的K线，例如7分钟、9分钟、11分钟等。
代码如下：
`
from typing import Callable
from vnpy.trader.constant import Interval
from vnpy_ctastrategy import BarData, BarGeneratorclass NewBarGenerator(BarGenerator):
def init(
self,
on_bar: Callable,
window: int = 0,
on_window_bar: Callable = None,
interval: Interval = Interval.MINUTE
):
super().init(on_bar, window, on_window_bar, interval)def update_bar_minute_window(self, bar: BarData) -> None: # If not inited, create window bar object if not self.window_bar: dt = bar.datetime.replace(second=0, microsecond=0) self.window_bar = BarData( symbol=bar.symbol, exchange=bar.exchange, datetime=dt, gateway_name=bar.gateway_name, open_price=bar.open_price, high_price=bar.high_price, low_price=bar.low_price ) # Otherwise, update high/low price into window bar else: self.window_bar.high_price = max( self.window_bar.high_price, bar.high_price ) self.window_bar.low_price = min( self.window_bar.low_price, bar.low_price ) # Update close price/volume/turnover into window bar self.window_bar.close_price = bar.close_price self.window_bar.volume += bar.volume self.window_bar.turnover += bar.turnover self.window_bar.open_interest = bar.open_interest # Check if window bar completed self.interval_count += 1 if not self.interval_count % self.window: self.interval_count = 0 self.on_window_bar(self.window_bar) self.window_bar = None`
区别如上图所示，看见有的人分享的K线自定义合成函数NewBarGenerator写的如下面代码所示
if not (bar.datetime.minute + 1) % self.window:在这一行代码与你写的有点点不同（ if not self.interval_count % self.window:），请问这具体的区别是？ 望指导一下，万分感激
你好，这是基于vnpy3.5的运行结果，如上图所示报错：BaseDatabase.load_bar_data() missing 1 required positional argument: 'self '，看了一些其他人的写法load_bar_data这么使用好像也没错，还需要怎么修改一下吗？望回复，万分感激'
from vnpy.trader.object import BarData因为vnpy3.0版本修改了数据库管理器的基类BaseDatabase，把上面数据库引用修改成了下面的这句
from vnpy.trader.database import BaseDatabaseimport multiprocessing
import time
from datetime import datetime
from vnpy.trader.constant import Exchange, Interval
from vnpy.chart import ChartWidget, VolumeItem, CandleItem
from vnpy.trader.ui import create_qapp, QtCore
from vnpy.trader.object import BarData
import os
from vnpy.trader.database import BaseDatabase
bar: BarData
def putbardata(q_1m,q_5m,q_30m,q_4h,su):
#从数据库中读取1分钟数据，你的数据库必须有下载好的数据。
bars = BaseDatabase.load_bar_data(
symbol="i888",
exchange=Exchange.DCE,
interval=Interval.MINUTE,
start=datetime(2020, 1, 1),
end=datetime(2025, 1, 1)
)
sudu = 0.055
i = 0
for bar in bars:
q_1m.put(bar)
q_5m.put(bar)
q_30m.put(bar)
q_4h.put(bar)
if i > 1200: #先快速播放一定数量的一分钟bar
if not su.empty():
sudu = int(su.get(True))
print("速度已经设定为：", sudu)
if i % 10 == 1 :
os.system("pause") #正常播放以后，每10个一分钟bar暂停一下，按任意键继续，不需要这个功能的可以删掉。
time.sleep(sudu)
i = i + 1
def MINUTE_5m(q):
app = create_qapp()
widget = ChartWidget()
widget.add_plot("candle", hide_x_axis=True)
widget.add_plot("volume", maximum_height=180)
widget.add_item(CandleItem, "candle", "candle")
widget.add_item(VolumeItem, "volume", "volume")
widget.add_cursor()
history : BarData
history = []
global i_5
i_5 = 0
global bar_
def update_bar():
global i_5
global bar_
if not q.empty():
bar = q.get(True)
if i_5 == 0 :
bar_ = bar
i_5 = 1
history.append(bar_)
if i_5 == 5 :
bar_ = bar
i_5 = 1
history.append(bar_)
else :
bar_.close_price = bar.close_price
if bar.high_price > bar_.high_price:
bar_.high_price = bar.high_price
if bar.low_price < bar_.low_price:
bar_.low_price = bar.low_price
bar_.volume = bar_.volume + bar.volume
i_5 = i_5 + 1
history[-1] = bar_ #这一段是把一分钟数据形成5分钟数据
widget.clear_all()
widget.update_history(history) #刷新图形数据
timer = QtCore.QTimer()
timer.timeout.connect(update_bar)
timer.start(50)
widget.setWindowTitle("五分钟") #设定五分钟窗口的标题和窗口大小以及位置
widget.setGeometry(0, 0, 900, 550)
widget.show()
app.exec_()
def MINUTE_30m(q): #30分钟和5分钟类似
app = create_qapp()
widget = ChartWidget()
widget.add_plot("candle", hide_x_axis=True)
widget.add_plot("volume", maximum_height=180)
widget.add_item(CandleItem, "candle", "candle")
widget.add_item(VolumeItem, "volume", "volume")
widget.add_cursor()
history: BarData
history = []
global i_30
i_30 = 0
global bar_
def update_bar():
global i_30
global bar_
if not q.empty():
bar = q.get(True)
if i_30 == 0 :
bar_ = bar
i_30 = 1
history.append(bar_)
if i_30 == 30 :
bar_ = bar
i_30 = 1
history.append(bar_)
else :
bar_.close_price = bar.close_price
if bar.high_price > bar_.high_price:
bar_.high_price = bar.high_price
if bar.low_price < bar_.low_price:
bar_.low_price = bar.low_price
bar_.volume = bar_.volume + bar.volume
i_30 = i_30 + 1
history[-1] = bar_
widget.clear_all()
widget.update_history(history)
timer = QtCore.QTimer()
timer.timeout.connect(update_bar)
timer.start(50)
widget.setWindowTitle("三十分钟")
widget.setGeometry(0, 560, 900, 550)
widget.show()
app.exec_()
def MINUTE_4h(q):
app = create_qapp()
widget = ChartWidget()
widget.add_plot("candle", hide_x_axis=True)
widget.add_plot("volume", maximum_height=180)
widget.add_item(CandleItem, "candle", "candle")
widget.add_item(VolumeItem, "volume", "volume")
widget.add_cursor()
history: BarData
history = []
global i_4h
i_4h = 0
global bar_
def update_bar():
global i_4h
global bar_
if not q.empty():
bar = q.get(True)
if i_4h == 0 :
bar_ = bar
i_4h = 1
history.append(bar_)
if i_4h == 240 :
bar_ = bar
i_4h = 1
history.append(bar_)
else :
bar_.close_price = bar.close_price
if bar.high_price > bar_.high_price:
bar_.high_price = bar.high_price
if bar.low_price < bar_.low_price:
bar_.low_price = bar.low_price
bar_.volume = bar_.volume + bar.volume
i_4h = i_4h + 1
history[-1] = bar_
widget.clear_all()
widget.update_history(history)
timer = QtCore.QTimer()
timer.timeout.connect(update_bar)
timer.start(50)
widget.setWindowTitle("四小时")
widget.setGeometry(860, 560, 1050, 530)
widget.show()
app.exec_()
def MINUTE(q): #一分钟的是最简单的，直接使用就好。
app = create_qapp()
widget = ChartWidget()
widget.add_plot("candle", hide_x_axis=True)
widget.add_plot("volume", maximum_height=180)
widget.add_item(CandleItem, "candle", "candle")
widget.add_item(VolumeItem, "volume", "volume")
widget.add_cursor()
def update_bar():
if not q.empty():
bar = q.get(True)
widget.update_bar(bar)
timer = QtCore.QTimer()
timer.timeout.connect(update_bar)
timer.start(50)
widget.setWindowTitle("一分钟")
widget.setGeometry(860, 15, 1050, 550)
widget.show()
app.exec_()
if __name__ == '__main__':
manager = multiprocessing.Manager()
q_1m = manager.Queue()
q_5m = manager.Queue()
q_30m = manager.Queue()
q_4h = manager.Queue()
su = manager.Queue()
pw = multiprocessing.Process(target=putbardata, args=(q_1m,q_5m,q_30m,q_4h,su))
pr_1m = multiprocessing.Process(target=MINUTE, args=(q_1m,))
pr_5m = multiprocessing.Process(target=MINUTE_5m, args=(q_5m,))
pr_30m = multiprocessing.Process(target=MINUTE_30m, args=(q_30m,))
pr_4h = multiprocessing.Process(target=MINUTE_4h, args=(q_4h,))
pw.start()
pr_1m.start()
pr_5m.start()
pr_30m.start()
pr_4h.start()
sudu = input("请输入速度:")
su.put(sudu)
time.sleep(1000000)
print('任务完成')
关于VNPY3.5版本可以下载tick，但是DataManagerApp无法查看tick，后续是否有新增DataManagerApp查看tick计划的考虑？
xiaohe wrote:
ib接口指数的查询后缀是IND
xiaohe wrote:
可以贴一下完整报错截图
VNPY3.5版本优化参数时，一旦参数多了就会报错，用的遗传算法优化，报错代码如下，报错截图如下，如何解决？
Traceback (most recent call last):
File "C:\veighna_studio\lib\threading.py", line 1319, in invoke_excepthook
hook(args)
TypeError: create_qapp.<locals>.threading_excepthook() missing 1 required positional argument: 'extra'
VNPY3.5版本优化参数时，一旦参数多了就会报错，报错代码如下，如何解决？
Traceback (most recent call last):
File "C:\veighna_studio\lib\threading.py", line 1319, in invoke_excepthook
hook(args)
TypeError: create_qapp.<locals>.threading_excepthook() missing 1 required positional argument: 'extra'买好包 wrote:
ib下载连续合约的问题已解决，但是没有复权选项呢，直接下载的数据和单个主力合约的数据还是有差别，在接近交割日的时候。
老师您好，咨询一下，ib下载连续合约的问题时具体怎么解决的呢？
用Python的交易员 wrote:
试试在策略里，把委托价格用round取整到整数好了
老师您好，我已在策略中用round函数取整，但是我反复测试，vnpy2.1.6版本中，即便是我人工下单（停止单），也提示The price does not conform to the minimum price variation for this contract人工下单也报错，去新加坡交易所查询了一下，好像新加坡交易所把A50的最小变动单位从最小变动2.5修改为1了，但是vnpy发出的单子还是2.5的倍数，这个该怎么解决了？ 所以系统一直报错提交中
升级vnstudio-2.1.6后，再运行2.1.3之前的所有版本就直接报错：AttributeError: module 'matplotlib' has no attribute 'plot'，大家有什么好的解决办法吗？
1：老师好，咨询一个问题，如图所示，使用的盈透IB的接口，策略发出的停止单或者限价单价格非交易所最小跳数（第一张图片日志显示），但是vnpy发单时已经取整了，盈透却一直显示提交中，老师具体知道原因吗？ 有什么好的解决方法吗？
报错：The price does not conform to the minimum price variation for this contract
