diff --git a/homeassistant/components/recorder/__init__.py b/homeassistant/components/recorder/__init__.py
index 17215eb9845..d045726bc22 100644
--- a/homeassistant/components/recorder/__init__.py
+++ b/homeassistant/components/recorder/__init__.py
@@ -49,7 +49,7 @@ from homeassistant.helpers.typing import ConfigType
 from homeassistant.loader import bind_hass
 import homeassistant.util.dt as dt_util
 
-from . import history, migration, purge, statistics
+from . import history, migration, purge, statistics, websocket_api
 from .const import CONF_DB_INTEGRITY_CHECK, DATA_INSTANCE, DOMAIN, SQLITE_URL_PREFIX
 from .models import (
     Base,
@@ -264,6 +264,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     _async_register_services(hass, instance)
     history.async_setup(hass)
     statistics.async_setup(hass)
+    websocket_api.async_setup(hass)
     await async_process_integration_platforms(hass, DOMAIN, _process_recorder_platform)
 
     return await instance.async_db_ready
diff --git a/homeassistant/components/recorder/statistics.py b/homeassistant/components/recorder/statistics.py
index c1b924ceeec..6ed612c60ad 100644
--- a/homeassistant/components/recorder/statistics.py
+++ b/homeassistant/components/recorder/statistics.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from collections import defaultdict
+import dataclasses
 from datetime import datetime, timedelta
 from itertools import groupby
 import logging
@@ -91,6 +92,18 @@ UNIT_CONVERSIONS = {
 _LOGGER = logging.getLogger(__name__)
 
 
+@dataclasses.dataclass
+class ValidationIssue:
+    """Error or warning message."""
+
+    type: str
+    data: dict[str, str | None] | None = None
+
+    def as_dict(self) -> dict:
+        """Return dictionary version."""
+        return dataclasses.asdict(self)
+
+
 def async_setup(hass: HomeAssistant) -> None:
     """Set up the history hooks."""
     hass.data[STATISTICS_BAKERY] = baked.bakery()
@@ -471,3 +484,13 @@ def _sorted_statistics_to_dict(
 
     # Filter out the empty lists if some states had 0 results.
     return {metadata[key]["statistic_id"]: val for key, val in result.items() if val}
+
+
+def validate_statistics(hass: HomeAssistant) -> dict[str, list[ValidationIssue]]:
+    """Validate statistics."""
+    platform_validation: dict[str, list[ValidationIssue]] = {}
+    for platform in hass.data[DOMAIN].values():
+        if not hasattr(platform, "validate_statistics"):
+            continue
+        platform_validation.update(platform.validate_statistics(hass))
+    return platform_validation
diff --git a/homeassistant/components/recorder/websocket_api.py b/homeassistant/components/recorder/websocket_api.py
new file mode 100644
index 00000000000..c5a332547cb
--- /dev/null
+++ b/homeassistant/components/recorder/websocket_api.py
@@ -0,0 +1,30 @@
+"""The Energy websocket API."""
+import voluptuous as vol
+
+from homeassistant.components import websocket_api
+from homeassistant.core import HomeAssistant, callback
+
+from .statistics import validate_statistics
+
+
+@callback
+def async_setup(hass: HomeAssistant) -> None:
+    """Set up the recorder websocket API."""
+    websocket_api.async_register_command(hass, ws_validate_statistics)
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "recorder/validate_statistics",
+    }
+)
+@websocket_api.async_response
+async def ws_validate_statistics(
+    hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
+) -> None:
+    """Fetch a list of available statistic_id."""
+    statistic_ids = await hass.async_add_executor_job(
+        validate_statistics,
+        hass,
+    )
+    connection.send_result(msg["id"], statistic_ids)
diff --git a/homeassistant/components/sensor/recorder.py b/homeassistant/components/sensor/recorder.py
index 7e3fb5ddd9f..02269439cb8 100644
--- a/homeassistant/components/sensor/recorder.py
+++ b/homeassistant/components/sensor/recorder.py
@@ -1,6 +1,7 @@
 """Statistics helper for sensor."""
 from __future__ import annotations
 
+from collections import defaultdict
 import datetime
 import itertools
 import logging
@@ -543,3 +544,56 @@ def list_statistic_ids(hass: HomeAssistant, statistic_type: str | None = None) -
         statistic_ids[entity_id] = statistics_unit
 
     return statistic_ids
+
+
+def validate_statistics(
+    hass: HomeAssistant,
+) -> dict[str, list[statistics.ValidationIssue]]:
+    """Validate statistics."""
+    validation_result = defaultdict(list)
+
+    entities = _get_entities(hass)
+
+    for (
+        entity_id,
+        _state_class,
+        device_class,
+    ) in entities:
+        state = hass.states.get(entity_id)
+        assert state is not None
+
+        metadata = statistics.get_metadata(hass, entity_id)
+        if not metadata:
+            continue
+
+        state_unit = state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
+        metadata_unit = metadata["unit_of_measurement"]
+
+        if device_class not in UNIT_CONVERSIONS:
+
+            if state_unit != metadata_unit:
+                validation_result[entity_id].append(
+                    statistics.ValidationIssue(
+                        "units_changed",
+                        {
+                            "statistic_id": entity_id,
+                            "state_unit": state_unit,
+                            "metadata_unit": metadata_unit,
+                        },
+                    )
+                )
+            continue
+
+        if state_unit not in UNIT_CONVERSIONS[device_class]:
+            validation_result[entity_id].append(
+                statistics.ValidationIssue(
+                    "unsupported_unit",
+                    {
+                        "statistic_id": entity_id,
+                        "device_class": device_class,
+                        "state_unit": state_unit,
+                    },
+                )
+            )
+
+    return validation_result
diff --git a/tests/components/recorder/test_websocket_api.py b/tests/components/recorder/test_websocket_api.py
new file mode 100644
index 00000000000..51334432121
--- /dev/null
+++ b/tests/components/recorder/test_websocket_api.py
@@ -0,0 +1,242 @@
+"""The tests for sensor recorder platform."""
+# pylint: disable=protected-access,invalid-name
+from datetime import timedelta
+
+import pytest
+
+from homeassistant.components.recorder.const import DATA_INSTANCE
+from homeassistant.components.recorder.models import StatisticsMeta
+from homeassistant.components.recorder.util import session_scope
+from homeassistant.setup import async_setup_component
+import homeassistant.util.dt as dt_util
+from homeassistant.util.unit_system import IMPERIAL_SYSTEM, METRIC_SYSTEM
+
+from tests.common import init_recorder_component
+
+BATTERY_SENSOR_ATTRIBUTES = {
+    "device_class": "battery",
+    "state_class": "measurement",
+    "unit_of_measurement": "%",
+}
+POWER_SENSOR_ATTRIBUTES = {
+    "device_class": "power",
+    "state_class": "measurement",
+    "unit_of_measurement": "kW",
+}
+NONE_SENSOR_ATTRIBUTES = {
+    "state_class": "measurement",
+}
+PRESSURE_SENSOR_ATTRIBUTES = {
+    "device_class": "pressure",
+    "state_class": "measurement",
+    "unit_of_measurement": "hPa",
+}
+TEMPERATURE_SENSOR_ATTRIBUTES = {
+    "device_class": "temperature",
+    "state_class": "measurement",
+    "unit_of_measurement": "°C",
+}
+
+
+@pytest.mark.parametrize(
+    "units, attributes, unit",
+    [
+        (IMPERIAL_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
+        (METRIC_SYSTEM, POWER_SENSOR_ATTRIBUTES, "W"),
+        (IMPERIAL_SYSTEM, TEMPERATURE_SENSOR_ATTRIBUTES, "°F"),
+        (METRIC_SYSTEM, TEMPERATURE_SENSOR_ATTRIBUTES, "°C"),
+        (IMPERIAL_SYSTEM, PRESSURE_SENSOR_ATTRIBUTES, "psi"),
+        (METRIC_SYSTEM, PRESSURE_SENSOR_ATTRIBUTES, "Pa"),
+    ],
+)
+async def test_validate_statistics_supported_device_class(
+    hass, hass_ws_client, units, attributes, unit
+):
+    """Test list_statistic_ids."""
+    id = 1
+
+    def next_id():
+        nonlocal id
+        id += 1
+        return id
+
+    async def assert_validation_result(client, expected_result):
+        await client.send_json(
+            {"id": next_id(), "type": "recorder/validate_statistics"}
+        )
+        response = await client.receive_json()
+        assert response["success"]
+        assert response["result"] == expected_result
+
+    now = dt_util.utcnow()
+
+    hass.config.units = units
+    await hass.async_add_executor_job(init_recorder_component, hass)
+    await async_setup_component(hass, "sensor", {})
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    client = await hass_ws_client()
+
+    # No statistics, no state - empty response
+    await assert_validation_result(client, {})
+
+    # No statistics, valid state - empty response
+    hass.states.async_set(
+        "sensor.test", 10, attributes={**attributes, **{"unit_of_measurement": unit}}
+    )
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_validation_result(client, {})
+
+    # No statistics, invalid state - empty response
+    hass.states.async_set(
+        "sensor.test", 11, attributes={**attributes, **{"unit_of_measurement": "dogs"}}
+    )
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_validation_result(client, {})
+
+    # Statistics has run, invalid state - expect error
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    hass.data[DATA_INSTANCE].do_adhoc_statistics(start=now)
+    hass.states.async_set(
+        "sensor.test", 12, attributes={**attributes, **{"unit_of_measurement": "dogs"}}
+    )
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    expected = {
+        "sensor.test": [
+            {
+                "data": {
+                    "device_class": attributes["device_class"],
+                    "state_unit": "dogs",
+                    "statistic_id": "sensor.test",
+                },
+                "type": "unsupported_unit",
+            }
+        ],
+    }
+    await assert_validation_result(client, expected)
+
+    # Valid state - empty response
+    hass.states.async_set(
+        "sensor.test", 13, attributes={**attributes, **{"unit_of_measurement": unit}}
+    )
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_validation_result(client, {})
+
+    # Valid state, statistic runs again - empty response
+    hass.data[DATA_INSTANCE].do_adhoc_statistics(start=now)
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_validation_result(client, {})
+
+    # Remove the state - empty response
+    hass.states.async_remove("sensor.test")
+    await assert_validation_result(client, {})
+
+
+@pytest.mark.parametrize(
+    "attributes",
+    [BATTERY_SENSOR_ATTRIBUTES, NONE_SENSOR_ATTRIBUTES],
+)
+async def test_validate_statistics_unsupported_device_class(
+    hass, hass_ws_client, attributes
+):
+    """Test list_statistic_ids."""
+    id = 1
+
+    def next_id():
+        nonlocal id
+        id += 1
+        return id
+
+    async def assert_validation_result(client, expected_result):
+        await client.send_json(
+            {"id": next_id(), "type": "recorder/validate_statistics"}
+        )
+        response = await client.receive_json()
+        assert response["success"]
+        assert response["result"] == expected_result
+
+    async def assert_statistic_ids(expected_result):
+        with session_scope(hass=hass) as session:
+            db_states = list(session.query(StatisticsMeta))
+            assert len(db_states) == len(expected_result)
+            for i in range(len(db_states)):
+                assert db_states[i].statistic_id == expected_result[i]["statistic_id"]
+                assert (
+                    db_states[i].unit_of_measurement
+                    == expected_result[i]["unit_of_measurement"]
+                )
+
+    now = dt_util.utcnow()
+
+    await hass.async_add_executor_job(init_recorder_component, hass)
+    await async_setup_component(hass, "sensor", {})
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    client = await hass_ws_client()
+    rec = hass.data[DATA_INSTANCE]
+
+    # No statistics, no state - empty response
+    await assert_validation_result(client, {})
+
+    # No statistics, original unit - empty response
+    hass.states.async_set("sensor.test", 10, attributes=attributes)
+    await assert_validation_result(client, {})
+
+    # No statistics, changed unit - empty response
+    hass.states.async_set(
+        "sensor.test", 11, attributes={**attributes, **{"unit_of_measurement": "dogs"}}
+    )
+    await assert_validation_result(client, {})
+
+    # Run statistics, no statistics will be generated because of conflicting units
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    rec.do_adhoc_statistics(start=now)
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_statistic_ids([])
+
+    # No statistics, changed unit - empty response
+    hass.states.async_set(
+        "sensor.test", 12, attributes={**attributes, **{"unit_of_measurement": "dogs"}}
+    )
+    await assert_validation_result(client, {})
+
+    # Run statistics one hour later, only the "dogs" state will be considered
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    rec.do_adhoc_statistics(start=now + timedelta(hours=1))
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_statistic_ids(
+        [{"statistic_id": "sensor.test", "unit_of_measurement": "dogs"}]
+    )
+    await assert_validation_result(client, {})
+
+    # Change back to original unit - expect error
+    hass.states.async_set("sensor.test", 13, attributes=attributes)
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    expected = {
+        "sensor.test": [
+            {
+                "data": {
+                    "metadata_unit": "dogs",
+                    "state_unit": attributes.get("unit_of_measurement"),
+                    "statistic_id": "sensor.test",
+                },
+                "type": "units_changed",
+            }
+        ],
+    }
+    await assert_validation_result(client, expected)
+
+    # Changed unit - empty response
+    hass.states.async_set(
+        "sensor.test", 14, attributes={**attributes, **{"unit_of_measurement": "dogs"}}
+    )
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_validation_result(client, {})
+
+    # Valid state, statistic runs again - empty response
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    hass.data[DATA_INSTANCE].do_adhoc_statistics(start=now)
+    await hass.async_add_executor_job(hass.data[DATA_INSTANCE].block_till_done)
+    await assert_validation_result(client, {})
+
+    # Remove the state - empty response
+    hass.states.async_remove("sensor.test")
+    await assert_validation_result(client, {})