diff --git a/api/QMT_API.py b/api/QMT_API.py index 9bf80de..a8be7b0 100644 --- a/api/QMT_API.py +++ b/api/QMT_API.py @@ -280,8 +280,8 @@ class FullTickHandler(BaseHandler): def post(self): data = json.loads(self.request.body) stocks = data.get('stocks', []) - if not stocks: - raise HTTPError(400, "need args stocks") + #if not stocks: + # raise HTTPError(400, "need args stocks") ret = safe_call(self.ctx().get_full_tick, stocks) if not ret: raise HTTPError(500, "获取分笔行情失败") @@ -913,7 +913,7 @@ class TradeDetailDataHandler(BaseHandler): data = json.loads(self.request.body) account = data.get('account', 'stock') datatype = data.get('datatype', 'position') - ret = safe_call(get_trade_detail_data, self.acc(), account, datatype, 'qmt') + ret = safe_call(get_trade_detail_data, self.acc(), account, datatype) if ret is None: ret = [] result = [] @@ -956,7 +956,7 @@ class LastOrderIdHandler(BaseHandler): data = json.loads(self.request.body) account = data.get('account', 'stock') datatype = data.get('datatype', 'ORDER') - ret = safe_call(get_last_order_id, self.acc(), account, datatype, 'qmt') + ret = safe_call(get_last_order_id, self.acc(), account, datatype) self.write(json.dumps({"last_order_id": ret}, ensure_ascii=False)) # can_cancel_order() - 查询委托是否可撤销 @@ -1163,7 +1163,8 @@ class BuyHandler(BaseHandler): price = float(data['price']) volume = int(data['volume']) pr_type = data.get('prType', 11) - order_ref = passorder(23, 1101, self.acc(), stock, pr_type, price, volume, 'qmt', 2, self.ctx()) + sn = data.get('sn','') + order_ref = passorder(23, 1101, self.acc(), stock, pr_type, price, volume, sn, 2, self.ctx()) self.write(json.dumps({ "status": "success", "action": "buy", "stock": stock, "order_ref": str(order_ref) if order_ref else "unknown" @@ -1181,7 +1182,8 @@ class SellHandler(BaseHandler): price = float(data['price']) volume = int(data['volume']) pr_type = data.get('prType', 11) - order_ref = passorder(24, 1101, self.acc(), stock, pr_type, price, volume, 'qmt', 2, self.ctx()) + sn = data.get('sn','') + order_ref = passorder(24, 1101, self.acc(), stock, pr_type, price, volume, sn, 2, self.ctx()) self.write(json.dumps({ "status": "success", "action": "sell", "stock": stock, "order_ref": str(order_ref) if order_ref else "unknown" @@ -1195,7 +1197,8 @@ class OrderStatusHandler(BaseHandler): def post(self): data = json.loads(self.request.body) account = data.get('account', 'stock') - orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', 'qmt') or [] + sn = data.get('sn','') + orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', sn) or [] rets = [] for order in orders: rets.append({ @@ -1212,7 +1215,8 @@ class CancelAllHandler(BaseHandler): try: data = json.loads(self.request.body) account = data.get('account', 'stock') - orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', 'qmt') or [] + sn = data.get('sn','') + orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', sn) or [] canceled_list = [] for order in orders: if can_cancel_order(order.m_strOrderSysID, self.acc(), account): @@ -1237,11 +1241,12 @@ class CancelByRuleHandler(BaseHandler): try: data = json.loads(self.request.body) stock = data.get('stock') + sn = data.get('sn','') cancel_volume = int(data.get('volume', 0)) account = data.get('account', 'stock') if not stock or cancel_volume <= 0: raise HTTPError(400, "参数错误：必须提供 stock 且 volume > 0") - orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', 'qmt') or [] + orders = safe_call(get_trade_detail_data, self.acc(), account, 'order', sn) or [] target_orders = [] for order in orders: order_code = f"{order.m_strInstrumentID}.{order.m_strExchangeID}" @@ -1293,7 +1298,8 @@ class DealHandler(BaseHandler): def post(self): data = json.loads(self.request.body) account = data.get('account', 'stock') - deals = safe_call(get_trade_detail_data, self.acc(), account, 'deal', 'qmt') or [] + sn = data.get('sn','') + deals = safe_call(get_trade_detail_data, self.acc(), account, 'deal', sn) or [] rets = [] for deal in deals: attrs = {} diff --git a/go-client/apps/zt/logic/boot.go b/go-client/apps/zt/logic/boot.go index 203e07e..04f2501 100644 --- a/go-client/apps/zt/logic/boot.go +++ b/go-client/apps/zt/logic/boot.go @@ -39,13 +39,13 @@ func Overview(assets *sdk.Assets, positions []sdk.Position) { } } -func RunOnce(ctx context.Context, client *sdk.Client, books *OrderBook, signals *libs.SignalResult) { +func RunOnce(ctx context.Context, client *sdk.Client, signals *libs.SignalResult) { if !libs.TradingTime(time.Now()) { return } // 1 取消过期订单 - books.CancelExpired(ctx, client) + OrderBook.CancelExpired(client) // 2 验证可用资金 assets, err := client.Assets(ctx) @@ -87,10 +87,10 @@ func RunOnce(ctx context.Context, client *sdk.Client, books *OrderBook, signals // 7 执行开仓:有开仓信号 && 大盘指数允许开仓 if len(allowOpen) > 0 && IsAllow { - openSignal(ctx, client, books, ticks, allowOpen) + openSignal(ctx, client, ticks, allowOpen) } // 8 持仓计算 buyBudget := assets.Available - managePositions(ctx, client, books, ticks, positions, IsAllow, &buyBudget) + managePositions(ctx, client, ticks, positions, IsAllow, &buyBudget) } diff --git a/go-client/apps/zt/logic/open.go b/go-client/apps/zt/logic/open.go index 80c7214..3cc0680 100644 --- a/go-client/apps/zt/logic/open.go +++ b/go-client/apps/zt/logic/open.go @@ -8,10 +8,10 @@ import ( "big-qmt/go-client/sdk" ) -func openSignal(ctx context.Context, client *sdk.Client, books *OrderBook, ticks map[string]sdk.Tick, openSignals []libs.SignalItem) { +func openSignal(ctx context.Context, client *sdk.Client, ticks map[string]sdk.Tick, openSignals []libs.SignalItem) { for _, item := range openSignals { // 是否有锁 - if _, err := QuantState.Get(item.Code); err == nil { + if OrderBook.IsLock("BUY", item.Code) { continue } // 验证价格 @@ -29,11 +29,11 @@ func openSignal(ctx context.Context, client *sdk.Client, books *OrderBook, ticks continue } // 开仓 - orderID := newOrderTag("base") - if !books.place(ctx, client, sideBuy, item.Code, volume, orderID) { + orderID := NewOrderID("base") + if !OrderBook.Place(client, sdk.OpBuy, item.Code, volume, orderID) { continue } - // 保存数量 + // 保存状态 QuantState.Set(&StateItem{Code: item.Code, BaseOrderId: orderID, BaseQty: volume, BaseCost: price, BaseStatus: StatusIng}) if err := QuantState.Save(); err != nil { logf("ERROR", "%v", err) diff --git a/go-client/apps/zt/logic/order.go b/go-client/apps/zt/logic/order.go index a59c82c..c0c88c0 100644 --- a/go-client/apps/zt/logic/order.go +++ b/go-client/apps/zt/logic/order.go @@ -4,415 +4,151 @@ import ( "context" "crypto/rand" "encoding/hex" - "encoding/json" "fmt" - "os" - "path/filepath" + "slices" "strconv" "strings" "sync" "time" - "big-qmt/go-client/config" "big-qmt/go-client/sdk" ) -type orderReceipt struct { - OrderID string `json:"order_id"` - QMTOrderID string `json:"qmt_order_id"` - StockCode string `json:"stock_code"` - Side string `json:"side"` - Status string `json:"status"` - RequestedVolume int `json:"requested_volume"` - TradedVolume int `json:"traded_volume"` -} +var ( + STOCK_DIRECTION = 48 + STOCK_SIDE_BUY = 48 + STOCK_SIDE_SELL = 49 + OffsetFlag = map[string]string{"23": "BUY", "24": "SELL", "48": "BUY", "49": "SELL"} + OrderTimeout = 5 * time.Minute -const ( - opBuyStock = 23 - opBuyAlt = 48 - sideBuy = "buy" - sideSell = "sell" + OrderBook *Books ) -var activeStatuses = map[int]struct{}{ - 48: {}, 49: {}, 50: {}, 51: {}, 52: {}, 55: {}, +type OrderItem struct { + ID string + Code string + Side string + Remark string + Status string + CreatedAt time.Time + Volume int } -type parsedOrder struct { - OrderID string - StockCode string - Side string - Active bool - OrderTime int64 - RemarkOwned bool - VolumeOrig int - VolumeLeft int - VolumeTraded int - Tag string +type Books struct { + mu sync.Mutex + Data map[string]*OrderItem + Index []string } -func (o parsedOrder) cancelVolume() int { - n := o.VolumeLeft + o.VolumeTraded - if n > 0 { - return n - } - return o.VolumeOrig +func NewOrderBook() { + OrderBook = &Books{Data: make(map[string]*OrderItem), Index: make([]string, 0)} } -type submission struct { - Code string - Side string -} - -type OrderBook struct { - mu sync.Mutex - cached []parsedOrder - hasCache bool - buyLocks map[string]time.Time - sellLocks map[string]time.Time - subs []submission - receipts map[string]time.Time -} - -func NewOrderBook() *OrderBook { - return &OrderBook{ - buyLocks: map[string]time.Time{}, - sellLocks: map[string]time.Time{}, - receipts: map[string]time.Time{}, - } -} - -// readReceipts 读取 QMT 回写并同步委托状态。 -func (o *OrderBook) readReceipts() { - paths, _ := filepath.Glob(filepath.Join(config.Global.QMTDataDir, "order_*.json")) - for _, path := range paths { - info, err := os.Stat(path) - if err != nil { - continue - } - o.mu.Lock() - last := o.receipts[path] - o.mu.Unlock() - if !info.ModTime().After(last) { - continue - } - receipt, err := loadReceipt(path) - if err != nil || !strings.HasPrefix(receipt.OrderID, "zt-") || receipt.StockCode == "" || receipt.Status == "" { - continue - } - o.mu.Lock() - o.receipts[path] = info.ModTime() - o.mu.Unlock() - - status := strings.ToLower(receipt.Status) - if (status == "filled" || status == "cancelled" || status == "rejected") && (receipt.Side == sideBuy || receipt.Side == sideSell) { - o.unlockSide(receipt.StockCode, receipt.Side) - o.invalidate() - } - logf("INFO", "[ZT][回写] %s status=%s traded=%d/%d", receipt.OrderID, status, receipt.TradedVolume, receipt.RequestedVolume) - } -} - -func loadReceipt(path string) (*orderReceipt, error) { - raw, err := os.ReadFile(path) - if err != nil { - return nil, err - } - var receipt orderReceipt - err = json.Unmarshal(raw, &receipt) - return &receipt, err -} - -func (o *OrderBook) invalidate() { - o.mu.Lock() - defer o.mu.Unlock() - o.hasCache = false - o.cached = nil -} - -func (o *OrderBook) query(ctx context.Context, client *sdk.Client) ([]parsedOrder, error) { - o.mu.Lock() - if o.hasCache { - out := append([]parsedOrder(nil), o.cached...) - o.mu.Unlock() - return out, nil - } - o.mu.Unlock() - raw, err := client.TradeDetailData(ctx, "order") - if err != nil { - logf("ERROR", "[ZT][委托] 查询失败: %v", err) - return nil, err - } - orders := make([]parsedOrder, 0, len(raw)) - for _, item := range raw { - orders = append(orders, parseOrder(item)) - } - o.mu.Lock() - o.cached = orders - o.hasCache = true - o.mu.Unlock() - return orders, nil -} - -func (o *OrderBook) activeSets(ctx context.Context, client *sdk.Client) (buys, sells map[string]struct{}, ok bool) { - orders, err := o.query(ctx, client) - if err != nil { - return nil, nil, false - } - buys, sells = map[string]struct{}{}, map[string]struct{}{} - for _, item := range orders { - if !item.Active || item.StockCode == "" { - continue - } - if item.Side == sideBuy { - buys[item.StockCode] = struct{}{} - } else { - sells[item.StockCode] = struct{}{} - } - } - return buys, sells, true -} - -func (o *OrderBook) CancelExpired(ctx context.Context, client *sdk.Client) bool { - o.invalidate() - orders, err := o.query(ctx, client) - if err != nil { - return false - } - state := QuantState - now := time.Now() - timeout := time.Duration(config.Account.OrderTimeoutSec) * time.Second - seen := map[string]struct{}{} - cancelled := false - for _, order := range orders { - if !order.Active || order.StockCode == "" { - continue - } - if !o.claimed(state, order) { - continue - } - if order.OrderTime <= 0 || now.Sub(time.Unix(order.OrderTime, 0)) <= timeout { - continue - } - vol := order.cancelVolume() - if vol <= 0 { - logf("WARNING", "[ZT][委托] 超时单缺少数量，跳过 %s %s", order.OrderID, order.StockCode) - continue - } - key := order.StockCode + "|" + strconv.Itoa(vol) - if _, ok := seen[key]; ok { - continue - } - seen[key] = struct{}{} - if order.OrderID != "" { - can, err := client.CanCancelOrder(ctx, order.OrderID) - if err != nil { - logf("ERROR", "[ZT][委托] 查询是否可撤失败 %s: %v", order.OrderID, err) - continue - } - if !truthy(can) { - logf("INFO", "[ZT][委托] 不可撤 %s %s", order.OrderID, order.StockCode) - continue - } - } - ret, err := client.CancelByRule(ctx, order.StockCode, vol) - if err != nil { - logf("ERROR", "[ZT][委托] 撤单失败 %s %s: %v", order.OrderID, order.StockCode, err) - continue - } - if ret == nil || ret.Status != "success" { - msg := "" - if ret != nil { - msg = ret.Message - } - logf("WARNING", "[ZT][委托] 规则撤单未命中 %s %s volume=%d %s", order.OrderID, order.StockCode, vol, msg) - continue - } - o.unlockSide(order.StockCode, order.Side) - cancelled = true - logf("INFO", "[ZT][委托] 撤销超时单 %s %s %s volume=%d", order.OrderID, order.StockCode, order.Side, vol) - } - if cancelled { - o.invalidate() - } - return true -} - -func (o *OrderBook) claimed(state *State, order parsedOrder) bool { - if order.RemarkOwned { - return true - } - o.mu.Lock() - for _, s := range o.subs { - if s.Code == order.StockCode && s.Side == order.Side { - o.mu.Unlock() - return true - } - } - o.mu.Unlock() - if state == nil { - return false - } - 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) { - o.mu.Lock() - defer o.mu.Unlock() - delete(o.locks(side), code) - n := 0 - for _, s := range o.subs { - if s.Code == code && s.Side == side { - continue - } - o.subs[n] = s - n++ - } - o.subs = o.subs[:n] -} - -func (o *OrderBook) sideBusy(code, side string, active map[string]struct{}) bool { - if _, ok := active[code]; ok { - return true - } - return o.locked(code, side) -} - -func (o *OrderBook) locked(code, side string) bool { - o.mu.Lock() - defer o.mu.Unlock() - ts, ok := o.locks(side)[code] - return ok && time.Since(ts) < time.Duration(config.Account.OrderTimeoutSec)*time.Second -} - -func (o *OrderBook) locks(side string) map[string]time.Time { - if side == sideBuy { - return o.buyLocks - } - return o.sellLocks -} - -func (o *OrderBook) hasActive(ctx context.Context, client *sdk.Client, code, side string) bool { - orders, err := o.query(ctx, client) - if err != nil { - return true - } - for _, item := range orders { - if item.StockCode == code && item.Active && item.Side == side { - return true - } - } - return false -} - -func (o *OrderBook) place(ctx context.Context, client *sdk.Client, side, code string, volume int, tag string) bool { - if volume <= 0 || volume%100 != 0 { - logf("ERROR", "[ZT][委托] %s 拒绝非整手数量=%d", code, volume) - return false - } - if o.locked(code, side) { - logf("INFO", "[ZT][委托] %s %s锁定中", code, side) - return false - } - if o.hasActive(ctx, client, code, side) { - logf("INFO", "[ZT][委托] %s 已有%s在途委托", code, side) - return false - } - _, err := client.PassorderLatestTagged(ctx, side == sideBuy, code, volume, tag) - if err != nil { - logf("ERROR", "[ZT][委托] %s 异常: %v", code, err) - return false - } - o.mu.Lock() - o.locks(side)[code] = time.Now() - o.subs = append(o.subs, submission{Code: code, Side: side}) - o.mu.Unlock() - logf("INFO", "[ZT][委托] 已提交 %s %s %d股 tag=%s", side, code, volume, tag) - return true -} - -func parseOrder(item map[string]string) parsedOrder { - operation, _ := strconv.Atoi(item["m_nOffsetFlag"]) - status, _ := strconv.Atoi(item["m_nOrderStatus"]) - tag := item["m_strRemark"] - orderTime, _ := strconv.ParseInt(item["m_nOrderTime"], 10, 64) - if orderTime > 1e11 { - orderTime /= 1000 - } - if orderTime <= 0 { - date := item["m_strInsertDate"] - clock := strings.ReplaceAll(item["m_strInsertTime"], ":", "") - if date != "" { - if len(clock) < 6 { - clock = strings.Repeat("0", 6-len(clock)) + clock - } - if t, err := time.ParseInLocation("20060102150405", date+clock, time.Local); err == nil { - orderTime = t.Unix() - } - } - } - side := sideSell - if operation == opBuyStock || operation == opBuyAlt { - side = sideBuy - } - left, _ := strconv.Atoi(item["m_nVolumeTotal"]) - traded, _ := strconv.Atoi(item["m_nVolumeTraded"]) - orig, _ := strconv.Atoi(item["m_nVolumeTotalOriginal"]) - _, active := activeStatuses[status] - return parsedOrder{ - OrderID: item["m_strOrderSysID"], - StockCode: item["m_strInstrumentID"], - Side: side, - Active: active, - OrderTime: orderTime, - RemarkOwned: strings.HasPrefix(tag, "zt-"), - VolumeOrig: orig, - VolumeLeft: left, - VolumeTraded: traded, - Tag: tag, - } -} - -func truthy(v any) bool { - if v == nil { - return false - } - switch x := v.(type) { - case bool: - return x - case string: - s := strings.ToLower(strings.TrimSpace(x)) - return s == "true" || s == "1" || s == "yes" - case float64: - return x != 0 - case int: - return x != 0 - default: - s := strings.ToLower(strings.TrimSpace(fmt.Sprint(v))) - return s == "true" || s == "1" - } -} - -func newOrderTag(leg string) string { - legCode := map[string]string{"base": "b", "add": "a", "take_profit": "t", "all": "s"}[leg] - if legCode == "" { - legCode = "x" - } - var buf [6]byte - _, _ = rand.Read(buf[:]) - // 订单号同时用于 Windows 回写文件名，因此只使用文件名安全字符。 - tag := fmt.Sprintf("zt-%s-%s", legCode, hex.EncodeToString(buf[:])) +func NewOrderID(leg string) string { + var random [6]byte + _, _ = rand.Read(random[:]) + tag := fmt.Sprintf("zt-%s-%s", leg, hex.EncodeToString(random[:])) if len(tag) > 24 { return tag[:24] } return tag } -func parseHM(now time.Time) int { - n, _ := strconv.Atoi(now.Format("1504")) - return n +func (o *Books) IsLock(side, code string) bool { + o.mu.Lock() + defer o.mu.Unlock() + + keyStr := fmt.Sprintf("%s-%s", side, code) + return slices.Contains(o.Index, keyStr) +} + +func (o *Books) Refresh(client *sdk.Client) error { + o.mu.Lock() + defer o.mu.Unlock() + raw, err := client.TradeDetailData(context.Background(), "order") + if err != nil { + return err + } + var idx []string + orders := make(map[string]*OrderItem) + for _, row := range raw { + keyStr, item := parseOrder(row) + orders[keyStr] = item + idx = append(idx, keyStr) + } + o.Data = orders + o.Index = idx + return nil +} + +func (o *Books) CancelExpired(client *sdk.Client) error { + ctx := context.Background() + err := o.Refresh(client) + if err != nil { + return fmt.Errorf("[委托] 查询失败: %v", err) + } + for _, order := range o.Data { + if order.CreatedAt.IsZero() || time.Since(order.CreatedAt) <= OrderTimeout { + continue + } + if order.ID != "" { + rs, err := client.CanCancelOrder(ctx, order.ID) + if err != nil { + logf("ERROR", "[委托] 撤销失败：%v", err) + continue + } else { + logf("INFO", "[委托] 撤销成功：%v", rs) + } + } + } + return nil +} + +func (o *Books) Place(client *sdk.Client, op int, code string, volume int, sn string) bool { + if _, err := client.PassorderLatestTagged(context.Background(), op, code, volume, sn); err != nil { + logf("ERROR", "[委托] %s 下单失败: %v", code, err) + return false + } + + o.mu.Lock() + defer o.mu.Unlock() + + keyStr := fmt.Sprintf("%s-%s", OffsetFlag[strconv.Itoa(op)], code) + o.Index = append(o.Index, keyStr) + + logf("INFO", "[委托] 下单已提交 %d %s %d股", op, code, volume) + return true +} + +func parseOrder(row map[string]string) (string, *OrderItem) { + left, _ := strconv.Atoi(row["m_nVolumeTotal"]) + traded, _ := strconv.Atoi(row["m_nVolumeTraded"]) + volume := left + traded + + item := &OrderItem{ + ID: row["m_strOrderSysID"], + Code: row["m_strInstrumentID"], + Side: OffsetFlag[row["m_nOffsetFlag"]], + Remark: row["m_strRemark"], + Status: row["m_nOrderStatus"], + Volume: volume, + CreatedAt: time.Unix(parseTimestamp(row), 0), + } + keyStr := fmt.Sprintf("%s-%s", item.Side, item.Code) + return keyStr, item +} + +func parseTimestamp(row map[string]string) int64 { + ts, _ := strconv.ParseInt(row["m_nOrderTime"], 10, 64) + if ts > 1e11 { + return ts / 1000 + } + if ts > 0 { + return ts + } + date := row["m_strInsertDate"] + clock := strings.ReplaceAll(row["m_strInsertTime"], ":", "") + clock = strings.Repeat("0", max(0, 6-len(clock))) + clock + t, _ := time.ParseInLocation("20060102150405", date+clock, time.Local) + return t.Unix() } diff --git a/go-client/apps/zt/logic/positions.go b/go-client/apps/zt/logic/positions.go index fd3da96..f250953 100644 --- a/go-client/apps/zt/logic/positions.go +++ b/go-client/apps/zt/logic/positions.go @@ -10,218 +10,215 @@ import ( "big-qmt/go-client/sdk" ) -var peakMu sync.Mutex -var peakGrids = map[string]int{} +const ( + legBase = "base" + legAdded = "add" +) -func peakKey(code, leg string) string { return code + "|" + leg } +var ( + peakMu sync.Mutex + peakGrids = make(map[string]int) +) -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][持仓] 持仓或状态不可用，本轮跳过") +func managePositions(_ context.Context, client *sdk.Client, ticks map[string]sdk.Tick, positions []sdk.Position, marketOK bool, budget *float64) { + if QuantState == nil || OrderBook == nil || positions == nil { return } - buys, sells, ok := books.activeSets(ctx, client) - if !ok { + if err := OrderBook.Refresh(client); err != nil { + logf("ERROR", "[持仓] 刷新委托失败: %v", err) return } - before := map[string]struct{}{} - for _, code := range stateCodes(QuantState) { - before[code] = struct{}{} - } - if ticks == nil { - ticks = map[string]sdk.Tick{} - } - type row struct { - volume, usable int - avg, price float64 - stock string - item *StateItem - } - rows := make([]row, 0, len(positions)) - seen := map[string]struct{}{} - for _, pos := range positions { - code := pos.StockCode - if code == "" { - continue - } - seen[code] = struct{}{} - 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}) + + current := make(map[string]sdk.Position, len(positions)) + for _, position := range positions { + if position.StockCode != "" { + current[position.StockCode] = position } } - for _, code := range stateCodes(QuantState) { - if _, ok := seen[code]; !ok { - syncItem(QuantState, code, 0, 0, buys, sells, books) - } + + for _, code := range stateCodes() { + syncPosition(code, current[code]) } - after := map[string]struct{}{} - for _, code := range stateCodes(QuantState) { - after[code] = struct{}{} - } - for code := range before { - if _, ok := after[code]; !ok { - forget(code) - } - } - for _, r := range rows { - 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.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.item.AddedQty > 0 { - addPnL := -999.0 - 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.AddedQty, "add", addPnL) - } - continue - } - basePnL := -999.0 - if r.item.BaseCost > 0 { - basePnL = (r.price - r.item.BaseCost) / r.item.BaseCost * 100 - } - if retreated(r.item, "base", basePnL) { - sellLeg(ctx, client, books, r.item, r.usable, r.item.BaseQty, "base", basePnL) - } else if basePnL <= config.Account.LossTriggerPct { - addOnRebound(ctx, client, books, r.item, r.price, marketOK, buyBudget) - } - } - if err := QuantState.Save(); err != nil { - logf("ERROR", "%v", err) + for code, position := range current { + managePosition(client, ticks[code], position, marketOK, budget) } + + saveState() } -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 +func syncPosition(code string, position sdk.Position) { + item, err := QuantState.Get(code) + if err != nil || orderBusy(code, "BUY") || orderBusy(code, "SELL") { + return } + if position.Volume <= 0 { + QuantState.Delete(code) + forget(code) + return + } + 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 + item.BaseQty = max(0, position.Volume-item.AddedQty) + item.BaseCost = position.OpenPrice + item.BaseStatus = StatusOk } - item, _ = state.Get(code) - return item + if item.AddedStatus == StatusIng { + syncAdded(item, position) + } + QuantState.Set(item) } -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) { +func syncAdded(item *StateItem, position sdk.Position) { + addedQty := position.Volume - item.BaseQty + if addedQty <= 0 { + item.BaseQty = position.Volume + item.BaseCost = position.OpenPrice + item.AddedQty = 0 + item.AddedCost = 0 + item.AddedStatus = StatusNone + clearPeak(item.Code, legAdded) return } - if volume <= 0 { - state.Delete(item.Code) - return - } - 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) + + item.AddedQty = addedQty + totalCost := position.OpenPrice * float64(position.Volume) + baseCost := item.BaseCost * float64(item.BaseQty) + item.AddedCost = math.Max(0, (totalCost-baseCost)/float64(addedQty)) + item.AddedStatus = StatusOk } -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) { +func managePosition(client *sdk.Client, tick sdk.Tick, position sdk.Position, marketOK bool, budget *float64) { + item, err := QuantState.Get(position.StockCode) + if err != nil || tick.LastPrice <= 0 || !positionReady(item, position) { return } - if volume <= 0 { - state.Delete(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.AddedQty, item.AddedCost, item.AddedStatus = 0, 0, StatusNone - peakMu.Lock() - delete(peakGrids, peakKey(item.Code, "add")) - peakMu.Unlock() - } - state.Set(item) -} - -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 - } - volume := libs.CalcBuyVolume(price, config.Account.BuyValue) - estimated := price * float64(volume) - if volume <= 0 || buyBudget == nil || estimated > *buyBudget { - return - } - orderID := newOrderTag("add") - if books.place(ctx, client, sideBuy, item.Code, volume, orderID) { - item.AddedOrderId, item.AddedQty, item.AddedCost, item.AddedStatus = orderID, volume, price, StatusIng - item.AddedNum++ - QuantState.Set(item) - *buyBudget -= estimated - if err := QuantState.Save(); err != nil { - logf("ERROR", "%v", err) + if item.AddedQty > 0 { + pnl := profit(tick.LastPrice, item.AddedCost) + if shouldSell(item.Code, legAdded, pnl) { + sell(client, item, position.CanUseVolume, item.AddedQty, legAdded, pnl) } + return + } + + pnl := profit(tick.LastPrice, item.BaseCost) + if shouldSell(item.Code, legBase, pnl) { + sell(client, item, position.CanUseVolume, item.BaseQty, legBase, pnl) + } else if pnl <= config.Account.LossTriggerPct { + buyAdded(client, item, tick.LastPrice, marketOK, budget) } } -func retreated(item *StateItem, leg string, pnl float64) bool { - if pnl < config.Account.MinProfitPct { +func positionReady(item *StateItem, position sdk.Position) bool { + return item.BaseStatus != StatusIng && item.AddedStatus != StatusIng && + position.Volume > 0 && position.Volume%100 == 0 && + position.Volume == item.BaseQty+item.AddedQty +} + +func buyAdded(client *sdk.Client, item *StateItem, price float64, marketOK bool, budget *float64) { + if !marketOK || budget == nil || PosbuyWatch == nil || !PosbuyWatch.Triggered("补仓", item.Code, price) { + return + } + volume := calcBuyVolume(price, config.Account.BuyValue) + amount := price * float64(volume) + if volume <= 0 || amount > *budget || orderBusy(item.Code, "BUY") { + return + } + + orderID := NewOrderID(legAdded) + if !OrderBook.Place(client, sdk.OpBuy, item.Code, volume, orderID) { + return + } + item.AddedOrderId = orderID + item.AddedNum++ + item.AddedQty = volume + item.AddedCost = price + item.AddedStatus = StatusIng + QuantState.Set(item) + *budget -= amount + saveState() +} + +func sell(client *sdk.Client, item *StateItem, usable, volume int, leg string, pnl float64) { + volume -= volume % 100 + if volume <= 0 || usable < volume || orderBusy(item.Code, "SELL") { + return + } + orderID := NewOrderID(leg) + if !OrderBook.Place(client, sdk.OpSell, item.Code, volume, orderID) { + return + } + if leg == legAdded { + item.AddedOrderId = orderID + item.AddedStatus = StatusIng + } else { + item.BaseOrderId = orderID + item.BaseStatus = StatusIng + } + QuantState.Set(item) + saveState() + logf("INFO", "[止盈] %s 卖出%d股，盈利=%.2f%%", item.Code, volume, pnl) +} + +func orderBusy(code, side string) bool { + OrderBook.mu.Lock() + defer OrderBook.mu.Unlock() + order := OrderBook.Data[side+"-"+code] + if order == nil { + return false + } + switch order.Status { + case "48", "49", "50", "51", "52", "55": + return true + default: + return false + } +} + +func shouldSell(code, leg string, pnl float64) bool { + if pnl < config.Account.MinProfitPct || config.Account.GridStepPct <= 0 { return false } grid := int(math.Floor(pnl / config.Account.GridStepPct)) - key := peakKey(item.Code, leg) + key := peakKey(code, leg) peakMu.Lock() defer peakMu.Unlock() - peak, ok := peakGrids[key] - if !ok || grid > peak { + peak, tracked := peakGrids[key] + if !tracked || grid > peak { peakGrids[key] = grid return false } return grid < peak } -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 { - return - } - orderID := newOrderTag(leg) - if !books.place(ctx, client, sideSell, item.Code, volume, orderID) { - return - } - if leg == "add" { - item.AddedOrderId, item.AddedStatus = orderID, StatusIng - } else { - item.BaseOrderId, item.BaseStatus = orderID, StatusIng - } - QuantState.Set(item) +func stateCodes() []string { + QuantState.mu.Lock() + defer QuantState.mu.Unlock() + return append([]string(nil), QuantState.Codes...) +} + +func saveState() { if err := QuantState.Save(); err != nil { logf("ERROR", "%v", err) } - logf("INFO", "[ZT][止盈] %s 卖出 %d 股，%s腿盈利=%.2f%%", item.Code, volume, leg, pnl) +} + +func profit(price, cost float64) float64 { + if cost <= 0 { + return math.Inf(-1) + } + return (price - cost) / cost * 100 +} + +func calcBuyVolume(price, value float64) int { + return libs.CalcBuyVolume(price, value) +} + +func peakKey(code, leg string) string { return code + "|" + leg } + +func clearPeak(code, leg string) { + peakMu.Lock() + delete(peakGrids, peakKey(code, leg)) + peakMu.Unlock() } func forget(code string) { @@ -235,8 +232,6 @@ func forget(code string) { delete(PosbuyWatch.Data, code) PosbuyWatch.mu.Unlock() } - peakMu.Lock() - delete(peakGrids, peakKey(code, "base")) - delete(peakGrids, peakKey(code, "add")) - peakMu.Unlock() + clearPeak(code, legBase) + clearPeak(code, legAdded) } diff --git a/go-client/apps/zt/logic/strategy_test.go b/go-client/apps/zt/logic/strategy_test.go deleted file mode 100644 index 56223eb..0000000 --- a/go-client/apps/zt/logic/strategy_test.go +++ /dev/null @@ -1,39 +0,0 @@ -package logic - -import ( - "os" - "path/filepath" - "strings" - "testing" -) - -func TestCalcBuyVolume(t *testing.T) { - for _, tt := range []struct { - price, value float64 - want int - }{{10, 5000, 500}, {33, 5000, 100}, {100, 5000, 100}, {0, 5000, 0}} { - if got := calcBuyVolume(tt.price, tt.value); got != tt.want { - t.Fatalf("calcBuyVolume(%v,%v)=%d, want %d", tt.price, tt.value, got, tt.want) - } - } -} - -func TestNewOrderTagIsShortAndFileSafe(t *testing.T) { - tag := newOrderTag("base") - if len(tag) > 24 || !strings.HasPrefix(tag, "zt-") || strings.ContainsAny(tag, `<>:"/\\|?*`) { - t.Fatalf("订单号不符合约束: %q", tag) - } -} - -func TestLoadReceipt(t *testing.T) { - dir := t.TempDir() - path := filepath.Join(dir, "order_zt.json") - raw := []byte(`{"order_id":"zt-b-123","qmt_order_id":"9","stock_code":"000001.SZ","side":"buy","requested_volume":500,"traded_volume":500,"status":"filled","updated_at":"2026-08-25T10:00:00+08:00"}`) - if err := os.WriteFile(path, raw, 0o644); err != nil { - t.Fatal(err) - } - got, err := loadReceipt(path) - if err != nil || got.Status != "filled" || got.TradedVolume != 500 { - t.Fatalf("loadReceipt()=%+v, %v", got, err) - } -} diff --git a/go-client/apps/zt/main.go b/go-client/apps/zt/main.go index b95f234..be6e728 100644 --- a/go-client/apps/zt/main.go +++ b/go-client/apps/zt/main.go @@ -39,6 +39,7 @@ func main() { } // 初始化 + logic.NewOrderBook() logic.InitWatch() if err := logic.InitState(StrategyName); err != nil { log.Panicln("ERROR", err.Error()) @@ -57,13 +58,12 @@ func main() { log.Printf("[INFO] [ZT] 已加载 %d 个开仓信号", len(signals.Data)) // 第四步：工作日每 30 秒触发，交易时段由 RunOnce 统一判断。 - books := logic.NewOrderBook() scheduler := cron.New( cron.WithSeconds(), cron.WithChain(cron.SkipIfStillRunning(cron.DefaultLogger)), ) if _, err := scheduler.AddFunc("0,30 * 9-15 * * 1-5", func() { - logic.RunOnce(ctx, client, books, signals) + logic.RunOnce(ctx, client, signals) }); err != nil { log.Fatalf("[ERROR] 创建计划任务失败: %v", err) } diff --git a/go-client/config/config.go b/go-client/config/config.go index 17a8ee9..304f48d 100644 --- a/go-client/config/config.go +++ b/go-client/config/config.go @@ -25,14 +25,13 @@ 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"` + AccountID string `yaml:"account_id"` + HostKey string `yaml:"host_key"` + 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 映射加载当前计算机的账户配置。 diff --git a/go-client/sdk/trade.go b/go-client/sdk/trade.go index b45bacb..7728adb 100644 --- a/go-client/sdk/trade.go +++ b/go-client/sdk/trade.go @@ -30,18 +30,14 @@ func (c *Client) Passorder(ctx context.Context, req PassorderRequest) (*OrderRef } // PassorderLatest 按最新价下单，不附加策略订单号。 -func (c *Client) PassorderLatest(ctx context.Context, buy bool, stock string, volume int) (*OrderRefResult, error) { - return c.PassorderLatestTagged(ctx, buy, stock, volume, "") +func (c *Client) PassorderLatest(ctx context.Context, side int, stock string, volume int) (*OrderRefResult, error) { + return c.PassorderLatestTagged(ctx, side, stock, volume, "") } // PassorderLatestTagged 使用 strategyName 将本地唯一订单号传给 QMT。 -func (c *Client) PassorderLatestTagged(ctx context.Context, buy bool, stock string, volume int, orderID string) (*OrderRefResult, error) { - op := OpSell - if buy { - op = OpBuy - } +func (c *Client) PassorderLatestTagged(ctx context.Context, side int, stock string, volume int, orderID string) (*OrderRefResult, error) { return c.Passorder(ctx, PassorderRequest{ - OpType: op, + OpType: side, OrderType: OrderTypeVolume, Stock: stock, PrType: PrTypeLatest,