From 63687723abe227921a7540f491c661797c707208 Mon Sep 17 00:00:00 2001 From: fish Date: Sat, 4 Jul 2026 11:54:07 +0800 Subject: [PATCH] =?UTF-8?q?=E7=B3=BB=E7=BB=9F=E7=AE=A1=E7=90=86=E5=91=98?= =?UTF-8?q?=E7=82=B9=E5=87=BB=E5=90=8C=E6=AD=A5=E6=97=B6=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E5=8D=8A=E5=B9=B4=E5=8E=86=E5=8F=B2=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/datasource/tickflow.go | 41 ++++++++ backend/internal/handlers/stock.go | 16 +++- backend/internal/services/stock_sync.go | 121 +++++++++++++++++++++++- 3 files changed, 176 insertions(+), 2 deletions(-) diff --git a/backend/internal/datasource/tickflow.go b/backend/internal/datasource/tickflow.go index d9c2ec7..0bf5cc2 100644 --- a/backend/internal/datasource/tickflow.go +++ b/backend/internal/datasource/tickflow.go @@ -74,6 +74,19 @@ type UniverseDetail struct { Symbols []string `json:"symbols"` } +// KlineData 对应 /v1/klines/batch 返回的单标的历史 K 线（紧凑列式）。 +type KlineData struct { + Symbol string `json:"-"` + Name string `json:"-"` + Timestamp []int64 `json:"timestamp"` + Open []float64 `json:"open"` + High []float64 `json:"high"` + Low []float64 `json:"low"` + Close []float64 `json:"close"` + Volume []int64 `json:"volume"` + Amount []float64 `json:"amount"` +} + // Client 封装 tickflow HTTP API 调用。 type Client struct { baseURL string @@ -120,6 +133,34 @@ func (c *Client) GetQuotesBySymbols(symbols []string) ([]Quote, error) { return quotes, nil } +// GetKlinesBatch 批量获取多只股票的历史日 K 线。 +// symbols 最多 100 只（tickflow 限制）。 +func (c *Client) GetKlinesBatch(symbols []string, period string, startMs, endMs int64) (map[string]KlineData, error) { + if len(symbols) == 0 { + return nil, nil + } + if len(symbols) > 100 { + return nil, fmt.Errorf("GetKlinesBatch supports up to 100 symbols, got %d", len(symbols)) + } + params := map[string]any{ + "symbols": strings.Join(symbols, ","), + "period": period, + "adjust": "none", + "start_time": startMs, + "end_time": endMs, + "count": 10000, + } + var result map[string]KlineData + if err := c.request("GET", "/v1/klines/batch", params, nil, &result); err != nil { + return nil, err + } + for sym, data := range result { + data.Symbol = sym + result[sym] = data + } + return result, nil +} + func (c *Client) request(method, path string, params, body map[string]any, result any) error { url := c.baseURL + path diff --git a/backend/internal/handlers/stock.go b/backend/internal/handlers/stock.go index 9706f8d..fb90c11 100644 --- a/backend/internal/handlers/stock.go +++ b/backend/internal/handlers/stock.go @@ -5,6 +5,7 @@ import ( "net/http" "time" + "stock-user-system/internal/middleware" "stock-user-system/internal/models" "stock-user-system/internal/services" @@ -146,8 +147,21 @@ func (h *StockHandler) SyncStatus(c *gin.Context) { } // TriggerSync 手动触发同步（管理员）。 +// 系统管理员点击时同步近半年历史数据，其他管理员只同步当天。 func (h *StockHandler) TriggerSync(c *gin.Context) { - count, err := h.Svc.Sync(c.Request.Context(), "manual") + user, ok := middleware.GetCurrentUser(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"success": false, "error": "未登录"}) + return + } + + var count int + var err error + if user.Role == models.RoleSystemAdmin { + count, err = h.Svc.SyncHistory(c.Request.Context(), "manual", 6) + } else { + count, err = h.Svc.Sync(c.Request.Context(), "manual") + } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"success": false, "error": err.Error()}) return diff --git a/backend/internal/services/stock_sync.go b/backend/internal/services/stock_sync.go index c67e788..7e9ad23 100644 --- a/backend/internal/services/stock_sync.go +++ b/backend/internal/services/stock_sync.go @@ -40,6 +40,19 @@ func NewStockSyncService(cfg *config.Config, db *gorm.DB) *StockSyncService { // Sync 执行一次盘后同步，返回写入的记录数。 func (s *StockSyncService) Sync(ctx context.Context, triggerBy string) (int, error) { + return s.runSyncJob(ctx, triggerBy, func() (int, error) { + return s.doSync(ctx) + }) +} + +// SyncHistory 同步近 N 个月的历史日 K 数据，返回写入的记录数。 +func (s *StockSyncService) SyncHistory(ctx context.Context, triggerBy string, months int) (int, error) { + return s.runSyncJob(ctx, triggerBy, func() (int, error) { + return s.doSyncHistory(ctx, months) + }) +} + +func (s *StockSyncService) runSyncJob(ctx context.Context, triggerBy string, fn func() (int, error)) (int, error) { s.mu.Lock() defer s.mu.Unlock() @@ -53,7 +66,7 @@ func (s *StockSyncService) Sync(ctx context.Context, triggerBy string) (int, err return 0, fmt.Errorf("create sync job: %w", err) } - recordsCount, err := s.doSync(ctx) + recordsCount, err := fn() now := time.Now().UTC() updates := map[string]any{ @@ -143,6 +156,105 @@ func (s *StockSyncService) LatestJob(ctx context.Context) (*models.StockSyncJob, return &job, nil } +func (s *StockSyncService) doSyncHistory(ctx context.Context, months int) (int, error) { + // 1. 获取全 A 股代码列表 + universe, err := s.client.GetUniverse("CN_Equity_A") + if err != nil { + return 0, fmt.Errorf("fetch universe: %w", err) + } + if universe == nil || len(universe.Symbols) == 0 { + return 0, fmt.Errorf("empty universe") + } + + // 2. 获取核心指数列表 + allSymbols := append([]string{}, universe.Symbols...) + allSymbols = append(allSymbols, coreIndexSymbols...) + + // 3. 计算时间范围（近 N 个月） + endMs := time.Now().In(chinaLoc).UnixMilli() + startMs := time.Now().In(chinaLoc).AddDate(0, -months, 0).UnixMilli() + + log.Printf("[stock sync] start history sync for %d months, symbols=%d, range=%d-%d", + months, len(allSymbols), startMs, endMs) + + // 4. 分批拉取并写入 + const batchSize = 50 + totalRecords := 0 + for i := 0; i < len(allSymbols); i += batchSize { + batch := allSymbols[i:min(i+batchSize, len(allSymbols))] + klinesMap, err := s.client.GetKlinesBatch(batch, "1d", startMs, endMs) + if err != nil { + log.Printf("[stock sync] klines batch %d-%d failed: %v", i+1, min(i+batchSize, len(allSymbols)), err) + continue + } + + records := make([]models.StockDailyQuote, 0) + for sym, data := range klinesMap { + name := sym + records = append(records, klineDataToRecords(sym, name, data)...) + } + + if len(records) == 0 { + continue + } + + if err := s.db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "symbol"}, {Name: "trade_date"}}, + UpdateAll: true, + }).CreateInBatches(records, 500).Error; err != nil { + return totalRecords, fmt.Errorf("upsert history batch %d: %w", i/batchSize, err) + } + totalRecords += len(records) + log.Printf("[stock sync] history batch %d/%d done, records=%d, total=%d", + (i/batchSize)+1, (len(allSymbols)+batchSize-1)/batchSize, len(records), totalRecords) + } + + return totalRecords, nil +} + +func klineDataToRecords(symbol, name string, data datasource.KlineData) []models.StockDailyQuote { + n := len(data.Timestamp) + if n == 0 { + return nil + } + if len(data.Open) < n { n = len(data.Open) } + if len(data.High) < n { n = len(data.High) } + if len(data.Low) < n { n = len(data.Low) } + if len(data.Close) < n { n = len(data.Close) } + if len(data.Volume) < n { n = len(data.Volume) } + if len(data.Amount) < n { n = len(data.Amount) } + + records := make([]models.StockDailyQuote, 0, n) + for i := 0; i < n; i++ { + open := round2(data.Open[i]) + high := round2(data.High[i]) + low := round2(data.Low[i]) + close := round2(data.Close[i]) + var prevClose float64 + if i > 0 { + prevClose = round2(data.Close[i-1]) + } + var changePct float64 + if prevClose != 0 { + changePct = (close - prevClose) / prevClose + } + records = append(records, models.StockDailyQuote{ + Symbol: symbol, + Name: name, + TradeDate: msToDate(data.Timestamp[i]), + Open: open, + High: high, + Low: low, + Close: close, + PrevClose: prevClose, + Volume: data.Volume[i], + Amount: data.Amount[i], + ChangePct: changePct, + }) + } + return records +} + func today() time.Time { return time.Now().In(chinaLoc).Truncate(24 * time.Hour) } @@ -154,3 +266,10 @@ func msToDate(ms int64) time.Time { func round2(v float64) float64 { return math.Round(v*100) / 100 } + +func min(a, b int) int { + if a < b { + return a + } + return b +}