1. 先看看CandleChartDialog的代码
class CandleChartDialog(QtWidgets.QDialog):
"""
"""
def __init__(self):
""""""
super().__init__()
self.dt_ix_map = {}
self.updated = False
self.init_ui()
def init_ui(self):
""""""
self.setWindowTitle("回测K线图表")
self.resize(1400, 800)
# Create chart widget
self.chart = 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_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)
# Set layout
vbox = QtWidgets.QVBoxLayout()
vbox.addWidget(self.chart)
self.setLayout(vbox)
def update_history(self, history: list):
""""""
self.updated = True
self.chart.update_history(history)
for ix, bar in enumerate(history):
self.dt_ix_map[bar.datetime] = ix
def update_trades(self, trades: list):
""""""
trade_data = []
for trade in trades:
ix = self.dt_ix_map[trade.datetime]
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):
""""""
self.updated = False
self.chart.clear_all()
self.dt_ix_map.clear()
self.trade_scatter.clear()
def is_updated(self):
""""""
return self.updated2. 再看看self.trade_scatter成员
2.1 一个特别的成员self.trade_scatter
它和其他的绘图部件不同，其他的都是ChartItem类型继承得到，只有它例外，它是一个ScatterPlotItem。
self.trade_scatter = pg.ScatterPlotItem()2.2 问题：CandleChartDialog不可以在非回测环境下使用
看看update_trades() 的代码中
for trade in trades:
ix = self.dt_ix_map[trade.datetime] # 查找一个成交单（trade）是属于哪个BarData的索引这里能够不出错，完全是因为在回测中，人为固定地把发出交易信号的那个bar的开始时间datetime赋值给了trade.datetime!
让我们来看看app\cta_strategy\backtesting.py中的class BacktestingEngine，它在bar模式的时候，是这样生成成交单的：
trade = TradeData(
symbol=order.symbol,
exchange=order.exchange,
orderid=order.orderid,
tradeid=str(self.trade_count),
direction=order.direction,
offset=order.offset,
price=trade_price,
volume=order.volume,
datetime=self.datetime,
gateway_name=self.gateway_name,
)其中trade中的datetime=self.datetime，而self.datetime是这样赋值的：
def run_backtesting(self):
""""""
if self.mode == BacktestingMode.BAR:
func = self.new_bar
else:
func = self.new_tick
self.strategy.on_init()
# Use the first [days] of history data for initializing strategy
day_count = 1
ix = 0
for ix, data in enumerate(self.history_data):
if self.datetime and data.datetime.day != self.datetime.day:
day_count += 1
if day_count >= self.days:
break
self.datetime = data.datetime # 它是用self.history_data中代表bar的data.datetime赋值的！
... ...可是我们知道trade.datetime是不可能总是恰好等于bar的开始时间datetime的，成交时间可能是一个bar形成期间的任何时间。
回到CandleChartDialog，由上面可知self.dt_ix_map的维护是update_history()维护的，它是由bar的开始时间datetime和其顺序ix构成的一个字典。
2.3 如果trade是实际的成交单，使用CandleChartDialog是一定会出错
原因是trade.datetime在self.dt_ix_map字典的键值中大概率是不存在的，因此ix = self.dt_ix_map[trade.datetime]语句会出错！而ScatterPlotItem是通过"pos": (ix, trade.price)来确定代表买卖的上下三角形来绘图的，因此CandleChartDialog是不可以简单地在显示实盘成交单的地方引用的。
3. 怎么修改既可显示回测成交单，又可以显示实盘成交单办？
思路：
3.1 增加如下成员：
self.trades:Dict[int,Dict[str,TradeData]] = {} # 其键值为成交发生bar的索引,内容为成交单字典3.2 增加add_trade()：
根据trade.datetime字段在self.dt_ix_map中所在位置的两个相邻时间dt0和dt1
满足：
def add_trade(self,trade:TradeData):
# 这里使用reverse=True，是考虑到实盘成交往往发生在最新的bar里，可以加快搜索速度
od = OrderedDict(sorted(self.dt_ix_map.items(),key = lambda t:t[0],reverse=True))
idx = 0
for dt,ix in od.items():
if dt <= trade.datetime:
idx = idx
break
# 注意：一个bar期间可能发生多个成交单
if idx not in self.trades:
self.trades[idx] = {trade.tradeid: trade}
else:
self.trades[idx][trade.tradeid] = trade3.3 update_trades()这么修改就OK了。
def update_trades(self, trades: list):
""""""
for trade in trades:
# 寻找每个成交单对应的bar的索引，并且保证为字典
self.add_trade(trade)
trade_data = []
for ix in self.trades:
for tradeid,trade in self.trades[ix].items():
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)4. 不一样的成交单显示解决办法
参见：典型绘图部件及使用方法
