mirror of
https://github.com/NoFxAiOS/nofx.git
synced 2026-07-11 14:56:57 +08:00
feat: add debate arena and fix multiple issues
- Add AI debate arena for multi-AI trading decisions - Fix debate consensus calculation and display - Fix vote parsing to support both <decision> and <final_vote> tags - Fix JSON field name compatibility (stop_loss/stop_loss_pct) - Fix symbol validation to prevent AI hallucinating invalid symbols - Fix Bybit position side display (was uppercase, now lowercase for consistency) - Fix NOFX logo navigation to home page - Add detailed logging for debugging trade execution
This commit is contained in:
tinkle-community
634
api/debate.go
Normal file
634
api/debate.go
Normal file
@@ -0,0 +1,634 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
|
||||
"nofx/debate"
|
||||
"nofx/logger"
|
||||
"nofx/pool"
|
||||
"nofx/store"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// DebateHandler handles debate-related API requests
|
||||
type DebateHandler struct {
|
||||
debateStore *store.DebateStore
|
||||
strategyStore *store.StrategyStore
|
||||
aiModelStore *store.AIModelStore
|
||||
engine *debate.DebateEngine
|
||||
|
||||
// Trader manager for execution
|
||||
traderManager DebateTraderManager
|
||||
|
||||
// SSE subscribers
|
||||
subscribers map[string]map[chan []byte]bool // sessionID -> channels
|
||||
subscribersMu sync.RWMutex
|
||||
}
|
||||
|
||||
// DebateTraderManager interface for getting trader executors
|
||||
type DebateTraderManager interface {
|
||||
GetTraderExecutor(traderID string) (debate.TraderExecutor, error)
|
||||
}
|
||||
|
||||
// NewDebateHandler creates a new DebateHandler
|
||||
func NewDebateHandler(debateStore *store.DebateStore, strategyStore *store.StrategyStore, aiModelStore *store.AIModelStore) *DebateHandler {
|
||||
handler := &DebateHandler{
|
||||
debateStore: debateStore,
|
||||
strategyStore: strategyStore,
|
||||
aiModelStore: aiModelStore,
|
||||
subscribers: make(map[string]map[chan []byte]bool),
|
||||
}
|
||||
|
||||
// Create debate engine with event callbacks
|
||||
handler.engine = debate.NewDebateEngine(debateStore, strategyStore, aiModelStore)
|
||||
handler.engine.OnRoundStart = handler.broadcastRoundStart
|
||||
handler.engine.OnMessage = handler.broadcastMessage
|
||||
handler.engine.OnRoundEnd = handler.broadcastRoundEnd
|
||||
handler.engine.OnVote = handler.broadcastVote
|
||||
handler.engine.OnConsensus = handler.broadcastConsensus
|
||||
handler.engine.OnError = handler.broadcastError
|
||||
|
||||
return handler
|
||||
}
|
||||
|
||||
// CreateDebateRequest represents a request to create a new debate
|
||||
type CreateDebateRequest struct {
|
||||
Name string `json:"name" binding:"required"`
|
||||
StrategyID string `json:"strategy_id" binding:"required"`
|
||||
Symbol string `json:"symbol"` // Optional: auto-selected based on strategy if empty
|
||||
MaxRounds int `json:"max_rounds"`
|
||||
IntervalMinutes int `json:"interval_minutes"`
|
||||
PromptVariant string `json:"prompt_variant"`
|
||||
AutoExecute bool `json:"auto_execute"`
|
||||
TraderID string `json:"trader_id"`
|
||||
Participants []ParticipantConfig `json:"participants" binding:"required,min=2"`
|
||||
}
|
||||
|
||||
// ParticipantConfig represents a participant configuration
|
||||
type ParticipantConfig struct {
|
||||
AIModelID string `json:"ai_model_id" binding:"required"`
|
||||
Personality string `json:"personality" binding:"required"`
|
||||
}
|
||||
|
||||
// HandleListDebates lists all debates for a user
|
||||
func (h *DebateHandler) HandleListDebates(c *gin.Context) {
|
||||
userID := c.GetString("user_id")
|
||||
if userID == "" {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
sessions, err := h.debateStore.GetSessionsByUser(userID)
|
||||
if err != nil {
|
||||
logger.Errorf("Failed to get debates for user %s: %v", userID, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get debates"})
|
||||
return
|
||||
}
|
||||
|
||||
// Return empty array instead of null
|
||||
if sessions == nil {
|
||||
sessions = []*store.DebateSession{}
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, sessions)
|
||||
}
|
||||
|
||||
// HandleGetDebate gets a specific debate with all details
|
||||
func (h *DebateHandler) HandleGetDebate(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSessionWithDetails(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
// Check ownership
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, session)
|
||||
}
|
||||
|
||||
// HandleCreateDebate creates a new debate
|
||||
func (h *DebateHandler) HandleCreateDebate(c *gin.Context) {
|
||||
userID := c.GetString("user_id")
|
||||
if userID == "" {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
var req CreateDebateRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// Validate strategy exists
|
||||
strategy, err := h.strategyStore.Get(userID, req.StrategyID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "strategy not found"})
|
||||
return
|
||||
}
|
||||
|
||||
// Validate strategy belongs to user or is default
|
||||
if strategy.UserID != userID && !strategy.IsDefault {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "strategy access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
// Auto-select symbol based on strategy if not provided
|
||||
if req.Symbol == "" {
|
||||
req.Symbol = "BTCUSDT" // default fallback
|
||||
if strategyConfig, err := strategy.ParseConfig(); err == nil {
|
||||
coinSource := strategyConfig.CoinSource
|
||||
switch coinSource.SourceType {
|
||||
case "static":
|
||||
if len(coinSource.StaticCoins) > 0 {
|
||||
req.Symbol = coinSource.StaticCoins[0]
|
||||
}
|
||||
case "coinpool":
|
||||
// Fetch from coin pool API
|
||||
if coinSource.CoinPoolAPIURL != "" {
|
||||
pool.SetCoinPoolAPI(coinSource.CoinPoolAPIURL)
|
||||
}
|
||||
if coins, err := pool.GetTopRatedCoins(1); err == nil && len(coins) > 0 {
|
||||
req.Symbol = coins[0]
|
||||
logger.Infof("Fetched coin from pool API: %s", req.Symbol)
|
||||
}
|
||||
case "oi_top":
|
||||
// Fetch from OI top API
|
||||
if coinSource.OITopAPIURL != "" {
|
||||
pool.SetOITopAPI(coinSource.OITopAPIURL)
|
||||
}
|
||||
if coins, err := pool.GetOITopSymbols(); err == nil && len(coins) > 0 {
|
||||
req.Symbol = coins[0]
|
||||
logger.Infof("Fetched coin from OI Top API: %s", req.Symbol)
|
||||
}
|
||||
case "mixed":
|
||||
// Try coin pool first, then OI top
|
||||
if coinSource.UseCoinPool && coinSource.CoinPoolAPIURL != "" {
|
||||
pool.SetCoinPoolAPI(coinSource.CoinPoolAPIURL)
|
||||
if coins, err := pool.GetTopRatedCoins(1); err == nil && len(coins) > 0 {
|
||||
req.Symbol = coins[0]
|
||||
logger.Infof("Fetched coin from pool API (mixed): %s", req.Symbol)
|
||||
}
|
||||
} else if coinSource.UseOITop && coinSource.OITopAPIURL != "" {
|
||||
pool.SetOITopAPI(coinSource.OITopAPIURL)
|
||||
if coins, err := pool.GetOITopSymbols(); err == nil && len(coins) > 0 {
|
||||
req.Symbol = coins[0]
|
||||
logger.Infof("Fetched coin from OI Top API (mixed): %s", req.Symbol)
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.Infof("Auto-selected symbol %s for debate based on strategy %s (source_type=%s)",
|
||||
req.Symbol, strategy.Name, coinSource.SourceType)
|
||||
}
|
||||
}
|
||||
|
||||
// Set defaults
|
||||
if req.MaxRounds <= 0 || req.MaxRounds > 5 {
|
||||
req.MaxRounds = 3
|
||||
}
|
||||
if req.IntervalMinutes <= 0 {
|
||||
req.IntervalMinutes = 5
|
||||
}
|
||||
if req.PromptVariant == "" {
|
||||
req.PromptVariant = "balanced"
|
||||
}
|
||||
|
||||
// Create session
|
||||
session := &store.DebateSession{
|
||||
UserID: userID,
|
||||
Name: req.Name,
|
||||
StrategyID: req.StrategyID,
|
||||
Symbol: req.Symbol,
|
||||
MaxRounds: req.MaxRounds,
|
||||
IntervalMinutes: req.IntervalMinutes,
|
||||
PromptVariant: req.PromptVariant,
|
||||
AutoExecute: req.AutoExecute,
|
||||
TraderID: req.TraderID,
|
||||
}
|
||||
|
||||
if err := h.debateStore.CreateSession(session); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create debate"})
|
||||
return
|
||||
}
|
||||
|
||||
// Add participants
|
||||
for i, p := range req.Participants {
|
||||
// Validate AI model exists and belongs to user
|
||||
aiModel, err := h.aiModelStore.GetByID(p.AIModelID)
|
||||
if err != nil {
|
||||
logger.Warnf("AI model not found: %s", p.AIModelID)
|
||||
continue
|
||||
}
|
||||
if aiModel.UserID != userID {
|
||||
logger.Warnf("AI model %s does not belong to user", p.AIModelID)
|
||||
continue
|
||||
}
|
||||
|
||||
// Validate personality
|
||||
personality := store.DebatePersonality(p.Personality)
|
||||
if _, ok := store.PersonalityColors[personality]; !ok {
|
||||
personality = store.PersonalityAnalyst
|
||||
}
|
||||
|
||||
participant := &store.DebateParticipant{
|
||||
SessionID: session.ID,
|
||||
AIModelID: p.AIModelID,
|
||||
AIModelName: aiModel.Name,
|
||||
Provider: aiModel.Provider,
|
||||
Personality: personality,
|
||||
Color: store.PersonalityColors[personality],
|
||||
SpeakOrder: i,
|
||||
}
|
||||
|
||||
if err := h.debateStore.AddParticipant(participant); err != nil {
|
||||
logger.Errorf("Failed to add participant: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Get full session with participants
|
||||
fullSession, _ := h.debateStore.GetSessionWithDetails(session.ID)
|
||||
|
||||
c.JSON(http.StatusCreated, fullSession)
|
||||
}
|
||||
|
||||
// HandleStartDebate starts a debate
|
||||
func (h *DebateHandler) HandleStartDebate(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.Status != store.DebateStatusPending {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "debate is not in pending status"})
|
||||
return
|
||||
}
|
||||
|
||||
// Start debate asynchronously
|
||||
if err := h.engine.StartDebate(debateID); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"message": "debate started", "id": debateID})
|
||||
}
|
||||
|
||||
// HandleCancelDebate cancels a running debate
|
||||
func (h *DebateHandler) HandleCancelDebate(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.engine.CancelDebate(debateID); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"message": "debate cancelled"})
|
||||
}
|
||||
|
||||
// HandleDeleteDebate deletes a debate
|
||||
func (h *DebateHandler) HandleDeleteDebate(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
// Don't allow deleting running debates
|
||||
if session.Status == store.DebateStatusRunning || session.Status == store.DebateStatusVoting {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "cannot delete running debate"})
|
||||
return
|
||||
}
|
||||
|
||||
if err := h.debateStore.DeleteSession(debateID); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete debate"})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"message": "debate deleted"})
|
||||
}
|
||||
|
||||
// HandleGetMessages gets all messages for a debate
|
||||
func (h *DebateHandler) HandleGetMessages(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
messages, err := h.debateStore.GetMessages(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get messages"})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, messages)
|
||||
}
|
||||
|
||||
// HandleGetVotes gets all votes for a debate
|
||||
func (h *DebateHandler) HandleGetVotes(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
votes, err := h.debateStore.GetVotes(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get votes"})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, votes)
|
||||
}
|
||||
|
||||
// HandleDebateStream handles SSE streaming for live debate updates
|
||||
func (h *DebateHandler) HandleDebateStream(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
// Set SSE headers
|
||||
c.Header("Content-Type", "text/event-stream")
|
||||
c.Header("Cache-Control", "no-cache")
|
||||
c.Header("Connection", "keep-alive")
|
||||
c.Header("Transfer-Encoding", "chunked")
|
||||
|
||||
// Create channel for this subscriber
|
||||
ch := make(chan []byte, 100)
|
||||
h.addSubscriber(debateID, ch)
|
||||
defer h.removeSubscriber(debateID, ch)
|
||||
|
||||
// Send initial state
|
||||
initialState, _ := h.debateStore.GetSessionWithDetails(debateID)
|
||||
initialData, _ := json.Marshal(map[string]interface{}{
|
||||
"event": "initial",
|
||||
"data": initialState,
|
||||
})
|
||||
c.Writer.Write([]byte(fmt.Sprintf("event: initial\ndata: %s\n\n", initialData)))
|
||||
c.Writer.Flush()
|
||||
|
||||
// Stream updates
|
||||
clientGone := c.Request.Context().Done()
|
||||
for {
|
||||
select {
|
||||
case <-clientGone:
|
||||
return
|
||||
case msg := <-ch:
|
||||
c.Writer.Write(msg)
|
||||
c.Writer.Flush()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// SetTraderManager sets the trader manager for executing trades
|
||||
func (h *DebateHandler) SetTraderManager(tm DebateTraderManager) {
|
||||
h.traderManager = tm
|
||||
}
|
||||
|
||||
// ExecuteDebateRequest represents a request to execute a debate's consensus
|
||||
type ExecuteDebateRequest struct {
|
||||
TraderID string `json:"trader_id" binding:"required"`
|
||||
}
|
||||
|
||||
// HandleExecuteDebate executes the consensus decision from a completed debate
|
||||
func (h *DebateHandler) HandleExecuteDebate(c *gin.Context) {
|
||||
debateID := c.Param("id")
|
||||
userID := c.GetString("user_id")
|
||||
|
||||
// Check trader manager is available
|
||||
if h.traderManager == nil {
|
||||
c.JSON(http.StatusServiceUnavailable, gin.H{"error": "trading service not available"})
|
||||
return
|
||||
}
|
||||
|
||||
// Get debate session
|
||||
session, err := h.debateStore.GetSession(debateID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "debate not found"})
|
||||
return
|
||||
}
|
||||
|
||||
// Check ownership
|
||||
if session.UserID != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "access denied"})
|
||||
return
|
||||
}
|
||||
|
||||
// Check status
|
||||
if session.Status != store.DebateStatusCompleted {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "debate is not completed"})
|
||||
return
|
||||
}
|
||||
|
||||
// Parse request
|
||||
var req ExecuteDebateRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// Get trader executor
|
||||
executor, err := h.traderManager.GetTraderExecutor(req.TraderID)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("trader not available: %v", err)})
|
||||
return
|
||||
}
|
||||
|
||||
// Execute consensus
|
||||
if err := h.engine.ExecuteConsensus(debateID, executor); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// Get updated session
|
||||
updatedSession, _ := h.debateStore.GetSessionWithDetails(debateID)
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"message": "consensus executed successfully",
|
||||
"session": updatedSession,
|
||||
})
|
||||
}
|
||||
|
||||
// GetPersonalities returns available AI personalities
|
||||
func (h *DebateHandler) HandleGetPersonalities(c *gin.Context) {
|
||||
personalities := []map[string]interface{}{
|
||||
{
|
||||
"id": "bull",
|
||||
"name": "Aggressive Bull",
|
||||
"emoji": "🐂",
|
||||
"color": store.PersonalityColors[store.PersonalityBull],
|
||||
"description": "Looks for long opportunities, optimistic about market",
|
||||
},
|
||||
{
|
||||
"id": "bear",
|
||||
"name": "Cautious Bear",
|
||||
"emoji": "🐻",
|
||||
"color": store.PersonalityColors[store.PersonalityBear],
|
||||
"description": "Skeptical, focuses on risks and short opportunities",
|
||||
},
|
||||
{
|
||||
"id": "analyst",
|
||||
"name": "Data Analyst",
|
||||
"emoji": "📊",
|
||||
"color": store.PersonalityColors[store.PersonalityAnalyst],
|
||||
"description": "Pure technical analysis, neutral and data-driven",
|
||||
},
|
||||
{
|
||||
"id": "contrarian",
|
||||
"name": "Contrarian",
|
||||
"emoji": "🔄",
|
||||
"color": store.PersonalityColors[store.PersonalityContrarian],
|
||||
"description": "Challenges majority opinion, looks for overlooked opportunities",
|
||||
},
|
||||
{
|
||||
"id": "risk_manager",
|
||||
"name": "Risk Manager",
|
||||
"emoji": "🛡️",
|
||||
"color": store.PersonalityColors[store.PersonalityRiskManager],
|
||||
"description": "Focuses on position sizing, stop losses, and risk control",
|
||||
},
|
||||
}
|
||||
c.JSON(http.StatusOK, personalities)
|
||||
}
|
||||
|
||||
// SSE broadcast helpers
|
||||
func (h *DebateHandler) addSubscriber(sessionID string, ch chan []byte) {
|
||||
h.subscribersMu.Lock()
|
||||
defer h.subscribersMu.Unlock()
|
||||
|
||||
if h.subscribers[sessionID] == nil {
|
||||
h.subscribers[sessionID] = make(map[chan []byte]bool)
|
||||
}
|
||||
h.subscribers[sessionID][ch] = true
|
||||
}
|
||||
|
||||
func (h *DebateHandler) removeSubscriber(sessionID string, ch chan []byte) {
|
||||
h.subscribersMu.Lock()
|
||||
defer h.subscribersMu.Unlock()
|
||||
|
||||
if h.subscribers[sessionID] != nil {
|
||||
delete(h.subscribers[sessionID], ch)
|
||||
close(ch)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcast(sessionID string, event string, data interface{}) {
|
||||
h.subscribersMu.RLock()
|
||||
defer h.subscribersMu.RUnlock()
|
||||
|
||||
subs := h.subscribers[sessionID]
|
||||
if subs == nil {
|
||||
return
|
||||
}
|
||||
|
||||
jsonData, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
msg := []byte(fmt.Sprintf("event: %s\ndata: %s\n\n", event, jsonData))
|
||||
for ch := range subs {
|
||||
select {
|
||||
case ch <- msg:
|
||||
default:
|
||||
// Channel full, skip
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcastRoundStart(sessionID string, round int) {
|
||||
h.broadcast(sessionID, "round_start", map[string]interface{}{
|
||||
"round": round,
|
||||
"status": "running",
|
||||
})
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcastMessage(sessionID string, msg *store.DebateMessage) {
|
||||
h.broadcast(sessionID, "message", msg)
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcastRoundEnd(sessionID string, round int) {
|
||||
h.broadcast(sessionID, "round_end", map[string]interface{}{
|
||||
"round": round,
|
||||
"status": "completed",
|
||||
})
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcastVote(sessionID string, vote *store.DebateVote) {
|
||||
h.broadcast(sessionID, "vote", vote)
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcastConsensus(sessionID string, decision *store.DebateDecision) {
|
||||
h.broadcast(sessionID, "consensus", decision)
|
||||
}
|
||||
|
||||
func (h *DebateHandler) broadcastError(sessionID string, err error) {
|
||||
h.broadcast(sessionID, "error", map[string]interface{}{
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
@@ -28,6 +28,7 @@ type Server struct {
|
||||
store *store.Store
|
||||
cryptoHandler *CryptoHandler
|
||||
backtestManager *backtest.Manager
|
||||
debateHandler *DebateHandler
|
||||
httpServer *http.Server
|
||||
port int
|
||||
}
|
||||
@@ -45,12 +46,21 @@ func NewServer(traderManager *manager.TraderManager, st *store.Store, cryptoServ
|
||||
// Create crypto handler
|
||||
cryptoHandler := NewCryptoHandler(cryptoService)
|
||||
|
||||
// Create debate store and handler
|
||||
debateStore := store.NewDebateStore(st.DB())
|
||||
if err := debateStore.InitSchema(); err != nil {
|
||||
logger.Errorf("Failed to initialize debate schema: %v", err)
|
||||
}
|
||||
debateHandler := NewDebateHandler(debateStore, st.Strategy(), st.AIModel())
|
||||
debateHandler.SetTraderManager(traderManager)
|
||||
|
||||
s := &Server{
|
||||
router: router,
|
||||
traderManager: traderManager,
|
||||
store: st,
|
||||
cryptoHandler: cryptoHandler,
|
||||
backtestManager: backtestManager,
|
||||
debateHandler: debateHandler,
|
||||
port: port,
|
||||
}
|
||||
|
||||
@@ -157,6 +167,19 @@ func (s *Server) setupRoutes() {
|
||||
protected.POST("/strategies/:id/activate", s.handleActivateStrategy)
|
||||
protected.POST("/strategies/:id/duplicate", s.handleDuplicateStrategy)
|
||||
|
||||
// Debate Arena
|
||||
protected.GET("/debates", s.debateHandler.HandleListDebates)
|
||||
protected.GET("/debates/personalities", s.debateHandler.HandleGetPersonalities)
|
||||
protected.GET("/debates/:id", s.debateHandler.HandleGetDebate)
|
||||
protected.POST("/debates", s.debateHandler.HandleCreateDebate)
|
||||
protected.POST("/debates/:id/start", s.debateHandler.HandleStartDebate)
|
||||
protected.POST("/debates/:id/cancel", s.debateHandler.HandleCancelDebate)
|
||||
protected.POST("/debates/:id/execute", s.debateHandler.HandleExecuteDebate)
|
||||
protected.DELETE("/debates/:id", s.debateHandler.HandleDeleteDebate)
|
||||
protected.GET("/debates/:id/messages", s.debateHandler.HandleGetMessages)
|
||||
protected.GET("/debates/:id/votes", s.debateHandler.HandleGetVotes)
|
||||
protected.GET("/debates/:id/stream", s.debateHandler.HandleDebateStream)
|
||||
|
||||
// Data for specified trader (using query parameter ?trader_id=xxx)
|
||||
protected.GET("/status", s.handleStatus)
|
||||
protected.GET("/account", s.handleAccount)
|
||||
|
||||
Reference in New Issue
Block a user