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

dev 1

Browse Source
This commit is contained in:
yanweidong
2026-08-25 16:40:18 +08:00
parent 4514646c6a
commit fa62436a73
28 changed files with 6401 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
go-client/apps/zt/boot.go Normal file
View File

@@ -0,0 +1,70 @@
package main
import (
"context"
"fmt"
"strings"
"time"
"big-qmt/go-client/sdk"
)
func overview(cfg Config, assets *sdk.Assets, positions []sdk.Position) {
fmt.Println("\n" + strings.Repeat("=", 80))
fmt.Printf("【时间】%s\n", time.Now().Format("2006-01-02 15:04:05"))
fmt.Printf("【配置】account_id: %s host_key: %s open_money: %.0f\n", cfg.AccountID, cfg.HostKey, cfg.OpenMoney)
if assets != nil {
fmt.Printf("【资金】总资产：%.2f元，可用资金：%.2f元\n", assets.Total, assets.Available)
} else {
fmt.Println("【资金】查询失败")
}
fmt.Printf("【持仓】%d只\n", len(positions))
fmt.Println(strings.Repeat("=", 80))
for _, p := range positions {
if p.Volume <= 0 {
continue
}
code := normalizeCode(p.StockCode, "")
fmt.Printf("【持仓】%s %s 持仓=%d 可用=%d 冻结=%d 在途=%d 昨仓=%d 成本=%.3f 现价=%.3f 市值=%.2f 浮盈=%.2f 盈亏比例=%.2f%%\n",
code, p.StockName, p.Volume, p.CanUseVolume, p.FrozenVolume, p.OnRoadVolume, p.YesterdayVolume,
p.OpenPrice, p.LastPrice, p.MarketValue, p.FloatProfit, p.ProfitRate*100)
}
}
func runRound(ctx context.Context, client *sdk.Client, books *orderBook, cfg Config, assets *sdk.Assets, ticks map[string]sdk.Tick, positions []sdk.Position) {
signals := fetchSignal(cfg, "dcm_signal")
books.cancelExpired(ctx, client, cfg)
hold := positionCodes(positions)
openSignals := map[string]map[string]any{}
for code, signal := range signals {
norm := normalizeCode(code, "")
if norm == "" {
norm = code
}
if _, held := hold[norm]; held {
continue
}
openSignals[norm] = signal
}
if len(openSignals) > 0 {
if books.refresh(ctx, client, cfg) {
buys, _, ok := books.activeSets(ctx, client, cfg)
if ok {
filtered := map[string]map[string]any{}
for code, signal := range openSignals {
if _, buying := buys[code]; buying {
continue
}
filtered[code] = signal
}
openSignals = filtered
}
}
}
marketOK := marketAllowOpen(cfg)
if len(openSignals) > 0 {
openSignal(ctx, client, books, cfg, assets, ticks, openSignals, marketOK)
}
managePositions(ctx, client, books, cfg, ticks, positions, marketOK)
}

118
go-client/apps/zt/config.go Normal file
View File

@@ -0,0 +1,118 @@
package main
import (
"os"
"strconv"
"strings"
"time"
)
type Config struct {
QMTBaseURL string
QMTToken string
AccountType string
AccountID string
HostKey string
APIHost string
DataDir string
HTTPTimeout time.Duration
OrderTimeout time.Duration
LoopInterval time.Duration
OpenMoney float64
MinCashRatio float64
LossTriggerPct float64
GridStepPct float64
MinProfitPct float64
AdoptExisting bool
ReadyCacheStart int
WatchTimeout time.Duration
ReboundThreshold float64
}
func loadConfig() Config {
cfg := Config{
QMTBaseURL: env("QMT_BASE_URL", "http://127.0.0.1:10086"),
QMTToken: env("QMT_TOKEN", "QMTbyYanweidong"),
AccountType: env("QMT_ACCOUNT", "stock"),
AccountID: env("ACCOUNT_ID", ""),
HostKey: env("HOST_KEY", ""),
APIHost: strings.TrimRight(env("API_HOST", "http://139.224.247.176:13499"), "/"),
DataDir: env("DATA_DIR", "D:/qmt_strategy_state"),
HTTPTimeout: durationEnv("HTTP_TIMEOUT_SEC", 5) * time.Second,
OrderTimeout: durationEnv("ORDER_TIMEOUT_SEC", 60) * time.Second,
LoopInterval: durationEnv("LOOP_INTERVAL_SEC", 30) * time.Second,
OpenMoney: floatEnv("OPEN_MONEY", 5000),
MinCashRatio: floatEnv("MIN_CASH_RATIO", 0.1),
LossTriggerPct: floatEnv("LOSS_TRIGGER_PCT", -30),
GridStepPct: floatEnv("GRID_STEP_PCT", 1),
MinProfitPct: floatEnv("MIN_PROFIT_PCT", 2),
AdoptExisting: boolEnv("ADOPT_EXISTING_POSITIONS", true),
ReadyCacheStart: intEnv("READY_CACHE_START", 925),
WatchTimeout: durationEnv("WATCH_TIMEOUT_SEC", 300) * time.Second,
ReboundThreshold: floatEnv("REBOUND_THRESHOLD", 0.61),
}
if strings.TrimSpace(cfg.AccountID) == "" {
logf("ERROR", "ACCOUNT_ID 为空")
os.Exit(1)
}
if strings.TrimSpace(cfg.HostKey) == "" {
logf("ERROR", "HOST_KEY 为空")
os.Exit(1)
}
if cfg.MinCashRatio < 0 || cfg.MinCashRatio >= 1 {
logf("ERROR", "MIN_CASH_RATIO 必须在 [0, 1)")
os.Exit(1)
}
if cfg.OpenMoney <= 0 {
logf("ERROR", "OPEN_MONEY 必须大于 0")
os.Exit(1)
}
if err := os.MkdirAll(cfg.DataDir, 0o755); err != nil {
logf("ERROR", "创建 DATA_DIR 失败: %v", err)
os.Exit(1)
}
return cfg
}
func env(key, fallback string) string {
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
return v
}
return fallback
}
func intEnv(key string, fallback int) int {
v := strings.TrimSpace(os.Getenv(key))
if v == "" {
return fallback
}
n, err := strconv.Atoi(v)
if err != nil {
return fallback
}
return n
}
func floatEnv(key string, fallback float64) float64 {
v := strings.TrimSpace(os.Getenv(key))
if v == "" {
return fallback
}
f, err := strconv.ParseFloat(v, 64)
if err != nil {
return fallback
}
return f
}
func durationEnv(key string, fallbackSec int) time.Duration {
return time.Duration(intEnv(key, fallbackSec))
}
func boolEnv(key string, fallback bool) bool {
v := strings.ToLower(strings.TrimSpace(os.Getenv(key)))
if v == "" {
return fallback
}
return v == "1" || v == "true" || v == "yes"
}

10
go-client/apps/zt/log.go Normal file
View File

@@ -0,0 +1,10 @@
package main
import (
"fmt"
"log"
)
func logf(level, format string, args ...any) {
log.Printf("[%s] %s", level, fmt.Sprintf(format, args...))
}

105
go-client/apps/zt/main.go Normal file
View File

@@ -0,0 +1,105 @@
package main
import (
"context"
"log"
"os"
"os/signal"
"strings"
"syscall"
"time"
"big-qmt/go-client/sdk"
)
func main() {
log.SetFlags(log.LstdFlags | log.Lmicroseconds)
cfg := loadConfig()
client := sdk.New(cfg.QMTBaseURL, cfg.QMTToken, cfg.AccountType, cfg.HTTPTimeout)
books := newOrderBook()
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
startup := context.Background()
assets, err := client.Assets(startup, cfg.AccountType)
if err != nil {
logf("ERROR", "启动获取资产失败: %v", err)
}
positions, err := client.Positions(startup, cfg.AccountType)
if err != nil {
logf("ERROR", "启动获取持仓失败: %v", err)
positions = []sdk.Position{}
}
overview(cfg, assets, positions)
logf("INFO", "[ZT] host_key=%s interval=%s", cfg.HostKey, cfg.LoopInterval)
logf("INFO", "[ZT] Init Success, waiting trading session")
ticker := time.NewTicker(cfg.LoopInterval)
defer ticker.Stop()
runOnce(ctx, client, books, cfg)
for {
select {
case <-ctx.Done():
logf("INFO", "[ZT] 停止")
return
case <-ticker.C:
runOnce(ctx, client, books, cfg)
}
}
}
func runOnce(ctx context.Context, client *sdk.Client, books *orderBook, cfg Config) {
if !tradingTime(time.Now()) {
return
}
roundCtx, cancel := context.WithTimeout(ctx, cfg.HTTPTimeout*4)
defer cancel()
assets, err := client.Assets(roundCtx, cfg.AccountType)
if err != nil {
logf("ERROR", "获取资产失败: %v", err)
return
}
positions, err := client.Positions(roundCtx, cfg.AccountType)
if err != nil {
logf("ERROR", "获取持仓失败: %v", err)
return
}
codes := passCodes(cfg)
seen := map[string]struct{}{}
stockList := make([]string, 0, len(codes)+len(positions))
addCode := func(code string) {
n := normalizeCode(code, "")
if n == "" {
n = strings.ToUpper(strings.TrimSpace(code))
}
if n == "" {
return
}
if _, ok := seen[n]; ok {
return
}
seen[n] = struct{}{}
stockList = append(stockList, n)
}
for _, code := range codes {
addCode(code)
}
for _, p := range positions {
addCode(p.StockCode)
}
ticks := map[string]sdk.Tick{}
if len(stockList) > 0 {
raw, err := client.FullTick(roundCtx, stockList)
if err != nil {
logf("ERROR", "获取行情失败: %v", err)
return
}
for code, tick := range raw {
ticks[normalizeCode(code, "")] = tick
ticks[code] = tick
}
}
runRound(roundCtx, client, books, cfg, assets, ticks, positions)
}

114
go-client/apps/zt/open.go Normal file
View File

@@ -0,0 +1,114 @@
package main
import (
"context"
"math"
"sync"
"time"
"big-qmt/go-client/sdk"
)
type dipWatch struct {
LastClose float64
ExpiresAt time.Time
}
var openDip = struct {
mu sync.Mutex
store map[string]dipWatch
}{store: map[string]dipWatch{}}
func openSignal(ctx context.Context, client *sdk.Client, books *orderBook, cfg Config, assets *sdk.Assets, ticks map[string]sdk.Tick, openSignals map[string]map[string]any, marketOK bool) {
if !marketOK {
return
}
if assets == nil {
return
}
if assets.Available < assets.Total*cfg.MinCashRatio {
return
}
state := getState(cfg)
if state.LoadError != "" {
logf("ERROR", "[ZT][开仓] 状态文件异常，禁止新开仓: %s", state.LoadError)
return
}
for signalCode, signal := range openSignals {
code := normalizeCode(signalCode, "")
if code == "" {
if c, ok := signal["code"].(string); ok {
code = normalizeCode(c, "")
}
}
if code == "" {
logf("ERROR", "[ZT][开仓] 无效股票代码=%s", signalCode)
continue
}
if state.Get(code) != nil {
continue
}
price := ticks[code].LastPrice
if price <= 0 {
continue
}
if !dipTriggered(&openDip.mu, openDip.store, cfg, "开仓", code, price) {
continue
}
volume := calcOpenVolume(price, cfg.OpenMoney)
if volume <= 0 {
continue
}
if !books.place(ctx, client, cfg, "buy", code, volume, newOrderTag("base")) {
continue
}
state.Ensure(code).Pending = "base_opening"
state.Save()
logf("INFO", "[ZT][开仓] %s 买入 %d 股", code, volume)
}
state.Save()
}
func calcOpenVolume(price, openMoney float64) int {
if price <= 0 || openMoney <= 0 {
return 0
}
hands := int(math.Floor(openMoney / (price * 100)))
if hands == 0 {
hands = 1
}
return hands * 100
}
func dipTriggered(mu *sync.Mutex, store map[string]dipWatch, cfg Config, 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(cfg.WatchTimeout)}
logf("INFO", "[%s-观察] %s 现价=%.2f", tag, code, price)
return false
}
if price < watch.LastClose {
watch.LastClose = price
watch.ExpiresAt = now.Add(cfg.WatchTimeout)
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 < cfg.ReboundThreshold {
logf("INFO", "[%s-等待] %s 反弹=%.2f%% 阈值=%.2f%%", tag, code, rebound, cfg.ReboundThreshold)
return false
}
delete(store, code)
logf("INFO", "[%s-触发] %s 反弹=%.2f%% 低点=%.2f", tag, code, rebound, watch.LastClose)
return true
}

374
go-client/apps/zt/order.go Normal file
View File

@@ -0,0 +1,374 @@
package main
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"strconv"
"strings"
"sync"
"time"
"big-qmt/go-client/sdk"
)
const (
opBuyStock = 23
opBuyAlt = 48
)
var activeStatuses = map[int]struct{}{
48: {}, 49: {}, 50: {}, 51: {}, 52: {}, 55: {},
}
type parsedOrder struct {
OrderID string
StockCode string
Side string
Active bool
OrderTime int64
RemarkOwned bool
VolumeOrig int
VolumeLeft int
VolumeTraded int
Tag string
}
func (o parsedOrder) cancelVolume() int {
n := o.VolumeLeft + o.VolumeTraded
if n > 0 {
return n
}
return o.VolumeOrig
}
type submission struct {
Code string
Side string
Volume int
At time.Time
Tag string
}
type orderBook struct {
mu sync.Mutex
cached []parsedOrder
hasCache bool
buyLocks map[string]time.Time
sellLocks map[string]time.Time
subs []submission
}
func newOrderBook() *orderBook {
return &orderBook{
buyLocks: map[string]time.Time{},
sellLocks: map[string]time.Time{},
}
}
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, cfg Config) ([]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, cfg.AccountType, "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) refresh(ctx context.Context, client *sdk.Client, cfg Config) bool {
o.invalidate()
_, err := o.query(ctx, client, cfg)
return err == nil
}
func (o *orderBook) activeSets(ctx context.Context, client *sdk.Client, cfg Config) (buys, sells map[string]struct{}, ok bool) {
orders, err := o.query(ctx, client, cfg)
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 == "buy" {
buys[item.StockCode] = struct{}{}
} else {
sells[item.StockCode] = struct{}{}
}
}
return buys, sells, true
}
func (o *orderBook) cancelExpired(ctx context.Context, client *sdk.Client, cfg Config) bool {
o.invalidate()
orders, err := o.query(ctx, client, cfg)
if err != nil {
return false
}
state := getState(cfg)
now := time.Now()
timeout := cfg.OrderTimeout
seen := map[string]struct{}{}
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, cfg.AccountType)
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, cfg.AccountType)
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)
logf("INFO", "[ZT][委托] 撤销超时单 %s %s %s volume=%d", order.OrderID, order.StockCode, order.Side, vol)
}
return true
}
func (o *orderBook) claimed(state *ZTState, 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 := state.Get(order.StockCode)
if item == nil || item.Pending == "" {
return false
}
switch item.Pending {
case "base_opening", "add":
return order.Side == "buy"
case "sell_add", "sell_base":
return order.Side == "sell"
default:
return false
}
}
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(cfg Config, code, side string, active map[string]struct{}) bool {
if _, ok := active[code]; ok {
return true
}
return o.locked(cfg, code, side)
}
func (o *orderBook) locked(cfg Config, code, side string) bool {
o.mu.Lock()
defer o.mu.Unlock()
ts, ok := o.locks(side)[code]
return ok && time.Since(ts) < cfg.OrderTimeout
}
func (o *orderBook) locks(side string) map[string]time.Time {
if side == "buy" {
return o.buyLocks
}
return o.sellLocks
}
func (o *orderBook) hasActive(ctx context.Context, client *sdk.Client, cfg Config, code, side string) bool {
orders, err := o.query(ctx, client, cfg)
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, cfg Config, 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(cfg, code, side) {
logf("INFO", "[ZT][委托] %s %s锁定中", code, side)
return false
}
if o.hasActive(ctx, client, cfg, code, side) {
logf("INFO", "[ZT][委托] %s 已有%s在途委托", code, side)
return false
}
_, err := client.PassorderLatest(ctx, side == "buy", code, volume)
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, Volume: volume, At: time.Now(), Tag: tag})
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 := asIntS(mapGet(item, "m_nOffsetFlag", "m_nOrderType", "order_type"))
status := asIntS(mapGet(item, "m_nOrderStatus", "order_status", "status"))
tag := mapGet(item, "m_strRemark", "m_strUserOrderId", "order_remark")
orderTime := int64(asIntS(mapGet(item, "m_nOrderTime", "order_time")))
if orderTime > 1e11 {
orderTime /= 1000
}
if orderTime <= 0 {
date := mapGet(item, "m_strInsertDate")
clock := strings.ReplaceAll(mapGet(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 := "sell"
if operation == opBuyStock || operation == opBuyAlt {
side = "buy"
}
left := asIntS(mapGet(item, "m_nVolumeTotal", "volume_left"))
traded := asIntS(mapGet(item, "m_nVolumeTraded", "volume_traded"))
orig := asIntS(mapGet(item, "m_nVolumeTotalOriginal", "volume"))
_, active := activeStatuses[status]
return parsedOrder{
OrderID: mapGet(item, "m_strOrderSysID", "m_nOrderID", "order_id"),
StockCode: stockCodeFromMap(item),
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[:])
tag := fmt.Sprintf("zt:%s:%s", legCode, hex.EncodeToString(buf[:]))
if len(tag) > 24 {
return tag[:24]
}
return tag
}
func parseHM(now time.Time) int {
n, _ := strconv.Atoi(now.Format("1504"))
return n
}
func tradingTime(now time.Time) bool {
hm := parseHM(now)
return (hm >= 930 && hm <= 1130) || (hm >= 1300 && hm <= 1500)
}

298
go-client/apps/zt/positions.go Normal file
View File

@@ -0,0 +1,298 @@
package main
import (
"context"
"math"
"sync"
"big-qmt/go-client/sdk"
)
var posDip = struct {
mu sync.Mutex
store map[string]dipWatch
}{store: map[string]dipWatch{}}
var peakMu sync.Mutex
var peakGrids = map[string]int{}
func peakKey(code, leg string) string { return code + "|" + leg }
func positionCodes(positions []sdk.Position) map[string]struct{} {
out := map[string]struct{}{}
for _, p := range positions {
if p.Volume <= 0 {
continue
}
code := normalizeCode(p.StockCode, "")
if code != "" {
out[code] = struct{}{}
}
}
return out
}
func managePositions(ctx context.Context, client *sdk.Client, books *orderBook, cfg Config, ticks map[string]sdk.Tick, positions []sdk.Position, marketOK bool) {
if positions == nil {
logf("ERROR", "[ZT][持仓] 持仓查询失败，本轮跳过")
return
}
state := getState(cfg)
if !books.cancelExpired(ctx, client, cfg) {
logf("ERROR", "[ZT][持仓] 委托查询失败，本轮跳过")
return
}
buys, sells, ok := books.activeSets(ctx, client, cfg)
if !ok {
return
}
before := map[string]struct{}{}
for _, code := range state.Codes() {
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
}
rows := make([]row, 0, len(positions))
seen := map[string]struct{}{}
for _, pos := range positions {
code := normalizeCode(pos.StockCode, "")
if code == "" {
continue
}
seen[code] = struct{}{}
item := syncItem(cfg, state, code, pos.Volume, pos.OpenPrice, buys, sells, books)
if pos.Volume <= 0 {
continue
}
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() {
if _, ok := seen[code]; !ok {
syncItem(cfg, state, code, 0, 0, buys, sells, books)
}
}
after := map[string]struct{}{}
for _, code := range state.Codes() {
after[code] = struct{}{}
}
for code := range before {
if _, ok := after[code]; !ok {
forget(code)
}
}
for _, r := range rows {
if r.item == nil || r.item.Pending != "" {
continue
}
if r.avg <= 0 || r.price <= 0 || r.volume%100 != 0 {
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 {
addPnL := -999.0
if r.item.AddCost > 0 {
addPnL = (r.price - r.item.AddCost) / r.item.AddCost * 100
}
if retreated(cfg, r.item, "add", addPnL) {
sellLeg(ctx, client, books, cfg, r.item, r.usable, r.item.AddQty, "add", addPnL)
}
continue
}
basePnL := -999.0
if r.item.BaseCost > 0 {
basePnL = (r.price - r.item.BaseCost) / r.item.BaseCost * 100
}
if retreated(cfg, r.item, "base", basePnL) {
sellLeg(ctx, client, books, cfg, r.item, r.usable, r.item.BaseQty, "base", basePnL)
} else if r.item.AddQty <= 0 && r.item.AddCost <= 0 && basePnL <= cfg.LossTriggerPct {
addOnRebound(ctx, client, books, cfg, r.item, r.price, marketOK)
}
}
state.Save()
}
func syncItem(cfg Config, state *ZTState, code string, volume int, avgPrice float64, buys, sells map[string]struct{}, books *orderBook) *SymbolState {
item := state.Get(code)
if item == nil {
if volume > 0 {
if cfg.AdoptExisting && avgPrice > 0 {
item = state.Ensure(code)
item.BaseQty, item.BaseCost, item.Pending = volume, avgPrice, ""
logf("WARNING", "[ZT][持仓] %s 接管为底仓", code)
return item
}
logf("ERROR", "[ZT][持仓] %s 无本地状态，跳过", code)
}
return nil
}
switch item.Pending {
case "base_opening":
syncOpen(cfg, state, item, volume, avgPrice, buys, books)
case "add":
syncAdd(cfg, state, item, volume, avgPrice, buys, books)
case "sell_add":
syncSellAdd(cfg, state, item, volume, avgPrice, sells, books)
case "sell_base":
syncSellBase(cfg, state, item, volume, avgPrice, sells, books)
default:
if volume <= 0 {
state.Remove(code)
logf("INFO", "[ZT][持仓] %s 已无持仓，清除状态", code)
return nil
}
}
return state.Get(code)
}
func syncOpen(cfg Config, 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(cfg, item.Code, "buy", buys) {
return
}
if volume <= 0 {
state.Remove(item.Code)
logf("INFO", "[ZT][委托] %s 开仓委托已失效，允许重新开仓", item.Code)
return
}
item.Pending = ""
logf("INFO", "[ZT][持仓] %s 开仓确认 数量=%d 成本=%.2f", item.Code, item.BaseQty, item.BaseCost)
}
func syncAdd(cfg Config, state *ZTState, item *SymbolState, volume int, avgPrice float64, buys map[string]struct{}, books *orderBook) {
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(cfg, item.Code, "buy", 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)
}
item.Pending = ""
}
func syncSellAdd(cfg Config, state *ZTState, item *SymbolState, volume int, avgPrice float64, sells map[string]struct{}, books *orderBook) {
if volume <= 0 {
if !books.sideBusy(cfg, item.Code, "sell", sells) {
state.Remove(item.Code)
logf("INFO", "[ZT][委托] %s 卖出后已无持仓，清除状态", item.Code)
}
return
}
if volume <= item.BaseQty {
item.BaseQty, item.BaseCost = volume, avgPrice
item.AddQty = 0
peakMu.Lock()
delete(peakGrids, peakKey(item.Code, "add"))
peakMu.Unlock()
} else {
item.AddQty = volume - item.BaseQty
}
if !books.sideBusy(cfg, item.Code, "sell", sells) {
item.Pending = ""
}
}
func syncSellBase(cfg Config, state *ZTState, item *SymbolState, volume int, avgPrice float64, sells map[string]struct{}, books *orderBook) {
if volume <= 0 {
if !books.sideBusy(cfg, item.Code, "sell", sells) {
state.Remove(item.Code)
logf("INFO", "[ZT][委托] %s 卖出后已无持仓，清除状态", item.Code)
}
return
}
item.BaseQty, item.BaseCost = volume, avgPrice
if !books.sideBusy(cfg, item.Code, "sell", sells) {
item.Pending = ""
}
}
func addOnRebound(ctx context.Context, client *sdk.Client, books *orderBook, cfg Config, item *SymbolState, price float64, marketOK bool) {
if !marketOK || !dipTriggered(&posDip.mu, posDip.store, cfg, "补仓", item.Code, price) {
return
}
if books.place(ctx, client, cfg, "buy", item.Code, item.BaseQty, newOrderTag("add")) {
item.AddCost = price
item.Pending = "add"
getState(cfg).Save()
logf("INFO", "[ZT][补仓] %s 买入 %d 股", item.Code, item.BaseQty)
}
}
func retreated(cfg Config, item *SymbolState, leg string, pnl float64) bool {
if pnl < cfg.MinProfitPct {
return false
}
grid := int(math.Floor(pnl / cfg.GridStepPct))
key := peakKey(item.Code, leg)
peakMu.Lock()
defer peakMu.Unlock()
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, cfg Config, item *SymbolState, 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
}
if !books.place(ctx, client, cfg, "sell", item.Code, volume, newOrderTag(leg)) {
return
}
if leg == "add" {
item.Pending = "sell_add"
} else {
item.Pending = "sell_base"
}
getState(cfg).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()
peakMu.Lock()
delete(peakGrids, peakKey(code, "base"))
delete(peakGrids, peakKey(code, "add"))
peakMu.Unlock()
}

278
go-client/apps/zt/remote.go Normal file
View File

@@ -0,0 +1,278 @@
package main
import (
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
type dailyCache struct {
Date string `json:"date"`
FetchedAt string `json:"fetched_at"`
OK bool `json:"ok"`
Data any `json:"data"`
}
var memCache sync.Map
func getJSON(rawURL string, params url.Values, timeout time.Duration) (map[string]any, error) {
if params != nil {
if strings.Contains(rawURL, "?") {
rawURL += "&" + params.Encode()
} else {
rawURL += "?" + params.Encode()
}
}
req, err := http.NewRequest(http.MethodGet, rawURL, nil)
if err != nil {
return nil, err
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "big-qmt-go-zt/1")
client := &http.Client{Timeout: timeout}
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
if resp.StatusCode >= 400 {
return nil, fmt.Errorf("http %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
}
out := map[string]any{}
if err := json.Unmarshal(body, &out); err != nil {
return nil, err
}
return out, nil
}
func daily(cfg Config, name, filename string, loader func() (any, error), now time.Time) *dailyCache {
if int(parseHM(now)) < cfg.ReadyCacheStart {
return nil
}
day := now.Format("20060102")
path := filepath.Join(cfg.DataDir, fmt.Sprintf(filename, day))
if v, ok := memCache.Load(path); ok {
if c, ok := v.(*dailyCache); ok && c.Date == day {
return c
}
}
cached := loadDailyFile(path)
if cached != nil && cached.Date == day {
memCache.Store(path, cached)
return cached
}
data, err := loader()
ok := err == nil
if err != nil {
logf("ERROR", "%s 当日请求失败: %v", name, err)
data = map[string]any{}
}
cached = &dailyCache{
Date: day,
FetchedAt: now.Format("2006-01-02 15:04:05"),
OK: ok,
Data: data,
}
raw, _ := json.MarshalIndent(map[string]any{"version": 1, "data": map[string]any{
"date": cached.Date, "fetched_at": cached.FetchedAt, "ok": cached.OK, "data": cached.Data,
}}, "", " ")
if err := os.WriteFile(path+".tmp", raw, 0o644); err == nil {
_ = os.Rename(path+".tmp", path)
}
memCache.Store(path, cached)
return cached
}
func loadDailyFile(path string) *dailyCache {
raw, err := os.ReadFile(path)
if err != nil {
return nil
}
var payload struct {
Version int `json:"version"`
Data map[string]any `json:"data"`
}
if json.Unmarshal(raw, &payload) != nil || payload.Version != 1 || payload.Data == nil {
return nil
}
c := &dailyCache{}
b, _ := json.Marshal(payload.Data)
if json.Unmarshal(b, c) != nil {
return nil
}
return c
}
func fetchSignal(cfg Config, name string) map[string]map[string]any {
cached := daily(cfg, name, "open_%s.json", func() (any, error) {
q := url.Values{"host_key": {cfg.HostKey}}
payload, err := getJSON(cfg.APIHost+"/a/"+name, q, cfg.HTTPTimeout)
if err != nil {
return nil, err
}
return normalizeZT(payload), nil
}, time.Now())
if cached == nil || !cached.OK {
return map[string]map[string]any{}
}
return asSignalMap(cached.Data)
}
func asSignalMap(data any) map[string]map[string]any {
out := map[string]map[string]any{}
switch v := data.(type) {
case map[string]map[string]any:
return v
case map[string]any:
for code, val := range v {
if m, ok := val.(map[string]any); ok {
out[code] = m
} else {
out[code] = map[string]any{"code": code}
}
}
}
return out
}
func normalizeZT(payload map[string]any) map[string]map[string]any {
data, _ := payload["data"]
out := map[string]map[string]any{}
switch v := data.(type) {
case []any:
for _, item := range v {
m, ok := item.(map[string]any)
if !ok {
continue
}
code, _ := m["code"].(string)
if code != "" {
out[code] = m
}
}
case map[string]any:
if code, _ := v["code"].(string); code != "" {
out[code] = v
return out
}
for code, val := range v {
if m, ok := val.(map[string]any); ok {
if _, has := m["code"]; !has {
m["code"] = code
}
out[code] = m
} else {
out[code] = map[string]any{"code": code}
}
}
}
return out
}
func passCodes(cfg Config) []string {
load := func() (any, error) {
payload, err := getJSON(cfg.APIHost+"/a/pass_codes", nil, cfg.HTTPTimeout)
if err != nil {
return nil, err
}
data, _ := payload["data"].([]any)
if data == nil {
return nil, fmt.Errorf("接口 data 不是数组")
}
codes := make([]string, 0, len(data))
for _, item := range data {
s := strings.ToUpper(strings.TrimSpace(fmt.Sprint(item)))
if s != "" && s != "<nil>" {
codes = append(codes, s)
}
}
return codes, nil
}
cached := daily(cfg, "pass_codes", "pass_codes_%s.json", load, time.Now())
codes := codesFromAny(cached)
if len(codes) > 0 {
return codes
}
logf("INFO", "pass_codes 为空，重新获取")
data, err := load()
if err != nil {
logf("ERROR", "pass_codes 重新获取失败: %v", err)
return nil
}
list, _ := data.([]string)
return list
}
func codesFromAny(cached *dailyCache) []string {
if cached == nil || !cached.OK {
return nil
}
switch v := cached.Data.(type) {
case []string:
return v
case []any:
out := make([]string, 0, len(v))
for _, item := range v {
s := strings.ToUpper(strings.TrimSpace(fmt.Sprint(item)))
if s != "" && s != "<nil>" {
out = append(out, s)
}
}
return out
}
return nil
}
func marketAllowOpen(cfg Config) bool {
payload, err := getJSON(cfg.APIHost+"/a/market", url.Values{"period": {"60m"}}, cfg.HTTPTimeout)
if err != nil {
logf("ERROR", "获取60m大盘信号失败: %s %v", cfg.APIHost+"/a/market", err)
return false
}
status := marketStatus(payload)
logf("INFO", "大盘信号: status=%s", status)
return status == "UP"
}
func marketStatus(payload map[string]any) string {
var value any = payload
if m, ok := value.(map[string]any); ok {
if d, exists := m["data"]; exists {
value = d
}
}
if arr, ok := value.([]any); ok {
if len(arr) == 0 {
value = nil
} else {
value = arr[len(arr)-1]
}
}
if m, ok := value.(map[string]any); ok {
if v, exists := m["action"]; exists {
value = v
} else if v, exists := m["status"]; exists {
value = v
} else if v, exists := m["signal"]; exists {
value = v
}
}
s := strings.ToUpper(strings.TrimSpace(fmt.Sprint(value)))
switch s {
case "UP", "DOWN", "NEUTRAL":
return s
default:
return "UNKNOWN"
}
}

179
go-client/apps/zt/state.go Normal file
View File

@@ -0,0 +1,179 @@
package main
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"sync"
)
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"`
}
type filePayload struct {
Version int `json:"version"`
Data map[string]any `json:"data"`
}
type ZTState struct {
path string
Items map[string]*SymbolState
LoadError string
mu sync.Mutex
}
var (
statesMu sync.Mutex
states = map[string]*ZTState{}
)
func getState(cfg Config) *ZTState {
statesMu.Lock()
defer statesMu.Unlock()
if s, ok := states[cfg.AccountID]; ok {
return s
}
s := loadZTState(cfg.DataDir, cfg.AccountID)
states[cfg.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) {
return st
}
st.rebuild(err)
return st
}
var payload filePayload
if err := json.Unmarshal(raw, &payload); err != nil || payload.Version != 1 {
st.rebuild(fmt.Errorf("状态文件版本无效"))
return st
}
data := payload.Data
if data == nil {
st.rebuild(fmt.Errorf("状态文件内容无效"))
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
}
}
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) rebuild(err error) {
if err := os.Remove(s.path); err != nil && !os.IsNotExist(err) {
s.LoadError = err.Error()
logf("ERROR", "[ZT][状态] 状态文件重建失败: %s", s.LoadError)
return
}
s.Items = map[string]*SymbolState{}
if saveErr := s.saveUnlocked(); saveErr != nil {
s.LoadError = fmt.Sprintf("%v；重建失败: %v", err, saveErr)
logf("ERROR", "[ZT][状态] 状态文件重建失败: %s", s.LoadError)
return
}
logf("WARNING", "[ZT][状态] 状态文件损坏，已删除并重建: %v", err)
}
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)
}
func (s *ZTState) Codes() []string {
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 {
logf("ERROR", "[ZT][状态] 保存失败: %v", err)
}
}
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
}
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)
}
func replaceFile(tmp, dest string) error {
if err := os.Rename(tmp, dest); err == nil {
return nil
}
_ = os.Remove(dest)
return os.Rename(tmp, dest)
}

104
go-client/apps/zt/stock.go Normal file
View File

@@ -0,0 +1,104 @@
package main
import (
"fmt"
"strings"
"unicode"
)
var exchangeAlias = map[string]string{
"SSE": "SH", "SHSE": "SH", "XSHG": "SH",
"SZSE": "SZ", "XSHE": "SZ",
"BSE": "BJ", "BJSE": "BJ",
}
func stockCodeFromMap(item map[string]string) string {
code := strings.ToUpper(strings.TrimSpace(mapGet(item, "m_strInstrumentID", "StockCode", "stock_code", "code")))
ex := mapGet(item, "m_strExchangeID", "exchange", "exchange_id")
return normalizeCode(code, ex)
}
func normalizeCode(code, exchange string) string {
code = strings.ToUpper(strings.TrimSpace(code))
if code == "" {
return ""
}
if i := strings.LastIndex(code, "."); i >= 0 {
symbol, ex := code[:i], code[i+1:]
ex = canonExchange(ex)
if ex == "SH" || ex == "SZ" || ex == "BJ" {
return symbol + "." + ex
}
return ""
}
ex := canonExchange(exchange)
if ex == "" && looksDigits(code, 6) {
switch {
case strings.HasPrefix(code, "92") || code[0] == '4' || code[0] == '8':
ex = "BJ"
case code[0] == '5' || code[0] == '6' || code[0] == '9' || strings.HasPrefix(code, "11"):
ex = "SH"
case code[0] == '0' || code[0] == '1' || code[0] == '2' || code[0] == '3':
ex = "SZ"
}
}
if ex == "SH" || ex == "SZ" || ex == "BJ" {
return code + "." + ex
}
return ""
}
func canonExchange(ex string) string {
ex = strings.ToUpper(strings.TrimSpace(ex))
if v, ok := exchangeAlias[ex]; ok {
return v
}
return ex
}
func looksDigits(s string, n int) bool {
if len(s) != n {
return false
}
for _, r := range s {
if !unicode.IsDigit(r) {
return false
}
}
return true
}
func mapGet(item map[string]string, names ...string) string {
for _, name := range names {
if v := strings.TrimSpace(item[name]); v != "" {
return v
}
}
return ""
}
func asIntS(s string) int {
s = strings.TrimSpace(s)
if s == "" {
return 0
}
var n int
_, _ = fmt.Sscanf(s, "%d", &n)
if n == 0 {
var f float64
if _, err := fmt.Sscanf(s, "%f", &f); err == nil {
return int(f)
}
}
return n
}
func asFloatS(s string) float64 {
s = strings.TrimSpace(s)
if s == "" {
return 0
}
var f float64
_, _ = fmt.Sscanf(s, "%f", &f)
return f
}