From ca78a3379a1518c9fddb0cb1fceb14c716e84848 Mon Sep 17 00:00:00 2001
From: c9s <yoanlin93@gmail.com>
Date: Sun, 4 Jun 2023 14:47:29 +0800
Subject: [PATCH] indicator: add cross stream

---
 pkg/indicator/float64series.go  | 71 +++++++++++++++++++++++++++++++++
 pkg/indicator/float64updater.go | 70 --------------------------------
 pkg/indicator/v2_cross.go       | 59 +++++++++++++++++++++++++++
 3 files changed, 130 insertions(+), 70 deletions(-)
 create mode 100644 pkg/indicator/float64series.go
 create mode 100644 pkg/indicator/v2_cross.go

diff --git a/pkg/indicator/float64series.go b/pkg/indicator/float64series.go
new file mode 100644
index 000000000..821c17666
--- /dev/null
+++ b/pkg/indicator/float64series.go
@@ -0,0 +1,71 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type Float64Series struct {
+	types.SeriesBase
+	Float64Updater
+	slice floats.Slice
+}
+
+func NewFloat64Series(v ...float64) Float64Series {
+	s := Float64Series{}
+	s.slice = v
+	s.SeriesBase.Series = s.slice
+	return s
+}
+
+func (f *Float64Series) Last(i int) float64 {
+	return f.slice.Last(i)
+}
+
+func (f *Float64Series) Index(i int) float64 {
+	return f.Last(i)
+}
+
+func (f *Float64Series) Length() int {
+	return len(f.slice)
+}
+
+func (f *Float64Series) Slice() floats.Slice {
+	return f.slice
+}
+
+func (f *Float64Series) PushAndEmit(x float64) {
+	f.slice.Push(x)
+	f.EmitUpdate(x)
+}
+
+func (f *Float64Series) Subscribe(source Float64Source, c func(x float64)) {
+	if sub, ok := source.(Float64Subscription); ok {
+		sub.AddSubscriber(c)
+	} else {
+		source.OnUpdate(c)
+	}
+}
+
+// Bind binds the source event to the target (Float64Calculator)
+// A Float64Calculator should be able to calculate the float64 result from a single float64 argument input
+func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
+	var c func(x float64)
+
+	// optimize the truncation check
+	trc, canTruncate := target.(Float64Truncator)
+	if canTruncate {
+		c = func(x float64) {
+			y := target.Calculate(x)
+			target.PushAndEmit(y)
+			trc.Truncate()
+		}
+	} else {
+		c = func(x float64) {
+			y := target.Calculate(x)
+			target.PushAndEmit(y)
+		}
+	}
+
+	f.Subscribe(source, c)
+}
diff --git a/pkg/indicator/float64updater.go b/pkg/indicator/float64updater.go
index 7afdef8aa..a9743538e 100644
--- a/pkg/indicator/float64updater.go
+++ b/pkg/indicator/float64updater.go
@@ -1,76 +1,6 @@
 package indicator
 
-import (
-	"github.com/c9s/bbgo/pkg/datatype/floats"
-	"github.com/c9s/bbgo/pkg/types"
-)
-
 //go:generate callbackgen -type Float64Updater
 type Float64Updater struct {
 	updateCallbacks []func(v float64)
 }
-
-type Float64Series struct {
-	types.SeriesBase
-	Float64Updater
-	slice floats.Slice
-}
-
-func NewFloat64Series(v ...float64) Float64Series {
-	s := Float64Series{}
-	s.slice = v
-	s.SeriesBase.Series = s.slice
-	return s
-}
-
-func (f *Float64Series) Last(i int) float64 {
-	return f.slice.Last(i)
-}
-
-func (f *Float64Series) Index(i int) float64 {
-	return f.Last(i)
-}
-
-func (f *Float64Series) Length() int {
-	return len(f.slice)
-}
-
-func (f *Float64Series) Slice() floats.Slice {
-	return f.slice
-}
-
-func (f *Float64Series) PushAndEmit(x float64) {
-	f.slice.Push(x)
-	f.EmitUpdate(x)
-}
-
-func (f *Float64Series) Subscribe(source Float64Source, c func(x float64)) {
-	if sub, ok := source.(Float64Subscription); ok {
-		sub.AddSubscriber(c)
-	} else {
-		source.OnUpdate(c)
-	}
-}
-
-// Bind binds the source event to the target (Float64Calculator)
-// A Float64Calculator should be able to calculate the float64 result from a single float64 argument input
-func (f *Float64Series) Bind(source Float64Source, target Float64Calculator) {
-	var c func(x float64)
-
-	// optimize the truncation check
-	trc, canTruncate := target.(Float64Truncator)
-	if canTruncate {
-		c = func(x float64) {
-			y := target.Calculate(x)
-			target.PushAndEmit(y)
-			trc.Truncate()
-		}
-	} else {
-		c = func(x float64) {
-			y := target.Calculate(x)
-			target.PushAndEmit(y)
-		}
-	}
-
-	f.Subscribe(source, c)
-}
diff --git a/pkg/indicator/v2_cross.go b/pkg/indicator/v2_cross.go
new file mode 100644
index 000000000..084130fdb
--- /dev/null
+++ b/pkg/indicator/v2_cross.go
@@ -0,0 +1,59 @@
+package indicator
+
+import (
+	"github.com/c9s/bbgo/pkg/datatype/floats"
+)
+
+type CrossType float64
+
+const (
+	CrossOver  CrossType = 1.0
+	CrossUnder CrossType = -1.0
+)
+
+// CrossStream subscribes 2 upstreams, and calculate the cross signal
+type CrossStream struct {
+	Float64Series
+
+	a, b floats.Slice
+}
+
+// Cross creates the CrossStream object:
+//
+// cross := Cross(fastEWMA, slowEWMA)
+func Cross(a, b Float64Source) *CrossStream {
+	s := &CrossStream{
+		Float64Series: NewFloat64Series(),
+	}
+	a.OnUpdate(func(v float64) {
+		s.a.Push(v)
+		s.calculate()
+	})
+	b.OnUpdate(func(v float64) {
+		s.b.Push(v)
+		s.calculate()
+	})
+	return s
+}
+
+func (s *CrossStream) calculate() {
+	if s.a.Length() != s.b.Length() {
+		return
+	}
+
+	current := s.a.Last(0) - s.b.Last(0)
+	previous := s.a.Last(1) - s.b.Last(1)
+
+	if previous == 0.0 {
+		return
+	}
+
+	// cross over or cross under
+	if current*previous < 0 {
+		if current > 0 {
+			s.PushAndEmit(float64(CrossOver))
+		} else {
+			s.PushAndEmit(float64(CrossUnder))
+		}
+	}
+}