quant/big-qmt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(zt): migrate strategy logic to new state model

Browse Source
This commit is contained in:
yanweidong
2026-08-26 12:48:48 +08:00
parent 9604126ee7
commit 566a07fea8
7 changed files with 131 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
go-client/apps/zt/logic/boot.go
View File

@@ -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)
}

55
go-client/apps/zt/logic/dip.go
View File

@@ -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
}

25
go-client/apps/zt/logic/open.go
View File

@@ -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)
}
}

22
go-client/apps/zt/logic/order.go
View File

@@ -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) {

237
go-client/apps/zt/logic/positions.go
View File

@@ -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"))

2
go-client/apps/zt/logic/watch.go
View File

@@ -19,7 +19,7 @@ type dipWatch struct {
}
type WatchMu struct {
mu *sync.Mutex
mu sync.Mutex
Data map[string]dipWatch
}

18
go-client/config/config.go
View File

@@ -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 映射加载当前计算机的账户配置。