· 8 years ago · Mar 29, 2018, 06:34 AM
1package main
2
3import (
4 "encoding/json"
5 "context"
6 "fmt"
7 "errors"
8 "net/http"
9 "strconv"
10 "time"
11 "database/sql"
12 _ "github.com/lib/pq"
13 "github.com/shopspring/decimal"
14 "net/url"
15 "github.com/gorilla/websocket"
16 "regexp"
17 "strings"
18 "github.com/fatih/color"
19 "log"
20)
21
22type DataPoint struct {
23 Time int `json:"time"`
24 Close decimal.Decimal `json:"close"`
25 High decimal.Decimal `json:"high"`
26 Low decimal.Decimal `json:"low"`
27 Open decimal.Decimal `json:"open"`
28 Volume1 decimal.Decimal `json:"volumefrom"`
29 Volume2 decimal.Decimal `json:"volumeto"`
30}
31
32type Trade struct {
33 Time time.Time
34 Rate decimal.Decimal
35 Amount decimal.Decimal
36}
37
38var tables = []string{
39 `
40 create table if not exists crypto
41 (
42 time timestamp NOT NULL,
43 fsym varchar(3) NOT NULL,
44 tsym varchar(3) NOT NULL,
45 open numeric(15, 2),
46 close numeric(15, 2),
47 high numeric(15, 2),
48 low numeric(15, 2),
49 volume1 numeric(15, 2),
50 volume2 numeric(15, 2),
51 range numeric(15, 2),
52 differential numeric(15, 2),
53 volumedelta numeric(15, 2)
54 );
55 `,
56}
57
58func initDB() (*sql.DB, error) {
59 psql := fmt.Sprintf("host=%s port=%d user=%s "+
60 "dbname=%s sslmode=disable",
61 "192.168.69.106", 5432, "postgres", "crypto")
62
63 db, err := sql.Open("postgres", psql)
64 if err != nil {
65 return nil, err
66 }
67
68 err = db.Ping()
69 if err != nil {
70 return nil, err
71 }
72
73 for _, k := range tables {
74 _, err = db.Query(k)
75 if err != nil {
76 return nil, err
77 }
78 }
79
80 return db, nil
81}
82
83func insertSeries(db *sql.DB, fsym, tsym string, series []DataPoint) error {
84 tx, err := db.Begin()
85 if err != nil {
86 return err
87 }
88
89 for _, point := range series {
90 t := time.Unix(int64(point.Time), 64)
91 if _, err := tx.Exec(`insert into crypto
92 (fsym, tsym, time, open, close, high, low, volume1, volume2, range, differential, volumedelta)
93 values
94 ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
95 fsym, tsym, t, point.Open, point.Close,
96 point.High, point.Low, point.Volume1, point.Volume2,
97 point.High.Sub(point.Low), point.Close.Sub(point.Open),
98 point.Volume2.Sub(point.Volume1)); err != nil {
99 return err
100 }
101 }
102
103 if err = tx.Commit(); err != nil {
104 return err
105 }
106
107 return nil
108}
109
110// Bitfinex Trades
111func getBFXTrades(ctx context.Context, out chan *Trade) error {
112 addr := url.URL{Scheme: "wss", Host: "api.bitfinex.com", Path: "/ws/2"}
113 conn, _, err := websocket.DefaultDialer.Dial(addr.String(), nil)
114 if err != nil {
115 return err
116 }
117
118 payload, err := json.Marshal(struct {
119 Event string `json:"event"`
120 Channel string `json:"channel"`
121 Symbol string `json:"symbol"`
122 }{
123 "subscribe",
124 "trades",
125 "ETHUSD",
126 })
127
128 if err != nil {
129 return err
130 }
131
132 conn.WriteMessage(websocket.TextMessage, payload)
133
134 reg := regexp.MustCompile(`(\[([0-9]{1,5}),"([A-z]+)"),\[(-?\d+).\d+,(-?\d+).\d+,(-?\d+).\d+]]`)
135 go func() {
136 for {
137 select {
138 case <- ctx.Done():
139 break
140 default:
141 }
142
143 _, m, err := conn.ReadMessage()
144 if err != nil {
145 log.Println(err)
146 }
147
148 msg := string(m)
149
150 if reg.MatchString(msg) {
151 fields := strings.Split(msg, ",")
152 if fields[1] == `"tu"` {
153 _, err = strconv.Atoi(fields[3])
154 if err != nil {
155 log.Println(err)
156 continue
157 }
158 amount, err := decimal.NewFromString(fields[4])
159 if err != nil {
160 log.Println(err)
161 continue
162 }
163 price, err := decimal.NewFromString(fields[5][:len(fields[5]) - 2])
164 if err != nil {
165 log.Println(err)
166 continue
167 }
168
169 out <- &Trade{
170 Rate: price,
171 Amount: amount,
172 }
173 }
174 }
175 }
176
177 }()
178 return nil
179}
180
181func getGDAXTrades(ctx context.Context, out chan *Trade) error {
182 addr := url.URL{Scheme: "wss", Host: "ws-feed.gdax.com", Path: "/"}
183 conn, _, err := websocket.DefaultDialer.Dial(addr.String(), nil)
184 if err != nil {
185 return err
186 }
187
188 payload, err := json.Marshal(struct {
189 Typ string `json:"type"`
190 Pairs []string `json:"product_ids"`
191 Channels []string `json:"channels"`
192 }{
193 "subscribe",
194 []string{"ETH-USD"},
195 []string{"level2", "heartbeat"},
196 })
197
198 if err != nil {
199 return err
200 }
201
202 if err = conn.WriteMessage(websocket.TextMessage, payload); err != nil {
203 return err
204 }
205
206 go func() {
207 for {
208 select {
209 case <- ctx.Done():
210 break
211 default:
212 }
213
214 msg := make(map[string]interface{})
215 if err = conn.ReadJSON(&msg); err != nil {
216 log.Println(err)
217 }
218
219 switch msg["type"] {
220 case "heartbeat":
221 case "l2update":
222 for _, c := range msg["changes"].([]interface{}) {
223 t := c.([]interface{})
224 if t[2] == "0" {
225 continue
226 }
227
228 rate, err := decimal.NewFromString(t[1].(string))
229 if err != nil {
230 log.Println(err)
231 continue
232 }
233
234 amount, err := decimal.NewFromString(t[2].(string))
235 if err != nil {
236 log.Println(err)
237 continue
238 }
239
240 if t[0].(string) == "sell" {
241 amount = amount.Neg()
242 }
243
244
245
246 out <- &Trade{
247 Rate: rate,
248 Amount: amount,
249 }
250 }
251 }
252 }
253 }()
254 return nil
255}
256
257func getGEMTrades(ctx context.Context, out chan *Trade) error {
258 addr := url.URL{Scheme: "wss", Host: "api.gemini.com", Path: "/v1/marketdata/ethusd"}
259 conn, _, err := websocket.DefaultDialer.Dial(addr.String(), nil)
260 if err != nil {
261 return err
262 }
263
264 go func() {
265 for {
266 select {
267 case <- ctx.Done():
268 break
269 default:
270 }
271
272 msg := make(map[string]interface{})
273 if err := conn.ReadJSON(&msg); err != nil {
274 log.Println(err)
275 continue
276 }
277
278 switch msg["type"] {
279 case "update":
280 us, ok := msg["events"].([]interface{})
281 if ok {
282 for _, k := range us {
283 t, ok := k.(map[string]interface{})
284 if ok {
285 if t["type"] == "change" && t["reason"] == "trade" {
286 amount, err := decimal.NewFromString(t["delta"].(string))
287 if err != nil {
288 continue
289 }
290 amount = amount.Abs()
291 if t["type"] == "ask" {
292 amount = amount.Neg()
293 }
294 pri, ok := t["price"].(string)
295 fmt.Println(ok)
296 if !ok {
297 continue
298 }
299 price, err := decimal.NewFromString(pri)
300 if err != nil {
301 continue
302 }
303 out <- &Trade{
304 Rate: price,
305 Amount: amount,
306 }
307 }
308 }
309 }
310 }
311 }
312 }
313 }()
314
315 return nil
316}
317
318
319// Cryptocompare
320func getSeriesHourly(fsym, tsym string, date time.Time) ([]DataPoint, error) {
321 d := strconv.FormatInt(date.Unix(), 10)
322
323 baseUrl := "https://min-api.cryptocompare.com/data/histohour"
324 url := fmt.Sprintf("%s?limit=200&fsym=%s&tsym=%s&toTs=%s", baseUrl, fsym, tsym, d)
325 fmt.Println(url)
326 resp, err := http.Get(url)
327 if err != nil {
328 panic(err)
329 }
330
331 type Response struct {
332 Response string `json:"Response"`
333 Type int `json:"Type"`
334 Aggregated bool `json:"Aggregated"`
335 Data []DataPoint `json:"Data"`
336 }
337
338 r := &Response{}
339 dec := json.NewDecoder(resp.Body)
340 if err = dec.Decode(r); err != nil {
341 return nil, err
342 }
343
344
345 if r.Response != "Success" {
346 return nil, errors.New(r.Response)
347 }
348
349 return r.Data, nil
350}
351
352func main() {
353 ctx := context.Background()
354 out := make(chan *Trade)
355
356 if err := getGEMTrades(ctx, out); err != nil {
357 panic(err)
358 }
359
360 // go func() {
361 // if err := getGDAXTrades(ctx, out); err != nil {
362 // panic(err)
363 // }
364 // }()
365
366 // go func() {
367 // if err := getBFXTrades(ctx, out); err != nil {
368 // panic(err)
369 // }
370 // }()
371
372 for t := range out {
373 if true {
374 cond := t.Amount.LessThan(decimal.NewFromFloat(0))
375 var outStr string
376 if t.Amount.Abs().GreaterThan(decimal.NewFromFloat(20)) {
377 outStr = fmt.Sprintf("Coins: %s\t\tRate: %s\t\tSize: %s\t\t*", t.Amount.Round(3), t.Rate.Round(2), t.Amount.Mul(t.Rate).Round(2))
378 } else {
379 outStr = fmt.Sprintf("Coins: %s\t\tRate: %s\t\tSize: %s", t.Amount.Round(3), t.Rate.Round(2), t.Amount.Mul(t.Rate).Round(2))
380 }
381 if cond {
382 color.Red(outStr)
383 } else {
384 color.Green(outStr)
385 }
386 }
387 }
388 // db, err := initDB()
389 // if err != nil {
390 // panic(err)
391 // }
392
393 // s, err := getSeriesHourly("ETH", "USD", time.Now())
394 // if err != nil {
395 // panic(err)
396 // }
397
398 // err = insertSeries(db, "ETH", "USD", s)
399 // if err != nil {
400 // panic(err)
401 // }
402}