Add trend indicator DPO

README.md

@@ -34,6 +34,7 @@ The following list of indicators are currently supported by this package:
 -	[Community Channel Index (CCI)](trend/README.md#type-cci)
 -   [Envelope](trend/README.md#type-envelope)
 -	[Hull Moving Average (HMA)](trend/README.md#type-hma)
+-   [Detrended Price Oscillator (DPO)](trend/README.md#type-dpo)
 -	[Double Exponential Moving Average (DEMA)](trend/README.md#type-dema)
 -	[Exponential Moving Average (EMA)](trend/README.md#type-ema)
 -	[Kaufman's Adaptive Moving Average (KAMA)](trend/README.md#type-kama)

helper/README.md

@@ -79,6 +79,7 @@ The information provided on this project is strictly for informational purposes
 - [func Sign\[T Number\]\(c \<\-chan T\) \<\-chan T](<#Sign>)
 - [func Since\[T comparable, R Number\]\(c \<\-chan T\) \<\-chan R](<#Since>)
 - [func Skip\[T any\]\(c \<\-chan T, count int\) \<\-chan T](<#Skip>)
+- [func SkipLast\[T any\]\(c \<\-chan T, count int\) \<\-chan T](<#SkipLast>)
 - [func SliceToChan\[T any\]\(slice \[\]T\) \<\-chan T](<#SliceToChan>)
 - [func Sqrt\[T Number\]\(c \<\-chan T\) \<\-chan T](<#Sqrt>)
 - [func Subtract\[T Number\]\(ac, bc \<\-chan T\) \<\-chan T](<#Subtract>)
@@ -949,6 +950,23 @@ actual := helper.Skip(c, 2)
 fmt.Println(helper.ChanToSlice(actual)) // [6, 8]
 ```
 
+<a name="SkipLast"></a>
+## func [SkipLast](<https://github.com/cinar/indicator/blob/master/helper/skip_last.go#L11>)
+
+```go
+func SkipLast[T any](c <-chan T, count int) <-chan T
+```
+
+SkipLast skips the specified number of elements from the end of the given channel.
+
+Example:
+
+```
+c := helper.SliceToChan([]int{2, 4, 6, 8})
+actual := helper.SkipLast(c, 2)
+fmt.Println(helper.ChanToSlice(actual)) // [2, 4]
+```
+
 <a name="SliceToChan"></a>
 ## func [SliceToChan](<https://github.com/cinar/indicator/blob/master/helper/slice_to_chan.go#L17>)
 

helper/skip_last.go

@@ -0,0 +1,32 @@
+package helper
+
+// SkipLast skips the specified number of elements
+// from the end of the given channel.
+//
+// Example:
+//
+//	c := helper.SliceToChan([]int{2, 4, 6, 8})
+//	actual := helper.SkipLast(c, 2)
+//	fmt.Println(helper.ChanToSlice(actual)) // [2, 4]
+func SkipLast[T any](c <-chan T, count int) <-chan T {
+	result := make(chan T, cap(c))
+
+	go func() {
+		defer close(result)
+
+		// Buffer to hold the last "count" elements
+		buf := make([]T, 0, count)
+
+		for v := range c {
+			buf = append(buf, v)
+			if len(buf) > count {
+				// send the oldest value
+				result <- buf[0]
+				buf = buf[1:]
+			}
+		}
+		// drop the last `count` elements automatically
+	}()
+
+	return result
+}

helper/skip_last_test.go

@@ -0,0 +1,23 @@
+// Copyright (c) 2021-2024 Onur Cinar.
+// The source code is provided under GNU AGPLv3 License.
+// https://github.com/cinar/indicator
+
+package helper_test
+
+import (
+    "testing"
+
+    "github.com/cinar/indicator/v2/helper"
+)
+
+func TestSkipLast(t *testing.T) {
+    input := helper.SliceToChan([]int{2, 4, 6, 8})
+    expected := helper.SliceToChan([]int{2, 4})
+
+    actual := helper.SkipLast(input, 2)
+
+    err := helper.CheckEquals(actual, expected)
+    if err != nil {
+        t.Fatal(err)
+    }
+}

trend/README.md

@@ -43,6 +43,12 @@ The information provided on this project is strictly for informational purposes
   - [func NewDema\[T helper.Number\]\(\) \*Dema\[T\]](<#NewDema>)
   - [func \(d \*Dema\[T\]\) Compute\(c \<\-chan T\) \<\-chan T](<#Dema[T].Compute>)
   - [func \(d \*Dema\[T\]\) IdlePeriod\(\) int](<#Dema[T].IdlePeriod>)
+- [type Dpo](<#Dpo>)
+  - [func NewDpo\[T helper.Float\]\(\) \*Dpo\[T\]](<#NewDpo>)
+  - [func NewDpoWithPeriod\[T helper.Float\]\(period int\) \*Dpo\[T\]](<#NewDpoWithPeriod>)
+  - [func \(d \*Dpo\[T\]\) Compute\(closing \<\-chan T\) \<\-chan T](<#Dpo[T].Compute>)
+  - [func \(d \*Dpo\[T\]\) IdlePeriod\(\) int](<#Dpo[T].IdlePeriod>)
+  - [func \(d \*Dpo\[T\]\) String\(\) string](<#Dpo[T].String>)
 - [type Ema](<#Ema>)
   - [func NewEma\[T helper.Number\]\(\) \*Ema\[T\]](<#NewEma>)
   - [func NewEmaWithPeriod\[T helper.Number\]\(period int\) \*Ema\[T\]](<#NewEmaWithPeriod>)
@@ -299,6 +305,12 @@ const (
 )
 ```
 
+<a name="DefaultDpoPeriod"></a>DefaultDpoPeriod is the default period for DPO calculation.
+
+```go
+const DefaultDpoPeriod = 20
+```
+
 <a name="DefaultRmaPeriod"></a>
 
 ```go
@@ -612,6 +624,73 @@ func (d *Dema[T]) IdlePeriod() int
 
 IdlePeriod is the initial period that DEMA won't yield any results.
 
+<a name="Dpo"></a>
+## type [Dpo](<https://github.com/cinar/indicator/blob/master/trend/dpo.go#L27-L31>)
+
+Dpo computes the Detrended Price Oscillator. Formula \(common approximation\): Let k = floor\(period/2\) \+ 1. For time index t \>= period\-1\+k:
+
+```
+DPO[t] = Price[t] - SMA[t - k]
+```
+
+Example:
+
+```
+dpo := trend.NewDpoWithPeriod[float64](20)
+out := dpo.Compute(c)
+```
+
+```go
+type Dpo[T helper.Float] struct {
+    // contains filtered or unexported fields
+}
+```
+
+<a name="NewDpo"></a>
+### func [NewDpo](<https://github.com/cinar/indicator/blob/master/trend/dpo.go#L34>)
+
+```go
+func NewDpo[T helper.Float]() *Dpo[T]
+```
+
+NewDpo creates a new DPO instance with default parameters.
+
+<a name="NewDpoWithPeriod"></a>
+### func [NewDpoWithPeriod](<https://github.com/cinar/indicator/blob/master/trend/dpo.go#L42>)
+
+```go
+func NewDpoWithPeriod[T helper.Float](period int) *Dpo[T]
+```
+
+NewDpoWithPeriod initializes a new DPO instance with the given period. Periods \<= 1 are clamped to DefaultDpoPeriod.
+
+<a name="Dpo[T].Compute"></a>
+### func \(\*Dpo\[T\]\) [Compute](<https://github.com/cinar/indicator/blob/master/trend/dpo.go#L53>)
+
+```go
+func (d *Dpo[T]) Compute(closing <-chan T) <-chan T
+```
+
+Compute calculates the DPO indicator over the input price channel.
+
+<a name="Dpo[T].IdlePeriod"></a>
+### func \(\*Dpo\[T\]\) [IdlePeriod](<https://github.com/cinar/indicator/blob/master/trend/dpo.go#L73>)
+
+```go
+func (d *Dpo[T]) IdlePeriod() int
+```
+
+IdlePeriod returns the number of leading samples to discard before the first DPO value is available.
+
+<a name="Dpo[T].String"></a>
+### func \(\*Dpo\[T\]\) [String](<https://github.com/cinar/indicator/blob/master/trend/dpo.go#L78>)
+
+```go
+func (d *Dpo[T]) String() string
+```
+
+String is the string representation of the DPO.
+
 <a name="Ema"></a>
 ## type [Ema](<https://github.com/cinar/indicator/blob/master/trend/ema.go#L29-L35>)
 

trend/dpo.go

@@ -0,0 +1,80 @@
+// Copyright (c) 2021-2024 Onur Cinar.
+// The source code is provided under GNU AGPLv3 License.
+// https://github.com/cinar/indicator
+
+package trend
+
+import (
+	"fmt"
+
+	"github.com/cinar/indicator/v2/helper"
+)
+
+// DefaultDpoPeriod is the default period for DPO calculation.
+const DefaultDpoPeriod = 20
+
+// Dpo computes the Detrended Price Oscillator.
+// Formula (common approximation):
+// Let k = floor(period/2) + 1.
+// For time index t >= period-1+k:
+//
+//	DPO[t] = Price[t] - SMA[t - k]
+//
+// Example:
+//
+//	dpo := trend.NewDpoWithPeriod[float64](20)
+//	out := dpo.Compute(c)
+type Dpo[T helper.Float] struct {
+	// Period is the SMA window length. Typical default is 20.
+	// Note: values <= 1 are considered invalid and will be replaced with DefaultDpoPeriod by constructors.
+	period int
+}
+
+// NewDpo creates a new DPO instance with default parameters.
+func NewDpo[T helper.Float]() *Dpo[T] {
+	return &Dpo[T]{
+		period: DefaultDpoPeriod,
+	}
+}
+
+// NewDpoWithPeriod initializes a new DPO instance with the given period.
+// Periods <= 1 are clamped to DefaultDpoPeriod.
+func NewDpoWithPeriod[T helper.Float](period int) *Dpo[T] {
+	if period <= 1 {
+		period = DefaultDpoPeriod
+	}
+
+	return &Dpo[T]{
+		period: period,
+	}
+}
+
+// Compute calculates the DPO indicator over the input price channel.
+func (d *Dpo[T]) Compute(closing <-chan T) <-chan T {
+	k := d.period/2 + 1
+	dup := helper.Duplicate(closing, 2)
+
+	// compute SMA on the first duplicated stream
+	sma := NewSma[T]()
+	sma.Period = d.period
+	smaOut := sma.Compute(dup[0])
+
+	// align the original price stream and the SMA stream according to DPO formula
+	skippedClosing := helper.Skip(dup[1], d.IdlePeriod())
+	smaDelayed := helper.SkipLast(smaOut, k)
+
+	// DPO = Price - shifted SMA
+	return helper.Operate(skippedClosing, smaDelayed, func(price, shiftedSma T) T {
+		return price - shiftedSma
+	})
+}
+
+// IdlePeriod returns the number of leading samples to discard before the first DPO value is available.
+func (d *Dpo[T]) IdlePeriod() int {
+	return (d.period - 1) + (d.period/2 + 1)
+}
+
+// String is the string representation of the DPO.
+func (d *Dpo[T]) String() string {
+	return fmt.Sprintf("DPO(%d)", d.period)
+}

trend/dpo_test.go

@@ -0,0 +1,42 @@
+package trend_test
+
+import (
+	"testing"
+
+	"github.com/cinar/indicator/v2/helper"
+	"github.com/cinar/indicator/v2/trend"
+)
+
+func TestDpo(t *testing.T) {
+	type DpoData struct {
+		Close float64
+		Dpo   float64
+	}
+
+	input, err := helper.ReadFromCsvFile[DpoData]("testdata/dpo.csv")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	inputs := helper.Duplicate(input, 2)
+	closing := helper.Map(inputs[0], func(d *DpoData) float64 { return d.Close })
+	expected := helper.Map(inputs[1], func(d *DpoData) float64 { return d.Dpo })
+
+	dpo := trend.NewDpo[float64]()
+	actual := helper.RoundDigits(dpo.Compute(closing), 2)
+	expected = helper.Skip(expected, dpo.IdlePeriod())
+
+	err = helper.CheckEquals(actual, expected)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestDpoString(t *testing.T) {
+	expected := "DPO(10)"
+	actual := trend.NewDpoWithPeriod[float64](10).String()
+
+	if actual != expected {
+		t.Fatalf("actual %v expected %v", actual, expected)
+	}
+}