diff --git a/homeassistant/components/history/__init__.py b/homeassistant/components/history/__init__.py
index 9b009e0fceaac..c92718a87e4ed 100644
--- a/homeassistant/components/history/__init__.py
+++ b/homeassistant/components/history/__init__.py
@@ -11,9 +11,11 @@
 from sqlalchemy import not_, or_
 import voluptuous as vol
 
+from homeassistant.components import websocket_api
 from homeassistant.components.http import HomeAssistantView
 from homeassistant.components.recorder import history
 from homeassistant.components.recorder.models import States
+from homeassistant.components.recorder.statistics import statistics_during_period
 from homeassistant.components.recorder.util import session_scope
 from homeassistant.const import (
     CONF_DOMAINS,
@@ -101,10 +103,56 @@ async def async_setup(hass, config):
     hass.components.frontend.async_register_built_in_panel(
         "history", "history", "hass:poll-box"
     )
+    hass.components.websocket_api.async_register_command(
+        ws_get_statistics_during_period
+    )
 
     return True
 
 
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "history/statistics_during_period",
+        vol.Required("start_time"): str,
+        vol.Optional("end_time"): str,
+        vol.Optional("statistic_id"): str,
+    }
+)
+@websocket_api.async_response
+async def ws_get_statistics_during_period(
+    hass: HomeAssistant, connection: websocket_api.ActiveConnection, msg: dict
+) -> None:
+    """Handle statistics websocket command."""
+    start_time_str = msg["start_time"]
+    end_time_str = msg.get("end_time")
+
+    start_time = dt_util.parse_datetime(start_time_str)
+    if start_time:
+        start_time = dt_util.as_utc(start_time)
+    else:
+        connection.send_error(msg["id"], "invalid_start_time", "Invalid start_time")
+        return
+
+    if end_time_str:
+        end_time = dt_util.parse_datetime(end_time_str)
+        if end_time:
+            end_time = dt_util.as_utc(end_time)
+        else:
+            connection.send_error(msg["id"], "invalid_end_time", "Invalid end_time")
+            return
+    else:
+        end_time = None
+
+    statistics = await hass.async_add_executor_job(
+        statistics_during_period,
+        hass,
+        start_time,
+        end_time,
+        msg.get("statistic_id"),
+    )
+    connection.send_result(msg["id"], {"statistics": statistics})
+
+
 class HistoryPeriodView(HomeAssistantView):
     """Handle history period requests."""
 
diff --git a/tests/components/history/test_init.py b/tests/components/history/test_init.py
index bf6f392b649cc..36dd3f30156ba 100644
--- a/tests/components/history/test_init.py
+++ b/tests/components/history/test_init.py
@@ -826,3 +826,120 @@ async def test_entity_ids_limit_via_api_with_skip_initial_state(hass, hass_clien
     assert len(response_json) == 2
     assert response_json[0][0]["entity_id"] == "light.kitchen"
     assert response_json[1][0]["entity_id"] == "light.cow"
+
+
+async def test_statistics_during_period(hass, hass_ws_client):
+    """Test statistics_during_period."""
+    now = dt_util.utcnow()
+
+    await hass.async_add_executor_job(init_recorder_component, hass)
+    await async_setup_component(
+        hass,
+        "history",
+        {"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.test",
+        10,
+        attributes={"device_class": "temperature", "state_class": "measurement"},
+    )
+    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(),
+            "end_time": now.isoformat(),
+            "statistic_id": "sensor.test",
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"] == {"statistics": {}}
+
+    client = await hass_ws_client()
+    await client.send_json(
+        {
+            "id": 1,
+            "type": "history/statistics_during_period",
+            "start_time": now.isoformat(),
+            "statistic_id": "sensor.test",
+        }
+    )
+    response = await client.receive_json()
+    assert response["success"]
+    assert response["result"] == {
+        "statistics": {
+            "sensor.test": [
+                {
+                    "statistic_id": "sensor.test",
+                    "start": now.isoformat(),
+                    "mean": 10.0,
+                    "min": 10.0,
+                    "max": 10.0,
+                    "last_reset": None,
+                    "state": None,
+                    "sum": None,
+                }
+            ]
+        }
+    }
+
+
+async def test_statistics_during_period_bad_start_time(hass, hass_ws_client):
+    """Test statistics_during_period."""
+    await hass.async_add_executor_job(init_recorder_component, hass)
+    await async_setup_component(
+        hass,
+        "history",
+        {"history": {}},
+    )
+    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": "cats",
+        }
+    )
+    response = await client.receive_json()
+    assert not response["success"]
+    assert response["error"]["code"] == "invalid_start_time"
+
+
+async def test_statistics_during_period_bad_end_time(hass, hass_ws_client):
+    """Test statistics_during_period."""
+    now = dt_util.utcnow()
+
+    await hass.async_add_executor_job(init_recorder_component, hass)
+    await async_setup_component(
+        hass,
+        "history",
+        {"history": {}},
+    )
+    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(),
+            "end_time": "dogs",
+        }
+    )
+    response = await client.receive_json()
+    assert not response["success"]
+    assert response["error"]["code"] == "invalid_end_time"