diff --git a/app/dao/notifications_dao.py b/app/dao/notifications_dao.py
index 13317afd7..13fcee367 100644
--- a/app/dao/notifications_dao.py
+++ b/app/dao/notifications_dao.py
@@ -11,13 +11,16 @@ from sqlalchemy.orm import joinedload
 
 from app import db, create_uuid
 from app.dao import days_ago
+from app.dao.provider_details_dao import get_provider_details_by_identifier
 from app.models import (
     Service,
     Notification,
     NotificationHistory,
     NotificationStatistics,
+    ProviderDetails,
     Template,
     NOTIFICATION_CREATED,
+    NOTIFICATION_DELIVERED,
     NOTIFICATION_SENDING,
     NOTIFICATION_PENDING,
     NOTIFICATION_TECHNICAL_FAILURE,
@@ -421,3 +424,32 @@ def get_total_sent_notifications_in_date_range(start_date, end_date, notificatio
     ).scalar()
 
     return result or 0
+
+
+def get_count_of_slow_delivery_sms_notifications_for_provider(
+    created_at,
+    sent_at,
+    provider,
+    threshold,
+    delivery_time,
+    service_id,
+    template_id
+):
+    count = db.session.query(
+        func.count().label('total'),
+        Notification.sent_by
+    ).filter(
+        Notification.service_id == service_id,
+        Notification.template_id == template_id,
+        Notification.created_at >= created_at,
+        Notification.sent_at >= sent_at,
+        Notification.status == NOTIFICATION_DELIVERED,
+        Notification.sent_by == provider,
+        (Notification.updated_at - Notification.sent_at) >= delivery_time,
+    ).group_by(
+        Notification.sent_by
+    ).having(
+        func.count().label('total') >= threshold,
+    ).first()
+
+    return count
diff --git a/tests/app/conftest.py b/tests/app/conftest.py
index d58cfef70..ea7319cce 100644
--- a/tests/app/conftest.py
+++ b/tests/app/conftest.py
@@ -421,29 +421,34 @@ def sample_notification_with_job(
 
 
 @pytest.fixture(scope='function')
-def sample_notification(notify_db,
-                        notify_db_session,
-                        service=None,
-                        template=None,
-                        job=None,
-                        job_row_number=None,
-                        to_field=None,
-                        status='created',
-                        reference=None,
-                        created_at=None,
-                        sent_at=None,
-                        billable_units=1,
-                        personalisation=None,
-                        api_key_id=None,
-                        key_type=KEY_TYPE_NORMAL,
-                        sent_by=None,
-                        client_reference=None):
+def sample_notification(
+    notify_db,
+    notify_db_session,
+    service=None,
+    template=None,
+    job=None,
+    job_row_number=None,
+    to_field=None,
+    status='created',
+    reference=None,
+    created_at=None,
+    sent_at=None,
+    billable_units=1,
+    personalisation=None,
+    api_key_id=None,
+    key_type=KEY_TYPE_NORMAL,
+    sent_by=None,
+    client_reference=None,
+    updated_at=None
+):
     if created_at is None:
         created_at = datetime.utcnow()
     if service is None:
         service = sample_service(notify_db, notify_db_session)
     if template is None:
         template = sample_template(notify_db, notify_db_session, service=service)
+    if not updated_at and status in NOTIFICATION_STATUS_TYPES_COMPLETED:
+        updated_at = created_at
 
     notification_id = uuid.uuid4()
 
@@ -472,7 +477,7 @@ def sample_notification(notify_db,
         'api_key_id': api_key_id,
         'key_type': key_type,
         'sent_by': sent_by,
-        'updated_at': created_at if status in NOTIFICATION_STATUS_TYPES_COMPLETED else None,
+        'updated_at': updated_at,
         'client_reference': client_reference
     }
     if job_row_number:
diff --git a/tests/app/dao/test_notification_dao.py b/tests/app/dao/test_notification_dao.py
index 83e8d1132..d9aa249a4 100644
--- a/tests/app/dao/test_notification_dao.py
+++ b/tests/app/dao/test_notification_dao.py
@@ -40,7 +40,9 @@ from app.dao.notifications_dao import (
     dao_delete_notifications_and_history_by_id,
     dao_timeout_notifications,
     get_financial_year,
-    get_april_fools)
+    get_april_fools,
+    get_count_of_slow_delivery_sms_notifications_for_provider
+)
 
 from app.dao.services_dao import dao_update_service
 from tests.app.conftest import (
@@ -1422,3 +1424,229 @@ def test_get_total_sent_notifications_for_email_excludes_sms_counts(
 
     total_count = get_total_sent_notifications_in_date_range(start_date, end_date, 'email')
     assert total_count == 2
+
+
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_get_count_of_slow_delivery_sms_notifications_returns_after_created_time(
+    notify_db,
+    notify_db_session,
+    sample_template
+):
+    now = datetime.utcnow()
+    a_second_ago = now - timedelta(seconds=1)
+    five_minutes_from_now = now + timedelta(minutes=5)
+
+    notification_five_minutes_to_deliver = partial(
+        sample_notification,
+        notify_db,
+        notify_db_session,
+        template=sample_template,
+        status='delivered',
+        sent_at=now,
+        sent_by='mmg',
+        updated_at=five_minutes_from_now
+    )
+
+    noti1 = notification_five_minutes_to_deliver(created_at=a_second_ago)
+    notification_five_minutes_to_deliver(created_at=a_second_ago)
+    notification_five_minutes_to_deliver(created_at=now)
+
+    count = get_count_of_slow_delivery_sms_notifications_for_provider(
+        created_at=now,
+        sent_at=now,
+        provider='mmg',
+        threshold=1,
+        delivery_time=timedelta(minutes=5),
+        service_id=noti1.service_id,
+        template_id=noti1.template_id
+    )
+
+    assert count.total == 1
+
+
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_get_count_of_slow_delivery_sms_notifications_returns_after_sent_time(
+    notify_db,
+    notify_db_session,
+    sample_template
+):
+    now = datetime.utcnow()
+    one_minute_from_now = now + timedelta(minutes=1)
+    five_minutes_from_now = now + timedelta(minutes=5)
+
+    notification_five_minutes_to_deliver = partial(
+        sample_notification,
+        notify_db,
+        notify_db_session,
+        template=sample_template,
+        status='delivered',
+        sent_by='mmg',
+        updated_at=five_minutes_from_now
+    )
+
+    noti1 = notification_five_minutes_to_deliver(created_at=now, sent_at=now)
+    notification_five_minutes_to_deliver(created_at=now, sent_at=one_minute_from_now)
+    notification_five_minutes_to_deliver(created_at=now, sent_at=one_minute_from_now)
+
+    count = get_count_of_slow_delivery_sms_notifications_for_provider(
+        created_at=now,
+        sent_at=one_minute_from_now,
+        provider='mmg',
+        threshold=2,
+        delivery_time=timedelta(minutes=3),
+        service_id=noti1.service_id,
+        template_id=noti1.template_id
+    )
+
+    assert count.total == 2
+
+
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_get_count_of_slow_delivery_sms_notifications_observes_threshold(
+    notify_db,
+    notify_db_session,
+    sample_template
+):
+    now = datetime.utcnow()
+    five_minutes_from_now = now + timedelta(minutes=5)
+
+    notification_five_minutes_to_deliver = partial(
+        sample_notification,
+        notify_db,
+        notify_db_session,
+        template=sample_template,
+        status='delivered',
+        sent_at=now,
+        sent_by='mmg',
+        updated_at=five_minutes_from_now
+    )
+
+    noti1 = notification_five_minutes_to_deliver(created_at=now)
+    notification_five_minutes_to_deliver(created_at=now)
+
+    count = get_count_of_slow_delivery_sms_notifications_for_provider(
+        created_at=now,
+        sent_at=now,
+        provider='mmg',
+        threshold=3,
+        delivery_time=timedelta(minutes=5),
+        service_id=noti1.service_id,
+        template_id=noti1.template_id
+    )
+
+    assert not count
+
+
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_get_count_of_slow_delivery_sms_notifications_returns_status_delivered_only(
+    notify_db,
+    notify_db_session,
+    sample_template
+):
+    now = datetime.utcnow()
+    five_minutes_from_now = now + timedelta(minutes=5)
+
+    notification_five_minutes_to_deliver = partial(
+        sample_notification,
+        notify_db,
+        notify_db_session,
+        template=sample_template,
+        sent_at=now,
+        sent_by='firetext',
+        created_at=now,
+        updated_at=five_minutes_from_now
+    )
+
+    noti1 = notification_five_minutes_to_deliver(status='created')
+    notification_five_minutes_to_deliver(status='sending')
+    notification_five_minutes_to_deliver(status='delivered')
+    notification_five_minutes_to_deliver(status='delivered')
+
+    count = get_count_of_slow_delivery_sms_notifications_for_provider(
+        created_at=now,
+        sent_at=now,
+        provider='firetext',
+        threshold=1,
+        delivery_time=timedelta(minutes=5),
+        service_id=noti1.service_id,
+        template_id=noti1.template_id
+    )
+
+    assert count.total == 2
+
+
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_get_count_of_slow_delivery_sms_notifications_returns_slow_delivery_time_only(
+    notify_db,
+    notify_db_session,
+    sample_template
+):
+    now = datetime.utcnow()
+    five_minutes_from_now = now + timedelta(minutes=5)
+
+    notification = partial(
+        sample_notification,
+        notify_db,
+        notify_db_session,
+        template=sample_template,
+        created_at=now,
+        sent_at=now,
+        sent_by='mmg',
+        status='delivered'
+    )
+
+    noti1 = notification(updated_at=five_minutes_from_now - timedelta(seconds=1))
+    noti1 = notification(updated_at=five_minutes_from_now - timedelta(seconds=1))
+    notification(updated_at=five_minutes_from_now)
+    notification(updated_at=five_minutes_from_now)
+    notification(updated_at=five_minutes_from_now)
+
+    count = get_count_of_slow_delivery_sms_notifications_for_provider(
+        created_at=now,
+        sent_at=now,
+        provider='mmg',
+        threshold=1,
+        delivery_time=timedelta(minutes=5),
+        service_id=noti1.service_id,
+        template_id=noti1.template_id
+    )
+
+    assert count.total == 3
+
+
+@freeze_time("2016-01-10 12:00:00.000000")
+def test_get_count_of_slow_delivery_sms_notifications_returns_provider_only(
+    notify_db,
+    notify_db_session,
+    sample_template
+):
+    now = datetime.utcnow()
+    five_minutes_from_now = now + timedelta(minutes=5)
+
+    notification = partial(
+        sample_notification,
+        notify_db,
+        notify_db_session,
+        template=sample_template,
+        created_at=now,
+        sent_at=now,
+        updated_at=five_minutes_from_now,
+        status='delivered'
+    )
+
+    noti1 = notification(sent_by='mmg')
+    notification(sent_by='firetext')
+    notification(sent_by='loadtesting')
+    notification(sent_by='loadtesting')
+
+    count = get_count_of_slow_delivery_sms_notifications_for_provider(
+        created_at=now,
+        sent_at=now,
+        provider='mmg',
+        threshold=1,
+        delivery_time=timedelta(minutes=5),
+        service_id=noti1.service_id,
+        template_id=noti1.template_id
+    )
+
+    assert count.sent_by == 'mmg'