diff --git a/esphome/components/sensor/__init__.py b/esphome/components/sensor/__init__.py
index 1585a6342f..ec5bf1364d 100644
--- a/esphome/components/sensor/__init__.py
+++ b/esphome/components/sensor/__init__.py
@@ -453,8 +453,12 @@ async def skip_initial_filter_to_code(config, filter_id):
     return cg.new_Pvariable(filter_id, config)
 
 
-@FILTER_REGISTRY.register("min", MinFilter, MIN_SCHEMA)
-async def min_filter_to_code(config, filter_id):
+def _create_sliding_window_filter(config, filter_id, sliding_class, streaming_class):
+    """Helper to create sliding window or streaming filter based on config.
+
+    When window_size == send_every, use streaming filter (O(1) memory).
+    Otherwise, use sliding window filter (O(n) memory).
+    """
     window_size = config[CONF_WINDOW_SIZE]
     send_every = config[CONF_SEND_EVERY]
     send_first_at = config[CONF_SEND_FIRST_AT]
@@ -462,17 +466,14 @@ async def min_filter_to_code(config, filter_id):
     # Optimization: Use streaming filter for batch windows (window_size == send_every)
     # Saves 99.98% memory for large windows (e.g., 20KB → 4 bytes for window_size=5000)
     if window_size == send_every:
-        return cg.new_Pvariable(
-            filter_id,
-            StreamingMinFilter,
-            window_size,
-            send_first_at,
-        )
-    return cg.new_Pvariable(
-        filter_id,
-        config[CONF_WINDOW_SIZE],
-        config[CONF_SEND_EVERY],
-        config[CONF_SEND_FIRST_AT],
+        return cg.new_Pvariable(filter_id, streaming_class, window_size, send_first_at)
+    return cg.new_Pvariable(filter_id, window_size, send_every, send_first_at)
+
+
+@FILTER_REGISTRY.register("min", MinFilter, MIN_SCHEMA)
+async def min_filter_to_code(config, filter_id):
+    return _create_sliding_window_filter(
+        config, filter_id, MinFilter, StreamingMinFilter
     )
 
 
@@ -490,24 +491,8 @@ MAX_SCHEMA = cv.All(
 
 @FILTER_REGISTRY.register("max", MaxFilter, MAX_SCHEMA)
 async def max_filter_to_code(config, filter_id):
-    window_size = config[CONF_WINDOW_SIZE]
-    send_every = config[CONF_SEND_EVERY]
-    send_first_at = config[CONF_SEND_FIRST_AT]
-
-    # Optimization: Use streaming filter for batch windows (window_size == send_every)
-    # Saves 99.98% memory for large windows (e.g., 20KB → 4 bytes for window_size=5000)
-    if window_size == send_every:
-        return cg.new_Pvariable(
-            filter_id,
-            StreamingMaxFilter,
-            window_size,
-            send_first_at,
-        )
-    return cg.new_Pvariable(
-        filter_id,
-        config[CONF_WINDOW_SIZE],
-        config[CONF_SEND_EVERY],
-        config[CONF_SEND_FIRST_AT],
+    return _create_sliding_window_filter(
+        config, filter_id, MaxFilter, StreamingMaxFilter
     )
 
 
@@ -529,24 +514,11 @@ SLIDING_AVERAGE_SCHEMA = cv.All(
     SLIDING_AVERAGE_SCHEMA,
 )
 async def sliding_window_moving_average_filter_to_code(config, filter_id):
-    window_size = config[CONF_WINDOW_SIZE]
-    send_every = config[CONF_SEND_EVERY]
-    send_first_at = config[CONF_SEND_FIRST_AT]
-
-    # Optimization: Use streaming filter for batch windows (window_size == send_every)
-    # Saves 99.94% memory for large windows (e.g., 20KB → 12 bytes for window_size=5000)
-    if window_size == send_every:
-        return cg.new_Pvariable(
-            filter_id,
-            StreamingMovingAverageFilter,
-            window_size,
-            send_first_at,
-        )
-    return cg.new_Pvariable(
+    return _create_sliding_window_filter(
+        config,
         filter_id,
-        config[CONF_WINDOW_SIZE],
-        config[CONF_SEND_EVERY],
-        config[CONF_SEND_FIRST_AT],
+        SlidingWindowMovingAverageFilter,
+        StreamingMovingAverageFilter,
     )