Added real-time balance time series analytics endpoint.

finnow-api/source/analytics/api.d

@@ -1,17 +1,16 @@
 module analytics.api;
 
 import handy_http_primitives;
-import handy_http_handlers.path_handler : PathMapping;
+import handy_http_handlers.path_handler : PathMapping, GetMapping;
+import handy_http_data : writeJsonBody;
+import std.datetime;
 
 import profile.data;
 import profile.service;
 import profile.api : PROFILE_PATH;
-
+import analytics.balances;
-@PathMapping(HttpMethod.GET, PROFILE_PATH ~ "/analytics/balance-time-series")
+import util.money;
-void handleGetBalanceTimeSeries(ref ServerHttpRequest request, ref ServerHttpResponse response) {
+import util.data;
-    auto ds = getProfileDataSource(request);
-    serveJsonFromProperty(response, ds, "analytics.balanceTimeSeries");
-}
 
 @PathMapping(HttpMethod.GET, PROFILE_PATH ~ "/analytics/category-spend-time-series")
 void handleGetCategorySpendTimeSeries(ref ServerHttpRequest request, ref ServerHttpResponse response) {
@@ -19,6 +18,15 @@ void handleGetCategorySpendTimeSeries(ref ServerHttpRequest request, ref ServerH
     serveJsonFromProperty(response, ds, "analytics.categorySpendTimeSeries");
 }
 
+@GetMapping(PROFILE_PATH ~ "/analytics/balance-time-series")
+void handleGetBalanceTimeSeriesV2(ref ServerHttpRequest request, ref ServerHttpResponse response) {
+    auto ds = getProfileDataSource(request);
+    Currency currency = Currency.ofCode(request.getParamAs!string("currency", Currencies.USD.code));
+    TimeRange timeRange = TimeRange(Optional!(SysTime).empty(), Optional!(SysTime).empty());
+    auto data = computeBalanceTimeSeriesV2(ds, currency, timeRange);
+    writeJsonBody(response, data);
+}
+
 /**
  * Helper method to serve JSON analytics data to a client by fetching it
  * directly from the user's profile properties table and writing it to the

finnow-api/source/analytics/balances.d

@@ -5,6 +5,9 @@ import std.datetime;
 import std.stdio;
 import std.path;
 import std.file;
+import std.algorithm;
+import std.array;
+import std.conv;
 import slf4d;
 import asdf;
 
@@ -17,6 +20,9 @@ import analytics.util;
 import transaction.model;
 import transaction.dto;
 import util.pagination;
+import util.money;
+import util.data;
+import analytics.data;
 
 struct TimeSeriesPoint {
     /// The millisecond UTC timestamp.
@@ -25,63 +31,129 @@ struct TimeSeriesPoint {
     long y;
 }
 
-alias CurrencyGroupedTimeSeries = TimeSeriesPoint[][string];
-
 struct AccountBalanceData {
     ulong accountId;
-    string currencyCode;
     TimeSeriesPoint[] data;
 }
 
 struct BalanceTimeSeriesAnalytics {
     AccountBalanceData[] accounts;
-    CurrencyGroupedTimeSeries totals;
+    TimeSeriesPoint[] totals;
 }
 
-void computeAccountBalanceTimeSeries(Profile profile, ProfileRepository profileRepo) {
+/**
-    ProfileDataSource ds = profileRepo.getDataSource(profile);
+ * Computes a time series tracking the balance of each account (and total of
-    Account[] accounts = ds.getAccountRepository().findAll();
+ * all accounts) over the given time range.
+ * Params:
+ *   ds = The profile data source.
+ *   currency = The currency to report data for.
+ *   timeRange = The time range to generate the time series for.
+ * Returns: An analytics response containing a "totals" time series, as well
+ * as a time series for each known account in the given time range.
+ */
+BalanceTimeSeriesAnalytics computeBalanceTimeSeriesV2(
+    ProfileDataSource ds,
+    in Currency currency,
+    in TimeRange timeRange
+) {
+    SysTime[] timestamps = generateTimeSeriesTimestamps(days(1), timeRange);
+    AnalyticsRepository repo = ds.getAnalyticsRepository();
+    JournalEntryStub[] journalEntries = repo.getJournalEntries(currency, timeRange);
+    BalanceRecordStub[] balanceRecords = repo.getBalanceRecords(currency, timeRange);
+    auto accountIds = balanceRecords.map!(br => br.accountId).uniq.array.sort;
 
-    // Initialize the data structure that'll store the analytics info.
+    BalanceTimeSeriesAnalytics result;
-    BalanceTimeSeriesAnalytics data;
+
-    foreach (account; accounts) {
+    foreach (timestamp; timestamps) {
-        AccountBalanceData accountData;
+        long totalBalance = 0;
-        accountData.accountId = account.id;
+        foreach (accountId; accountIds) {
-        accountData.currencyCode = account.currency.code.idup;
+            Optional!long optionalBalance = deriveBalance(accountId, journalEntries, balanceRecords, timestamp);
-        data.accounts ~= accountData;
+            if (!optionalBalance.isNull) {
+                TimeSeriesPoint p = TimeSeriesPoint(timestamp.toUnixMillis(), optionalBalance.value);
+                bool isAccountDataPresent = false;
+                foreach (ref accountData; result.accounts) {
+                    if (accountData.accountId == accountId) {
+                        accountData.data ~= p;
+                        isAccountDataPresent = true;
+                        break;
+                    }
+                }
+                if (!isAccountDataPresent) {
+                    result.accounts ~= AccountBalanceData(accountId, [p]);
+                }
+                totalBalance += optionalBalance.value;
+            }
+        }
+        result.totals ~= TimeSeriesPoint(timestamp.toUnixMillis(), totalBalance);
     }
 
-    foreach (timestamp; generateTimeSeriesTimestamps(days(1), 365)) {
+    return result;
-        // Compute the balance of each account at this timestamp.
-        foreach (idx, account; accounts) {
-            auto balance = getBalance(ds, account.id, timestamp);
-            if (!balance.isNull) {
-                data.accounts[idx].data ~= TimeSeriesPoint(
-                    timestamp.toUnixMillis(),
-                    balance.value
-                );
-            }
 }
 
-        // Compute total balances for this timestamp.
+private Optional!long deriveBalance(
-        auto totalBalances = getTotalBalanceForAllAccounts(ds, timestamp);
+    ulong accountId,
-        foreach (CurrencyBalance bal; totalBalances) {
+    in JournalEntryStub[] journalEntries,
-            data.totals[bal.currency.code.idup] ~= TimeSeriesPoint(timestamp.toUnixMillis(), bal.balance);
+    in BalanceRecordStub[] balanceRecords,
+    in SysTime timestamp
+) {
+    import core.time : abs;
+    Optional!BalanceRecordStub nearestBalanceRecord;
+    foreach (br; balanceRecords) {
+        if (br.accountId == accountId) {
+            if (nearestBalanceRecord.isNull) {
+                nearestBalanceRecord = Optional!(BalanceRecordStub).of(br);
+            } else {
+                Duration currentDiff = abs(nearestBalanceRecord.value.timestamp - timestamp);
+                Duration newDiff = abs(br.timestamp - timestamp);
+                if (newDiff < currentDiff) {
+                    nearestBalanceRecord = Optional!(BalanceRecordStub).of(br);
                 }
             }
+        }
+    }
+    if (nearestBalanceRecord.isNull) {
+        return Optional!(long).empty();
+    }
+    if (timestamp == nearestBalanceRecord.value.timestamp) {
+        return Optional!(long).of(nearestBalanceRecord.value.value);
+    }
 
-    ds.doTransaction(() {
+    // Now that we have a balance record, work our way towards the desired
-        ds.getPropertiesRepository().deleteProperty("analytics.balanceTimeSeries");
+    // timestamp, applying journal entry changes.
-        ds.getPropertiesRepository().setProperty(
+    SysTime startTimestamp;
-            "analytics.balanceTimeSeries",
+    SysTime endTimestamp;
-            serializeToJsonPretty(data)
+    long balance = nearestBalanceRecord.value.value;
-        );
+    if (timestamp > nearestBalanceRecord.value.timestamp) {
-    });
+        startTimestamp = nearestBalanceRecord.value.timestamp;
-    infoF!"Computed account balance analytics for user %s, profile %s."(
+        endTimestamp = timestamp;
-        profile.username,
+    } else {
-        profile.name
+        startTimestamp = timestamp;
-    );
+        endTimestamp = nearestBalanceRecord.value.timestamp;
     }
+    auto relevantJournalEntries = journalEntries
+        .filter!(je => (
+            je.accountId == accountId &&
+            je.timestamp >= startTimestamp &&
+            je.timestamp <= endTimestamp
+        ));
+    foreach (je; relevantJournalEntries) {
+        long entryValue = je.amount;
+        if (je.type == AccountJournalEntryType.CREDIT) {
+            entryValue *= -1;
+        }
+        if (!je.accountType.debitsPositive) {
+            entryValue *= -1;
+        }
+        if (je.timestamp < nearestBalanceRecord.value.timestamp) {
+            entryValue *= -1;
+        }
+        balance += entryValue;
+    }
+    return Optional!(long).of(balance);
+}
+
+
+alias CurrencyGroupedTimeSeries = TimeSeriesPoint[][string];
 
 struct CategorySpendData {
     ulong categoryId;

finnow-api/source/analytics/data.d

@@ -0,0 +1,37 @@
+module analytics.data;
+
+import std.datetime;
+
+import util.money;
+import util.data;
+import analytics.balances;
+import account.model;
+
+struct JournalEntryStub {
+    SysTime timestamp;
+    ulong accountId;
+    AccountType accountType;
+    ulong amount;
+    AccountJournalEntryType type;
+}
+
+struct BalanceRecordStub {
+    SysTime timestamp;
+    ulong accountId;
+    long value;
+}
+
+/**
+ * Repository that provides various functions for fetching data that's used in
+ * the calculation of analytics, separate from usual app functionality.
+ */
+interface AnalyticsRepository {
+    JournalEntryStub[] getJournalEntries(
+        in Currency currency,
+        in TimeRange timeRange
+    );
+    BalanceRecordStub[] getBalanceRecords(
+        in Currency currency,
+        in TimeRange timeRange
+    );
+}

finnow-api/source/analytics/data_impl_sqlite.d

@@ -0,0 +1,104 @@
+module analytics.data_impl_sqlite;
+
+import d2sqlite3;
+import std.array;
+import std.datetime;
+
+import util.money;
+import util.data;
+import util.sqlite;
+import account.model : AccountJournalEntryType, AccountType;
+import analytics.balances;
+import analytics.data;
+
+class SqliteAnalyticsRepository : AnalyticsRepository {
+    private Database db;
+    this(Database db) {
+        this.db = db;
+    }
+
+    JournalEntryStub[] getJournalEntries(
+        in Currency currency,
+        in TimeRange timeRange
+    ) {
+        QueryBuilder qb = QueryBuilder("account_journal_entry je")
+            .select("je.timestamp,je.account_id,je.amount,je.type,account.type")
+            .join("LEFT JOIN account ON account.id = je.account_id")
+            .where("UPPER(je.currency) = ?")
+            .withArgBinding((ref stmt, ref idx) {
+                stmt.bind(idx++, currency.codeString);
+            });
+        if (timeRange.fromTime) {
+            qb.where("je.timestamp >= ?")
+                .withArgBinding((ref stmt, ref idx) {
+                    stmt.bind(idx++, timeRange.fromTime.value);
+                });
+        }
+        if (timeRange.toTime) {
+            qb.where("je.timestamp <= ?")
+                .withArgBinding((ref stmt, ref idx) {
+                    stmt.bind(idx++, timeRange.fromTime.value);
+                });
+        }
+        string query = qb.build() ~ " ORDER BY je.timestamp";
+        Statement stmt = db.prepare(query);
+        qb.applyArgBindings(stmt);
+        ResultRange result = stmt.execute();
+        Appender!(JournalEntryStub[]) app;
+        foreach (row; result) {
+            auto journalEntryTypeStr = row.peek!(string, PeekMode.slice)(3);
+            AccountJournalEntryType type;
+            if (journalEntryTypeStr == AccountJournalEntryType.CREDIT) {
+                type = AccountJournalEntryType.CREDIT;
+            } else {
+                type = AccountJournalEntryType.DEBIT;
+            }
+            app ~= JournalEntryStub(
+                SysTime.fromISOExtString(row.peek!(string, PeekMode.slice)(0)),
+                row.peek!ulong(1),
+                AccountType.fromId(row.peek!(string, PeekMode.slice)(4)),
+                row.peek!ulong(2),
+                type
+            );
+        }
+        return app[];
+    }
+
+    BalanceRecordStub[] getBalanceRecords(
+        in Currency currency,
+        in TimeRange timeRange
+    ) {
+        QueryBuilder qb = QueryBuilder("account_value_record")
+            .select("timestamp,account_id,value,type")
+            .where("UPPER(currency) = ?")
+            .withArgBinding((ref stmt, ref idx) {
+                stmt.bind(idx++, currency.codeString);
+            })
+            .where("UPPER(type) = 'BALANCE'");
+        if (timeRange.fromTime) {
+            qb.where("timestamp >= ?")
+                .withArgBinding((ref stmt, ref idx) {
+                    stmt.bind(idx++, timeRange.fromTime.value);
+                });
+        }
+        if (timeRange.toTime) {
+            qb.where("timestamp <= ?")
+                .withArgBinding((ref stmt, ref idx) {
+                    stmt.bind(idx++, timeRange.fromTime.value);
+                });
+        }
+        string query = qb.build() ~ " ORDER BY timestamp";
+        Statement stmt = db.prepare(query);
+        qb.applyArgBindings(stmt);
+        ResultRange result = stmt.execute();
+        Appender!(BalanceRecordStub[]) app;
+        foreach (row; result) {
+            app ~= BalanceRecordStub(
+                SysTime.fromISOExtString(row.peek!(string, PeekMode.slice)(0)),
+                row.peek!ulong(1),
+                row.peek!long(2)
+            );
+        }
+        return app[];
+    }
+}

finnow-api/source/analytics/util.d

@@ -1,6 +1,8 @@
 module analytics.util;
 
 import std.datetime;
+import std.array;
+import util.data;
 
 SysTime[] generateTimeSeriesTimestamps(Duration intervalSize, int intervalCount) {
     const SysTime now = Clock.currTime(UTC());
@@ -18,6 +20,34 @@ SysTime[] generateTimeSeriesTimestamps(Duration intervalSize, int intervalCount)
     return timestamps;
 }
 
+SysTime[] generateTimeSeriesTimestamps(Duration intervalSize, in TimeRange timeRange) {
+    SysTime endOfRange;
+    SysTime startOfRange;
+    if (timeRange.toTime.isNull) {
+        endOfRange = Clock.currTime(UTC());
+    } else {
+        endOfRange = timeRange.toTime.value;
+    }
+    if (timeRange.fromTime.isNull) {
+        startOfRange = endOfRange;
+        startOfRange.add!"years"(-5);
+    } else {
+        startOfRange = timeRange.fromTime.value;
+    }
+    import std.stdio;
+    writefln!"start = %s, end = %s"(startOfRange, endOfRange);
+
+    Appender!(SysTime[]) app;
+    app ~= startOfRange;
+
+    SysTime timestamp = startOfRange;
+    while (timestamp + intervalSize <= endOfRange) {
+        timestamp += intervalSize;
+        app ~= timestamp;
+    }
+    return app[];
+}
+
 ulong toUnixMillis(in SysTime ts) {
     return (ts - SysTime(unixTimeToStdTime(0))).total!"msecs";
 }

finnow-api/source/profile/data.d

@@ -40,6 +40,7 @@ interface ProfileDataSource {
     import account.data;
     import transaction.data;
     import attachment.data;
+    import analytics.data;
 
     PropertiesRepository getPropertiesRepository();
     AttachmentRepository getAttachmentRepository();
@@ -53,6 +54,8 @@ interface ProfileDataSource {
     TransactionTagRepository getTransactionTagRepository();
     TransactionRepository getTransactionRepository();
 
+    AnalyticsRepository getAnalyticsRepository();
+
     void doTransaction(void delegate () dg);
 }
 
@@ -61,6 +64,7 @@ version(unittest) {
         import account.data;
         import transaction.data;
         import attachment.data;
+        import analytics.data;
 
         PropertiesRepository getPropertiesRepository() {
             throw new Exception("Not implemented");
@@ -89,6 +93,9 @@ version(unittest) {
         TransactionRepository getTransactionRepository() {
             throw new Exception("Not implemented");
         }
+        AnalyticsRepository getAnalyticsRepository() {
+            throw new Exception("Not implemented");
+        }
         void doTransaction(void delegate () dg) {
             throw new Exception("Not implemented");
         }

finnow-api/source/profile/data_impl_sqlite.d

@@ -181,6 +181,8 @@ class SqliteProfileDataSource : ProfileDataSource {
     import transaction.data_impl_sqlite;
     import attachment.data;
     import attachment.data_impl_sqlite;
+    import analytics.data;
+    import analytics.data_impl_sqlite;
 
     private const string dbPath;
     Database db;
@@ -195,6 +197,7 @@ class SqliteProfileDataSource : ProfileDataSource {
     TransactionCategoryRepository transactionCategoryRepo;
     TransactionTagRepository transactionTagRepo;
     TransactionRepository transactionRepo;
+    AnalyticsRepository analyticsRepo;
 
     this(string path) {
         this.dbPath = path;
@@ -276,6 +279,13 @@ class SqliteProfileDataSource : ProfileDataSource {
         return transactionRepo;
     }
 
+    AnalyticsRepository getAnalyticsRepository() {
+        if (analyticsRepo is null) {
+            analyticsRepo = new SqliteAnalyticsRepository(db);
+        }
+        return analyticsRepo;
+    }
+
     void doTransaction(void delegate () dg) {
         util.sqlite.doTransaction(db, dg);
     }

finnow-api/source/scheduled_jobs.d

@@ -15,7 +15,6 @@ void startScheduledJobs() {
 	JobScheduler jobScheduler = new TaskPoolScheduler();
 	jobScheduler.addJob(() {
 		info("Computing account balance time series analytics for all users...");
-		doForAllUserProfiles(&computeAccountBalanceTimeSeries);
 		doForAllUserProfiles(&computeCategorySpendTimeSeries);
 		info("Done computing analytics!");
 	}, analyticsSchedule);

finnow-api/source/util/data.d

@@ -3,6 +3,7 @@ module util.data;
 import handy_http_primitives;
 import handy_http_data.multipart;
 import std.typecons;
+import std.datetime;
 
 Optional!T toOptional(T)(Nullable!T value) {
     if (value.isNull) {
@@ -125,3 +126,13 @@ private MultipartFile parseMultipartFile(in MultipartElement e) {
         cast(ubyte[]) e.content
     );
 }
+
+/**
+ * Struct representing a user-provided time range with optional "from" and "to"
+ * timestamps. If both are empty, it is assumed that the user is requesting all
+ * data.
+ */
+struct TimeRange {
+    Optional!SysTime fromTime;
+    Optional!SysTime toTime;
+}

finnow-api/source/util/money.d

@@ -27,6 +27,11 @@ struct Currency {
         }
         throw new Exception("Unknown currency code: " ~ code);
     }
+
+    string codeString() const {
+        import std.conv : to;
+        return code.to!string;
+    }
 }
 
 /**