143 lines
3.5 KiB
Go
143 lines
3.5 KiB
Go
package services
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"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")
|
|
}
|
|
|
|
log.Printf("[stock sync] fetched %d stock quotes from CN_Equity_A, %d index quotes, total %d",
|
|
len(stockQuotes), len(indexQuotes), len(quotes))
|
|
|
|
// 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)
|
|
}
|