diff --git a/go-client/apps/zt/logic/state.go b/go-client/apps/zt/logic/state.go index 13fd3e2..f1bc9d3 100644 --- a/go-client/apps/zt/logic/state.go +++ b/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 ( - pendingNone = "" - pendingBaseOpening = "base_opening" - pendingAdd = "add" - pendingSellAdd = "sell_add" - pendingSellBase = "sell_base" -) - -type SymbolState struct { - Code string `json:"code"` - BaseQty int `json:"base_qty"` - BaseCost float64 `json:"base_cost"` - AddQty int `json:"add_qty"` - AddCost float64 `json:"add_cost"` - Pending string `json:"pending"` - PendingOrderID string `json:"pending_order_id,omitempty"` - OrderStatus string `json:"order_status,omitempty"` -} - -func setPending(item *SymbolState, pending, orderID string) { - item.Pending = pending - item.PendingOrderID = orderID - item.OrderStatus = "submitted" -} - -func clearPending(item *SymbolState) { - item.Pending = pendingNone - item.PendingOrderID = "" - item.OrderStatus = "" -} - -type filePayload struct { - Version int `json:"version"` - Data map[string]any `json:"data"` -} - -type ZTState struct { - path string - Items map[string]*SymbolState - LoadError string - fresh bool - mu sync.Mutex -} - var ( - statesMu sync.Mutex - states = map[string]*ZTState{} + StatusNone = "" + StatusIng = "ING" // 处理中 + StatusOk = "OK" // 成功 + QuantState *State ) -// BootstrapState 在状态文件首次不存在时，将启动前已有持仓登记为底仓。 -func BootstrapState(positions []sdk.Position) { - state := getState() - if !state.Fresh() { - return +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"` + BaseStatus string `json:"base_status,omitempty"` + AddedOrderId string `json:"added_order_id"` + AddedNum int `json:"add_num"` + AddedQty int `json:"add_qty"` + AddedCost float64 `json:"add_cost"` + AddedStatus string `json:"added_status,omitempty"` +} + +func InitState(sn string) error { + absPath := path.Join(config.Global.QMTDataDir, fmt.Sprintf("%s_%s_state.json", sn, config.Account.AccountID)) + items, err := loadStateFile(absPath) + if err != nil { + return err } + + var codes []string + for code, _ := range items { + codes = append(codes, code) + } + + QuantState = &State{ + AbsPath: absPath, + Items: items, + Codes: codes, + } + return nil +} + +func loadStateFile(fp string) (map[string]*StateItem, error) { + raw, err := os.ReadFile(fp) + if err != nil { + return nil, fmt.Errorf("[状态] 读取失败: %v", err) + } + var items map[string]*StateItem + if err := json.Unmarshal(raw, &items); err != nil { + return nil, fmt.Errorf("[状态] 解析失败：%s", err) + } + return items, nil + +} + +func SyncPositions(positions []sdk.Position) error { for _, pos := range positions { code := pos.StockCode if code == "" || pos.Volume <= 0 || pos.OpenPrice <= 0 { continue } - item := state.Ensure(code) - item.BaseQty, item.BaseCost = pos.Volume, pos.OpenPrice - logf("WARNING", "[ZT][状态] %s 首次接管为底仓 数量=%d 成本=%.2f", code, pos.Volume, pos.OpenPrice) - } - state.completeBootstrap() - state.Save() -} - -func getState() *ZTState { - 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 { - st := &ZTState{ - path: filepath.Join(dataDir, fmt.Sprintf("zt_%s_state.json", accountID)), - Items: map[string]*SymbolState{}, - } - raw, err := os.ReadFile(st.path) - if err != nil { - if os.IsNotExist(err) { - st.fresh = true - return st - } - st.LoadError = err.Error() - 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 + if !slices.Contains(QuantState.Codes, code) { + item := &StateItem{ + Code: code, + BaseQty: pos.Volume, + BaseCost: pos.OpenPrice, + BaseStatus: StatusOk, + } + QuantState.Append(item) + logf("WARNING", "[状态] %s 首次接管为底仓 数量=%d 成本=%.2f", code, pos.Volume, pos.OpenPrice) } } - 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 QuantState.Save() +} + +func (s *State) Append(i *StateItem) { + s.mu.Lock() + defer s.mu.Unlock() + + s.Items[i.Code] = i + s.Codes = append(s.Codes, i.Code) +} + +func (s *State) Get(code string) (*StateItem, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if i, ok := s.Items[code]; ok { + return i, nil + } else { + return nil, fmt.Errorf("%s not found.", code) } - return st } -func (s *ZTState) Fresh() bool { +func (s *State) Set(i *StateItem) { s.mu.Lock() defer s.mu.Unlock() - return s.fresh -} -func (s *ZTState) completeBootstrap() { - 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 + if _, ok := s.Items[i.Code]; !ok { + s.Codes = append(s.Codes, i.Code) } - item := &SymbolState{Code: code} - s.Items[code] = item - return item + s.Items[i.Code] = i } -func (s *ZTState) Remove(code string) { +func (s *State) Delete(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() { - s.mu.Lock() - 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) + // 写入AbsPath文件 + f, err := os.OpenFile(s.AbsPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644) 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 { - if err := os.Rename(tmp, dest); err == nil { - return nil + encoder := json.NewEncoder(f) + encoder.SetIndent("", " ") + if err := encoder.Encode(s.Items); err != nil { + return fmt.Errorf("[状态] 写入失败: %v", err) } - _ = os.Remove(dest) - return os.Rename(tmp, dest) + return nil } diff --git a/go-client/apps/zt/logic/watch.go b/go-client/apps/zt/logic/watch.go new file mode 100644 index 0000000..326eabb --- /dev/null +++ b/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 +} diff --git a/go-client/apps/zt/main.go b/go-client/apps/zt/main.go index a6de989..b95f234 100644 --- a/go-client/apps/zt/main.go +++ b/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 { @@ -57,7 +69,7 @@ func main() { } scheduler.Start() log.Printf("[INFO] [ZT] 计划任务已启动") - + <-ctx.Done() <-scheduler.Stop().Done() log.Printf("[INFO] [ZT] 停止")