1.改为多线程抓取数据

2.数据库表名重定义，股票代码有可能命中PostgreSQL的保留3.关键字，加市场后缀，如US或HK
4.删除了插入数据时存在检查以提高效率
5.将 io/ioutil 的使用移除，并替换为 Go 1.16+ 推荐的实现

README.md

@@ -18,9 +18,9 @@
 一个基于长桥证券接口（LongPort OpenAPI）的抓取工具，按配置批量抓取指定股票在指定日期的历史 K 线并写入 PostgreSQL。支持多只股票、多日期、去重插入、进度显示。
 
 ### 功能特性
-- 批量抓取指定标的的历史 K 线（分钟级）
+- 批量多线程抓取指定标的的历史 K 线（分钟级）
 - 多日期批处理，自动去重（`ON CONFLICT DO NOTHING`）
-- 自动根据标的生成数据表（如 `AAPL.US` → `aapl`）
+- 自动根据标的生成数据表（如 `AAPL.US` → `aapl_us`）
 - 支持 `go run` 与二进制部署两种运行方式
 
 ### 环境要求
@@ -99,6 +99,9 @@ longport:
   app_key: "xxxxx"
   app_secret: "xxxxx"
   access_token: "xxxxx"
+  region: "cn"
+  threads: 5
+  rps: 10
 ```
 也可用环境变量（PowerShell）：
 ```powershell
@@ -108,12 +111,12 @@ setx LONGPORT_ACCESS_TOKEN "your-access-token"
 ```
 
 ### 数据库表结构与写入规则
-- 表名：取股票代码点号前部分并转小写。例如 `AAPL.US` → `aapl`。
+- 表名：取股票代码点号前部分并转小写，加上下划线分隔符。例如 `AAPL.US` → `aapl_us`。
 - 去重：以 `timestamp` 为唯一键，`ON CONFLICT (timestamp) DO NOTHING` 跳过重复。
 
-示例表结构（以 `aapl` 为例）：
+示例表结构（以 `aapl_us` 为例）：
 ```sql
-CREATE TABLE IF NOT EXISTS aapl (
+CREATE TABLE IF NOT EXISTS aapl_us (
   timestamp   TIMESTAMPTZ PRIMARY KEY,
   open        DOUBLE PRECISION,
   close       DOUBLE PRECISION,
@@ -149,9 +152,9 @@ LongPort 不针对接口服务额外收取开通或使用费用，只需开通 L
 A data fetcher powered by LongPort OpenAPI. It batches historical K-line (candlestick) data for specified symbols and dates, and writes into PostgreSQL. Supports multiple symbols, multiple dates, de-dup insert, and progress display.
 
 ### Features
-- Batch fetch minute-level historical K-line for selected symbols
+- Batch multi-thread fetch minute-level historical K-line for selected symbols
 - Multi-date processing with de-dup (`ON CONFLICT DO NOTHING`)
-- Auto-create table per symbol (e.g. `AAPL.US` → `aapl`)
+- Auto-create table per symbol (e.g. `AAPL.US` → `aapl_us`)
 - Works with both `go run` and compiled binary deployments
 
 ### Requirements
@@ -230,6 +233,9 @@ longport:
   app_key: "xxxxx"
   app_secret: "xxxxx"
   access_token: "xxxxx"
+  region: "cn"
+  threads: 5
+  rps: 10
 ```
 Or use environment variables (PowerShell):
 ```powershell
@@ -239,12 +245,12 @@ setx LONGPORT_ACCESS_TOKEN "your-access-token"
 ```
 
 ### Table Schema & Insert Rules
-- Table name: take the part before the dot and lowercase (e.g. `AAPL.US` → `aapl`).
+- Table name: take the part before the dot and lowercase, with underscore separator (e.g. `AAPL.US` → `aapl_us`).
 - De-dup: `timestamp` as primary key; `ON CONFLICT (timestamp) DO NOTHING`.
 
-Example (for `aapl`):
+Example (for `aapl_us`):
 ```sql
-CREATE TABLE IF NOT EXISTS aapl (
+CREATE TABLE IF NOT EXISTS aapl_us (
   timestamp   TIMESTAMPTZ PRIMARY KEY,
   open        DOUBLE PRECISION,
   close       DOUBLE PRECISION,

appconfig/config.go

@@ -2,7 +2,7 @@ package appconfig
 
 import (
 	"fmt"
-	"io/ioutil"
+	"os"
 	"time"
 
 	"gopkg.in/yaml.v3"
@@ -20,7 +20,7 @@ type Config struct {
 
 // Load 从YAML文件加载配置
 func Load(filepath string) (*Config, error) {
-	data, err := ioutil.ReadFile(filepath)
+	data, err := os.ReadFile(filepath)
 	if err != nil {
 		return nil, fmt.Errorf("读取配置文件失败: %v", err)
 	}
@@ -45,4 +45,4 @@ func (c *Config) ParseDates() ([]time.Time, error) {
 		dates = append(dates, date)
 	}
 	return dates, nil
-}
+}

appconfig/longport.go

@@ -0,0 +1,70 @@
+package appconfig
+
+import (
+	"os"
+
+	"gopkg.in/yaml.v3"
+)
+
+// AccountConfig 表示单个账号的配置（单账号模式）
+type AccountConfig struct {
+	Name        string
+	AppKey      string
+	AppSecret   string
+	AccessToken string
+	Region      string
+	Threads     int
+	RPS         int
+	TimeoutMS   int
+}
+
+// longportYAML 用于解析 longport.yaml 文件（仅支持单账号）
+type longportYAML struct {
+	Longport struct {
+		AppKey      string `yaml:"app_key"`
+		AppSecret   string `yaml:"app_secret"`
+		AccessToken string `yaml:"access_token"`
+		Region      string `yaml:"region"`
+		Threads     int    `yaml:"threads"`
+		RPS         int    `yaml:"rps"`
+		TimeoutMS   int    `yaml:"timeout_ms"`
+	} `yaml:"longport"`
+}
+
+// LoadLongportAccount 读取 longport.yaml 并返回单账号配置
+func LoadLongportAccount(path string) (AccountConfig, error) {
+	var out AccountConfig
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return out, err
+	}
+	var cfg longportYAML
+	if err := yaml.Unmarshal(data, &cfg); err != nil {
+		return out, err
+	}
+
+	threads := cfg.Longport.Threads
+	if threads <= 0 {
+		threads = 5
+	}
+	rps := cfg.Longport.RPS
+	if rps <= 0 {
+		rps = 10
+	}
+	timeout := cfg.Longport.TimeoutMS
+	if timeout <= 0 {
+		timeout = 10000
+	}
+
+	out = AccountConfig{
+		Name:        "default",
+		AppKey:      cfg.Longport.AppKey,
+		AppSecret:   cfg.Longport.AppSecret,
+		AccessToken: cfg.Longport.AccessToken,
+		Region:      cfg.Longport.Region,
+		Threads:     threads,
+		RPS:         rps,
+		TimeoutMS:   timeout,
+	}
+	return out, nil
+}

config/config.yaml

@@ -6,10 +6,9 @@ symbols:
 
 # 查询日期列表 (格式: YYYY-MM-DD)
 dates:
-  - "2025-10-24"
-  - "2025-10-23"
-
+  - "2025-10-28"
+  - "2025-10-29"
 # 查询设置
 settings:
   period: "OneMinute"  # 可选: OneMinute, FiveMinute, FifteenMinute, ThirtyMinute
-  adjust_type: "No"    # 可选: No, ForwardAdjust
+  adjust_type: "No"    # 可选: No, ForwardAdjust

config/longport.yaml

@@ -1,10 +1,8 @@
 longport:
-  app_key: ""
-  app_secret: ""
-  access_token: ""
-  # Optional overrides:
-  # quote_url: ""
-  # trade_url: ""
-  # enable_overnight: false
-  # language: 0  # 0: zh-CN, 1: zh-TW, 2: en
-  # region: "cn"
+  app_key: "YOUR_APP_KEY"
+  app_secret: "YOUR_APP_SECRET"
+  access_token: "YOUR_ACCESS_TOKEN"
+  region: "cn"      # HK 或 CN 
+  threads: 5        # 每账号并发线程（最大 5）
+  rps: 10           # 每账号每秒最大请求数（最大 10）
+  timeout_ms: 10000 # 每账号请求超时毫秒（默认 10000）

dbconn/dbconn.go

@@ -75,4 +75,4 @@ func OpenFromFile(path string) (*sql.DB, error) {
 		return nil, err
 	}
 	return Open(cfg)
-}
+}

go.mod

@@ -7,6 +7,8 @@ require (
 	github.com/longportapp/openapi-go v0.16.4
 )
 
+require golang.org/x/time v0.14.0 // indirect
+
 require (
 	github.com/Allenxuxu/ringbuffer v0.0.11 // indirect
 	github.com/BurntSushi/toml v1.3.2 // indirect

go.sum

@@ -42,6 +42,8 @@ github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5g
 github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
+golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

main.go

@@ -8,6 +8,8 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"getk/appconfig"
@@ -27,181 +29,55 @@ type Candlestick struct {
 	Volume    int64
 	Turnover  float64
 }
+type job struct {
+	symbol string
+	date   time.Time
+}
 
-// 移除 GetOrCreateTable，改为直接根据 symbol 推导表名
-func InsertCandlesticks(db *sql.DB, symbol string, candlesticks []Candlestick) error {
-	tableName := strings.Split(symbol, ".")[0]
-	tableName = strings.ToLower(tableName)
+// 使用 appconfig.AccountConfig 的类型别名，避免重复定义
+type AccountConfig = appconfig.AccountConfig
 
-	// Prepare existence check and insert statements for efficiency
-	existsStmt, err := db.Prepare(fmt.Sprintf(`SELECT EXISTS(SELECT 1 FROM %s WHERE timestamp = $1)`, tableName))
-	if err != nil {
-		return err
-	}
-	defer existsStmt.Close()
+type RetryConfig struct {
+	MaxAttempts int
+	BaseDelayMS int
+	MaxDelayMS  int
+}
 
-	insertStmt, err := db.Prepare(fmt.Sprintf(
-		`INSERT INTO %s (timestamp, open, close, high, low, volume, turnover)
-		 VALUES ($1, $2, $3, $4, $5, $6, $7)
-		 ON CONFLICT (timestamp) DO NOTHING`,
-		tableName,
-	))
-	if err != nil {
-		return err
-	}
-	defer insertStmt.Close()
+type FetchResult struct {
+	Account string
+	Worker  int
+	Symbol  string
+	Ok      bool
+	Err     error
+	Records int
+	Elapsed time.Duration
+}
 
-	for _, c := range candlesticks {
-		var exists bool
-		if err := existsStmt.QueryRow(c.Timestamp).Scan(&exists); err != nil {
-			log.Printf("Existence check failed: %v\n", err)
-			continue
-		}
-		if exists {
-			// Skip duplicates
-			continue
-		}
-
-		_, err := insertStmt.Exec(c.Timestamp, c.Open, c.Close, c.High, c.Low, c.Volume, c.Turnover)
-		if err != nil {
-			log.Printf("Insert failed: %v\n", err)
-		}
-	}
-
-	return nil
+// 统一生成安全的表名，避免与保留关键字冲突
+func safeTableName(symbol string) string {
+	base := strings.Split(symbol, ".")[0]
+	region := strings.Split(symbol, ".")[1]
+	name := strings.ToLower(base)
+	regionName := strings.ToLower(region)
+	return name + "_" + regionName
 }
 
 // 自动创建数据表（若不存在）
 func EnsureTable(db *sql.DB, symbol string) error {
-	tableName := strings.Split(symbol, ".")[0]
-	tableName = strings.ToLower(tableName)
+	tableName := safeTableName(symbol)
 	createSQL := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
-		timestamp   TIMESTAMPTZ PRIMARY KEY,
-		open        DOUBLE PRECISION,
-		close       DOUBLE PRECISION,
-		high        DOUBLE PRECISION,
-		low         DOUBLE PRECISION,
-		volume      BIGINT,
-		turnover    DOUBLE PRECISION
-	)`, tableName)
+timestamp   TIMESTAMPTZ PRIMARY KEY,
+open        DOUBLE PRECISION,
+close       DOUBLE PRECISION,
+high        DOUBLE PRECISION,
+low         DOUBLE PRECISION,
+volume      BIGINT,
+turnover    DOUBLE PRECISION
+)`, tableName)
 	_, err := db.Exec(createSQL)
 	return err
 }
 
-func main() {
-	// 加载应用配置
-	appCfg, err := appconfig.Load(configPath("config.yaml"))
-	if err != nil {
-		log.Fatalf("读取应用配置失败: %v", err)
-	}
-
-	// 解析日期
-	dates, err := appCfg.ParseDates()
-	if err != nil {
-		log.Fatalf("解析日期失败: %v", err)
-	}
-
-	// 通过独立包加载数据库配置并建立连接
-	dbCfg, err := dbconn.Load(configPath("db.yaml"))
-	if err != nil {
-		log.Fatalf("读取数据库配置失败: %v", err)
-	}
-	db, err := dbconn.Open(dbCfg)
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer db.Close()
-
-	// 通过配置文件加载 Longbridge 配置 (YAML/TOML)
-	conf, err := config.New(config.WithFilePath(configPath("longport.yaml")))
-	if err != nil {
-		log.Fatalf("加载配置文件失败: %v", err)
-	}
-
-	qctx, err := quote.NewFromCfg(conf)
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer qctx.Close()
-
-	// 获取期间类型
-	period := getPeriodFromConfig(appCfg.Settings.Period)
-	adjustType := getAdjustTypeFromConfig(appCfg.Settings.AdjustType)
-
-	fmt.Printf("开始数据获取任务...\n")
-	fmt.Printf("配置的股票数量: %d\n", len(appCfg.Symbols))
-	fmt.Printf("配置的日期数量: %d\n", len(dates))
-	fmt.Printf("总任务数: %d\n\n", len(appCfg.Symbols)*len(dates))
-
-	totalTasks := len(appCfg.Symbols) * len(dates)
-	currentTask := 0
-
-	// 循环处理每个股票的每个日期
-	for _, symbol := range appCfg.Symbols {
-		fmt.Printf("📈 开始处理股票: %s\n", symbol)
-
-		// 确保目标表存在（自动创建）
-		if err := EnsureTable(db, symbol); err != nil {
-			fmt.Printf("  ❌ 创建表失败: %v\n", err)
-			continue
-		}
-
-		for _, date := range dates {
-			currentTask++
-
-			// 显示进度
-			fmt.Printf("  [%d/%d] 正在查询 %s 的 %s 数据...",
-				currentTask, totalTasks, symbol, date.Format("2006-01-02"))
-
-			// 设置查询时间范围（一整天）
-			start := time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, time.UTC)
-			end := time.Date(date.Year(), date.Month(), date.Day(), 23, 59, 59, 0, time.UTC)
-
-			// 获取K线数据
-			candlesticks, err := qctx.HistoryCandlesticksByDate(
-				context.Background(),
-				symbol,
-				period,
-				adjustType,
-				&start,
-				&end,
-			)
-			if err != nil {
-				fmt.Printf(" ❌ 失败: %v\n", err)
-				continue
-			}
-
-			// 转换数据格式
-			var records []Candlestick
-			for _, c := range candlesticks {
-				records = append(records, Candlestick{
-					Symbol:    symbol,
-					Timestamp: time.Unix(c.Timestamp, 0),
-					Open:      c.Open.InexactFloat64(),
-					Close:     c.Close.InexactFloat64(),
-					High:      c.High.InexactFloat64(),
-					Low:       c.Low.InexactFloat64(),
-					Volume:    c.Volume,
-					Turnover:  c.Turnover.InexactFloat64(),
-				})
-			}
-
-			// 插入数据
-			err = InsertCandlesticks(db, symbol, records)
-			if err != nil {
-				fmt.Printf(" ❌ 数据库插入失败: %v\n", err)
-				continue
-			}
-
-			fmt.Printf(" ✅ 完成 (获取 %d 条记录)\n", len(records))
-		}
-		fmt.Printf("✅ 股票 %s 处理完成\n\n", symbol)
-	}
-
-	fmt.Printf("🎉 所有数据已保存到PostgreSQL！\n")
-	fmt.Printf("总共处理了 %d 个任务\n", totalTasks)
-}
-
 // getPeriodFromConfig 根据配置字符串返回对应的Period类型
 func getPeriodFromConfig(periodStr string) quote.Period {
 	switch periodStr {
@@ -257,3 +133,211 @@ func configPath(filename string) string {
 	// 4) 最后回退到相对路径，交由后续读取时报错并提示
 	return filepath.Join(".", "config", filename)
 }
+
+func InsertCandlesticks(db *sql.DB, symbol string, candlesticks []Candlestick) error {
+	tableName := safeTableName(symbol)
+	insertStmt, err := db.Prepare(fmt.Sprintf(
+		`INSERT INTO %s (timestamp, open, close, high, low, volume, turnover)
+ VALUES ($1, $2, $3, $4, $5, $6, $7)
+ ON CONFLICT (timestamp) DO NOTHING`,
+		tableName,
+	))
+	if err != nil {
+		return err
+	}
+	defer insertStmt.Close()
+
+	for _, c := range candlesticks {
+		_, err := insertStmt.Exec(c.Timestamp, c.Open, c.Close, c.High, c.Low, c.Volume, c.Turnover)
+		if err != nil {
+			log.Printf("Insert failed: %v\n", err)
+		}
+	}
+
+	return nil
+}
+
+func main() {
+	// 加载应用配置
+	appCfg, err := appconfig.Load(configPath("config.yaml"))
+	if err != nil {
+		log.Fatalf("读取应用配置失败: %v", err)
+	}
+
+	// 解析日期
+	dates, err := appCfg.ParseDates()
+	if err != nil {
+		log.Fatalf("解析日期失败: %v", err)
+	}
+
+	// 通过独立包加载数据库配置并建立连接
+	dbCfg, err := dbconn.Load(configPath("db.yaml"))
+	if err != nil {
+		log.Fatalf("读取数据库配置失败: %v", err)
+	}
+	db, err := dbconn.Open(dbCfg)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer db.Close()
+
+	// 获取期间类型
+	period := getPeriodFromConfig(appCfg.Settings.Period)
+	adjustType := getAdjustTypeFromConfig(appCfg.Settings.AdjustType)
+
+	fmt.Printf("开始数据获取任务...\n")
+	fmt.Printf("配置的股票数量: %d\n", len(appCfg.Symbols))
+	fmt.Printf("配置的日期数量: %d\n", len(dates))
+	fmt.Printf("总任务数: %d\n\n", len(appCfg.Symbols)*len(dates))
+
+	// 预先确保所有表存在（可并发安全，若不存在则创建）
+	for _, symbol := range appCfg.Symbols {
+		if err := EnsureTable(db, symbol); err != nil {
+			fmt.Printf("   创建表失败: %v\n", err)
+		}
+	}
+
+	// 单账号模式：加载 longport.yaml
+	acc, err := appconfig.LoadLongportAccount(configPath("longport.yaml"))
+	if err != nil {
+		log.Fatalf("读取 longport.yaml 失败: %v", err)
+	}
+
+	// 创建单账号的 QuoteContext（通过临时配置文件）
+	tmpPath := filepath.Join(os.TempDir(), "longport_single.yaml")
+	content := fmt.Sprintf(
+		"longport:\n  app_key: %q\n  app_secret: %q\n  access_token: %q\n  region: %q\n",
+		acc.AppKey, acc.AppSecret, acc.AccessToken, acc.Region,
+	)
+	if err := os.WriteFile(tmpPath, []byte(content), 0600); err != nil {
+		log.Fatalf("写入临时配置失败: %v", err)
+	}
+	confAcc, err := config.New(config.WithFilePath(tmpPath))
+	if err != nil {
+		log.Fatalf("加载行情配置失败: %v", err)
+	}
+	accCtx, err := quote.NewFromCfg(confAcc)
+	if err != nil {
+		log.Fatalf("创建行情上下文失败: %v", err)
+	}
+	defer accCtx.Close()
+
+	// 重试策略（用于 SDK 拉取失败时的指数退避重试）
+	retryCfg := RetryConfig{MaxAttempts: 3, BaseDelayMS: 500, MaxDelayMS: 2000}
+
+	// 构建任务队列（所有 symbols x dates）
+	totalTasks := len(appCfg.Symbols) * len(dates)
+	var currentTask int64
+	jobs := make(chan job, totalTasks)
+	for _, symbol := range appCfg.Symbols {
+		for _, date := range dates {
+			jobs <- job{symbol: symbol, date: date}
+		}
+	}
+	close(jobs)
+
+	// 账号级速率限制器
+	rps := acc.RPS
+	if rps <= 0 {
+		rps = 10
+	}
+	interval := time.Second / time.Duration(rps)
+	limiter := time.NewTicker(interval)
+	defer limiter.Stop()
+
+	// 启动 worker 池
+	threads := acc.Threads
+	if threads <= 0 {
+		threads = 5
+	}
+	var wg sync.WaitGroup
+	var successCount int64
+	var failCount int64
+	wg.Add(threads)
+	for w := 0; w < threads; w++ {
+		workerID := w + 1
+		go func(workerID int, jobs <-chan job) {
+			defer wg.Done()
+			for j := range jobs {
+				t0 := time.Now()
+				<-limiter.C
+				cur := atomic.AddInt64(&currentTask, 1)
+				fmt.Printf("  [进度=%d/%d] 线程=%d 正在查询 %s 的 %s 数据...\n", cur, totalTasks, workerID, j.symbol, j.date.Format("2006-01-02"))
+
+				start := time.Date(j.date.Year(), j.date.Month(), j.date.Day(), 0, 0, 0, 0, time.UTC)
+				end := time.Date(j.date.Year(), j.date.Month(), j.date.Day(), 23, 59, 59, 0, time.UTC)
+
+				var fetched bool
+				for attempt := 1; attempt <= retryCfg.MaxAttempts; attempt++ {
+					candlesticks, err := accCtx.HistoryCandlesticksByDate(
+						context.Background(),
+						j.symbol,
+						period,
+						adjustType,
+						&start,
+						&end,
+					)
+					if err != nil {
+						if attempt < retryCfg.MaxAttempts {
+							base := retryCfg.BaseDelayMS * (1 << (attempt - 1))
+							if base > retryCfg.MaxDelayMS {
+								base = retryCfg.MaxDelayMS
+							}
+							backoff := time.Duration(base) * time.Millisecond
+							fmt.Printf("  线程=%d 重试 %d/%d (错误: %v, 等待=%dms)\n", workerID, attempt, retryCfg.MaxAttempts, err, backoff.Milliseconds())
+							time.Sleep(backoff)
+							continue
+						}
+						elapsed := time.Since(t0)
+						atomic.AddInt64(&failCount, 1)
+						fmt.Printf("  线程=%d 失败 (耗时=%dms): %v\n", workerID, elapsed.Milliseconds(), err)
+						break
+					}
+
+					var records []Candlestick
+					for _, c := range candlesticks {
+						if c.Open == nil || c.Close == nil || c.High == nil || c.Low == nil {
+							continue
+						}
+						turnover := 0.0
+						if c.Turnover != nil {
+							turnover = c.Turnover.InexactFloat64()
+						}
+						records = append(records, Candlestick{
+							Symbol:    j.symbol,
+							Timestamp: time.Unix(c.Timestamp, 0),
+							Open:      c.Open.InexactFloat64(),
+							Close:     c.Close.InexactFloat64(),
+							High:      c.High.InexactFloat64(),
+							Low:       c.Low.InexactFloat64(),
+							Volume:    c.Volume,
+							Turnover:  turnover,
+						})
+					}
+
+					if err := InsertCandlesticks(db, j.symbol, records); err != nil {
+						elapsed := time.Since(t0)
+						fmt.Printf("  线程=%d 数据库插入失败 (耗时=%dms): %v\n", workerID, elapsed.Milliseconds(), err)
+						atomic.AddInt64(&failCount, 1)
+						break
+					}
+
+					elapsed := time.Since(t0)
+					fmt.Printf("  线程=%d 完成 (记录=%d, 耗时=%dms)\n", workerID, len(records), elapsed.Milliseconds())
+					atomic.AddInt64(&successCount, 1)
+					fetched = true
+					break
+				}
+
+				if !fetched {
+					// 已在失败路径计数与打印
+				}
+			}
+		}(workerID, jobs)
+	}
+
+	wg.Wait()
+	fmt.Printf(" 所有数据已保存到PostgreSQL！\n")
+	fmt.Printf("成功=%d 失败=%d 总任务=%d\n", successCount, failCount, totalTasks)
+
+}