fiat: add Bitfinex price backend

fiat/bitfinex_api.go

@@ -0,0 +1,193 @@
+package fiat
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/shopspring/decimal"
+)
+
+const (
+	// bitfinexHistoryAPI is the endpoint for historical candle data.
+	// The URL path encodes the time-frame and trading pair:
+	//   /v2/candles/trade:<timeframe>:<symbol>/hist
+	bitfinexHistoryAPI = "https://api-pub.bitfinex.com" +
+		"/v2/candles/trade:%s:%s/hist"
+
+	// bitfinexDefaultPair is the trading pair used to obtain BTC/USD
+	// prices. Trading pair symbols are formed prepending a "t".
+	bitfinexDefaultPair = "tBTCUSD"
+
+	// bitfinexDefaultCurrency is the fiat currency returned.
+	bitfinexDefaultCurrency = "USD"
+
+	// bitfinexCandleCap is the maximum number of candles the API returns
+	// per request.
+	bitfinexCandleCap = 10000
+)
+
+// bitfinexTimeframe maps a Granularity to the Bitfinex candle key string.
+var bitfinexTimeframe = map[Granularity]string{
+	GranularityHour: "1h",
+	GranularityDay:  "1D",
+}
+
+// bitfinexAPI implements the fiatBackend interface using the Bitfinex v2
+// public candles endpoint.
+type bitfinexAPI struct {
+	// granularity controls the candle bucket size (hour or day).
+	granularity Granularity
+
+	// pair is the Bitfinex symbol, e.g. "tBTCUSD".
+	pair string
+
+	// client is the HTTP client used to make requests.
+	client *http.Client
+}
+
+// newBitfinexAPI returns a bitfinexAPI that satisfies fiatBackend.
+func newBitfinexAPI(g Granularity) *bitfinexAPI {
+	return &bitfinexAPI{
+		granularity: g,
+		pair:        bitfinexDefaultPair,
+		client: &http.Client{
+			Timeout: 10 * time.Second,
+		},
+	}
+}
+
+// queryBitfinex performs one HTTP request for a single window of up to
+// bitfinexCandleCap candles. Timestamps are in milliseconds. The sort=1
+// parameter requests ascending order.
+func queryBitfinex(start, end time.Time, pair, timeframe string,
+	cl *http.Client) ([]byte, error) {
+
+	base := fmt.Sprintf(bitfinexHistoryAPI, timeframe, pair)
+	params := url.Values{}
+	params.Set("limit", strconv.Itoa(bitfinexCandleCap))
+	params.Set("start", strconv.FormatInt(start.UnixMilli(), 10))
+	params.Set("end", strconv.FormatInt(end.UnixMilli(), 10))
+	params.Set("sort", "1")
+
+	// #nosec G107 – public data
+	resp, err := cl.Get(base + "?" + params.Encode())
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	return io.ReadAll(resp.Body)
+}
+
+// parseBitfinexData parses the JSON response from the Bitfinex candles
+// endpoint.
+//
+// Bitfinex v2 public candles endpoint
+//
+//	GET https://api-pub.bitfinex.com
+//		/v2/candles/trade:<timeframe>:<symbol>/hist
+//
+// Response body -- array of fixed-width arrays (when sort=1, ascending):
+//
+//	[
+//	  [ MTS,   OPEN,  CLOSE,  HIGH,  LOW,   VOLUME ],
+//	  ...
+//	]
+//
+// Field meanings:
+//   - MTS    -- millisecond timestamp (bucket open).
+//   - OPEN   -- first execution price during the bucket interval.
+//   - CLOSE  -- last execution price during the bucket interval.
+//   - HIGH   -- highest execution price during the bucket interval.
+//   - LOW    -- lowest execution price during the bucket interval.
+//   - VOLUME -- quantity of base asset traded during the bucket interval.
+//
+// We use the CLOSE price (index 2) to be consistent with the other
+// backends.
+func parseBitfinexData(data []byte) ([]*Price, error) {
+	var raw [][]float64
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return nil, err
+	}
+
+	prices := make([]*Price, 0, len(raw))
+	for _, c := range raw {
+		if len(c) < 3 {
+			continue
+		}
+
+		ts := time.UnixMilli(int64(c[0])).UTC()
+		closePx := decimal.NewFromFloat(c[2])
+
+		prices = append(prices, &Price{
+			Timestamp: ts,
+			Price:     closePx,
+			Currency:  bitfinexDefaultCurrency,
+		})
+	}
+
+	return prices, nil
+}
+
+// rawPriceData satisfies the fiatBackend interface.
+func (b *bitfinexAPI) rawPriceData(ctx context.Context,
+	startTime, endTime time.Time) ([]*Price, error) {
+
+	tf, ok := bitfinexTimeframe[b.granularity]
+	if !ok {
+		return nil, fmt.Errorf("bitfinex: unsupported granularity %v",
+			b.granularity.label)
+	}
+
+	// Each request returns at most bitfinexCandleCap candles. We page
+	// forward by advancing start past the last received timestamp.
+	chunk := b.granularity.aggregation * bitfinexCandleCap
+	start := startTime.Truncate(b.granularity.aggregation)
+	end := start.Add(chunk)
+	if end.After(endTime) {
+		end = endTime
+	}
+
+	var all []*Price
+	seen := make(map[int64]struct{})
+	for start.Before(endTime) {
+		queryStart, queryEnd := start, end
+		query := func() ([]byte, error) {
+			return queryBitfinex(
+				queryStart, queryEnd, b.pair, tf, b.client,
+			)
+		}
+
+		records, err := retryQuery(ctx, query, parseBitfinexData)
+		if err != nil {
+			return nil, err
+		}
+
+		// Bitfinex candles can include boundary timestamps for both
+		// start and end. Filter duplicates across page boundaries by
+		// timestamp.
+		for _, record := range records {
+			ts := record.Timestamp.UnixMilli()
+			if _, ok := seen[ts]; ok {
+				continue
+			}
+
+			seen[ts] = struct{}{}
+			all = append(all, record)
+		}
+
+		start = end
+		end = start.Add(chunk)
+		if end.After(endTime) {
+			end = endTime
+		}
+	}
+
+	return all, nil
+}

fiat/bitfinex_api_test.go

@@ -0,0 +1,172 @@
+package fiat
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/jarcoal/httpmock"
+	"github.com/shopspring/decimal"
+	"github.com/stretchr/testify/require"
+)
+
+// TestParseBitfinexData tests parsing of the candle array format returned
+// by the Bitfinex v2 public candles endpoint.
+func TestParseBitfinexData(t *testing.T) {
+	t.Parallel()
+
+	// Bitfinex returns: [MTS, OPEN, CLOSE, HIGH, LOW, VOLUME].
+	// We use the CLOSE field (index 2).
+	ts1 := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
+	ts2 := time.Date(2024, 1, 1, 1, 0, 0, 0, time.UTC)
+
+	input := []byte(`[
+		[` + fmt.Sprintf("%d", ts1.UnixMilli()) +
+		`, 42000.0, 42100.5, 42200.0, 41900.0, 12.5],
+		[` + fmt.Sprintf("%d", ts2.UnixMilli()) +
+		`, 42100.0, 42300.0, 42400.0, 42050.0, 8.3]
+	]`)
+
+	prices, err := parseBitfinexData(input)
+	require.NoError(t, err)
+
+	expected := []*Price{
+		{
+			Timestamp: ts1,
+			Price:     decimal.NewFromFloat(42100.5),
+			Currency:  "USD",
+		},
+		{
+			Timestamp: ts2,
+			Price:     decimal.NewFromFloat(42300.0),
+			Currency:  "USD",
+		},
+	}
+	require.Equal(t, expected, prices)
+}
+
+// TestParseBitfinexDataShortRow verifies that rows with fewer than 3
+// elements are silently skipped.
+func TestParseBitfinexDataShortRow(t *testing.T) {
+	t.Parallel()
+
+	input := []byte(`[[1700000000000, 42000.0], [1700003600000, ` +
+		`42100.0, 42200.0, 42300.0, 42050.0, 5.0]]`)
+	prices, err := parseBitfinexData(input)
+	require.NoError(t, err)
+	require.Len(t, prices, 1)
+	require.Equal(t, decimal.NewFromFloat(42200.0), prices[0].Price)
+}
+
+// TestBitfinexRawPriceData tests the paging logic of rawPriceData using a
+// mocked HTTP transport.
+func TestBitfinexRawPriceData(t *testing.T) {
+	now := time.Now().UTC().Truncate(time.Hour)
+	start := now.Add(-time.Hour * 4)
+
+	mock := httpmock.NewMockTransport()
+	client := &http.Client{Transport: mock}
+
+	const numCandles = 4
+	candles := make([][]float64, numCandles)
+	for i := range candles {
+		ts := start.Add(time.Duration(i) * time.Hour)
+		candles[i] = []float64{
+			float64(ts.UnixMilli()),
+			float64(45000 + i), // open
+			float64(50000 + i), // close
+			float64(55000 + i), // high
+			float64(44000 + i), // low
+			1.0,                // volume
+		}
+	}
+
+	expected := make([]*Price, numCandles)
+	for i := range expected {
+		expected[i] = &Price{
+			Timestamp: start.Add(time.Hour * time.Duration(i)),
+			Price:     decimal.NewFromFloat(float64(50000 + i)),
+			Currency:  "USD",
+		}
+	}
+
+	mock.RegisterResponder(
+		"GET", `=~https://api-pub.bitfinex.com/.*`,
+		httpmock.NewJsonResponderOrPanic(200, candles),
+	)
+
+	api := newBitfinexAPI(GranularityHour)
+	api.client = client
+
+	ctx := context.Background()
+	out, err := api.rawPriceData(ctx, start, now)
+	require.NoError(t, err)
+	require.EqualValues(t, expected, out)
+}
+
+// TestBitfinexRawPriceDataNoDuplicateBoundaries verifies that page boundaries
+// do not produce duplicate timestamps when multiple requests are made.
+func TestBitfinexRawPriceDataNoDuplicateBoundaries(t *testing.T) {
+	t.Parallel()
+
+	end := time.Now().UTC().Truncate(time.Hour)
+	start := end.Add(-time.Duration(bitfinexCandleCap+1) * time.Hour)
+
+	mock := httpmock.NewMockTransport()
+	client := &http.Client{Transport: mock}
+
+	var calls int
+	mock.RegisterResponder(
+		"GET", `=~https://api-pub.bitfinex.com/.*`,
+		func(req *http.Request) (*http.Response, error) {
+			calls++
+
+			query := req.URL.Query()
+			startMS, err := strconv.ParseInt(
+				query.Get("start"), 10, 64,
+			)
+			if err != nil {
+				return nil, err
+			}
+
+			endMS, err := strconv.ParseInt(query.Get("end"), 10, 64)
+			if err != nil {
+				return nil, err
+			}
+
+			// Simulate an inclusive API that returns both
+			// boundaries.
+			candles := [][]float64{
+				{
+					float64(startMS), 0, float64(startMS),
+					0, 0, 1,
+				},
+				{
+					float64(endMS), 0, float64(endMS),
+					0, 0, 1,
+				},
+			}
+
+			return httpmock.NewJsonResponse(200, candles)
+		},
+	)
+
+	api := newBitfinexAPI(GranularityHour)
+	api.client = client
+
+	out, err := api.rawPriceData(context.Background(), start, end)
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, calls, 2, "expected paging to occur")
+
+	seen := make(map[int64]struct{}, len(out))
+	for _, price := range out {
+		ts := price.Timestamp.UnixMilli()
+		_, ok := seen[ts]
+		require.False(t, ok, "duplicate timestamp: %v", price.Timestamp)
+
+		seen[ts] = struct{}{}
+	}
+}

fiat/prices.go

@@ -115,6 +115,16 @@ func (cfg *PriceSourceConfig) validatePriceSourceConfig() error {
 				"daily granularity only",
 				errGranularityUnsupported)
 		}
+
+	case BitfinexPriceBackend:
+		if cfg.Granularity == nil ||
+			(*cfg.Granularity != GranularityHour &&
+				*cfg.Granularity != GranularityDay) {
+
+			return fmt.Errorf("%w: bitfinex supports hourly or "+
+				"daily granularity only",
+				errGranularityUnsupported)
+		}
 	}
 
 	return nil
@@ -175,6 +185,9 @@ const (
 
 	// CoinbasePriceBackend uses Coinbase's API for fiat price data.
 	CoinbasePriceBackend
+
+	// BitfinexPriceBackend uses Bitfinex's API for fiat price data.
+	BitfinexPriceBackend
 )
 
 var priceBackendNames = map[PriceBackend]string{
@@ -184,6 +197,7 @@ var priceBackendNames = map[PriceBackend]string{
 	CustomPriceBackend:    "custom",
 	CoinGeckoPriceBackend: "coingecko",
 	CoinbasePriceBackend:  "coinbase",
+	BitfinexPriceBackend:  "bitfinex",
 }
 
 // String returns the string representation of a price backend.
@@ -231,6 +245,11 @@ func NewPriceSource(cfg *PriceSourceConfig) (*PriceSource, error) {
 		return &PriceSource{
 			impl: newCoinbaseAPI(*cfg.Granularity),
 		}, nil
+
+	case BitfinexPriceBackend:
+		return &PriceSource{
+			impl: newBitfinexAPI(*cfg.Granularity),
+		}, nil
 	}
 
 	return nil, errUnknownPriceBackend

fiat/prices_test.go

@@ -232,6 +232,35 @@ func TestValidatePriceSourceConfig(t *testing.T) {
 			},
 			expectedErr: errGranularityUnsupported,
 		},
+		{
+			name: "bitfinex hourly granularity",
+			cfg: &PriceSourceConfig{
+				Backend:     BitfinexPriceBackend,
+				Granularity: &GranularityHour,
+			},
+		},
+		{
+			name: "bitfinex daily granularity",
+			cfg: &PriceSourceConfig{
+				Backend:     BitfinexPriceBackend,
+				Granularity: &GranularityDay,
+			},
+		},
+		{
+			name: "bitfinex no granularity disallowed",
+			cfg: &PriceSourceConfig{
+				Backend: BitfinexPriceBackend,
+			},
+			expectedErr: errGranularityUnsupported,
+		},
+		{
+			name: "bitfinex minute granularity disallowed",
+			cfg: &PriceSourceConfig{
+				Backend:     BitfinexPriceBackend,
+				Granularity: &GranularityMinute,
+			},
+			expectedErr: errGranularityUnsupported,
+		},
 	}
 
 	for _, test := range tests {