diff --git a/homeassistant/components/recorder/db_schema.py b/homeassistant/components/recorder/db_schema.py
index dbe2b775297..fb57a1c73e2 100644
--- a/homeassistant/components/recorder/db_schema.py
+++ b/homeassistant/components/recorder/db_schema.py
@@ -691,12 +691,14 @@ class StatisticsBase:
     duration: timedelta
 
     @classmethod
-    def from_stats(cls, metadata_id: int, stats: StatisticData) -> Self:
+    def from_stats(
+        cls, metadata_id: int, stats: StatisticData, now_timestamp: float | None = None
+    ) -> Self:
         """Create object from a statistics with datetime objects."""
         return cls(  # type: ignore[call-arg]
             metadata_id=metadata_id,
             created=None,
-            created_ts=time.time(),
+            created_ts=now_timestamp or time.time(),
             start=None,
             start_ts=stats["start"].timestamp(),
             mean=stats.get("mean"),
@@ -709,12 +711,17 @@ class StatisticsBase:
         )
 
     @classmethod
-    def from_stats_ts(cls, metadata_id: int, stats: StatisticDataTimestamp) -> Self:
+    def from_stats_ts(
+        cls,
+        metadata_id: int,
+        stats: StatisticDataTimestamp,
+        now_timestamp: float | None = None,
+    ) -> Self:
         """Create object from a statistics with timestamps."""
         return cls(  # type: ignore[call-arg]
             metadata_id=metadata_id,
             created=None,
-            created_ts=time.time(),
+            created_ts=now_timestamp or time.time(),
             start=None,
             start_ts=stats["start_ts"],
             mean=stats.get("mean"),
diff --git a/homeassistant/components/recorder/statistics.py b/homeassistant/components/recorder/statistics.py
index 9f01fd0399c..3f1d5b981e3 100644
--- a/homeassistant/components/recorder/statistics.py
+++ b/homeassistant/components/recorder/statistics.py
@@ -11,6 +11,7 @@ from itertools import chain, groupby
 import logging
 from operator import itemgetter
 import re
+from time import time as time_time
 from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
 
 from sqlalchemy import Select, and_, bindparam, func, lambda_stmt, select, text
@@ -446,8 +447,9 @@ def _compile_hourly_statistics(session: Session, start: datetime) -> None:
                 }
 
     # Insert compiled hourly statistics in the database
+    now_timestamp = time_time()
     session.add_all(
-        Statistics.from_stats_ts(metadata_id, summary_item)
+        Statistics.from_stats_ts(metadata_id, summary_item, now_timestamp)
         for metadata_id, summary_item in summary.items()
     )
 
@@ -578,6 +580,7 @@ def _compile_statistics(
 
     new_short_term_stats: list[StatisticsBase] = []
     updated_metadata_ids: set[int] = set()
+    now_timestamp = time_time()
     # Insert collected statistics in the database
     for stats in platform_stats:
         modified_statistic_id, metadata_id = statistics_meta_manager.update_or_add(
@@ -587,10 +590,7 @@ def _compile_statistics(
             modified_statistic_ids.add(modified_statistic_id)
         updated_metadata_ids.add(metadata_id)
         if new_stat := _insert_statistics(
-            session,
-            StatisticsShortTerm,
-            metadata_id,
-            stats["stat"],
+            session, StatisticsShortTerm, metadata_id, stats["stat"], now_timestamp
         ):
             new_short_term_stats.append(new_stat)
 
@@ -666,10 +666,11 @@ def _insert_statistics(
     table: type[StatisticsBase],
     metadata_id: int,
     statistic: StatisticData,
+    now_timestamp: float,
 ) -> StatisticsBase | None:
     """Insert statistics in the database."""
     try:
-        stat = table.from_stats(metadata_id, statistic)
+        stat = table.from_stats(metadata_id, statistic, now_timestamp)
         session.add(stat)
     except SQLAlchemyError:
         _LOGGER.exception(
@@ -2347,11 +2348,12 @@ def _import_statistics_with_session(
     _, metadata_id = statistics_meta_manager.update_or_add(
         session, metadata, old_metadata_dict
     )
+    now_timestamp = time_time()
     for stat in statistics:
         if stat_id := _statistics_exists(session, table, metadata_id, stat["start"]):
             _update_statistics(session, table, stat_id, stat)
         else:
-            _insert_statistics(session, table, metadata_id, stat)
+            _insert_statistics(session, table, metadata_id, stat, now_timestamp)
 
     if table != StatisticsShortTerm:
         return True
diff --git a/tests/components/recorder/test_statistics.py b/tests/components/recorder/test_statistics.py
index bdf39c5ef4a..6b1e1a655db 100644
--- a/tests/components/recorder/test_statistics.py
+++ b/tests/components/recorder/test_statistics.py
@@ -337,12 +337,12 @@ def mock_from_stats():
     counter = 0
     real_from_stats = StatisticsShortTerm.from_stats
 
-    def from_stats(metadata_id, stats):
+    def from_stats(metadata_id, stats, now_timestamp):
         nonlocal counter
         if counter == 0 and metadata_id == 2:
             counter += 1
             return None
-        return real_from_stats(metadata_id, stats)
+        return real_from_stats(metadata_id, stats, now_timestamp)
 
     with patch(
         "homeassistant.components.recorder.statistics.StatisticsShortTerm.from_stats",