jack/exchange-monitor-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: persist positions to DB before goroutine so restarts don't lose open trades

Browse Source
This commit is contained in:
jackyu66git
2026-05-04 20:16:44 +08:00
parent 411365b66e
commit 06a7e37836
2 changed files with 95 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/trade_repo.go
+8 -2
View File
@@ -105,14 +105,20 @@ func (d *DB) UpdateTradeStatus(id int64, t *TradeRecord) error {
return err
}
// GetOpenTrades returns all trades with status='open'.
// SetTradeStatus updates only the status field of a trade.
func (d *DB) SetTradeStatus(id int64, status string) error {
_, err := d.Exec("UPDATE trades SET status=? WHERE id=?", status, id)
return err
}
// GetOpenTrades returns all non-closed trades (status='open' or status='entering').
func (d *DB) GetOpenTrades() ([]TradeRecord, error) {
rows, err := d.Query(`SELECT id, coin, direction, status, entry_spread, exit_spread,
long_exchange, short_exchange, long_entry, long_exit, short_entry, short_exit,
long_pnl, short_pnl, fee_entry, fee_exit, net_pnl,
amount_usd, scale_count, exit_reason, convergence, opened_at, closed_at,
pnl_long_usd, pnl_short_usd, fee_long_usd, fee_short_usd
FROM trades WHERE status='open'`)
FROM trades WHERE status IN ('open','entering')`)
if err != nil {
return nil, err
}
trader.go
+87 -49
View File
@@ -471,9 +471,32 @@ func (t *Trader) TryEntry(opp *ArbOpportunity, store *PriceStore, notifier *Noti
t.entering[opp.Coin] = true
t.mu.Unlock()
// Persist an "entering" record to DB BEFORE the goroutine,
// so even if the process is killed mid-entry, the position survives restart.
var pendingTradeID int64
if t.db != nil {
now := time.Now()
entrySpread := opp.NetProfit
dbTrade := &db.TradeRecord{
Coin: opp.Coin,
Direction: opp.Direction,
Status: "entering",
EntrySpread: &entrySpread,
LongExchange: opp.BuyEx,
ShortExchange: opp.SellEx,
AmountUSD: t.cfg.TradeAmountUSD,
OpenedAt: now,
}
if id, err := t.db.SaveTrade(dbTrade); err == nil {
pendingTradeID = id
} else {
log.Printf("[Trader] Failed to save entering trade for %s: %v", opp.Coin, err)
}
}
// Async goroutine — placeOrder calls (REST or mock) don't block the main loop
go func() {
t.executeEntry(opp, store, notifier)
t.executeEntry(opp, store, notifier, pendingTradeID)
t.mu.Lock()
delete(t.entering, opp.Coin)
t.mu.Unlock()
@@ -484,7 +507,7 @@ func (t *Trader) TryEntry(opp *ArbOpportunity, store *PriceStore, notifier *Noti
// executeEntry places both legs using the scan-time prices from ArbOpportunity.
// Synchronous — runs in the scanner tick to avoid WS price movement between
// detection and execution.
func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier *Notifier) bool {
func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier *Notifier, pendingTradeID int64) bool {
// Use scan-time prices directly to avoid WS jitter killing the entry
bgP, hlP := opp.BuyPrice, opp.SellPrice
if opp.BuyEx == ExHyperLiquid {
@@ -586,55 +609,64 @@ func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier *
// Persist entry to DB immediately (incremental — not batch at close)
if t.db != nil {
now := time.Now()
status := "filled"
tradeUnit := t.cfg.TradeAmountUSD
es := pos.EntrySpread
dbTrade := &db.TradeRecord{
Coin: pos.Coin,
Direction: pos.Direction,
Status: "open",
EntrySpread: &es,
LongExchange: pos.LongLeg.Exchange,
ShortExchange: pos.ShortLeg.Exchange,
LongEntry: &pos.LongLeg.EntryPrice,
ShortEntry: &pos.ShortLeg.EntryPrice,
AmountUSD: t.cfg.TradeAmountUSD,
OpenedAt: now,
}
if tradeID, err := t.db.SaveTrade(dbTrade); err == nil {
pos.DBTradeID = tradeID
// Use actual fee from exchange (fetched in placeOrder), fall back to estimate
if longFeeUSD <= 0 {
longFeeUSD = tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
if pendingTradeID > 0 {
// Trade was already saved with "entering" status — update to "open"
if err := t.db.SetTradeStatus(pendingTradeID, "open"); err == nil {
pos.DBTradeID = pendingTradeID
log.Printf("[Trader] %s: DB status entering->open (id=%d)", pos.Coin, pendingTradeID)
}
if shortFeeUSD <= 0 {
shortFeeUSD = tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
}
longShares := tradeUnit / pos.LongLeg.EntryPrice
shortShares := tradeUnit / pos.ShortLeg.EntryPrice
} else {
// Fallback: no pending record, insert fresh
now := time.Now()
status := "filled"
tradeUnit := t.cfg.TradeAmountUSD
es := pos.EntrySpread
longOID, _ := t.db.SaveOrder(&db.OrderRecord{
TradeID: tradeID, Leg: "long", Type: "entry",
Exchange: pos.LongLeg.Exchange, Side: "buy",
Price: &pos.LongLeg.EntryPrice, Size: &longShares,
Fee: &longFeeUSD, Status: &status, CreatedAt: now,
})
shortOID, _ := t.db.SaveOrder(&db.OrderRecord{
TradeID: tradeID, Leg: "short", Type: "entry",
Exchange: pos.ShortLeg.Exchange, Side: "sell",
Price: &pos.ShortLeg.EntryPrice, Size: &shortShares,
Fee: &shortFeeUSD, Status: &status, CreatedAt: now,
})
t.db.SaveSystemOrder(&db.SystemOrderRecord{
TradeID: tradeID, Type: "entry", Status: "filled",
Spread: &es,
LongPrice: &pos.LongLeg.EntryPrice, ShortPrice: &pos.ShortLeg.EntryPrice,
LongOrderID: &longOID, ShortOrderID: &shortOID,
CreatedAt: now,
})
dbTrade := &db.TradeRecord{
Coin: pos.Coin,
Direction: pos.Direction,
Status: "open",
EntrySpread: &es,
LongExchange: pos.LongLeg.Exchange,
ShortExchange: pos.ShortLeg.Exchange,
LongEntry: &pos.LongLeg.EntryPrice,
ShortEntry: &pos.ShortLeg.EntryPrice,
AmountUSD: t.cfg.TradeAmountUSD,
OpenedAt: now,
}
if tradeID, err := t.db.SaveTrade(dbTrade); err == nil {
pos.DBTradeID = tradeID
// Use actual fee from exchange (fetched in placeOrder), fall back to estimate
if longFeeUSD <= 0 {
longFeeUSD = tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
}
if shortFeeUSD <= 0 {
shortFeeUSD = tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
}
longShares := tradeUnit / pos.LongLeg.EntryPrice
shortShares := tradeUnit / pos.ShortLeg.EntryPrice
longOID, _ := t.db.SaveOrder(&db.OrderRecord{
TradeID: tradeID, Leg: "long", Type: "entry",
Exchange: pos.LongLeg.Exchange, Side: "buy",
Price: &pos.LongLeg.EntryPrice, Size: &longShares,
Fee: &longFeeUSD, Status: &status, CreatedAt: now,
})
shortOID, _ := t.db.SaveOrder(&db.OrderRecord{
TradeID: tradeID, Leg: "short", Type: "entry",
Exchange: pos.ShortLeg.Exchange, Side: "sell",
Price: &pos.ShortLeg.EntryPrice, Size: &shortShares,
Fee: &shortFeeUSD, Status: &status, CreatedAt: now,
})
t.db.SaveSystemOrder(&db.SystemOrderRecord{
TradeID: tradeID, Type: "entry", Status: "filled",
Spread: &es,
LongPrice: &pos.LongLeg.EntryPrice, ShortPrice: &pos.ShortLeg.EntryPrice,
LongOrderID: &longOID, ShortOrderID: &shortOID,
CreatedAt: now,
})
}
}
}
@@ -1440,6 +1472,12 @@ func (t *Trader) restoreOpenPositions() {
break
}
tr := &openTrades[i]
// Skip "entering" trades — process was killed mid-entry, orders not confirmed
if tr.Status == "entering" {
log.Printf("[Trader] Skipping incomplete trade %d (%s status='entering'), marking as failed", tr.ID, tr.Coin)
t.db.SetTradeStatus(tr.ID, "failed")
continue
}
// Recreate position structure from DB record
pos := &ArbPosition{
Coin: tr.Coin,