update kline

Author: haipome

accessws/aw_config.c

@@ -7,6 +7,56 @@
 
 struct settings settings;
 
+static int read_depth_merge_cfg(json_t *root, const char *key)
+{
+    json_t *obj = json_object_get(root, key);
+    if (obj == NULL || !json_is_object(obj))
+        return -__LINE__;
+
+    settings.depth_market_count = json_object_size(obj);
+    settings.depth_market_lists = malloc(sizeof(depth_merge_cfg) * settings.depth_market_count);
+
+    char *market;
+    json_t *value;
+    int index = 0;
+    json_object_foreach(obj, market, value) {
+        depth_merge_cfg *cfg = &settings.depth_market_lists[index++];
+        cfg->market = strdup(market);
+        if (!json_is_array(value))
+            return -__LINE__;
+        cfg->count = json_array_size(value);
+        cfg->limit = malloc(sizeof(void *) * cfg->count);
+        for (int i = 0; i < cfg->count; ++i) {
+            const char *str = json_string_value(json_array_get(value, i));
+            if (str == NULL)
+                return -__LINE__;
+            cfg->limit[i] = decimal(str, 0);
+            if (cfg->limit[i] == NULL)
+                return -__LINE__;
+        }
+    }
+
+    return 0;
+}
+
+static int read_depth_limit_cfg(json_t *root, const char *key)
+{
+    json_t *obj = json_object_get(root, key);
+    if (obj == NULL || !json_is_array(obj))
+        return -__LINE__;
+
+    settings.depth_limit.count = json_array_size(obj);
+    settings.depth_limit.limit = malloc(sizeof(int) * settings.depth_limit.count);
+
+    for (int i = 0; i < settings.depth_limit.count; ++i) {
+        settings.depth_limit.limit[i] = json_integer_value(json_array_get(obj, i));
+        if (settings.depth_limit.limit[i] == 0)
+            return -__LINE__;
+    }
+
+    return 0;
+}
+
 static int read_config_from_json(json_t *root)
 {
     int ret;
@@ -72,7 +122,8 @@ static int read_config_from_json(json_t *root)
     ERR_RET(read_cfg_real(root, "kline_interval", &settings.kline_interval, false, 0.5));
     ERR_RET(read_cfg_real(root, "depth_interval", &settings.depth_interval, false, 0.5));
 
-    ERR_RET(read_cfg_int(root, "depth_subscribe_max", &settings.depth_subscribe_max, false, 100));
+    ERR_RET(read_depth_merge_cfg(root, "depth_merge"));
+    ERR_RET(read_depth_limit_cfg(root, "depth_limit"));
 
     return 0;
 }

accessws/aw_config.h

@@ -44,6 +44,17 @@
 
 # define AW_LISTENER_BIND   "seqpacket@/tmp/accessws_listener.sock"
 
+typedef struct depth_merge_cfg {
+    char    *market;
+    int     count;
+    mpd_t   **limit;
+} depth_merge_cfg;
+
+typedef struct depth_limit_cfg {
+    int     count;
+    int     *limit;
+} depth_limit_cfg;
+
 struct settings {
     process_cfg         process;
     log_cfg             log;
@@ -67,7 +78,9 @@ struct settings {
     double              kline_interval;
     double              depth_interval;
 
-    int                 depth_subscribe_max;
+    int                 depth_market_count;
+    depth_merge_cfg     *depth_market_lists;
+    depth_limit_cfg     depth_limit;
 };
 
 extern struct settings settings;

accessws/aw_depth.c

@@ -387,8 +387,46 @@ int init_depth(void)
     return 0;
 }
 
+static bool is_good_limit(int limit)
+{
+    for (int i = 0; i < settings.depth_limit.count; ++i) {
+        if (settings.depth_limit.limit[i] == limit) {
+            return true;
+        }
+    }
+
+    return false;
+}
+
+static bool is_good_interval(const char *market, const char *interval)
+{
+    mpd_t *merge = decimal(interval, 0);
+    if (merge == NULL)
+        return false;
+
+    for (int i = 0; i < settings.depth_market_count; ++i) {
+        depth_merge_cfg *cfg = &settings.depth_market_lists[i];
+        if (strcmp(cfg->market, market) != 0)
+            continue;
+        for (int j = 0; j < cfg->count; ++j) {
+            if (mpd_cmp(cfg->limit[j], merge, &mpd_ctx) == 0) {
+                mpd_del(merge);
+                return true;
+            }
+        }
+    }
+
+    mpd_del(merge);
+    return false;
+}
+
 int depth_subscribe(nw_ses *ses, const char *market, uint32_t limit, const char *interval)
 {
+    if (!is_good_limit(limit))
+        return -1;
+    if (!is_good_interval(market, interval))
+        return -1;
+
     struct depth_key key;
     memset(&key, 0, sizeof(key));
     strncpy(key.market, market, MARKET_NAME_MAX_LEN - 1);
@@ -397,9 +435,6 @@ int depth_subscribe(nw_ses *ses, const char *market, uint32_t limit, const char
 
     dict_entry *entry = dict_find(dict_depth, &key);
     if (entry == NULL) {
-        if (dict_depth->used >= settings.depth_subscribe_max)
-            return -__LINE__;
-
         struct depth_val val;
         memset(&val, 0, sizeof(val));
 

accessws/aw_server.c

@@ -150,7 +150,9 @@ static int process_cache(nw_ses *ses, uint64_t id, sds key)
         return 0;
     }
 
+    json_incref(cache->result);
     send_result(ses, id, cache->result);
+
     return 1;
 }
 
@@ -267,8 +269,12 @@ static int on_method_depth_subscribe(nw_ses *ses, uint64_t id, struct clt_info *
         return send_error_invalid_argument(ses, id);
 
     depth_unsubscribe(ses);
-    if (depth_subscribe(ses, market, limit, interval) < 0)
+    int ret = depth_subscribe(ses, market, limit, interval);
+    if (ret == -1) {
+        return send_error_invalid_argument(ses, id);
+    } else if (ret < 0) {
         return send_error_internal_error(ses, id);
+    }
 
     send_success(ses, id);
     depth_send_clean(ses, market, limit, interval);

accessws/config.json

@@ -63,5 +63,11 @@
     },
     "backend_timeout": 1.0,
     "cache_timeout": 10.0,
-    "auth_url": "http://192.168.1.6:8000/internal/exchange/user/auth"
+    "auth_url": "http://192.168.1.6:8000/internal/exchange/user/auth",
+    "depth_merge": {
+        "BTCCNY": ["0", "1", "5", "10"],
+        "ETHCNY": ["0", "0.1", "1", "5"],
+        "ETHBTC": ["0", "0.00001", "0.0001", "0.0005"]
+    },
+    "depth_limit": [1, 5, 10, 20, 30, 50, 100]
 }

marketprice/mp_config.c

@@ -46,11 +46,12 @@ static int read_config_from_json(json_t *root)
         return -__LINE__;
     }
 
-    read_cfg_int(root, "timezone", &settings.timezone, false, -28800);
     read_cfg_int(root, "sec_max", &settings.sec_max, false, 86400 * 7);
     read_cfg_int(root, "min_max", &settings.min_max, false, 60 * 24 * 365);
     read_cfg_int(root, "hour_max", &settings.hour_max, false, 24 * 365 * 10);
 
+    settings.timezone = get_timezone_offset();
+
     return 0;
 }
 

marketprice/mp_message.c

@@ -355,6 +355,17 @@ static void add_update(struct market_info *info, int type, time_t timestamp)
     dict_add(info->update, &key, NULL);
 }
 
+static time_t get_day_start(time_t timestamp)
+{
+    struct tm *timeinfo = localtime(&timestamp);
+    struct tm dtm;
+    memset(&dtm, 0, sizeof(dtm));
+    dtm.tm_year = timeinfo->tm_year;
+    dtm.tm_mon  = timeinfo->tm_mon;
+    dtm.tm_mday = timeinfo->tm_mday;
+    return mktime(&dtm);
+}
+
 static int market_update(const char *market, double timestamp, mpd_t *price, mpd_t *amount, int side, uint64_t id)
 {
     struct market_info *info = market_query(market);
@@ -415,7 +426,7 @@ static int market_update(const char *market, double timestamp, mpd_t *price, mpd
     add_update(info, KLINE_HOUR, time_hour);
 
     // update day
-    time_t time_day = time_sec / 86400 * 86400 + settings.timezone;
+    time_t time_day = get_day_start(time_sec);
     entry = dict_find(info->day, &time_day);
     if (entry) {
         kinfo = entry->val;
@@ -935,7 +946,7 @@ json_t *get_market_status_today(const char *market)
 
     json_t *result = json_object();
     time_t now = time(NULL);
-    time_t start = now / 86400 * 86400 + settings.timezone;
+    time_t start = get_day_start(now);
     struct kline_info *klast = get_last_kline(info->day, start - 86400, start - 86400 * 30, 86400);
     dict_entry *entry = dict_find(info->day, &start);
     if (entry) {
@@ -1072,11 +1083,10 @@ json_t *get_market_kline_hour(const char *market, time_t start, time_t end, int
     time_t start_min = now / 3600 * 3600 - settings.hour_max * 3600;
     if (start < start_min)
         start = start_min;
-    time_t base = start / 86400 * 86400 - 86400 + settings.timezone;
-    while (base <= start) {
+    time_t base = get_day_start(start);
+    while ((base + interval) <= start)
         base += interval;
-    }
-    start = base - interval;
+    start = base;
 
     struct kline_info *klast = get_last_kline(info->hour, start - 3600, start_min, 3600);
     int step = interval / 3600;
@@ -1112,10 +1122,7 @@ json_t *get_market_kline_day(const char *market, time_t start, time_t end, int i
         return NULL;
 
     json_t *result = json_array();
-    time_t base = start / interval * interval + settings.timezone;
-    if (base > start)
-        base -= interval;
-    start = base;
+    start = (start - settings.timezone) / interval * interval + settings.timezone;
 
     struct kline_info *klast = get_last_kline(info->day, start - 86400, start - 86400 * 30, 86400);
     int step = interval / 86400;
@@ -1151,10 +1158,10 @@ json_t *get_market_kline_week(const char *market, time_t start, time_t end, int
         return NULL;
 
     json_t *result = json_array();
-    time_t base = start / interval * interval - 3 * 86400 + settings.timezone;
-    while (base <= start)
+    time_t base = (start - settings.timezone) / interval * interval - 3 * 86400 + settings.timezone;
+    while ((base + interval) <= start)
         base += interval;
-    start = base - interval;
+    start = base;
 
     struct kline_info *klast = get_last_kline(info->day, start - 86400, start - 86400 * 30, 86400);
     int step = interval / 86400;

matchengine/me_cli.c

@@ -9,9 +9,22 @@
 # include "me_market.h"
 # include "me_trade.h"
 # include "me_persist.h"
+# include "me_operlog.h"
+# include "me_history.h"
+# include "me_message.h"
 
 static cli_svr *svr;
 
+static sds on_cmd_status(const char *cmd, int argc, sds *argv)
+{
+    sds reply = sdsempty();
+    reply = market_status(reply);
+    reply = operlog_status(reply);
+    reply = history_status(reply);
+    reply = message_status(reply);
+    return reply;
+}
+
 static sds on_cmd_balance_list(const char *cmd, int argc, sds *argv)
 {
     sds reply = sdsempty();
@@ -169,6 +182,7 @@ int init_cli(void)
         return -__LINE__;
     }
 
+    cli_svr_add_cmd(svr, "status", on_cmd_status);
     cli_svr_add_cmd(svr, "balance", on_cmd_balance);
     cli_svr_add_cmd(svr, "market",  on_cmd_market);
     cli_svr_add_cmd(svr, "makeslice", on_cmd_makeslice);

matchengine/me_history.c

@@ -322,3 +322,8 @@ bool is_history_block(void)
     return false;
 }
 
+sds history_status(sds reply)
+{
+    return sdscatprintf(reply, "history pending %d\n", job->request_count);
+}
+

matchengine/me_history.h

@@ -18,6 +18,7 @@ int append_order_deal_history(double t, uint64_t ask, int ask_role, uint64_t bid
 int append_user_balance_history(double t, uint32_t user_id, const char *asset, const char *business, mpd_t *change, const char *detail);
 
 bool is_history_block(void);
+sds history_status(sds reply);
 
 # endif