From 566a07fea82b4e9a3df6eba22401f95468f4ed00 Mon Sep 17 00:00:00 2001 From: yanweidong Date: Wed, 26 Aug 2026 12:48:48 +0800 Subject: [PATCH] refactor(zt): migrate strategy logic to new state model --- go-client/apps/zt/logic/boot.go | 5 +- go-client/apps/zt/logic/dip.go | 55 ------- go-client/apps/zt/logic/open.go | 25 +-- go-client/apps/zt/logic/order.go | 22 +-- go-client/apps/zt/logic/positions.go | 237 +++++++++++---------------- go-client/apps/zt/logic/watch.go | 2 +- go-client/config/config.go | 18 +- 7 files changed, 131 insertions(+), 233 deletions(-) delete mode 100644 go-client/apps/zt/logic/dip.go diff --git a/go-client/apps/zt/logic/boot.go b/go-client/apps/zt/logic/boot.go index 468125c..203e07e 100644 --- a/go-client/apps/zt/logic/boot.go +++ b/go-client/apps/zt/logic/boot.go @@ -87,9 +87,10 @@ func RunOnce(ctx context.Context, client *sdk.Client, books *OrderBook, signals // 7 执行开仓:有开仓信号 && 大盘指数允许开仓 if len(allowOpen) > 0 && IsAllow { - openSignal(client, books, ticks, allowOpen) + openSignal(ctx, client, books, ticks, allowOpen) } // 8 持仓计算 - managePositions(client, books, ticks, positions, IsAllow) + buyBudget := assets.Available + managePositions(ctx, client, books, ticks, positions, IsAllow, &buyBudget) } diff --git a/go-client/apps/zt/logic/dip.go b/go-client/apps/zt/logic/dip.go deleted file mode 100644 index b228e72..0000000 --- a/go-client/apps/zt/logic/dip.go +++ /dev/null @@ -1,55 +0,0 @@ -package logic - -import ( - "big-qmt/go-client/config" - "sync" - "time" -) - -type dipWatch struct { - LastClose float64 - ExpiresAt time.Time -} - -var openDip = struct { - mu sync.Mutex - store map[string]dipWatch -}{store: map[string]dipWatch{}} - -var posDip = struct { - mu sync.Mutex - store map[string]dipWatch -}{store: map[string]dipWatch{}} - -func dipTriggered(mu *sync.Mutex, store map[string]dipWatch, tag, code string, price float64) bool { - if price <= 0 { - return false - } - mu.Lock() - defer mu.Unlock() - now := time.Now() - watch, ok := store[code] - if !ok || now.After(watch.ExpiresAt) || now.Equal(watch.ExpiresAt) { - store[code] = dipWatch{LastClose: price, ExpiresAt: now.Add(time.Duration(config.Account.WatchTimeoutSec) * time.Second)} - logf("INFO", "[%s-观察] %s 现价=%.2f", tag, code, price) - return false - } - if price < watch.LastClose { - watch.LastClose = price - watch.ExpiresAt = now.Add(time.Duration(config.Account.WatchTimeoutSec) * time.Second) - store[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 < config.Account.ReboundThreshold { - logf("INFO", "[%s-等待] %s 反弹=%.2f%% 阈值=%.2f%%", tag, code, rebound, config.Account.ReboundThreshold) - return false - } - delete(store, code) - logf("INFO", "[%s-触发] %s 反弹=%.2f%% 低点=%.2f", tag, code, rebound, watch.LastClose) - return true -} diff --git a/go-client/apps/zt/logic/open.go b/go-client/apps/zt/logic/open.go index 2f0a3b1..80c7214 100644 --- a/go-client/apps/zt/logic/open.go +++ b/go-client/apps/zt/logic/open.go @@ -1,38 +1,43 @@ package logic import ( + "context" + "big-qmt/go-client/config" "big-qmt/go-client/libs" "big-qmt/go-client/sdk" ) -func openSignal(client *sdk.Client, books *OrderBook, ticks map[string]sdk.Tick, openSignals []libs.SignalItem) { - state := getState() - if state.LoadError != "" { - logf("ERROR", "[ZT][开仓] 状态文件异常，禁止新开仓: %s", state.LoadError) - return - } +func openSignal(ctx context.Context, client *sdk.Client, books *OrderBook, ticks map[string]sdk.Tick, openSignals []libs.SignalItem) { for _, item := range openSignals { - if state.Get(item.Code) != nil { + // 是否有锁 + if _, err := QuantState.Get(item.Code); err == nil { continue } + // 验证价格 price := ticks[item.Code].LastPrice if price <= 0 { continue } - if !dipTriggered(&openDip.mu, openDip.store, "开仓", item.Code, price) { + // 防止接飞刀 + if !OpenWatch.Triggered("开仓", item.Code, price) { continue } + // 计算开仓数量 volume := libs.CalcBuyVolume(price, config.Account.BuyValue) if volume <= 0 { continue } + // 开仓 orderID := newOrderTag("base") if !books.place(ctx, client, sideBuy, item.Code, volume, orderID) { continue } - setPending(state.Ensure(item.Code), pendingBaseOpening, orderID) - state.Save() + // 保存数量 + QuantState.Set(&StateItem{Code: item.Code, BaseOrderId: orderID, BaseQty: volume, BaseCost: price, BaseStatus: StatusIng}) + if err := QuantState.Save(); err != nil { + logf("ERROR", "%v", err) + } logf("INFO", "[ZT][开仓] %s 买入 %d 股", item.Code, volume) } } diff --git a/go-client/apps/zt/logic/order.go b/go-client/apps/zt/logic/order.go index a3fdc9e..a59c82c 100644 --- a/go-client/apps/zt/logic/order.go +++ b/go-client/apps/zt/logic/order.go @@ -105,11 +105,6 @@ func (o *OrderBook) readReceipts() { o.mu.Unlock() status := strings.ToLower(receipt.Status) - state := getState() - if item := state.Get(receipt.StockCode); item != nil && item.PendingOrderID == receipt.OrderID { - item.OrderStatus = status - state.Save() - } if (status == "filled" || status == "cancelled" || status == "rejected") && (receipt.Side == sideBuy || receipt.Side == sideSell) { o.unlockSide(receipt.StockCode, receipt.Side) o.invalidate() @@ -184,7 +179,7 @@ func (o *OrderBook) CancelExpired(ctx context.Context, client *sdk.Client) bool if err != nil { return false } - state := getState() + state := QuantState now := time.Now() timeout := time.Duration(config.Account.OrderTimeoutSec) * time.Second seen := map[string]struct{}{} @@ -243,7 +238,7 @@ func (o *OrderBook) CancelExpired(ctx context.Context, client *sdk.Client) bool return true } -func (o *OrderBook) claimed(state *ZTState, order parsedOrder) bool { +func (o *OrderBook) claimed(state *State, order parsedOrder) bool { if order.RemarkOwned { return true } @@ -258,18 +253,11 @@ func (o *OrderBook) claimed(state *ZTState, order parsedOrder) bool { if state == nil { return false } - item := state.Get(order.StockCode) - if item == nil || item.Pending == "" { - return false - } - switch item.Pending { - case pendingBaseOpening, pendingAdd: - return order.Side == sideBuy - case pendingSellAdd, pendingSellBase: - return order.Side == sideSell - default: + item, err := state.Get(order.StockCode) + if err != nil { return false } + return order.OrderID == item.BaseOrderId || order.OrderID == item.AddedOrderId || item.BaseStatus == StatusIng || item.AddedStatus == StatusIng } func (o *OrderBook) unlockSide(code, side string) { diff --git a/go-client/apps/zt/logic/positions.go b/go-client/apps/zt/logic/positions.go index 8ce3bef..fd3da96 100644 --- a/go-client/apps/zt/logic/positions.go +++ b/go-client/apps/zt/logic/positions.go @@ -6,6 +6,7 @@ import ( "sync" "big-qmt/go-client/config" + "big-qmt/go-client/libs" "big-qmt/go-client/sdk" ) @@ -14,29 +15,37 @@ var peakGrids = map[string]int{} func peakKey(code, leg string) string { return code + "|" + leg } -func managePositions(client *sdk.Client, books *OrderBook, ticks map[string]sdk.Tick, positions []sdk.Position, marketOK bool) { - if positions == nil { - logf("ERROR", "[ZT][持仓] 持仓查询失败，本轮跳过") +func calcBuyVolume(price, value float64) int { + return libs.CalcBuyVolume(price, value) +} + +func stateCodes(state *State) []string { + state.mu.Lock() + defer state.mu.Unlock() + return append([]string(nil), state.Codes...) +} + +func managePositions(ctx context.Context, client *sdk.Client, books *OrderBook, ticks map[string]sdk.Tick, positions []sdk.Position, marketOK bool, buyBudget *float64) { + if positions == nil || QuantState == nil { + logf("ERROR", "[ZT][持仓] 持仓或状态不可用，本轮跳过") return } - state := getState() - if state.LoadError != "" { - logf("ERROR", "[ZT][持仓] 状态文件异常，本轮停止交易: %s", state.LoadError) + buys, sells, ok := books.activeSets(ctx, client) + if !ok { return } before := map[string]struct{}{} - for _, code := range state.Codes() { + for _, code := range stateCodes(QuantState) { before[code] = struct{}{} } if ticks == nil { ticks = map[string]sdk.Tick{} } - logf("INFO", "[ZT][持仓] 开始处理 %d 只", len(positions)) type row struct { volume, usable int avg, price float64 stock string - item *SymbolState + item *StateItem } rows := make([]row, 0, len(positions)) seen := map[string]struct{}{} @@ -46,20 +55,18 @@ func managePositions(client *sdk.Client, books *OrderBook, ticks map[string]sdk. continue } seen[code] = struct{}{} - item := syncItem(state, code, pos.Volume, pos.OpenPrice, buys, sells, books) - if pos.Volume <= 0 { - continue + item := syncItem(QuantState, code, pos.Volume, pos.OpenPrice, buys, sells, books) + if pos.Volume > 0 { + rows = append(rows, row{stock: code, volume: pos.Volume, usable: pos.CanUseVolume, avg: pos.OpenPrice, price: ticks[code].LastPrice, item: item}) } - price := ticks[code].LastPrice - rows = append(rows, row{stock: code, volume: pos.Volume, usable: pos.CanUseVolume, avg: pos.OpenPrice, price: price, item: item}) } - for _, code := range state.Codes() { + for _, code := range stateCodes(QuantState) { if _, ok := seen[code]; !ok { - syncItem(state, code, 0, 0, buys, sells, books) + syncItem(QuantState, code, 0, 0, buys, sells, books) } } after := map[string]struct{}{} - for _, code := range state.Codes() { + for _, code := range stateCodes(QuantState) { after[code] = struct{}{} } for code := range before { @@ -68,29 +75,20 @@ func managePositions(client *sdk.Client, books *OrderBook, ticks map[string]sdk. } } for _, r := range rows { - if r.item == nil || r.item.Pending != "" { + if r.item == nil || r.item.BaseStatus == StatusIng || r.item.AddedStatus == StatusIng || r.avg <= 0 || r.price <= 0 || r.volume%100 != 0 { continue } - if r.avg <= 0 || r.price <= 0 || r.volume%100 != 0 { + if r.volume != r.item.BaseQty+r.item.AddedQty { + logf("INFO", "[ZT][持仓] %s 数量异常，底仓=%d 补仓=%d 现有=%d", r.stock, r.item.BaseQty, r.item.AddedQty, r.volume) continue } - if r.volume != r.item.BaseQty+r.item.AddQty { - logf("INFO", "[ZT][持仓] %s 数量异常，底仓=%d 补仓=%d 现有=%d", r.stock, r.item.BaseQty, r.item.AddQty, r.volume) - continue - } - holdingAdd := r.item.AddQty > 0 - legName := "底仓" - if holdingAdd { - legName = "补仓腿" - } - logf("INFO", "[ZT][持仓] %s 现价=%.2f 成本=%.2f 可用=%d %s", r.stock, r.price, r.avg, r.usable, legName) - if holdingAdd { + if r.item.AddedQty > 0 { addPnL := -999.0 - if r.item.AddCost > 0 { - addPnL = (r.price - r.item.AddCost) / r.item.AddCost * 100 + if r.item.AddedCost > 0 { + addPnL = (r.price - r.item.AddedCost) / r.item.AddedCost * 100 } if retreated(r.item, "add", addPnL) { - sellLeg(ctx, client, books, r.item, r.usable, r.item.AddQty, "add", addPnL) + sellLeg(ctx, client, books, r.item, r.usable, r.item.AddedQty, "add", addPnL) } continue } @@ -100,138 +98,96 @@ func managePositions(client *sdk.Client, books *OrderBook, ticks map[string]sdk. } if retreated(r.item, "base", basePnL) { sellLeg(ctx, client, books, r.item, r.usable, r.item.BaseQty, "base", basePnL) - } else if r.item.AddQty <= 0 && r.item.AddCost <= 0 && basePnL <= config.Account.LossTriggerPct { + } else if basePnL <= config.Account.LossTriggerPct { addOnRebound(ctx, client, books, r.item, r.price, marketOK, buyBudget) } } - // 首次没有状态文件时，本轮已将启动前持仓全部接管为底仓。 - state.completeBootstrap() - state.Save() + if err := QuantState.Save(); err != nil { + logf("ERROR", "%v", err) + } } -func syncItem(state *ZTState, code string, volume int, avgPrice float64, buys, sells map[string]struct{}, books *OrderBook) *SymbolState { - item := state.Get(code) - if item == nil { +func syncItem(state *State, code string, volume int, avgPrice float64, buys, sells map[string]struct{}, books *OrderBook) *StateItem { + item, err := state.Get(code) + if err != nil { if volume > 0 { logf("ERROR", "[ZT][持仓] %s 无本地状态，跳过", code) } return nil } - switch item.Pending { - case pendingBaseOpening: - syncOpen(state, item, volume, avgPrice, buys, books) - case pendingAdd: - syncAdd(state, item, volume, avgPrice, buys, books) - case pendingSellAdd: - syncSellAdd(state, item, volume, avgPrice, sells, books) - case pendingSellBase: - syncSellBase(state, item, volume, avgPrice, sells, books) - default: - if volume <= 0 { - state.Remove(code) - logf("INFO", "[ZT][持仓] %s 已无持仓，清除状态", code) - return nil - } + if item.BaseStatus == StatusIng { + syncBase(state, item, volume, avgPrice, buys, sells, books) + } else if item.AddedStatus == StatusIng { + syncAdded(state, item, volume, avgPrice, buys, sells, books) + } else if volume <= 0 { + state.Delete(code) + return nil } - return state.Get(code) + item, _ = state.Get(code) + return item } -func syncOpen(state *ZTState, item *SymbolState, volume int, avgPrice float64, buys map[string]struct{}, books *OrderBook) { - if volume > 0 { - item.BaseQty, item.BaseCost = volume, avgPrice - } - if books.sideBusy(item.Code, sideBuy, buys) { +func syncBase(state *State, item *StateItem, volume int, avgPrice float64, buys, sells map[string]struct{}, books *OrderBook) { + if books.sideBusy(item.Code, sideBuy, buys) || books.sideBusy(item.Code, sideSell, sells) { return } if volume <= 0 { - state.Remove(item.Code) - logf("INFO", "[ZT][委托] %s 开仓委托已失效，允许重新开仓", item.Code) + state.Delete(item.Code) return } - clearPending(item) - logf("INFO", "[ZT][持仓] %s 开仓确认 数量=%d 成本=%.2f", item.Code, item.BaseQty, item.BaseCost) + item.BaseQty = volume - item.AddedQty + if item.BaseQty < 0 { + item.BaseQty, item.AddedQty, item.AddedCost, item.AddedStatus = volume, 0, 0, StatusNone + } + item.BaseCost = avgPrice + item.BaseStatus = StatusOk + state.Set(item) } -func syncAdd(state *ZTState, item *SymbolState, volume int, avgPrice float64, buys map[string]struct{}, books *OrderBook) { +func syncAdded(state *State, item *StateItem, volume int, avgPrice float64, buys, sells map[string]struct{}, books *OrderBook) { + if books.sideBusy(item.Code, sideBuy, buys) || books.sideBusy(item.Code, sideSell, sells) { + return + } + if volume <= 0 { + state.Delete(item.Code) + return + } if volume > item.BaseQty { - item.AddQty = volume - item.BaseQty - if item.AddQty > 0 { - item.AddCost = math.Max(0, (avgPrice*float64(volume)-item.BaseCost*float64(item.BaseQty))/float64(item.AddQty)) - } - } - if books.sideBusy(item.Code, sideBuy, buys) { - return - } - if volume <= 0 { - state.Remove(item.Code) - logf("INFO", "[ZT][委托] %s 补仓后无持仓，清除状态", item.Code) - return - } - if volume <= item.BaseQty { - item.AddQty = 0 - item.AddCost = 0 - logf("INFO", "[ZT][持仓] %s 补仓未成交，回退底仓", item.Code) - } - clearPending(item) -} - -func syncSellAdd(state *ZTState, item *SymbolState, volume int, avgPrice float64, sells map[string]struct{}, books *OrderBook) { - if volume <= 0 { - if !books.sideBusy(item.Code, sideSell, sells) { - state.Remove(item.Code) - logf("INFO", "[ZT][委托] %s 卖出后已无持仓，清除状态", item.Code) - } - return - } - if volume <= item.BaseQty { + item.AddedQty = volume - item.BaseQty + item.AddedCost = math.Max(0, (avgPrice*float64(volume)-item.BaseCost*float64(item.BaseQty))/float64(item.AddedQty)) + item.AddedStatus = StatusOk + } else { item.BaseQty, item.BaseCost = volume, avgPrice - item.AddQty = 0 + item.AddedQty, item.AddedCost, item.AddedStatus = 0, 0, StatusNone peakMu.Lock() delete(peakGrids, peakKey(item.Code, "add")) peakMu.Unlock() - } else { - item.AddQty = volume - item.BaseQty - } - if !books.sideBusy(item.Code, sideSell, sells) { - clearPending(item) } + state.Set(item) } -func syncSellBase(state *ZTState, item *SymbolState, volume int, avgPrice float64, sells map[string]struct{}, books *OrderBook) { - if volume <= 0 { - if !books.sideBusy(item.Code, sideSell, sells) { - state.Remove(item.Code) - logf("INFO", "[ZT][委托] %s 卖出后已无持仓，清除状态", item.Code) - } +func addOnRebound(ctx context.Context, client *sdk.Client, books *OrderBook, item *StateItem, price float64, marketOK bool, buyBudget *float64) { + if !marketOK || PosbuyWatch == nil || !PosbuyWatch.Triggered("补仓", item.Code, price) { return } - item.BaseQty, item.BaseCost = volume, avgPrice - if !books.sideBusy(item.Code, sideSell, sells) { - clearPending(item) - } -} - -func addOnRebound(ctx context.Context, client *sdk.Client, books *OrderBook, item *SymbolState, price float64, marketOK bool, buyBudget *float64) { - if !marketOK || !dipTriggered(&posDip.mu, posDip.store, "补仓", item.Code, price) { - return - } - volume := calcBuyVolume(price, config.Account.BuyValue) + volume := libs.CalcBuyVolume(price, config.Account.BuyValue) estimated := price * float64(volume) - if buyBudget == nil || estimated > *buyBudget { - logf("INFO", "[ZT][补仓] %s 可用买入预算不足，需要=%.2f", item.Code, estimated) + if volume <= 0 || buyBudget == nil || estimated > *buyBudget { return } orderID := newOrderTag("add") if books.place(ctx, client, sideBuy, item.Code, volume, orderID) { - item.AddCost = price - setPending(item, pendingAdd, orderID) + item.AddedOrderId, item.AddedQty, item.AddedCost, item.AddedStatus = orderID, volume, price, StatusIng + item.AddedNum++ + QuantState.Set(item) *buyBudget -= estimated - getState().Save() - logf("INFO", "[ZT][补仓] %s 买入 %d 股", item.Code, volume) + if err := QuantState.Save(); err != nil { + logf("ERROR", "%v", err) + } } } -func retreated(item *SymbolState, leg string, pnl float64) bool { +func retreated(item *StateItem, leg string, pnl float64) bool { if pnl < config.Account.MinProfitPct { return false } @@ -242,16 +198,14 @@ func retreated(item *SymbolState, leg string, pnl float64) bool { peak, ok := peakGrids[key] if !ok || grid > peak { peakGrids[key] = grid - logf("INFO", "[ZT][止盈] %s %s峰值网格=%d", item.Code, leg, grid) return false } return grid < peak } -func sellLeg(ctx context.Context, client *sdk.Client, books *OrderBook, item *SymbolState, usable, volume int, leg string, pnl float64) { +func sellLeg(ctx context.Context, client *sdk.Client, books *OrderBook, item *StateItem, usable, volume int, leg string, pnl float64) { volume -= volume % 100 if volume <= 0 || usable < volume { - logf("INFO", "[ZT][止盈] %s 可用股数不足，需要=%d 可用=%d", item.Code, volume, usable) return } orderID := newOrderTag(leg) @@ -259,21 +213,28 @@ func sellLeg(ctx context.Context, client *sdk.Client, books *OrderBook, item *Sy return } if leg == "add" { - setPending(item, pendingSellAdd, orderID) + item.AddedOrderId, item.AddedStatus = orderID, StatusIng } else { - setPending(item, pendingSellBase, orderID) + item.BaseOrderId, item.BaseStatus = orderID, StatusIng + } + QuantState.Set(item) + if err := QuantState.Save(); err != nil { + logf("ERROR", "%v", err) } - getState().Save() logf("INFO", "[ZT][止盈] %s 卖出 %d 股，%s腿盈利=%.2f%%", item.Code, volume, leg, pnl) } func forget(code string) { - openDip.mu.Lock() - delete(openDip.store, code) - openDip.mu.Unlock() - posDip.mu.Lock() - delete(posDip.store, code) - posDip.mu.Unlock() + if OpenWatch != nil { + OpenWatch.mu.Lock() + delete(OpenWatch.Data, code) + OpenWatch.mu.Unlock() + } + if PosbuyWatch != nil { + PosbuyWatch.mu.Lock() + delete(PosbuyWatch.Data, code) + PosbuyWatch.mu.Unlock() + } peakMu.Lock() delete(peakGrids, peakKey(code, "base")) delete(peakGrids, peakKey(code, "add")) diff --git a/go-client/apps/zt/logic/watch.go b/go-client/apps/zt/logic/watch.go index 326eabb..132d899 100644 --- a/go-client/apps/zt/logic/watch.go +++ b/go-client/apps/zt/logic/watch.go @@ -19,7 +19,7 @@ type dipWatch struct { } type WatchMu struct { - mu *sync.Mutex + mu sync.Mutex Data map[string]dipWatch } diff --git a/go-client/config/config.go b/go-client/config/config.go index 178a31d..17a8ee9 100644 --- a/go-client/config/config.go +++ b/go-client/config/config.go @@ -25,16 +25,14 @@ type GlobalConfig struct { } type AccountConfig struct { - AccountID string `yaml:"account_id"` - HostKey string `yaml:"host_key"` - OrderTimeoutSec int `yaml:"order_timeout_seconds"` - BuyValue float64 `yaml:"buy_value"` - MinCashRatio float64 `yaml:"min_cash_ratio"` - LossTriggerPct float64 `yaml:"loss_trigger_pct"` - GridStepPct float64 `yaml:"grid_step_pct"` - MinProfitPct float64 `yaml:"min_profit_pct"` - WatchTimeoutSec int `yaml:"watch_timeout_seconds"` - ReboundThreshold float64 `yaml:"rebound_threshold"` + AccountID string `yaml:"account_id"` + HostKey string `yaml:"host_key"` + OrderTimeoutSec int `yaml:"order_timeout_seconds"` + BuyValue float64 `yaml:"buy_value"` + MinCashRatio float64 `yaml:"min_cash_ratio"` + LossTriggerPct float64 `yaml:"loss_trigger_pct"` + GridStepPct float64 `yaml:"grid_step_pct"` + MinProfitPct float64 `yaml:"min_profit_pct"` } // Load 根据 global.yaml 中的 hosts 映射加载当前计算机的账户配置。