This repository has been archived by the owner on Feb 1, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 262
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
* 1 - write trades to a sqlite database * 2 - switch over to postgres db and use lib/pq library * 3 - log failed queries when interacting with the db * 4 - dbExec directly instead of using a prepared statement * 5 - save date_utc as a date type in db as UTC (no time zone so db doesn't convert automatically) * 6 - Kraken timestamp for trades is in seconds, convert to millis * 8 - Kraken should request trade cursor in milliseconds, not seconds * 9 - log when lastCursor is updated * 10 - Kraken use transaction IDs for updates to cursor * 7 - FillDBWriter should ignore duplicate insert values * 11 - fillDbWriter should register full asset string in db table * 12 - threading to move markets data to it's own table * 13 - write markets data to it's on table and fix schemas across db * 14 - set precision for floats to 15 decimals (max allowed by db for double precision types) * 15 - composite primary key for trades table * 16 - graceful failure when postgres database not running, remove pointer indirection to port in postgres config * 17 - db_version table and refactoring of db helper functions * 18 - update README to add postgres as a dependency * 19 - use better method to calculate elapsed time in millis
- Loading branch information
1 parent
335d191
commit 493b4b0
Showing
16 changed files
with
629 additions
and
40 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package database | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "fmt" | ||
| "log" | ||
| ) | ||
|
|
||
| /* | ||
| tables | ||
| */ | ||
| const sqlDbVersionTableCreate = "CREATE TABLE IF NOT EXISTS db_version (version INTEGER NOT NULL, date_completed_utc TIMESTAMP WITHOUT TIME ZONE NOT NULL, num_scripts INTEGER NOT NULL, time_elapsed_millis BIGINT NOT NULL, PRIMARY KEY (version))" | ||
| const sqlMarketsTableCreate = "CREATE TABLE IF NOT EXISTS markets (market_id TEXT PRIMARY KEY, exchange_name TEXT NOT NULL, base TEXT NOT NULL, quote TEXT NOT NULL)" | ||
| const sqlTradesTableCreate = "CREATE TABLE IF NOT EXISTS trades (market_id TEXT NOT NULL, txid TEXT NOT NULL, date_utc TIMESTAMP WITHOUT TIME ZONE NOT NULL, action TEXT NOT NULL, type TEXT NOT NULL, counter_price DOUBLE PRECISION NOT NULL, base_volume DOUBLE PRECISION NOT NULL, counter_cost DOUBLE PRECISION NOT NULL, fee DOUBLE PRECISION NOT NULL, PRIMARY KEY (market_id, txid))" | ||
|
|
||
| /* | ||
| indexes | ||
| */ | ||
| const sqlTradesIndexCreate = "CREATE INDEX IF NOT EXISTS date ON trades (market_id, date_utc)" | ||
|
|
||
| /* | ||
| insert statements | ||
| */ | ||
| // sqlDbVersionTableInsertTemplate inserts into the db_version table | ||
| const sqlDbVersionTableInsertTemplate = "INSERT INTO db_version (version, date_completed_utc, num_scripts, time_elapsed_millis) VALUES (%d, '%s', %d, %d)" | ||
|
|
||
| // SqlMarketsInsertTemplate inserts into the markets table | ||
| const SqlMarketsInsertTemplate = "INSERT INTO markets (market_id, exchange_name, base, quote) VALUES ('%s', '%s', '%s', '%s')" | ||
|
|
||
| // SqlTradesInsertTemplate inserts into the trades table | ||
| const SqlTradesInsertTemplate = "INSERT INTO trades (market_id, txid, date_utc, action, type, counter_price, base_volume, counter_cost, fee) VALUES ('%s', '%s', '%s', '%s', '%s', %.15f, %.15f, %.15f, %.15f)" | ||
|
|
||
| /* | ||
| queries | ||
| */ | ||
| // SqlQueryMarketsById queries the markets table | ||
| const SqlQueryMarketsById = "SELECT market_id, exchange_name, base, quote FROM markets WHERE market_id = $1 LIMIT 1" | ||
|
|
||
| // sqlQueryDbVersion queries the db_version table | ||
| const sqlQueryDbVersion = "SELECT version FROM db_version ORDER BY version desc LIMIT 1" | ||
|
|
||
| /* | ||
| query helper functions | ||
| */ | ||
| // QueryDbVersion queries for the version of the database | ||
| func QueryDbVersion(db *sql.DB) (uint32, error) { | ||
| rows, e := db.Query(sqlQueryDbVersion) | ||
| if e != nil { | ||
| return 0, fmt.Errorf("could not execute sql select query (%s): %s", sqlQueryDbVersion, e) | ||
| } | ||
| defer rows.Close() | ||
|
|
||
| for rows.Next() { | ||
| var dbVersion uint32 | ||
| e = rows.Scan(&dbVersion) | ||
| if e != nil { | ||
| return 0, fmt.Errorf("could not scan row to get the db version: %s", e) | ||
| } | ||
|
|
||
| log.Printf("fetched dbVersion from db: %d", dbVersion) | ||
| return dbVersion, nil | ||
| } | ||
|
|
||
| return 0, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| package database | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "fmt" | ||
| "log" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/stellar/kelp/support/postgresdb" | ||
| "github.com/stellar/kelp/support/utils" | ||
| ) | ||
|
|
||
| var upgradeScripts = []*UpgradeScript{ | ||
| makeUpgradeScript(1, sqlDbVersionTableCreate), | ||
| makeUpgradeScript(2, | ||
| sqlMarketsTableCreate, | ||
| sqlTradesTableCreate, | ||
| sqlTradesIndexCreate, | ||
| ), | ||
| } | ||
|
|
||
| // UpgradeScript encapsulates a script to be run to upgrade the database from one version to the next | ||
| type UpgradeScript struct { | ||
| version uint32 | ||
| commands []string | ||
| } | ||
|
|
||
| // makeUpgradeScript encapsulates a script to be run to upgrade the database from one version to the next | ||
| func makeUpgradeScript(version uint32, command string, moreCommands ...string) *UpgradeScript { | ||
| allCommands := []string{command} | ||
| allCommands = append(allCommands, moreCommands...) | ||
|
|
||
| return &UpgradeScript{ | ||
| version: version, | ||
| commands: allCommands, | ||
| } | ||
| } | ||
|
|
||
| // ConnectInitializedDatabase creates a database with the required metadata tables | ||
| func ConnectInitializedDatabase(postgresDbConfig *postgresdb.Config) (*sql.DB, error) { | ||
| dbCreated, e := postgresdb.CreateDatabaseIfNotExists(postgresDbConfig) | ||
| if e != nil { | ||
| if strings.Contains(e.Error(), "connect: connection refused") { | ||
| utils.PrintErrorHintf("ensure your postgres database is available on %s:%d, or remove the 'POSTGRES_DB' config from your trader config file\n", postgresDbConfig.GetHost(), postgresDbConfig.GetPort()) | ||
| } | ||
| return nil, fmt.Errorf("error when creating database from config (%+v), created=%v: %s", *postgresDbConfig, dbCreated, e) | ||
| } | ||
| if dbCreated { | ||
| log.Printf("created database '%s'", postgresDbConfig.GetDbName()) | ||
| } else { | ||
| log.Printf("did not create db '%s' because it already exists", postgresDbConfig.GetDbName()) | ||
| } | ||
|
|
||
| db, e := sql.Open("postgres", postgresDbConfig.MakeConnectString()) | ||
| if e != nil { | ||
| return nil, fmt.Errorf("could not open database: %s", e) | ||
| } | ||
| // don't defer db.Close() here becuase we want it open for the life of the application for now | ||
|
|
||
| log.Printf("creating db schema and running upgrade scripts ...\n") | ||
| e = runUpgradeScripts(db, upgradeScripts) | ||
| if e != nil { | ||
| return nil, fmt.Errorf("could not run upgrade scripts: %s", e) | ||
| } | ||
| log.Printf("... finished creating db schema and running upgrade scripts\n") | ||
|
|
||
| return db, nil | ||
| } | ||
|
|
||
| func runUpgradeScripts(db *sql.DB, scripts []*UpgradeScript) error { | ||
| currentDbVersion, e := QueryDbVersion(db) | ||
| if e != nil { | ||
| if !strings.Contains(e.Error(), "relation \"db_version\" does not exist") { | ||
| return fmt.Errorf("could not fetch current db version: %s", e) | ||
| } | ||
| currentDbVersion = 0 | ||
| } | ||
|
|
||
| for _, script := range scripts { | ||
| if script.version <= currentDbVersion { | ||
| log.Printf(" skipping upgrade script for version %d because current db version (%d) is equal or ahead\n", script.version, currentDbVersion) | ||
| continue | ||
| } | ||
|
|
||
| // start transaction | ||
| e = postgresdb.ExecuteStatement(db, "BEGIN") | ||
| if e != nil { | ||
| return fmt.Errorf("could not start transaction before upgrading db to version %d: %s", script.version, e) | ||
| } | ||
|
|
||
| startTime := time.Now() | ||
| startTimeMillis := startTime.UnixNano() / int64(time.Millisecond) | ||
| for ci, command := range script.commands { | ||
| e = postgresdb.ExecuteStatement(db, command) | ||
| if e != nil { | ||
| return fmt.Errorf("could not execute sql statement at index %d for db version %d (%s): %s", ci, script.version, command, e) | ||
| } | ||
| log.Printf(" executed sql statement at index %d for db version %d", ci, script.version) | ||
| } | ||
| endTimeMillis := time.Now().UnixNano() / int64(time.Millisecond) | ||
| elapsedMillis := endTimeMillis - startTimeMillis | ||
|
|
||
| // add entry to db_version table | ||
| sqlInsertDbVersion := fmt.Sprintf(sqlDbVersionTableInsertTemplate, | ||
| script.version, | ||
| startTime.Format(postgresdb.DateFormatString), | ||
| len(script.commands), | ||
| elapsedMillis, | ||
| ) | ||
| _, e = db.Exec(sqlInsertDbVersion) | ||
| if e != nil { | ||
| // duplicate insert should return an error | ||
| return fmt.Errorf("could not execute sql insert values statement in db_version table for db version %d (%s): %s", script.version, sqlInsertDbVersion, e) | ||
| } | ||
|
|
||
| // commit transaction | ||
| e = postgresdb.ExecuteStatement(db, "COMMIT") | ||
| if e != nil { | ||
| return fmt.Errorf("could not commit transaction before upgrading db to version %d: %s", script.version, e) | ||
| } | ||
| log.Printf(" successfully ran %d upgrade commands and upgraded to version %d of the database in %d milliseconds\n", len(script.commands), script.version, elapsedMillis) | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.