update

cmd/haobase/clearing/clearing.go

@@ -8,22 +8,38 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"github.com/yzimhao/trading_engine/cmd/haobase/base"
+	"github.com/yzimhao/trading_engine/cmd/haobase/base/symbols"
 	"github.com/yzimhao/trading_engine/trading_core"
 	"github.com/yzimhao/trading_engine/types"
 	"github.com/yzimhao/trading_engine/utils"
 )
 
-func RunClearing(symbol string) {
-	watch_redis_list(symbol)
+func Run() {
+	//load symbols
+	db := base.DB().NewSession()
+	defer db.Close()
+
+	var rows []symbols.TradingVarieties
+	db.Table(new(symbols.TradingVarieties)).Find(&rows)
+
+	for _, row := range rows {
+		run_clearing(row.Symbol)
+	}
+}
+
+func run_clearing(symbol string) {
+	go watch_redis_list(symbol)
 }
 
 func watch_redis_list(symbol string) {
 	key := types.FormatTradeResult.Format(symbol)
+	quote_key := types.FormatQuoteTradeResult.Format(symbol)
 	logrus.Infof("结算，正在监听%s成交日志...", symbol)
 	for {
 		func() {
 			cx := context.Background()
 			rdc := base.RDC()
+			defer rdc.Close()
 
 			if n, _ := rdc.LLen(cx, key).Result(); n == 0 {
 				time.Sleep(time.Duration(50) * time.Millisecond)
@@ -39,7 +55,17 @@ func watch_redis_list(symbol string) {
 				return
 			}
 
-			newClean(data)
+			logrus.Infof("%s成交记录 ask: %s bid: %s price: %s vol: %s", data.Symbol, data.AskOrderId, data.BidOrderId, data.TradePrice.String(), data.TradeQuantity.String())
+
+			err = newClean(data)
+			if err != nil {
+				logrus.Warnf("结算错误: %s %s", raw, err.Error())
+				return
+			}
+
+			//通知kline系统
+			rdc.RPush(cx, quote_key, raw)
+
 			// if !data.Last {
 			// 	go newClean(data)
 			// } else {

cmd/haobase/clearing/clearing_flow.go

@@ -22,7 +22,7 @@ type clean struct {
 	err               error
 }
 
-func newClean(raw trading_core.TradeResult) {
+func newClean(raw trading_core.TradeResult) error {
 	db := base.DB().NewSession()
 	defer db.Close()
 
@@ -34,10 +34,10 @@ func newClean(raw trading_core.TradeResult) {
 		tlog:              raw,
 	}
 
-	item.flow()
+	return item.flow()
 }
 
-func (c *clean) flow() {
+func (c *clean) flow() error {
 
 	c.db.Begin()
 	defer func() {
@@ -54,6 +54,7 @@ func (c *clean) flow() {
 	c.err = c.update_order(trading_core.OrderSideBuy)
 
 	c.err = c.transfer()
+	return c.err
 }
 
 func (c *clean) check_order() error {

cmd/haobase/main.go

@@ -10,6 +10,7 @@ import (
 	"github.com/yzimhao/trading_engine/cmd/haobase/assets"
 	"github.com/yzimhao/trading_engine/cmd/haobase/base"
 	"github.com/yzimhao/trading_engine/cmd/haobase/base/symbols"
+	"github.com/yzimhao/trading_engine/cmd/haobase/clearing"
 	"github.com/yzimhao/trading_engine/cmd/haobase/www"
 	"github.com/yzimhao/trading_engine/utils/app"
 )
@@ -76,6 +77,8 @@ func main() {
 
 			}
 
+			initDemoBaseData()
+			clearing.Run()
 			www.Run()
 			return nil
 		},
@@ -85,3 +88,9 @@ func main() {
 		logrus.Fatal(err)
 	}
 }
+
+func initDemoBaseData() {
+	if app.RunMode == app.ModeDemo {
+		symbols.DemoData()
+	}
+}

cmd/haobase/www/route.go

@@ -8,8 +8,6 @@ import (
 )
 
 func Run() {
-	demoBaseData()
-
 	g := gin.New()
 	router(g)
 	g.Run(viper.GetString("haobase.http.host"))

example/config.toml

@@ -5,6 +5,6 @@ mode="test"
 host="0.0.0.0:8080"
 
 [api]
-haobase_host = "//0.0.0.0:8082"
-haoquote_host = "//0.0.0.0:8081"
-haoquote_ws_host = "ws://0.0.0.0:8081"
+haobase_host = "//127.0.0.1:8082"
+haoquote_host = "//127.0.0.1:8081"
+haoquote_ws_host = "ws://127.0.0.1:8081"

haotrader/tengine.go

@@ -9,7 +9,6 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/shopspring/decimal"
 	"github.com/sirupsen/logrus"
-	"github.com/spf13/viper"
 	"github.com/yzimhao/trading_engine/trading_core"
 	"github.com/yzimhao/trading_engine/types"
 )
@@ -267,15 +266,7 @@ func (t *tengine) monitor_result() {
 		select {
 		case data := <-t.tp.ChTradeResult:
 			go func() {
-				relog := map[string]any{
-					"trade_price":    t.tp.Price2String(data.TradePrice),
-					"trade_quantity": t.tp.Qty2String(data.TradeQuantity),
-					"trade_time":     data.TradeTime,
-					"ask":            data.AskOrderId,
-					"bid":            data.BidOrderId,
-				}
-
-				raw, _ := json.Marshal(relog)
+				raw, _ := json.Marshal(data)
 				t.push_match_result(raw)
 			}()
 		case uniq := <-t.tp.ChCancelResult:
@@ -305,8 +296,8 @@ func (t *tengine) push_match_result(data []byte) {
 	key := types.FormatTradeResult.Format(t.symbol)
 	err := rdc.RPush(cx, key, data).Err()
 	logrus.Infof("往%s队列RPush: %s %s", key, data, err)
-	if viper.GetBool("haotrader.notify_quote") {
-		quote_key := types.FormatQuoteTradeResult.Format(t.symbol)
-		rdc.RPush(cx, quote_key, data)
-	}
+	// if viper.GetBool("haotrader.notify_quote") {
+	// 	quote_key := types.FormatQuoteTradeResult.Format(t.symbol)
+	// 	rdc.RPush(cx, quote_key, data)
+	// }
 }