tickserver/server/tick/ds_cfix.go

// Copyright 2013-2014 Fuzamei tech Ltd. All rights reserved.

package tick

// 本文件实现cfix数据源接口, 实时数据和历史数据的获取和保存

import (
	"fmt"
	"log"
	"strings"

	"tickserver/api/fix"
	"tickserver/markinfo"
	"tickserver/server/market"
)

// CFixDS实现了dataSource接口, 并对cfix的历史数据和实时数据保存
type CFixDS struct {
	*DSBase
	conf *DsConf
}

var cfixInss = []int{
	markinfo.AUDCAD,
	markinfo.AUDCHF,
	markinfo.AUDJPY,
	markinfo.AUDNZD,
	markinfo.AUDUSD,
	markinfo.CADJPY,
	markinfo.CHFJPY,
	markinfo.EURAUD,
	markinfo.EURCAD,
	markinfo.EURCHF,
	markinfo.EURGBP,
	markinfo.EURJPY,
	markinfo.EURNZD,
	markinfo.EURUSD,
	markinfo.GBPAUD,
	markinfo.GBPCAD,
	markinfo.GBPCHF,
	markinfo.GBPJPY,
	markinfo.GBPUSD,
	markinfo.NZDJPY,
	markinfo.NZDUSD,
	markinfo.OILUSD,
	markinfo.SGDJPY,
	markinfo.USDCAD,
	markinfo.USDCHF,
	markinfo.USDJPY,
	markinfo.USDPLN,
	markinfo.XAGUSD,
	markinfo.XAGEUR,
	markinfo.XAUUSD,
	markinfo.XAUEUR,
	markinfo.ZARJPY,
}

func init() {
	drivers[CFix] = newCFixDS
}

func newCFixDS(conf *DsConf) (DataSource, error) {
	cfixds := &CFixDS{
		DSBase: NewDsBase(conf),
		conf:   conf,
	}
	cfixds.insMap = cfixInsMap()

	return cfixds, nil
}

func (cfixds *CFixDS) Name() string {
	return CFix
}

func (cfixds *CFixDS) Run() {
	log.Println("CFixDS.Run")
	ss, err := fix.NewSessionSettings(cfixds.conf.CfgFile)
	if err != nil {
		panic(err)
	}
	sourdatach := make(chan *fix.TickFull, 1024)
	go cfixds.onMarket(sourdatach)
	app := fix.NewMarketInfoApp(sourdatach)
	//fmt.Println(app, ss, cfixds.conf.CfgFile)
	trade, err := fix.NewAppTradeClient(app, ss)
	if err != nil {
		panic(err)
	}
	fmt.Println("cfix run start")
	trade.Start()
	trade.Run()
	fmt.Println("cfix run end")
	trade.Stop()
	trade.Free()
	fmt.Println("cfix stop")
}

func (cfixds *CFixDS) onMarket(sourdatach chan *fix.TickFull) {
	for {
		tf := <-sourdatach

		mk := &Market{}
		mk.Type = IntCFix
		symbol := string(tf.Symbol[:7])
		ss := strings.Split(symbol, "/")
		if len(ss) != 2 {
			continue
		}
		symbol = ss[0] + ss[1]
		symbolId, err := markinfo.SymbolId(symbol)
		if err != nil {
			continue
		}
		mk.InsId = int64(symbolId)
		mk.Asks = make([]PP, tf.AskCount)
		mk.Bids = make([]PP, tf.BidCount)
		for i := 0; i < int(tf.AskCount); i++ {
			mk.Asks[i][0] = tf.AskPrice[i]
			mk.Asks[i][1] = tf.AskVolume[i]
		}
		for i := 0; i < int(tf.BidCount); i++ {
			mk.Bids[i][0] = tf.BidPrice[i]
			mk.Bids[i][1] = tf.BidVolume[i]
		}
		mk.Timestamp = int64(tf.Time)*1000 + int64(tf.Millisecond)
		mk.Close = tf.BidPrice[0]
		mk.High = tf.BidPrice[0]
		mk.Low = tf.BidPrice[0]
		mk.Open = tf.BidPrice[0]
		mk.AllAmount = tf.BidVolume[0]
		mk.AllVolume = tf.BidVolume[0]
		mk.LastPrice = tf.BidPrice[0]
		mk.LastVolume = tf.BidVolume[0]
		cfixds.Save(mk)
	}
}

func cfixInsMap() map[int64]*Instrument {
	insMap := make(map[int64]*Instrument)
	for _, id := range cfixInss {
		x, _ := markinfo.SymbolName(id)
		u, _ := markinfo.SymbolUint(x)
		ins := &Instrument{
			Id:       int64(id),
			Name:     x,
			ExId:     CFix,
			Type:     market.Forex,
			PriceInc: u,
			//StartTime: time.Now().Unix() * 1000,
		}
		insMap[int64(id)] = ins
	}
	return insMap
}