wip csv source stream

Author: go-dockly

pkg/datasource/csvsource/csv_tick_converter.go

@@ -1,7 +1,6 @@
 package csvsource
 
 import (
-	"encoding/csv"
 	"time"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
@@ -11,14 +10,12 @@ import (
 type ICSVTickConverter interface {
 	LatestKLine() (k *types.KLine)
 	GetKLineResult() []types.KLine
-	CsvTickToKLine(tick *CsvTick, interval types.Interval)
+	CsvTickToKLine(tick *CsvTick, interval types.Interval) (closesKLine bool)
 }
 
 // CSVTickConverter takes a tick and internally converts it to a KLine slice
 type CSVTickConverter struct {
-	csv     *csv.Reader
-	decoder CSVTickDecoder
-	klines  []types.KLine
+	klines []types.KLine
 }
 
 func NewCSVTickConverter() ICSVTickConverter {
@@ -41,7 +38,7 @@ func (c *CSVTickConverter) GetKLineResult() []types.KLine {
 }
 
 // Convert ticks to KLine with interval
-func (c *CSVTickConverter) CsvTickToKLine(tick *CsvTick, interval types.Interval) {
+func (c *CSVTickConverter) CsvTickToKLine(tick *CsvTick, interval types.Interval) (closesKLine bool) {
 	var (
 		currentCandle = types.KLine{}
 		high          = fixedpoint.Zero
@@ -50,15 +47,25 @@ func (c *CSVTickConverter) CsvTickToKLine(tick *CsvTick, interval types.Interval
 	isOpen, t := c.detCandleStart(tick.Timestamp.Time(), interval)
 
 	if isOpen {
+		k := c.LatestKLine()
 		c.klines = append(c.klines, types.KLine{
-			StartTime: types.NewTimeFromUnix(t.Unix(), 0),
-			EndTime:   types.NewTimeFromUnix(t.Add(interval.Duration()).Unix(), 0),
-			Open:      tick.Price,
-			High:      tick.Price,
-			Low:       tick.Price,
-			Close:     tick.Price,
-			Volume:    tick.HomeNotional,
+			Exchange:    tick.Exchange,
+			Symbol:      tick.Symbol,
+			Interval:    interval,
+			StartTime:   types.NewTimeFromUnix(t.Unix(), 0),
+			EndTime:     types.NewTimeFromUnix(t.Add(interval.Duration()).Unix(), 0),
+			Open:        tick.Price,
+			High:        tick.Price,
+			Low:         tick.Price,
+			Close:       tick.Price,
+			Volume:      tick.HomeNotional,
+			QuoteVolume: tick.ForeignNotional,
+			Closed:      false,
 		})
+		if k != nil {
+			k.Closed = true // k is pointer
+			closesKLine = true
+		}
 		return
 	}
 
@@ -77,14 +84,21 @@ func (c *CSVTickConverter) CsvTickToKLine(tick *CsvTick, interval types.Interval
 	}
 
 	c.klines[len(c.klines)-1] = types.KLine{
-		StartTime: currentCandle.StartTime,
-		EndTime:   currentCandle.EndTime,
-		Open:      currentCandle.Open,
-		High:      high,
-		Low:       low,
-		Close:     tick.Price,
-		Volume:    currentCandle.Volume.Add(tick.HomeNotional),
+		StartTime:   currentCandle.StartTime,
+		EndTime:     currentCandle.EndTime,
+		Exchange:    tick.Exchange,
+		Symbol:      tick.Symbol,
+		Interval:    interval,
+		Open:        currentCandle.Open,
+		High:        high,
+		Low:         low,
+		Close:       tick.Price,
+		Volume:      currentCandle.Volume.Add(tick.HomeNotional),
+		QuoteVolume: currentCandle.QuoteVolume.Add(tick.ForeignNotional),
+		Closed:      false,
 	}
+
+	return
 }
 
 func (c *CSVTickConverter) detCandleStart(ts time.Time, interval types.Interval) (isOpen bool, t time.Time) {

pkg/datasource/csvsource/csv_tick_decoder.go

@@ -33,7 +33,6 @@ func BinanceCSVTickDecoder(row []string, _ int) (*CsvTick, error) {
 	qty := fixedpoint.MustNewFromString(row[2])
 	baseQty := fixedpoint.MustNewFromString(row[2])
 	price := fixedpoint.MustNewFromString(row[1])
-	// isBuyerMaker=false trade will qualify as BUY.
 	id, err := strconv.ParseUint(row[0], 10, 64)
 	if err != nil {
 		return nil, err
@@ -42,6 +41,7 @@ func BinanceCSVTickDecoder(row []string, _ int) (*CsvTick, error) {
 	if err != nil {
 		return nil, err
 	}
+	// isBuyerMaker=false trade will qualify as BUY.
 	side := types.SideTypeBuy
 	if isBuyerMaker {
 		side = types.SideTypeSell

pkg/datasource/csvsource/stream.go

@@ -5,6 +5,7 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
@@ -22,6 +23,7 @@ type Stream struct {
 
 type CsvStreamConfig struct {
 	Interval     types.Interval
+	RateLimit    time.Duration    `json:"csvPath"`
 	CsvPath      string           `json:"csvPath"`
 	Symbol       string           `json:"symbol"`
 	BaseCoin     string           `json:"baseCoin"`
@@ -45,8 +47,11 @@ func NewStream(cfg *CsvStreamConfig) *Stream {
 	return stream
 }
 
-func (s *Stream) simulateEvents() error {
+func (s *Stream) Simulate() error {
 	var i int
+	converter := NewCSVTickConverter()
+
+	// iterate equity series at csv path and stream
 	err := filepath.WalkDir(s.config.CsvPath, func(path string, d fs.DirEntry, err error) error {
 		if err != nil {
 			return err
@@ -63,7 +68,6 @@ func (s *Stream) simulateEvents() error {
 		}
 		//nolint:errcheck // Read ops only so safe to ignore err return
 		defer file.Close()
-		converter := NewCSVTickConverter()
 		reader := NewBybitCSVTickReader(csv.NewReader(file))
 		tick, err := reader.Read(i)
 		if err != nil {
@@ -76,14 +80,17 @@ func (s *Stream) simulateEvents() error {
 		}
 		s.StandardStream.EmitMarketTrade(*trade)
 
-		converter.CsvTickToKLine(tick, s.config.Interval)
 		kline := converter.LatestKLine()
-		if kline.Closed {
+		closesKline := converter.CsvTickToKLine(tick, s.config.Interval)
+		if closesKline {
 			s.StandardStream.EmitKLineClosed(*kline)
 		} else {
+			kline = converter.LatestKLine()
 			s.StandardStream.EmitKLine(*kline)
 		}
-
+		// allow for execution time of indicators and strategy
+		time.Sleep(s.config.RateLimit) // Max execution time for tradingview strategy
+		// to optimize exec time consider callback channel once a strategy has finished running another tick is emitted
 		return nil
 	})
 	if err != nil {