Merge remote-tracking branch 'origin/master'

client.go

@@ -181,6 +181,11 @@ func (this *Client) handlerDealMessage(c *client.Client, msg ios.Acker) {
 
 }
 
+// SetTimeout 设置超时时间
+func (this *Client) SetTimeout(t time.Duration) {
+	this.Wait.SetTimeout(t)
+}
+
 // SendFrame 发送数据,并等待响应
 func (this *Client) SendFrame(f *protocol.Frame, cache ...any) (any, error) {
 	f.MsgID = atomic.AddUint32(&this.msgID, 1)
@@ -440,7 +445,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 +461,20 @@ 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 {
+		for i := 0; i < 3; i++ {
+			res, err = this.GetHistoryTradeDay(t.Format("20060102"), code)
+			if err == nil {
+				break
+			}
+		}
 		if err != nil {
-			return nil, err
+			return false
 		}
 		ls = append(ls, res.List...)
-	}
-	return ls, nil
+		return true
+	})
+	return ls, err
 }
 
 // GetHistoryTradeDay 获取历史某天分时全部交易,通过多次请求来拼接,只能获取昨天及之前的数据

codes.go

@@ -34,12 +34,6 @@ func NewCodesMysql(c *Client, dsn string) (*Codes, error) {
 		return nil, err
 	}
 	db.SetMapper(core.SameMapper{})
-	if err := db.Sync2(new(CodeModel)); err != nil {
-		return nil, err
-	}
-	if err := db.Sync2(new(UpdateModel)); err != nil {
-		return nil, err
-	}
 
 	return NewCodes(c, db)
 }
@@ -62,6 +56,12 @@ func NewCodesSqlite(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
 	}
@@ -69,17 +69,13 @@ func NewCodesSqlite(c *Client, filenames ...string) (*Codes, error) {
 		return nil, err
 	}
 
-	return NewCodes(c, db)
-}
-
-func NewCodes(c *Client, db *xorm.Engine) (*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
 			}
@@ -127,83 +123,6 @@ func NewCodes(c *Client, db *xorm.Engine) (*Codes, error) {
 	return cc, cc.Update(true)
 }
 
-//func NewCodes(c *Client, filenames ...string) (*Codes, error) {
-//
-//	//如果没有指定文件名,则使用默认
-//	defaultFilename := filepath.Join(DefaultDatabaseDir, "codes.db")
-//	filename := conv.Default(defaultFilename, filenames...)
-//	filename = conv.Select(filename == "", defaultFilename, filename)
-//
-//	//如果文件夹不存在就创建
-//	dir, _ := filepath.Split(filename)
-//	_ = os.MkdirAll(dir, 0777)
-//
-//	//连接数据库
-//	db, err := xorm.NewEngine("sqlite", filename)
-//	if err != nil {
-//		return nil, err
-//	}
-//	db.SetMapper(core.SameMapper{})
-//	db.DB().SetMaxOpenConns(1)
-//	if err := db.Sync2(new(CodeModel)); err != nil {
-//		return nil, err
-//	}
-//	if err := db.Sync2(new(UpdateModel)); err != nil {
-//		return nil, err
-//	}
-//
-//	update := new(UpdateModel)
-//	{ //查询或者插入一条数据
-//		has, err := db.Get(update)
-//		if err != nil {
-//			return nil, err
-//		} else if !has {
-//			if _, err := db.Insert(update); err != nil {
-//				return nil, err
-//			}
-//		}
-//	}
-//
-//	cc := &Codes{
-//		Client: c,
-//		db:     db,
-//	}
-//
-//	{ //设置定时器,每天早上9点更新数据
-//		task := cron.New(cron.WithSeconds())
-//		task.AddFunc("10 0 9 * * *", func() {
-//			for i := 0; i < 3; i++ {
-//				if err := cc.Update(); err == nil {
-//					return
-//				}
-//				logs.Err(err)
-//				<-time.After(time.Minute * 5)
-//			}
-//		})
-//		task.Start()
-//	}
-//
-//	{ //判断是否更新过,更新过则不更新
-//		now := time.Now()
-//		node := time.Date(now.Year(), now.Month(), now.Day(), 9, 0, 0, 0, time.Local)
-//		updateTime := time.Unix(update.Time, 0)
-//		if now.Sub(node) > 0 {
-//			//当前时间在9点之后,且更新时间在9点之前,需要更新
-//			if updateTime.Sub(node) < 0 {
-//				return cc, cc.Update()
-//			}
-//		} else {
-//			//当前时间在9点之前,且更新时间在上个节点之前
-//			if updateTime.Sub(node.Add(time.Hour*24)) < 0 {
-//				return cc, cc.Update()
-//			}
-//		}
-//	}
-//
-//	//从缓存中加载
-//	return cc, cc.Update(true)
-//}
-
 type Codes struct {
 	*Client                         //客户端
 	db        *xorm.Engine          //数据库实例
@@ -276,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
 }
 
@@ -339,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/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")
+}

extend/codes-bj.go

@@ -4,6 +4,14 @@ import (
 	"github.com/injoyai/tdx"
 )
 
-func GetBjCodes() ([]*tdx.BjCode, error) {
-	return tdx.GetBjCodes()
+func GetBjCodes() ([]string, error) {
+	cs, err := tdx.GetBjCodes()
+	if err != nil {
+		return nil, err
+	}
+	ls := []string(nil)
+	for _, v := range cs {
+		ls = append(ls, "bj"+v.Code)
+	}
+	return ls, 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 {

protocol/model_kline.go

@@ -293,6 +293,7 @@ func (this Klines) Merge(n int) Klines {
 	if n <= 1 {
 		return this
 	}
+
 	ks := Klines(nil)
 	ls := Klines(nil)
 	for i := 0; ; i++ {

protocol/model_trade.go

@@ -210,7 +210,7 @@ func (this Trades) klinesForDay(date time.Time) Klines {
 	//分组,按
 	for _, v := range this {
 		ms := minutes(v.Time)
-		t := conv.Select(ms <= _930, _930, ms)
+		t := conv.Select(ms < _930, _930, ms)
 		t++
 		t = conv.Select(t > _1130 && t <= _1300, _1130, t)
 		t = conv.Select(t > _1500, _1500, t)

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
 	}
 }
 
@@ -281,11 +281,11 @@ func IsETF(code string) bool {
 	code = strings.ToLower(code)
 	switch {
 	case code[0:2] == ExchangeSH.String() &&
-		(code[2:5] == "510" || code[2:5] == "511" || code[2:5] == "512" || code[2:5] == "513" || code[2:5] == "515"):
+		(code[2:4] == "51" || code[2:4] == "56" || code[2:4] == "58"):
 		return true
 
 	case code[0:2] == ExchangeSZ.String() &&
-		(code[2:5] == "159"):
+		(code[2:4] == "15" || code[2:4] == "16"):
 		return true
 	}
 	return false

workday.go

@@ -6,6 +6,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/injoyai/base/maps"
 	"github.com/injoyai/conv"
+	"github.com/injoyai/ios/client"
 	"github.com/injoyai/logs"
 	"github.com/injoyai/tdx/protocol"
 	"github.com/robfig/cron/v3"
@@ -16,6 +17,14 @@ import (
 	"xorm.io/xorm"
 )
 
+func DialWorkday(op ...client.Option) (*Workday, error) {
+	c, err := DialDefault(op...)
+	if err != nil {
+		return nil, err
+	}
+	return NewWorkdaySqlite(c)
+}
+
 func NewWorkdayMysql(c *Client, dsn string) (*Workday, error) {
 
 	//连接数据库
@@ -24,9 +33,6 @@ func NewWorkdayMysql(c *Client, dsn string) (*Workday, error) {
 		return nil, err
 	}
 	db.SetMapper(core.SameMapper{})
-	if err := db.Sync2(new(WorkdayModel)); err != nil {
-		return nil, err
-	}
 
 	return NewWorkday(c, db)
 }
@@ -47,14 +53,15 @@ func NewWorkdaySqlite(c *Client, filenames ...string) (*Workday, error) {
 	}
 	db.SetMapper(core.SameMapper{})
 	db.DB().SetMaxOpenConns(1)
-	if err := db.Sync2(new(WorkdayModel)); err != nil {
-		return nil, err
-	}
 
 	return NewWorkday(c, db)
 }
 
 func NewWorkday(c *Client, db *xorm.Engine) (*Workday, error) {
+	if err := db.Sync2(new(WorkdayModel)); err != nil {
+		return nil, err
+	}
+
 	w := &Workday{
 		Client: c,
 		db:     db,
@@ -76,48 +83,6 @@ func NewWorkday(c *Client, db *xorm.Engine) (*Workday, error) {
 	return w, w.Update()
 }
 
-//func NewWorkday(c *Client, filenames ...string) (*Workday, error) {
-//
-//	defaultFilename := filepath.Join(DefaultDatabaseDir, "workday.db")
-//	filename := conv.Default(defaultFilename, filenames...)
-//
-//	//如果文件夹不存在就创建
-//	dir, _ := filepath.Split(filename)
-//	_ = os.MkdirAll(dir, 0777)
-//
-//	//连接数据库
-//	db, err := xorm.NewEngine("sqlite", filename)
-//	if err != nil {
-//		return nil, err
-//	}
-//	db.SetMapper(core.SameMapper{})
-//	db.DB().SetMaxOpenConns(1)
-//	if err := db.Sync2(new(WorkdayModel)); err != nil {
-//		return nil, err
-//	}
-//
-//	w := &Workday{
-//		Client: c,
-//		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 {
-//				return
-//			}
-//			logs.Err(err)
-//			<-time.After(time.Minute * 5)
-//		}
-//	})
-//	task.Start()
-//
-//	return w, w.Update()
-//}
-
 type Workday struct {
 	*Client
 	db    *xorm.Engine
@@ -148,27 +113,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
 }
 
@@ -191,11 +159,11 @@ func (this *Workday) RangeYear(year int, f func(t time.Time) bool) {
 	)
 }
 
-// Range 遍历指定范围的工作日
+// Range 遍历指定范围的工作日,推荐start带上时间15:00,这样当天小于15点不会触发
 func (this *Workday) Range(start, end time.Time, f func(t time.Time) bool) {
 	start = conv.Select(start.Before(protocol.ExchangeEstablish), protocol.ExchangeEstablish, start)
-	now := IntegerDay(time.Now())
-	end = conv.Select(end.After(now), now, end).Add(1)
+	//now := IntegerDay(time.Now())
+	//end = conv.Select(end.After(now), now, end).Add(1)
 	for ; start.Before(end); start = start.Add(time.Hour * 24) {
 		if this.Is(start) {
 			if !f(start) {
@@ -222,7 +190,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 {