Compare commits

...

4 Commits

Author SHA1 Message Date
injoyai
26e2479e2f 细节优化 2025-11-21 16:08:46 +08:00
injoyai
6d0125afef 优化manage,修改成Option的方式,老版本的名称从NewManage改成MewManageSqlite 2025-11-21 14:17:09 +08:00
injoyai
2d77d769fd 增加Codes2的Option名称前缀 2025-11-21 14:14:54 +08:00
injoyai
33627c3d6c 增加新版迭代器,Iter和IterYear 2025-11-21 14:13:40 +08:00
7 changed files with 273 additions and 163 deletions

View File

@@ -22,50 +22,50 @@ import (
type Codes2Option func(*Codes2)
func WithDBFilename(filename string) Codes2Option {
func WithCodes2Database(filename string) Codes2Option {
return func(c *Codes2) {
c.dbFilename = filename
}
}
func WithTempDir(dir string) Codes2Option {
func WithCodes2TempDir(dir string) Codes2Option {
return func(c *Codes2) {
c.tempDir = dir
}
}
func WithSpec(spec string) Codes2Option {
func WithCodes2Spec(spec string) Codes2Option {
return func(c *Codes2) {
c.spec = spec
}
}
func WithUpdateKey(key string) Codes2Option {
func WithCodes2UpdateKey(key string) Codes2Option {
return func(c *Codes2) {
c.updateKey = key
}
}
func WithRetry(retry int) Codes2Option {
func WithCodes2Retry(retry int) Codes2Option {
return func(c *Codes2) {
c.retry = retry
}
}
func WithClient(c *Client) Codes2Option {
func WithCodes2Client(c *Client) Codes2Option {
return func(cs *Codes2) {
cs.c = c
}
}
func WithDial(dial ios.DialFunc, op ...client.Option) Codes2Option {
func WithCodes2Dial(dial ios.DialFunc, op ...client.Option) Codes2Option {
return func(c *Codes2) {
c.dial = dial
c.dialOption = op
}
}
func WithDialOption(op ...client.Option) Codes2Option {
func WithCodes2DialOption(op ...client.Option) Codes2Option {
return func(c *Codes2) {
c.dialOption = op
}
@@ -77,7 +77,7 @@ func NewCodes2(op ...Codes2Option) (*Codes2, error) {
tempDir: filepath.Join(DefaultDataDir, "temp"),
spec: "10 0 9 * * *",
updateKey: "codes",
retry: 3,
retry: DefaultRetry,
dial: NewRangeDial(Hosts),
dialOption: nil,
m: maps.NewGeneric[string, *CodeModel](),
@@ -117,7 +117,7 @@ func NewCodes2(op ...Codes2Option) (*Codes2, error) {
// 定时更新
cr := cron.New(cron.WithSeconds())
_, err = cr.AddFunc(cs.spec, func() {
for i := 0; i < 3; i++ {
for i := 0; i == 0 || i < cs.retry; i++ {
if err := cs.Update(); err != nil {
logs.Err(err)
<-time.After(time.Minute * 5)

View File

@@ -1,13 +1,14 @@
package main
import (
"time"
"github.com/injoyai/logs"
"github.com/injoyai/tdx"
"time"
)
func main() {
m, err := tdx.NewManage(nil)
m, err := tdx.NewManage()
logs.PanicErr(err)
codes := m.Codes.GetStocks().Codes()

23
example/Manage/main.go Normal file
View File

@@ -0,0 +1,23 @@
package main
import (
"github.com/injoyai/logs"
"github.com/injoyai/tdx"
)
func main() {
m, err := tdx.NewManage()
logs.PanicErr(err)
err = m.Do(func(c *tdx.Client) error {
resp, err := c.GetIndexDayAll("sh000001")
if err != nil {
return err
}
for _, v := range resp.List {
logs.Debug(v)
}
return nil
})
logs.PanicErr(err)
}

View File

@@ -2,16 +2,17 @@ package main
import (
"context"
"path/filepath"
"time"
"github.com/injoyai/logs"
"github.com/injoyai/tdx"
"github.com/injoyai/tdx/extend"
"path/filepath"
"time"
)
func main() {
m, err := tdx.NewManage(nil)
m, err := tdx.NewManage()
logs.PanicErr(err)
err = extend.NewPullKline(extend.PullKlineConfig{

View File

@@ -2,6 +2,7 @@ package main
import (
"context"
"github.com/injoyai/logs"
"github.com/injoyai/tdx"
"github.com/injoyai/tdx/extend"
@@ -11,7 +12,7 @@ func main() {
pt := extend.NewPullTrade("./data/trade")
m, err := tdx.NewManage(nil)
m, err := tdx.NewManage()
logs.PanicErr(err)
err = pt.PullYear(context.Background(), m, 2025, "sz000001")

329
manage.go
View File

@@ -2,170 +2,224 @@ package tdx
import (
"errors"
"sync"
"github.com/injoyai/conv"
"github.com/injoyai/ios/client"
"github.com/robfig/cron/v3"
"sync"
"time"
)
const (
DefaultClients = 1
DefaultRetry = 3
DefaultDataDir = "./data"
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,
}, nil
func NewManageMysql(op ...Option) (*Manage, error) {
return NewManage(
WithOptions(op...),
WithDialCodes(func(c *Client, database string) (ICodes, error) {
if database == "" {
return nil, errors.New("未配置Codes的数据库")
}
return NewCodesMysql(c, database)
}),
WithDialWorkday(func(c *Client, database string) (*Workday, error) {
if database == "" {
return nil, errors.New("未配置Workday的数据库")
}
return NewWorkdayMysql(c, database)
}),
)
}
func NewManage(cfg *ManageConfig, op ...client.Option) (*Manage, error) {
//初始化配置
if cfg == nil {
cfg = &ManageConfig{}
}
if cfg.CodesFilename == "" {
cfg.CodesFilename = DefaultDatabaseDir + "/codes.db"
}
if cfg.WorkdayFileName == "" {
cfg.WorkdayFileName = DefaultDatabaseDir + "/workday.db"
}
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 := NewCodesSqlite(commonClient, cfg.CodesFilename)
if err != nil {
return nil, err
}
//工作日管理
workday, err := NewWorkdaySqlite(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,
}, nil
func NewManageSqlite(op ...Option) (*Manage, error) {
return NewManage(
WithCodesDatabase(DefaultDatabaseDir+"/codes.db"),
WithWorkdayDatabase(DefaultDatabaseDir+"/workday.db"),
WithOptions(op...),
WithDialCodes(func(c *Client, database string) (ICodes, error) {
return NewCodesSqlite(c, database)
}),
WithDialWorkday(func(c *Client, database string) (*Workday, error) {
return NewWorkdaySqlite(c, database)
}),
)
}
func NewManage2(cfg *ManageConfig, op ...client.Option) (*Manage, error) {
//初始化配置
if cfg == nil {
cfg = &ManageConfig{}
}
if cfg.CodesFilename == "" {
cfg.CodesFilename = DefaultDatabaseDir + "/codes2.db"
}
if cfg.WorkdayFileName == "" {
cfg.WorkdayFileName = DefaultDatabaseDir + "/workday.db"
}
if cfg.Dial == nil {
cfg.Dial = DialDefault
func NewManageSqlite2(op ...Option) (*Manage, error) {
return NewManage(
WithCodesDatabase(DefaultDatabaseDir+"/codes2.db"),
WithWorkdayDatabase(DefaultDatabaseDir+"/workday.db"),
WithOptions(op...),
WithDialCodes(func(c *Client, database string) (ICodes, error) {
return NewCodes2(
WithCodes2Client(c),
WithCodes2Database(database),
)
}),
WithDialWorkday(func(c *Client, database string) (*Workday, error) {
return NewWorkdaySqlite(c, database)
}),
)
}
func NewManage(op ...Option) (m *Manage, err error) {
m = &Manage{
clients: DefaultClients,
dial: DialDefault,
dialOptions: nil,
dialCodes: nil,
codesDatabase: DefaultDatabaseDir + "/codes2.db",
dialWorkday: nil,
workdayDatabase: DefaultDatabaseDir + "/workday.db",
Pool: nil,
Codes: nil,
Workday: nil,
cron: nil,
once: sync.Once{},
}
//通用客户端
commonClient, err := cfg.Dial(op...)
for _, v := range op {
if v != nil {
v(m)
}
}
m.clients = conv.Select(m.clients <= 0, 1, m.clients)
m.dial = conv.Select(m.dial == nil, DialDefault, m.dial)
//连接池
m.Pool, err = NewPool(func() (*Client, error) { return m.dial(m.dialOptions...) }, m.clients)
if err != nil {
return nil, err
}
commonClient.Wait.SetTimeout(time.Second * 5)
//代码管理
codes, err := NewCodes2(WithClient(commonClient), WithDBFilename(cfg.CodesFilename))
if err != nil {
return nil, err
if m.Codes == nil {
if m.dialCodes == nil {
m.dialCodes = func(c *Client, database string) (ICodes, error) {
return NewCodes2(WithCodes2Client(c), WithCodes2Database(database))
}
}
err = m.Pool.Do(func(c *Client) error {
m.Codes, err = m.dialCodes(c, m.codesDatabase)
return err
})
if err != nil {
return nil, err
}
}
//工作日管理
workday, err := NewWorkdaySqlite(commonClient, cfg.WorkdayFileName)
if err != nil {
return nil, err
if m.Workday == nil {
if m.dialWorkday == nil {
m.dialWorkday = func(c *Client, database string) (*Workday, error) {
return NewWorkdaySqlite(c, database)
}
}
err = m.Pool.Do(func(c *Client) error {
m.Workday, err = m.dialWorkday(c, m.workdayDatabase)
return err
})
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
}
return &Manage{
Pool: p,
Config: cfg,
Codes: codes,
Workday: workday,
}, nil
/*
*/
type Option func(m *Manage)
type DialWorkdayFunc func(c *Client, database string) (*Workday, error)
type DialCodesFunc func(c *Client, database string) (ICodes, error)
func WithClients(clients int) Option {
return func(m *Manage) {
m.clients = clients
}
}
func WithDial(dial func(op ...client.Option) (*Client, error), op ...client.Option) Option {
return func(m *Manage) {
m.dial = dial
m.dialOptions = op
}
}
func WithDialOptions(op ...client.Option) Option {
return func(m *Manage) {
m.dialOptions = op
}
}
func WithCodes(codes ICodes) Option {
return func(m *Manage) {
m.Codes = codes
}
}
func WithDialCodes(dial DialCodesFunc) Option {
return func(m *Manage) {
m.dialCodes = dial
}
}
func WithCodesDatabase(database string) Option {
return func(m *Manage) {
m.codesDatabase = database
}
}
func WithWorkday(w *Workday) Option {
return func(m *Manage) {
m.Workday = w
}
}
func WithDialWorkday(dial DialWorkdayFunc) Option {
return func(m *Manage) {
m.dialWorkday = dial
}
}
func WithWorkdayDatabase(database string) Option {
return func(m *Manage) {
m.workdayDatabase = database
}
}
func WithOptions(op ...Option) Option {
return func(m *Manage) {
for _, v := range op {
v(m)
}
}
}
type Manage struct {
clients int
dial func(op ...client.Option) (cli *Client, err error)
dialOptions []client.Option
dialCodes func(c *Client, database string) (ICodes, error)
codesDatabase string
dialWorkday DialWorkdayFunc
workdayDatabase string
/*
*/
*Pool
Config *ManageConfig
Codes ICodes
Workday *Workday
cron *cron.Cron
@@ -186,6 +240,13 @@ func (this *Manage) RangeETFs(f func(code string)) {
}
}
// RangeIndexes 遍历所有指数
func (this *Manage) RangeIndexes(f func(code string)) {
for _, v := range this.Codes.GetETFs() {
f(v.FullCode())
}
}
// AddWorkdayTask 添加工作日任务
func (this *Manage) AddWorkdayTask(spec string, f func(m *Manage)) {
this.once.Do(func() {

View File

@@ -2,6 +2,11 @@ package tdx
import (
"errors"
"iter"
"os"
"path/filepath"
"time"
_ "github.com/glebarez/go-sqlite"
_ "github.com/go-sql-driver/mysql"
"github.com/injoyai/base/maps"
@@ -10,9 +15,6 @@ import (
"github.com/injoyai/logs"
"github.com/injoyai/tdx/protocol"
"github.com/robfig/cron/v3"
"os"
"path/filepath"
"time"
"xorm.io/core"
"xorm.io/xorm"
)
@@ -154,7 +156,7 @@ func (this *Workday) TodayIs() bool {
func (this *Workday) RangeYear(year int, f func(t time.Time) bool) {
this.Range(
time.Date(year, 1, 1, 0, 0, 0, 0, time.Local),
time.Date(year, 12, 31, 0, 0, 0, 0, time.Local),
time.Date(year, 12, 31, 0, 0, 0, 1, time.Local),
f,
)
}
@@ -162,8 +164,6 @@ func (this *Workday) RangeYear(year int, f func(t time.Time) bool) {
// 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)
for ; start.Before(end); start = start.Add(time.Hour * 24) {
if this.Is(start) {
if !f(start) {
@@ -173,13 +173,36 @@ func (this *Workday) Range(start, end time.Time, f func(t time.Time) bool) {
}
}
// RangeDesc 倒序遍历工作日,从今天-1990年12月19日(上海交易所成立时间)
func (this *Workday) RangeDesc(f func(t time.Time) bool) {
t := IntegerDay(time.Now())
for ; t.After(time.Date(1990, 12, 18, 0, 0, 0, 0, time.Local)); t = t.Add(-time.Hour * 24) {
if this.Is(t) {
if !f(t) {
return
func (this *Workday) IterYear(year int, desc ...bool) iter.Seq[time.Time] {
return this.Iter(
time.Date(year, 1, 1, 0, 0, 0, 0, time.Local),
time.Date(year, 12, 31, 0, 0, 0, 1, time.Local),
desc...,
)
}
// Iter 遍历指定范围的工作日,推荐start带上时间15:00,这样当天小于15点不会触发
func (this *Workday) Iter(start, end time.Time, desc ...bool) iter.Seq[time.Time] {
start = conv.Select(start.Before(protocol.ExchangeEstablish), protocol.ExchangeEstablish, start)
if len(desc) > 0 && desc[0] {
//倒序遍历
return func(yield func(time.Time) bool) {
for ; end.After(start); end = end.Add(-time.Hour * 24) {
if this.Is(end) {
if !yield(end) {
return
}
}
}
}
}
//正序遍历
return func(yield func(time.Time) bool) {
for ; start.Before(end); start = start.Add(time.Hour * 24) {
if this.Is(start) {
if !yield(start) {
return
}
}
}
}