diff --git a/cmd/longbridge-fs/main.go b/cmd/longbridge-fs/main.go index 7d8c25f..bfdf5fc 100644 --- a/cmd/longbridge-fs/main.go +++ b/cmd/longbridge-fs/main.go @@ -18,6 +18,7 @@ import ( "longbridge-fs/internal/portfolio" "longbridge-fs/internal/research" "longbridge-fs/internal/risk" + signalpkg "longbridge-fs/internal/signal" "github.com/longbridge/openapi-go/quote" "github.com/longbridge/openapi-go/trade" @@ -371,11 +372,12 @@ func runInit(root string) error { // controllerCmd creates the controller subcommand func controllerCmd() *cobra.Command { var ( - root string - interval time.Duration - credFile string - mock bool - compactAfter int + root string + interval time.Duration + credFile string + mock bool + compactAfter int + autoRebalance bool ) cmd := &cobra.Command{ @@ -399,7 +401,7 @@ The controller monitors the file system and automatically: # Custom polling interval longbridge-fs controller --root ./fs --interval 5s`, RunE: func(cmd *cobra.Command, args []string) error { - return runController(root, interval, credFile, mock, compactAfter) + return runController(root, interval, credFile, mock, compactAfter, autoRebalance) }, } @@ -408,11 +410,12 @@ The controller monitors the file system and automatically: cmd.Flags().StringVar(&credFile, "credential", "credential", "Credential file path") cmd.Flags().BoolVar(&mock, "mock", false, "Use mock execution without API") cmd.Flags().IntVar(&compactAfter, "compact-after", 10, "Compact after N executed orders, 0=disable") + cmd.Flags().BoolVar(&autoRebalance, "auto-rebalance", false, "Automatically create rebalance orders when portfolio drift is detected") return cmd } -func runController(root string, interval time.Duration, credFile string, mock bool, compactAfter int) error { +func runController(root string, interval time.Duration, credFile string, mock bool, compactAfter int, autoRebalance bool) error { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -475,6 +478,7 @@ func runController(root string, interval time.Duration, credFile string, mock bo log.Printf(" Interval: %s", interval) log.Printf(" Compact after: %d orders", compactAfter) log.Printf(" Mock mode: %v", useMock) + log.Printf(" Auto-rebalance: %v", autoRebalance) } log.Printf("🚀 Controller started (interval=%s, compact-after=%d)", interval, compactAfter) @@ -540,6 +544,15 @@ func runController(root string, interval time.Duration, credFile string, mock bo } } + // Phase 3: Compute builtin signals from signal/definitions/ + if err := signalpkg.ComputeAll(root); err != nil { + if verbose { + log.Printf("⚠ Signal computation failed: %v", err) + } + } else if verbose { + log.Printf("✓ Signals computed") + } + // Generate PnL report (positions + current prices — file-only, works in mock) if err := account.GeneratePnL(root); err != nil { log.Printf("❌ PnL generation failed: %v", err) @@ -568,6 +581,15 @@ func runController(root string, interval time.Duration, credFile string, mock bo log.Printf("✓ Portfolio diff computed") } + // Phase 2: Auto-rebalance mode: create pending.json from diff when drift detected + if autoRebalance { + if err := portfolio.AutoCreatePending(root); err != nil { + log.Printf("❌ Auto-rebalance failed: %v", err) + } else if verbose { + log.Printf("✓ Auto-rebalance check complete") + } + } + // Phase 2: Process pending rebalance orders if err := portfolio.ProcessRebalance(root); err != nil { log.Printf("❌ Rebalance processing failed: %v", err) diff --git a/internal/model/types.go b/internal/model/types.go index 602f9f8..1455829 100644 --- a/internal/model/types.go +++ b/internal/model/types.go @@ -325,6 +325,49 @@ type RebalanceOrder struct { TIF string `json:"tif"` } +// --- Phase 3: Signal types --- + +// SignalDefinition is the JSON structure for /signal/definitions/{name}.json +type SignalDefinition struct { + Name string `json:"name"` + Type string `json:"type"` // builtin, external + Version int `json:"version"` + Params map[string]interface{} `json:"params,omitempty"` // indicator-specific params + Symbols []string `json:"symbols"` + Enabled bool `json:"enabled"` + Description string `json:"description,omitempty"` +} + +// SignalOutput is the JSON structure for /signal/output/{SYMBOL}/latest.json +type SignalOutput struct { + Symbol string `json:"symbol"` + UpdatedAt string `json:"updated_at"` + Signals []SignalEntry `json:"signals"` +} + +// SignalEntry represents a computed signal value +type SignalEntry struct { + Name string `json:"name"` + Value string `json:"value"` // BULLISH, BEARISH, NEUTRAL, POSITIVE, NEGATIVE + Strength float64 `json:"strength"` // 0.0 ~ 1.0 + Detail string `json:"detail"` + ComputedAt string `json:"computed_at"` +} + +// ActiveSignals is the JSON structure for /signal/active.json +type ActiveSignals struct { + UpdatedAt string `json:"updated_at"` + Signals []ActiveSignalEntry `json:"signals"` +} + +// ActiveSignalEntry is one active signal in the aggregated view +type ActiveSignalEntry struct { + Symbol string `json:"symbol"` + Name string `json:"name"` + Value string `json:"value"` + Strength float64 `json:"strength"` +} + // --- Phase 3: Research & Signal types --- // Watchlist is the JSON structure for /research/watchlist.json diff --git a/internal/portfolio/rebalance.go b/internal/portfolio/rebalance.go index c30720c..7699e6b 100644 --- a/internal/portfolio/rebalance.go +++ b/internal/portfolio/rebalance.go @@ -11,6 +11,79 @@ import ( "longbridge-fs/internal/model" ) +// AutoCreatePending reads portfolio/diff.json and, when requires_rebalance is true, +// automatically generates portfolio/rebalance/pending.json so the controller can +// process it on the next cycle. It is a no-op when pending.json already exists. +func AutoCreatePending(root string) error { + // Read diff.json + diffPath := filepath.Join(root, "portfolio", "diff.json") + data, err := os.ReadFile(diffPath) + if err != nil { + if os.IsNotExist(err) { + return nil // No diff available + } + return fmt.Errorf("read diff: %w", err) + } + + var diff model.PortfolioDiff + if err := json.Unmarshal(data, &diff); err != nil { + return fmt.Errorf("parse diff: %w", err) + } + + if !diff.RequiresRebalance || len(diff.Adjustments) == 0 { + return nil // Nothing to do + } + + // Don't overwrite an existing pending.json + pendingPath := filepath.Join(root, "portfolio", "rebalance", "pending.json") + if _, err := os.Stat(pendingPath); err == nil { + return nil // Already pending + } + + // Convert diff adjustments to rebalance orders + rebalanceID := fmt.Sprintf("rebal-%s", time.Now().UTC().Format("20060102-150405")) + pending := model.RebalancePending{ + RebalanceID: rebalanceID, + CreatedAt: time.Now().UTC().Format(time.RFC3339), + CreatedBy: "controller-auto-rebalance", + AutoExecute: true, + Orders: make([]model.RebalanceOrder, 0, len(diff.Adjustments)), + } + + for _, adj := range diff.Adjustments { + if adj.EstimatedQty <= 0 { + continue + } + + order := model.RebalanceOrder{ + Symbol: adj.Symbol, + Side: adj.EstimatedSide, + Qty: adj.EstimatedQty, + Type: "MARKET", + TIF: "DAY", + } + + pending.Orders = append(pending.Orders, order) + } + + if len(pending.Orders) == 0 { + return nil + } + + // Write pending.json + rebalDir := filepath.Join(root, "portfolio", "rebalance") + if err := os.MkdirAll(rebalDir, 0755); err != nil { + return fmt.Errorf("create rebalance dir: %w", err) + } + + out, err := json.MarshalIndent(pending, "", " ") + if err != nil { + return fmt.Errorf("marshal pending: %w", err) + } + + return os.WriteFile(pendingPath, append(out, '\n'), 0644) +} + // ProcessRebalance reads portfolio/rebalance/pending.json and converts it to ORDER entries // in trade/beancount.txt, then archives the pending file to history/ func ProcessRebalance(root string) error { @@ -65,7 +138,7 @@ func writeRebalanceOrders(root string, pending *model.RebalancePending) error { var orderLines strings.Builder orderLines.WriteString("\n") - orderLines.WriteString(fmt.Sprintf("%s * \"ORDER\" \"%s %s %d %s\"\n", + orderLines.WriteString(fmt.Sprintf("%s * \"ORDER\" \"%s %d %s %s\"\n", timestamp, order.Side, order.Qty, order.Symbol, pending.RebalanceID)) orderLines.WriteString(fmt.Sprintf(" ; intent_id: %s\n", intentID)) orderLines.WriteString(fmt.Sprintf(" ; side: %s\n", order.Side)) diff --git a/internal/signal/definitions.go b/internal/signal/definitions.go new file mode 100644 index 0000000..9463d9e --- /dev/null +++ b/internal/signal/definitions.go @@ -0,0 +1,94 @@ +package signal + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + + "longbridge-fs/internal/model" +) + +// ListDefinitions scans signal/definitions/ and returns all enabled signal definitions. +func ListDefinitions(root string) ([]*model.SignalDefinition, error) { + defsDir := filepath.Join(root, "signal", "definitions") + + entries, err := os.ReadDir(defsDir) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, fmt.Errorf("read definitions dir: %w", err) + } + + var defs []*model.SignalDefinition + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") { + continue + } + + def, err := ParseDefinition(filepath.Join(defsDir, entry.Name())) + if err != nil { + // Skip invalid definitions but log + fmt.Printf("Warning: skipping invalid signal definition %s: %v\n", entry.Name(), err) + continue + } + + if def.Enabled { + defs = append(defs, def) + } + } + + return defs, nil +} + +// ParseDefinition reads and parses a single signal definition file. +func ParseDefinition(path string) (*model.SignalDefinition, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + var def model.SignalDefinition + if err := json.Unmarshal(data, &def); err != nil { + return nil, fmt.Errorf("parse definition: %w", err) + } + + if def.Name == "" { + return nil, fmt.Errorf("definition missing name") + } + + if def.Type != "builtin" && def.Type != "external" { + return nil, fmt.Errorf("unknown definition type %q (must be builtin or external)", def.Type) + } + + return &def, nil +} + +// paramFloat extracts a float64 parameter from a signal definition's Params map. +// Returns defaultVal if the key is absent or not numeric. +func paramFloat(params map[string]interface{}, key string, defaultVal float64) float64 { + if params == nil { + return defaultVal + } + v, ok := params[key] + if !ok { + return defaultVal + } + switch val := v.(type) { + case float64: + return val + case int: + return float64(val) + case int64: + return float64(val) + } + return defaultVal +} + +// paramInt extracts an int parameter from a signal definition's Params map. +func paramInt(params map[string]interface{}, key string, defaultVal int) int { + f := paramFloat(params, key, float64(defaultVal)) + return int(f) +} diff --git a/internal/signal/indicator.go b/internal/signal/indicator.go new file mode 100644 index 0000000..9d3e113 --- /dev/null +++ b/internal/signal/indicator.go @@ -0,0 +1,139 @@ +package signal + +import ( + "fmt" + "math" +) + +// strengthScaleFactorSMA scales the SMA distance to a 0-1 strength value. +// A 5% relative difference between fast and slow SMA maps to strength 1.0. +const strengthScaleFactorSMA = 20.0 +// Returns: value (BULLISH/BEARISH/NEUTRAL), strength (0.0-1.0), detail string. +func ComputeSMACross(prices []float64, fastPeriod, slowPeriod int) (string, float64, string) { + if len(prices) < slowPeriod+1 { + return "NEUTRAL", 0.0, fmt.Sprintf("insufficient data (need %d, have %d)", slowPeriod+1, len(prices)) + } + + fastNow := sma(prices, fastPeriod) + slowNow := sma(prices, slowPeriod) + + // Previous bar values + prevPrices := prices[:len(prices)-1] + fastPrev := sma(prevPrices, fastPeriod) + slowPrev := sma(prevPrices, slowPeriod) + + // Detect crossover + crossedAbove := fastPrev <= slowPrev && fastNow > slowNow + crossedBelow := fastPrev >= slowPrev && fastNow < slowNow + + // Strength: relative distance between SMAs + diff := math.Abs(fastNow-slowNow) / slowNow + strength := math.Min(1.0, diff*strengthScaleFactorSMA) + + if crossedAbove { + return "BULLISH", strength, fmt.Sprintf("SMA%d (%.2f) crossed above SMA%d (%.2f)", fastPeriod, fastNow, slowPeriod, slowNow) + } + if crossedBelow { + return "BEARISH", strength, fmt.Sprintf("SMA%d (%.2f) crossed below SMA%d (%.2f)", fastPeriod, fastNow, slowPeriod, slowNow) + } + + if fastNow > slowNow { + return "BULLISH", strength * 0.5, fmt.Sprintf("SMA%d (%.2f) above SMA%d (%.2f), no fresh cross", fastPeriod, fastNow, slowPeriod, slowNow) + } + return "BEARISH", strength * 0.5, fmt.Sprintf("SMA%d (%.2f) below SMA%d (%.2f), no fresh cross", fastPeriod, fastNow, slowPeriod, slowNow) +} + +// ComputeRSI computes the Relative Strength Index signal. +// Returns: value (BULLISH/BEARISH/NEUTRAL), strength (0.0-1.0), detail string. +func ComputeRSI(prices []float64, period int, overbought, oversold float64) (string, float64, string) { + if len(prices) < period+1 { + return "NEUTRAL", 0.0, fmt.Sprintf("insufficient data (need %d, have %d)", period+1, len(prices)) + } + + rsiVal := rsi(prices, period) + detail := fmt.Sprintf("RSI(%d) = %.1f", period, rsiVal) + + if rsiVal >= overbought { + strength := math.Min(1.0, (rsiVal-overbought)/(100-overbought)) + return "BEARISH", strength, detail + fmt.Sprintf(" (overbought > %.0f)", overbought) + } + if rsiVal <= oversold { + strength := math.Min(1.0, (oversold-rsiVal)/oversold) + return "BULLISH", strength, detail + fmt.Sprintf(" (oversold < %.0f)", oversold) + } + + // Neutral: strength represents distance from midpoint + mid := (overbought + oversold) / 2 + strength := math.Abs(rsiVal-mid) / (overbought - mid) + return "NEUTRAL", strength, detail +} + +// ComputePriceChange computes the price change signal over a window. +// Returns: value (BULLISH/BEARISH/NEUTRAL), strength (0.0-1.0), detail string. +func ComputePriceChange(prices []float64, thresholdPct float64, window int) (string, float64, string) { + if len(prices) < window+1 { + return "NEUTRAL", 0.0, fmt.Sprintf("insufficient data (need %d, have %d)", window+1, len(prices)) + } + + current := prices[len(prices)-1] + past := prices[len(prices)-1-window] + + if past == 0 { + return "NEUTRAL", 0.0, "base price is zero" + } + + changePct := (current - past) / past * 100 + detail := fmt.Sprintf("price change %.2f%% over %d bars (%.2f → %.2f)", changePct, window, past, current) + + absChange := math.Abs(changePct) + strength := math.Min(1.0, absChange/thresholdPct) + + if changePct >= thresholdPct { + return "BULLISH", strength, detail + } + if changePct <= -thresholdPct { + return "BEARISH", strength, detail + } + return "NEUTRAL", strength, detail +} + +// sma computes the Simple Moving Average of the last n prices. +func sma(prices []float64, n int) float64 { + if len(prices) < n { + return 0 + } + sum := 0.0 + for _, p := range prices[len(prices)-n:] { + sum += p + } + return sum / float64(n) +} + +// rsi computes the RSI of the last period+1 prices using the Wilder method. +func rsi(prices []float64, period int) float64 { + if len(prices) < period+1 { + return 50 + } + + gains := 0.0 + losses := 0.0 + + for i := len(prices) - period; i < len(prices); i++ { + diff := prices[i] - prices[i-1] + if diff > 0 { + gains += diff + } else { + losses -= diff + } + } + + avgGain := gains / float64(period) + avgLoss := losses / float64(period) + + if avgLoss == 0 { + return 100 + } + + rs := avgGain / avgLoss + return 100 - (100 / (1 + rs)) +} diff --git a/internal/signal/output.go b/internal/signal/output.go new file mode 100644 index 0000000..b1b6ded --- /dev/null +++ b/internal/signal/output.go @@ -0,0 +1,57 @@ +package signal + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + + "longbridge-fs/internal/model" +) + +// WriteOutput writes signal output to signal/output/{SYMBOL}/latest.json +func WriteOutput(root, symbol string, output *model.SignalOutput) error { + outDir := filepath.Join(root, "signal", "output", symbol) + if err := os.MkdirAll(outDir, 0755); err != nil { + return fmt.Errorf("create output dir: %w", err) + } + + data, err := json.MarshalIndent(output, "", " ") + if err != nil { + return fmt.Errorf("marshal signal output: %w", err) + } + + return os.WriteFile(filepath.Join(outDir, "latest.json"), append(data, '\n'), 0644) +} + +// AppendHistory appends a signal output snapshot to signal/output/{SYMBOL}/history.jsonl +func AppendHistory(root, symbol string, output *model.SignalOutput) error { + outDir := filepath.Join(root, "signal", "output", symbol) + if err := os.MkdirAll(outDir, 0755); err != nil { + return fmt.Errorf("create output dir: %w", err) + } + + line, err := json.Marshal(output) + if err != nil { + return fmt.Errorf("marshal signal output: %w", err) + } + + f, err := os.OpenFile(filepath.Join(outDir, "history.jsonl"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) + if err != nil { + return err + } + defer f.Close() + + _, err = fmt.Fprintf(f, "%s\n", line) + return err +} + +// WriteActiveSignals writes the aggregated active signals to signal/active.json +func WriteActiveSignals(root string, active *model.ActiveSignals) error { + data, err := json.MarshalIndent(active, "", " ") + if err != nil { + return fmt.Errorf("marshal active signals: %w", err) + } + + return os.WriteFile(filepath.Join(root, "signal", "active.json"), append(data, '\n'), 0644) +} diff --git a/internal/signal/signal.go b/internal/signal/signal.go new file mode 100644 index 0000000..0c14838 --- /dev/null +++ b/internal/signal/signal.go @@ -0,0 +1,146 @@ +package signal + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" + + "longbridge-fs/internal/model" +) + +// ComputeAll scans signal/definitions/, computes builtin signals from quote data, +// writes per-symbol output files, and regenerates signal/active.json. +func ComputeAll(root string) error { + defs, err := ListDefinitions(root) + if err != nil { + return fmt.Errorf("list signal definitions: %w", err) + } + + if len(defs) == 0 { + return nil // Nothing to compute + } + + active := &model.ActiveSignals{ + UpdatedAt: time.Now().UTC().Format(time.RFC3339), + Signals: []model.ActiveSignalEntry{}, + } + + // Group definitions by symbol so we load kline data once per symbol + symbolDefs := make(map[string][]*model.SignalDefinition) + for _, def := range defs { + for _, sym := range def.Symbols { + symbolDefs[sym] = append(symbolDefs[sym], def) + } + } + + now := active.UpdatedAt + + for symbol, sdefs := range symbolDefs { + // Load kline data (daily close prices) + prices, err := loadClosePrices(root, symbol) + if err != nil { + fmt.Printf("Warning: skipping signal computation for %s: %v\n", symbol, err) + continue + } + + output := &model.SignalOutput{ + Symbol: symbol, + UpdatedAt: now, + Signals: []model.SignalEntry{}, + } + + for _, def := range sdefs { + if def.Type != "builtin" { + continue // External signals are written by agents, not computed here + } + + entry, err := computeBuiltin(def, prices, now) + if err != nil { + fmt.Printf("Warning: failed to compute signal %s for %s: %v\n", def.Name, symbol, err) + continue + } + + output.Signals = append(output.Signals, *entry) + + active.Signals = append(active.Signals, model.ActiveSignalEntry{ + Symbol: symbol, + Name: def.Name, + Value: entry.Value, + Strength: entry.Strength, + }) + } + + if err := WriteOutput(root, symbol, output); err != nil { + fmt.Printf("Warning: failed to write signal output for %s: %v\n", symbol, err) + } + if err := AppendHistory(root, symbol, output); err != nil { + fmt.Printf("Warning: failed to append signal history for %s: %v\n", symbol, err) + } + } + + return WriteActiveSignals(root, active) +} + +// computeBuiltin dispatches to the appropriate indicator function based on the definition params. +func computeBuiltin(def *model.SignalDefinition, prices []float64, now string) (*model.SignalEntry, error) { + indicator, _ := def.Params["indicator"].(string) + + var value, detail string + var strength float64 + + switch indicator { + case "SMA_CROSS": + fastPeriod := paramInt(def.Params, "fast_period", 5) + slowPeriod := paramInt(def.Params, "slow_period", 20) + value, strength, detail = ComputeSMACross(prices, fastPeriod, slowPeriod) + + case "RSI": + period := paramInt(def.Params, "period", 14) + overbought := paramFloat(def.Params, "overbought", 70) + oversold := paramFloat(def.Params, "oversold", 30) + value, strength, detail = ComputeRSI(prices, period, overbought, oversold) + + case "PRICE_CHANGE": + thresholdPct := paramFloat(def.Params, "threshold_pct", 5.0) + window := paramInt(def.Params, "window", 5) + value, strength, detail = ComputePriceChange(prices, thresholdPct, window) + + default: + return nil, fmt.Errorf("unknown indicator %q", indicator) + } + + return &model.SignalEntry{ + Name: def.Name, + Value: value, + Strength: strength, + Detail: detail, + ComputedAt: now, + }, nil +} + +// loadClosePrices reads the daily candlestick JSON for a symbol and returns close prices. +func loadClosePrices(root, symbol string) ([]float64, error) { + klinePath := filepath.Join(root, "quote", "hold", symbol, "D.json") + data, err := os.ReadFile(klinePath) + if err != nil { + return nil, fmt.Errorf("read kline data: %w", err) + } + + var sticks []model.Candlestick + if err := json.Unmarshal(data, &sticks); err != nil { + return nil, fmt.Errorf("parse kline data: %w", err) + } + + if len(sticks) == 0 { + return nil, fmt.Errorf("no kline data available") + } + + prices := make([]float64, len(sticks)) + for i, s := range sticks { + prices[i] = s.Close + } + + return prices, nil +} diff --git a/task.md b/task.md index e6008c1..31cc157 100644 --- a/task.md +++ b/task.md @@ -25,16 +25,16 @@ - [x] `portfolio/target.json` → `current.json` → `diff.json` 计算 - [x] `rebalance/pending.json` → ORDER 转化 - [x] `portfolio/history/` 快照归档 -- [ ] `--auto-rebalance` 模式 +- [x] `--auto-rebalance` 模式 ### Phase 3: 研究与信号 (v0.5.0) 目标：数据管道自动化 - [x] `research/watchlist.json` + Content API 集成 -- [ ] `signal/definitions/` 声明式配置 -- [ ] 内置指标计算 (SMA_CROSS, RSI, PRICE_CHANGE) -- [ ] `signal/active.json` 聚合 +- [x] `signal/definitions/` 声明式配置 +- [x] 内置指标计算 (SMA_CROSS, RSI, PRICE_CHANGE) +- [x] `signal/active.json` 聚合 ### Phase 4: 算法执行 (v0.6.0)