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

Phase 1: SQLite persistence layer

Browse Source 
- Add modernc.org/sqlite (pure Go, no CGO)
- db/ package: trades, orders, config_log tables + CRUD
- Trade persistence: every closed trade saved to SQLite
- Restart recovery: open positions restored from DB
- Automatic migration on startup
This commit is contained in:
jackyu66git
2026-05-03 17:28:08 +08:00
parent 189fc0d9b6
commit b09314f317
7 changed files with 457 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/trades.db
BIN
View File
Binary file not shown.
db/db.go
+109
View File
@@ -0,0 +1,109 @@
package db
import (
"database/sql"
"log"
"os"
"path/filepath"
"time"
_ "modernc.org/sqlite"
)
// DB wraps the sql.DB connection with our schema.
type DB struct {
*sql.DB
Path string
}
// Open opens or creates the SQLite database at the given path.
func Open(path string) (*DB, error) {
if path == "" {
home, _ := os.UserHomeDir()
path = filepath.Join(home, "Project", "exchange-monitor-go", "data", "trades.db")
}
// Ensure directory exists
os.MkdirAll(filepath.Dir(path), 0755)
db, err := sql.Open("sqlite", path)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(1) // SQLite single-writer
db.SetMaxIdleConns(1)
d := &DB{DB: db, Path: path}
if err := d.migrate(); err != nil {
return nil, err
}
return d, nil
}
func (d *DB) migrate() error {
schema := `
CREATE TABLE IF NOT EXISTS trades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
coin TEXT NOT NULL,
direction TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'open',
entry_spread REAL,
exit_spread REAL,
long_exchange TEXT,
short_exchange TEXT,
long_entry REAL,
long_exit REAL,
short_entry REAL,
short_exit REAL,
long_pnl REAL,
short_pnl REAL,
fee_entry REAL,
fee_exit REAL,
net_pnl REAL,
amount_usd REAL,
scale_count INTEGER DEFAULT 0,
exit_reason TEXT,
convergence TEXT,
opened_at DATETIME NOT NULL,
closed_at DATETIME
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
trade_id INTEGER NOT NULL REFERENCES trades(id),
leg TEXT NOT NULL,
type TEXT NOT NULL,
exchange TEXT NOT NULL,
side TEXT NOT NULL,
price REAL,
size REAL,
fee REAL,
order_id TEXT,
status TEXT,
created_at DATETIME NOT NULL
);
CREATE TABLE IF NOT EXISTS config_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
key TEXT NOT NULL,
value TEXT NOT NULL,
changed_at DATETIME NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_trades_coin ON trades(coin);
CREATE INDEX IF NOT EXISTS idx_trades_status ON trades(status);
CREATE INDEX IF NOT EXISTS idx_trades_opened ON trades(opened_at);
CREATE INDEX IF NOT EXISTS idx_orders_trade_id ON orders(trade_id);
`
_, err := d.Exec(schema)
if err != nil {
return err
}
log.Printf("[DB] SQLite ready: %s", d.Path)
return nil
}
// Now returns the current time in UTC.
func Now() time.Time {
return time.Now().UTC()
}
db/trade_repo.go
+208
View File
@@ -0,0 +1,208 @@
package db
import (
"database/sql"
"time"
)
// TradeRecord mirrors the database row for trades table.
type TradeRecord struct {
ID int64
Coin string
Direction string
Status string // open / closed
EntrySpread *float64
ExitSpread *float64
LongExchange string
ShortExchange string
LongEntry *float64
LongExit *float64
ShortEntry *float64
ShortExit *float64
LongPnl *float64
ShortPnl *float64
FeeEntry *float64
FeeExit *float64
NetPnl *float64
AmountUSD float64
ScaleCount int
ExitReason *string
Convergence *string
OpenedAt time.Time
ClosedAt *time.Time
}
// OrderRecord mirrors the database row for orders table.
type OrderRecord struct {
ID int64
TradeID int64
Leg string // long / short
Type string // entry / exit / scale
Exchange string
Side string // buy / sell
Price *float64
Size *float64
Fee *float64
OrderID *string
Status *string
CreatedAt time.Time
}
// SaveTrade inserts a new trade and returns its ID.
func (d *DB) SaveTrade(t *TradeRecord) (int64, error) {
res, err := d.Exec(`INSERT INTO trades (
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
) VALUES (?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?)`,
t.Coin, t.Direction, t.Status, t.EntrySpread, t.ExitSpread,
t.LongExchange, t.ShortExchange, t.LongEntry, t.LongExit, t.ShortEntry, t.ShortExit,
t.LongPnl, t.ShortPnl, t.FeeEntry, t.FeeExit, t.NetPnl,
t.AmountUSD, t.ScaleCount, t.ExitReason, t.Convergence, t.OpenedAt, t.ClosedAt,
)
if err != nil {
return 0, err
}
return res.LastInsertId()
}
// UpdateTradeStatus updates an existing trade's close data.
func (d *DB) UpdateTradeStatus(id int64, t *TradeRecord) error {
_, err := d.Exec(`UPDATE trades SET
status=?, exit_spread=?, long_exit=?, short_exit=?,
long_pnl=?, short_pnl=?, net_pnl=?,
scale_count=?, exit_reason=?, convergence=?, closed_at=?
WHERE id=?`,
t.Status, t.ExitSpread,
t.LongExit, t.ShortExit,
t.LongPnl, t.ShortPnl, t.NetPnl,
t.ScaleCount, t.ExitReason, t.Convergence, t.ClosedAt,
id,
)
return err
}
// GetOpenTrades returns all trades with status='open'.
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
FROM trades WHERE status='open'`)
if err != nil {
return nil, err
}
defer rows.Close()
return scanTrades(rows)
}
// GetTrades returns paginated closed trades.
func (d *DB) GetTrades(page, limit int, coin string) ([]TradeRecord, int, error) {
// Count total
var total int
countSQL := "SELECT COUNT(*) FROM trades WHERE status='closed'"
args := []interface{}{}
if coin != "" {
countSQL += " AND coin=?"
args = append(args, coin)
}
if err := d.QueryRow(countSQL, args...).Scan(&total); err != nil {
return nil, 0, err
}
// Fetch page
offset := (page - 1) * limit
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
FROM trades WHERE status='closed'`
if coin != "" {
query += " AND coin=?"
}
query += " ORDER BY closed_at DESC LIMIT ? OFFSET ?"
allArgs := args
allArgs = append(allArgs, limit, offset)
rows, err := d.Query(query, allArgs...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
trades, err := scanTrades(rows)
return trades, total, err
}
// SaveOrder inserts an order record.
func (d *DB) SaveOrder(o *OrderRecord) (int64, error) {
res, err := d.Exec(`INSERT INTO orders
(trade_id, leg, type, exchange, side, price, size, fee, order_id, status, created_at)
VALUES (?,?,?,?,?, ?,?,?,?,?, ?)`,
o.TradeID, o.Leg, o.Type, o.Exchange, o.Side,
o.Price, o.Size, o.Fee, o.OrderID, o.Status, o.CreatedAt,
)
if err != nil {
return 0, err
}
return res.LastInsertId()
}
// GetTradeByID returns a single trade with its orders.
func (d *DB) GetTradeByID(id int64) (*TradeRecord, []OrderRecord, error) {
row := d.QueryRow(`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
FROM trades WHERE id=?`, id)
var t TradeRecord
err := row.Scan(
&t.ID, &t.Coin, &t.Direction, &t.Status, &t.EntrySpread, &t.ExitSpread,
&t.LongExchange, &t.ShortExchange, &t.LongEntry, &t.LongExit, &t.ShortEntry, &t.ShortExit,
&t.LongPnl, &t.ShortPnl, &t.FeeEntry, &t.FeeExit, &t.NetPnl,
&t.AmountUSD, &t.ScaleCount, &t.ExitReason, &t.Convergence, &t.OpenedAt, &t.ClosedAt,
)
if err != nil {
return nil, nil, err
}
// Fetch orders
oRows, err := d.Query(`SELECT id, trade_id, leg, type, exchange, side,
price, size, fee, order_id, status, created_at
FROM orders WHERE trade_id=? ORDER BY id`, id)
if err != nil {
return nil, nil, err
}
defer oRows.Close()
var orders []OrderRecord
for oRows.Next() {
var o OrderRecord
if err := oRows.Scan(&o.ID, &o.TradeID, &o.Leg, &o.Type, &o.Exchange, &o.Side,
&o.Price, &o.Size, &o.Fee, &o.OrderID, &o.Status, &o.CreatedAt); err != nil {
return nil, nil, err
}
orders = append(orders, o)
}
return &t, orders, nil
}
func scanTrades(rows *sql.Rows) ([]TradeRecord, error) {
var trades []TradeRecord
for rows.Next() {
var t TradeRecord
if err := rows.Scan(
&t.ID, &t.Coin, &t.Direction, &t.Status, &t.EntrySpread, &t.ExitSpread,
&t.LongExchange, &t.ShortExchange, &t.LongEntry, &t.LongExit, &t.ShortEntry, &t.ShortExit,
&t.LongPnl, &t.ShortPnl, &t.FeeEntry, &t.FeeExit, &t.NetPnl,
&t.AmountUSD, &t.ScaleCount, &t.ExitReason, &t.Convergence, &t.OpenedAt, &t.ClosedAt,
); err != nil {
return nil, err
}
trades = append(trades, t)
}
return trades, rows.Err()
}
go.mod
+14 -2
View File
@@ -1,5 +1,17 @@
module exchange-monitor
go 1.23.0
go 1.25.0
require github.com/gorilla/websocket v1.5.3 // indirect
require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
golang.org/x/sys v0.42.0 // indirect
modernc.org/libc v1.72.0 // indirect
modernc.org/mathutil v1.7.1 // indirect
modernc.org/memory v1.11.0 // indirect
modernc.org/sqlite v1.50.0 // indirect
)
go.sum
+21
View File
@@ -1,2 +1,23 @@
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
modernc.org/libc v1.72.0 h1:IEu559v9a0XWjw0DPoVKtXpO2qt5NVLAnFaBbjq+n8c=
modernc.org/libc v1.72.0/go.mod h1:tTU8DL8A+XLVkEY3x5E/tO7s2Q/q42EtnNWda/L5QhQ=
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
modernc.org/sqlite v1.50.0 h1:eMowQSWLK0MeiQTdmz3lqoF5dqclujdlIKeJA11+7oM=
modernc.org/sqlite v1.50.0/go.mod h1:m0w8xhwYUVY3H6pSDwc3gkJ/irZT/0YEXwBlhaxQEew=
main.go
+10 -1
View File
@@ -11,6 +11,7 @@ import (
"syscall"
"time"
"exchange-monitor/db"
"exchange-monitor/exchange"
)
@@ -34,8 +35,16 @@ func main() {
store := NewPriceStore()
notifier := NewNotifier(cfg.TelegramBotToken, cfg.TelegramChatID)
// Initialize SQLite database
database, err := db.Open("")
if err != nil {
log.Printf("[DB] Failed to open database: %v", err)
} else {
defer database.Close()
}
// Initialize trader
trader := NewTrader(cfg)
trader := NewTrader(cfg, database)
if trader.IsConfigured() {
modeLabel := trader.ModeLabel()
log.Printf("[Trader] %s mode: automated trading ENABLED (threshold >= %.2f%%, $%.0f/trade)",
trader.go
+95 -2
View File
@@ -6,6 +6,7 @@ import (
"sync"
"time"
"exchange-monitor/db"
"exchange-monitor/exchange"
)
@@ -57,6 +58,7 @@ type Trader struct {
bitget *exchange.BitgetTrade
hyperliquid *exchange.HyperLiquidTrade
db *db.DB
mu sync.Mutex
positions map[string]*ArbPosition // coin -> position
lastTradeTime map[string]time.Time
@@ -79,20 +81,28 @@ type TradeRecord struct {
AmountUSD float64
}
func NewTrader(cfg *Config) *Trader {
func NewTrader(cfg *Config, database *db.DB) *Trader {
var bt *exchange.BitgetTrade
if cfg.BitgetAPIKey != "" {
bt = exchange.NewBitgetTrade(cfg.BitgetAPIKey, cfg.BitgetAPISecret, cfg.BitgetPassphrase)
}
hl, _ := exchange.NewHyperLiquidTrade(cfg.HLPrivateKey, cfg.HLAddress)
return &Trader{
t := &Trader{
cfg: cfg,
db: database,
bitget: bt,
hyperliquid: hl,
positions: make(map[string]*ArbPosition),
lastTradeTime: make(map[string]time.Time),
}
// Restore open positions from DB on restart
if database != nil {
t.restoreOpenPositions()
}
return t
}
func (t *Trader) IsConfigured() bool {
@@ -395,6 +405,11 @@ func (t *Trader) checkExit(pos *ArbPosition, bgP, hlP, diffPct float64, notifier
t.closedTrades = append(t.closedTrades, record)
t.mu.Unlock()
// Persist to SQLite
if t.db != nil {
go t.persistTrade(pos, diffPct, convergenceLabel, exitReason, netPnl, longPnl, shortPnl, totalFees)
}
msg := fmt.Sprintf(
"<b>[平仓]</b> %s/USDT %s\n"+
" 持仓: %s 加仓: %d次\n"+
@@ -555,3 +570,81 @@ func (t *Trader) GetClosedTrades() []TradeRecord {
copy(r, t.closedTrades)
return r
}
// persistTrade saves a completed trade to SQLite.
func (t *Trader) persistTrade(pos *ArbPosition, exitSpread float64, convergence, exitReason string, netPnl, longPnl, shortPnl, totalFees float64) {
var entrySpread, fe float64
if pos.LongLeg != nil {
entrySpread = pos.EntrySpread
}
fe = totalFees / 2 // split into entry/exit halves
now := time.Now()
dbTrade := &db.TradeRecord{
Coin: pos.Coin,
Direction: pos.Direction,
Status: "closed",
EntrySpread: &entrySpread,
ExitSpread: &exitSpread,
LongExchange: pos.LongLeg.Exchange,
ShortExchange: pos.ShortLeg.Exchange,
LongEntry: &pos.LongLeg.EntryPrice,
LongExit: &pos.LongLeg.ExitPrice,
ShortEntry: &pos.ShortLeg.EntryPrice,
ShortExit: &pos.ShortLeg.ExitPrice,
LongPnl: &longPnl,
ShortPnl: &shortPnl,
FeeEntry: &fe,
FeeExit: &fe,
NetPnl: &netPnl,
AmountUSD: pos.AmountUSD,
ScaleCount: pos.ScaleLevels,
ExitReason: &exitReason,
Convergence: &convergence,
OpenedAt: pos.StartedAt,
ClosedAt: &now,
}
if _, err := t.db.SaveTrade(dbTrade); err != nil {
log.Printf("[Trader] Failed to save trade to DB: %v", err)
}
}
// restoreOpenPositions loads open trades from DB and recreates their positions.
func (t *Trader) restoreOpenPositions() {
openTrades, err := t.db.GetOpenTrades()
if err != nil {
log.Printf("[Trader] Failed to load open trades: %v", err)
return
}
for i := range openTrades {
tr := &openTrades[i]
// Recreate position structure from DB record
pos := &ArbPosition{
Coin: tr.Coin,
Direction: tr.Direction,
AmountUSD: tr.AmountUSD,
EntrySpread: *tr.EntrySpread,
ScaleLevels: tr.ScaleCount,
StartedAt: tr.OpenedAt,
Status: "open",
}
if tr.LongEntry != nil {
pos.LongLeg = &PositionLeg{
Coin: tr.Coin, Exchange: tr.LongExchange, Side: Long,
EntryPrice: *tr.LongEntry, EntryTime: tr.OpenedAt,
}
}
if tr.ShortEntry != nil {
pos.ShortLeg = &PositionLeg{
Coin: tr.Coin, Exchange: tr.ShortExchange, Side: Short,
EntryPrice: *tr.ShortEntry, EntryTime: tr.OpenedAt,
}
}
t.positions[tr.Coin] = pos
// Prevent immediate re-trading of the same coin
t.lastTradeTime[tr.Coin] = tr.OpenedAt
}
if len(openTrades) > 0 {
log.Printf("[Trader] Restored %d open positions from DB", len(openTrades))
}
}