Skip to content

Commit 1664c16

Browse files
martonpmartonp
authored
mm: Market Making RPC fixes (#3492)
* mm: Market Making RPC fixes - Allow starting / stopping all bots in config file - Make stopping async so that RPC command doesn't time out - Update UI to support new async bot stopping - Export RPC config * Make sure balance allocation is provided. * Deprecate RPCConfig. * Remove RPCConfig
1 parent f0cea18 commit 1664c16

14 files changed

Lines changed: 267 additions & 195 deletions

File tree

‎client/mm/config.go‎

Lines changed: 31 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -117,18 +117,6 @@ func balanceDiffsToAllocation(diffs *BotInventoryDiffs) *BotBalanceAllocation {
117117
// should be created and the event log db should be updated to support both
118118
// versions.
119119

120-
type rpcConfig struct {
121-
Alloc *BotBalanceAllocation `json:"alloc"`
122-
AutoRebalance *AutoRebalanceConfig `json:"autoRebalance"`
123-
}
124-
125-
func (r *rpcConfig) copy() *rpcConfig {
126-
return &rpcConfig{
127-
Alloc: r.Alloc.copy(),
128-
AutoRebalance: r.AutoRebalance.copy(),
129-
}
130-
}
131-
132120
// BotConfig is the configuration for a market making bot.
133121
// The balance fields are the initial amounts that will be reserved to use for
134122
// this bot. As the bot trades, the amounts reserved for it will be updated.
@@ -156,13 +144,14 @@ type BotConfig struct {
156144
// of the quote asset.
157145
QuoteBridgeName string `json:"quoteBridgeName,omitempty"`
158146

159-
// UIConfig is settings defined and used by the front end to determine
160-
// allocations.
147+
// UIConfig is settings defined and used by the front end for UI state.
161148
UIConfig json.RawMessage `json:"uiConfig,omitempty"`
162149

163-
// RPCConfig can be used for file-based initial allocations and
164-
// auto-rebalance settings.
165-
RPCConfig *rpcConfig `json:"rpcConfig"`
150+
// Alloc is the balance allocation for this bot.
151+
Alloc *BotBalanceAllocation `json:"alloc,omitempty"`
152+
153+
// AutoRebalance configures automatic rebalancing between DEX and CEX.
154+
AutoRebalance *AutoRebalanceConfig `json:"autoRebalance,omitempty"`
166155

167156
// LotSize is the lot size of the market at the time this configuration
168157
// was created. It is used to notify the user if the lot size changes
@@ -185,8 +174,11 @@ func (c *BotConfig) copy() *BotConfig {
185174
b.UIConfig = make(json.RawMessage, len(c.UIConfig))
186175
copy(b.UIConfig, c.UIConfig)
187176
}
188-
if c.RPCConfig != nil {
189-
b.RPCConfig = c.RPCConfig.copy()
177+
if c.Alloc != nil {
178+
b.Alloc = c.Alloc.copy()
179+
}
180+
if c.AutoRebalance != nil {
181+
b.AutoRebalance = c.AutoRebalance.copy()
190182
}
191183
if c.BasicMMConfig != nil {
192184
b.BasicMMConfig = c.BasicMMConfig.copy()
@@ -212,23 +204,26 @@ func (c *BotConfig) updateLotSize(oldLotSize, newLotSize uint64) {
212204
}
213205

214206
func (c *BotConfig) validate(configuredBridgesSupported func([]*configuredBridge) error) error {
215-
bridges := make([]*configuredBridge, 0, 2)
216-
if c.BaseID != c.CEXBaseID {
217-
bridges = append(bridges, &configuredBridge{
218-
dexAssetID: c.BaseID,
219-
cexAssetID: c.CEXBaseID,
220-
bridgeName: c.BaseBridgeName,
221-
})
222-
}
223-
if c.QuoteID != c.CEXQuoteID {
224-
bridges = append(bridges, &configuredBridge{
225-
dexAssetID: c.QuoteID,
226-
cexAssetID: c.CEXQuoteID,
227-
bridgeName: c.QuoteBridgeName,
228-
})
229-
}
230-
if err := configuredBridgesSupported(bridges); err != nil {
231-
return err
207+
// Only validate bridges if CEX is configured (DEX-only bots don't need bridge validation)
208+
if c.CEXName != "" {
209+
bridges := make([]*configuredBridge, 0, 2)
210+
if c.BaseID != c.CEXBaseID {
211+
bridges = append(bridges, &configuredBridge{
212+
dexAssetID: c.BaseID,
213+
cexAssetID: c.CEXBaseID,
214+
bridgeName: c.BaseBridgeName,
215+
})
216+
}
217+
if c.QuoteID != c.CEXQuoteID {
218+
bridges = append(bridges, &configuredBridge{
219+
dexAssetID: c.QuoteID,
220+
cexAssetID: c.CEXQuoteID,
221+
bridgeName: c.QuoteBridgeName,
222+
})
223+
}
224+
if err := configuredBridgesSupported(bridges); err != nil {
225+
return err
226+
}
232227
}
233228

234229
if c.BasicMMConfig != nil {

‎client/mm/mm.go‎

Lines changed: 77 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,9 @@ import (
1212
"maps"
1313
"os"
1414
"slices"
15+
"strings"
1516
"sync"
17+
"sync/atomic"
1618
"time"
1719

1820
"decred.org/dcrdex/client/asset"
@@ -119,8 +121,9 @@ type bot interface {
119121

120122
type runningBot struct {
121123
bot
122-
cm *dex.ConnectionMaster
123-
cexCfg *CEXConfig
124+
cm *dex.ConnectionMaster
125+
cexCfg *CEXConfig
126+
stopping atomic.Bool
124127
}
125128

126129
func (rb *runningBot) assets() map[uint32]any {
@@ -343,8 +346,9 @@ func newCEXProblems() *CEXProblems {
343346

344347
// BotStatus is state information about a configured bot.
345348
type BotStatus struct {
346-
Config *BotConfig `json:"config"`
347-
Running bool `json:"running"`
349+
Config *BotConfig `json:"config"`
350+
Running bool `json:"running"`
351+
Stopping bool `json:"stopping"`
348352
// RunStats being non-nil means the bot is running.
349353
RunStats *RunStats `json:"runStats"`
350354
LatestEpoch *EpochReport `json:"latestEpoch"`
@@ -366,14 +370,17 @@ func (m *MarketMaker) Status() *Status {
366370
var stats *RunStats
367371
var epochReport *EpochReport
368372
var cexProblems *CEXProblems
373+
var stopping bool
369374
if rb != nil {
370375
stats = rb.stats()
371376
epochReport = rb.latestEpoch()
372377
cexProblems = rb.latestCEXProblems()
378+
stopping = rb.stopping.Load()
373379
}
374380
status.Bots = append(status.Bots, &BotStatus{
375381
Config: botCfg,
376382
Running: rb != nil,
383+
Stopping: stopping,
377384
RunStats: stats,
378385
LatestEpoch: epochReport,
379386
CEXProblems: cexProblems,
@@ -409,6 +416,7 @@ func (m *MarketMaker) RunningBotsStatus() *Status {
409416
status.Bots = append(status.Bots, &BotStatus{
410417
Config: rb.botCfg(),
411418
Running: true,
419+
Stopping: rb.stopping.Load(),
412420
RunStats: rb.stats(),
413421
LatestEpoch: rb.latestEpoch(),
414422
CEXProblems: rb.latestCEXProblems(),
@@ -718,6 +726,10 @@ func (m *MarketMaker) Connect(ctx context.Context) (*sync.WaitGroup, error) {
718726
}
719727

720728
func (m *MarketMaker) balancesSufficient(balances *BotBalanceAllocation, mkt *MarketWithHost, botCfg *BotConfig, cexCfg *CEXConfig) error {
729+
if balances == nil {
730+
return fmt.Errorf("no balance allocation provided")
731+
}
732+
721733
availableDEXBalances, availableCEXBalances, err := m.availableBalances(mkt, botCfg.CEXBaseID, botCfg.CEXQuoteID, cexCfg)
722734
if err != nil {
723735
return fmt.Errorf("error getting available balances: %v", err)
@@ -863,28 +875,19 @@ func (m *MarketMaker) newBot(cfg *BotConfig, adaptorCfg *exchangeAdaptorCfg) (bo
863875
}
864876
}
865877

866-
// StartConfig contains the data that must be submitted with a call to StartBot.
867-
type StartConfig struct {
868-
MarketWithHost
869-
AutoRebalance *AutoRebalanceConfig `json:"autoRebalance"`
870-
Alloc *BotBalanceAllocation `json:"alloc"`
871-
}
872-
873878
// StartBot starts a market making bot.
874-
func (m *MarketMaker) StartBot(startCfg *StartConfig, alternateConfigPath *string, appPW []byte, overrideLotSizeChange bool) (err error) {
875-
mkt := startCfg.MarketWithHost
876-
879+
func (m *MarketMaker) StartBot(mkt *MarketWithHost, alternateConfigPath *string, appPW []byte, overrideLotSizeChange bool) (err error) {
877880
m.startUpdateMtx.Lock()
878881
defer m.startUpdateMtx.Unlock()
879882

880883
m.runningBotsMtx.RLock()
881-
_, found := m.runningBots[startCfg.MarketWithHost]
884+
_, found := m.runningBots[*mkt]
882885
m.runningBotsMtx.RUnlock()
883886
if found {
884887
return fmt.Errorf("bot for %s already running", mkt)
885888
}
886889

887-
coreMkt, err := m.core.ExchangeMarket(startCfg.Host, startCfg.BaseID, startCfg.QuoteID)
890+
coreMkt, err := m.core.ExchangeMarket(mkt.Host, mkt.BaseID, mkt.QuoteID)
888891
if err != nil {
889892
return fmt.Errorf("error getting market: %v", err)
890893
}
@@ -898,16 +901,11 @@ func (m *MarketMaker) StartBot(startCfg *StartConfig, alternateConfigPath *strin
898901
}
899902
}
900903

901-
botCfg, cexCfg, err := m.configsForMarket(&startCfg.MarketWithHost, alternateConfigPath)
904+
botCfg, cexCfg, err := m.configsForMarket(mkt, alternateConfigPath)
902905
if err != nil {
903906
return err
904907
}
905908

906-
if botCfg.RPCConfig != nil {
907-
startCfg.Alloc = botCfg.RPCConfig.Alloc
908-
startCfg.AutoRebalance = botCfg.RPCConfig.AutoRebalance
909-
}
910-
911909
// Lot size may be zero if started from RPC. If the lot size in the config
912910
// is set, then we check if the lot size has changed since the configuration
913911
// was saved. If so, and overrideLotSizeChange is false, we return an error.
@@ -933,12 +931,11 @@ func (m *MarketMaker) StartBot(startCfg *StartConfig, alternateConfigPath *strin
933931
botCfg.LotSize = mktInfo.LotSize
934932
}
935933

936-
return m.startBot(startCfg, botCfg, cexCfg, appPW)
934+
return m.startBot(mkt, botCfg, cexCfg, appPW)
937935
}
938936

939-
func (m *MarketMaker) startBot(startCfg *StartConfig, botCfg *BotConfig, cexCfg *CEXConfig, appPW []byte) (err error) {
940-
mwh := &startCfg.MarketWithHost
941-
if err := m.balancesSufficient(startCfg.Alloc, mwh, botCfg, cexCfg); err != nil {
937+
func (m *MarketMaker) startBot(mkt *MarketWithHost, botCfg *BotConfig, cexCfg *CEXConfig, appPW []byte) (err error) {
938+
if err := m.balancesSufficient(botCfg.Alloc, mkt, botCfg, cexCfg); err != nil {
942939
return err
943940
}
944941

@@ -971,10 +968,10 @@ func (m *MarketMaker) startBot(startCfg *StartConfig, botCfg *BotConfig, cexCfg
971968

972969
adaptorCfg := &exchangeAdaptorCfg{
973970
botID: dexMarketID(botCfg.Host, botCfg.BaseID, botCfg.QuoteID),
974-
mwh: mwh,
975-
baseDexBalances: startCfg.Alloc.DEX,
976-
baseCexBalances: startCfg.Alloc.CEX,
977-
autoRebalanceConfig: startCfg.AutoRebalance,
971+
mwh: mkt,
972+
baseDexBalances: botCfg.Alloc.DEX,
973+
baseCexBalances: botCfg.Alloc.CEX,
974+
autoRebalanceConfig: botCfg.AutoRebalance,
978975
core: m.core,
979976
cex: cex,
980977
log: m.botSubLogger(botCfg),
@@ -997,14 +994,14 @@ func (m *MarketMaker) startBot(startCfg *StartConfig, botCfg *BotConfig, cexCfg
997994
go func() {
998995
cm.Wait()
999996
m.runningBotsMtx.Lock()
1000-
if bot, found := m.runningBots[*mwh]; found {
997+
if bot, found := m.runningBots[*mkt]; found {
1001998
if bot.botCfg().requiresPriceOracle() {
1002-
m.oracle.stopAutoSyncingMarket(mwh.BaseID, mwh.QuoteID)
999+
m.oracle.stopAutoSyncingMarket(mkt.BaseID, mkt.QuoteID)
10031000
}
1004-
delete(m.runningBots, *mwh)
1001+
delete(m.runningBots, *mkt)
10051002
}
10061003
m.runningBotsMtx.Unlock()
1007-
m.core.Broadcast(newRunStatsNote(mwh.Host, mwh.BaseID, mwh.QuoteID, nil))
1004+
m.core.Broadcast(newRunStatsNote(mkt.Host, mkt.BaseID, mkt.QuoteID, nil))
10081005
}()
10091006

10101007
startedBot = true
@@ -1016,24 +1013,66 @@ func (m *MarketMaker) startBot(startCfg *StartConfig, botCfg *BotConfig, cexCfg
10161013
}
10171014

10181015
m.runningBotsMtx.Lock()
1019-
m.runningBots[*mwh] = rb
1016+
m.runningBots[*mkt] = rb
10201017
m.runningBotsMtx.Unlock()
10211018

10221019
return nil
10231020
}
10241021

1025-
// StopBot stops a running bot.
1022+
// StopBot stops a running bot. The function returns immediately; the bot
1023+
// stops asynchronously. A notification is broadcast when the bot finishes
1024+
// stopping.
10261025
func (m *MarketMaker) StopBot(mkt *MarketWithHost) error {
10271026
runningBots := m.runningBotsLookup()
10281027
bot, found := runningBots[*mkt]
10291028
if !found {
10301029
return fmt.Errorf("no bot running on market: %s", mkt)
10311030
}
1032-
bot.cm.Disconnect()
1033-
m.core.Broadcast(newRunStatsNote(mkt.Host, mkt.BaseID, mkt.QuoteID, nil))
1031+
bot.stopping.Store(true)
1032+
go bot.cm.Disconnect()
10341033
return nil
10351034
}
10361035

1036+
// StartBots starts all bots in the provided config file.
1037+
func (m *MarketMaker) StartBots(cfgPath *string, appPW []byte) (started int, err error) {
1038+
cfg, err := getMarketMakingConfig(*cfgPath)
1039+
if err != nil {
1040+
return 0, fmt.Errorf("error loading config: %w", err)
1041+
}
1042+
if len(cfg.BotConfigs) == 0 {
1043+
return 0, fmt.Errorf("no bots configured in %s", *cfgPath)
1044+
}
1045+
var errs []string
1046+
for _, botCfg := range cfg.BotConfigs {
1047+
mkt := &MarketWithHost{botCfg.Host, botCfg.BaseID, botCfg.QuoteID}
1048+
if err := m.StartBot(mkt, cfgPath, appPW, true); err != nil {
1049+
errs = append(errs, fmt.Sprintf("%s: %v", mkt, err))
1050+
} else {
1051+
started++
1052+
}
1053+
}
1054+
if len(errs) > 0 {
1055+
return started, fmt.Errorf("some bots failed to start: %s", strings.Join(errs, "; "))
1056+
}
1057+
return started, nil
1058+
}
1059+
1060+
// StopBots signals all running bots to stop and returns immediately.
1061+
// The bots will finish stopping asynchronously.
1062+
func (m *MarketMaker) StopBots() (stopped int, err error) {
1063+
runningBots := m.runningBotsLookup()
1064+
if len(runningBots) == 0 {
1065+
return 0, fmt.Errorf("no bots running")
1066+
}
1067+
for mkt := range runningBots {
1068+
if err := m.StopBot(&mkt); err != nil {
1069+
continue
1070+
}
1071+
stopped++
1072+
}
1073+
return stopped, nil
1074+
}
1075+
10371076
func getMarketMakingConfig(path string) (*MarketMakingConfig, error) {
10381077
if path == "" {
10391078
return nil, fmt.Errorf("no config file provided")

‎client/mm/sample-arb-mm.json‎

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -14,16 +14,14 @@
1414
"multisplit": "true",
1515
"multisplitbuffer": "5"
1616
},
17-
"rpcConfig": {
18-
"alloc": {
19-
"dex": {
20-
"60": 1000000000,
21-
"0": 10000000
22-
},
23-
"cex": {
24-
"60": 1000000000,
25-
"0": 10000000
26-
}
17+
"alloc": {
18+
"dex": {
19+
"60": 1000000000,
20+
"0": 10000000
21+
},
22+
"cex": {
23+
"60": 1000000000,
24+
"0": 10000000
2725
}
2826
},
2927
"arbMarketMakingConfig": {

‎client/mm/sample-arb.json‎

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -20,16 +20,14 @@
2020
"minQuoteTransfer": 10000000
2121
}
2222
},
23-
"rpcConfig": {
24-
"alloc": {
25-
"dex": {
26-
"60": 1000000000,
27-
"0": 10000000
28-
},
29-
"cex": {
30-
"60": 1000000000,
31-
"0": 10000000
32-
}
23+
"alloc": {
24+
"dex": {
25+
"60": 1000000000,
26+
"0": 10000000
27+
},
28+
"cex": {
29+
"60": 1000000000,
30+
"0": 10000000
3331
}
3432
},
3533
"simpleArbConfig": {

0 commit comments

Comments
 (0)