diff --git a/homeassistant/core.py b/homeassistant/core.py
index bfb88ab6bfd5a7..8bc37f1f6bba64 100644
--- a/homeassistant/core.py
+++ b/homeassistant/core.py
@@ -538,7 +538,7 @@ def __init__(
         event_type: str,
         data: Optional[Dict[str, Any]] = None,
         origin: EventOrigin = EventOrigin.local,
-        time_fired: Optional[int] = None,
+        time_fired: Optional[datetime.datetime] = None,
         context: Optional[Context] = None,
     ) -> None:
         """Initialize a new event."""
diff --git a/homeassistant/helpers/event.py b/homeassistant/helpers/event.py
index d7b1f171a5128c..68af250a96ff77 100644
--- a/homeassistant/helpers/event.py
+++ b/homeassistant/helpers/event.py
@@ -74,10 +74,12 @@ class TrackTemplate:
 
     The template is template to calculate.
     The variables are variables to pass to the template.
+    The rate_limit is a rate limit on how often the template is re-rendered.
     """
 
     template: Template
     variables: TemplateVarsType
+    rate_limit: Optional[timedelta] = None
 
 
 @dataclass
@@ -560,7 +562,8 @@ def __init__(
         self._listeners: Dict[str, Callable] = {}
 
         self._last_result: Dict[Template, Union[str, TemplateError]] = {}
-        self._last_info: Dict[Template, RenderInfo] = {}
+        self._last_rendered: Dict[Template, datetime] = {}
+        self._rate_limit_timers: Dict[Template, asyncio.TimerHandle] = {}
         self._info: Dict[Template, RenderInfo] = {}
         self._last_domains: Set = set()
         self._last_entities: Set = set()
@@ -581,7 +584,6 @@ def async_setup(self, raise_on_template_error: bool) -> None:
                     exc_info=self._info[template].exception,
                 )
 
-        self._last_info = self._info.copy()
         self._create_listeners()
         _LOGGER.debug(
             "Template group %s listens for %s",
@@ -600,28 +602,23 @@ def listeners(self) -> Dict:
 
     @property
     def _needs_all_listener(self) -> bool:
-        for track_template_ in self._track_templates:
-            template = track_template_.template
-
+        for info in self._info.values():
             # Tracking all states
-            if (
-                self._info[template].all_states
-                or self._info[template].all_states_lifecycle
-            ):
+            if info.all_states or info.all_states_lifecycle:
                 return True
 
             # Previous call had an exception
             # so we do not know which states
             # to track
-            if self._info[template].exception:
+            if info.exception:
                 return True
 
         return False
 
     @property
     def _all_templates_are_static(self) -> bool:
-        for track_template_ in self._track_templates:
-            if not self._info[track_template_.template].is_static:
+        for info in self._info.values():
+            if not info.is_static:
                 return False
 
         return True
@@ -648,6 +645,13 @@ def _cancel_listener(self, listener_name: str) -> None:
 
         self._listeners.pop(listener_name)()
 
+    @callback
+    def _cancel_rate_limit_timer(self, template: Template) -> None:
+        if template not in self._rate_limit_timers:
+            return
+
+        self._rate_limit_timers.pop(template).cancel()
+
     @callback
     def _update_listeners(self) -> None:
         had_all_listener = _TEMPLATE_ALL_LISTENER in self._listeners
@@ -712,9 +716,10 @@ def _setup_all_listener(self) -> None:
     @callback
     def async_remove(self) -> None:
         """Cancel the listener."""
-        self._cancel_listener(_TEMPLATE_ALL_LISTENER)
-        self._cancel_listener(_TEMPLATE_DOMAINS_LISTENER)
-        self._cancel_listener(_TEMPLATE_ENTITIES_LISTENER)
+        for key in list(self._listeners):
+            self._listeners.pop(key)()
+        for track_template_ in self._track_templates:
+            self._cancel_rate_limit_timer(track_template_.template)
 
     @callback
     def async_refresh(self) -> None:
@@ -722,32 +727,86 @@ def async_refresh(self) -> None:
         self._refresh(None)
 
     @callback
-    def _refresh(self, event: Optional[Event]) -> None:
-        entity_id = event and event.data.get(ATTR_ENTITY_ID)
-        lifecycle_event = event and (
-            event.data.get("new_state") is None or event.data.get("old_state") is None
+    def _handle_rate_limit(
+        self,
+        track_template_: TrackTemplate,
+        event: Event,
+        now: datetime,
+    ) -> bool:
+        """Check rate limits and call later if the rate limit is hit.
+
+        If there is already a call later scheduled for the template
+        we do not setup a second one.
+
+        Returns True if the rate limit has been hit or False on miss.
+        """
+        template = track_template_.template
+        rate_limit = self._info[template].rate_limit or track_template_.rate_limit
+
+        if not rate_limit or template not in self._last_rendered:
+            return False
+
+        next_render_time = self._last_rendered[template] + rate_limit
+
+        if next_render_time <= now:
+            self._cancel_rate_limit_timer(template)
+            return False
+
+        _LOGGER.debug(
+            "Template rate_limit %s hit by event %s deferred by rate_limit %s to %s",
+            template.template,
+            event,
+            rate_limit,
+            next_render_time,
+        )
+
+        if template not in self._rate_limit_timers:
+            self._rate_limit_timers[template] = self.hass.loop.call_later(
+                (next_render_time - now).total_seconds() + MAX_TIME_TRACKING_ERROR,
+                self._refresh,
+                event,
+            )
+
+        return True
+
+    @callback
+    def _event_triggers_template(self, template: Template, event: Event) -> bool:
+        """Determine if a template should be re-rendered from an event."""
+        entity_id = event.data.get(ATTR_ENTITY_ID)
+        return (
+            self._info[template].filter(entity_id)
+            or event.data.get("new_state") is None
+            or event.data.get("old_state") is None
+            and self._info[template].filter_lifecycle(entity_id)
         )
+
+    @callback
+    def _refresh(self, event: Optional[Event]) -> None:
         updates = []
         info_changed = False
+        now = dt_util.utcnow()
 
         for track_template_ in self._track_templates:
             template = track_template_.template
-            if (
-                entity_id
-                and not self._last_info[template].filter(entity_id)
-                and (
-                    not lifecycle_event
-                    or not self._last_info[template].filter_lifecycle(entity_id)
+            if event:
+                if (
+                    template not in self._rate_limit_timers
+                    and not self._event_triggers_template(template, event)
+                ):
+                    continue
+
+                if self._handle_rate_limit(track_template_, event, now):
+                    continue
+
+                _LOGGER.debug(
+                    "Template update %s triggered by event: %s",
+                    template.template,
+                    event,
                 )
-            ):
-                continue
-
-            _LOGGER.debug(
-                "Template update %s triggered by event: %s",
-                template.template,
-                event,
-            )
+            else:
+                self._cancel_rate_limit_timer(template)
 
+            self._last_rendered[template] = now
             self._info[template] = template.async_render_to_info(
                 track_template_.variables
             )
@@ -778,7 +837,6 @@ def _refresh(self, event: Optional[Event]) -> None:
                 self._track_templates,
                 self.listeners,
             )
-            self._last_info = self._info.copy()
 
         if not updates:
             return
diff --git a/homeassistant/helpers/template.py b/homeassistant/helpers/template.py
index 721c1407f3785e..70151c7a51977e 100644
--- a/homeassistant/helpers/template.py
+++ b/homeassistant/helpers/template.py
@@ -187,10 +187,11 @@ def __init__(self, template):
         self.domains = set()
         self.domains_lifecycle = set()
         self.entities = set()
+        self.rate_limit = None
 
     def __repr__(self) -> str:
         """Representation of RenderInfo."""
-        return f"<RenderInfo {self.template} all_states={self.all_states} all_states_lifecycle={self.all_states_lifecycle} domains={self.domains} domains_lifecycle={self.domains_lifecycle} entities={self.entities}>"
+        return f"<RenderInfo {self.template} all_states={self.all_states} all_states_lifecycle={self.all_states_lifecycle} domains={self.domains} domains_lifecycle={self.domains_lifecycle} entities={self.entities} rate_limit={self.rate_limit}>"
 
     def _filter_domains_and_entities(self, entity_id: str) -> bool:
         """Template should re-render if the entity state changes when we match specific domains or entities."""
@@ -467,6 +468,28 @@ def __repr__(self) -> str:
         return 'Template("' + self.template + '")'
 
 
+class RateLimit:
+    """Class to control update rate limits."""
+
+    def __init__(self, hass: HomeAssistantType):
+        """Initialize rate limit."""
+        self._hass = hass
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Optional[timedelta]:
+        """Handle a call to the class."""
+        delta = timedelta(*args, **kwargs)
+
+        render_info = self._hass.data.get(_RENDER_INFO)
+        if render_info is not None:
+            render_info.rate_limit = delta
+
+        return delta
+
+    def __repr__(self) -> str:
+        """Representation of a RateLimit."""
+        return "<template RateLimit>"
+
+
 class AllStates:
     """Class to expose all HA states as attributes."""
 
@@ -1201,10 +1224,11 @@ def wrapper(*args, **kwargs):
         self.globals["is_state_attr"] = hassfunction(is_state_attr)
         self.globals["state_attr"] = hassfunction(state_attr)
         self.globals["states"] = AllStates(hass)
+        self.globals["rate_limit"] = RateLimit(hass)
 
     def is_safe_callable(self, obj):
         """Test if callback is safe."""
-        return isinstance(obj, AllStates) or super().is_safe_callable(obj)
+        return isinstance(obj, (AllStates, RateLimit)) or super().is_safe_callable(obj)
 
     def is_safe_attribute(self, obj, attr, value):
         """Test if attribute is safe."""
diff --git a/tests/helpers/test_event.py b/tests/helpers/test_event.py
index 8bdf9cb891c390..94a8a3f32b9a80 100644
--- a/tests/helpers/test_event.py
+++ b/tests/helpers/test_event.py
@@ -1259,6 +1259,298 @@ def refresh_listener(event, updates):
     assert refresh_runs == ["static"]
 
 
+async def test_track_template_rate_limit(hass):
+    """Test template rate limit."""
+    template_refresh = Template("{{ states | count }}", hass)
+
+    refresh_runs = []
+
+    @ha.callback
+    def refresh_listener(event, updates):
+        refresh_runs.append(updates.pop().result)
+
+    info = async_track_template_result(
+        hass,
+        [TrackTemplate(template_refresh, None, timedelta(seconds=1))],
+        refresh_listener,
+    )
+    await hass.async_block_till_done()
+    info.async_refresh()
+    await hass.async_block_till_done()
+
+    assert refresh_runs == ["0"]
+    hass.states.async_set("sensor.one", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0"]
+    info.async_refresh()
+    assert refresh_runs == ["0", "1"]
+    hass.states.async_set("sensor.two", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.three", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.four", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "4"]
+    hass.states.async_set("sensor.five", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "4"]
+
+
+async def test_track_template_rate_limit_overridden(hass):
+    """Test template rate limit can be overridden from the template."""
+    template_refresh = Template(
+        "{% set x = rate_limit(seconds=1) %}{{ states | count }}", hass
+    )
+
+    refresh_runs = []
+
+    @ha.callback
+    def refresh_listener(event, updates):
+        refresh_runs.append(updates.pop().result)
+
+    info = async_track_template_result(
+        hass,
+        [TrackTemplate(template_refresh, None, timedelta(seconds=5))],
+        refresh_listener,
+    )
+    await hass.async_block_till_done()
+    info.async_refresh()
+    await hass.async_block_till_done()
+
+    assert refresh_runs == ["0"]
+    hass.states.async_set("sensor.one", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0"]
+    info.async_refresh()
+    assert refresh_runs == ["0", "1"]
+    hass.states.async_set("sensor.two", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.three", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.four", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "4"]
+    hass.states.async_set("sensor.five", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "4"]
+
+
+async def test_track_template_rate_limit_five(hass):
+    """Test template rate limit of 5 seconds."""
+    template_refresh = Template("{{ states | count }}", hass)
+
+    refresh_runs = []
+
+    @ha.callback
+    def refresh_listener(event, updates):
+        refresh_runs.append(updates.pop().result)
+
+    info = async_track_template_result(
+        hass,
+        [TrackTemplate(template_refresh, None, timedelta(seconds=5))],
+        refresh_listener,
+    )
+    await hass.async_block_till_done()
+    info.async_refresh()
+    await hass.async_block_till_done()
+
+    assert refresh_runs == ["0"]
+    hass.states.async_set("sensor.one", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0"]
+    info.async_refresh()
+    assert refresh_runs == ["0", "1"]
+    hass.states.async_set("sensor.two", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1"]
+    hass.states.async_set("sensor.three", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1"]
+
+
+async def test_track_template_rate_limit_changes(hass):
+    """Test template rate limit can be changed."""
+    template_refresh = Template(
+        """
+        {% if states.sensor.two.state == "any" %}
+            {% set x = rate_limit(seconds=5) %}
+        {% else %}
+            {% set x = rate_limit(seconds=1) %}
+        {% endif %}
+        {{ states | count }}
+        """,
+        hass,
+    )
+
+    refresh_runs = []
+
+    @ha.callback
+    def refresh_listener(event, updates):
+        refresh_runs.append(updates.pop().result)
+
+    info = async_track_template_result(
+        hass,
+        [TrackTemplate(template_refresh, None)],
+        refresh_listener,
+    )
+    await hass.async_block_till_done()
+    info.async_refresh()
+    await hass.async_block_till_done()
+
+    assert refresh_runs == ["0"]
+    hass.states.async_set("sensor.one", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0"]
+    info.async_refresh()
+    assert refresh_runs == ["0", "1"]
+    hass.states.async_set("sensor.two", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.three", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.four", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.five", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+
+
+async def test_track_template_rate_limit_removed(hass):
+    """Test template rate limit can be removed."""
+    template_refresh = Template(
+        """
+        {% if states.sensor.two.state == "any" %}
+            {% set x = rate_limit(0) %}
+        {% else %}
+            {% set x = rate_limit(seconds=1) %}
+        {% endif %}
+        {{ states | count }}
+        """,
+        hass,
+    )
+
+    refresh_runs = []
+
+    @ha.callback
+    def refresh_listener(event, updates):
+        refresh_runs.append(updates.pop().result)
+
+    info = async_track_template_result(
+        hass,
+        [TrackTemplate(template_refresh, None)],
+        refresh_listener,
+    )
+    await hass.async_block_till_done()
+    info.async_refresh()
+    await hass.async_block_till_done()
+
+    assert refresh_runs == ["0"]
+    hass.states.async_set("sensor.one", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0"]
+    info.async_refresh()
+    assert refresh_runs == ["0", "1"]
+    hass.states.async_set("sensor.two", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2"]
+    hass.states.async_set("sensor.three", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "3"]
+    hass.states.async_set("sensor.four", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "3", "4"]
+    hass.states.async_set("sensor.five", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs == ["0", "1", "2", "3", "4", "5"]
+
+
+async def test_track_two_templates_with_different_rate_limits(hass):
+    """Test two templates with different rate limits."""
+    template_one = Template(
+        "{% set x = rate_limit(seconds=1) %}{{ states | count }}", hass
+    )
+    template_five = Template(
+        "{% set x = rate_limit(seconds=5) %}{{ states | count }}", hass
+    )
+
+    refresh_runs = {
+        template_one: [],
+        template_five: [],
+    }
+
+    @ha.callback
+    def refresh_listener(event, updates):
+        for update in updates:
+            refresh_runs[update.template].append(update.result)
+
+    info = async_track_template_result(
+        hass,
+        [TrackTemplate(template_one, None), TrackTemplate(template_five, None)],
+        refresh_listener,
+    )
+
+    await hass.async_block_till_done()
+    info.async_refresh()
+    await hass.async_block_till_done()
+
+    assert refresh_runs[template_one] == ["0"]
+    assert refresh_runs[template_five] == ["0"]
+    hass.states.async_set("sensor.one", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs[template_one] == ["0"]
+    assert refresh_runs[template_five] == ["0"]
+    info.async_refresh()
+    assert refresh_runs[template_one] == ["0", "1"]
+    assert refresh_runs[template_five] == ["0", "1"]
+    hass.states.async_set("sensor.two", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs[template_one] == ["0", "1"]
+    assert refresh_runs[template_five] == ["0", "1"]
+    await asyncio.sleep(1.2)
+    await hass.async_block_till_done()
+    assert refresh_runs[template_one] == ["0", "1", "2"]
+    assert refresh_runs[template_five] == ["0", "1"]
+    hass.states.async_set("sensor.three", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs[template_one] == ["0", "1", "2"]
+    assert refresh_runs[template_five] == ["0", "1"]
+    hass.states.async_set("sensor.four", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs[template_one] == ["0", "1", "2"]
+    assert refresh_runs[template_five] == ["0", "1"]
+    hass.states.async_set("sensor.five", "any")
+    await hass.async_block_till_done()
+    assert refresh_runs[template_one] == ["0", "1", "2"]
+    assert refresh_runs[template_five] == ["0", "1"]
+
+
 async def test_string(hass):
     """Test a string."""
     template_refresh = Template("no_template", hass)
diff --git a/tests/helpers/test_template.py b/tests/helpers/test_template.py
index be6b1bd2ecf05e..175660d5026901 100644
--- a/tests/helpers/test_template.py
+++ b/tests/helpers/test_template.py
@@ -1,5 +1,5 @@
 """Test Home Assistant template helper methods."""
-from datetime import datetime
+from datetime import datetime, timedelta
 import math
 import random
 
@@ -2483,3 +2483,18 @@ async def test_template_timeout(hass):
 """
     tmp5 = template.Template(slow_template_str, hass)
     assert await tmp5.async_render_will_timeout(0.000001) is True
+
+
+async def test_rate_limit(hass):
+    """Test we can pickup a rate limit directive."""
+    tmp = template.Template("{{ states | count }}", hass)
+
+    info = tmp.async_render_to_info()
+    assert info.rate_limit is None
+
+    tmp = template.Template(
+        "{% set x = rate_limit(minutes=1) %}{{ states | count }}", hass
+    )
+
+    info = tmp.async_render_to_info()
+    assert info.rate_limit == timedelta(minutes=1)