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.
This commit is contained in:
CarterPerez-dev
parent
687e13389e
commit
ae5a83c2f8
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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=
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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))
|
||||
}
|
||||
Loading…
Reference in New Issue