系统管理员点击同步时同步半年历史数据
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user