diff --git a/homeassistant/components/history/__init__.py b/homeassistant/components/history/__init__.py
index 3ae71602dc..5120fb2f10 100644
--- a/homeassistant/components/history/__init__.py
+++ b/homeassistant/components/history/__init__.py
@@ -16,6 +16,7 @@ from homeassistant.components import websocket_api
 from homeassistant.components.http import HomeAssistantView
 from homeassistant.components.recorder import history, models as history_models
 from homeassistant.components.recorder.statistics import (
+    clear_statistics,
     list_statistic_ids,
     statistics_during_period,
 )
@@ -104,6 +105,7 @@ async def async_setup(hass, config):
         ws_get_statistics_during_period
     )
     hass.components.websocket_api.async_register_command(ws_get_list_statistic_ids)
+    hass.components.websocket_api.async_register_command(ws_clear_statistics)
 
     return True
 
@@ -177,6 +179,25 @@ async def ws_get_list_statistic_ids(
     connection.send_result(msg["id"], statistic_ids)
 
 
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "history/clear_statistics",
+        vol.Required("statistic_ids"): [str],
+    }
+)
+@websocket_api.async_response
+async def ws_clear_statistics(
+    hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
+) -> None:
+    """Clear statistics for a list of statistic_ids."""
+    await hass.async_add_executor_job(
+        clear_statistics,
+        hass,
+        msg["statistic_ids"],
+    )
+    connection.send_result(msg["id"])
+
+
 class HistoryPeriodView(HomeAssistantView):
     """Handle history period requests."""
 
diff --git a/homeassistant/components/recorder/statistics.py b/homeassistant/components/recorder/statistics.py
index 74d27282ea..66e96ed4a0 100644
--- a/homeassistant/components/recorder/statistics.py
+++ b/homeassistant/components/recorder/statistics.py
@@ -458,6 +458,18 @@ def _configured_unit(unit: str, units: UnitSystem) -> str:
     return unit
 
 
+def clear_statistics(hass: HomeAssistant, statistic_ids: list[str]) -> None:
+    """Clear statistics for a list of statistic_ids."""
+    with session_scope(hass=hass) as session:
+        metadata_ids = _get_metadata_ids(hass, session, statistic_ids)
+        session.query(Statistics).filter(
+            Statistics.metadata_id.in_(metadata_ids)
+        ).delete(synchronize_session=False)
+        session.query(StatisticsMeta).filter(
+            StatisticsMeta.id.in_(metadata_ids)
+        ).delete(synchronize_session=False)
+
+
 def list_statistic_ids(
     hass: HomeAssistant,
     statistic_type: Literal["mean"] | Literal["sum"] | None = None,
diff --git a/tests/components/history/test_init.py b/tests/components/history/test_init.py
index 661d703725..260b22d631 100644
--- a/tests/components/history/test_init.py
+++ b/tests/components/history/test_init.py
@@ -1048,3 +1048,120 @@ async def test_list_statistic_ids(hass, hass_ws_client, units, attributes, unit)
     response = await client.receive_json()
     assert response["success"]
     assert response["result"] == []
+
+
+async def test_clear_statistics_during_period(hass, hass_ws_client):
+    """Test removing statistics."""
+    now = dt_util.utcnow()
+
+    units = METRIC_SYSTEM
+    attributes = POWER_SENSOR_ATTRIBUTES
+    state = 10
+    value = 10000
+
+    hass.config.units = units
+    await hass.async_add_executor_job(init_recorder_component, hass)
+    await async_setup_component(hass, "history", {})
+    await async_setup_component(hass, "sensor", {})
+    await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
+    hass.states.async_set("sensor.test1", state, attributes=attributes)
+    hass.states.async_set("sensor.test2", state * 2, attributes=attributes)
+    hass.states.async_set("sensor.test3", state * 3, attributes=attributes)
+    await hass.async_block_till_done()
+
+    await hass.async_add_executor_job(trigger_db_commit, hass)
+    await hass.async_block_till_done()
+
+    hass.data[recorder.DATA_INSTANCE].do_adhoc_statistics(period="hourly", start=now)
+    await hass.async_add_executor_job(hass.data[recorder.DATA_INSTANCE].block_till_done)
+
+    client = await hass_ws_client()
+    await client.send_json(
+        {
+            "id": 1,
+            "type": "history/statistics_during_period",
+            "start_time": now.isoformat(),
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    expected_response = {
+        "sensor.test1": [
+            {
+                "statistic_id": "sensor.test1",
+                "start": now.isoformat(),
+                "mean": approx(value),
+                "min": approx(value),
+                "max": approx(value),
+                "state": None,
+                "sum": None,
+            }
+        ],
+        "sensor.test2": [
+            {
+                "statistic_id": "sensor.test2",
+                "start": now.isoformat(),
+                "mean": approx(value * 2),
+                "min": approx(value * 2),
+                "max": approx(value * 2),
+                "state": None,
+                "sum": None,
+            }
+        ],
+        "sensor.test3": [
+            {
+                "statistic_id": "sensor.test3",
+                "start": now.isoformat(),
+                "mean": approx(value * 3),
+                "min": approx(value * 3),
+                "max": approx(value * 3),
+                "state": None,
+                "sum": None,
+            }
+        ],
+    }
+    assert response["result"] == expected_response
+
+    await client.send_json(
+        {
+            "id": 2,
+            "type": "history/clear_statistics",
+            "statistic_ids": ["sensor.test"],
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+
+    client = await hass_ws_client()
+    await client.send_json(
+        {
+            "id": 3,
+            "type": "history/statistics_during_period",
+            "start_time": now.isoformat(),
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"] == expected_response
+
+    await client.send_json(
+        {
+            "id": 4,
+            "type": "history/clear_statistics",
+            "statistic_ids": ["sensor.test1", "sensor.test3"],
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+
+    client = await hass_ws_client()
+    await client.send_json(
+        {
+            "id": 5,
+            "type": "history/statistics_during_period",
+            "start_time": now.isoformat(),
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"sensor.test2": expected_response["sensor.test2"]}