diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/aggregator.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/aggregator.go new file mode 100644 index 00000000..2da928b2 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/aggregator.go @@ -0,0 +1,70 @@ +// ©AngelaMos | 2026 +// aggregator.go + +package coinbase + +import "time" + +type Aggregator struct { + open map[string]MinuteBar +} + +func NewAggregator() *Aggregator { + return &Aggregator{open: make(map[string]MinuteBar)} +} + +func (a *Aggregator) Push(t Tick) (*MinuteBar, MinuteBar) { + minute := t.TS.UTC().Truncate(time.Minute) + cur, exists := a.open[t.Symbol] + + if !exists { + cur = MinuteBar{ + Symbol: t.Symbol, + Minute: minute, + Open: t.Price, + High: t.Price, + Low: t.Price, + Close: t.Price, + Volume: t.Volume24h, + } + a.open[t.Symbol] = cur + return nil, cur + } + + if minute.After(cur.Minute) { + closed := cur + cur = MinuteBar{ + Symbol: t.Symbol, + Minute: minute, + Open: t.Price, + High: t.Price, + Low: t.Price, + Close: t.Price, + Volume: t.Volume24h, + } + a.open[t.Symbol] = cur + return &closed, cur + } + + if t.Price.GreaterThan(cur.High) { + cur.High = t.Price + } + if t.Price.LessThan(cur.Low) { + cur.Low = t.Price + } + cur.Close = t.Price + if t.Volume24h.GreaterThan(cur.Volume) { + cur.Volume = t.Volume24h + } + a.open[t.Symbol] = cur + return nil, cur +} + +func (a *Aggregator) Flush() []MinuteBar { + out := make([]MinuteBar, 0, len(a.open)) + for _, bar := range a.open { + out = append(out, bar) + } + a.open = make(map[string]MinuteBar) + return out +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/aggregator_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/aggregator_test.go new file mode 100644 index 00000000..60d46741 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/aggregator_test.go @@ -0,0 +1,88 @@ +// ©AngelaMos | 2026 +// aggregator_test.go + +package coinbase_test + +import ( + "testing" + "time" + + "github.com/shopspring/decimal" + "github.com/stretchr/testify/require" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/coinbase" +) + +func TestAggregator_FirstTickOpensBar(t *testing.T) { + a := coinbase.NewAggregator() + ts := time.Date(2026, 5, 1, 12, 0, 30, 0, time.UTC) + closed, cur := a.Push(coinbase.Tick{ + Symbol: "BTC-USD", TS: ts, + Price: decimal.RequireFromString("42000"), + Volume24h: decimal.RequireFromString("100"), + }) + require.Nil(t, closed) + require.Equal(t, ts.Truncate(time.Minute), cur.Minute) + require.True(t, cur.Open.Equal(decimal.RequireFromString("42000"))) + require.True(t, cur.Close.Equal(decimal.RequireFromString("42000"))) +} + +func TestAggregator_FoldUpdatesHighLowClose(t *testing.T) { + a := coinbase.NewAggregator() + min := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) + a.Push(coinbase.Tick{Symbol: "BTC-USD", TS: min.Add(10 * time.Second), Price: decimal.RequireFromString("42000"), Volume24h: decimal.RequireFromString("100")}) + a.Push(coinbase.Tick{Symbol: "BTC-USD", TS: min.Add(20 * time.Second), Price: decimal.RequireFromString("42050"), Volume24h: decimal.RequireFromString("110")}) + closed, cur := a.Push(coinbase.Tick{ + Symbol: "BTC-USD", TS: min.Add(30 * time.Second), + Price: decimal.RequireFromString("41950"), Volume24h: decimal.RequireFromString("120"), + }) + require.Nil(t, closed) + require.True(t, cur.Open.Equal(decimal.RequireFromString("42000"))) + require.True(t, cur.High.Equal(decimal.RequireFromString("42050"))) + require.True(t, cur.Low.Equal(decimal.RequireFromString("41950"))) + require.True(t, cur.Close.Equal(decimal.RequireFromString("41950"))) + require.True(t, cur.Volume.Equal(decimal.RequireFromString("120"))) +} + +func TestAggregator_MinuteRolloverEmitsClosedBar(t *testing.T) { + a := coinbase.NewAggregator() + min1 := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) + min2 := min1.Add(time.Minute) + + a.Push(coinbase.Tick{Symbol: "BTC-USD", TS: min1.Add(10 * time.Second), Price: decimal.RequireFromString("42000"), Volume24h: decimal.RequireFromString("100")}) + a.Push(coinbase.Tick{Symbol: "BTC-USD", TS: min1.Add(50 * time.Second), Price: decimal.RequireFromString("42100"), Volume24h: decimal.RequireFromString("105")}) + + closed, cur := a.Push(coinbase.Tick{ + Symbol: "BTC-USD", TS: min2.Add(5 * time.Second), + Price: decimal.RequireFromString("42150"), Volume24h: decimal.RequireFromString("110"), + }) + + require.NotNil(t, closed) + require.Equal(t, min1, closed.Minute) + require.True(t, closed.Close.Equal(decimal.RequireFromString("42100"))) + + require.Equal(t, min2, cur.Minute) + require.True(t, cur.Open.Equal(decimal.RequireFromString("42150"))) +} + +func TestAggregator_BarsArePerProduct(t *testing.T) { + a := coinbase.NewAggregator() + min := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) + + _, btc := a.Push(coinbase.Tick{Symbol: "BTC-USD", TS: min.Add(10 * time.Second), Price: decimal.RequireFromString("42000"), Volume24h: decimal.RequireFromString("100")}) + _, eth := a.Push(coinbase.Tick{Symbol: "ETH-USD", TS: min.Add(11 * time.Second), Price: decimal.RequireFromString("2300"), Volume24h: decimal.RequireFromString("80")}) + + require.Equal(t, "BTC-USD", btc.Symbol) + require.Equal(t, "ETH-USD", eth.Symbol) + require.True(t, btc.Open.Equal(decimal.RequireFromString("42000"))) + require.True(t, eth.Open.Equal(decimal.RequireFromString("2300"))) +} + +func TestAggregator_FlushReturnsAllOpenBars(t *testing.T) { + a := coinbase.NewAggregator() + min := time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) + a.Push(coinbase.Tick{Symbol: "BTC-USD", TS: min.Add(10 * time.Second), Price: decimal.RequireFromString("42000"), Volume24h: decimal.RequireFromString("100")}) + a.Push(coinbase.Tick{Symbol: "ETH-USD", TS: min.Add(15 * time.Second), Price: decimal.RequireFromString("2300"), Volume24h: decimal.RequireFromString("80")}) + bars := a.Flush() + require.Len(t, bars, 2) +}