Update

backtest/backtest.go

@@ -37,8 +37,13 @@ type PlotData struct {
 	Equities  []float64
 }
 
+type DataState struct {
+	Time  int64 // ns
+	Index int   // datas 中的索引
+}
+
 type Backtest struct {
-	data             *dataloader.Data
+	datas            []*dataloader.Data
 	symbol           string
 	strategy         Strategy
 	exchanges        []ExchangeSim
@@ -50,8 +55,10 @@ type Backtest struct {
 
 	eLoggers []ExchangeLogger
 
-	start time.Time // 开始时间
-	end   time.Time // 结束时间
+	start         time.Time // 开始时间
+	end           time.Time // 结束时间
+	currentTimeNS int64     // ns
+	sortedDatas   []*DataState
 
 	startedAt time.Time // 运行开始时间
 	endedAt   time.Time // 运行结束时间
@@ -82,9 +89,9 @@ func init() {
 // NewBacktest Create backtest
 // data: The data
 // outputDir: 日志输出目录
-func NewBacktest(data *dataloader.Data, symbol string, start time.Time, end time.Time, strategy Strategy, exchanges []ExchangeSim, outputDir string) *Backtest {
+func NewBacktest(datas []*dataloader.Data, symbol string, start time.Time, end time.Time, strategy Strategy, exchanges []ExchangeSim, outputDir string) *Backtest {
 	b := &Backtest{
-		data:          data,
+		datas:         datas,
 		symbol:        symbol,
 		start:         start,
 		end:           end,
@@ -105,16 +112,16 @@ func NewBacktest(data *dataloader.Data, symbol string, start time.Time, end time
 }
 
 // SetData Set data for backtest
-func (b *Backtest) SetData(data *dataloader.Data) {
-	b.data = data
+func (b *Backtest) SetDatas(datas []*dataloader.Data) {
+	b.datas = datas
 }
 
 // GetTime get current time
 func (b *Backtest) GetTime() time.Time {
-	if b.data == nil {
+	if len(b.datas) == 0 {
 		return time.Now()
 	}
-	return b.data.GetOrderBook().Time
+	return time.Unix(0, b.currentTimeNS)
 }
 
 // Run Run backtest
@@ -151,20 +158,31 @@ func (b *Backtest) Run() {
 
 	b.startedAt = time.Now()
 
-	b.data.Reset(b.start, b.end)
+	for _, data := range b.datas {
+		data.Reset(b.start, b.end)
+	}
+
+	// 初始化数据
+	b.sortedDatas = make([]*DataState, 0, len(b.datas))
+	for i := 0; i < len(b.datas); i++ {
+		b.sortedDatas = append(b.sortedDatas, &DataState{
+			Time:  0,
+			Index: i,
+		})
+	}
+
+	// 设置数据缓存
+	b.setCacheData()
 
 	// 初始净值
-	if ob := b.data.GetOrderBook(); ob != nil {
-		item := &LogItem{
-			Time:    ob.Time,
-			RawTime: ob.Time,
-			Ask:     ob.AskPrice(),
-			Bid:     ob.BidPrice(),
-			Stats:   nil,
-		}
-		b.fetchItemStats(item)
-		b.logs = append(b.logs, item)
+	item := &LogItem{
+		Time:    time.Unix(0, b.currentTimeNS),
+		RawTime: time.Unix(0, b.currentTimeNS),
+		Prices:  b.getPrices(),
+		Stats:   nil,
 	}
+	b.fetchItemStats(item)
+	b.logs = append(b.logs, item)
 
 	// Init
 	b.strategy.OnInit()
@@ -173,7 +191,7 @@ func (b *Backtest) Run() {
 		b.strategy.OnTick()
 		b.runEventLoopOnce()
 		b.addItemStats()
-		if !b.data.Next() {
+		if !b.next() {
 			break
 		}
 	}
@@ -191,15 +209,72 @@ func (b *Backtest) Run() {
 	b.endedAt = time.Now()
 }
 
+func (b *Backtest) next() bool {
+	if b.currentTimeNS == 0 {
+		return b.nextInternal()
+	}
+
+	for _, data := range b.sortedDatas {
+		if b.currentTimeNS < data.Time {
+			b.currentTimeNS = data.Time
+			return true
+		}
+	}
+
+	return b.nextInternal()
+}
+
+func (b *Backtest) nextInternal() bool {
+	if len(b.datas) == 1 {
+		ret := b.datas[0].Next()
+		if ret {
+			b.currentTimeNS = b.datas[0].GetOrderBook().Time.UnixNano()
+		}
+		return ret
+	}
+
+	for _, data := range b.datas {
+		if !data.Next() {
+			return false
+		}
+	}
+
+	b.setCacheData()
+
+	return true
+}
+
+func (b *Backtest) setCacheData() {
+	// 数据对齐，提前排序
+	n := len(b.datas)
+
+	for i := 0; i < n; i++ {
+		b.sortedDatas[i].Time = b.datas[i].GetOrderBook().Time.UnixNano()
+		b.sortedDatas[i].Index = i
+	}
+
+	sort.Slice(b.sortedDatas, func(i, j int) bool {
+		return b.sortedDatas[i].Time < b.sortedDatas[j].Time
+	})
+
+	b.currentTimeNS = b.sortedDatas[0].Time
+}
+
+func (b *Backtest) getPrices() (result []float64) {
+	for _, data := range b.datas {
+		result = append(result, data.GetOrderBook().Price())
+	}
+	return
+}
+
 func (b *Backtest) runEventLoopOnce() {
 	for _, exchange := range b.exchanges {
 		exchange.RunEventLoopOnce()
 	}
 }
 
 func (b *Backtest) addItemStats() {
-	ob := b.data.GetOrderBook()
-	tm := ob.Time
+	tm := b.GetTime()
 	update := false
 	timestamp := time.Date(tm.Year(), tm.Month(), tm.Day(), tm.Hour(), tm.Minute()+1, 0, 0, time.UTC)
 	var lastItem *LogItem
@@ -214,17 +289,15 @@ func (b *Backtest) addItemStats() {
 	var item *LogItem
 	if update {
 		item = lastItem
-		item.RawTime = ob.Time
-		item.Ask = ob.AskPrice()
-		item.Bid = ob.BidPrice()
+		item.RawTime = tm
+		item.Prices = b.getPrices()
 		item.Stats = nil
 		b.fetchItemStats(item)
 	} else {
 		item = &LogItem{
 			Time:    timestamp,
-			RawTime: ob.Time,
-			Ask:     ob.AskPrice(),
-			Bid:     ob.BidPrice(),
+			RawTime: tm,
+			Prices:  b.getPrices(),
 			Stats:   nil,
 		}
 		b.fetchItemStats(item)
@@ -267,8 +340,8 @@ func (b *Backtest) ComputeStats() (result *Stats) {
 	result.End = logs[n-1].Time
 	result.Duration = result.End.Sub(result.Start)
 	result.RunDuration = b.endedAt.Sub(b.startedAt)
-	result.EntryPrice = logs[0].Price()
-	result.ExitPrice = logs[n-1].Price()
+	result.EntryPrice = logs[0].Prices[0]
+	result.ExitPrice = logs[n-1].Prices[0]
 	result.EntryEquity = logs[0].TotalEquity()
 	result.ExitEquity = logs[n-1].TotalEquity()
 	result.BaHReturn = (result.ExitPrice - result.EntryPrice) / result.EntryPrice * result.EntryEquity
@@ -521,7 +594,7 @@ func (b *Backtest) Plot() {
 
 	for _, v := range b.logs {
 		plotData.NameItems = append(plotData.NameItems, v.Time.Format(SimpleDateTimeFormat))
-		plotData.Prices = append(plotData.Prices, v.Price())
+		plotData.Prices = append(plotData.Prices, v.Prices[0])
 		plotData.Equities = append(plotData.Equities, v.TotalEquity())
 	}
 
@@ -576,7 +649,7 @@ func (b *Backtest) PlotOld() {
 
 	for _, v := range b.logs {
 		nameItems = append(nameItems, v.Time.Format(SimpleDateTimeFormat))
-		prices = append(prices, v.Price())
+		prices = append(prices, v.Prices[0])
 		equities = append(equities, v.TotalEquity())
 	}
 

dataloader/csv_dataloader.go

@@ -23,7 +23,7 @@ func (l *CsvDataLoader) Setup(start time.Time, end time.Time) error {
 	return nil
 }
 
-func (l *CsvDataLoader) ReadData() (result []*OrderBook) {
+func (l *CsvDataLoader) ReadOrderBooks() (result []*OrderBook) {
 	if !l.hasMoreData {
 		return nil
 	}
@@ -54,6 +54,10 @@ func (l *CsvDataLoader) ReadData() (result []*OrderBook) {
 	return
 }
 
+func (l *CsvDataLoader) ReadRecords(limit int) []*Record {
+	return nil
+}
+
 func (l *CsvDataLoader) HasMoreData() bool {
 	return l.hasMoreData
 }

dataloader/data.go

@@ -37,6 +37,10 @@ func (d *Data) GetOrderBook() *OrderBook {
 	return d.data[d.index]
 }
 
+func (d *Data) GetRecords(size int) []*Record {
+	return nil
+}
+
 func (d *Data) Next() bool {
 	if d.index < d.maxIndex {
 		d.index++
@@ -54,7 +58,7 @@ func (d *Data) readMore() int {
 	if !d.dataLoader.HasMoreData() {
 		return 0
 	}
-	data := d.dataLoader.ReadData()
+	data := d.dataLoader.ReadOrderBooks()
 	if len(data) == 0 {
 		return 0
 	}

dataloader/dataloader.go

@@ -7,6 +7,7 @@ import (
 
 type DataLoader interface {
 	Setup(start time.Time, end time.Time) error
-	ReadData() []*OrderBook
+	ReadOrderBooks() []*OrderBook
+	ReadRecords(limit int) []*Record
 	HasMoreData() bool
 }

dataloader/mongodb_dataloader.go

@@ -45,7 +45,7 @@ func (l *MongoDBDataLoader) Setup(start time.Time, end time.Time) error {
 	return nil
 }
 
-func (l *MongoDBDataLoader) ReadData() (result []*OrderBook) {
+func (l *MongoDBDataLoader) ReadOrderBooks() (result []*OrderBook) {
 	if !l.hasMoreData {
 		return nil
 	}
@@ -65,6 +65,10 @@ func (l *MongoDBDataLoader) ReadData() (result []*OrderBook) {
 	return l.convert(batch...)
 }
 
+func (l *MongoDBDataLoader) ReadRecords(limit int) []*Record {
+	return nil
+}
+
 func (l *MongoDBDataLoader) convert(r ...bson.Raw) (result []*OrderBook) {
 	n := len(r)
 	var wg sync.WaitGroup

dataloader/mongodb_dataloader_test.go

@@ -14,7 +14,7 @@ func TestMongoDBDataLoader_Setup(t *testing.T) {
 
 	t.Log("------------")
 	s := time.Now()
-	result := loader.ReadData()
+	result := loader.ReadOrderBooks()
 	//for _, v := range result {
 	//	t.Logf("%v", v.Time)
 	//}
@@ -23,7 +23,7 @@ func TestMongoDBDataLoader_Setup(t *testing.T) {
 
 	t.Log("------------")
 	s = time.Now()
-	result = loader.ReadData()
+	result = loader.ReadOrderBooks()
 	//for _, v := range result {
 	//	t.Logf("%v", v.Time)
 	//}

examples/backtest/main.go

@@ -52,16 +52,17 @@ func main() {
 	start, _ := time.Parse("2006-01-02 15:04:05", "2019-10-01 00:00:00")
 	end, _ := time.Parse("2006-01-02 15:04:05", "2019-10-02 00:00:00")
 
-	//data := dataloader.NewCsvData("../../data-samples/deribit/deribit_BTC-PERPETUAL_and_futures_tick_by_tick_book_snapshots_10_levels_2019-10-01_2019-11-01.csv")
-
+	var datas []*dataloader.Data
 	var exchanges []ExchangeSim
 	for i := 0; i < 2; i++ {
+		datas = append(datas, data)
 		ex := exsim.NewExSim(data, 5.0, -0.00025, 0.00075, false)
 		exchanges = append(exchanges, ex)
 	}
+
 	s := &BasicStrategy{}
 	outputDir := "./output"
-	bt := backtest.NewBacktest(data,
+	bt := backtest.NewBacktest(datas,
 		"BTC",
 		start,
 		end,

exchanges/exsim/exsim.go

@@ -3,6 +3,7 @@ package exsim
 import (
 	"errors"
 	"fmt"
+	"github.com/chuckpreslar/emission"
 	. "github.com/coinrust/crex"
 	"github.com/coinrust/crex/dataloader"
 	"log"
@@ -36,7 +37,8 @@ type ExSim struct {
 	historyOrders  map[string]*Order     // History orders
 	positions      map[string]*Positions // Position key: symbol
 
-	eLog ExchangeLogger
+	emitter *emission.Emitter
+	eLog    ExchangeLogger
 }
 
 func (b *ExSim) GetName() (name string) {
@@ -164,7 +166,12 @@ func (b *ExSim) PlaceOrder(symbol string, direction Direction, orderType OrderTy
 
 	b.orders[id] = order
 	result = order
+
 	b.logOrderInfo("Place order", SimEventOrder, order)
+
+	var orders = []*Order{order}
+	b.emitter.Emit(WSEventOrder, orders)
+
 	return
 }
 
@@ -691,6 +698,8 @@ func (b *ExSim) RunEventLoopOnce() (err error) {
 		match, err = b.matchOrder(order, false)
 		if match {
 			b.logOrderInfo("Match order", SimEventDeal, order)
+			var orders = []*Order{order}
+			b.emitter.Emit(WSEventOrder, orders)
 		}
 	}
 	return
@@ -727,5 +736,6 @@ func NewExSim(data *dataloader.Data, cash float64, makerFeeRate float64, takerFe
 		openOrders:     make(map[string]*Order),
 		historyOrders:  make(map[string]*Order),
 		positions:      make(map[string]*Positions),
+		emitter:        emission.NewEmitter(),
 	}
 }