FEATURE: [xfundingv2] strategy state persistence support

pkg/strategy/xfundingv2/arb_round_fee.go

@@ -11,9 +11,13 @@ import (
 )
 
 type PendingRound struct {
-	Round         *ArbitrageRound
-	RetryCount    int
-	LastRetryTime time.Time
+	Round         *ArbitrageRound `json:"round"`
+	RetryCount    int             `json:"retryCount"`
+	LastRetryTime time.Time       `json:"lastRetryTime"`
+}
+
+func (r *PendingRound) Initialize(ctx context.Context, s *Strategy) error {
+	return r.Round.Initialize(ctx, s)
 }
 
 func (s *Strategy) processPendingRounds(ctx context.Context, currentTime time.Time) {

pkg/strategy/xfundingv2/arb_round_pnl.go

@@ -54,12 +54,12 @@ func (r *ArbitrageRound) PnL(ctx context.Context, currentTime time.Time) RoundPn
 	if r.futuresExchangeFeeRates != nil {
 		futuresPosition.ExchangeFeeRates = r.futuresExchangeFeeRates
 	}
-	if !r.avgFeeCost.IsZero() {
+	if !r.syncState.AvgFeeCost.IsZero() {
 		spotPosition.FeeAverageCosts = map[string]fixedpoint.Value{
-			r.feeSymbol: r.avgFeeCost,
+			r.syncState.FeeSymbol: r.syncState.AvgFeeCost,
 		}
 		futuresPosition.FeeAverageCosts = map[string]fixedpoint.Value{
-			r.feeSymbol: r.avgFeeCost,
+			r.syncState.FeeSymbol: r.syncState.AvgFeeCost,
 		}
 	}
 

pkg/strategy/xfundingv2/arb_round_pnl_test.go

@@ -33,10 +33,10 @@ func TestArbitrageRound_TradePnL(t *testing.T) {
 	t.Run("returns zero profit when position is only opened", func(t *testing.T) {
 		// Add orders first so AddTrade accepts them
 		spotExecutor := round.spotWorker.Executor()
-		spotExecutor.orders[1] = types.OrderQuery{OrderID: "1"}
+		spotExecutor.syncState.Orders[1] = types.OrderQuery{OrderID: "1"}
 
 		futuresExecutor := round.futuresWorker.Executor()
-		futuresExecutor.orders[2] = types.OrderQuery{OrderID: "2"}
+		futuresExecutor.syncState.Orders[2] = types.OrderQuery{OrderID: "2"}
 
 		// Opening trades: buy spot at 40000, sell futures at 40100
 		spotExecutor.AddTrade(types.Trade{
@@ -74,10 +74,10 @@ func TestArbitrageRound_TradePnL(t *testing.T) {
 
 	t.Run("calculates realized profit after closing trades", func(t *testing.T) {
 		spotExecutor := round.spotWorker.Executor()
-		spotExecutor.orders[3] = types.OrderQuery{OrderID: "3"}
+		spotExecutor.syncState.Orders[3] = types.OrderQuery{OrderID: "3"}
 
 		futuresExecutor := round.futuresWorker.Executor()
-		futuresExecutor.orders[4] = types.OrderQuery{OrderID: "4"}
+		futuresExecutor.syncState.Orders[4] = types.OrderQuery{OrderID: "4"}
 
 		// Closing trades: sell spot at 41000 (profit), buy futures at 39900 (profit)
 		spotExecutor.AddTrade(types.Trade{

pkg/strategy/xfundingv2/arb_round_sync.go

@@ -0,0 +1,125 @@
+package xfundingv2
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func (r *ArbitrageRound) Initialize(ctx context.Context, s *Strategy) error {
+	r.SetLogger(s.logger)
+	r.SetFuturesExchangeFeeRates(
+		map[types.ExchangeName]types.ExchangeFee{
+			s.futuresSession.Exchange.Name(): {
+				MakerFeeRate: s.futuresSession.MakerFeeRate,
+				TakerFeeRate: s.futuresSession.TakerFeeRate,
+			},
+		},
+	)
+	r.SetSpotExchangeFeeRates(
+		map[types.ExchangeName]types.ExchangeFee{
+			s.spotSession.Exchange.Name(): {
+				MakerFeeRate: s.spotSession.MakerFeeRate,
+				TakerFeeRate: s.spotSession.TakerFeeRate,
+			},
+		},
+	)
+	r.retryTransferTickC = make(chan time.Time)
+	if !r.HasStarted() {
+		// the round has been started before, we need to start the retry worker
+		go r.retryTransferWorker(ctx, r.retryTransferTickC)
+	}
+	if service, ok := s.futuresSession.Exchange.(FuturesService); ok {
+		r.futuresService = service
+	} else {
+		return errors.New("[ArbitrageRound] futures exchange does not implement FuturesService")
+	}
+	if r.spotWorker != nil {
+		if err := r.spotWorker.Initialize(ctx, s); err != nil {
+			return fmt.Errorf("[ArbitrageRound] spot load strategy error: %w", err)
+		}
+	} else {
+		// should not happend
+		// by the time we create the round, the spot worker is never nil
+		// the restored round should always have the spot worker restored as well.
+		return errors.New("[ArbitrageRound] spot worker is nil")
+	}
+	if r.futuresWorker != nil {
+		if err := r.futuresWorker.Initialize(ctx, s); err != nil {
+			return fmt.Errorf("[ArbitrageRound] futures load strategy error: %w", err)
+		}
+	} else {
+		// should not happend
+		// by the time we create the round, the futures worker is never nil
+		// the restored round should always have the futures worker restored as well.
+		return errors.New("[ArbitrageRound] futures worker is nil")
+	}
+
+	return nil
+}
+
+type ArbitrageRoundSyncState struct {
+	TriggeredFundingRate        fixedpoint.Value     `json:"triggeredFundingRate"`
+	TriggeredSpotTargetPosition fixedpoint.Value     `json:"triggeredSpotTargetPosition"`
+	MinHoldingIntervals         int                  `json:"minHoldingIntervals"`
+	FundingIntervalHours        int                  `json:"fundingIntervalHours"`
+	FundingIntervalStart        time.Time            `json:"fundingIntervalStart"`
+	FundingIntervalEnd          time.Time            `json:"fundingIntervalEnd"`
+	FundingFeeRecords           map[int64]FundingFee `json:"fundingFeeRecords"`
+
+	Symbol              string             `json:"symbol"`
+	SpotExchangeName    types.ExchangeName `json:"spotExchangeName"`
+	FuturesExchangeName types.ExchangeName `json:"futuresExchangeName"`
+	Asset               string             `json:"asset"` // base asset, e.g. "BTC"
+
+	SpotFeeAssetAmount    fixedpoint.Value `json:"spotFeeAssetAmount"`
+	FuturesFeeAssetAmount fixedpoint.Value `json:"futuresFeeAssetAmount"`
+	FeeSymbol             string           `json:"feeSymbol"`
+	AvgFeeCost            fixedpoint.Value `json:"avgFeeCost"`
+
+	RetryDuration  time.Duration            `json:"retryDuration"`
+	RetryTransfers map[uint64]transferRetry `json:"retryTransfers"`
+
+	State RoundState `json:"state"`
+
+	// StartTime is the time when the round is started
+	StartTime time.Time `json:"startTime"`
+	// ClosingTime is the time when the round is entered closing state
+	ClosingTime     time.Time     `json:"closingTime"`
+	ClosingDuration time.Duration `json:"closingDuration"`
+	// LastUpdateTime is the last time when the round is updated
+	LastUpdateTime time.Time `json:"lastUpdateTime"`
+}
+
+func (r *ArbitrageRound) MarshalJSON() ([]byte, error) {
+	v := struct {
+		SyncState     ArbitrageRoundSyncState `json:"syncState"`
+		SpotWorker    *TWAPWorker             `json:"spotWorker,omitempty"`
+		FuturesWorker *TWAPWorker             `json:"futuresWorker,omitempty"`
+	}{
+		SyncState:     r.syncState,
+		SpotWorker:    r.spotWorker,
+		FuturesWorker: r.futuresWorker,
+	}
+	return json.Marshal(v)
+}
+
+func (r *ArbitrageRound) UnmarshalJSON(b []byte) error {
+	v := struct {
+		SyncState     ArbitrageRoundSyncState `json:"syncState"`
+		SpotWorker    *TWAPWorker             `json:"spotWorker,omitempty"`
+		FuturesWorker *TWAPWorker             `json:"futuresWorker,omitempty"`
+	}{}
+	if err := json.Unmarshal(b, &v); err != nil {
+		return err
+	}
+	r.syncState = v.SyncState
+	r.spotWorker = v.SpotWorker
+	r.futuresWorker = v.FuturesWorker
+	return nil
+}

pkg/strategy/xfundingv2/arb_round_sync_test.go

@@ -0,0 +1,108 @@
+package xfundingv2
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func TestArbitrageRound_MarshalUnmarshalJSON(t *testing.T) {
+	t.Run("round_trip_preserves_all_fields", func(t *testing.T) {
+		startTime := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
+		closingTime := time.Date(2025, 1, 2, 0, 0, 0, 0, time.UTC)
+		lastUpdateTime := time.Date(2025, 1, 2, 1, 0, 0, 0, time.UTC)
+		fundingStart := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
+		fundingEnd := time.Date(2025, 1, 1, 7, 59, 59, 0, time.UTC)
+
+		round := &ArbitrageRound{
+			syncState: ArbitrageRoundSyncState{
+				TriggeredFundingRate:        fixedpoint.NewFromFloat(0.001),
+				TriggeredSpotTargetPosition: fixedpoint.NewFromFloat(0.5),
+				MinHoldingIntervals:         3,
+				FundingIntervalHours:        8,
+				FundingIntervalStart:        fundingStart,
+				FundingIntervalEnd:          fundingEnd,
+				FundingFeeRecords: map[int64]FundingFee{
+					100: {
+						Asset:  "BTC",
+						Amount: fixedpoint.NewFromFloat(0.0001),
+						Txn:    100,
+						Time:   startTime,
+					},
+				},
+				Symbol:              "BTCUSDT",
+				SpotExchangeName:    types.ExchangeBinance,
+				FuturesExchangeName: types.ExchangeBinance,
+				Asset:               "BTC",
+
+				SpotFeeAssetAmount:    fixedpoint.NewFromFloat(0.01),
+				FuturesFeeAssetAmount: fixedpoint.NewFromFloat(0.02),
+				FeeSymbol:             "BNB",
+				AvgFeeCost:            fixedpoint.NewFromFloat(600.0),
+
+				RetryDuration: 5 * time.Minute,
+				RetryTransfers: map[uint64]transferRetry{
+					1: {
+						Trade: types.Trade{
+							ID:       1,
+							Exchange: types.ExchangeBinance,
+							Side:     types.SideTypeBuy,
+						},
+						LastTried: startTime,
+					},
+				},
+
+				State:           RoundReady,
+				StartTime:       startTime,
+				ClosingTime:     closingTime,
+				ClosingDuration: 30 * time.Minute,
+				LastUpdateTime:  lastUpdateTime,
+			},
+		}
+
+		data, err := json.Marshal(round)
+		require.NoError(t, err)
+
+		var restored ArbitrageRound
+		err = json.Unmarshal(data, &restored)
+		require.NoError(t, err)
+
+		assert.Equal(t, round.syncState, restored.syncState)
+	})
+
+	t.Run("nil_workers_are_preserved", func(t *testing.T) {
+		round := &ArbitrageRound{
+			syncState: ArbitrageRoundSyncState{
+				TriggeredFundingRate: fixedpoint.NewFromFloat(0.0005),
+				SpotExchangeName:    types.ExchangeBinance,
+				FuturesExchangeName: types.ExchangeBinance,
+				State:               RoundPending,
+				Asset:               "ETH",
+				FundingFeeRecords:   make(map[int64]FundingFee),
+			},
+		}
+
+		data, err := json.Marshal(round)
+		require.NoError(t, err)
+
+		var restored ArbitrageRound
+		err = json.Unmarshal(data, &restored)
+		require.NoError(t, err)
+
+		assert.Nil(t, restored.spotWorker)
+		assert.Nil(t, restored.futuresWorker)
+		assert.Equal(t, round.syncState, restored.syncState)
+	})
+
+	t.Run("unmarshal_invalid_json_returns_error", func(t *testing.T) {
+		var round ArbitrageRound
+		err := json.Unmarshal([]byte(`{invalid`), &round)
+		assert.Error(t, err)
+	})
+}

pkg/strategy/xfundingv2/arb_round_test.go

@@ -76,7 +76,7 @@ func TestArbitrageRound_CollectedFunding(t *testing.T) {
 	})
 
 	t.Run("sums funding fee records", func(t *testing.T) {
-		round.startTime = time.Date(2024, 1, 1, 1, 0, 0, 0, time.UTC)
+		round.syncState.StartTime = time.Date(2024, 1, 1, 1, 0, 0, 0, time.UTC)
 
 		// Simulate funding fee income returned by the service
 		mockService.incomeHistory = []binanceapi.FuturesIncome{