diff --git a/resotocore/resotocore/cli/cli.py b/resotocore/resotocore/cli/cli.py
index c1b4c4616b..6654fe1ac2 100644
--- a/resotocore/resotocore/cli/cli.py
+++ b/resotocore/resotocore/cli/cli.py
@@ -394,7 +394,7 @@ async def parse_query(query_arg: str) -> Query:
             query_part = "all" if query_part.strip() == "" else query_part
             # section expansion is disabled here: it will happen on the final query after all parts have been combined
             return await self.dependencies.template_expander.parse_query(
-                "".join(query_part), None, omit_section_expansion=True, **ctx.env
+                "".join(query_part), None, omit_section_expansion=True, env=ctx.env
             )
 
         query: Query = Query.by(AllTerm())
diff --git a/resotocore/resotocore/cli/command.py b/resotocore/resotocore/cli/command.py
index 87790ed0a1..13f1bcd91c 100644
--- a/resotocore/resotocore/cli/command.py
+++ b/resotocore/resotocore/cli/command.py
@@ -51,7 +51,7 @@
 from attr import evolve
 from attrs import define, field
 from dateutil import parser as date_parser
-from parsy import Parser, string
+from parsy import Parser, string, ParseError
 from resotoclient.models import Model as RCModel, Kind as RCKind
 from resotodatalink import EngineConfig
 from resotodatalink.batch_stream import BatchStream
@@ -140,7 +140,7 @@
     NavigateUntilLeaf,
     IsTerm,
 )
-from resotocore.query.query_parser import parse_query, aggregate_parameter_parser
+from resotocore.query.query_parser import parse_query, aggregate_parameter_parser, predicate_term
 from resotocore.query.template_expander import tpl_props_p
 from resotocore.report import BenchmarkConfigPrefix, ReportSeverity
 from resotocore.report.benchmark_renderer import respond_benchmark_result
@@ -170,6 +170,7 @@
 )
 from resotocore.worker_task_queue import WorkerTask, WorkerTaskName
 from resotolib.core import CLIEnvelope
+from resotolib.durations import parse_duration
 from resotolib.parse_util import (
     double_quoted_or_simple_string_dp,
     space_dp,
@@ -1431,7 +1432,7 @@ def parse_at(x: Optional[str]) -> Optional[datetime]:
         at: Optional[datetime] = parse_at(parsed.get("at", None))
 
         # all templates are expanded at this point, so we can call the parser directly.
-        query = parse_query(rest, **ctx.env)
+        query = parse_query(rest, ctx.env)
 
         async def get_db(at: Optional[datetime], graph_name: GraphName) -> Tuple[GraphDB, GraphName]:
             db_access = self.dependencies.db_access
@@ -3904,7 +3905,7 @@ async def put_template(name: str, template_query: str) -> AsyncIterator[str]:
             # try to render_console the template with dummy values and see if the search can be parsed
             try:
                 rendered_query = self.dependencies.template_expander.render(template_query, defaultdict(lambda: True))
-                parse_query(rendered_query, **ctx.env)
+                parse_query(rendered_query, ctx.env)
             except Exception as ex:
                 raise CLIParseError(f"Given template does not define a valid search: {template_query}") from ex
             await self.dependencies.template_expander.put_template(Template(name, template_query))
@@ -6036,6 +6037,115 @@ def key_fn(node: Json) -> Union[str, Tuple[str, str]]:
             raise AttributeError("Wrong or insufficient arguments. Execute `help db` to get more information.")
 
 
+class TimeSeriesCommand(CLICommand):
+    """
+    ```
+    timeseries snapshot --name <time series> <aggregate search>
+    timeseries get --name <time series> --start <time> --end <time> --granularity <duration|integer>
+                   --group <group> --filter <var><op><value>
+    ```
+
+    The timeseries command can be used to add data to a time series or to retrieve data from a time series.
+
+    To append values to a time series, it's necessary to perform an aggregate search query on a graph.
+    As this graph reflects the present state of affairs,
+    running an aggregate search on it essentially captures the current state into the time series.
+    By regularly taking snapshots of the time series using the same aggregate search query,
+    you can effectively track how things evolve over time.
+
+    To retrieve data from a time series, you need to specify the name of the time series, as well as a time range.
+    It is possible to group the data by a certain variable, and to filter the data by a certain variable.
+
+    ## Parameters
+
+    - `--name` - The name of the time series.
+    - `--start` - The start of the time range.
+    - `--end` - The end of the time range.
+    - `--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.
+
+
+    ## Examples
+
+    ```
+    # compute the number of instances per instance type and store it in a time series
+    > timeseries snapshot --name instances aggregate(instance_type: sum(1)): is(instance)
+
+    # retrieve the number of instances per instance type since yesterday
+    > timeseries get --name instances --start @yesterday@ --end @now@
+    ```
+    """
+
+    @property
+    def name(self) -> str:
+        return "timeseries"
+
+    def info(self) -> str:
+        return "Operations on time series."
+
+    def args_info(self) -> ArgsInfo:
+        return {
+            "snapshot": [
+                ArgInfo("--name", help_text="<time series>."),
+                ArgInfo(expects_value=True, value_hint="search"),
+            ],
+            "get": [
+                ArgInfo("--name", expects_value=True, help_text="<time series>"),
+                ArgInfo("--start", expects_value=True, value_hint="datetime", help_text="<time series>"),
+                ArgInfo("--end", expects_value=True, value_hint="datetime", help_text="<time series>"),
+                ArgInfo("--group", expects_value=True, can_occur_multiple_times=True, help_text="<group>"),
+                ArgInfo("--filter", expects_value=True, can_occur_multiple_times=True, help_text="<var><op><value>"),
+                ArgInfo("--granularity", expects_value=True, help_text="<duration|integer>"),
+            ],
+        }
+
+    def parse(self, arg: Optional[str] = None, ctx: CLIContext = EmptyContext, **kwargs: Any) -> CLIAction:
+        async def snapshot_time_series(part: str) -> AsyncIterator[str]:
+            parser = NoExitArgumentParser()
+            parser.add_argument("--name", type=str, required=True)
+            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)
+            yield f"{res} entries added to time series {parsed.name}."
+
+        async def load_time_series(part: str) -> Tuple[CLISourceContext, AsyncIterator[Json]]:
+            def parse_duration_or_int(s: str) -> Union[int, timedelta]:
+                try:
+                    return parse_duration(s)
+                except ParseError:
+                    return int(s)
+
+            parser = NoExitArgumentParser()
+            parser.add_argument("--name", type=str, required=True)
+            parser.add_argument("--start", type=parse_time_or_delta, required=True)
+            parser.add_argument("--end", type=parse_time_or_delta, required=True)
+            parser.add_argument("--group", type=str, nargs="*", default=None)
+            parser.add_argument("--filter", type=predicate_term.parse, nargs="*", default=None)
+            parser.add_argument("--granularity", type=parse_duration_or_int, default=5)
+            p = parser.parse_args(args_parts_unquoted_parser.parse(part))
+            cursor = await self.dependencies.db_access.time_series_db.load_time_series(
+                p.name, p.start, p.end, group_by=p.group, filter_by=p.filter, granularity=p.granularity
+            )
+            return CLISourceContext(cursor.count(), cursor.full_count()), cursor
+
+        args = re.split("\\s+", arg, maxsplit=1) if arg else []
+        if arg and len(args) == 2 and args[0] == "snapshot":
+            return CLISource.single(
+                partial(snapshot_time_series, args[1].strip()), required_permissions={Permission.read}
+            )
+        elif arg and len(args) == 2 and args[0] == "get":
+            return CLISource(partial(load_time_series, args[1].strip()), required_permissions={Permission.read})
+        else:
+            return CLISource.single(
+                lambda: stream.just(self.rendered_help(ctx)), required_permissions={Permission.read}
+            )
+
+
 def all_commands(d: TenantDependencies) -> List[CLICommand]:
     commands = [
         AggregateCommand(d, "search"),
@@ -6077,6 +6187,7 @@ def all_commands(d: TenantDependencies) -> List[CLICommand]:
         SystemCommand(d, "setup", allowed_in_source_position=True),
         TagCommand(d, "action"),
         TailCommand(d, "search"),
+        TimeSeriesCommand(d, "action", allowed_in_source_position=True),
         UniqCommand(d, "misc"),
         UserCommand(d, "setup", allowed_in_source_position=True),
         WorkflowsCommand(d, "action", allowed_in_source_position=True),
@@ -6114,6 +6225,7 @@ def alias_names() -> Dict[str, str]:
         "job": "jobs",
         "lists": "list",
         "template": "templates",
+        "ts": "timeseries",
         "workflow": "workflows",
         "app": "apps",
         "man": "help",
diff --git a/resotocore/resotocore/cli/model.py b/resotocore/resotocore/cli/model.py
index 77c2241755..980b6aa304 100644
--- a/resotocore/resotocore/cli/model.py
+++ b/resotocore/resotocore/cli/model.py
@@ -115,6 +115,10 @@ class CLIContext:
     def graph_name(self) -> GraphName:
         return GraphName(self.env["graph"])
 
+    @property
+    def section(self) -> str:
+        return self.env.get("section", PathRoot)
+
     @property
     def user_permissions(self) -> Set[Permission]:
         return self.user.permissions if self.user else set()
diff --git a/resotocore/resotocore/core_config.py b/resotocore/resotocore/core_config.py
index 6c33f668a3..1c0c63b0b9 100644
--- a/resotocore/resotocore/core_config.py
+++ b/resotocore/resotocore/core_config.py
@@ -192,6 +192,9 @@ class DatabaseConfig(ConfigObject):
         default=False, metadata={"description": "If the connection should not be verified (default: False)"}
     )
     request_timeout: int = field(default=900, metadata={"description": "Request timeout in seconds (default: 900)"})
+    time_series_ttl: timedelta = field(
+        default=timedelta(days=90), metadata={"description": "Time series TTL (default: 90d)"}
+    )
 
 
 @define(order=True, hash=True, frozen=True)
diff --git a/resotocore/resotocore/db/arango_query.py b/resotocore/resotocore/db/arango_query.py
index ec1dd1b6c6..0f900f40a9 100644
--- a/resotocore/resotocore/db/arango_query.py
+++ b/resotocore/resotocore/db/arango_query.py
@@ -2,8 +2,9 @@
 import logging
 import re
 from collections import defaultdict
+from datetime import datetime, timedelta
 from textwrap import dedent
-from typing import Union, List, Tuple, Any, Optional, Dict, Set, Literal
+from typing import Union, List, Tuple, Any, Optional, Dict, Set, Literal, Collection
 
 from arango.typings import Json
 from attrs import evolve
@@ -82,6 +83,28 @@
 array_marker_in_path_regexp = re.compile(r"\[]|\[[*]](?=[.])")
 
 
+class ArangoQueryContext:
+    def __init__(self) -> None:
+        self.counters: Dict[str, int] = defaultdict(lambda: 0)
+        self.bind_vars: Dict[str, Any] = {}
+
+    def next_counter(self, name: str) -> int:
+        count = self.counters[name]
+        self.counters[name] = count + 1
+        return count
+
+    def next_crs(self, name: str = "m") -> str:
+        return f"{name}{self.next_counter(name)}"
+
+    def next_bind_var_name(self) -> str:
+        return f'b{self.next_counter("bind_vars")}'
+
+    def add_bind_var(self, value: Any) -> str:
+        bvn = self.next_bind_var_name()
+        self.bind_vars[bvn] = value
+        return bvn
+
+
 def to_query(
     db: Any,
     query_model: QueryModel,
@@ -89,15 +112,14 @@ def to_query(
     from_collection: Optional[str] = None,
     id_column: str = "_key",
 ) -> Tuple[str, Json]:
-    count: Dict[str, int] = defaultdict(lambda: 0)
+    ctx = ArangoQueryContext()
     query = query_model.query
-    bind_vars: Json = {}
     start = from_collection or f"`{db.vertex_name}`"
-    cursor, query_str = query_string(db, query, query_model, start, with_edges, bind_vars, count, id_column=id_column)
+    cursor, query_str = query_string(db, query, query_model, start, with_edges, ctx, id_column=id_column)
     last_limit = (
         f" LIMIT {ll.offset}, {ll.length}" if ((ll := query.current_part.limit) and not query.is_aggregate()) else ""
     )
-    return f"""{query_str} FOR result in {cursor}{last_limit} RETURN UNSET(result, {unset_props})""", bind_vars
+    return f"""{query_str} FOR result in {cursor}{last_limit} RETURN UNSET(result, {unset_props})""", ctx.bind_vars
 
 
 def query_string(
@@ -106,8 +128,7 @@ def query_string(
     query_model: QueryModel,
     start_cursor: str,
     with_edges: bool,
-    bind_vars: Json,
-    counters: Dict[str, int],
+    ctx: ArangoQueryContext,
     *,
     outer_merge: Optional[str] = None,
     id_column: str = "_key",
@@ -118,17 +139,6 @@ def query_string(
     # combine merge names from the query as well as the default ancestor merge names
     merge_names: Set[str] = query_model.query.merge_names | ancestor_merges
 
-    def next_counter(name: str) -> int:
-        count = counters[name]
-        counters[name] = count + 1
-        return count
-
-    def next_crs(name: str = "m") -> str:
-        return f"{name}{next_counter(name)}"
-
-    def next_bind_var_name() -> str:
-        return f'b{next_counter("bind_vars")}'
-
     def prop_name_kind(
         path: str, context_path: Optional[str] = None
     ) -> Tuple[str, ResolvedProperty, Optional[str]]:  # prop_name, prop, merge_name
@@ -161,7 +171,7 @@ def aggregate(in_cursor: str, a: Aggregate) -> Tuple[str, str]:
         nested = {name for agg in a.group_by for name in agg.all_names() if array_marker_in_path_regexp.search(name)}
         # If we have a nested array, we need to unfold the array and create a new for loop for each array access.
         if nested:
-            cursor = next_crs("agg")
+            cursor = ctx.next_crs("agg")
             for_loop = f"for {cursor} in {in_cursor}"
             internals = []
             for ag in nested:
@@ -172,14 +182,14 @@ def aggregate(in_cursor: str, a: Aggregate) -> Tuple[str, str]:
                     ar_parts.append(ar)
                     if tuple(ar_parts) in cursor_lookup:
                         continue
-                    nxt_crs = next_crs("pre")
+                    nxt_crs = ctx.next_crs("pre")
                     cursor_lookup[tuple(ar_parts)] = nxt_crs
                     for_loop += f" FOR {nxt_crs} IN APPEND(TO_ARRAY({inner_crsr}.{ar}), {{_internal: true}})"
                     internals.append(f"{nxt_crs}._internal!=true")
                     inner_crsr = nxt_crs
             for_loop += f" FILTER {' OR '.join(internals)}"
         else:
-            cursor = next_crs("agg")
+            cursor = ctx.next_crs("agg")
             for_loop = f"for {cursor} in {in_cursor}"
 
         # the property needs to be accessed from the correct cursor
@@ -229,7 +239,7 @@ def unfold_array_func_term(fn: AggregateFunction) -> Optional[str]:
             if len(ars) == 1:
                 return None
             # array access without a related group variable.
-            res = next_crs("agg_let")
+            res = ctx.next_crs("agg_let")
             nested_function_lookup[fn] = res
             pre = ""
             current = cursor
@@ -240,7 +250,7 @@ def unfold_array_func_term(fn: AggregateFunction) -> Optional[str]:
                 if tcar in cursor_lookup:
                     current = cursor_lookup[tcar]
                     continue
-                nxt_crs = next_crs("inner")
+                nxt_crs = ctx.next_crs("inner")
                 pre += f" FOR {nxt_crs} IN TO_ARRAY({current}.{ar})"
                 current = nxt_crs
             return f"LET {res} = {fn.function}({pre} RETURN {current}.{ars[-1]})"
@@ -279,7 +289,7 @@ def predicate(cursor: str, p: Predicate, context_path: Optional[str] = None) ->
         ars_stmts = []
         prop_name = ars.pop()
         for ar in ars:
-            nxt_crs = next_crs("pre")
+            nxt_crs = ctx.next_crs("pre")
             ars_stmts.append(f"{nxt_crs}._internal!=true")
             # append an internal element to make sure this for loop always yields at least one result
             # this value will be filtered out explicitly later on.
@@ -290,12 +300,12 @@ def predicate(cursor: str, p: Predicate, context_path: Optional[str] = None) ->
             pre += f" FOR {nxt_crs} IN APPEND(TO_ARRAY({cursor}.{ar}), {{_internal: true}})"
             cursor = nxt_crs
 
-        bvn = next_bind_var_name()
+        bvn = ctx.next_bind_var_name()
         op = lgt_ops[p.op] if prop.simple_kind.reverse_order and p.op in lgt_ops else p.op
         if op in ["in", "not in"] and isinstance(p.value, list):
-            bind_vars[bvn] = [prop.kind.coerce(a) for a in p.value]
+            ctx.bind_vars[bvn] = [prop.kind.coerce(a) for a in p.value]
         else:
-            bind_vars[bvn] = prop.kind.coerce(p.value)
+            ctx.bind_vars[bvn] = prop.kind.coerce(p.value)
         var_name = f"{cursor}.{prop_name}"
         if op == "=~":  # use regex_test to do case-insensitive matching
             p_term = f"REGEX_TEST({var_name}, @{bvn}, true)"
@@ -318,7 +328,7 @@ def context_term(cursor: str, aep: ContextTerm, context_path: Optional[str] = No
             if spath[-1] == "":
                 spath = spath[:-1]
             for ar in [a.lstrip(".") for a in spath]:
-                nxt_crs = next_crs("pre")
+                nxt_crs = ctx.next_crs("pre")
                 predicate_statement += f" FOR {nxt_crs} IN TO_ARRAY({path_cursor}.{ar})"
                 path_cursor = nxt_crs
             ps, fs = term(path_cursor, aep.term, context_path)
@@ -334,12 +344,12 @@ def context_term(cursor: str, aep: ContextTerm, context_path: Optional[str] = No
         return predicate_statement, filter_statement
 
     def with_id(cursor: str, t: IdTerm) -> str:
-        bvn = next_bind_var_name()
+        bvn = ctx.next_bind_var_name()
         if len(t.ids) == 1:
-            bind_vars[bvn] = t.ids[0]
+            ctx.bind_vars[bvn] = t.ids[0]
             return f"{cursor}.{id_column} == @{bvn}"
         else:
-            bind_vars[bvn] = t.ids
+            ctx.bind_vars[bvn] = t.ids
             return f"{cursor}.{id_column} in @{bvn}"
 
     def is_term(cursor: str, t: IsTerm) -> str:
@@ -347,8 +357,8 @@ def is_term(cursor: str, t: IsTerm) -> str:
         for kind in t.kinds:
             if kind not in model:
                 raise AttributeError(f"Given kind does not exist: {kind}")
-            bvn = next_bind_var_name()
-            bind_vars[bvn] = kind
+            bvn = ctx.next_bind_var_name()
+            ctx.bind_vars[bvn] = kind
             is_results.append(f"@{bvn} IN {cursor}.kinds")
         is_result = " or ".join(is_results)
         return is_result if len(is_results) == 1 else f"({is_result})"
@@ -357,9 +367,9 @@ def fulltext_term(cursor: str, t: FulltextTerm) -> str:
         # This fulltext filter can not take advantage of the fulltext search index.
         # Instead, we filter the resulting entry to match a regular expression derived from the term.
         # The flat property is used via a regexp search.
-        bvn = next_bind_var_name()
+        bvn = ctx.next_bind_var_name()
         dl = fulltext_delimiter_regexp
-        bind_vars[bvn] = dl.pattern.join(f"{re.escape(w)}" for w in dl.split(t.text))
+        ctx.bind_vars[bvn] = dl.pattern.join(f"{re.escape(w)}" for w in dl.split(t.text))
         return f"REGEX_TEST({cursor}.flat, @{bvn}, true)"
 
     def not_term(cursor: str, t: NotTerm, context_path: Optional[str] = None) -> Tuple[Optional[str], str]:
@@ -374,7 +384,7 @@ def term(cursor: str, ab_term: Term, context_path: Optional[str] = None) -> Tupl
         if isinstance(ab_term, ContextTerm):
             return context_term(cursor, ab_term, context_path)
         elif isinstance(ab_term, FunctionTerm):
-            return None, as_arangodb_function(cursor, bind_vars, ab_term, query_model)
+            return None, as_arangodb_function(cursor, ctx.bind_vars, ab_term, query_model)
         elif isinstance(ab_term, IdTerm):
             return None, with_id(cursor, ab_term)
         elif isinstance(ab_term, IsTerm):
@@ -392,8 +402,8 @@ def term(cursor: str, ab_term: Term, context_path: Optional[str] = None) -> Tupl
             raise AttributeError(f"Do not understand: {ab_term}")
 
     def merge(cursor: str, merge_queries: List[MergeQuery]) -> Tuple[str, str]:  # cursor, query
-        result_cursor = next_crs("merge_result")
-        merge_cursor = next_crs()
+        result_cursor = ctx.next_crs("merge_result")
+        merge_cursor = ctx.next_crs()
         merge_result = f"LET {result_cursor} = (FOR {merge_cursor} in {cursor} "
         merge_parts: Json = {}
 
@@ -404,7 +414,7 @@ def add_merge_query(mq: MergeQuery, part_result: str) -> None:
             assert (
                 f.term == AllTerm() and not f.sort and not f.limit and not f.with_clause and not f.tag
             ), "Merge query needs to start with navigation!"
-            merge_crsr = next_crs("merge_part")
+            merge_crsr = ctx.next_crs("merge_part")
             # make sure the limit only yields one element
             mg_crs, mg_query = query_string(
                 db,
@@ -412,8 +422,7 @@ def add_merge_query(mq: MergeQuery, part_result: str) -> None:
                 query_model,
                 merge_cursor,
                 with_edges,
-                bind_vars,
-                counters,
+                ctx,
                 outer_merge=merge_crsr,
                 id_column=id_column,
             )
@@ -452,7 +461,7 @@ def check_is(t: IsTerm) -> Optional[str]:
             )
 
         for mq_in in merge_queries:
-            part_res = next_crs("part_res")
+            part_res = ctx.next_crs("part_res")
             resolved = is_already_resolved(mq_in.query)
             if resolved:
                 merge_result += f'LET {part_res} = DOCUMENT("{db.vertex_name}", {merge_cursor}.refs.{resolved}_id)'
@@ -476,8 +485,8 @@ def filter_statement(current_cursor: str, part_term: Term, limit: Optional[Limit
             if isinstance(part_term, AllTerm) and limit is None and not p.sort:
                 return current_cursor
             nonlocal query_part, filtered_out
-            crsr = next_crs()
-            filtered_out = next_crs("filter")
+            crsr = ctx.next_crs()
+            filtered_out = ctx.next_crs("filter")
             md = f"NOT_NULL({crsr}.metadata, {{}})"
             limited = f" LIMIT {limit.offset}, {limit.length} " if limit else " "
             pre, term_string = term(crsr, part_term)
@@ -485,9 +494,9 @@ def filter_statement(current_cursor: str, part_term: Term, limit: Optional[Limit
             for_stmt = f"FOR {crsr} in {current_cursor} {pre_string} {term_string}"
             # in case nested properties get unfolded, we need to make the list distinct again
             if pre:
-                nested_distinct = next_crs("nested_distinct")
+                nested_distinct = ctx.next_crs("nested_distinct")
                 for_stmt = f"LET {nested_distinct} = ({for_stmt} RETURN DISTINCT {crsr})"
-                crsr = next_crs()
+                crsr = ctx.next_crs()
                 sort_by = sort(crsr, p.sort) if p.sort else " "
                 for_stmt = f"{for_stmt} FOR {crsr} in {nested_distinct}{sort_by}{limited}"
             else:
@@ -509,17 +518,17 @@ def with_usage(in_crsr: str, usage: WithUsage, term: Term, limit: Optional[Limit
             after_filter_cursor = filter_statement(in_crsr, before_term, limit=limit if after_term.is_all else None)
 
             # add the usage predicates
-            usage_crs = next_crs("with_usage")
-            start = next_bind_var_name()
-            end = next_bind_var_name()
-            start_s = next_bind_var_name()
-            duration = next_bind_var_name()
+            usage_crs = ctx.next_crs("with_usage")
+            start = ctx.next_bind_var_name()
+            end = ctx.next_bind_var_name()
+            start_s = ctx.next_bind_var_name()
+            duration = ctx.next_bind_var_name()
             start_time = usage.start_from_now()
             end_time = usage.end_from_now()
-            bind_vars[start] = start_time.timestamp()
-            bind_vars[end] = end_time.timestamp()
-            bind_vars[start_s] = utc_str(start_time)
-            bind_vars[duration] = duration_str(end_time - start_time)
+            ctx.bind_vars[start] = start_time.timestamp()
+            ctx.bind_vars[end] = end_time.timestamp()
+            ctx.bind_vars[start_s] = utc_str(start_time)
+            ctx.bind_vars[duration] = duration_str(end_time - start_time)
             avgs = []
             merges = []
             for mn in usage.metrics:
@@ -565,7 +574,7 @@ def with_clause(in_crsr: str, clause: WithClause, limit: Optional[Limit]) -> str
             # COLLECT l2_cloud = l3_cloud WITH COUNT INTO counter1
             # FILTER (counter1>=0) //counter is +1 since the node itself is always bypassed
             # RETURN ({cloud: l2_cloud._key, count:counter1})
-            current = next_counter("with_clause")
+            current = ctx.next_counter("with_clause")
 
             def cursor_in(depth: int) -> str:
                 return f"c{current}_{depth}"
@@ -608,7 +617,7 @@ def collect_filter(cl: WithClause, depth: int) -> str:
                 inner = collect_filter(cl.with_clause, depth + 1) if cl.with_clause else ""
                 return inner + f"COLLECT {collects} WITH COUNT INTO counter{depth} {filter_term} "
 
-            out = next_crs()
+            out = ctx.next_crs()
 
             limited = f" LIMIT {limit.offset}, {limit.length} " if limit else " "
             query_part += (
@@ -622,10 +631,10 @@ def collect_filter(cl: WithClause, depth: int) -> str:
 
         def inout(in_crsr: str, start: int, until: int, edge_type: str, direction: str) -> str:
             nonlocal query_part
-            in_c = next_crs("io_in")
-            out = next_crs("io_out")
-            out_crsr = next_crs("io_crs")
-            link = next_crs("io_link")
+            in_c = ctx.next_crs("io_in")
+            out = ctx.next_crs("io_out")
+            out_crsr = ctx.next_crs("io_crs")
+            link = ctx.next_crs("io_link")
             unique = "uniqueEdges: 'path'" if with_edges else "uniqueVertices: 'global'"
             link_str = f", {link}" if with_edges else ""
             dir_bound = "OUTBOUND" if direction == Direction.outbound else "INBOUND"
@@ -664,7 +673,7 @@ def navigation(in_crsr: str, nav: Navigation) -> str:
             if len(all_walks) == 1:
                 return all_walks[0]
             else:
-                nav_crsr = next_crs()
+                nav_crsr = ctx.next_crs()
                 all_walks_combined = ",".join(all_walks)
                 query_part += f"LET {nav_crsr} = UNION_DISTINCT({all_walks_combined})"
                 return nav_crsr
@@ -708,8 +717,8 @@ def ft_term(cursor: str, ab_term: Term) -> str:
             if isinstance(ab_term, NotTerm):
                 return f"NOT ({ft_term(cursor, ab_term.term)})"
             elif isinstance(ab_term, FulltextTerm):
-                bvn = next_bind_var_name()
-                bind_vars[bvn] = ab_term.text
+                bvn = ctx.next_bind_var_name()
+                ctx.bind_vars[bvn] = ab_term.text
                 # the fulltext index is based on the flat property. The full text term is tokenized.
                 return f"PHRASE({cursor}.flat, @{bvn})"
             elif isinstance(ab_term, CombinedTerm):
@@ -721,7 +730,7 @@ def ft_term(cursor: str, ab_term: Term) -> str:
 
         # Since fulltext filtering is handled separately, we replace the remaining filter term in the first part
         query_parts[0] = evolve(query_parts[0], term=filter_term)
-        crs = next_crs()
+        crs = ctx.next_crs()
         doc = f"search_{db.vertex_name}"
         ftt = ft_term("ft", ft_part)
         q = f"LET {crs}=(FOR ft in {doc} SEARCH ANALYZER({ftt}, 'delimited') SORT BM25(ft) DESC RETURN ft)"
@@ -737,7 +746,7 @@ def ft_term(cursor: str, ab_term: Term) -> str:
 
     query_str = fulltext_part + " ".join(p[3] for p in parts)
     resulting_cursor = crsr
-    nxt = next_crs()
+    nxt = ctx.next_crs()
     if query.aggregate:  # return aggregate
         resulting_cursor, aggregation = aggregate(resulting_cursor, query.aggregate)
         query_str += aggregation
@@ -765,24 +774,14 @@ def possible_values(
     skip: Optional[int] = None,
 ) -> Tuple[str, Json]:
     path = path_or_predicate if isinstance(path_or_predicate, str) else path_or_predicate.name
-    counters: Dict[str, int] = defaultdict(lambda: 0)
-
-    def next_counter(name: str) -> int:
-        count = counters[name]
-        counters[name] = count + 1
-        return count
-
-    def next_crs(name: str = "m") -> str:
-        return f"{name}{next_counter(name)}"
-
-    bind_vars: Json = {}
     start = f"`{db.vertex_name}`"
-    cursor, query_str = query_string(db, query.query, query, start, False, bind_vars, counters, id_column="_key")
+    ctx = ArangoQueryContext()
+    cursor, query_str = query_string(db, query.query, query, start, False, ctx, id_column="_key")
 
     # iterate over the result
-    let_cursor = next_crs()
+    let_cursor = ctx.next_crs()
     query_str += f" LET {let_cursor} = ("
-    next_cursor = next_crs()
+    next_cursor = ctx.next_crs()
     query_str += f" FOR {next_cursor} in {cursor}"
     cursor = next_cursor
 
@@ -790,14 +789,14 @@ def next_crs(name: str = "m") -> str:
     ars = [a.lstrip(".") for a in array_marker_in_path_regexp.split(path)]
     prop_name = None if path.endswith("[]") or path.endswith("[*]") else ars.pop()
     for ar in ars:
-        nxt_crs = next_crs()
+        nxt_crs = ctx.next_crs()
         query_str += f" FOR {nxt_crs} IN TO_ARRAY({cursor}.{ar})"
         cursor = nxt_crs
     access_path = f"{cursor}.{prop_name}" if prop_name is not None else cursor
 
     # access the detail
     if detail == "attributes":
-        cursor = next_crs()
+        cursor = ctx.next_crs()
         query_str += (
             f" FILTER IS_OBJECT({access_path}) FOR {cursor} IN ATTRIBUTES({access_path}, true) RETURN {cursor})"
         )
@@ -807,29 +806,92 @@ def next_crs(name: str = "m") -> str:
         raise AttributeError(f"Unknown detail: {detail}")
 
     # result stream of matching entries: filter and sort
-    sorted_let = next_crs()
-    next_cursor = next_crs()
+    sorted_let = ctx.next_crs()
+    next_cursor = ctx.next_crs()
     query_str += f" LET {sorted_let} = (FOR {next_cursor} IN {let_cursor} FILTER {next_cursor}!=null"
     cursor = next_cursor
     if isinstance(path_or_predicate, Predicate):
         p: Predicate = path_or_predicate
-        bvn = f'b{next_counter("bind_vars")}'
+        bvn = f'b{ctx.next_counter("bind_vars")}'
         prop = query.model.property_by_path(Section.without_section(p.name))
         pk = prop.kind
         op = lgt_ops[p.op] if prop.simple_kind.reverse_order and p.op in lgt_ops else p.op
-        bind_vars[bvn] = [pk.coerce(a) for a in p.value] if isinstance(p.value, list) else pk.coerce(p.value)
+        ctx.bind_vars[bvn] = [pk.coerce(a) for a in p.value] if isinstance(p.value, list) else pk.coerce(p.value)
         if op == "=~":  # use regex_test to do case-insensitive matching
             query_str += f" FILTER REGEX_TEST({cursor}, @{bvn}, true)"
         else:
             query_str += f" FILTER {cursor} {op} @{bvn}"
     query_str += f" RETURN DISTINCT {cursor})"
     cursor = sorted_let
-    next_cursor = next_crs()
+    next_cursor = ctx.next_crs()
     query_str += f"FOR {next_cursor} IN {cursor} SORT {next_cursor} ASC"
     if limit:
         query_str += f" LIMIT {skip if skip else 0}, {limit}"
     query_str += f" RETURN {next_cursor}"
-    return query_str, bind_vars
+    return query_str, ctx.bind_vars
+
+
+def create_time_series(
+    query_model: QueryModel, db: Any, time_series_collection: str, time_series: str, at: int
+) -> Tuple[str, Json]:
+    query = query_model.query
+    ctx = ArangoQueryContext()
+    start = f"`{db.vertex_name}`"
+    cursor, query_str = query_string(db, query, query_model, start, False, ctx)
+    next_crs = ctx.next_crs()
+    at_bvn = ctx.add_bind_var(at)
+    ts_bvn = ctx.add_bind_var(time_series)
+    insert = (
+        query_str + f" for {next_crs} in {cursor} insert MERGE({next_crs}, {{at:@{at_bvn}, ts:@{ts_bvn}}})"
+        f" into `{time_series_collection}` collect with count into length return length"
+    )
+    return insert, ctx.bind_vars
+
+
+def load_time_series(
+    time_series_collection: str,
+    time_series: str,
+    start: datetime,
+    end: datetime,
+    granularity: timedelta,
+    group_by: Optional[Collection[str]] = None,
+    group_filter: Optional[List[Predicate]] = None,
+) -> Tuple[str, Json]:
+    ctx = ArangoQueryContext()
+    bv_name = ctx.add_bind_var(time_series)
+    bv_start = ctx.add_bind_var(int(start.timestamp()))
+    bv_end = ctx.add_bind_var(int(end.timestamp()))
+
+    query = f"FOR d in `{time_series_collection}` FILTER d.ts==@{bv_name} AND d.at>=@{bv_start} AND d.at<=@{bv_end}"
+    if group_filter:
+        parts = []
+        for f in group_filter:
+            bv = ctx.add_bind_var(f.value)
+            parts.append(f"d.group.{f.name}{f.op}@{bv}")
+        query += f" FILTER {' AND '.join(parts)}"
+    time_slot = ctx.next_crs()
+    slotter = int(granularity.total_seconds())
+    gran = ctx.add_bind_var(slotter)
+    offset = start.timestamp() - ((start.timestamp() // slotter) * slotter)
+    query += f" LET {time_slot} = (FLOOR(d.at / @{gran}) * @{gran}) + @{ctx.add_bind_var(offset)}"
+    # create the groups to collect
+    collect = [f"group_slot={time_slot}"]
+    group = ""
+    if group_by is None:
+        collect.append("complete_group=d.group")
+        group = "group: complete_group,"
+    elif len(group_by) == 0:
+        pass  # no other groups
+    else:
+        parts = []
+        for g in group_by:
+            collect.append(f"group_{g}=d.group.{g}")
+            parts.append(f"{g}: group_{g}")
+        group = f"group: {{ {', '.join(parts)} }},"
+
+    query += f" COLLECT {', '.join(collect)} INTO group"
+    query += f" SORT group_slot RETURN {{at: group_slot, {group} v: AVG(group[*].d.v)}}"
+    return query, ctx.bind_vars
 
 
 async def query_cost(graph_db: Any, model: QueryModel, with_edges: bool) -> EstimatedSearchCost:
diff --git a/resotocore/resotocore/db/async_arangodb.py b/resotocore/resotocore/db/async_arangodb.py
index 0f219fb0ee..2916c51557 100644
--- a/resotocore/resotocore/db/async_arangodb.py
+++ b/resotocore/resotocore/db/async_arangodb.py
@@ -35,7 +35,9 @@
 
 
 class AsyncCursor(AsyncIterator[Any]):
-    def __init__(self, cursor: Cursor, trafo: Optional[Callable[[Json], Optional[Any]]]):
+    def __init__(
+        self, cursor: Cursor, trafo: Optional[Callable[[Json], Optional[Any]]], flatten_nodes_and_edges: bool = False
+    ):
         self.cursor = cursor
         self.visited_node: Set[str] = set()
         self.visited_edge: Set[str] = set()
@@ -44,7 +46,9 @@ def __init__(self, cursor: Cursor, trafo: Optional[Callable[[Json], Optional[Any
         self.trafo: Callable[[Json], Optional[Any]] = trafo if trafo else identity  # type: ignore
         self.vt_len: Optional[int] = None
         self.on_hold: Optional[Json] = None
-        self.get_next: Callable[[], Awaitable[Optional[Json]]] = self.next_filtered if trafo else self.next_from_db
+        self.get_next: Callable[[], Awaitable[Optional[Json]]] = (
+            self.next_filtered if flatten_nodes_and_edges else self.next_element
+        )
 
     async def __anext__(self) -> Any:
         # if there is an on-hold element: unset and return it
@@ -76,6 +80,10 @@ def count(self) -> Optional[int]:
     def full_count(self) -> Optional[int]:
         return stats.get("fullCount") if (stats := self.cursor.statistics()) else None
 
+    async def next_element(self) -> Optional[Json]:
+        element = await self.next_from_db()
+        return self.trafo(element)
+
     async def next_filtered(self) -> Optional[Json]:
         element = await self.next_from_db()
         vertex: Optional[Json] = None
@@ -141,13 +149,8 @@ async def next_deferred_edge(self) -> Json:
 
 
 class AsyncCursorContext(AsyncContextManager[AsyncCursor]):
-    def __init__(self, cursor: Cursor, trafo: Optional[Callable[[Json], Optional[Any]]]):
-        self._cursor = cursor
-        self._trafo = trafo
-
-    @property
-    def cursor(self) -> AsyncCursor:
-        return AsyncCursor(self._cursor, self._trafo)
+    def __init__(self, cursor: Cursor, trafo: Optional[Callable[[Json], Optional[Any]]], flatten_nodes_and_edges: bool):
+        self.cursor = AsyncCursor(cursor, trafo, flatten_nodes_and_edges)
 
     async def __aenter__(self) -> AsyncCursor:
         return self.cursor
@@ -164,6 +167,7 @@ async def aql_cursor(
         self,
         query: str,
         trafo: Optional[Callable[[Json], Optional[Any]]] = None,
+        flatten_nodes_and_edges: Optional[bool] = None,
         count: bool = False,
         batch_size: Optional[int] = None,
         ttl: Optional[Number] = None,
@@ -207,7 +211,7 @@ async def aql_cursor(
             skip_inaccessible_cols,
             max_runtime,
         )
-        return AsyncCursorContext(cursor, trafo)
+        return AsyncCursorContext(cursor, trafo, flatten_nodes_and_edges or False)
 
     async def aql(
         self,
diff --git a/resotocore/resotocore/db/db_access.py b/resotocore/resotocore/db/db_access.py
index e88ea734b3..d54a465327 100644
--- a/resotocore/resotocore/db/db_access.py
+++ b/resotocore/resotocore/db/db_access.py
@@ -26,6 +26,7 @@
 from resotocore.db.runningtaskdb import running_task_db
 from resotocore.db.system_data_db import SystemDataDb
 from resotocore.db.templatedb import template_entity_db
+from resotocore.db.timeseriesdb import TimeSeriesDB
 from resotocore.error import NoSuchGraph, RequiredDependencyMissingError
 from resotocore.ids import GraphName
 from resotocore.model.adjust_node import AdjustNode
@@ -53,6 +54,7 @@ def __init__(
         configs_model: str = "configs_model",
         template_entity: str = "templates",
         infra_app_packages: str = "infra_app_packages",
+        time_series: str = "ts",
     ):
         super().__init__()
         self.event_sender = event_sender
@@ -69,6 +71,7 @@ def __init__(
         self.configs_model_db = model_db(self.db, configs_model)
         self.template_entity_db = template_entity_db(self.db, template_entity)
         self.package_entity_db = app_package_entity_db(self.db, infra_app_packages)
+        self.time_series_db = TimeSeriesDB(self.db, time_series, config.db.time_series_ttl)
         self.graph_dbs: Dict[str, GraphDB] = {}
         self.config = config
         self.cleaner = Periodic("outdated_updates_cleaner", self.check_outdated_updates, timedelta(seconds=60))
@@ -83,6 +86,7 @@ async def start(self) -> None:
             await self.template_entity_db.create_update_schema()
             await self.deferred_outer_edge_db.create_update_schema()
             await self.package_entity_db.create_update_schema()
+            await self.time_series_db.create_update_schema()
             for graph in cast(List[Json], self.database.graphs()):
                 graph_name = GraphName(graph["name"])
 
diff --git a/resotocore/resotocore/db/graphdb.py b/resotocore/resotocore/db/graphdb.py
index 1f6571ceb9..bfe01c7dfc 100644
--- a/resotocore/resotocore/db/graphdb.py
+++ b/resotocore/resotocore/db/graphdb.py
@@ -602,6 +602,7 @@ async def search_list(
         return await self.db.aql_cursor(
             query=q_string,
             trafo=None if kwargs.get("no_trafo") else self.document_to_instance_fn(query.model, query),
+            flatten_nodes_and_edges=True,
             count=with_count,
             full_count=with_count,
             bind_vars=bind,
@@ -663,6 +664,7 @@ async def search_graph_gen(
         return await self.db.aql_cursor(
             query=query_string,
             trafo=self.document_to_instance_fn(query.model, query),
+            flatten_nodes_and_edges=True,
             bind_vars=bind,
             count=with_count,
             full_count=with_count,
@@ -1471,6 +1473,10 @@ def __init__(self, real: ArangoGraphDB, event_sender: AnalyticsEventSender):
     def name(self) -> GraphName:
         return self.real.name
 
+    @property
+    def vertex_name(self) -> str:
+        return self.real.vertex_name
+
     async def get_node(self, model: Model, node_id: NodeId) -> Optional[Json]:
         return await self.real.get_node(model, node_id)
 
diff --git a/resotocore/resotocore/db/timeseriesdb.py b/resotocore/resotocore/db/timeseriesdb.py
new file mode 100644
index 0000000000..ae7969482d
--- /dev/null
+++ b/resotocore/resotocore/db/timeseriesdb.py
@@ -0,0 +1,95 @@
+import time
+from datetime import timedelta, datetime, timezone
+from typing import Optional, List, Set, Union, cast
+
+from attr import evolve
+
+from resotocore.db import arango_query
+from resotocore.db.async_arangodb import AsyncArangoDB, AsyncCursor
+from resotocore.db.graphdb import GraphDB
+from resotocore.db.model import QueryModel
+from resotocore.query.model import Predicate
+from resotocore.types import Json
+from resotocore.util import utc_str
+
+
+class TimeSeriesDB:
+    def __init__(self, db: AsyncArangoDB, collection_name: str, keep_history: timedelta) -> None:
+        self.db = db
+        self.collection_name = collection_name
+        self.keep_history = keep_history
+
+    async def add_entries(self, name: str, query_model: QueryModel, graph_db: GraphDB, 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."
+        assert len(query.aggregate.group_func) == 1, "Only a single group function is supported for time series."
+        # make sure the final value is called "v"
+        query = evolve(
+            query,
+            aggregate=evolve(query.aggregate, group_func=[evolve(query.aggregate.group_func[0], as_name="v")]),
+        )
+        at = at if at is not None else int(time.time())  # only use seconds
+        qs, bv = arango_query.create_time_series(QueryModel(query, model), graph_db, self.collection_name, name, at)
+        async with await self.db.aql_cursor(qs, bind_vars=bv) as crsr:
+            async for v in crsr:
+                return cast(int, v)  # The query will return the number of inserted documents
+        return 0
+
+    async def load_time_series(
+        self,
+        name: str,
+        start: datetime,
+        end: datetime,
+        *,
+        group_by: Optional[Set[str]] = None,
+        filter_by: Optional[List[Predicate]] = None,
+        granularity: Optional[Union[timedelta, int]] = None,
+    ) -> AsyncCursor:
+        """
+        Load time series data.
+        :param name: The name of the time series.
+        :param start: Filter data after this time.
+        :param end: Filter data before this time.
+        :param group_by: Combine defined group properties. Only existing group properties can be used.
+        :param filter_by: Filter specific group properties by predicate.
+        :param granularity: Optional timedelta to retrieve data in a specific granularity.
+               The minimum granularity is one hour.
+               In case this number is an integer, it is interpreted as the number of steps between start and end.
+        :return: A cursor to iterate over the time series data.
+        """
+        assert start < end, "start must be before end"
+        assert name, "name must not be empty"
+        duration = end - start
+
+        # compute effective granularity
+        if isinstance(granularity, int):
+            grl = duration / granularity
+        elif isinstance(granularity, timedelta):
+            grl = granularity
+        else:
+            grl = duration / 20  # default to 20 datapoints if nothing is specified
+        if grl < timedelta(hours=1):
+            grl = timedelta(hours=1)
+
+        qs, bv = arango_query.load_time_series(self.collection_name, name, start, end, grl, group_by, filter_by)
+
+        def result_trafo(js: Json) -> Json:
+            js["at"] = utc_str(datetime.fromtimestamp(js["at"], timezone.utc))
+            return js
+
+        async with await self.db.aql_cursor(qs, bind_vars=bv, trafo=result_trafo) as crsr:
+            return crsr
+
+    async def create_update_schema(self) -> None:
+        if not await self.db.has_collection(self.collection_name):
+            await self.db.create_collection(self.collection_name)
+        collection = self.db.collection(self.collection_name)
+        indexes = {idx["name"]: idx for idx in cast(List[Json], collection.indexes())}
+        if "ttl" not in indexes:
+            collection.add_ttl_index(["at"], expiry_time=int(self.keep_history.total_seconds()), name="ttl")
+        if "access" not in indexes:
+            collection.add_persistent_index(["ts", "at"], name="access")
+
+    async def wipe(self) -> bool:
+        return await self.db.truncate(self.collection_name)
diff --git a/resotocore/resotocore/query/__init__.py b/resotocore/resotocore/query/__init__.py
index 1578673549..3731ab0feb 100644
--- a/resotocore/resotocore/query/__init__.py
+++ b/resotocore/resotocore/query/__init__.py
@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import Optional
+from typing import Optional, Dict
 
 from resotocore.query.model import Query
 
@@ -11,7 +11,12 @@ class QueryParser(ABC):
 
     @abstractmethod
     async def parse_query(
-        self, to_parse: str, on_section: Optional[str], *, omit_section_expansion: bool = False, **env: str
+        self,
+        to_parse: str,
+        on_section: Optional[str],
+        *,
+        omit_section_expansion: bool = False,
+        env: Optional[Dict[str, str]] = None,
     ) -> Query:
         """
         Parse given string into a query.
@@ -20,5 +25,6 @@ async def parse_query(
         :param on_section: interpret the query relative to given section.
                If the section is none, all paths are considered absolute.
         :param omit_section_expansion: if this flag is true, the query is not interpreted relative to section.
+        :param env: optional environment variables to use.
         :return: the parsed query.
         """
diff --git a/resotocore/resotocore/query/query_parser.py b/resotocore/resotocore/query/query_parser.py
index 4b9531edaa..16a6deffc2 100644
--- a/resotocore/resotocore/query/query_parser.py
+++ b/resotocore/resotocore/query/query_parser.py
@@ -1,6 +1,6 @@
 from datetime import timedelta
 from functools import reduce
-from typing import List
+from typing import List, Optional, Dict
 
 import parsy
 from attrs import evolve
@@ -466,7 +466,7 @@ def query_parser() -> Parser:
     return Query(parts[::-1], preamble, maybe_aggregate)
 
 
-def parse_query(query: str, **env: str) -> Query:
+def parse_query(query: str, env: Optional[Dict[str, str]] = None) -> Query:
     def set_edge_type_if_not_set(part: Part, edge_types: List[EdgeType]) -> Part:
         def set_in_with_clause(wc: WithClause) -> WithClause:
             nav = wc.navigation
@@ -485,6 +485,7 @@ def set_in_with_clause(wc: WithClause) -> WithClause:
     try:
         parsed: Query = query_parser.parse(query.strip())
         pre = parsed.preamble
+        env = env or {}
         ets: List[EdgeType] = pre.get("edge_type", env.get("edge_type", EdgeTypes.default)).split(",")  # type: ignore
         for et in ets:
             if et not in EdgeTypes.all:
diff --git a/resotocore/resotocore/query/template_expander.py b/resotocore/resotocore/query/template_expander.py
index d4d5c1ef15..744a067b14 100644
--- a/resotocore/resotocore/query/template_expander.py
+++ b/resotocore/resotocore/query/template_expander.py
@@ -2,7 +2,7 @@
 import json
 import re
 from abc import abstractmethod
-from typing import Any, Union, Iterable, Tuple, List, Optional, cast
+from typing import Any, Union, Iterable, Tuple, List, Optional, cast, Dict
 
 from parsy import string, Parser, regex, any_char
 from ustache import default_getter, default_virtuals, render, PropertyGetter, TagsTuple, default_tags
@@ -123,16 +123,21 @@ def change_well_known_names(name: str) -> str:
         return name
 
     async def parse_query(
-        self, to_parse: str, on_section: Optional[str], *, omit_section_expansion: bool = False, **env: str
+        self,
+        to_parse: str,
+        on_section: Optional[str],
+        *,
+        omit_section_expansion: bool = False,
+        env: Optional[Dict[str, str]] = None,
     ) -> Query:
         # if the query starts with the term "search " then we parse it as command line
         if to_parse.strip().startswith("search "):
             in_section = PathRoot if omit_section_expansion else on_section or PathRoot
-            to_parse = await self.parse_query_from_command_line(to_parse, in_section, **env)
+            to_parse = await self.parse_query_from_command_line(to_parse, in_section, env=env)
             omit_section_expansion = True  # already done
         rendered = self.render(to_parse, env) if env else to_parse
         expanded, _ = await self.expand(rendered)
-        result = query_parser.parse_query(expanded, **env)
+        result = query_parser.parse_query(expanded, env)
         result = result.change_variable(self.change_well_known_names)
         return result if omit_section_expansion else result.on_section(on_section)
 
@@ -164,7 +169,9 @@ def default_props(self) -> Optional[Json]:
         pass
 
     @abstractmethod
-    async def parse_query_from_command_line(self, to_parse: str, on_section: str, **env: str) -> str:
+    async def parse_query_from_command_line(
+        self, to_parse: str, on_section: str, env: Optional[Dict[str, str]] = None
+    ) -> str:
         pass
 
 
diff --git a/resotocore/resotocore/query/template_expander_service.py b/resotocore/resotocore/query/template_expander_service.py
index 6e97f2239d..d3544e7d1d 100644
--- a/resotocore/resotocore/query/template_expander_service.py
+++ b/resotocore/resotocore/query/template_expander_service.py
@@ -1,4 +1,4 @@
-from typing import Optional, List
+from typing import Optional, List, Dict
 
 from resotocore.cli import strip_quotes
 from resotocore.cli.model import CLI, CLIContext
@@ -34,8 +34,10 @@ async def get_template(self, name: str) -> Optional[Template]:
     async def list_templates(self) -> List[Template]:
         return [t async for t in self.db.all()]
 
-    async def parse_query_from_command_line(self, to_parse: str, on_section: str, **env: str) -> str:
-        final_env = {**env, "section": on_section, "no_rewrite": "true"}
+    async def parse_query_from_command_line(
+        self, to_parse: str, on_section: str, env: Optional[Dict[str, str]] = None
+    ) -> str:
+        final_env = {**(env or {}), "section": on_section, "no_rewrite": "true"}
         parsed = await self.cli.evaluate_cli_command(to_parse, CLIContext(env=final_env), replace_place_holder=False)
         if len(parsed) == 1:
             first_line = parsed[0]
diff --git a/resotocore/resotocore/report/inspector_service.py b/resotocore/resotocore/report/inspector_service.py
index 0b72555431..c74c318adb 100644
--- a/resotocore/resotocore/report/inspector_service.py
+++ b/resotocore/resotocore/report/inspector_service.py
@@ -289,7 +289,7 @@ async def __list_failing_resources(
 
         async def perform_search(search: str) -> AsyncIterator[Json]:
             # parse query
-            query = await self.template_expander.parse_query(search, on_section="reported", **env)
+            query = await self.template_expander.parse_query(search, on_section="reported", env=env)
             # filter only relevant accounts if provided
             if context.accounts:
                 query = Query.by(P.single(account_id_prop).is_in(context.accounts)).combine(query)
@@ -415,7 +415,7 @@ async def validate_check_collection_config(self, json: Json) -> Optional[Json]:
                 try:
                     env = check.default_values or {}
                     if search := check.detect.get("resoto"):
-                        await self.template_expander.parse_query(search, on_section="reported", **env)
+                        await self.template_expander.parse_query(search, on_section="reported", env=env)
                     elif cmd := check.detect.get("resoto_cmd"):
                         await self.cli.evaluate_cli_command(cmd, CLIContext(env=env))
                     elif check.detect.get("manual"):
diff --git a/resotocore/resotocore/static/api-doc.yaml b/resotocore/resotocore/static/api-doc.yaml
index fa8ad58a44..97d71ac463 100644
--- a/resotocore/resotocore/static/api-doc.yaml
+++ b/resotocore/resotocore/static/api-doc.yaml
@@ -32,6 +32,8 @@ tags:
     description: Endpoints to get information about the system.
   - name: debug
     description: Endpoints to debug the system.
+  - name: timeseries
+    description: Endpoints to access time series data.
   - name: tsdb
     description: Endpoints to access the time series database.
   - name: authorization
@@ -3094,6 +3096,81 @@ paths:
                 example: "1.1 node1"
           description: "The response from the tsdb."
 
+  # endregion
+
+  # region timeseries
+
+  /timeseries/{timeseries}:
+    post:
+      summary: Load time series data.
+      parameters:
+        - name: timeseries
+          description: "The name of the time series."
+          in: path
+          required: true
+          schema:
+            type: string
+            example: "some_name"
+      requestBody:
+        description: Parameters of the time series request.
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                start:
+                  type: string
+                  format: date-time
+                  description: The start of the time series. If not defined 7 days from now into the past.
+                end:
+                  type: string
+                  format: date-time
+                  description: The end of the time series. If not defined now.
+                group:
+                  type: array
+                  items:
+                      type: string
+                  description: Reduce the available groups to the set of defined ones.
+                filter:
+                  type: array
+                  items:
+                      type: string
+                  description: |
+                    Filter available group members by predicate.
+                    Only time series values with matching group criteria will be selected.
+                  example: group_var_foo == "bla"
+                granularity:
+                  type: string
+                  format: duration
+                  description: |
+                    The granularity of the time series.
+                    This can be either a duration or the number of items for the time window that is selected.
+                    The granularity can not go below one hour.
+                    It defaults to (end-start) / 20.
+      tags:
+        - timeseries
+      responses:
+        "400":
+          description: "A required property is missing or the request can not be read."
+          content:
+            text/plain:
+              schema:
+                type: string
+        default:
+          description: The time series data
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: true
+              example: |
+                {
+                  "time": "2021-01-01T00:00:00Z",
+                  "value": 1
+                }
+
+
+
   # endregion
 
   # region authorization
diff --git a/resotocore/resotocore/web/api.py b/resotocore/resotocore/web/api.py
index 02b6e06606..dcfa0b3e09 100644
--- a/resotocore/resotocore/web/api.py
+++ b/resotocore/resotocore/web/api.py
@@ -28,6 +28,7 @@
     Awaitable,
     Literal,
     cast,
+    Set,
 )
 from urllib.parse import urlencode, urlparse, parse_qs, urlunparse
 
@@ -275,6 +276,8 @@ def __add_routes(self, prefix: str) -> None:
                 web.get(prefix + "/report/checks/graph/{graph_id}", require(self.perform_benchmark_on_checks, r)),
                 web.get(prefix + "/report/check/{check_id}/graph/{graph_id}", require(self.inspection_results, r)),
                 web.get(prefix + "/report/benchmark/{benchmark}/graph/{graph_id}", require(self.perform_benchmark, r)),
+                # time series
+                web.post(prefix + "/timeseries/{timeseries}", require(self.load_time_series, r)),
                 # CLI
                 web.post(prefix + "/cli/evaluate", require(self.evaluate, r)),
                 web.post(prefix + "/cli/execute", require(self.execute, r)),
@@ -704,6 +707,29 @@ async def inspection_results(self, request: Request, deps: TenantDependencies) -
         inspections = await deps.inspector.list_failing_resources(graph, check_id, accounts)
         return await self.stream_response_from_gen(request, inspections)
 
+    async def load_time_series(self, request: Request, deps: TenantDependencies) -> StreamResponse:
+        def parse_duration_or_int(s: str) -> Union[int, timedelta]:
+            try:
+                return duration(s)
+            except Exception:
+                return int(s)
+
+        name = request.match_info["timeseries"]
+        body = await request.json() if request.content_length else {}
+        start = if_set(body.get("start"), parse_utc, utc() - timedelta(days=7))
+        end = if_set(body.get("end"), parse_utc, utc())
+        group_by: Optional[Set[str]] = if_set(body.get("group"), set)
+        filter_by: Optional[List[Predicate]] = if_set(
+            body.get("filter"), lambda x: [predicate_term.parse(y) for y in x]  # type: ignore
+        )
+        granularity: Optional[Union[int, timedelta]] = if_set(body.get("granularity"), parse_duration_or_int)
+        cursor = await deps.db_access.time_series_db.load_time_series(
+            name, start, end, group_by=group_by, filter_by=filter_by, granularity=granularity
+        )
+        return await self.stream_response_from_gen(
+            request, cursor, count=cursor.count(), total_count=cursor.full_count()
+        )
+
     async def handle_events(self, request: Request, deps: TenantDependencies) -> StreamResponse:
         show = request.query["show"].split(",") if "show" in request.query else ["*"]
         return await self.listen_to_events(request, deps, SubscriberId(str(uuid.uuid1())), show)
@@ -1047,7 +1073,7 @@ async def graph_query_model_from_request(
         query_string = await request.text()
         graph_name, model = await self.graph_model_from_request(request, deps)
         graph_db = deps.db_access.get_graph_db(graph_name)
-        q = await deps.template_expander.parse_query(query_string, section, **request.query)
+        q = await deps.template_expander.parse_query(query_string, section, env=cast(Dict[str, str], request.query))
         return graph_db, QueryModel(q, model, cast(Dict[str, Any], request.query))
 
     async def raw(self, request: Request, deps: TenantDependencies) -> StreamResponse:
diff --git a/resotocore/tests/resotocore/cli/command_test.py b/resotocore/tests/resotocore/cli/command_test.py
index 01a4fbe92f..71978bc39e 100644
--- a/resotocore/tests/resotocore/cli/command_test.py
+++ b/resotocore/tests/resotocore/cli/command_test.py
@@ -1393,3 +1393,34 @@ async def check(in_: Stream) -> None:
 
     # make sure argsinfo is available
     assert "sync" in cli.direct_commands["db"].args_info()
+
+
+@pytest.mark.asyncio
+async def test_timeseries(cli: CLI) -> None:
+    async def exec(cmd: str) -> List[JsonElement]:
+        res = await cli.execute_cli_command(cmd, stream.list)
+        return cast(List[JsonElement], res[0])
+
+    tsdb = cli.dependencies.db_access.time_series_db
+    await tsdb.wipe()  # create a clean slate
+    now = utc()
+    in_one_min = utc_str(now + timedelta(minutes=1))
+    one_min_ago = utc_str(now - timedelta(minutes=1))
+    # Create a time series based on all foo entries
+    res = await exec("timeseries snapshot --name test 'search aggregate(reported.some_int, reported.identifier: sum(1)): is(foo)'")  # fmt: skip
+    assert res[0] == "10 entries added to time series test."
+    # Get the time series combined with each complete group
+    res = await exec(f"timeseries get --name test --start {one_min_ago} --end {in_one_min}")
+    assert len(res) == 10
+    # Get the time series combined over all groups --> only one entry for one timestamp
+    res = await exec(f"timeseries get --name test --start {one_min_ago} --end {in_one_min} --group")
+    assert len(res) == 1
+    # Combine over some_int (which has only one) --> only one entry for one timestamp
+    res = await exec(f"timeseries get --name test --start {one_min_ago} --end {in_one_min} --group some_int")
+    assert len(res) == 1
+    # Combine over identifier (which is unique for each entry) --> 10 entries for one timestamp
+    res = await exec(f"timeseries get --name test --start {one_min_ago} --end {in_one_min} --group identifier")
+    assert len(res) == 10
+    # Combine over identifier (which is unique for each entry), filter for identifier==2 --> 1 entry for one timestamp
+    res = await exec(f'timeseries get --name test --start {one_min_ago} --end {in_one_min} --group identifier --filter identifier=="2"')  # fmt: skip
+    assert len(res) == 1
diff --git a/resotocore/tests/resotocore/conftest.py b/resotocore/tests/resotocore/conftest.py
index 6ea9f5bce1..1e719e594e 100644
--- a/resotocore/tests/resotocore/conftest.py
+++ b/resotocore/tests/resotocore/conftest.py
@@ -52,6 +52,7 @@
 from resotocore.db.packagedb import PackageEntityDb, app_package_entity_db
 from resotocore.db.runningtaskdb import RunningTaskDb
 from resotocore.db.system_data_db import SystemDataDb
+from resotocore.db.timeseriesdb import TimeSeriesDB
 from resotocore.dependencies import DirectTenantDependencyProvider, TenantDependencies
 from resotocore.graph_manager.graph_manager import GraphManager
 from resotocore.ids import SubscriberId, WorkerId, TaskDescriptorId, ConfigId, GraphName
@@ -210,6 +211,15 @@ async def pending_deferred_edge_db(async_db: AsyncArangoDB) -> DeferredOuterEdge
     return edges_db
 
 
+@fixture
+async def timeseries_db(async_db: AsyncArangoDB) -> TimeSeriesDB:
+    db = TimeSeriesDB(async_db, "ts", timedelta(hours=1))
+    await db.create_update_schema()
+    with suppress(Exception):
+        async_db.db.collection("ts").delete_index("ttl", ignore_missing=True)  # remove index for tests
+    return db
+
+
 @fixture()
 def db_access(graph_db: ArangoGraphDB) -> DbAccess:
     access = DbAccess(graph_db.db.db, NoEventSender(), NoAdjust(), empty_config())
diff --git a/resotocore/tests/resotocore/db/arango_query_test.py b/resotocore/tests/resotocore/db/arango_query_test.py
index 36b24cb2b6..56766498c4 100644
--- a/resotocore/tests/resotocore/db/arango_query_test.py
+++ b/resotocore/tests/resotocore/db/arango_query_test.py
@@ -1,7 +1,9 @@
+from datetime import timedelta, datetime
+
 import pytest
 
 from resotocore.db import EstimatedSearchCost, EstimatedQueryCostRating
-from resotocore.db.arango_query import to_query, query_cost, fulltext_term_combine, possible_values
+from resotocore.db.arango_query import to_query, query_cost, fulltext_term_combine, possible_values, load_time_series
 from resotocore.db.graphdb import GraphDB
 from resotocore.db.model import QueryModel
 from resotocore.model.model import Model
@@ -279,3 +281,51 @@ def test_possible_values(foo_model: Model, graph_db: GraphDB) -> None:
     # skip and limit the result
     pv, bv = possible_values(graph_db, model, "reported.tags", "values", limit=10, skip=20)
     assert pv.endswith("SORT m6 ASC LIMIT 20, 10 RETURN m6")
+
+
+def test_load_time_series() -> None:
+    now = datetime.fromtimestamp(1700000000)
+    one_hour = timedelta(hours=1)
+    # group_by=[] --> no group by any value
+    q, bv = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour, group_by=[])
+    assert (
+        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<=@b2 "
+        "LET m0 = (FLOOR(d.at / @b3) * @b3) + @b4 "
+        "COLLECT group_slot=m0 INTO group "
+        "SORT group_slot "
+        "RETURN {at: group_slot,  v: AVG(group[*].d.v)}"
+    )
+    assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": 3600, "b4": 800}
+    # no group by defined --> group by all values
+    q, bv = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour)
+    assert (
+        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<=@b2 "
+        "LET m0 = (FLOOR(d.at / @b3) * @b3) + @b4 "
+        "COLLECT group_slot=m0, complete_group=d.group INTO group "
+        "SORT group_slot "
+        "RETURN {at: group_slot, group: complete_group, v: AVG(group[*].d.v)}"
+    )
+    assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": 3600, "b4": 800}
+    # group by specific group variables
+    q, bv = load_time_series("ts", "foo", now - (24 * one_hour), now, one_hour, group_by=["a", "b"])
+    assert (
+        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<=@b2 "
+        "LET m0 = (FLOOR(d.at / @b3) * @b3) + @b4 "
+        "COLLECT group_slot=m0, group_a=d.group.a, group_b=d.group.b INTO group "
+        "SORT group_slot "
+        "RETURN {at: group_slot, group: { a: group_a, b: group_b }, v: AVG(group[*].d.v)}"
+    )
+    assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": 3600, "b4": 800}
+    # group by specific group variables and filter by group variables
+    q, bv = load_time_series(
+        "ts", "foo", now - (24 * one_hour), now, one_hour, group_by=["a", "b"], group_filter=[P("a").eq("a")]
+    )
+    assert (
+        q == "FOR d in `ts` FILTER d.ts==@b0 AND d.at>=@b1 AND d.at<=@b2 "
+        "FILTER d.group.a==@b3 "
+        "LET m0 = (FLOOR(d.at / @b4) * @b4) + @b5 "
+        "COLLECT group_slot=m0, group_a=d.group.a, group_b=d.group.b INTO group "
+        "SORT group_slot "
+        "RETURN {at: group_slot, group: { a: group_a, b: group_b }, v: AVG(group[*].d.v)}"
+    )
+    assert bv == {"b0": "foo", "b1": 1699913600, "b2": 1700000000, "b3": "a", "b4": 3600, "b5": 800}
diff --git a/resotocore/tests/resotocore/db/timeseriesdb_test.py b/resotocore/tests/resotocore/db/timeseriesdb_test.py
new file mode 100644
index 0000000000..08474e6e38
--- /dev/null
+++ b/resotocore/tests/resotocore/db/timeseriesdb_test.py
@@ -0,0 +1,50 @@
+from datetime import datetime, timedelta
+from typing import Any, List
+
+from resotocore.db.graphdb import GraphDB
+from resotocore.db.model import QueryModel
+from resotocore.db.timeseriesdb import TimeSeriesDB
+from resotocore.model.model import Model
+from resotocore.query.model import P
+from resotocore.query.query_parser import parse_query
+from resotocore.types import Json
+
+
+async def test_create_time_series(timeseries_db: TimeSeriesDB, foo_model: Model, filled_graph_db: GraphDB) -> None:
+    await timeseries_db.wipe()  # clean up
+    qm = QueryModel(parse_query("aggregate(reported.some_int, reported.identifier: sum(1)): is(foo)"), foo_model)
+    # create timeseries at 10 different points in time
+    for a in range(10):
+        await timeseries_db.add_entries("test", qm, filled_graph_db, at=3600 * a)
+    begin = datetime.fromtimestamp(0)
+    after5h = datetime.fromtimestamp(3600 * 5 - 1)
+
+    async def load_ts(**kwargs: Any) -> List[Json]:
+        return [t async for t in await timeseries_db.load_time_series(**kwargs)]
+
+    ## check start end is working
+    # first 5 hours with 10 entries each: 50 entries
+    assert len(await load_ts(name="test", start=begin, end=datetime.fromtimestamp(3600 * 1 - 1))) == 10
+    assert len(await load_ts(name="test", start=begin, end=datetime.fromtimestamp(3600 * 2 - 1))) == 20
+    assert len(await load_ts(name="test", start=begin, end=datetime.fromtimestamp(3600 * 4 - 1))) == 40
+    assert len(await load_ts(name="test", start=datetime.fromtimestamp(3600 * 4), end=after5h)) == 10
+
+    ## check granularity is working
+    # first 5 hours with 10 entries each: 50 entries
+    assert len(await load_ts(name="test", start=begin, end=after5h)) == 50
+    # first 5 hours with 10 entries each: granularity 2 hours: at 0, 2 and 4: 30 entries
+    assert len(await load_ts(name="test", start=begin, end=after5h, granularity=timedelta(hours=2))) == 30
+    # first 5 hours with 10 entries each: granularity 5 hours: at 0: 10 entries
+    assert len(await load_ts(name="test", start=begin, end=after5h, granularity=timedelta(hours=5))) == 10
+
+    ## check group_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, group_by=["some_int"])) == 5
+    # identifier is different for each entry: 50 entries
+    assert len(await load_ts(name="test", start=begin, end=after5h, group_by=["identifier"])) == 50
+    # do not use any groups: 5 entries
+    assert len(await load_ts(name="test", start=begin, end=after5h, group_by=[])) == 5
+
+    ## 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("identifier").eq("1"))])) == 5
diff --git a/resotocore/tests/resotocore/query/query_parser_test.py b/resotocore/tests/resotocore/query/query_parser_test.py
index 782426ee13..3c28fb54c5 100644
--- a/resotocore/tests/resotocore/query/query_parser_test.py
+++ b/resotocore/tests/resotocore/query/query_parser_test.py
@@ -235,7 +235,7 @@ def test_query_with_preamble() -> None:
     q1 = parse_query('(edge_type=delete): id("root") -[0:1]->')
     assert q1.parts[0].navigation.edge_types == ["delete"]  # type: ignore
     # edge type can be defined via kwargs
-    q2 = parse_query('id("root") -[0:1]->', edge_type="delete")
+    q2 = parse_query('id("root") -[0:1]->', dict(edge_type="delete"))
     assert q2.parts[0].navigation.edge_types == ["delete"]  # type: ignore
     # aggregation and preamble
     q3 = parse_query('aggregate(region: sum(cpu))(edge_type=delete): id("root") -[0:1]->')
diff --git a/resotocore/tests/resotocore/query/template_expander_test.py b/resotocore/tests/resotocore/query/template_expander_test.py
index c6e5b07cd3..e578a73787 100644
--- a/resotocore/tests/resotocore/query/template_expander_test.py
+++ b/resotocore/tests/resotocore/query/template_expander_test.py
@@ -35,7 +35,9 @@ async def list_templates(self) -> List[Template]:
     def default_props(self) -> Optional[Json]:
         return self.props
 
-    async def parse_query_from_command_line(self, to_parse: str, on_section: str, **env: str) -> str:
+    async def parse_query_from_command_line(
+        self, to_parse: str, on_section: str, env: Optional[Dict[str, str]] = None
+    ) -> str:
         # straight forward implementation: simply strip the "search" command
         return re.sub("^search\\s+", "", to_parse)
 
diff --git a/resotocore/tests/resotocore/web/api_test.py b/resotocore/tests/resotocore/web/api_test.py
index 4d97d4aa0a..307e8fed00 100644
--- a/resotocore/tests/resotocore/web/api_test.py
+++ b/resotocore/tests/resotocore/web/api_test.py
@@ -388,7 +388,7 @@ async def test_cli(core_client: ResotoClient) -> None:
 
     # list all cli commands
     info = AccessJson(await core_client.cli_info())
-    assert len(info.commands) == 44
+    assert len(info.commands) == 45
 
 
 @pytest.mark.asyncio