修复深圳指数历史分时成交价格小10倍的问题

client.go

@@ -422,7 +422,7 @@ func (this *Client) GetHistoryTrade(date, code string, start, count uint16) (*pr
 
 // GetHistoryMinuteTrade 获取历史分时交易
 // 只能获取昨天及之前的数据,服务器最多返回2000条,count-start<=2000,如果日期输入错误,则返回0
-// 历史数据sz000001在20241116只能查到21111112,13年差几天,3141天,或者其他规则
+// 历史数据只能查到20000609
 func (this *Client) GetHistoryMinuteTrade(date, code string, start, count uint16) (*protocol.TradeResp, error) {
 	code = protocol.AddPrefix(code)
 	f, err := protocol.MHistoryTrade.Frame(date, code, start, count)
@@ -439,13 +439,36 @@ func (this *Client) GetHistoryMinuteTrade(date, code string, start, count uint16
 	return result.(*protocol.TradeResp), nil
 }
 
-func (this *Client) GetHistoryTradeAll(date, code string) (*protocol.TradeResp, error) {
+// GetHistoryTradeFull 获取上市至今的分时成交
-	return this.GetHistoryMinuteTradeAll(date, code)
+func (this *Client) GetHistoryTradeFull(code string) (protocol.Trades, error) {
+	ls := protocol.Trades(nil)
+	resp, err := this.GetKlineMonthAll(code)
+	if err != nil {
+		return nil, err
+	}
+	if len(resp.List) == 0 {
+		return nil, nil
+	}
+	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)
+		if err != nil {
+			return nil, err
+		}
+		ls = append(ls, res.List...)
+	}
+	return ls, nil
 }
 
-// GetHistoryMinuteTradeAll 获取历史分时全部交易,通过多次请求来拼接,只能获取昨天及之前的数据
+// GetHistoryTradeDay 获取历史某天分时全部交易,通过多次请求来拼接,只能获取昨天及之前的数据
-// 历史数据sz000001在20241116只能查到21111112,13年差几天,3141天,或者其他规则
+func (this *Client) GetHistoryTradeDay(date, code string) (*protocol.TradeResp, error) {
-func (this *Client) GetHistoryMinuteTradeAll(date, code string) (*protocol.TradeResp, error) {
+	return this.GetHistoryMinuteTradeDay(date, code)
+}
+
+// GetHistoryMinuteTradeDay 获取历史某天分时全部交易,通过多次请求来拼接,只能获取昨天及之前的数据
+// 历史数据只能查到20000609
+func (this *Client) GetHistoryMinuteTradeDay(date, code string) (*protocol.TradeResp, error) {
 	resp := &protocol.TradeResp{}
 	size := uint16(2000)
 	for start := uint16(0); ; start += size {

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. 插入或者更新数据库
+	switch this.db.Dialect().URI().DBType {
-	err := NewSessionFunc(this.db, func(session *xorm.Session) error {
+	case "mysql":
-		for _, v := range insert {
+		// 1️⃣ 清空
-			if _, err := session.Insert(v); err != nil {
+		if _, err := this.db.Exec("TRUNCATE TABLE codes"); err != nil {
-				return err
+			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 {
+	case "sqlite3":
-			if _, err := session.Where("Exchange=? and Code=? ", v.Exchange, v.Code).Cols("Name,LastPrice").Update(v); err != nil {
+		//4. 插入或者更新数据库
-				return err
+		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.GetHistoryMinuteTradeAll("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/model_kline.go

@@ -257,12 +257,16 @@ func (this Klines) Sort() {
 	sort.Sort(this)
 }
 
-// Kline 计算多个K线,成一个K线
+func (this Klines) Kline(t time.Time, last Price) *Kline {
-func (this Klines) Kline() *Kline {
+	k := &Kline{
-	if this == nil {
+		Time:   t,
-		return new(Kline)
+		Open:   last,
+		High:   last,
+		Low:    last,
+		Close:  last,
+		Volume: 0,
+		Amount: 0,
 	}
-	k := new(Kline)
 	for i, v := range this {
 		switch i {
 		case 0:
@@ -270,34 +274,87 @@ func (this Klines) Kline() *Kline {
 			k.High = v.High
 			k.Low = v.Low
 			k.Close = v.Close
-		case len(this) - 1:
+		default:
-			k.Close = v.Close
+			if k.Open == 0 {
-			k.Time = v.Time
+				k.Open = v.Open
-		}
+			}
-		if v.High > k.High {
+			k.High = conv.Select(k.High < v.High, v.High, k.High)
-			k.High = v.High
+			k.Low = conv.Select(k.Low > v.Low, v.Low, k.Low)
-		}
-		if v.Low < k.Low {
-			k.Low = v.Low
 		}
+		k.Close = v.Close
 		k.Volume += v.Volume
 		k.Amount += v.Amount
 	}
 	return k
 }
 
-// Merge 合并K线,1分钟转成5,15,30分钟等
+// Merge 合并成其他类型的K线
 func (this Klines) Merge(n int) Klines {
-	if this == nil {
+	if n <= 1 {
-		return nil
+		return this
 	}
-	ks := []*Kline(nil)
+	ks := Klines(nil)
-	for i := 0; i < len(this); i += n {
+	ls := Klines(nil)
-		if i+n > len(this) {
+	for i := 0; ; i++ {
-			ks = append(ks, this[i:].Kline())
+		if len(this) <= i*n {
-		} else {
+			break
-			ks = append(ks, this[i:i+n].Kline())
 		}
+		if len(this) < (i+1)*n {
+			ls = this[i*n:]
+		} else {
+			ls = this[i*n : (i+1)*n]
+		}
+		if len(ls) == 0 {
+			break
+		}
+		last := ls[len(ls)-1]
+		k := ls.Kline(last.Time, ls[0].Open)
+		ks = append(ks, k)
 	}
 	return ks
 }
+
+//// Kline 计算多个K线,成一个K线
+//func (this Klines) Kline() *Kline {
+//	if this == nil {
+//		return new(Kline)
+//	}
+//	k := new(Kline)
+//	for i, v := range this {
+//		switch i {
+//		case 0:
+//			k.Open = v.Open
+//			k.High = v.High
+//			k.Low = v.Low
+//			k.Close = v.Close
+//		case len(this) - 1:
+//			k.Close = v.Close
+//			k.Time = v.Time
+//		}
+//		if v.High > k.High {
+//			k.High = v.High
+//		}
+//		if v.Low < k.Low {
+//			k.Low = v.Low
+//		}
+//		k.Volume += v.Volume
+//		k.Amount += v.Amount
+//	}
+//	return k
+//}
+
+//// Merge 合并K线,1分钟转成5,15,30分钟等
+//func (this Klines) Merge(n int) Klines {
+//	if this == nil {
+//		return nil
+//	}
+//	ks := []*Kline(nil)
+//	for i := 0; i < len(this); i += n {
+//		if i+n > len(this) {
+//			ks = append(ks, this[i:].Kline())
+//		} else {
+//			ks = append(ks, this[i:i+n].Kline())
+//		}
+//	}
+//	return ks
+//}

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 {
+		inserts := []any(nil)
-			for _, v := range resp.List {
+		for _, v := range resp.List {
-				if unix := v.Time.Unix(); unix > lastWorkday.Unix {
+			if unix := v.Time.Unix(); unix > lastWorkday.Unix {
-					_, err = session.Insert(&WorkdayModel{Unix: unix, Date: v.Time.Format("20060102"), Is: true})
+				inserts = append(inserts, &WorkdayModel{Unix: unix, Date: v.Time.Format("20060102")})
-					if err != nil {
+				this.cache.Set(uint64(unix), true)
-						return err
-					}
-					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 {