diff --git a/resotocore/resotocore/cli/command.py b/resotocore/resotocore/cli/command.py
index 388537297f..3fff04947c 100644
--- a/resotocore/resotocore/cli/command.py
+++ b/resotocore/resotocore/cli/command.py
@@ -6038,6 +6038,8 @@ 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>
+    timeseries list
+    timeseries downsample
     ```
 
     The timeseries command can be used to add data to a time series or to retrieve data from a time series.
@@ -6051,6 +6053,13 @@ class TimeSeriesCommand(CLICommand):
     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.
 
+    All available timeseries can be retrieved with the `timeseries list` command.
+
+    The `timeseries downsample` command can be used to downsample all time series.
+    See resoto.core db configuration to configure the buckets that are used for downsampling.
+    General idea of downsample: higher resolution and more datapoints for recent data,
+                                lower resolution and less datapoints for older data.
+
     ## Parameters
 
     - `--name` - The name of the time series.
@@ -6069,6 +6078,12 @@ class TimeSeriesCommand(CLICommand):
 
     # retrieve the number of instances per instance type since yesterday
     > timeseries get --name instances --start @yesterday@ --end @now@
+
+    # downsample existing time series
+    > timeseries downsample
+
+    # list all time series
+    > timeseries list
     ```
     """
 
@@ -6081,6 +6096,7 @@ def info(self) -> str:
 
     def args_info(self) -> ArgsInfo:
         return {
+            "list": [],
             "snapshot": [
                 ArgInfo("--name", help_text="<time series>."),
                 ArgInfo(expects_value=True, value_hint="search"),
@@ -6093,6 +6109,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>"),
             ],
+            "downsample": [],
         }
 
     def parse(self, arg: Optional[str] = None, ctx: CLIContext = EmptyContext, **kwargs: Any) -> CLIAction:
@@ -6128,6 +6145,19 @@ def parse_duration_or_int(s: str) -> Union[int, timedelta]:
             )
             return CLISourceContext(cursor.count(), cursor.full_count()), cursor
 
+        async def list_ts() -> Tuple[int, JsGen]:
+            ts = await self.dependencies.db_access.time_series_db.list_time_series()
+            return len(ts), stream.iterate([to_js(a) for a in ts])
+
+        async def downsample() -> Tuple[int, JsGen]:
+            ts = await self.dependencies.db_access.time_series_db.downsample()
+            if isinstance(ts, str):
+                return 1, stream.just(ts)
+            elif ts:
+                return len(ts), stream.iterate([{k: v} for k, v in ts.items()])
+            else:
+                return 1, stream.just("No time series to downsample.")
+
         args = re.split("\\s+", arg, maxsplit=1) if arg else []
         if arg and len(args) == 2 and args[0] == "snapshot":
             return CLISource.single(
@@ -6135,6 +6165,10 @@ def parse_duration_or_int(s: str) -> Union[int, timedelta]:
             )
         elif arg and len(args) == 2 and args[0] == "get":
             return CLISource(partial(load_time_series, args[1].strip()), required_permissions={Permission.read})
+        elif arg and len(args) == 1 and args[0] == "list":
+            return CLISource.only_count(list_ts, required_permissions={Permission.read})
+        elif arg and len(args) == 1 and args[0] == "downsample":
+            return CLISource.only_count(downsample, required_permissions={Permission.read})
         else:
             return CLISource.single(
                 lambda: stream.just(self.rendered_help(ctx)), required_permissions={Permission.read}
diff --git a/resotocore/resotocore/core_config.py b/resotocore/resotocore/core_config.py
index 1c0c63b0b9..0a6268bedc 100644
--- a/resotocore/resotocore/core_config.py
+++ b/resotocore/resotocore/core_config.py
@@ -171,6 +171,13 @@ def access_token_expiration(self) -> timedelta:
 )
 
 
+@define()
+class TimeSeriesBucketConfig(ConfigObject):
+    kind: ClassVar[str] = f"{ResotoCoreRoot}_timeseries_bucket_config"
+    start: int = field(metadata={"description": "Start of the bucket in seconds."})
+    resolution: int = field(metadata={"description": "Resolution of the bucket in seconds."})
+
+
 @define()
 class DatabaseConfig(ConfigObject):
     kind: ClassVar[str] = f"{ResotoCoreRoot}_database_config"
@@ -192,9 +199,6 @@ 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)
@@ -572,6 +576,26 @@ class RunConfig(ConfigObject):
     verify: Union[bool, str, None] = None
 
 
+@define
+class TimeSeriesConfig(ConfigObject):
+    kind: ClassVar[str] = f"{ResotoCoreRoot}_timeseries_config"
+
+    buckets: List[TimeSeriesBucketConfig] = field(
+        factory=lambda: [
+            TimeSeriesBucketConfig(
+                start=int(timedelta(days=2).total_seconds()), resolution=int(timedelta(hours=4).total_seconds())
+            ),
+            TimeSeriesBucketConfig(
+                start=int(timedelta(days=30).total_seconds()), resolution=int(timedelta(days=1).total_seconds())
+            ),
+            TimeSeriesBucketConfig(
+                start=int(timedelta(days=180).total_seconds()), resolution=int(timedelta(days=3).total_seconds())
+            ),
+        ],
+        metadata={"description": "List of time series buckets."},
+    )
+
+
 @define()
 class CoreConfig(ConfigObject):
     api: ApiConfig
@@ -584,6 +608,7 @@ class CoreConfig(ConfigObject):
     snapshots: SnapshotsScheduleConfig
     args: Namespace
     run: RunConfig
+    timeseries: TimeSeriesConfig
 
     @property
     def multi_tenant_setup(self) -> bool:
@@ -595,7 +620,7 @@ def no_scheduling(self) -> bool:
 
     @property
     def editable(self) -> "EditableConfig":
-        return EditableConfig(self.api, self.cli, self.graph_update, self.runtime, self.workflows)
+        return EditableConfig(self.api, self.cli, self.graph_update, self.runtime, self.workflows, self.timeseries)
 
     def json(self) -> Json:
         return {ResotoCoreRoot: to_js(self.editable, strip_attr="kind")}
@@ -627,6 +652,10 @@ class EditableConfig(ConfigObject):
         factory=lambda: {"collect_and_cleanup": WorkflowConfig(schedule="0 * * * *")},
         metadata={"description": "Workflow related properties."},
     )
+    timeseries: TimeSeriesConfig = field(
+        factory=TimeSeriesConfig,
+        metadata={"description": "Time series related properties."},
+    )
 
 
 def config_model() -> List[Json]:
@@ -740,6 +769,7 @@ def parse_config(
         runtime=ed.runtime,
         workflows=ed.workflows,
         run=RunConfig(),  # overridden for each run
+        timeseries=ed.timeseries,
     )
 
 
diff --git a/resotocore/resotocore/db/arango_query.py b/resotocore/resotocore/db/arango_query.py
index 0f900f40a9..91ff6b9186 100644
--- a/resotocore/resotocore/db/arango_query.py
+++ b/resotocore/resotocore/db/arango_query.py
@@ -862,7 +862,7 @@ def load_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}"
+    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:
diff --git a/resotocore/resotocore/db/db_access.py b/resotocore/resotocore/db/db_access.py
index d54a465327..2a59f9456e 100644
--- a/resotocore/resotocore/db/db_access.py
+++ b/resotocore/resotocore/db/db_access.py
@@ -71,7 +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.time_series_db = TimeSeriesDB(self.db, time_series, config)
         self.graph_dbs: Dict[str, GraphDB] = {}
         self.config = config
         self.cleaner = Periodic("outdated_updates_cleaner", self.check_outdated_updates, timedelta(seconds=60))
diff --git a/resotocore/resotocore/db/timeseriesdb.py b/resotocore/resotocore/db/timeseriesdb.py
index ae7969482d..1470382f82 100644
--- a/resotocore/resotocore/db/timeseriesdb.py
+++ b/resotocore/resotocore/db/timeseriesdb.py
@@ -1,23 +1,77 @@
+import logging
 import time
+from collections import defaultdict
 from datetime import timedelta, datetime, timezone
-from typing import Optional, List, Set, Union, cast
-
-from attr import evolve
+from functools import partial
+from typing import Optional, List, Set, Union, cast, Callable, Dict
 
+from attr import evolve, define
+from resotocore.core_config import CoreConfig
 from resotocore.db import arango_query
-from resotocore.db.async_arangodb import AsyncArangoDB, AsyncCursor
+from resotocore.db.async_arangodb import AsyncArangoDB, AsyncCursor, AsyncArangoTransactionDB
 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
+from resotocore.types import Json, JsonElement
+from resotocore.util import utc_str, utc, if_set, parse_utc
+from resotolib.durations import duration_str
+
+log = logging.getLogger(__name__)
+max_delta = timedelta(days=730)
+last_run_name = "last_run"
+
+
+@define(repr=False, str=False)
+class TimeSeriesBucket:
+    start: timedelta
+    end: timedelta
+    resolution: timedelta
+
+    def __repr__(self) -> str:
+        return f"{duration_str(self.start)}:{duration_str(self.end)}:{duration_str(self.resolution)}"
+
+    def __str__(self) -> str:
+        return (
+            f"Bucket(start={duration_str(self.start)}, "
+            f"end={duration_str(self.end)}, resolution={duration_str(self.resolution)})"
+        )
+
+
+@define
+class TimeSeriesMeta:
+    name: str
+    created_at: datetime
+    last_updated: datetime
+    downsample_times: Dict[str, datetime]
 
 
 class TimeSeriesDB:
-    def __init__(self, db: AsyncArangoDB, collection_name: str, keep_history: timedelta) -> None:
+    def __init__(self, db: AsyncArangoDB, collection_name: str, config: CoreConfig) -> None:
         self.db = db
         self.collection_name = collection_name
-        self.keep_history = keep_history
+        self.meta_db = f"{collection_name}_meta"
+        self.names_db = f"{collection_name}_names"
+        self.config = config
+        self.buckets = self._buckets()
+        self.smallest_resolution = min(bucket.resolution for bucket in self.buckets)
+
+    async def __execute_aql(
+        self, query: str, bind_vars: Optional[Json] = None, tx: Optional[AsyncArangoTransactionDB] = None
+    ) -> List[JsonElement]:
+        async with await (tx or self.db).aql_cursor(query, bind_vars=bind_vars) as crsr:
+            return [el async for el in crsr]
+
+    async def list_time_series(self) -> List[TimeSeriesMeta]:
+        all_ts = cast(List[Json], await self.__execute_aql(f"FOR d IN {self.names_db} return d"))
+        return [
+            TimeSeriesMeta(
+                e["_key"],
+                datetime.fromtimestamp(e["created_at"] / 1000.0, timezone.utc),
+                datetime.fromtimestamp(e["last_updated"] / 1000.0, timezone.utc),
+                {k: datetime.fromtimestamp(v / 1000.0, timezone.utc) for k, v in e.get("downsample_times", {}).items()},
+            )
+            for e in all_ts
+        ]
 
     async def add_entries(self, name: str, query_model: QueryModel, graph_db: GraphDB, at: Optional[int] = None) -> int:
         query = query_model.query
@@ -31,10 +85,17 @@ async def add_entries(self, name: str, query_model: QueryModel, graph_db: GraphD
         )
         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
+        result, *_ = cast(List[int], await self.__execute_aql(qs, bv))
+        if result > 0:
+            # 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),
+            )
+
+        return result
 
     async def load_time_series(
         self,
@@ -45,6 +106,7 @@ async def load_time_series(
         group_by: Optional[Set[str]] = None,
         filter_by: Optional[List[Predicate]] = None,
         granularity: Optional[Union[timedelta, int]] = None,
+        trafo: Optional[Callable[[Json], Json]] = None,
     ) -> AsyncCursor:
         """
         Load time series data.
@@ -56,6 +118,7 @@ async def load_time_series(
         :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.
+        :param trafo: Optional transformation function to apply to each result.
         :return: A cursor to iterate over the time series data.
         """
         assert start < end, "start must be before end"
@@ -78,18 +141,120 @@ 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:
+        async with await self.db.aql_cursor(qs, bind_vars=bv, trafo=trafo or result_trafo) as crsr:
             return crsr
 
+    async def downsample(self, now: Optional[datetime] = None) -> Union[str, Json]:
+        def ts_format(ts: str, js: Json) -> Json:
+            js["ts"] = ts
+            js["at"] = int(js["at"])
+            return js
+
+        now = now or utc()
+        oldest = now - max_delta
+        # check if there is something to do
+        last_run = if_set(await self.db.get(self.meta_db, last_run_name), lambda d: parse_utc(d[last_run_name]), oldest)
+        if (now - last_run) < self.smallest_resolution:
+            return "No changes since last downsample run"
+        # acquire a lock to ensure exclusive access
+        try:
+            ttl = int((now + timedelta(minutes=15)).timestamp())
+            await self.db.insert(self.meta_db, dict(_key="lock", expires=ttl), sync=True)
+        except Exception:
+            return "Another downsample run is already in progress."
+        # If we come here, the lock is acquired: exclusive access.
+        # We only touch time series that are older than the minimal resolution (>1h).
+        # So we never interfere with snapshots that are eventually created concurrently.
+        result: Json = defaultdict(list)
+        try:
+            for ts in await self.list_time_series():
+                dst = ts.downsample_times.copy()
+                for bucket in self.buckets:
+                    ts_bucket_last = ts.downsample_times.get(repr(bucket), oldest)
+                    c_start = max(now - bucket.end, ts_bucket_last - bucket.start)
+                    c_end = now - bucket.start
+                    # Only downsample when the resolution duration is reached
+                    if c_end <= c_start or (c_end - c_start) < bucket.resolution or ts.last_updated < c_start:
+                        continue
+                    if ts_data := [
+                        e
+                        async for e in await self.load_time_series(
+                            ts.name,
+                            c_start,
+                            c_end,
+                            granularity=bucket.resolution,
+                            trafo=partial(ts_format, ts.name),
+                        )
+                    ]:
+                        log.info(f"Compact {ts.name} bucket {bucket} to {len(ts_data)} entries (last={ts_bucket_last})")
+                        result[ts.name].append(
+                            {
+                                "bucket": str(bucket),
+                                "start": utc_str(c_start),
+                                "end": utc_str(c_end),
+                                "data_points": len(ts_data),
+                            }
+                        )
+                        dst[repr(bucket)] = now  # update last downsample time in this bucket
+                        async with self.db.begin_transaction(write=[self.collection_name, self.names_db]) as tx:
+                            await self.__execute_aql(
+                                "FOR a in @@coll FILTER a.ts==@ts and a.at>=@start and a.at<@end REMOVE a IN @@coll",
+                                {
+                                    "ts": ts.name,
+                                    "start": c_start.timestamp(),
+                                    "end": c_end.timestamp(),
+                                    "@coll": self.collection_name,
+                                },
+                                tx=tx,
+                            )
+                            await tx.insert_many(self.collection_name, ts_data)
+                            await self.__execute_aql(
+                                "UPDATE {_key: @key, downsample_times: @dst} IN @@coll",
+                                bind_vars={
+                                    "@coll": self.names_db,
+                                    "key": ts.name,
+                                    "dst": {k: int(v.timestamp() * 1000) for k, v in dst.items()},
+                                },
+                                tx=tx,
+                            )
+            # update last run
+            await self.db.insert(self.meta_db, {"_key": last_run_name, last_run_name: utc_str(now)}, overwrite=True)
+        finally:
+            await self.db.delete(self.meta_db, "lock", ignore_missing=True)
+        return result
+
     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 "ttl" in indexes:
+            collection.delete_index("ttl")
         if "access" not in indexes:
             collection.add_persistent_index(["ts", "at"], name="access")
+        if not await self.db.has_collection(self.meta_db):
+            await self.db.create_collection(self.meta_db)
+        # meta collection: store information to handle ts
+        collection = self.db.collection(self.meta_db)
+        indexes = {idx["name"]: idx for idx in cast(List[Json], collection.indexes())}
+        if "ttl" not in indexes:
+            collection.add_ttl_index(["expires"], expiry_time=int(timedelta(hours=3).total_seconds()), name="ttl")
+        # names collection: store all names of time series
+        if not await self.db.has_collection(self.names_db):
+            await self.db.create_collection(self.names_db)
 
     async def wipe(self) -> bool:
-        return await self.db.truncate(self.collection_name)
+        return (
+            await self.db.truncate(self.collection_name)
+            and await self.db.truncate(self.names_db)
+            and await self.db.truncate(self.meta_db)
+        )
+
+    def _buckets(self) -> List[TimeSeriesBucket]:
+        result: List[TimeSeriesBucket] = []
+        if bs := self.config.timeseries.buckets:
+            cfg = sorted(bs, key=lambda b: b.start)
+            for a, z in zip(cfg, cfg[1:] + [None]):
+                end = timedelta(seconds=z.start) if z else max_delta
+                result.append(TimeSeriesBucket(timedelta(seconds=a.start), end, timedelta(seconds=a.resolution)))
+        return result
diff --git a/resotocore/tests/resotocore/conftest.py b/resotocore/tests/resotocore/conftest.py
index 1e719e594e..8f4e9ce730 100644
--- a/resotocore/tests/resotocore/conftest.py
+++ b/resotocore/tests/resotocore/conftest.py
@@ -126,6 +126,7 @@ def default_config() -> CoreConfig:
         snapshots=SnapshotsScheduleConfig(),
         args=parse_args(["--analytics-opt-out"]),
         run=RunConfig(),
+        timeseries=ed.timeseries,
     )
 
 
@@ -212,11 +213,9 @@ async def pending_deferred_edge_db(async_db: AsyncArangoDB) -> DeferredOuterEdge
 
 
 @fixture
-async def timeseries_db(async_db: AsyncArangoDB) -> TimeSeriesDB:
-    db = TimeSeriesDB(async_db, "ts", timedelta(hours=1))
+async def timeseries_db(async_db: AsyncArangoDB, default_config: CoreConfig) -> TimeSeriesDB:
+    db = TimeSeriesDB(async_db, "ts", default_config)
     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
 
 
diff --git a/resotocore/tests/resotocore/core_config_test.py b/resotocore/tests/resotocore/core_config_test.py
index 2d7f48ef20..572fd46c0c 100644
--- a/resotocore/tests/resotocore/core_config_test.py
+++ b/resotocore/tests/resotocore/core_config_test.py
@@ -123,6 +123,8 @@ def test_model() -> None:
         "resotocore_cli_config",
         "resotocore_graph_update_config",
         "resotocore_runtime_config",
+        "resotocore_timeseries_bucket_config",
+        "resotocore_timeseries_config",
         "resotocore_workflow_config",
     }
 
@@ -217,6 +219,13 @@ def config_json() -> Json:
                 "plantuml_server": "https://foo",
                 "start_collect_on_subscriber_connect": True,
             },
+            "timeseries": {
+                "buckets": [
+                    {"resolution": 14400, "start": 172800},
+                    {"resolution": 86400, "start": 2592000},
+                    {"resolution": 259200, "start": 15552000},
+                ]
+            },
             "workflows": {
                 "collect_and_cleanup": {
                     "schedule": "0 0 0 0 *",
diff --git a/resotocore/tests/resotocore/db/arango_query_test.py b/resotocore/tests/resotocore/db/arango_query_test.py
index 56766498c4..0360ec28b6 100644
--- a/resotocore/tests/resotocore/db/arango_query_test.py
+++ b/resotocore/tests/resotocore/db/arango_query_test.py
@@ -289,7 +289,7 @@ def test_load_time_series() -> None:
     # 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 "
+        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 "
@@ -299,7 +299,7 @@ def test_load_time_series() -> None:
     # 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 "
+        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 "
@@ -309,7 +309,7 @@ def test_load_time_series() -> None:
     # 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 "
+        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 "
@@ -321,7 +321,7 @@ def test_load_time_series() -> None:
         "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 "
+        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 "
diff --git a/resotocore/tests/resotocore/db/timeseriesdb_test.py b/resotocore/tests/resotocore/db/timeseriesdb_test.py
index 08474e6e38..5f37e5a92d 100644
--- a/resotocore/tests/resotocore/db/timeseriesdb_test.py
+++ b/resotocore/tests/resotocore/db/timeseriesdb_test.py
@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, List
 
 from resotocore.db.graphdb import GraphDB
@@ -48,3 +48,80 @@ 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("identifier").eq("1"))])) == 5
+
+
+async def test_compact_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)
+    now = datetime(2023, 12, 1, tzinfo=timezone.utc)
+
+    async def create_ts(before_now: timedelta, granularity: timedelta, number_of_entries: int) -> None:
+        start = now - before_now
+        for a in range(number_of_entries):
+            await timeseries_db.add_entries("test", qm, filled_graph_db, at=int((start - a * granularity).timestamp()))
+
+    # after 180 days, we want a 3-day resolution: 24 daily entries --> reduce by factor 3
+    await create_ts(timedelta(days=182), timedelta(days=1), 24)
+    # after 30 days, we want a 1-day resolution: 24 entries every 4 hour --> reduce by factor 4
+    await create_ts(timedelta(days=31), timedelta(hours=4), 24)
+    # after 2 days we want a 4-hour resolution: 24 entries --> reduce by factor 4
+    await create_ts(timedelta(days=3), timedelta(hours=1), 24)
+    # after now we want a 1-hour resolution: 24 entries --> no reduction
+    await create_ts(timedelta(), timedelta(hours=1), 24)
+
+    assert timeseries_db.db.collection(timeseries_db.collection_name).count() == 960
+    assert await timeseries_db.downsample(now) == {
+        "test": [
+            {
+                "bucket": "Bucket(start=2d, end=30d, resolution=4h)",
+                "start": "2023-11-01T00:00:00Z",
+                "end": "2023-11-29T00:00:00Z",
+                "data_points": 70,  # 24 hours (1h->4h) --> 0,4,8,12,16,20,24 ==> 7 with 10 entries each ==> 70
+            },
+            {
+                "bucket": "Bucket(start=30d, end=5mo25d, resolution=1d)",
+                "start": "2023-06-04T00:00:00Z",
+                "end": "2023-11-01T00:00:00Z",
+                "data_points": 50,  # 96 hours (4h->1d) --> 0,24,48,72,96 ==> 5 with 10 entries each ==> 50
+            },
+            {
+                "bucket": "Bucket(start=5mo25d, end=2yr, resolution=3d)",
+                "start": "2021-12-01T00:00:00Z",
+                "end": "2023-06-04T00:00:00Z",
+                "data_points": 90,  # 24 days (1d->3d) --> 0,3,6,9,12,15,18,21,24 ==> 9 with 10 entries each ==> 90
+            },
+        ]
+    }
+    assert timeseries_db.db.collection(timeseries_db.collection_name).count() == 450
+    assert await timeseries_db.downsample(now) == "No changes since last downsample run"
+    assert timeseries_db.db.collection(timeseries_db.collection_name).count() == 450
+    assert await timeseries_db.downsample(now=now + timedelta(days=27)) == {
+        "test": [
+            {
+                "bucket": "Bucket(start=2d, end=30d, resolution=4h)",
+                "start": "2023-11-29T00:00:00Z",
+                "end": "2023-12-26T00:00:00Z",
+                "data_points": 70,
+            },
+            {
+                "bucket": "Bucket(start=30d, end=5mo25d, resolution=1d)",
+                "start": "2023-11-01T00:00:00Z",
+                "end": "2023-11-28T00:00:00Z",
+                "data_points": 10,
+            },
+        ]
+    }
+    assert timeseries_db.db.collection(timeseries_db.collection_name).count() == 230
+    assert await timeseries_db.downsample(now=now + timedelta(days=200)) == {
+        "test": [
+            {
+                "bucket": "Bucket(start=5mo25d, end=2yr, resolution=3d)",
+                "start": "2023-06-04T00:00:00Z",
+                "end": "2023-12-21T00:00:00Z",
+                "data_points": 50,
+            }
+        ]
+    }
+    assert timeseries_db.db.collection(timeseries_db.collection_name).count() == 140
+    assert await timeseries_db.downsample(now=now + timedelta(days=400)) == {}
+    assert timeseries_db.db.collection(timeseries_db.collection_name).count() == 140