diff --git a/tasks/listing_tasks.py b/tasks/listing_tasks.py
index 51a70c0..1e4847c 100644
--- a/tasks/listing_tasks.py
+++ b/tasks/listing_tasks.py
@@ -635,27 +635,33 @@ async def _dump_listings_full_inner(
 
 @app.on_after_finalize.connect
 def setup_periodic_tasks(sender, **kwargs):
-    """Register periodic tasks from environment configuration."""
+    """Register periodic tasks from environment configuration.
+
+    The daily market aggregator is registered unconditionally so it stays
+    healthy even when the user's `SCRAPE_SCHEDULES` env var is malformed —
+    the two systems are orthogonal.
+    """
     try:
-        config = SchedulesConfig.from_env()
+        config: SchedulesConfig | None = SchedulesConfig.from_env()
     except ValueError as e:
         celery_logger.error(f"Failed to load schedule configuration: {e}")
-        return
+        config = None
 
-    for schedule in config.get_enabled_schedules():
-        celery_logger.info(
-            f"Registering periodic task: {schedule.name} at {schedule.hour}:{schedule.minute}"
-        )
+    if config is not None:
+        for schedule in config.get_enabled_schedules():
+            celery_logger.info(
+                f"Registering periodic task: {schedule.name} at {schedule.hour}:{schedule.minute}"
+            )
 
-        sender.add_periodic_task(
-            crontab(
-                minute=schedule.minute,
-                hour=schedule.hour,
-                day_of_week=schedule.day_of_week,
-            ),
-            dump_listings_task.s(schedule.to_query_parameters().model_dump_json()),
-            name=schedule.name,
-        )
+            sender.add_periodic_task(
+                crontab(
+                    minute=schedule.minute,
+                    hour=schedule.hour,
+                    day_of_week=schedule.day_of_week,
+                ),
+                dump_listings_task.s(schedule.to_query_parameters().model_dump_json()),
+                name=schedule.name,
+            )
 
     # Daily market aggregator — fires after the 03:00 RENT scrape so the
     # snapshot reflects today's freshly-pulled data. Imported lazily to
diff --git a/tests/unit/test_listing_tasks.py b/tests/unit/test_listing_tasks.py
index c044303..07a7de6 100644
--- a/tests/unit/test_listing_tasks.py
+++ b/tests/unit/test_listing_tasks.py
@@ -162,6 +162,10 @@ class TestDumpListingsTask:
 class TestSetupPeriodicTasks:
     """Tests for setup_periodic_tasks."""
 
+    # NOTE: every call to setup_periodic_tasks also registers the unconditional
+    # `daily-market-aggregator` task (one extra call per invocation), so
+    # call_count assertions below account for that +1.
+
     @patch("tasks.listing_tasks.SchedulesConfig.from_env")
     def test_registers_enabled_schedules(self, mock_from_env):
         from config.schedule_config import ScheduleConfig
@@ -180,18 +184,23 @@ class TestSetupPeriodicTasks:
         sender = MagicMock()
         module.setup_periodic_tasks(sender)
 
-        sender.add_periodic_task.assert_called_once()
-        call_args = sender.add_periodic_task.call_args
-        assert call_args[1]["name"] == "Test Schedule"
+        # 1 schedule + 1 market aggregator.
+        assert sender.add_periodic_task.call_count == 2
+        names = [c.kwargs["name"] for c in sender.add_periodic_task.call_args_list]
+        assert "Test Schedule" in names
+        assert "daily-market-aggregator" in names
 
     @patch("tasks.listing_tasks.SchedulesConfig.from_env")
     def test_handles_config_error_gracefully(self, mock_from_env):
+        """A malformed SCRAPE_SCHEDULES must not block the market aggregator."""
         mock_from_env.side_effect = ValueError("bad config")
 
         sender = MagicMock()
         module.setup_periodic_tasks(sender)
 
-        sender.add_periodic_task.assert_not_called()
+        # Aggregator still registers (the two systems are independent).
+        assert sender.add_periodic_task.call_count == 1
+        assert sender.add_periodic_task.call_args.kwargs["name"] == "daily-market-aggregator"
 
     @patch("tasks.listing_tasks.SchedulesConfig.from_env")
     def test_registers_nothing_when_no_schedules(self, mock_from_env):
@@ -202,7 +211,9 @@ class TestSetupPeriodicTasks:
         sender = MagicMock()
         module.setup_periodic_tasks(sender)
 
-        sender.add_periodic_task.assert_not_called()
+        # Only the market aggregator registered — no user schedules.
+        assert sender.add_periodic_task.call_count == 1
+        assert sender.add_periodic_task.call_args.kwargs["name"] == "daily-market-aggregator"
 
     @patch("tasks.listing_tasks.SchedulesConfig.from_env")
     def test_registers_multiple_schedules(self, mock_from_env):
@@ -220,7 +231,8 @@ class TestSetupPeriodicTasks:
         sender = MagicMock()
         module.setup_periodic_tasks(sender)
 
-        assert sender.add_periodic_task.call_count == 2
+        # 2 schedules + 1 market aggregator.
+        assert sender.add_periodic_task.call_count == 3
 
 
 class TestPipelineState: