[core][feat] TimeSeries: allow defining a factor for computing the average (#2126)

Author: aquamatthias

fixcore/fixcore/cli/command.py

@@ -6186,9 +6186,9 @@ def key_fn(node: Json) -> Union[str, Tuple[str, str]]:
 class TimeSeriesCommand(CLICommand):
     """
     ```
-    timeseries snapshot --name <time series> <aggregate search>
+    timeseries snapshot --name <time series> [--avg-factor <factor>] <aggregate search>
     timeseries get --name <time series> --start <time> --end <time> --granularity <duration|integer>
-                   --group <group> --filter <var><op><value>
+                   --group <group> --filter <var><op><value> --avg-factor <factor>
     timeseries list
     timeseries downsample
     ```
@@ -6219,7 +6219,8 @@ class TimeSeriesCommand(CLICommand):
     - `--group` - The variable(s) to select in the group. Can be repeated multiple times.
     - `--filter` - The variable(s) to filter. Can be repeated multiple times.
     - `--granularity` - The granularity of the time series. Can be a duration or an integer.
-
+    - `--avg-factor` - Factor used to divide values to compute the average.
+                       This is required for very big numbers to avoid overflow.
 
     ## Examples
 
@@ -6250,6 +6251,7 @@ def args_info(self) -> ArgsInfo:
             "list": [],
             "snapshot": [
                 ArgInfo("--name", help_text="<time series>."),
+                ArgInfo("--avg-factor", help_text="<avg factor>."),
                 ArgInfo(expects_value=True, value_hint="search"),
             ],
             "get": [
@@ -6260,6 +6262,7 @@ def args_info(self) -> ArgsInfo:
                 ArgInfo("--filter", expects_value=True, can_occur_multiple_times=True, help_text="<var><op><value>"),
                 ArgInfo("--granularity", expects_value=True, help_text="<duration|integer>"),
                 ArgInfo("--aggregation", expects_value=True, possible_values=["avg", "sum", "min", "max"]),
+                ArgInfo("--avg-factor", help_text="<avg factor>."),
             ],
             "downsample": [],
         }
@@ -6268,13 +6271,16 @@ def parse(self, arg: Optional[str] = None, ctx: CLIContext = EmptyContext, **kwa
         async def snapshot_time_series(part: str) -> AsyncIterator[str]:
             parser = NoExitArgumentParser()
             parser.add_argument("--name", type=str, required=True)
+            parser.add_argument("--avg-factor", type=int)
             parsed, rest = parser.parse_known_args(args_parts_parser.parse(part))
             graph_name = ctx.graph_name
             graph_db = self.dependencies.db_access.get_graph_db(graph_name)
             model = await self.dependencies.model_handler.load_model(graph_name)
             query = await self.dependencies.template_expander.parse_query(" ".join(rest), ctx.section, env=ctx.env)
             query_model = QueryModel(query, model, ctx.env)
-            res = await self.dependencies.db_access.time_series_db.add_entries(parsed.name, query_model, graph_db)
+            res = await self.dependencies.db_access.time_series_db.add_entries(
+                parsed.name, query_model, graph_db, avg_factor=parsed.avg_factor
+            )
             yield f"{res} entries added to time series {parsed.name}."
 
         async def load_time_series(part: str) -> Tuple[CLISourceContext, AsyncIterator[Json]]:
@@ -6292,6 +6298,7 @@ def parse_duration_or_int(s: str) -> Union[int, timedelta]:
             parser.add_argument("--filter", type=predicate_term.parse, nargs="*", default=None)
             parser.add_argument("--granularity", type=parse_duration_or_int, default=5)
             parser.add_argument("--aggregation", choices=["avg", "sum", "min", "max"], default="avg")
+            parser.add_argument("--avg-factor", type=int)
             p = parser.parse_args(args_parts_unquoted_parser.parse(part))
             timeout = if_set(ctx.env.get("search_timeout"), duration)
             cursor = await self.dependencies.db_access.time_series_db.load_time_series(
@@ -6303,6 +6310,7 @@ def parse_duration_or_int(s: str) -> Union[int, timedelta]:
                 granularity=p.granularity,
                 timeout=timeout,
                 aggregation=p.aggregation,
+                avg_factor=p.avg_factor,
             )
             return CLISourceContext(cursor.count(), cursor.full_count(), cursor.stats()), cursor
 

fixcore/fixcore/db/arango_query.py

@@ -871,6 +871,7 @@ def load_time_series(
     group_aggregation: Literal["avg", "sum", "min", "max"] = "avg",
     group_by: Optional[Collection[str]] = None,
     group_filter: Optional[List[Predicate]] = None,
+    avg_factor: Optional[int] = None,
 ) -> Tuple[str, Json]:
     ctx = ArangoQueryContext()
     bv_name = ctx.add_bind_var(time_series)
@@ -887,12 +888,18 @@ def load_time_series(
     time_slot = ctx.next_crs()
     slotter = int(granularity.total_seconds())
     gran = ctx.add_bind_var(slotter)
-    offset = start.timestamp() - ((start.timestamp() // slotter) * slotter)
+    offset = int(start.timestamp() - ((start.timestamp() // slotter) * slotter))
     # slot the time by averaging each single group
     query += f" LET {time_slot} = (FLOOR(d.at / @{gran}) * @{gran}) + @{ctx.add_bind_var(offset)}"
     query += f" COLLECT group_slot={time_slot}, complete_group=d.group"
-    query += " AGGREGATE slot_avg = AVG(d.v)"
-    query += " RETURN {at: group_slot, group: complete_group, v: slot_avg}"
+    if avg_factor:  # Required as long as https://github.com/arangodb/arangodb/issues/21096 is not fixed
+        assert avg_factor > 0, "Given average factor must be greater than 0!"
+        bvf = ctx.add_bind_var(avg_factor)
+        query += f" AGGREGATE slot_avg = AVG(d.v / @{bvf})"
+        query += f" RETURN {{at: group_slot, group: complete_group, v: slot_avg * @{bvf}}}"
+    else:
+        query += " AGGREGATE slot_avg = AVG(d.v)"
+        query += " RETURN {at: group_slot, group: complete_group, v: slot_avg}"
 
     # short circuit: no additional grouping and aggregation is avg
     if group_by is None and group_aggregation == "avg":

fixcore/fixcore/db/timeseriesdb.py

@@ -77,7 +77,15 @@ async def list_time_series(self) -> List[TimeSeriesMeta]:
             for e in all_ts
         ]
 
-    async def add_entries(self, name: str, query_model: QueryModel, graph_db: GraphDB, at: Optional[int] = None) -> int:
+    async def add_entries(
+        self,
+        name: str,
+        query_model: QueryModel,
+        graph_db: GraphDB,
+        *,
+        avg_factor: Optional[int] = None,
+        at: Optional[int] = None,
+    ) -> int:
         query = query_model.query
         model = query_model.model
         assert query.aggregate is not None, "Only aggregate queries are supported for time series."
@@ -97,9 +105,9 @@ async def add_entries(self, name: str, query_model: QueryModel, graph_db: GraphD
                 # update meta information
                 await self.__execute_aql(
                     "UPSERT { _key: @key } "
-                    "INSERT { _key: @key, created_at: DATE_NOW(), last_updated: DATE_NOW(), count: @count } "
-                    f"UPDATE {{ last_updated: DATE_NOW(), count: @count }} IN `{self.names_db}`",
-                    dict(key=name, count=result),
+                    "INSERT { _key: @key, created_at: DATE_NOW(), last_updated: DATE_NOW(), count: @count, factor: @factor } "  # noqa
+                    f"UPDATE {{ last_updated: DATE_NOW(), count: @count, factor: @factor }} IN `{self.names_db}`",
+                    dict(key=name, count=result, factor=avg_factor),
                 )
             return result
 
@@ -116,6 +124,7 @@ async def load_time_series(
         batch_size: Optional[int] = None,
         timeout: Optional[timedelta] = None,
         aggregation: Literal["avg", "sum", "min", "max"] = "avg",
+        avg_factor: Optional[int] = None,
     ) -> AsyncCursor:
         """
         Load time series data.
@@ -131,6 +140,8 @@ async def load_time_series(
         :param batch_size: Optional batch size for the query.
         :param timeout: Timeout for the query to run.
         :param aggregation: The aggregation function to use.
+        :param avg_factor: Apply this avg_factor to all values to compute the average in a slot.
+                       Required for big numbers to avoid overflows.
         :return: A cursor to iterate over the time series data.
         """
         assert start < end, "start must be before end"
@@ -145,10 +156,14 @@ async def load_time_series(
         else:
             grl = duration / 20  # default to 20 datapoints if nothing is specified
 
+        # no avg_factor given, try to get it from the database
+        if avg_factor is None and (nd := await self.db.get(self.names_db, name)) and (fc := nd.get("factor")):
+            avg_factor = int(fc)
+
         grl = max(grl, timedelta(hours=1))
 
         qs, bv = arango_query.load_time_series(
-            self.collection_name, name, start, end, grl, aggregation, group_by, filter_by
+            self.collection_name, name, start, end, grl, aggregation, group_by, filter_by, avg_factor=avg_factor
         )
 
         def result_trafo(js: Json) -> Json:

fixcore/tests/fixcore/db/arango_query_test.py

@@ -367,3 +367,7 @@ def test_load_time_series() -> None:
         "SORT group_slot RETURN {at: group_slot,group: { a: group_a, b: group_b }, v: agg_val}"
     )
     assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": "a", "b4": 3600, "b5": 800}
+    # use avg-factor
+    q, _ = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour, avg_factor=1000)
+    assert "slot_avg = AVG(d.v / @b" in q  # factor divides average
+    assert "v: slot_avg * @b" in q  # factor multiplies result

fixcore/tests/fixcore/db/timeseriesdb_test.py

@@ -49,6 +49,8 @@ async def load_ts(**kwargs: Any) -> List[Json]:
     ## check filter_by is working
     # some_int is the same for all entries: one every hour: 5 entries
     assert len(await load_ts(name="test", start=begin, end=after5h, filter_by=[(P("id").eq("1"))])) == 5
+    # avg factor does not change the result size
+    assert len(await load_ts(name="test", start=begin, end=after5h, filter_by=[(P("id").eq("1"))], avg_factor=10)) == 5
 
 
 async def test_compact_time_series(timeseries_db: TimeSeriesDB, foo_model: Model, filled_graph_db: GraphDB) -> None: