tickserver/server/tick/ds_oanda.go

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

package tick

// 本文件实现oanda数据源的tick数据获取下载和保存

import (
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"regexp"
	"strconv"
	"strings"
	"time"

	"tickserver/markinfo"
	"tickserver/server/market"
)

type oandaPairStatistics struct {
	Date       time.Time
	BuyPercent float64
	Rates      float64
	Symbol     string
}

var oanda_base = int(200)

var oanda_prev = make(map[string]*oandaPairStatistics)
var fetchoandaurl = "http://fxtrade.oanda.com/lang/cns/analysis/open-position-ratios"

func getOandaStatistics(ch chan<- *oandaPairStatistics) (err error) {
	html, err := httpDownload(fetchoandaurl)
	if err != nil {
		return err
	}
	regex := regexp.MustCompile("(?is)<ol[\\s]+class=\"position-ratio-list\"[^>]*>(.*?)<\\/ol>")
	matches := regex.FindAllStringSubmatch(html, -1)
	if len(matches) != 2 {
		return errors.New("regex matche error.")
	}
	//货币对--BuyPercent--SellPercent
	ratio_graph := get_span(matches[0][0])
	//货币对--Rates
	rates := get_span(matches[1][0])
	//精确到微妙
	t := match_time(html)
	if len(ratio_graph)%3 == 0 && len(rates)%2 == 0 && len(ratio_graph)/3 == len(rates)/2 {
		for i := 0; i < len(ratio_graph)/3; i++ {
			ops := &oandaPairStatistics{}
			ops.Symbol = strings.Replace(ratio_graph[3*i], "/", "", -1)
			ops.BuyPercent, _ = strconv.ParseFloat(ratio_graph[3*i+1], 64)
			ops.Rates, _ = strconv.ParseFloat(rates[2*i+1], 64)
			ops.Date = time.Unix(t/1000, (t%1000)*(1e6))
			sendOanda(ch, ops)
		}
	} else {
		return errors.New("invalid data.")
	}
	return
}

func sendOanda(ch chan<- *oandaPairStatistics, data *oandaPairStatistics) {
	if prev_data, ok := oanda_prev[data.Symbol]; ok {
		if *prev_data == *data {
			return
		}
	}
	// log.Println("oa client", data)
	oanda_prev[data.Symbol] = data
	select {
	case ch <- data:
	default:
	}
}

func httpDownload(url string) (string, error) {
	body, err := http.Get(url)
	if err != nil {
		fmt.Println(err)
		return "", err
	}
	defer body.Body.Close()
	b, err := ioutil.ReadAll(body.Body)
	if err != nil {
		fmt.Println(err)
		return "", err
	}
	return string(b), nil
}

var timeregex = regexp.MustCompile("(?is)<input.*?id=\"generation-timestamp\".*?value=\"(\\d+)\".*?>")

func match_time(html string) int64 {
	result := timeregex.FindAllStringSubmatch(html, -1)
	var time int64
	if len(result) == 1 && len(result[0]) == 2 {
		time, _ = strconv.ParseInt(result[0][1], 10, 64)
	}
	return time
}

var spanregex = regexp.MustCompile("(?is)<span[^>]+>.*?</span>")

func get_span(text string) []string {
	result := spanregex.FindAllString(text, -1)
	for i := 0; i < len(result); i++ {
		result[i] = strings.Trim(strip_tags(result[i]), " \t\r\n%")
	}
	return result
}

var stripregex = regexp.MustCompile("(?is)<[^>]+>")

func strip_tags(text string) string {
	return stripregex.ReplaceAllString(text, "")
}

var oandaInss = []string{
	"EURCHF",
	"XAGUSD",
	"EURGBP",
	"USDJPY",
	"USDCAD",
	"AUDJPY",
	"USDCHF",
	"XAUUSD",
	"EURJPY",
	"EURUSD",
	"GBPCHF",
	"AUDUSD",
	"EURAUD",
	"NZDUSD",
	"GBPJPY",
	"GBPUSD",
}

// OandaDS 实现数据源dataSource接口的定义
type OandaDS struct {
	*DSBase
	conf *DsConf
	//insMap map[string]*market.Instrument
}

func init() {
	drivers[Oanda] = newOandaDS
}

func newOandaDS(conf *DsConf) (DataSource, error) {
	log.Println("newOandaDS")
	ods := &OandaDS{
		DSBase: NewDsBase(conf),
		conf:   conf,
		//insMap: oandaInsMap(),
	}
	ods.insMap = oandaInsMap()
	return ods, nil
}

func (ods *OandaDS) Name() string {
	return Oanda
}

//func (ods *OandaDS) SubIns() *event.Event {
//return ods.insPublisher.Event()
//}

func (ods *OandaDS) onMarket(ps *oandaPairStatistics) {
	//insId := market.OandaPrefix + ps.Symbol
	insId, _ := markinfo.SymbolId(ps.Symbol)
	_, ok := ods.insMap[int64(insId)]
	if !ok {
		log.Fatal("OandaDS.onMarket error: insId is NOT in insMap:", insId)
	}
	//mk := ins.GetMk()
	mk := &Market{}
	mk.Type = IntOanda
	mk.InsId = int64(insId)
	mk.Timestamp = ps.Date.Unix() * 1000
	mk.LastPrice = 100 - ps.BuyPercent
	mk.LastVolume = 1
	//ins.SetMk(mk)
	ods.Save(mk)
}

/*func (ods *OandaDS) runHour() {
	ht := time.Tick(time.Hour)
	for _ = range ht {
		for _, ins := range ods.insMap {
			ods.Save(ins.GetMk())
		}
	}
}*/

func (ods *OandaDS) Run() {
	log.Println("OandaDS.run")
	//for _, ins := range ods.insMap {
	//ods.insPublisher.Publish(ins)
	//}
	// go ods.runHour()
	//go ods.RunSave(4)
	ch := make(chan *oandaPairStatistics, 1024)
	go func() {
		for {
			ps := <-ch
			ods.onMarket(ps)
		}
	}()
	for {
		var err error
		for {
			err = getOandaStatistics(ch)
			if err != nil {
				log.Println(err)
			}
			time.Sleep(3 * time.Second)
		}
	}
}

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