exchange-monitor-go/main.go

package main

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"log"
	"math/rand"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"exchange-monitor/db"
	"exchange-monitor/exchange"
)

func main() {
	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)

	// Set up multi-writer: stdout + log file
	logPath := os.ExpandEnv("$HOME/Project/exchange-monitor-go/exchange-monitor.log")
	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
	if err == nil {
		multi := io.MultiWriter(os.Stdout, logFile)
		log.SetOutput(multi)
	} else {
		log.SetOutput(os.Stdout)
	}
	log.Println("[Exchange Monitor] Starting...")

	loadDotEnv()
	cfg := LoadConfig()

	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, database)

	// Initialize dashboard (web server + SSE)
	dashboard := NewDashboard(store, trader, database, ":8888")
	go dashboard.Run()

	// Spread window tracker — measures how long spreads stay above threshold
	spreadTracker := NewSpreadWindowTracker()

	// P3-4: wire real-time trade event broadcast
	trader.OnTradeEvent = dashboard.BroadcastEvent
	if trader.IsConfigured() {
		modeLabel := trader.ModeLabel()
		log.Printf("[Trader] %s mode: automated trading ENABLED (threshold >= %.2f%%, $%.0f/trade)",
			modeLabel, cfg.TradeThreshold, cfg.TradeAmountUSD)
		if cfg.TestMode {
			log.Printf("[Trader] Using mock orders with %.3f%% slippage per leg", cfg.MockSlippagePct)
		}
		log.Printf("[Trader] Bitget+HL: BG->HL / HL->BG only")
	} else {
		log.Printf("[Trader] Automated trading DISABLED (set TRADE_ENABLED=1 or TEST_MODE=true in .env)")
	}

	// Context for graceful shutdown — replaces shared sigCh (B#1)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, os.Interrupt, syscall.SIGUSR1)

	// Collect symbols
	var bnSymbols, bgSymbols, hlSymbols, dydxSymbols []string
	for _, c := range TrackedCoins {
		bnSymbols = append(bnSymbols, c.BN)
		bgSymbols = append(bgSymbols, c.BG)
		hlSymbols = append(hlSymbols, c.HL)
		dydxSymbols = append(dydxSymbols, c.HL)
	}

	// Start all exchange WS connections
	startExchange := func(name string, runner func(func(string, float64, float64, float64)) error) {
		go func() {
			for {
				err := runner(func(coin string, price, bid, ask float64) {
					store.SetWithSpread(coin, name, price, bid, ask)
					dashboard.RecordPrice(coin, name, price)
					dashboard.RecordConnStatus(name) // P3-5
				})
				log.Printf("[%s] WS error: %v (reconnecting...)", name, err)
				select {
				case <-ctx.Done():
					return
				case <-time.After(3 * time.Second):
				}
			}
		}()
	}

	startExchange("Binance", exchange.NewBinanceWS(bnSymbols).Run)
	startExchange("HyperLiquid", exchange.NewHyperLiquidWS(hlSymbols).Run)
	startExchange("Bitget", exchange.NewBitgetWS(bgSymbols).Run)
	startExchange("dYdX", exchange.NewDydxWS(dydxSymbols).Run) // B#9: use dedicated symbol list

	log.Println("[Monitor] Waiting for initial data...")
	time.Sleep(10 * time.Second)

	// Main loop
	lastHour := -1

	// Random jitter 50-250ms to avoid lock-step with exchange push cycles
	jitterMin, jitterMax := 50, 250
	randInterval := func() time.Duration {
		return time.Duration(jitterMin+rand.Intn(jitterMax-jitterMin+1)) * time.Millisecond
	}
	scannerTick := time.NewTimer(randInterval())
	statusTick := time.NewTicker(30 * time.Second)

	log.Printf("[Monitor] Scanner running every %d-%dms (random jitter)", jitterMin, jitterMax)

	runLoop := true
	for runLoop {
		select {
		case sig := <-sigCh:
			if sig == syscall.SIGUSR1 {
				// Dump stats on request
				converged, diverged, flat, total := trader.GetClosedStats()
				stats := fmt.Sprintf("=== 收敛统计 ===  %s\n", time.Now().Format("2006-01-02 15:04"))
				stats += fmt.Sprintf("  总交易数: %d\n", total)
				stats += fmt.Sprintf("  价差收敛: %d\n", converged)
				stats += fmt.Sprintf("  价差持平: %d\n", flat)
				stats += fmt.Sprintf("  价差发散: %d\n", diverged)
				if total > 0 {
					stats += fmt.Sprintf("  收敛率: %.1f%%\n", float64(converged)/float64(total)*100)
				}
				log.Printf("[Monitor] SIGUSR1 received — wrote stats to trade_stats.txt")
				statsPath := os.ExpandEnv("$HOME/Project/exchange-monitor-go/trade_stats.txt")
				os.WriteFile(statsPath, []byte(stats), 0644)
				continue
			}
			log.Println("[Monitor] Shutting down...")
			cancel() // B#1: cancel context to stop all WS goroutines
			runLoop = false

		case <-statusTick.C:
			snap := store.GetAll()
			count := 0
			for _, exMap := range snap {
				count += len(exMap)
			}
			log.Printf("[Status] %d prices / %d coins connected", count, len(snap))

			// Show open positions (read from decoupled snapshot)
			if positions := trader.ReadSnapshot(); len(positions) > 0 {
				for _, pos := range positions {
					log.Printf("  [Position] %s %s open %d scales $%.0f since %s",
						pos.Coin, pos.Direction, pos.ScaleLevels, pos.AmountUSD,
						time.Since(pos.StartedAt).Round(time.Second).String())
				}
			}

		case <-scannerTick.C:
			now := time.Now()
			t0 := now

			// Tick the trader (monitor open positions for exit)
			trader.Tick(store, notifier)
			trader.RefreshSnapshot() // decoupled snapshot for display
			t1 := time.Now()

			// Scan for arbitrage entries using maker fees (limit orders)
			makerOpps := ScanBGHL(store)
			dashboard.UpdateScan(makerOpps)
			t2 := time.Now()

			// Track spread window durations (how long each opportunity stays alive)
			snap := store.GetAll()
			spreadTracker.Tick(snap, cfg.TradeThreshold)

			for _, opp := range makerOpps {
				if opp.NetProfit < cfg.ArbThreshold {
					continue
				}
				if trader.TryEntry(opp, store, notifier) {
					log.Printf("[Trader] %s: entry initiated for %.4f%%", opp.Coin, opp.NetProfit)
				}
			}
			t3 := time.Now()

			// Profile: warn if any step is slow
			tickDur := t3.Sub(t0)
			tickMs := tickDur.Milliseconds()
			if tickMs > 100 || t1.Sub(t0) > 50*time.Millisecond || t2.Sub(t1) > 50*time.Millisecond || t3.Sub(t2) > 50*time.Millisecond {
				log.Printf("[Profile] tick=%dms trader=%dms scan=%dms entry=%dms",
					tickMs, t1.Sub(t0).Milliseconds(), t2.Sub(t1).Milliseconds(), t3.Sub(t2).Milliseconds())
			}

			// Hourly trade summary — use hour-based tracking (wider window than second-granularity)
			hour := now.Hour()
			if hour != lastHour && now.Minute() < 1 {
				positions := trader.ReadSnapshot()
				notifier.SendTradeSummary(positions, now.Format("2006-01-02 15:04"))
				lastHour = hour
			}

			scannerTick.Reset(randInterval())
		}
	}

	log.Println("[Monitor] Stopped.")
}

func loadDotEnv() {
	envPath := os.ExpandEnv("$HOME/Project/exchange-monitor-go/.env")
	if _, err := os.Stat(envPath); err != nil {
		return
	}
	data, err := os.ReadFile(envPath)
	if err != nil {
		return
	}
	for _, line := range bytes.Split(data, []byte("\n")) {
		line = bytes.TrimSpace(line)
		if len(line) == 0 || line[0] == '#' {
			continue
		}
		parts := bytes.SplitN(line, []byte("="), 2)
		if len(parts) != 2 {
			continue
		}
		key := string(bytes.TrimSpace(parts[0]))
		val := string(bytes.TrimSpace(parts[1]))
		// Strip inline comments
		if idx := strings.Index(val, "#"); idx >= 0 {
			val = strings.TrimSpace(val[:idx])
		}
		if os.Getenv(key) == "" {
			os.Setenv(key, val)
		}
	}
}