diff --git a/app/celery/tasks.py b/app/celery/tasks.py
index 20b3088c2..f317c634b 100644
--- a/app/celery/tasks.py
+++ b/app/celery/tasks.py
@@ -1,6 +1,6 @@
 import json
 from datetime import datetime
-from collections import namedtuple
+from collections import namedtuple, defaultdict
 
 from celery.signals import worker_process_shutdown
 from flask import current_app
@@ -31,6 +31,7 @@ from app import (
 from app.aws import s3
 from app.celery import provider_tasks, letters_pdf_tasks, research_mode_tasks
 from app.config import QueueNames
+from app.dao.daily_sorted_letter_dao import dao_create_or_update_daily_sorted_letter
 from app.dao.inbound_sms_dao import dao_get_inbound_sms_by_id
 from app.dao.jobs_dao import (
     dao_update_job,
@@ -66,9 +67,11 @@ from app.models import (
     NOTIFICATION_TEMPORARY_FAILURE,
     NOTIFICATION_TECHNICAL_FAILURE,
     SMS_TYPE,
+    DailySortedLetter,
 )
 from app.notifications.process_notifications import persist_notification
 from app.service.utils import service_allowed_to_send_to
+from app.utils import convert_utc_to_bst
 
 
 @worker_process_shutdown.connect
@@ -404,6 +407,7 @@ def get_template_class(template_type):
 def update_letter_notifications_statuses(self, filename):
     bucket_location = '{}-ftp'.format(current_app.config['NOTIFY_EMAIL_DOMAIN'])
     response_file_content = s3.get_s3_file(bucket_location, filename)
+    sorted_letter_counts = defaultdict(int)
 
     try:
         notification_updates = process_updates_from_file(response_file_content)
@@ -414,6 +418,7 @@ def update_letter_notifications_statuses(self, filename):
         for update in notification_updates:
             check_billable_units(update)
             update_letter_notification(filename, temporary_failures, update)
+            sorted_letter_counts[update.cost_threshold] += 1
 
         if temporary_failures:
             # This will alert Notify that DVLA was unable to deliver the letters, we need to investigate
@@ -421,6 +426,32 @@ def update_letter_notifications_statuses(self, filename):
                 filename=filename, failures=temporary_failures)
             raise DVLAException(message)
 
+        if sorted_letter_counts.keys() - {'Unsorted', 'Sorted'}:
+            unknown_status = sorted_letter_counts.keys() - {'Unsorted', 'Sorted'}
+
+            message = 'DVLA response file: {} contains unknown Sorted status {}'.format(
+                filename, unknown_status
+            )
+            raise DVLAException(message)
+
+        billing_date = get_billing_date_in_bst_from_filename(filename)
+        persist_daily_sorted_letter_counts(billing_date, sorted_letter_counts)
+
+
+def get_billing_date_in_bst_from_filename(filename):
+    datetime_string = filename.split('.')[1]
+    datetime_obj = datetime.strptime(datetime_string, '%Y%m%d%H%M%S')
+    return convert_utc_to_bst(datetime_obj).date()
+
+
+def persist_daily_sorted_letter_counts(day, sorted_letter_counts):
+    daily_letter_count = DailySortedLetter(
+        billing_day=day,
+        unsorted_count=sorted_letter_counts['Unsorted'],
+        sorted_count=sorted_letter_counts['Sorted']
+    )
+    dao_create_or_update_daily_sorted_letter(daily_letter_count)
+
 
 def process_updates_from_file(response_file):
     NotificationUpdate = namedtuple('NotificationUpdate', ['reference', 'status', 'page_count', 'cost_threshold'])
diff --git a/app/dao/daily_sorted_letter_dao.py b/app/dao/daily_sorted_letter_dao.py
new file mode 100644
index 000000000..3afad4b2a
--- /dev/null
+++ b/app/dao/daily_sorted_letter_dao.py
@@ -0,0 +1,37 @@
+from datetime import datetime
+
+from sqlalchemy.dialects.postgresql import insert
+
+from app import db
+from app.dao.dao_utils import transactional
+from app.models import DailySortedLetter
+
+
+def dao_get_daily_sorted_letter_by_billing_day(billing_day):
+    return DailySortedLetter.query.filter_by(
+        billing_day=billing_day
+    ).first()
+
+
+@transactional
+def dao_create_or_update_daily_sorted_letter(new_daily_sorted_letter):
+    '''
+    This uses the Postgres upsert to avoid race conditions when two threads try and insert
+    at the same row. The excluded object refers to values that we tried to insert but were
+    rejected.
+    http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html#insert-on-conflict-upsert
+    '''
+    table = DailySortedLetter.__table__
+    stmt = insert(table).values(
+        billing_day=new_daily_sorted_letter.billing_day,
+        unsorted_count=new_daily_sorted_letter.unsorted_count,
+        sorted_count=new_daily_sorted_letter.sorted_count)
+    stmt = stmt.on_conflict_do_update(
+        index_elements=[table.c.billing_day],
+        set_={
+            'unsorted_count': table.c.unsorted_count + stmt.excluded.unsorted_count,
+            'sorted_count': table.c.sorted_count + stmt.excluded.sorted_count,
+            'updated_at': datetime.utcnow()
+        }
+    )
+    db.session.connection().execute(stmt)
diff --git a/app/models.py b/app/models.py
index 6678d91e6..47256a63d 100644
--- a/app/models.py
+++ b/app/models.py
@@ -1752,3 +1752,13 @@ class StatsTemplateUsageByMonth(db.Model):
             'year': self.year,
             'count': self.count
         }
+
+
+class DailySortedLetter(db.Model):
+    __tablename__ = "daily_sorted_letter"
+
+    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    billing_day = db.Column(db.Date, nullable=False, index=True, unique=True)
+    unsorted_count = db.Column(db.Integer, nullable=False, default=0)
+    sorted_count = db.Column(db.Integer, nullable=False, default=0)
+    updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.datetime.utcnow)
diff --git a/migrations/versions/0173_create_daily_sorted_letter.py b/migrations/versions/0173_create_daily_sorted_letter.py
new file mode 100644
index 000000000..3215134b9
--- /dev/null
+++ b/migrations/versions/0173_create_daily_sorted_letter.py
@@ -0,0 +1,30 @@
+"""
+
+Revision ID: 0173_create_daily_sorted_letter
+Revises: 0172_deprioritise_examples
+Create Date: 2018-03-01 11:53:32.964256
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision = '0173_create_daily_sorted_letter'
+down_revision = '0172_deprioritise_examples'
+
+
+def upgrade():
+    op.create_table('daily_sorted_letter',
+    sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
+    sa.Column('billing_day', sa.Date(), nullable=False),
+    sa.Column('unsorted_count', sa.Integer(), nullable=False),
+    sa.Column('sorted_count', sa.Integer(), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), nullable=True),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_daily_sorted_letter_billing_day'), 'daily_sorted_letter', ['billing_day'], unique=True)
+
+
+def downgrade():
+    op.drop_index(op.f('ix_daily_sorted_letter_billing_day'), table_name='daily_sorted_letter')
+    op.drop_table('daily_sorted_letter')
diff --git a/tests/app/celery/test_ftp_update_tasks.py b/tests/app/celery/test_ftp_update_tasks.py
index 05b73329f..435ccab93 100644
--- a/tests/app/celery/test_ftp_update_tasks.py
+++ b/tests/app/celery/test_ftp_update_tasks.py
@@ -1,5 +1,5 @@
-from collections import namedtuple
-from datetime import datetime
+from collections import namedtuple, defaultdict
+from datetime import datetime, date
 
 import pytest
 from freezegun import freeze_time
@@ -18,12 +18,15 @@ from app.models import (
 )
 from app.celery.tasks import (
     check_billable_units,
+    get_billing_date_in_bst_from_filename,
+    persist_daily_sorted_letter_counts,
     process_updates_from_file,
     update_dvla_job_to_error,
     update_letter_notifications_statuses,
     update_letter_notifications_to_error,
     update_letter_notifications_to_sent_to_dvla
 )
+from app.dao.daily_sorted_letter_dao import dao_get_daily_sorted_letter_by_billing_day
 
 from tests.app.db import create_notification, create_service_callback_api
 from tests.conftest import set_config
@@ -56,8 +59,8 @@ def test_update_letter_notifications_statuses_raises_for_invalid_format(notify_a
     mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=invalid_file)
 
     with pytest.raises(DVLAException) as e:
-        update_letter_notifications_statuses(filename='foo.txt')
-    assert 'DVLA response file: {} has an invalid format'.format('foo.txt') in str(e)
+        update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
+    assert 'DVLA response file: {} has an invalid format'.format('NOTIFY.20170823160812.RSP.TXT') in str(e)
 
 
 def test_update_letter_notification_statuses_when_notification_does_not_exist_updates_notification_history(
@@ -70,7 +73,7 @@ def test_update_letter_notification_statuses_when_notification_does_not_exist_up
                                        billable_units=1)
     Notification.query.filter_by(id=notification.id).delete()
 
-    update_letter_notifications_statuses(filename="older_than_7_days.txt")
+    update_letter_notifications_statuses(filename="NOTIFY.20170823160812.RSP.TXT")
 
     updated_history = NotificationHistory.query.filter_by(id=notification.id).one()
     assert updated_history.status == NOTIFICATION_DELIVERED
@@ -90,12 +93,35 @@ def test_update_letter_notifications_statuses_raises_dvla_exception(notify_api,
     ) in str(e)
 
 
+def test_update_letter_notifications_statuses_raises_error_for_unknown_sorted_status(
+    notify_api,
+    mocker,
+    sample_letter_template
+):
+    sent_letter_1 = create_notification(sample_letter_template, reference='ref-foo', status=NOTIFICATION_SENDING)
+    sent_letter_2 = create_notification(sample_letter_template, reference='ref-bar', status=NOTIFICATION_SENDING)
+    valid_file = '{}|Sent|1|Unsorted\n{}|Sent|2|Error'.format(
+        sent_letter_1.reference, sent_letter_2.reference)
+
+    mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
+
+    with pytest.raises(DVLAException) as e:
+        update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
+
+    assert "DVLA response file: {filename} contains unknown Sorted status {unknown_status}".format(
+        filename="NOTIFY.20170823160812.RSP.TXT", unknown_status="{'Error'}"
+    ) in str(e)
+
+
 def test_update_letter_notifications_statuses_calls_with_correct_bucket_location(notify_api, mocker):
     s3_mock = mocker.patch('app.celery.tasks.s3.get_s3_object')
 
     with set_config(notify_api, 'NOTIFY_EMAIL_DOMAIN', 'foo.bar'):
-        update_letter_notifications_statuses(filename='foo.txt')
-        s3_mock.assert_called_with('{}-ftp'.format(current_app.config['NOTIFY_EMAIL_DOMAIN']), 'foo.txt')
+        update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
+        s3_mock.assert_called_with('{}-ftp'.format(
+            current_app.config['NOTIFY_EMAIL_DOMAIN']),
+            'NOTIFY.20170823160812.RSP.TXT'
+        )
 
 
 def test_update_letter_notifications_statuses_builds_updates_from_content(notify_api, mocker):
@@ -103,7 +129,7 @@ def test_update_letter_notifications_statuses_builds_updates_from_content(notify
     mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
     update_mock = mocker.patch('app.celery.tasks.process_updates_from_file')
 
-    update_letter_notifications_statuses(filename='foo.txt')
+    update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
 
     update_mock.assert_called_with('ref-foo|Sent|1|Unsorted\nref-bar|Sent|2|Sorted')
 
@@ -136,7 +162,7 @@ def test_update_letter_notifications_statuses_persisted(notify_api, mocker, samp
     mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
 
     with pytest.raises(expected_exception=DVLAException) as e:
-        update_letter_notifications_statuses(filename='foo.txt')
+        update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
 
     assert sent_letter.status == NOTIFICATION_DELIVERED
     assert sent_letter.billable_units == 1
@@ -145,7 +171,45 @@ def test_update_letter_notifications_statuses_persisted(notify_api, mocker, samp
     assert failed_letter.billable_units == 2
     assert failed_letter.updated_at
     assert "DVLA response file: {filename} has failed letters with notification.reference {failures}".format(
-        filename="foo.txt", failures=[format(failed_letter.reference)]) in str(e)
+        filename="NOTIFY.20170823160812.RSP.TXT", failures=[format(failed_letter.reference)]) in str(e)
+
+
+def test_update_letter_notifications_statuses_persists_daily_sorted_letter_count(
+    notify_api,
+    mocker,
+    sample_letter_template
+):
+    sent_letter_1 = create_notification(sample_letter_template, reference='ref-foo', status=NOTIFICATION_SENDING)
+    sent_letter_2 = create_notification(sample_letter_template, reference='ref-bar', status=NOTIFICATION_SENDING)
+    valid_file = '{}|Sent|1|Unsorted\n{}|Sent|2|Sorted'.format(
+        sent_letter_1.reference, sent_letter_2.reference)
+
+    mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
+    persist_letter_count_mock = mocker.patch('app.celery.tasks.persist_daily_sorted_letter_counts')
+
+    update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
+
+    persist_letter_count_mock.assert_called_once_with(date(2017, 8, 23), {'Unsorted': 1, 'Sorted': 1})
+
+
+def test_update_letter_notifications_statuses_persists_daily_sorted_letter_count_with_no_sorted_values(
+    notify_api,
+    mocker,
+    sample_letter_template,
+    notify_db_session
+):
+    sent_letter_1 = create_notification(sample_letter_template, reference='ref-foo', status=NOTIFICATION_SENDING)
+    sent_letter_2 = create_notification(sample_letter_template, reference='ref-bar', status=NOTIFICATION_SENDING)
+    valid_file = '{}|Sent|1|Unsorted\n{}|Sent|2|Unsorted'.format(
+        sent_letter_1.reference, sent_letter_2.reference)
+    mocker.patch('app.celery.tasks.s3.get_s3_file', return_value=valid_file)
+
+    update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
+
+    daily_sorted_letter = dao_get_daily_sorted_letter_by_billing_day(date(2017, 8, 23))
+
+    assert daily_sorted_letter.unsorted_count == 2
+    assert daily_sorted_letter.sorted_count == 0
 
 
 def test_update_letter_notifications_does_not_call_send_callback_if_no_db_entry(notify_api, mocker,
@@ -159,7 +223,7 @@ def test_update_letter_notifications_does_not_call_send_callback_if_no_db_entry(
         'app.celery.service_callback_tasks.send_delivery_status_to_service.apply_async'
     )
 
-    update_letter_notifications_statuses(filename='foo.txt')
+    update_letter_notifications_statuses(filename='NOTIFY.20170823160812.RSP.TXT')
     send_mock.assert_not_called()
 
 
@@ -230,3 +294,24 @@ def test_check_billable_units_when_billable_units_does_not_match_page_count(
     mock_logger.assert_called_once_with(
         'Notification with id {} had 3 billable_units but a page count of 1'.format(notification.id)
     )
+
+
+@pytest.mark.parametrize('filename_date, billing_date', [
+    ('20170820230000', date(2017, 8, 21)),
+    ('20170120230000', date(2017, 1, 20))
+])
+def test_get_billing_date_in_bst_from_filename(filename_date, billing_date):
+    filename = 'NOTIFY.{}.RSP.TXT'.format(filename_date)
+    result = get_billing_date_in_bst_from_filename(filename)
+
+    assert result == billing_date
+
+
+@freeze_time("2018-01-11 09:00:00")
+def test_persist_daily_sorted_letter_counts_saves_sorted_and_unsorted_values(client, notify_db_session):
+    letter_counts = defaultdict(int, **{'Unsorted': 5, 'Sorted': 1})
+    persist_daily_sorted_letter_counts(date.today(), letter_counts)
+    day = dao_get_daily_sorted_letter_by_billing_day(date.today())
+
+    assert day.unsorted_count == 5
+    assert day.sorted_count == 1
diff --git a/tests/app/dao/test_daily_sorted_letter_dao.py b/tests/app/dao/test_daily_sorted_letter_dao.py
new file mode 100644
index 000000000..2a4ceb318
--- /dev/null
+++ b/tests/app/dao/test_daily_sorted_letter_dao.py
@@ -0,0 +1,47 @@
+from datetime import date
+
+from app.dao.daily_sorted_letter_dao import (
+    dao_create_or_update_daily_sorted_letter,
+    dao_get_daily_sorted_letter_by_billing_day,
+)
+from app.models import DailySortedLetter
+from tests.app.db import create_daily_sorted_letter
+
+
+def test_dao_get_daily_sorted_letter_by_billing_day(notify_db, notify_db_session):
+    billing_day = date(2018, 2, 1)
+    other_day = date(2017, 9, 8)
+
+    daily_sorted_letters = create_daily_sorted_letter(billing_day=billing_day)
+
+    assert dao_get_daily_sorted_letter_by_billing_day(billing_day) == daily_sorted_letters
+    assert not dao_get_daily_sorted_letter_by_billing_day(other_day)
+
+
+def test_dao_create_or_update_daily_sorted_letter_creates_a_new_entry(notify_db, notify_db_session):
+    billing_day = date(2018, 2, 1)
+    dsl = DailySortedLetter(billing_day=billing_day, unsorted_count=2, sorted_count=0)
+    dao_create_or_update_daily_sorted_letter(dsl)
+
+    daily_sorted_letter = dao_get_daily_sorted_letter_by_billing_day(billing_day)
+
+    assert daily_sorted_letter.billing_day == billing_day
+    assert daily_sorted_letter.unsorted_count == 2
+    assert daily_sorted_letter.sorted_count == 0
+    assert not daily_sorted_letter.updated_at
+
+
+def test_dao_create_or_update_daily_sorted_letter_updates_an_existing_entry(
+    notify_db,
+    notify_db_session
+):
+    create_daily_sorted_letter(unsorted_count=2, sorted_count=3)
+
+    dsl = DailySortedLetter(billing_day=date(2018, 1, 18), unsorted_count=5, sorted_count=17)
+    dao_create_or_update_daily_sorted_letter(dsl)
+
+    daily_sorted_letter = dao_get_daily_sorted_letter_by_billing_day(dsl.billing_day)
+
+    assert daily_sorted_letter.unsorted_count == 7
+    assert daily_sorted_letter.sorted_count == 20
+    assert daily_sorted_letter.updated_at
diff --git a/tests/app/db.py b/tests/app/db.py
index 6a09413a8..b159976da 100644
--- a/tests/app/db.py
+++ b/tests/app/db.py
@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, date
 import uuid
 
 from app import db
@@ -9,6 +9,7 @@ from app.dao.service_sms_sender_dao import update_existing_sms_sender_with_inbou
 from app.dao.invited_org_user_dao import save_invited_org_user
 from app.models import (
     ApiKey,
+    DailySortedLetter,
     InboundSms,
     InboundNumber,
     Job,
@@ -506,3 +507,16 @@ def create_invited_org_user(organisation, invited_by, email_address='invite@exam
     )
     save_invited_org_user(invited_org_user)
     return invited_org_user
+
+
+def create_daily_sorted_letter(billing_day=date(2018, 1, 18), unsorted_count=0, sorted_count=0):
+    daily_sorted_letter = DailySortedLetter(
+        billing_day=billing_day,
+        unsorted_count=unsorted_count,
+        sorted_count=sorted_count
+    )
+
+    db.session.add(daily_sorted_letter)
+    db.session.commit()
+
+    return daily_sorted_letter