Skip to content

Commit 654e678

Browse files
a0sa0s
committed
mass refactoring
1 parent a594f81 commit 654e678

File tree

6 files changed

+428
-339
lines changed

6 files changed

+428
-339
lines changed

‎config/config.go

+68-21
Original file line numberDiff line numberDiff line change
@@ -2,35 +2,35 @@ package config
22

33
import (
44
"flag"
5-
"fmt"
65
"github.com/octago/sflags/gen/gflag"
7-
"log"
8-
"os"
6+
"path/filepath"
7+
"strings"
98
"time"
9+
dict "tinkoff-invest-dumper/dictionary"
1010
)
1111

1212
var VersionString = "development"
1313

14-
var Conf Config
14+
type Logger interface {
15+
Fatalln(v ...interface{})
16+
}
1517

1618
type Config struct {
17-
Token string `flag:"token" desc:"your sandbox's token"`
18-
Path string `flag:"path" desc:"path to storage dir"`
19-
19+
Token string `flag:"token" desc:"your sandbox's token"`
20+
Path string `flag:"path" desc:"path to storage dir"`
2021
TimeSuffixEnabled bool `flag:"time-suffix-enabled" desc:"add the time suffix to every filename on (re)start"`
2122
TimeSuffixFormat string `flag:"time-suffix-format" desc:"go format of the time suffix (see https://golang.org/src/time/format.go)"`
2223
TimeSuffixStartedAt time.Time `flag:"-"`
24+
Orderbook string `flag:"orderbook" desc:"list of tickers to subscribe for orderbooks"`
25+
OrderbookDepth int `flag:"orderbook-depth" desc:"depth of orderbook: from 1 to 20"`
26+
Candle string `flag:"candle" desc:"list of tickers to subscribe for candles"`
27+
CandleInterval string `flag:"candle-interval" desc:"interval of candles: 1min,2min,3min,5min,10min,15min,30min,hour,2hour,4hour,day,week,month"`
28+
Version bool `flag:"version" desc:"show version info"`
2329

24-
Orderbook string `flag:"orderbook" desc:"list of tickers to subscribe for orderbooks"`
25-
OrderbookDepth int `flag:"orderbook-depth" desc:"depth of orderbook: from 1 to 20"`
26-
27-
Candle string `flag:"candle" desc:"list of tickers to subscribe for candles"`
28-
CandleInterval string `flag:"candle-interval" desc:"interval of candles: 1min,2min,3min,5min,10min,15min,30min,hour,2hour,4hour,day,week,month"`
29-
30-
Version bool `flag:"version" desc:"show version info"`
30+
logger Logger
3131
}
3232

33-
func NewConfig() *Config {
33+
func NewConfig(l Logger) *Config {
3434
config := &Config{
3535
Path: ".",
3636
TimeSuffixEnabled: false,
@@ -39,20 +39,67 @@ func NewConfig() *Config {
3939
OrderbookDepth: 20,
4040
CandleInterval: "1min",
4141
Version: false,
42+
43+
logger: l,
4244
}
4345
err := gflag.ParseToDef(config)
4446
if err != nil {
45-
log.Fatalf("config: %v", err)
47+
l.Fatalln("new config:", err)
4648
}
4749
flag.Parse()
4850

4951
return config
5052
}
5153

52-
func init() {
53-
Conf = *NewConfig()
54-
if Conf.Version {
55-
fmt.Printf("%s\n", VersionString)
56-
os.Exit(0)
54+
func (c *Config) BuildOrderbookPath(ticker dict.Ticker) string {
55+
var arr []string
56+
57+
arr = append(arr, string(ticker))
58+
if c.TimeSuffixEnabled {
59+
startedAt := c.TimeSuffixStartedAt.Format(c.TimeSuffixFormat)
60+
arr = append(arr, startedAt)
61+
}
62+
arr = append(arr, "obk")
63+
64+
path, err := filepath.Abs(filepath.Join(c.Path, strings.Join(arr, "-")))
65+
if err != nil {
66+
c.logger.Fatalln(err)
67+
}
68+
return path
69+
}
70+
71+
func (c *Config) BuildCandlePath(ticker dict.Ticker) string {
72+
var arr []string
73+
74+
arr = append(arr, string(ticker))
75+
if c.TimeSuffixEnabled {
76+
startedAt := c.TimeSuffixStartedAt.Format(c.TimeSuffixFormat)
77+
arr = append(arr, startedAt)
78+
}
79+
arr = append(arr, "cdl")
80+
81+
path, err := filepath.Abs(filepath.Join(c.Path, strings.Join(arr, "-")))
82+
if err != nil {
83+
c.logger.Fatalln(err)
84+
}
85+
return path
86+
}
87+
88+
func (c *Config) GetOrderbookTickers() []dict.Ticker {
89+
return stringToTickerList(c.Orderbook)
90+
}
91+
92+
func (c *Config) GetCandleTickers() []dict.Ticker {
93+
return stringToTickerList(c.Candle)
94+
}
95+
96+
func stringToTickerList(flag string) []dict.Ticker {
97+
var tickers []dict.Ticker
98+
flags := strings.Split(flag, ",")
99+
for _, f := range flags {
100+
if f != "" {
101+
tickers = append(tickers, dict.Ticker(f))
102+
}
57103
}
104+
return tickers
58105
}

‎dictionary/dict.go

+14-4
Original file line numberDiff line numberDiff line change
@@ -51,12 +51,22 @@ TICKERS:
5151
return &Dictionary{figiInstrument: fs, tickerInstrument: ts}, nil
5252
}
5353

54-
func (d *Dictionary) GetFIGIByTicker(t Ticker) Figi {
55-
return Figi(d.tickerInstrument[t].FIGI)
54+
func (d *Dictionary) GetFIGIByTicker(t Ticker) (Figi, error) {
55+
ins, ok := d.tickerInstrument[t]
56+
if !ok {
57+
return "", errors.New(fmt.Sprint("ticker not found:", t))
58+
}
59+
60+
return Figi(ins.FIGI), nil
5661
}
5762

58-
func (d *Dictionary) GetTickerByFIGI(f Figi) Ticker {
59-
return Ticker(d.figiInstrument[f].Ticker)
63+
func (d *Dictionary) GetTickerByFIGI(f Figi) (Ticker, error) {
64+
ins, ok := d.figiInstrument[f]
65+
if !ok {
66+
return "", errors.New(fmt.Sprint("figi not found:", f))
67+
}
68+
69+
return Ticker(ins.Ticker), nil
6070
}
6171

6272
type sortTicker []Ticker

‎eventer/event_receiver.go

+197
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,197 @@
1+
package eventer
2+
3+
import (
4+
sdk "github.com/TinkoffCreditSystems/invest-openapi-go-sdk"
5+
"math/rand"
6+
"reflect"
7+
"time"
8+
dict "tinkoff-invest-dumper/dictionary"
9+
)
10+
11+
type Logger interface {
12+
Fatalln(v ...interface{})
13+
}
14+
15+
type Dictionary interface {
16+
GetFIGIByTicker(t dict.Ticker) (dict.Figi, error)
17+
GetTickerByFIGI(figi dict.Figi) (dict.Ticker, error)
18+
}
19+
20+
type OrderbookEvent struct {
21+
Figi dict.Figi
22+
Ticker dict.Ticker
23+
LocalTime time.Time
24+
Event sdk.OrderBookEvent
25+
}
26+
27+
type CandleEvent struct {
28+
Figi dict.Figi
29+
Ticker dict.Ticker
30+
LocalTime time.Time
31+
Event sdk.CandleEvent
32+
}
33+
34+
type EventReceiver struct {
35+
streamingClient *sdk.StreamingClient
36+
logger Logger
37+
dictionary Dictionary
38+
39+
orderbooks map[dict.Ticker][]chan OrderbookEvent
40+
candles map[dict.Ticker][]chan CandleEvent
41+
}
42+
43+
func NewEventReceiver(lg Logger, sc *sdk.StreamingClient, dc Dictionary, ) *EventReceiver {
44+
return &EventReceiver{
45+
streamingClient: sc,
46+
logger: lg,
47+
dictionary: dc,
48+
orderbooks: make(map[dict.Ticker][]chan OrderbookEvent),
49+
candles: make(map[dict.Ticker][]chan CandleEvent),
50+
}
51+
}
52+
53+
func (l *EventReceiver) SubscribeToOrderbook(ticker dict.Ticker, depth int) chan OrderbookEvent {
54+
_, ok := l.orderbooks[ticker]
55+
if !ok {
56+
figi, err := l.dictionary.GetFIGIByTicker(ticker)
57+
if err != nil {
58+
l.logger.Fatalln("new subscription to orderbook:", err)
59+
}
60+
61+
err = l.streamingClient.SubscribeOrderbook(string(figi), depth, requestID())
62+
if err != nil {
63+
l.logger.Fatalln("new subscription to orderbook:", err)
64+
}
65+
l.orderbooks[ticker] = []chan OrderbookEvent{}
66+
}
67+
68+
ch := make(chan OrderbookEvent)
69+
l.orderbooks[ticker] = append(l.orderbooks[ticker], ch)
70+
return ch
71+
}
72+
73+
func (l *EventReceiver) SubscribeToCandle(ticker dict.Ticker, interval string) chan CandleEvent {
74+
_, ok := l.candles[ticker]
75+
if !ok {
76+
figi, err := l.dictionary.GetFIGIByTicker(ticker)
77+
if err != nil {
78+
l.logger.Fatalln("new candle subscription:", err)
79+
}
80+
81+
err = l.streamingClient.SubscribeCandle(string(figi), sdk.CandleInterval(interval), requestID())
82+
if err != nil {
83+
l.logger.Fatalln("new candle subscription:", err)
84+
}
85+
l.candles[ticker] = []chan CandleEvent{}
86+
}
87+
88+
ch := make(chan CandleEvent)
89+
l.candles[ticker] = append(l.candles[ticker], ch)
90+
return ch
91+
}
92+
93+
func (l *EventReceiver) WrapOrderbookEvent(e sdk.OrderBookEvent) *OrderbookEvent {
94+
figi := dict.Figi(e.OrderBook.FIGI)
95+
96+
ticker, err := l.dictionary.GetTickerByFIGI(figi)
97+
if err != nil {
98+
l.logger.Fatalln("create orderbook event:", err)
99+
}
100+
101+
return &OrderbookEvent{
102+
Figi: figi,
103+
Ticker: ticker,
104+
LocalTime: time.Now(),
105+
Event: e,
106+
}
107+
}
108+
109+
func (l *EventReceiver) WrapCandleEvent(e sdk.CandleEvent) *CandleEvent {
110+
figi := dict.Figi(e.Candle.FIGI)
111+
112+
ticker, err := l.dictionary.GetTickerByFIGI(figi)
113+
if err != nil {
114+
l.logger.Fatalln("create candle event:", err)
115+
}
116+
117+
return &CandleEvent{
118+
Figi: figi,
119+
Ticker: ticker,
120+
LocalTime: time.Now(),
121+
Event: e,
122+
}
123+
}
124+
125+
func (l *EventReceiver) Start() {
126+
for {
127+
err := l.streamingClient.RunReadLoop(func(event interface{}) error {
128+
switch sdkEvent := event.(type) {
129+
case sdk.OrderBookEvent:
130+
ob := l.WrapOrderbookEvent(sdkEvent)
131+
channels, ok := l.orderbooks[ob.Ticker]
132+
if !ok {
133+
l.logger.Fatalln("event receiver unknown channel:", ob.Ticker)
134+
}
135+
for _, ch := range channels {
136+
ch <- *ob
137+
}
138+
139+
case sdk.CandleEvent:
140+
cd := l.WrapCandleEvent(sdkEvent)
141+
channels, ok := l.candles[cd.Ticker]
142+
if !ok {
143+
l.logger.Fatalln("event receiver unknown channel:", cd.Ticker)
144+
}
145+
for _, ch := range channels {
146+
ch <- *cd
147+
}
148+
149+
default:
150+
l.logger.Fatalln("event receiver unsupported event type:", reflect.TypeOf(event))
151+
}
152+
153+
return nil
154+
})
155+
if err != nil {
156+
l.logger.Fatalln("event lister:", err)
157+
}
158+
}
159+
}
160+
161+
func requestID() string {
162+
var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
163+
164+
b := make([]rune, 12)
165+
for i := range b {
166+
b[i] = letterRunes[rand.Intn(len(letterRunes))]
167+
}
168+
169+
return string(b)
170+
}
171+
172+
func init() {
173+
rand.Seed(time.Now().UnixNano()) // for requestID
174+
}
175+
//
176+
//func (s *mainScope) unsubscribeOrderbook(streamingClient *sdk.StreamingClient) {
177+
// for _, ticker := range s.orderbookTickers {
178+
// figi := s.dict.GetFIGIByTicker(ticker)
179+
// err := streamingClient.UnsubscribeOrderbook(string(figi), config.Conf.OrderbookDepth, requestID())
180+
// if err != nil {
181+
// s.logger.Fatalln(err)
182+
// }
183+
// s.logger.Println("Unsubscribed from orderbook", ticker, figi)
184+
// }
185+
//}
186+
//
187+
//
188+
//func (s *mainScope) unsubscribeCandles(streamingClient *sdk.StreamingClient) {
189+
// for _, ticker := range s.candleTickers {
190+
// figi := s.dict.GetFIGIByTicker(ticker)
191+
// err := streamingClient.UnsubscribeCandle(string(figi), sdk.CandleInterval(config.Conf.CandleInterval), requestID())
192+
// if err != nil {
193+
// s.logger.Fatalln(err)
194+
// }
195+
// s.logger.Println("Unsubscribed from candles", ticker, figi)
196+
// }
197+
//}

0 commit comments

Comments
 (0)