// Copyright 2013-2014 Fuzamei tech Ltd. All rights reserved.
package tick
// 本文件实现大智慧数据源接口, 实时数据和历史数据的获取和保存
import (
"encoding/binary"
"errors"
"io"
"log"
"net"
"strconv"
"strings"
"time"
"tickserver/server/market"
"golang.org/x/text/encoding/simplifiedchinese"
)
// DzhDS实现了dataSource接口, 并对dzh的历史数据和实时数据保存
type DzhDS struct {
*DSBase
conf *DsConf
//insMap map[string]*market.Instrument
tcp net.Conn
}
func init() {
drivers[Dzh] = newDzhDS
}
func newDzhDS(conf *DsConf) (DataSource, error) {
tcp, err := net.DialTimeout("tcp", conf.Url, 2*time.Second)
if err != nil {
return nil, err
}
return &DzhDS{
DSBase: NewDsBase(conf),
conf: conf,
tcp: tcp,
//insMap: make(map[string]*market.Instrument),
}, nil
}
func (dds *DzhDS) Name() string {
return Dzh
}
//func (dds *DzhDS) SubIns() *event.Event {
//return dds.insPublisher.Event()
//}
func (dds *DzhDS) Run() {
log.Println("DzhDS.Run")
//go dds.RunSave(64)
for {
err := dds.read(dds.tcp)
if err != nil {
log.Println(err)
dds.tcp, err = net.Dial("tcp", dds.conf.Url)
if err != nil {
log.Fatal(err)
}
continue
}
}
}
func (dds *DzhDS) read(r io.Reader) error {
var t int32
err := binary.Read(r, binary.LittleEndian, &t)
if err != nil {
return errors.New("read data TYPE error:" + err.Error())
}
var count, size uint32
err = binary.Read(r, binary.LittleEndian, &count)
if err != nil {
return errors.New("read data COUNT error:" + err.Error())
}
err = binary.Read(r, binary.LittleEndian, &size)
if err != nil {
return errors.New("read data SIZE error:" + err.Error())
}
switch t {
case 0:
return dds.resoleL1(r, int(count))
case 1:
return dds.resoleL2(r, int(count))
case 2:
for i := 0; i < int(count); i++ {
mbi := MarketBoardInfo{}
err = binary.Read(r, binary.LittleEndian, &mbi)
if err != nil {
log.Println("read MarketBoardInfo error:", err)
}
}
default:
log.Fatal("can't go here")
}
return nil
}
type Ext struct {
Code [16]byte
Name [16]byte
PrevClose uint32
TopLtd uint32
BotLtd uint32
}
// 五档行情数据结构
type QuoteL1 struct {
Ext
Number int16
Tim32 int32
Open int32
High int32
Low int32
New int32
AllVol int32
AllAmount int32
MarketVal int32
Reserver2 int32
PricesOfBid [5]int32
VolumeOfBid [5]int32
PricesOfAsk [5]int32
VolumeOfAsk [5]int32
Zero [2]int32
}
func QuoteL1ToMD(ql1 *QuoteL1, ins *Instrument, insIdStr string) *Market {
mk := &Market{}
mk.InsId = ins.Id
mk.Type = IntDzh
factor := 100.0
sid := market.RealInsId(insIdStr)
if strings.HasPrefix(sid, "SH510") || strings.HasPrefix(sid, "SZ160") {
factor = 1000.0
}
mk.Asks = make([]PP, 10)
mk.Bids = make([]PP, 10)
for i := 0; i < 5; i++ {
mk.Asks[i][0] = float64(ql1.PricesOfAsk[i]) / factor
mk.Asks[i][1] = float64(ql1.VolumeOfAsk[i]) * 100.
mk.Bids[i][0] = float64(ql1.PricesOfBid[i]) / factor
mk.Bids[i][1] = float64(ql1.VolumeOfBid[i]) * 100.
}
mk.Close = float64(ql1.PrevClose) / factor
mk.Open = float64(ql1.Open) / factor
mk.LastPrice = float64(ql1.New) / factor
mk.Timestamp = int64(ql1.Tim32) * 1000
oldVol := mk.AllVolume
mk.AllVolume = float64(ql1.AllVol) * 100.
mk.LastVolume = mk.AllVolume - oldVol
if mk.LastVolume < 0 {
mk.LastVolume = mk.AllVolume
}
mk.AllAmount = float64(ql1.AllAmount)
return mk
}
// 十档行情数据结构
type QuoteL2 struct {
Ext
Market [2]byte
Number int16
BuyEven float32
BuyVol float32
SellEven float32
SellVol float32
PricesOfBid [5]float32
VolumeOfBid [5]float32
PricesOfAsk [5]float32
VolumeOfAsk [5]float32
}
func QuoteL2ToMD(ql2 *QuoteL2, ins *Instrument) *Market {
//mk := ins.GetMk()
mk := &Market{}
mk.Type = IntDzh
mk.InsId = ins.Id
if len(mk.Asks) == 0 {
mk.Asks = make([]PP, 10)
}
if len(mk.Bids) == 0 {
mk.Bids = make([]PP, 10)
}
for i := 0; i < 5; i++ {
mk.Asks[i+5][0] = float64(ql2.PricesOfAsk[i])
mk.Asks[i+5][1] = float64(ql2.VolumeOfAsk[i]) * 100
mk.Bids[i+5][0] = float64(ql2.PricesOfBid[i])
mk.Bids[i+5][1] = float64(ql2.VolumeOfBid[i]) * 100
}
//ins.SetMk(mk)
// log.Printf("@@@@@@@:%+v\n", mk)
return mk
}
type MarketBoardInfo struct {
Code [16]byte
Info [48]byte
}
func (dds *DzhDS) resole(r io.Reader, count int, isL1 bool) ([]*QuoteL1, []*QuoteL2, error) {
ql1s := make([]*QuoteL1, count)
ql2s := make([]*QuoteL2, count)
for i := 0; i < int(count); i++ {
if isL1 {
ql1 := &QuoteL1{}
err := binary.Read(r, binary.LittleEndian, ql1)
if err != nil {
return nil, nil, errors.New("read QuoteL1 error:" + err.Error())
}
ql1s[i] = ql1
} else {
ql2 := &QuoteL2{}
err := binary.Read(r, binary.LittleEndian, ql2)
if err != nil {
return nil, nil, errors.New("read QuoteL2 error:" + err.Error())
}
ql2s[i] = ql2
}
}
return ql1s, ql2s, nil
}
func (dds *DzhDS) addIns(ext Ext) (*Instrument, string) {
sid := b2s(ext.Code[:])
exid := market.SHEX
if strings.Contains(sid, market.SZEX) {
exid = market.SZEX
}
sname := b2s(ext.Name[:])
trans := simplifiedchinese.GBK.NewDecoder()
dst := make([]byte, 1024)
insIdStr := market.DzhPrefix + sid
if len(sid) != 8 {
log.Println("wrong sid", sid)
return nil, ""
}
insId, _ := strconv.ParseInt(sid[2:], 10, 64)
ins, ok := dds.insMap[insId]
if !ok {
nDst, _, err := trans.Transform(dst, []byte(sname), true)
if err == nil {
sname = string(dst[0:nDst])
}
ins = &Instrument{
Id: insId,
Name: sname,
Type: market.Securities,
ExId: exid,
PriceInc: 0.01,
}
dds.insMap[insId] = ins
//log.Println(ins)
//dds.insPublisher.Publish(ins)
}
return ins, insIdStr
}
func (dds *DzhDS) resoleL1(r io.Reader, count int) error {
ql1s, _, err := dds.resole(r, count, true)
if err != nil {
log.Println(err)
return err
}
for _, q := range ql1s {
ins, insIdStr := dds.addIns(q.Ext)
if ins == nil {
//log.Fatal("@@@: DzhDS.resoleL1 ins == nil")
continue
}
mk := QuoteL1ToMD(q, ins, insIdStr)
//ins.SetMk(mk)
if mk.LastVolume == 0 {
continue // 交易量为0, 不存储
}
dds.Save(mk)
}
return nil
}
func b2s(b []byte) string {
for i, c := range b {
if c == 0 {
return string(b[:i])
}
}
return string(b[:])
}
func (dds *DzhDS) resoleL2(r io.Reader, count int) error {
_, ql2s, err := dds.resole(r, count, false)
if err != nil {
log.Println(err)
return err
}
for _, q := range ql2s {
ins, _ := dds.addIns(q.Ext)
if ins == nil {
continue
}
mk := QuoteL2ToMD(q, ins)
dds.Save(mk)
}
return nil
}