diff --git a/cmd/longbridge-fs/main.go b/cmd/longbridge-fs/main.go index bfdf5fc..6ff9cf5 100644 --- a/cmd/longbridge-fs/main.go +++ b/cmd/longbridge-fs/main.go @@ -483,6 +483,12 @@ func runController(root string, interval time.Duration, credFile string, mock bo log.Printf("🚀 Controller started (interval=%s, compact-after=%d)", interval, compactAfter) + // Phase 4: Initialize algorithm scheduler + bcPath := filepath.Join(root, "trade", "beancount.txt") + algoScheduler := broker.NewAlgoScheduler(bcPath, tc, useMock) + defer algoScheduler.Shutdown() + log.Println("✓ Algorithm scheduler initialized") + executedCount := 0 ticker := time.NewTicker(interval) defer ticker.Stop() @@ -503,7 +509,7 @@ func runController(root string, interval time.Duration, credFile string, mock bo } // Process trade ledger - n, err := broker.ProcessLedger(ctx, tc, root, useMock) + n, err := broker.ProcessLedgerWithScheduler(ctx, tc, root, useMock, algoScheduler) if err != nil { log.Printf("❌ Order processing failed: %v", err) } else if n > 0 && verbose { @@ -511,6 +517,9 @@ func runController(root string, interval time.Duration, credFile string, mock bo } executedCount += n + // Cleanup completed algo tasks periodically + algoScheduler.CleanupCompleted() + // Refresh account state (only with real API) if tc != nil { if err := account.RefreshState(ctx, tc, root); err != nil { diff --git a/internal/audit/audit.go b/internal/audit/audit.go index 8a3dde7..1ee99ac 100644 --- a/internal/audit/audit.go +++ b/internal/audit/audit.go @@ -100,12 +100,12 @@ func (l *Logger) SetRiskStep(checked, passed, rejected int, rejections []Rejecti } // SetExecutionStep sets the execution step data -func (l *Logger) SetExecutionStep(submitted, executions, rejections int) { +func (l *Logger) SetExecutionStep(submitted, executions, rejections, algoTasksActive int) { l.log.Steps.Execution = ExecutionStep{ OrdersSubmitted: submitted, Executions: executions, Rejections: rejections, - AlgoTasksActive: 0, + AlgoTasksActive: algoTasksActive, } } diff --git a/internal/broker/algo.go b/internal/broker/algo.go new file mode 100644 index 0000000..24efde7 --- /dev/null +++ b/internal/broker/algo.go @@ -0,0 +1,210 @@ +package broker + +import ( + "context" + "fmt" + "log" + "strconv" + "sync" + "time" + + "longbridge-fs/internal/ledger" + "longbridge-fs/internal/model" + + "github.com/longbridge/openapi-go/trade" +) + +// AlgoTask represents an active algorithmic order execution task +type AlgoTask struct { + IntentID string + Order model.ParsedOrder + TotalQty int64 + SliceQty int64 + TotalSlices int + CurrentSlice int + Interval time.Duration + CreatedAt time.Time + LastSliceAt time.Time + Done bool + Cancel context.CancelFunc + mu sync.Mutex +} + +// AlgoScheduler manages active algorithmic order execution tasks +type AlgoScheduler struct { + tasks map[string]*AlgoTask + mu sync.RWMutex + bcPath string + tc *trade.TradeContext + useMock bool + ctx context.Context + cancelFunc context.CancelFunc +} + +// NewAlgoScheduler creates a new algorithm scheduler +func NewAlgoScheduler(bcPath string, tc *trade.TradeContext, useMock bool) *AlgoScheduler { + ctx, cancel := context.WithCancel(context.Background()) + return &AlgoScheduler{ + tasks: make(map[string]*AlgoTask), + bcPath: bcPath, + tc: tc, + useMock: useMock, + ctx: ctx, + cancelFunc: cancel, + } +} + +// CreateTask creates and starts an algorithmic order task +func (s *AlgoScheduler) CreateTask(o model.ParsedOrder) error { + s.mu.Lock() + defer s.mu.Unlock() + + // Check if task already exists + if _, exists := s.tasks[o.IntentID]; exists { + return fmt.Errorf("algo task already exists for intent_id: %s", o.IntentID) + } + + // Parse total quantity + totalQty, err := strconv.ParseInt(o.Qty, 10, 64) + if err != nil { + return fmt.Errorf("invalid qty %q: %w", o.Qty, err) + } + + // Validate algo parameters + if o.AlgoSlices <= 0 { + return fmt.Errorf("algo_slices must be > 0, got %d", o.AlgoSlices) + } + + // Calculate slice quantity + sliceQty := totalQty / int64(o.AlgoSlices) + if sliceQty <= 0 { + return fmt.Errorf("slice qty too small: total=%d slices=%d", totalQty, o.AlgoSlices) + } + + // Parse duration for TWAP + var interval time.Duration + if o.Algo == "TWAP" { + if o.AlgoDuration == "" { + return fmt.Errorf("TWAP requires algo_duration") + } + interval, err = time.ParseDuration(o.AlgoDuration) + if err != nil { + return fmt.Errorf("invalid algo_duration %q: %w", o.AlgoDuration, err) + } + // Calculate interval between slices + interval = interval / time.Duration(o.AlgoSlices) + } + + // Create task context + taskCtx, taskCancel := context.WithCancel(s.ctx) + + task := &AlgoTask{ + IntentID: o.IntentID, + Order: o, + TotalQty: totalQty, + SliceQty: sliceQty, + TotalSlices: o.AlgoSlices, + CurrentSlice: 0, + Interval: interval, + CreatedAt: time.Now(), + Done: false, + Cancel: taskCancel, + } + + s.tasks[o.IntentID] = task + + // Start execution based on algo type + switch o.Algo { + case "TWAP": + go s.executeTWAP(taskCtx, task) + case "ICEBERG": + go s.executeICEBERG(taskCtx, task) + default: + taskCancel() + delete(s.tasks, o.IntentID) + return fmt.Errorf("unsupported algo type: %s", o.Algo) + } + + log.Printf("Created %s task for intent=%s: %d slices of %d shares", o.Algo, o.IntentID, o.AlgoSlices, sliceQty) + return nil +} + +// GetActiveCount returns the number of active algo tasks +func (s *AlgoScheduler) GetActiveCount() int { + s.mu.RLock() + defer s.mu.RUnlock() + count := 0 + for _, task := range s.tasks { + if !task.Done { + count++ + } + } + return count +} + +// CleanupCompleted removes completed tasks from the scheduler +func (s *AlgoScheduler) CleanupCompleted() { + s.mu.Lock() + defer s.mu.Unlock() + + for intentID, task := range s.tasks { + if task.Done { + delete(s.tasks, intentID) + } + } +} + +// Shutdown gracefully stops all active tasks +func (s *AlgoScheduler) Shutdown() { + s.mu.Lock() + defer s.mu.Unlock() + + log.Printf("Shutting down AlgoScheduler with %d active tasks", len(s.tasks)) + s.cancelFunc() + + // Wait a bit for goroutines to finish + time.Sleep(100 * time.Millisecond) +} + +// executeSlice submits a single slice of an algorithmic order +func (s *AlgoScheduler) executeSlice(task *AlgoTask, sliceNum int, qty int64) error { + task.mu.Lock() + order := task.Order + intentID := task.IntentID + task.mu.Unlock() + + sym := ledger.FullSymbol(order.Symbol, order.Market) + sliceLabel := fmt.Sprintf("%d/%d", sliceNum, task.TotalSlices) + + var orderID string + var price string + var err error + + if s.useMock { + orderID, price = ExecuteOrderMock(order) + } else if s.tc != nil { + // Create slice order with adjusted quantity + sliceOrder := order + sliceOrder.Qty = strconv.FormatInt(qty, 10) + orderID, err = ExecuteOrder(context.Background(), s.tc, sliceOrder) + if err != nil { + log.Printf("Algo slice execution failed: intent=%s slice=%s err=%v", intentID, sliceLabel, err) + return err + } + price = order.Price + } + + // Append execution with slice metadata + AppendSliceExecution(s.bcPath, intentID, orderID, sym, order.Side, price, strconv.FormatInt(qty, 10), sliceLabel, order.Algo) + log.Printf("Algo slice executed: intent=%s slice=%s order_id=%s", intentID, sliceLabel, orderID) + + return nil +} + +// AppendSliceExecution appends an EXECUTION entry with slice metadata +func AppendSliceExecution(bcPath, intentID, orderID, symbol, side, price, qty, slice, algo string) { + AppendExecutionWithMeta(bcPath, intentID, orderID, symbol, side, price, qty, map[string]string{ + "slice": slice, + "algo": algo, + }) +} diff --git a/internal/broker/algo_iceberg.go b/internal/broker/algo_iceberg.go new file mode 100644 index 0000000..ecd746e --- /dev/null +++ b/internal/broker/algo_iceberg.go @@ -0,0 +1,70 @@ +package broker + +import ( + "context" + "log" + "time" +) + +// executeICEBERG executes an Iceberg algorithm +// Submits partial orders sequentially, revealing only a portion at a time +// Each slice is submitted after the previous one, simulating fill-based progression +func (s *AlgoScheduler) executeICEBERG(ctx context.Context, task *AlgoTask) { + defer func() { + task.mu.Lock() + task.Done = true + task.mu.Unlock() + }() + + log.Printf("Starting ICEBERG execution: intent=%s total_qty=%d slices=%d visible_qty=%d", + task.IntentID, task.TotalQty, task.TotalSlices, task.SliceQty) + + // For ICEBERG, we submit slices sequentially without time delay + // In a real implementation, we would wait for fills, but for this implementation + // we simulate by submitting each slice with a small delay to represent processing time + for i := 1; i <= task.TotalSlices; i++ { + select { + case <-ctx.Done(): + log.Printf("ICEBERG execution cancelled: intent=%s at slice %d/%d", task.IntentID, i, task.TotalSlices) + return + default: + } + + // Calculate quantity for this slice + var sliceQty int64 + if i == task.TotalSlices { + // Last slice gets remainder to handle rounding + task.mu.Lock() + executedQty := task.SliceQty * int64(i-1) + sliceQty = task.TotalQty - executedQty + task.mu.Unlock() + } else { + sliceQty = task.SliceQty + } + + // Execute the slice + if err := s.executeSlice(task, i, sliceQty); err != nil { + log.Printf("ICEBERG slice %d/%d failed: intent=%s err=%v", i, task.TotalSlices, task.IntentID, err) + // Continue with remaining slices even if one fails + } + + task.mu.Lock() + task.CurrentSlice = i + task.LastSliceAt = time.Now() + task.mu.Unlock() + + // Small delay between slices to simulate fill detection and next order submission + // In a real implementation, this would wait for order fill confirmation + if i < task.TotalSlices { + select { + case <-ctx.Done(): + log.Printf("ICEBERG execution cancelled: intent=%s after slice %d/%d", task.IntentID, i, task.TotalSlices) + return + case <-time.After(2 * time.Second): + // Continue to next slice + } + } + } + + log.Printf("ICEBERG execution completed: intent=%s total_slices=%d", task.IntentID, task.TotalSlices) +} diff --git a/internal/broker/algo_test.go b/internal/broker/algo_test.go new file mode 100644 index 0000000..e2e1484 --- /dev/null +++ b/internal/broker/algo_test.go @@ -0,0 +1,163 @@ +package broker + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" +) + +func TestTWAPExecution(t *testing.T) { + // Create temporary test directory + tmpDir := t.TempDir() + tradeDir := filepath.Join(tmpDir, "trade") + if err := os.MkdirAll(tradeDir, 0755); err != nil { + t.Fatalf("Failed to create trade dir: %v", err) + } + + bcPath := filepath.Join(tradeDir, "beancount.txt") + + // Create a TWAP order + order := `2026-03-31 * "ORDER" "BUY AAPL.US 500 via TWAP" + ; intent_id: test-twap-001 + ; side: BUY + ; symbol: AAPL + ; qty: 500 + ; type: LIMIT + ; price: 182.00 + ; tif: DAY + ; algo: TWAP + ; algo_duration: 1s + ; algo_slices: 5 +` + if err := os.WriteFile(bcPath, []byte(order), 0644); err != nil { + t.Fatalf("Failed to write order: %v", err) + } + + // Create scheduler + scheduler := NewAlgoScheduler(bcPath, nil, true) + defer scheduler.Shutdown() + + ctx := context.Background() + + // Process ledger + n, err := ProcessLedgerWithScheduler(ctx, nil, tmpDir, true, scheduler) + if err != nil { + t.Fatalf("ProcessLedger failed: %v", err) + } + + if n != 1 { + t.Errorf("Expected 1 order processed, got %d", n) + } + + // Check that algo task was created + activeCount := scheduler.GetActiveCount() + if activeCount != 1 { + t.Errorf("Expected 1 active algo task, got %d", activeCount) + } + + // Wait for TWAP to complete + time.Sleep(2 * time.Second) + + // Verify executions were written + data, err := os.ReadFile(bcPath) + if err != nil { + t.Fatalf("Failed to read beancount: %v", err) + } + + content := string(data) + + // Should have 5 EXECUTION entries for 5 slices + executionCount := 0 + for i := 1; i <= 5; i++ { + sliceLabel := "slice " + string(rune('0'+i)) + "/5" + if contains(content, sliceLabel) { + executionCount++ + } + } + + if executionCount != 5 { + t.Errorf("Expected 5 TWAP slice executions, found %d", executionCount) + t.Logf("Beancount content:\n%s", content) + } + + // Verify algo tasks are completed + finalActiveCount := scheduler.GetActiveCount() + if finalActiveCount != 0 { + t.Errorf("Expected 0 active algo tasks after completion, got %d", finalActiveCount) + } +} + +func TestICEBERGExecution(t *testing.T) { + // Create temporary test directory + tmpDir := t.TempDir() + tradeDir := filepath.Join(tmpDir, "trade") + if err := os.MkdirAll(tradeDir, 0755); err != nil { + t.Fatalf("Failed to create trade dir: %v", err) + } + + bcPath := filepath.Join(tradeDir, "beancount.txt") + + // Create an ICEBERG order + order := `2026-03-31 * "ORDER" "BUY TSLA.US 300 via ICEBERG" + ; intent_id: test-iceberg-001 + ; side: BUY + ; symbol: TSLA + ; qty: 300 + ; type: LIMIT + ; price: 250.00 + ; tif: DAY + ; algo: ICEBERG + ; algo_slices: 3 +` + if err := os.WriteFile(bcPath, []byte(order), 0644); err != nil { + t.Fatalf("Failed to write order: %v", err) + } + + // Create scheduler + scheduler := NewAlgoScheduler(bcPath, nil, true) + defer scheduler.Shutdown() + + ctx := context.Background() + + // Process ledger + n, err := ProcessLedgerWithScheduler(ctx, nil, tmpDir, true, scheduler) + if err != nil { + t.Fatalf("ProcessLedger failed: %v", err) + } + + if n != 1 { + t.Errorf("Expected 1 order processed, got %d", n) + } + + // Wait for ICEBERG to complete (3 slices * 2s delay) + time.Sleep(8 * time.Second) + + // Verify executions were written + data, err := os.ReadFile(bcPath) + if err != nil { + t.Fatalf("Failed to read beancount: %v", err) + } + + content := string(data) + + // Should have 3 EXECUTION entries for 3 slices + executionCount := 0 + for i := 1; i <= 3; i++ { + sliceLabel := "slice " + string(rune('0'+i)) + "/3" + if contains(content, sliceLabel) { + executionCount++ + } + } + + if executionCount != 3 { + t.Errorf("Expected 3 ICEBERG slice executions, found %d", executionCount) + t.Logf("Beancount content:\n%s", content) + } +} + +func contains(s, substr string) bool { + return len(s) >= len(substr) && (s == substr || len(s) > len(substr) && + (s[:len(substr)] == substr || contains(s[1:], substr))) +} diff --git a/internal/broker/algo_twap.go b/internal/broker/algo_twap.go new file mode 100644 index 0000000..753f6f3 --- /dev/null +++ b/internal/broker/algo_twap.go @@ -0,0 +1,65 @@ +package broker + +import ( + "context" + "log" + "time" +) + +// executeTWAP executes a Time-Weighted Average Price algorithm +// Splits the order into equal slices and submits them at regular intervals +func (s *AlgoScheduler) executeTWAP(ctx context.Context, task *AlgoTask) { + defer func() { + task.mu.Lock() + task.Done = true + task.mu.Unlock() + }() + + log.Printf("Starting TWAP execution: intent=%s total_qty=%d slices=%d interval=%s", + task.IntentID, task.TotalQty, task.TotalSlices, task.Interval) + + for i := 1; i <= task.TotalSlices; i++ { + select { + case <-ctx.Done(): + log.Printf("TWAP execution cancelled: intent=%s at slice %d/%d", task.IntentID, i, task.TotalSlices) + return + default: + } + + // Calculate quantity for this slice + var sliceQty int64 + if i == task.TotalSlices { + // Last slice gets remainder to handle rounding + task.mu.Lock() + executedQty := task.SliceQty * int64(i-1) + sliceQty = task.TotalQty - executedQty + task.mu.Unlock() + } else { + sliceQty = task.SliceQty + } + + // Execute the slice + if err := s.executeSlice(task, i, sliceQty); err != nil { + log.Printf("TWAP slice %d/%d failed: intent=%s err=%v", i, task.TotalSlices, task.IntentID, err) + // Continue with remaining slices even if one fails + } + + task.mu.Lock() + task.CurrentSlice = i + task.LastSliceAt = time.Now() + task.mu.Unlock() + + // Wait for interval before next slice (except after last slice) + if i < task.TotalSlices { + select { + case <-ctx.Done(): + log.Printf("TWAP execution cancelled: intent=%s after slice %d/%d", task.IntentID, i, task.TotalSlices) + return + case <-time.After(task.Interval): + // Continue to next slice + } + } + } + + log.Printf("TWAP execution completed: intent=%s total_slices=%d", task.IntentID, task.TotalSlices) +} diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 69e205a..91f14d0 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -22,6 +22,11 @@ import ( // ProcessLedger reads the beancount ledger, finds unprocessed ORDER entries, // and executes them. Returns the number of new executions. func ProcessLedger(ctx context.Context, tc *trade.TradeContext, root string, useMock bool) (int, error) { + return ProcessLedgerWithScheduler(ctx, tc, root, useMock, nil) +} + +// ProcessLedgerWithScheduler processes ledger with optional algo scheduler +func ProcessLedgerWithScheduler(ctx context.Context, tc *trade.TradeContext, root string, useMock bool, scheduler *AlgoScheduler) (int, error) { bcPath := filepath.Join(root, "trade", "beancount.txt") entries, err := ledger.ParseEntries(bcPath) if err != nil { @@ -107,6 +112,21 @@ func ProcessLedger(ctx context.Context, tc *trade.TradeContext, root string, use } } + // Phase 4: Check if this is an algorithmic order + if scheduler != nil && (o.Algo == "TWAP" || o.Algo == "ICEBERG") { + // Create algo task instead of direct execution + if err := scheduler.CreateTask(o); err != nil { + sym := ledger.FullSymbol(o.Symbol, o.Market) + reason := fmt.Sprintf("ALGO_ERROR: %s", err.Error()) + AppendRejection(bcPath, o.IntentID, sym, o.Side, o.Qty, reason) + log.Printf("algo task creation failed: intent=%s err=%v", o.IntentID, err) + } + // Mark as processed so we don't try to execute it again + processed[o.IntentID] = true + executed++ + continue + } + sym := ledger.FullSymbol(o.Symbol, o.Market) if useMock { @@ -191,6 +211,11 @@ func ExecuteOrderMock(o model.ParsedOrder) (orderID string, price string) { // AppendExecution appends an EXECUTION entry to the beancount ledger. func AppendExecution(bcPath, intentID, orderID, symbol, side, price, qty string) { + AppendExecutionWithMeta(bcPath, intentID, orderID, symbol, side, price, qty, nil) +} + +// AppendExecutionWithMeta appends an EXECUTION entry with additional metadata +func AppendExecutionWithMeta(bcPath, intentID, orderID, symbol, side, price, qty string, meta map[string]string) { f, err := os.OpenFile(bcPath, os.O_APPEND|os.O_WRONLY, 0644) if err != nil { log.Printf("append execution failed: %v", err) @@ -199,7 +224,15 @@ func AppendExecution(bcPath, intentID, orderID, symbol, side, price, qty string) defer f.Close() date := time.Now().Format("2006-01-02") - text := fmt.Sprintf("\n%s * \"EXECUTION\" \"%s %s\"\n", date, side, symbol) + executedAt := time.Now().Format(time.RFC3339) + + // Build description with algo slice info if present + desc := fmt.Sprintf("%s %s", side, symbol) + if meta != nil && meta["slice"] != "" && meta["algo"] != "" { + desc = fmt.Sprintf("%s slice %s %s %s", meta["algo"], meta["slice"], side, symbol) + } + + text := fmt.Sprintf("\n%s * \"EXECUTION\" \"%s\"\n", date, desc) text += fmt.Sprintf(" ; intent_id: %s\n", intentID) text += fmt.Sprintf(" ; order_id: %s\n", orderID) text += fmt.Sprintf(" ; status: FILLED\n") @@ -209,6 +242,17 @@ func AppendExecution(bcPath, intentID, orderID, symbol, side, price, qty string) if price != "" { text += fmt.Sprintf(" ; price: %s\n", price) } + text += fmt.Sprintf(" ; executed_at: %s\n", executedAt) + + // Add additional metadata + if meta != nil { + for k, v := range meta { + if k != "" && v != "" { + text += fmt.Sprintf(" ; %s: %s\n", k, v) + } + } + } + text += "\n" f.WriteString(text) } diff --git a/internal/ledger/parser.go b/internal/ledger/parser.go index 8000af3..93160b2 100644 --- a/internal/ledger/parser.go +++ b/internal/ledger/parser.go @@ -3,6 +3,7 @@ package ledger import ( "os" "regexp" + "strconv" "strings" "longbridge-fs/internal/model" @@ -79,6 +80,9 @@ func OrderFromEntry(e model.Entry) model.ParsedOrder { // Phase 1: Extended metadata fields Source: e.Meta["source"], RebalanceID: e.Meta["rebalance_id"], + // Phase 4: Algorithm execution fields + Algo: strings.ToUpper(e.Meta["algo"]), + AlgoDuration: e.Meta["algo_duration"], } // Parse signal_refs (comma-separated) @@ -90,6 +94,13 @@ func OrderFromEntry(e model.Entry) model.ParsedOrder { o.SignalRefs = refs } + // Parse algo_slices (integer) + if algoSlices := e.Meta["algo_slices"]; algoSlices != "" { + if n, err := strconv.Atoi(algoSlices); err == nil && n > 0 { + o.AlgoSlices = n + } + } + if o.Market == "" { o.Market = "US" } diff --git a/internal/model/types.go b/internal/model/types.go index 1455829..6d6fd91 100644 --- a/internal/model/types.go +++ b/internal/model/types.go @@ -55,6 +55,10 @@ type ParsedOrder struct { Source string // manual, rebalance, risk_trigger RebalanceID string // links to portfolio rebalance SignalRefs []string // triggering signals (comma-separated in beancount) + // Phase 4: Algorithm execution fields + Algo string // NONE (default), TWAP, ICEBERG + AlgoDuration string // e.g., "30m", "1h" + AlgoSlices int // number of slices to split order into } // --- Quote JSON types --- diff --git a/task.md b/task.md index 31cc157..f02859b 100644 --- a/task.md +++ b/task.md @@ -40,9 +40,9 @@ 目标：大单拆分执行 -- [ ] TWAP 实现 (goroutine scheduler) -- [ ] ICEBERG 实现 -- [ ] 算法子单关联与审计 +- [x] TWAP 实现 (goroutine scheduler) +- [x] ICEBERG 实现 +- [x] 算法子单关联与审计 ### Phase 5: Agent 集成增强 (v0.7.0)