cybercongress / cyber-markets Goto Github PK

• Subscription/Unsubscripted mechanism (ws message)
• Broadcasting mechanism
• Kafka consumer

Using docker stages: build containers \w java. Run java applications in docker compose.

Implement simple windowed algorithm for calculation tickers for 1h, 3h, 6h, 24h tickers with 5 sec precision.

Implementation variants:

1. Using apache spark
2. Own jvm implementation

• Read kafka trades channels
• Calculate tickers (for this task just hold all info in memory, no storage required)
• Push tickers to kafka

After completing #71 #72 #73 we should update prod without data losing.

About:
https://deepstream.io/

Deepstream Java SDK
https://deepstream.io/install/java

Goal of this issue is stream trades to deepstream server and check them on client with simple page

Create wiki page with ws endpoints descriptions.
Optionally choose, 3dparty saas.

• [+] Understand cassandra tables( partition key, composite key)
• [+] Chose cassandra partition key for trades
• [+] Chose cassandra composite key for trades
• [+] Add table difinition to devops/elassandra/bootstrap.cql

C* query examples.
select * from markets.trades where date=[date], exchange=[exchange1,exchange2], pair=[pair1,pair2]
select * from markets.trades where date=[date], exchange=[exchange1,exchange2]
select * from markets.trades where date=[date], pair=[pair1,pair2]

We should adopt our [exchange-connectors] for usage outside of cyber-markets ecosystem. Users should be able to use it not only with kafka. Any community feedback here will be useful.

• Token Pair creating by dictionary

• HitBtc, Poloniex, Bitfinex, Bitstamp creating token pair in unified view

• add to Trade isReverted flag

• More flexible subscription mechanism (Select pairs on given exchanges)
• Return only N last trades

For cybernode docs I need to reference cyber-markets parameters such as ENV['DEEPSTREAMURL'], so when anything changes/breaks, cybernode admins knew where to look for.

• enable Travis at https://travis-ci.org/cyberFund (@litvintech, need your help)
• add .travis.yml with language: java to the top (it will pick up gradle automatically)
• add status badge to README.md

• Setup minicube
• Setup k8s
• Setup kafka
• Setup kafka manager
• Setup zookeeper
• Add documentation & bash script for bootstrap env

Use http://docs.confluent.io/current/cp-docker-images/docs/index.html images for:
kafka, zookeper

https://docs.gdax.com/

Fix error:
WebSocket connection to 'ws://localhost:8082/' failed: Connection closed before receiving a handshake response

Try to revert to previous master version and find bug

There are couple of format of data which we get from exchanges
There is proposal for general format of markets data for cyber-markets

• tradeId (trade id from exchange)
• timestamp
• exchange (need a convention about naming)
• type (only sell/buy, converting partial fill and other to this 2 types)
• baseToken (example: BTC, token in which we measure and buy other tokens - quote tokens)
• quoteToken (example: ETH, token which we buy/sell with baseToken)
• baseAmount (example: 1)
• quoteAmount (example: 10)
• spotPrice (example: 0.1 baseToken per one quoteToken)

For Poloniex there are only for base tokens BTC, ETH, XMR, USDT
total = baseAmount
amount = quoteAmount
rate = spotPrice

We expect that trade.timestamp was in seconds (as well as on others exchanges)
but actual Bitstamp value: timestamp 1970-01-18T10:26:59Z and that isn't correct
check it and convert it to unified view.

https://hub.docker.com/u/cybernode

Add Intellij Idea code style settings for kotlin sources.

--reconnection mechanism
--history gaps introduced

lensesio/stream-reactor#284

It's impossible to sync two kafka topics(trades-a,trades-b) in one cassandra table (trades).
If i try to sync it's separately (for example trades-a to trades) it work perfect.

that is props for separetely attemp:
connector.class=com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector connect.cassandra.key.space=markets topics=TRADES-a tasks.max=1 connect.cassandra.kcql=INSERT INTO trade SELECT * FROM TRADES-a; connect.cassandra.fetch.size=5000 connect.cassandra.max.retries=20 connect.progress.enabled=false connect.cassandra.error.policy=throw connect.cassandra.contact.points=docker.for.mac.localhost connect.cassandra.port=9042 connect.cassandra.threadpool.size=10 name=CassandraSinkConnector connect.cassandra.retry.interval=30000 connect.progress.enabled=true connect.cassandra.consistency.level=LOCAL_ONE value.converter=org.apache.kafka.connect.json.JsonConverter key.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false key.converter.schemas.enable=false

As I Say it’s work fine.
And if i delete this connector and create TRADES-b to trade connector it’s work fine too!

but If I try to create this

connector.class=com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkConnector connect.cassandra.key.space=markets topics=TRADES-a,TRADES-b tasks.max=1 connect.cassandra.kcql=INSERT INTO trade SELECT * FROM TRADES-a;INSERT INTO trade SELECT * FROM TRADES-b; connect.cassandra.fetch.size=5000 connect.cassandra.max.retries=20 connect.progress.enabled=false connect.cassandra.error.policy=throw connect.cassandra.contact.points=docker.for.mac.localhost connect.cassandra.port=9042 connect.cassandra.threadpool.size=10 name=CassandraSinkConnector connect.cassandra.retry.interval=30000 connect.progress.enabled=true connect.cassandra.consistency.level=LOCAL_ONE value.converter=org.apache.kafka.connect.json.JsonConverter key.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=false key.converter.schemas.enable=false

It’s doesn’t work..

`TRACE:

org.apache.kafka.connect.errors.ConnectException: No tables found in Cassandra for topics trades,trades
at com.datamountaineer.streamreactor.connect.cassandra.utils.CassandraUtils$.checkCassandraTables(CassandraUtils.scala:61)
at com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraJsonWriter.(CassandraJsonWriter.scala:54)
at com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraWriter$.apply(CassandraWriter.scala:43)
at com.datamountaineer.streamreactor.connect.cassandra.sink.CassandraSinkTask.start(CassandraSinkTask.scala:58)
at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:232)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:145)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)`

Strange log..

I repeat, the error is obtained only if I make the insertion into the same table from different topics.
If I do an insertion into different tables from different topics, this works well

0x
For trades needs to listen event LogFill:
Exchange contract

EtherDelta
Exchange
For trades needs to listen event Trade
Exchange contract
https://coinmarketcap.com/exchanges/etherdelta/

Java Ethereum Library
One issue which we need to resolve - choose way of integration

1. integration of connectors for dex's smart contracts to cyber-markets (which create dependency for whole cyber-markets to ethereum node)
2. side stand-alone services

Provider quality

• LVL 1 no guarantee / no history recovery
• LVL 2 guaranteed gap detection / no history recovery
• LVL 3 gap detection / gap recovery
  Need to think about this functionality in cyber-markets

• Selected methodology how to tests ws
• Check current impl throuput

Current Kafka + Kafka manager combination do not allow to measure topics loads.

%subj%

• Message parser for Poloniex
• Separate ws channel for order book
• Full drop order book every N minutes
• Kafka producers + Kafka messages

https://plugins.gradle.org/plugin/io.spring.dependency-management

https://travis-ci.org/cyberFund/cyber-markets#L486 output contains unhandled exceptions, but Travis is green. What is going on?