新增盘后 A 股数据同步与看板市场概览

2026-07-04 10:36:59 +08:00
parent 1fd9e97fe1
commit d18aa79a47
13 changed files with 968 additions and 2 deletions
@@ -0,0 +1,138 @@
+package services
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"stock-user-system/internal/config"
+	"stock-user-system/internal/datasource"
+	"stock-user-system/internal/models"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+var (
+	coreIndexSymbols = []string{"000001.SH", "399001.SZ", "399006.SZ", "000688.SH"}
+	chinaLoc, _     = time.LoadLocation("Asia/Shanghai")
+)
+
+// StockSyncService 负责盘后行情同步。
+type StockSyncService struct {
+	cfg    *config.Config
+	db     *gorm.DB
+	client *datasource.Client
+	mu     sync.Mutex
+}
+
+// NewStockSyncService 创建同步服务。
+func NewStockSyncService(cfg *config.Config, db *gorm.DB) *StockSyncService {
+	return &StockSyncService{
+		cfg:    cfg,
+		db:     db,
+		client: datasource.NewClient(cfg),
+	}
+}
+
+// Sync 执行一次盘后同步，返回写入的记录数。
+func (s *StockSyncService) Sync(ctx context.Context, triggerBy string) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	job := models.StockSyncJob{
+		JobDate:   today(),
+		Status:    "running",
+		StartedAt: time.Now().UTC(),
+		TriggerBy: triggerBy,
+	}
+	if err := s.db.WithContext(ctx).Create(&job).Error; err != nil {
+		return 0, fmt.Errorf("create sync job: %w", err)
+	}
+
+	recordsCount, err := s.doSync(ctx)
+
+	now := time.Now().UTC()
+	updates := map[string]any{
+		"records_count": recordsCount,
+		"finished_at":   &now,
+	}
+	if err != nil {
+		updates["status"] = "failed"
+		updates["error_message"] = err.Error()
+	} else {
+		updates["status"] = "success"
+	}
+	s.db.WithContext(ctx).Model(&job).Updates(updates)
+
+	return recordsCount, err
+}
+
+func (s *StockSyncService) doSync(ctx context.Context) (int, error) {
+	// 1. 拉取全 A 股行情快照
+	stockQuotes, err := s.client.GetQuotesByUniverses([]string{"CN_Equity_A"})
+	if err != nil {
+		return 0, fmt.Errorf("fetch stock quotes: %w", err)
+	}
+
+	// 2. 拉取核心指数行情
+	indexQuotes, err := s.client.GetQuotesBySymbols(coreIndexSymbols)
+	if err != nil {
+		return 0, fmt.Errorf("fetch index quotes: %w", err)
+	}
+
+	quotes := append(stockQuotes, indexQuotes...)
+	if len(quotes) == 0 {
+		return 0, fmt.Errorf("no quotes returned")
+	}
+
+	// 3. 归一化并写入 DB
+	records := make([]models.StockDailyQuote, 0, len(quotes))
+	for _, q := range quotes {
+		tradeDate := msToDate(q.Timestamp)
+		records = append(records, models.StockDailyQuote{
+			Symbol:       q.Symbol,
+			Name:         q.DisplayName(),
+			TradeDate:    tradeDate,
+			Open:         q.Open,
+			High:         q.High,
+			Low:          q.Low,
+			Close:        q.Close,
+			PrevClose:    q.PrevClose,
+			Volume:       q.Volume,
+			Amount:       q.Amount,
+			ChangePct:    q.ChangePct(),
+			TurnoverRate: q.TurnoverRate(),
+			Region:       q.Region,
+		})
+	}
+
+	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 0, fmt.Errorf("upsert daily quotes: %w", err)
+	}
+
+	return len(records), nil
+}
+
+func (s *StockSyncService) LatestJob(ctx context.Context) (*models.StockSyncJob, error) {
+	var job models.StockSyncJob
+	if err := s.db.WithContext(ctx).Order("started_at DESC").First(&job).Error; err != nil {
+		if err == gorm.ErrRecordNotFound {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &job, nil
+}
+
+func today() time.Time {
+	return time.Now().In(chinaLoc).Truncate(24 * time.Hour)
+}
+
+func msToDate(ms int64) time.Time {
+	return time.UnixMilli(ms).In(chinaLoc).Truncate(24 * time.Hour)
+}