refactor(zt): update state and watch management

go-client/apps/zt/logic/state.go

@@ -4,225 +4,149 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
-	"path/filepath"
+	"path"
+	"slices"
 	"sync"
 
 	"big-qmt/go-client/config"
 	"big-qmt/go-client/sdk"
 )
 
-const (
+var (
-	pendingNone        = ""
+	StatusNone = ""
-	pendingBaseOpening = "base_opening"
+	StatusIng  = "ING" // 处理中
-	pendingAdd         = "add"
+	StatusOk   = "OK"  // 成功
-	pendingSellAdd     = "sell_add"
+	QuantState *State
-	pendingSellBase    = "sell_base"
 )
 
-type SymbolState struct {
+type State struct {
+	AbsPath string
+	mu      sync.Mutex
+	Items   map[string]*StateItem
+	Codes   []string
+}
+
+type StateItem struct {
 	Code         string  `json:"code"`
+	BaseOrderId  string  `json:"base_order_id"`
 	BaseQty      int     `json:"base_qty"`
 	BaseCost     float64 `json:"base_cost"`
-	AddQty         int     `json:"add_qty"`
+	BaseStatus   string  `json:"base_status,omitempty"`
-	AddCost        float64 `json:"add_cost"`
+	AddedOrderId string  `json:"added_order_id"`
-	Pending        string  `json:"pending"`
+	AddedNum     int     `json:"add_num"`
-	PendingOrderID string  `json:"pending_order_id,omitempty"`
+	AddedQty     int     `json:"add_qty"`
-	OrderStatus    string  `json:"order_status,omitempty"`
+	AddedCost    float64 `json:"add_cost"`
+	AddedStatus  string  `json:"added_status,omitempty"`
 }
 
-func setPending(item *SymbolState, pending, orderID string) {
+func InitState(sn string) error {
-	item.Pending = pending
+	absPath := path.Join(config.Global.QMTDataDir, fmt.Sprintf("%s_%s_state.json", sn, config.Account.AccountID))
-	item.PendingOrderID = orderID
+	items, err := loadStateFile(absPath)
-	item.OrderStatus = "submitted"
+	if err != nil {
+		return err
 	}
 
-func clearPending(item *SymbolState) {
+	var codes []string
-	item.Pending = pendingNone
+	for code, _ := range items {
-	item.PendingOrderID = ""
+		codes = append(codes, code)
-	item.OrderStatus = ""
 	}
 
-type filePayload struct {
+	QuantState = &State{
-	Version int            `json:"version"`
+		AbsPath: absPath,
-	Data    map[string]any `json:"data"`
+		Items:   items,
+		Codes:   codes,
+	}
+	return nil
 }
 
-type ZTState struct {
+func loadStateFile(fp string) (map[string]*StateItem, error) {
-	path      string
+	raw, err := os.ReadFile(fp)
-	Items     map[string]*SymbolState
+	if err != nil {
-	LoadError string
+		return nil, fmt.Errorf("[状态] 读取失败: %v", err)
-	fresh     bool
+	}
-	mu        sync.Mutex
+	var items map[string]*StateItem
+	if err := json.Unmarshal(raw, &items); err != nil {
+		return nil, fmt.Errorf("[状态] 解析失败：%s", err)
+	}
+	return items, nil
+
 }
 
-var (
+func SyncPositions(positions []sdk.Position) error {
-	statesMu sync.Mutex
-	states   = map[string]*ZTState{}
-)
-
-// BootstrapState 在状态文件首次不存在时，将启动前已有持仓登记为底仓。
-func BootstrapState(positions []sdk.Position) {
-	state := getState()
-	if !state.Fresh() {
-		return
-	}
 	for _, pos := range positions {
 		code := pos.StockCode
 		if code == "" || pos.Volume <= 0 || pos.OpenPrice <= 0 {
 			continue
 		}
-		item := state.Ensure(code)
+		if !slices.Contains(QuantState.Codes, code) {
-		item.BaseQty, item.BaseCost = pos.Volume, pos.OpenPrice
+			item := &StateItem{
-		logf("WARNING", "[ZT][状态] %s 首次接管为底仓 数量=%d 成本=%.2f", code, pos.Volume, pos.OpenPrice)
+				Code:       code,
+				BaseQty:    pos.Volume,
+				BaseCost:   pos.OpenPrice,
+				BaseStatus: StatusOk,
+			}
+			QuantState.Append(item)
+			logf("WARNING", "[状态] %s 首次接管为底仓 数量=%d 成本=%.2f", code, pos.Volume, pos.OpenPrice)
 		}
-	state.completeBootstrap()
-	state.Save()
 	}
 
-func getState() *ZTState {
+	return QuantState.Save()
-	statesMu.Lock()
-	defer statesMu.Unlock()
-	accountID := config.Account.AccountID
-	if s, ok := states[accountID]; ok {
-		return s
-	}
-	s := loadZTState(config.Global.QMTDataDir, accountID)
-	states[accountID] = s
-	return s
 }
 
-func loadZTState(dataDir, accountID string) *ZTState {
+func (s *State) Append(i *StateItem) {
-	st := &ZTState{
+	s.mu.Lock()
-		path:  filepath.Join(dataDir, fmt.Sprintf("zt_%s_state.json", accountID)),
+	defer s.mu.Unlock()
-		Items: map[string]*SymbolState{},
+
+	s.Items[i.Code] = i
+	s.Codes = append(s.Codes, i.Code)
 }
-	raw, err := os.ReadFile(st.path)
+
-	if err != nil {
+func (s *State) Get(code string) (*StateItem, error) {
-		if os.IsNotExist(err) {
+	s.mu.Lock()
-			st.fresh = true
+	defer s.mu.Unlock()
-			return st
+
-		}
+	if i, ok := s.Items[code]; ok {
-		st.LoadError = err.Error()
+		return i, nil
-		logf("ERROR", "[ZT][状态] 读取状态文件失败: %v", err)
-		return st
-	}
-	var payload filePayload
-	if err := json.Unmarshal(raw, &payload); err != nil || payload.Version != 1 {
-		st.LoadError = "状态文件版本无效"
-		logf("ERROR", "[ZT][状态] %s", st.LoadError)
-		return st
-	}
-	data := payload.Data
-	if data == nil {
-		st.LoadError = "状态文件内容无效"
-		logf("ERROR", "[ZT][状态] %s", st.LoadError)
-		return st
-	}
-	symbolsAny, _ := data["symbols"]
-	symbols, _ := symbolsAny.(map[string]any)
-	if symbols == nil {
-		if _, ok := data["code"]; ok {
-			symbols = map[string]any{}
 	} else {
-			symbols = data
+		return nil, fmt.Errorf("%s not found.", code)
 	}
 }
-	for code, value := range symbols {
-		m, ok := value.(map[string]any)
-		if !ok {
-			continue
-		}
-		item := &SymbolState{Code: code}
-		b, _ := json.Marshal(m)
-		_ = json.Unmarshal(b, item)
-		item.Code = code
-		st.Items[code] = item
-	}
-	return st
-}
 
-func (s *ZTState) Fresh() bool {
+func (s *State) Set(i *StateItem) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return s.fresh
+
+	if _, ok := s.Items[i.Code]; !ok {
+		s.Codes = append(s.Codes, i.Code)
+	}
+	s.Items[i.Code] = i
 }
 
-func (s *ZTState) completeBootstrap() {
+func (s *State) Delete(code string) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.fresh = false
-}
 
-func (s *ZTState) Get(code string) *SymbolState {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.Items[code]
-}
-
-func (s *ZTState) Ensure(code string) *SymbolState {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if item, ok := s.Items[code]; ok {
-		return item
-	}
-	item := &SymbolState{Code: code}
-	s.Items[code] = item
-	return item
-}
-
-func (s *ZTState) Remove(code string) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
 	delete(s.Items, code)
+	if index := slices.Index(s.Codes, code); index >= 0 {
+		s.Codes = slices.Delete(s.Codes, index, index+1)
+	}
 }
 
-func (s *ZTState) Codes() []string {
+func (s *State) Save() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	out := make([]string, 0, len(s.Items))
-	for code := range s.Items {
-		out = append(out, code)
-	}
-	return out
-}
 
-func (s *ZTState) Save() {
+	// 写入AbsPath文件
-	s.mu.Lock()
+	f, err := os.OpenFile(s.AbsPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
-	defer s.mu.Unlock()
-	if s.LoadError != "" {
-		return
-	}
-	if err := s.saveUnlocked(); err != nil {
-		s.LoadError = err.Error()
-		logf("ERROR", "[ZT][状态] 保存失败: %v", err)
-		return
-	}
-}
-
-func (s *ZTState) saveUnlocked() error {
-	symbols := map[string]any{}
-	for code, item := range s.Items {
-		symbols[code] = item
-	}
-	payload := filePayload{Version: 1, Data: map[string]any{"symbols": symbols}}
-	raw, err := json.Marshal(payload)
 	if err != nil {
-		return err
+		return fmt.Errorf("[状态] 打开文件失败: %v", err)
-	}
-	if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
-		return err
-	}
-	tmp := s.path + ".tmp"
-	if err := os.WriteFile(tmp, raw, 0o644); err != nil {
-		return err
-	}
-	return replaceFile(tmp, s.path)
 	}
+	defer f.Close()
 
-func replaceFile(tmp, dest string) error {
+	encoder := json.NewEncoder(f)
-	if err := os.Rename(tmp, dest); err == nil {
+	encoder.SetIndent("", "  ")
+	if err := encoder.Encode(s.Items); err != nil {
+		return fmt.Errorf("[状态] 写入失败: %v", err)
+	}
 	return nil
 }
-	_ = os.Remove(dest)
-	return os.Rename(tmp, dest)
-}

go-client/apps/zt/logic/watch.go

@@ -0,0 +1,66 @@
+package logic
+
+import (
+	"sync"
+	"time"
+)
+
+var (
+	WatchExpireTime  = 5 * time.Minute
+	WatchReThreshold = 0.61
+
+	OpenWatch   *WatchMu
+	PosbuyWatch *WatchMu
+)
+
+type dipWatch struct {
+	LastClose float64
+	ExpiresAt time.Time
+}
+
+type WatchMu struct {
+	mu   *sync.Mutex
+	Data map[string]dipWatch
+}
+
+func InitWatch() {
+	OpenWatch = &WatchMu{
+		Data: make(map[string]dipWatch),
+	}
+	PosbuyWatch = &WatchMu{
+		Data: make(map[string]dipWatch),
+	}
+}
+
+func (w *WatchMu) Triggered(tag, code string, price float64) bool {
+	if price <= 0 {
+		return false
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	now := time.Now()
+	watch, ok := w.Data[code]
+	if !ok || now.After(watch.ExpiresAt) || now.Equal(watch.ExpiresAt) {
+		w.Data[code] = dipWatch{LastClose: price, ExpiresAt: now.Add(WatchExpireTime)}
+		logf("INFO", "[%s-观察] %s 现价=%.2f", tag, code, price)
+		return false
+	}
+	if price < watch.LastClose {
+		watch.LastClose = price
+		watch.ExpiresAt = now.Add(WatchExpireTime)
+		w.Data[code] = watch
+		logf("INFO", "[%s-下跌] %s 刷新低点=%.2f", tag, code, price)
+		return false
+	}
+	rebound := (price - watch.LastClose) / watch.LastClose * 100
+	if rebound <= 0 {
+		return false
+	}
+	if rebound < WatchReThreshold {
+		logf("INFO", "[%s-等待] %s 反弹=%.2f%% 阈值=%.2f%%", tag, code, rebound, WatchReThreshold)
+		return false
+	}
+	delete(w.Data, code)
+	logf("INFO", "[%s-触发] %s 反弹=%.2f%% 低点=%.2f", tag, code, rebound, watch.LastClose)
+	return true
+}

go-client/apps/zt/main.go

@@ -16,6 +16,10 @@ import (
 	"github.com/robfig/cron/v3"
 )
 
+var (
+	StrategyName = "zt"
+)
+
 func main() {
 	log.SetFlags(log.LstdFlags | log.Lmicroseconds)
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
@@ -34,8 +38,16 @@ func main() {
 		return
 	}
 
+	// 初始化
+	logic.InitWatch()
+	if err := logic.InitState(StrategyName); err != nil {
+		log.Panicln("ERROR", err.Error())
+	}
+	if err := logic.SyncPositions(positions); err != nil {
+		log.Panicln("ERROR", err.Error())
+	}
+
 	// 第三步：连接成功后接管首次持仓并打印账户概览。
-	logic.BootstrapState(positions)
 	logic.Overview(assets, positions)
 	signals, err := libs.FetchSignal(libs.Dcm_Signal, config.Account.HostKey)
 	if err != nil {