From ea5b24d0cff811a0665ef084cbc8da43c5174928 Mon Sep 17 00:00:00 2001 From: dennis Date: Fri, 31 Oct 2025 01:23:54 +0800 Subject: [PATCH] =?UTF-8?q?1.=E6=94=B9=E4=B8=BA=E5=A4=9A=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E6=8A=93=E5=8F=96=E6=95=B0=E6=8D=AE=202.=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93=E8=A1=A8=E5=90=8D=E9=87=8D=E5=AE=9A=E4=B9=89=EF=BC=8C?= =?UTF-8?q?=E8=82=A1=E7=A5=A8=E4=BB=A3=E7=A0=81=E6=9C=89=E5=8F=AF=E8=83=BD?= =?UTF-8?q?=E5=91=BD=E4=B8=ADPostgreSQL=E7=9A=84=E4=BF=9D=E7=95=993.?= =?UTF-8?q?=E5=85=B3=E9=94=AE=E5=AD=97=EF=BC=8C=E5=8A=A0=E5=B8=82=E5=9C=BA?= =?UTF-8?q?=E5=90=8E=E7=BC=80=EF=BC=8C=E5=A6=82US=E6=88=96HK=204.=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E4=BA=86=E6=8F=92=E5=85=A5=E6=95=B0=E6=8D=AE=E6=97=B6?= =?UTF-8?q?=E5=AD=98=E5=9C=A8=E6=A3=80=E6=9F=A5=E4=BB=A5=E6=8F=90=E9=AB=98?= =?UTF-8?q?=E6=95=88=E7=8E=87=205.=E5=B0=86=20io/ioutil=20=E7=9A=84?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E7=A7=BB=E9=99=A4=EF=BC=8C=E5=B9=B6=E6=9B=BF?= =?UTF-8?q?=E6=8D=A2=E4=B8=BA=20Go=201.16+=20=E6=8E=A8=E8=8D=90=E7=9A=84?= =?UTF-8?q?=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 26 +-- appconfig/config.go | 6 +- appconfig/longport.go | 70 ++++++++ config/config.yaml | 7 +- config/longport.yaml | 16 +- dbconn/dbconn.go | 2 +- go.mod | 2 + go.sum | 2 + main.go | 408 +++++++++++++++++++++++++----------------- 9 files changed, 350 insertions(+), 189 deletions(-) create mode 100644 appconfig/longport.go diff --git a/README.md b/README.md index 413e508..02e5b38 100644 --- a/README.md +++ b/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, diff --git a/appconfig/config.go b/appconfig/config.go index 0f7ddc9..98224a3 100644 --- a/appconfig/config.go +++ b/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 -} \ No newline at end of file +} diff --git a/appconfig/longport.go b/appconfig/longport.go new file mode 100644 index 0000000..1d02885 --- /dev/null +++ b/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 +} diff --git a/config/config.yaml b/config/config.yaml index 9f8fc8f..631cec0 100644 --- a/config/config.yaml +++ b/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 \ No newline at end of file + adjust_type: "No" # 可选: No, ForwardAdjust diff --git a/config/longport.yaml b/config/longport.yaml index 0f5a483..c46df5c 100644 --- a/config/longport.yaml +++ b/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" \ No newline at end of file + 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） \ No newline at end of file diff --git a/dbconn/dbconn.go b/dbconn/dbconn.go index 45b048d..6045d2b 100644 --- a/dbconn/dbconn.go +++ b/dbconn/dbconn.go @@ -75,4 +75,4 @@ func OpenFromFile(path string) (*sql.DB, error) { return nil, err } return Open(cfg) -} \ No newline at end of file +} diff --git a/go.mod b/go.mod index 0f4b0ad..514ba48 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3b6633b..a4beb8a 100644 --- a/go.sum +++ b/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= diff --git a/main.go b/main.go index f8b30c1..2096a2f 100644 --- a/main.go +++ b/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(¤tTask, 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) + +}