From ae5a83c2f8621d225ef90c0ff92bdfd9f5c75aeb Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sat, 2 May 2026 03:24:55 -0400 Subject: [PATCH] feat(monitor/collectors/coinbase): ticks + minute OHLC repo with 1h history reads Adds shopspring/decimal as a require (folded with the first consuming task per Plan 2 convention). cenkalti/backoff/v4 is also now consumed transitively; remains indirect until Task 5 (reconnect) imports it. --- .../backend/go.mod | 1 + .../backend/go.sum | 2 + .../internal/collectors/coinbase/repo.go | 101 ++++++++++++ .../internal/collectors/coinbase/repo_test.go | 148 ++++++++++++++++++ 4 files changed, 252 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod index 01de56e5..fbb03443 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod @@ -91,6 +91,7 @@ require ( github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect github.com/segmentio/asm v1.2.1 // indirect github.com/shirou/gopsutil/v4 v4.26.3 // indirect + github.com/shopspring/decimal v1.4.0 // indirect github.com/sirupsen/logrus v1.9.4 // indirect github.com/testcontainers/testcontainers-go v0.42.0 // indirect github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0 // indirect diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum index 3a4aa362..cafc7d97 100644 --- a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum @@ -179,6 +179,8 @@ github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= github.com/shirou/gopsutil/v4 v4.26.3 h1:2ESdQt90yU3oXF/CdOlRCJxrP+Am1aBYubTMTfxJ1qc= github.com/shirou/gopsutil/v4 v4.26.3/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/sony/gobreaker/v2 v2.4.0 h1:g2KJRW1Ubty3+ZOcSEUN7K+REQJdN6yo6XvaML+jptg= diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo.go new file mode 100644 index 00000000..6c91fcf3 --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo.go @@ -0,0 +1,101 @@ +// ©AngelaMos | 2026 +// repo.go + +package coinbase + +import ( + "context" + "fmt" + "time" + + "github.com/jmoiron/sqlx" + "github.com/shopspring/decimal" +) + +const ( + history1hLimit = 60 +) + +type Tick struct { + Symbol string `db:"symbol"` + TS time.Time `db:"ts"` + Price decimal.Decimal `db:"price"` + Volume24h decimal.Decimal `db:"volume_24h"` +} + +type MinuteBar struct { + Symbol string `db:"symbol"` + Minute time.Time `db:"minute"` + Open decimal.Decimal `db:"open"` + High decimal.Decimal `db:"high"` + Low decimal.Decimal `db:"low"` + Close decimal.Decimal `db:"close"` + Volume decimal.Decimal `db:"volume"` +} + +type Repo struct { + db *sqlx.DB +} + +func NewRepo(db *sqlx.DB) *Repo { return &Repo{db: db} } + +func (r *Repo) InsertTick(ctx context.Context, t Tick) error { + _, err := r.db.ExecContext(ctx, ` + INSERT INTO btc_eth_ticks (symbol, ts, price, volume_24h) + VALUES ($1, $2, $3, $4) + ON CONFLICT (symbol, ts) DO NOTHING`, + t.Symbol, t.TS, t.Price, t.Volume24h, + ) + if err != nil { + return fmt.Errorf("insert tick %s @ %s: %w", t.Symbol, t.TS, err) + } + return nil +} + +func (r *Repo) UpsertMinute(ctx context.Context, b MinuteBar) error { + _, err := r.db.ExecContext(ctx, ` + INSERT INTO btc_eth_minute (symbol, minute, open, high, low, close, volume) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (symbol, minute) DO UPDATE SET + open = EXCLUDED.open, + high = EXCLUDED.high, + low = EXCLUDED.low, + close = EXCLUDED.close, + volume = EXCLUDED.volume`, + b.Symbol, b.Minute, b.Open, b.High, b.Low, b.Close, b.Volume, + ) + if err != nil { + return fmt.Errorf("upsert minute %s @ %s: %w", b.Symbol, b.Minute, err) + } + return nil +} + +func (r *Repo) LatestTick(ctx context.Context, symbol string) (Tick, error) { + var t Tick + err := r.db.GetContext(ctx, &t, ` + SELECT symbol, ts, price, volume_24h + FROM btc_eth_ticks + WHERE symbol = $1 + ORDER BY ts DESC LIMIT 1`, symbol) + if err != nil { + return Tick{}, err + } + return t, nil +} + +func (r *Repo) History1h(ctx context.Context, symbol string) ([]MinuteBar, error) { + var rows []MinuteBar + err := r.db.SelectContext(ctx, &rows, ` + SELECT symbol, minute, open, high, low, close, volume + FROM btc_eth_minute + WHERE symbol = $1 + ORDER BY minute DESC LIMIT $2`, symbol, history1hLimit, + ) + if err != nil { + return nil, fmt.Errorf("history1h %s: %w", symbol, err) + } + for i, j := 0, len(rows)-1; i < j; i, j = i+1, j-1 { + rows[i], rows[j] = rows[j], rows[i] + } + return rows, nil +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo_test.go new file mode 100644 index 00000000..ed26fe3b --- /dev/null +++ b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/repo_test.go @@ -0,0 +1,148 @@ +// ©AngelaMos | 2026 +// repo_test.go + +package coinbase_test + +import ( + "context" + "database/sql" + "errors" + "testing" + "time" + + _ "github.com/jackc/pgx/v5/stdlib" + "github.com/jmoiron/sqlx" + "github.com/shopspring/decimal" + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go/modules/postgres" + + "github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/coinbase" +) + +func setupDB(t *testing.T) *sqlx.DB { + t.Helper() + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + pg, err := postgres.Run(ctx, "postgres:17-alpine", + postgres.WithDatabase("monitor"), + postgres.WithUsername("monitor"), + postgres.WithPassword("monitor"), + postgres.BasicWaitStrategies(), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = pg.Terminate(context.Background()) }) + + dsn, err := pg.ConnectionString(ctx, "sslmode=disable") + require.NoError(t, err) + + db, err := sqlx.ConnectContext(ctx, "pgx", dsn) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + + _, err = db.ExecContext(ctx, ` + CREATE TABLE btc_eth_ticks ( + symbol text NOT NULL, + ts timestamptz NOT NULL, + price numeric(18,8) NOT NULL, + volume_24h numeric(20,8), + PRIMARY KEY (symbol, ts) + ); + CREATE TABLE btc_eth_minute ( + symbol text NOT NULL, + minute timestamptz NOT NULL, + open numeric(18,8) NOT NULL, + high numeric(18,8) NOT NULL, + low numeric(18,8) NOT NULL, + close numeric(18,8) NOT NULL, + volume numeric(20,8), + PRIMARY KEY (symbol, minute) + );`) + require.NoError(t, err) + return db +} + +func TestRepo_InsertTickIdempotent(t *testing.T) { + db := setupDB(t) + repo := coinbase.NewRepo(db) + ctx := context.Background() + + ts := time.Now().UTC().Truncate(time.Microsecond) + tk := coinbase.Tick{ + Symbol: "BTC-USD", + TS: ts, + Price: decimal.RequireFromString("42163.45"), + Volume24h: decimal.RequireFromString("19834.91230000"), + } + require.NoError(t, repo.InsertTick(ctx, tk)) + require.NoError(t, repo.InsertTick(ctx, tk)) + + got, err := repo.LatestTick(ctx, "BTC-USD") + require.NoError(t, err) + require.True(t, got.TS.Equal(ts)) + require.True(t, got.Price.Equal(tk.Price)) + require.True(t, got.Volume24h.Equal(tk.Volume24h)) +} + +func TestRepo_UpsertMinuteUpdatesOHLC(t *testing.T) { + db := setupDB(t) + repo := coinbase.NewRepo(db) + ctx := context.Background() + + minute := time.Now().UTC().Truncate(time.Minute) + bar := coinbase.MinuteBar{ + Symbol: "ETH-USD", + Minute: minute, + Open: decimal.RequireFromString("2310.00"), + High: decimal.RequireFromString("2315.50"), + Low: decimal.RequireFromString("2308.10"), + Close: decimal.RequireFromString("2312.75"), + Volume: decimal.RequireFromString("88.12300000"), + } + require.NoError(t, repo.UpsertMinute(ctx, bar)) + + bar.Close = decimal.RequireFromString("2316.00") + bar.High = decimal.RequireFromString("2317.00") + require.NoError(t, repo.UpsertMinute(ctx, bar)) + + hist, err := repo.History1h(ctx, "ETH-USD") + require.NoError(t, err) + require.Len(t, hist, 1) + require.True(t, hist[0].Close.Equal(decimal.RequireFromString("2316.00"))) + require.True(t, hist[0].High.Equal(decimal.RequireFromString("2317.00"))) +} + +func TestRepo_History1hReturnsLast60MinutesOldestFirst(t *testing.T) { + db := setupDB(t) + repo := coinbase.NewRepo(db) + ctx := context.Background() + + now := time.Now().UTC().Truncate(time.Minute) + for i := 0; i < 75; i++ { + bar := coinbase.MinuteBar{ + Symbol: "BTC-USD", + Minute: now.Add(-time.Duration(i) * time.Minute), + Open: decimal.NewFromInt(int64(40000 + i)), + High: decimal.NewFromInt(int64(40050 + i)), + Low: decimal.NewFromInt(int64(39950 + i)), + Close: decimal.NewFromInt(int64(40010 + i)), + Volume: decimal.NewFromInt(int64(i)), + } + require.NoError(t, repo.UpsertMinute(ctx, bar)) + } + + hist, err := repo.History1h(ctx, "BTC-USD") + require.NoError(t, err) + require.Len(t, hist, 60) + require.True(t, hist[0].Minute.Before(hist[len(hist)-1].Minute), "history must be oldest → newest") +} + +func TestRepo_LatestTickMissingReturnsErrNoRows(t *testing.T) { + db := setupDB(t) + repo := coinbase.NewRepo(db) + ctx := context.Background() + + _, err := repo.LatestTick(ctx, "DOES-NOT-EXIST") + require.True(t, errors.Is(err, sql.ErrNoRows)) +}