package datasource import ( "bytes" "encoding/json" "fmt" "io" "net/http" "strings" "time" "stock-user-system/internal/config" ) const ( defaultTimeout = 60 * time.Second maxRetries = 3 retryBaseDelay = 1 * time.Second ) // Quote 对应 tickflow /v1/quotes 返回的单条行情数据。 type Quote struct { Symbol string `json:"symbol"` Name string `json:"name"` Open float64 `json:"open"` High float64 `json:"high"` Low float64 `json:"low"` Close float64 `json:"last_price"` PrevClose float64 `json:"prev_close"` Volume int64 `json:"volume"` Amount float64 `json:"amount"` Timestamp int64 `json:"timestamp"` Region string `json:"region"` Ext struct { ChangePct float64 `json:"change_pct"` TurnoverRate float64 `json:"turnover_rate"` Name string `json:"name"` } `json:"ext"` } // ChangePct 优先使用 ext.change_pct(tickflow 返回小数,如 0.30 = 30%), // 否则根据 close/prev_close 计算并统一返回小数形式。 func (q *Quote) ChangePct() float64 { if q.Ext.ChangePct != 0 { return q.Ext.ChangePct } if q.PrevClose != 0 { return (q.Close - q.PrevClose) / q.PrevClose } return 0 } // TurnoverRate 返回 ext.turnover_rate(仅 A 股有效)。 func (q *Quote) TurnoverRate() float64 { return q.Ext.TurnoverRate } // DisplayName 优先使用 ext.name,否则使用 symbol。 func (q *Quote) DisplayName() string { if q.Ext.Name != "" { return q.Ext.Name } if q.Name != "" { return q.Name } return q.Symbol } // UniverseDetail 对应 /v1/universes/:id 返回的标的池详情。 type UniverseDetail struct { ID string `json:"id"` Name string `json:"name"` SymbolCount int `json:"symbol_count"` 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 apiKey string client *http.Client } // NewClient 从配置创建 tickflow 客户端。 func NewClient(cfg *config.Config) *Client { return &Client{ baseURL: strings.TrimRight(cfg.TickFlowBaseURL, "/"), apiKey: cfg.TickFlowAPIKey, client: &http.Client{Timeout: defaultTimeout}, } } // GetUniverse 获取指定标的池的完整代码列表。 func (c *Client) GetUniverse(id string) (*UniverseDetail, error) { path := fmt.Sprintf("/v1/universes/%s", id) var detail UniverseDetail if err := c.request("GET", path, nil, nil, &detail); err != nil { return nil, err } return &detail, nil } // GetQuotesByUniverses 按标的池批量获取行情快照。 func (c *Client) GetQuotesByUniverses(universes []string) ([]Quote, error) { body := map[string]any{"universes": universes} var quotes []Quote if err := c.request("POST", "/v1/quotes", nil, body, "es); err != nil { return nil, err } return quotes, nil } // GetQuotesBySymbols 按代码列表批量获取行情快照。 func (c *Client) GetQuotesBySymbols(symbols []string) ([]Quote, error) { body := map[string]any{"symbols": symbols} var quotes []Quote if err := c.request("POST", "/v1/quotes", nil, body, "es); err != nil { return nil, err } 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 var lastErr error for attempt := 0; attempt <= maxRetries; attempt++ { if attempt > 0 { time.Sleep(retryBaseDelay * time.Duration(1<<(attempt-1))) } var bodyReader io.Reader if len(body) > 0 && (method == "POST" || method == "PUT") { data, err := json.Marshal(body) if err != nil { return err } bodyReader = bytes.NewReader(data) } req, err := http.NewRequest(method, url, bodyReader) if err != nil { return err } if c.apiKey != "" { req.Header.Set("x-api-key", c.apiKey) } req.Header.Set("Accept", "application/json") req.Header.Set("Content-Type", "application/json") if len(params) > 0 { q := req.URL.Query() for k, v := range params { q.Set(k, fmt.Sprintf("%v", v)) } req.URL.RawQuery = q.Encode() } resp, err := c.client.Do(req) if err != nil { lastErr = err continue } respBody, err := io.ReadAll(resp.Body) resp.Body.Close() if err != nil { lastErr = err continue } if resp.StatusCode >= 500 || resp.StatusCode == 429 { lastErr = fmt.Errorf("tickflow %s %s returned %d: %s", method, path, resp.StatusCode, string(respBody)) continue } if resp.StatusCode >= 400 { return fmt.Errorf("tickflow %s %s returned %d: %s", method, path, resp.StatusCode, string(respBody)) } var wrapper struct { Data json.RawMessage `json:"data"` } if err := json.Unmarshal(respBody, &wrapper); err != nil { return fmt.Errorf("decode tickflow response: %w", err) } if result != nil { if err := json.Unmarshal(wrapper.Data, result); err != nil { return fmt.Errorf("decode tickflow data: %w", err) } } return nil } if lastErr != nil { return fmt.Errorf("tickflow request failed after %d retries: %w", maxRetries, lastErr) } return fmt.Errorf("tickflow request failed") }