优化GetHistoryTradeFull,需要传入Workday,减少非工作日的查询操作

client.go

@@ -440,7 +440,12 @@ func (this *Client) GetHistoryMinuteTrade(date, code string, start, count uint16
 }
 
 // GetHistoryTradeFull 获取上市至今的分时成交
-func (this *Client) GetHistoryTradeFull(code string) (protocol.Trades, error) {
+func (this *Client) GetHistoryTradeFull(code string, w *Workday) (protocol.Trades, error) {
+	return this.GetHistoryTradeBefore(code, w, time.Now())
+}
+
+// GetHistoryTradeBefore 获取上市至今的分时成交
+func (this *Client) GetHistoryTradeBefore(code string, w *Workday, before time.Time) (protocol.Trades, error) {
 	ls := protocol.Trades(nil)
 	resp, err := this.GetKlineMonthAll(code)
 	if err != nil {
@@ -451,14 +456,15 @@ func (this *Client) GetHistoryTradeFull(code string) (protocol.Trades, error) {
 	}
 	start := time.Date(resp.List[0].Time.Year(), resp.List[0].Time.Month(), 1, 0, 0, 0, 0, resp.List[0].Time.Location())
 	var res *protocol.TradeResp
-	for ; start.Before(time.Now()); start = start.Add(time.Hour * 24) {
-		res, err = this.GetHistoryTradeDay(start.Format("20060102"), code)
+	w.Range(start, before, func(t time.Time) bool {
+		res, err = this.GetHistoryTradeDay(t.Format("20060102"), code)
 		if err != nil {
-			return nil, err
+			return false
 		}
 		ls = append(ls, res.List...)
-	}
-	return ls, nil
+		return true
+	})
+	return ls, err
 }
 
 // GetHistoryTradeDay 获取历史某天分时全部交易,通过多次请求来拼接,只能获取昨天及之前的数据

codes.go

@@ -23,10 +23,22 @@ func DialCodes(filename string, op ...client.Option) (*Codes, error) {
 	if err != nil {
 		return nil, err
 	}
-	return NewCodes(c, filename)
+	return NewCodesSqlite(c, filename)
 }
 
-func NewCodes(c *Client, filenames ...string) (*Codes, error) {
+func NewCodesMysql(c *Client, dsn string) (*Codes, error) {
+
+	//连接数据库
+	db, err := xorm.NewEngine("mysql", dsn)
+	if err != nil {
+		return nil, err
+	}
+	db.SetMapper(core.SameMapper{})
+
+	return NewCodes(c, db)
+}
+
+func NewCodesSqlite(c *Client, filenames ...string) (*Codes, error) {
 
 	//如果没有指定文件名,则使用默认
 	defaultFilename := filepath.Join(DefaultDatabaseDir, "codes.db")
@@ -44,6 +56,12 @@ func NewCodes(c *Client, filenames ...string) (*Codes, error) {
 	}
 	db.SetMapper(core.SameMapper{})
 	db.DB().SetMaxOpenConns(1)
+
+	return NewCodes(c, db)
+}
+
+func NewCodes(c *Client, db *xorm.Engine) (*Codes, error) {
+
 	if err := db.Sync2(new(CodeModel)); err != nil {
 		return nil, err
 	}
@@ -53,10 +71,11 @@ func NewCodes(c *Client, filenames ...string) (*Codes, error) {
 
 	update := new(UpdateModel)
 	{ //查询或者插入一条数据
-		has, err := db.Get(update)
+		has, err := db.Where("`Key`=?", "codes").Get(update)
 		if err != nil {
 			return nil, err
 		} else if !has {
+			update.Key = "codes"
 			if _, err := db.Insert(update); err != nil {
 				return nil, err
 			}
@@ -72,7 +91,8 @@ func NewCodes(c *Client, filenames ...string) (*Codes, error) {
 		task := cron.New(cron.WithSeconds())
 		task.AddFunc("10 0 9 * * *", func() {
 			for i := 0; i < 3; i++ {
-				if err := cc.Update(); err == nil {
+				err := cc.Update()
+				if err == nil {
 					return
 				}
 				logs.Err(err)
@@ -155,53 +175,8 @@ func (this *Codes) Get(code string) *CodeModel {
 	return this.Map[code]
 }
 
-//// GetExchange 获取股票交易所,这里的参数不需要带前缀
-//func (this *Codes) GetExchange(code string) protocol.Exchange {
-//	if len(code) == 6 {
-//		switch {
-//		case code[:1] == "6":
-//			return protocol.ExchangeSH
-//		case code[:1] == "0":
-//			return protocol.ExchangeSZ
-//		case code[:2] == "30":
-//			return protocol.ExchangeSZ
-//		}
-//	}
-//	var exchange string
-//	exchanges := this.exchanges[code]
-//	if len(exchanges) >= 1 {
-//		exchange = exchanges[0]
-//	}
-//	if len(code) == 8 {
-//		exchange = code[0:2]
-//	}
-//	switch exchange {
-//	case protocol.ExchangeSH.String():
-//		return protocol.ExchangeSH
-//	case protocol.ExchangeSZ.String():
-//		return protocol.ExchangeSZ
-//	default:
-//		return protocol.ExchangeSH
-//	}
-//}
-
 func (this *Codes) AddExchange(code string) string {
 	return protocol.AddPrefix(code)
-	//if exchanges := this.exchanges[code]; len(exchanges) == 1 {
-	//	return exchanges[0] + code
-	//}
-	//if len(code) == 6 {
-	//	switch {
-	//	case code[:1] == "6":
-	//		return protocol.ExchangeSH.String() + code
-	//	case code[:1] == "0":
-	//		return protocol.ExchangeSZ.String() + code
-	//	case code[:2] == "30":
-	//		return protocol.ExchangeSZ.String() + code
-	//	}
-	//	return this.GetExchange(code).String() + code
-	//}
-	//return code
 }
 
 // Update 更新数据,从服务器或者数据库
@@ -220,7 +195,7 @@ func (this *Codes) Update(byDB ...bool) error {
 	this.list = codes
 	this.exchanges = exchanges
 	//更新时间
-	_, err = this.db.Update(&UpdateModel{Time: time.Now().Unix()})
+	_, err = this.db.Where("`Key`=?", "codes").Update(&UpdateModel{Time: time.Now().Unix()})
 	return err
 }
 
@@ -283,29 +258,52 @@ func (this *Codes) GetCodes(byDatabase bool) ([]*CodeModel, error) {
 		}
 	}
 
-	//4. 插入或者更新数据库
-	err := NewSessionFunc(this.db, func(session *xorm.Session) error {
-		for _, v := range insert {
-			if _, err := session.Insert(v); err != nil {
-				return err
+	switch this.db.Dialect().URI().DBType {
+	case "mysql":
+		// 1️⃣ 清空
+		if _, err := this.db.Exec("TRUNCATE TABLE codes"); err != nil {
+			return nil, err
+		}
+
+		data := append(insert, update...)
+		// 2️⃣ 直接批量插入
+		batchSize := 3000 // 8000(2m16s) 5000(43s) 3000(11s) 1000(59s)
+		for i := 0; i < len(data); i += batchSize {
+			end := i + batchSize
+			if end > len(data) {
+				end = len(data)
+			}
+
+			slice := conv.Array(data[i:end])
+			if _, err := this.db.Insert(slice); err != nil {
+				return nil, err
 			}
 		}
-		for _, v := range update {
-			if _, err := session.Where("Exchange=? and Code=? ", v.Exchange, v.Code).Cols("Name,LastPrice").Update(v); err != nil {
-				return err
+	case "sqlite3":
+		//4. 插入或者更新数据库
+		err := NewSessionFunc(this.db, func(session *xorm.Session) error {
+			for _, v := range insert {
+				if _, err := session.Insert(v); err != nil {
+					return err
+				}
 			}
+			for _, v := range update {
+				if _, err := session.Where("Exchange=? and Code=? ", v.Exchange, v.Code).Cols("Name,LastPrice").Update(v); err != nil {
+					return err
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, err
 		}
-		return nil
-	})
-	if err != nil {
-		return nil, err
 	}
 
 	return list, nil
-
 }
 
 type UpdateModel struct {
+	Key  string
 	Time int64 //更新时间
 }
 

example/GetHistoryTradeAll/main.go

@@ -8,7 +8,7 @@ import (
 
 func main() {
 	common.Test(func(c *tdx.Client) {
-		resp, err := c.GetHistoryMinuteTradeDay("20241025", "sz000001")
+		resp, err := c.GetHistoryMinuteTradeDay("20251010", "sh000001")
 		logs.PanicErr(err)
 
 		for _, v := range resp.List {

example/GetQuote/main.go

@@ -10,7 +10,7 @@ func main() {
 	c, err := tdx.Dial("124.71.187.122:7709", tdx.WithDebug())
 	logs.PanicErr(err)
 
-	tdx.DefaultCodes, err = tdx.NewCodes(c, "./codes.db")
+	tdx.DefaultCodes, err = tdx.NewCodesSqlite(c, "./codes.db")
 	logs.PanicErr(err)
 
 	_ = c

example/ManageMysql/main.go

@@ -0,0 +1,17 @@
+package main
+
+import (
+	"github.com/injoyai/logs"
+	"github.com/injoyai/tdx"
+)
+
+func main() {
+	_, err := tdx.NewManageMysql(&tdx.ManageConfig{
+		Number:          2,
+		CodesFilename:   "root:root@tcp(192.168.1.105:3306)/stock?charset=utf8mb4&parseTime=True&loc=Local",
+		WorkdayFileName: "root:root@tcp(192.168.1.105:3306)/stock?charset=utf8mb4&parseTime=True&loc=Local",
+		Dial:            nil,
+	})
+	logs.PanicErr(err)
+	logs.Debug("done")
+}

example/TradesToKlines/main.go

@@ -9,7 +9,7 @@ import (
 func main() {
 	common.Test(func(c *tdx.Client) {
 
-		resp, err := c.GetHistoryTradeAll("20251010", "sz000001")
+		resp, err := c.GetHistoryTradeDay("20251010", "sz000001")
 		logs.PanicErr(err)
 
 		ks := resp.List.Klines()

example/WorkdayAndCodes/main.go

@@ -9,10 +9,10 @@ import (
 func main() {
 	common.Test(func(c *tdx.Client) {
 
-		_, err := tdx.NewWorkday(c) //"./workday.db"
+		_, err := tdx.NewWorkdaySqlite(c) //"./workday.db"
 		logs.PanicErr(err)
 
-		_, err = tdx.NewCodes(c) //"./codes.db"
+		_, err = tdx.NewCodesSqlite(c) //"./codes.db"
 		logs.PanicErr(err)
 
 		c.Close()

extend/pull-trade.go

@@ -56,7 +56,7 @@ func (this *PullTrade) PullYear(ctx context.Context, m *tdx.Manage, year int, co
 
 		var resp *protocol.TradeResp
 		err = m.Do(func(c *tdx.Client) error {
-			resp, err = c.GetHistoryTradeAll(date, code)
+			resp, err = c.GetHistoryTradeDay(date, code)
 			return err
 		})
 		if err != nil {

manage.go

@@ -1,6 +1,7 @@
 package tdx
 
 import (
+	"errors"
 	"github.com/injoyai/ios/client"
 	"github.com/robfig/cron/v3"
 	"time"
@@ -10,6 +11,57 @@ const (
 	DefaultDatabaseDir = "./data/database"
 )
 
+func NewManageMysql(cfg *ManageConfig, op ...client.Option) (*Manage, error) {
+	//初始化配置
+	if cfg == nil {
+		cfg = &ManageConfig{}
+	}
+	if cfg.CodesFilename == "" {
+		return nil, errors.New("未配置Codes的数据库")
+	}
+	if cfg.WorkdayFileName == "" {
+		return nil, errors.New("未配置Workday的数据库")
+	}
+	if cfg.Dial == nil {
+		cfg.Dial = DialDefault
+	}
+
+	//通用客户端
+	commonClient, err := cfg.Dial(op...)
+	if err != nil {
+		return nil, err
+	}
+	commonClient.Wait.SetTimeout(time.Second * 5)
+
+	//代码管理
+	codes, err := NewCodesMysql(commonClient, cfg.CodesFilename)
+	if err != nil {
+		return nil, err
+	}
+
+	//工作日管理
+	workday, err := NewWorkdayMysql(commonClient, cfg.WorkdayFileName)
+	if err != nil {
+		return nil, err
+	}
+
+	//连接池
+	p, err := NewPool(func() (*Client, error) {
+		return cfg.Dial(op...)
+	}, cfg.Number)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Manage{
+		Pool:    p,
+		Config:  cfg,
+		Codes:   codes,
+		Workday: workday,
+		Cron:    cron.New(cron.WithSeconds()),
+	}, nil
+}
+
 func NewManage(cfg *ManageConfig, op ...client.Option) (*Manage, error) {
 	//初始化配置
 	if cfg == nil {
@@ -33,13 +85,13 @@ func NewManage(cfg *ManageConfig, op ...client.Option) (*Manage, error) {
 	commonClient.Wait.SetTimeout(time.Second * 5)
 
 	//代码管理
-	codes, err := NewCodes(commonClient, cfg.CodesFilename)
+	codes, err := NewCodesSqlite(commonClient, cfg.CodesFilename)
 	if err != nil {
 		return nil, err
 	}
 
 	//工作日管理
-	workday, err := NewWorkday(commonClient, cfg.WorkdayFileName)
+	workday, err := NewWorkdaySqlite(commonClient, cfg.WorkdayFileName)
 	if err != nil {
 		return nil, err
 	}

protocol/unit.go

@@ -135,10 +135,10 @@ func basePrice(code string) Price {
 		return 1
 	}
 	switch code[:2] {
-	case "60", "30", "68", "00", "92", "43":
+	case "60", "30", "68", "00", "92", "43", "39":
 		return 1
 	default:
-		return 10
+		return 1
 	}
 }
 
@@ -262,7 +262,7 @@ func IsStock(code string) bool {
 }
 
 func IsSZStock(code string) bool {
-	return len(code) == 8 && strings.ToLower(code[0:2]) == ExchangeSZ.String() && code[2:3] == "0"
+	return len(code) == 8 && strings.ToLower(code[0:2]) == ExchangeSZ.String() && (code[2:3] == "0" || code[2:4] == "30")
 }
 
 func IsSHStock(code string) bool {

workday.go

@@ -3,6 +3,7 @@ package tdx
 import (
 	"errors"
 	_ "github.com/glebarez/go-sqlite"
+	_ "github.com/go-sql-driver/mysql"
 	"github.com/injoyai/base/maps"
 	"github.com/injoyai/conv"
 	"github.com/injoyai/logs"
@@ -15,7 +16,19 @@ import (
 	"xorm.io/xorm"
 )
 
-func NewWorkday(c *Client, filenames ...string) (*Workday, error) {
+func NewWorkdayMysql(c *Client, dsn string) (*Workday, error) {
+
+	//连接数据库
+	db, err := xorm.NewEngine("mysql", dsn)
+	if err != nil {
+		return nil, err
+	}
+	db.SetMapper(core.SameMapper{})
+
+	return NewWorkday(c, db)
+}
+
+func NewWorkdaySqlite(c *Client, filenames ...string) (*Workday, error) {
 
 	defaultFilename := filepath.Join(DefaultDatabaseDir, "workday.db")
 	filename := conv.Default(defaultFilename, filenames...)
@@ -31,6 +44,11 @@ func NewWorkday(c *Client, filenames ...string) (*Workday, error) {
 	}
 	db.SetMapper(core.SameMapper{})
 	db.DB().SetMaxOpenConns(1)
+
+	return NewWorkday(c, db)
+}
+
+func NewWorkday(c *Client, db *xorm.Engine) (*Workday, error) {
 	if err := db.Sync2(new(WorkdayModel)); err != nil {
 		return nil, err
 	}
@@ -40,12 +58,12 @@ func NewWorkday(c *Client, filenames ...string) (*Workday, error) {
 		db:     db,
 		cache:  maps.NewBit(),
 	}
-
 	//设置定时器,每天早上9点更新数据,8点多获取不到今天的数据
 	task := cron.New(cron.WithSeconds())
 	task.AddFunc("0 0 9 * * *", func() {
 		for i := 0; i < 3; i++ {
-			if err := w.Update(); err == nil {
+			err := w.Update()
+			if err == nil {
 				return
 			}
 			logs.Err(err)
@@ -53,7 +71,6 @@ func NewWorkday(c *Client, filenames ...string) (*Workday, error) {
 		}
 	})
 	task.Start()
-
 	return w, w.Update()
 }
 
@@ -87,27 +104,30 @@ func (this *Workday) Update() error {
 	}
 
 	now := time.Now()
-	if lastWorkday == nil || lastWorkday.Unix < IntegerDay(now).Unix() {
+	if lastWorkday.Unix < IntegerDay(now).Unix() {
 		resp, err := this.Client.GetIndexDayAll("sh000001")
 		if err != nil {
 			logs.Err(err)
 			return err
 		}
 
-		return NewSessionFunc(this.db, func(session *xorm.Session) error {
-			for _, v := range resp.List {
-				if unix := v.Time.Unix(); unix > lastWorkday.Unix {
-					_, err = session.Insert(&WorkdayModel{Unix: unix, Date: v.Time.Format("20060102"), Is: true})
-					if err != nil {
-						return err
-					}
-					this.cache.Set(uint64(unix), true)
-				}
+		inserts := []any(nil)
+		for _, v := range resp.List {
+			if unix := v.Time.Unix(); unix > lastWorkday.Unix {
+				inserts = append(inserts, &WorkdayModel{Unix: unix, Date: v.Time.Format("20060102")})
+				this.cache.Set(uint64(unix), true)
 			}
+		}
+
+		if len(inserts) == 0 {
 			return nil
-		})
+		}
+
+		_, err = this.db.Insert(inserts)
+		return err
 
 	}
+
 	return nil
 }
 
@@ -161,7 +181,6 @@ type WorkdayModel struct {
 	ID   int64  `json:"id"`   //主键
 	Unix int64  `json:"unix"` //时间戳
 	Date string `json:"date"` //日期
-	Is   bool   `json:"is"`   //是否是工作日
 }
 
 func (this *WorkdayModel) TableName() string {