diff --git a/openedx/core/djangoapps/notifications/management/commands/delete_notifications.py b/openedx/core/djangoapps/notifications/management/commands/delete_notifications.py
new file mode 100644
index 000000000000..62f2069100c9
--- /dev/null
+++ b/openedx/core/djangoapps/notifications/management/commands/delete_notifications.py
@@ -0,0 +1,76 @@
+"""
+Management command for deleting notifications with parameters
+"""
+import datetime
+import logging
+
+from django.core.management.base import BaseCommand
+
+from openedx.core.djangoapps.notifications.base_notification import (
+    COURSE_NOTIFICATION_APPS,
+    COURSE_NOTIFICATION_TYPES
+)
+from openedx.core.djangoapps.notifications.tasks import delete_notifications
+
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Invoke with:
+
+        python manage.py lms delete_notifications
+    """
+    help = (
+        "Delete notifications that meets a criteria. Requires app_name, notification_type and created (duration)"
+    )
+
+    def add_arguments(self, parser):
+        """
+        Add command line arguments to management command
+        """
+        parser.add_argument('--app_name', choices=list(COURSE_NOTIFICATION_APPS.keys()), required=True)
+        parser.add_argument('--notification_type', choices=list(COURSE_NOTIFICATION_TYPES.keys()),
+                            required=True)
+        parser.add_argument('--created', type=argparse_date, required=True,
+                            help="Allowed date formats YYYY-MM-DD. YYYY Year. MM Month. DD Date."
+                                 "Duration can be specified with ~. Maximum 15 days duration is allowed")
+        parser.add_argument('--course_id', required=False)
+
+    def handle(self, *args, **kwargs):
+        """
+        Calls delete notifications task
+        """
+        delete_notifications.delay(kwargs)
+        logger.info('Deletion task is in progress please check logs to verify')
+
+
+def argparse_date(string: str):
+    """
+    Argparse Type to return datetime dictionary from string
+    Returns {
+    'created__gte': datetime,
+    'created__lte': datetime,
+    }
+    """
+    ret_dict = {}
+    if '~' in string:
+        start_date, end_date = string.split('~')
+        start_date = parse_date(start_date)
+        end_date = parse_date(end_date)
+        if (end_date - start_date).days > 15:
+            raise ValueError('More than 15 days duration is not allowed')
+    else:
+        start_date = parse_date(string)
+        end_date = start_date
+    ret_dict['created__gte'] = datetime.datetime.combine(start_date, datetime.time.min)
+    ret_dict['created__lte'] = datetime.datetime.combine(end_date, datetime.time.max)
+    return ret_dict
+
+
+def parse_date(string):
+    """
+    Converts string date to datetime object
+    """
+    return datetime.datetime.strptime(string.strip(), "%Y-%m-%d")
diff --git a/openedx/core/djangoapps/notifications/management/tests/test_delete_notifications.py b/openedx/core/djangoapps/notifications/management/tests/test_delete_notifications.py
new file mode 100644
index 000000000000..d4b65fa565c6
--- /dev/null
+++ b/openedx/core/djangoapps/notifications/management/tests/test_delete_notifications.py
@@ -0,0 +1,56 @@
+"""
+Tests delete_notifications management command
+"""
+import ddt
+
+from unittest import mock
+
+from django.core.management import call_command
+
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+
+
+@ddt.ddt
+class TestDeleteNotifications(ModuleStoreTestCase):
+    """
+    Tests delete notifications management command
+    """
+
+    @ddt.data('app_name', 'notification_type', 'created')
+    def test_management_command_fails_if_required_param_is_missing(self, param):
+        """
+        Tests if all required params are available when running management command
+        """
+        default_dict = {
+            'app_name': 'discussion',
+            'notification_type': 'new_comment',
+            'created': '2024-02-01'
+        }
+        default_dict.pop(param)
+        try:
+            call_command('delete_notifications', **default_dict)
+            assert False
+        except Exception:    # pylint: disable=broad-except
+            pass
+
+    @ddt.data('course_id', None)
+    @mock.patch(
+        'openedx.core.djangoapps.notifications.tasks.delete_notifications.delay'
+    )
+    def test_management_command_runs_for_valid_params(self, key_to_remove, mock_func):
+        """
+        Tests management command runs with valid params optional
+        """
+        default_dict = {
+            'app_name': 'discussion',
+            'notification_type': 'new_comment',
+            'created': '2024-02-01',
+            'course_id': 'test-course'
+        }
+        if key_to_remove is not None:
+            default_dict.pop(key_to_remove)
+        call_command('delete_notifications', **default_dict)
+        assert mock_func.called
+        args = mock_func.call_args[0][0]
+        for key, value in default_dict.items():
+            assert args[key] == value
diff --git a/openedx/core/djangoapps/notifications/tasks.py b/openedx/core/djangoapps/notifications/tasks.py
index 137303fd5a61..a876b513cbcf 100644
--- a/openedx/core/djangoapps/notifications/tasks.py
+++ b/openedx/core/djangoapps/notifications/tasks.py
@@ -26,7 +26,7 @@
     Notification,
     get_course_notification_preference_config_version,
 )
-from openedx.core.djangoapps.notifications.utils import get_list_in_batches
+from openedx.core.djangoapps.notifications.utils import clean_arguments, get_list_in_batches
 
 logger = get_task_logger(__name__)
 
@@ -58,6 +58,33 @@ def create_course_notification_preferences_for_courses(self, course_ids):
     logger.info('Completed task create_course_notification_preferences')
 
 
+@shared_task(ignore_result=True)
+@set_code_owner_attribute
+def delete_notifications(kwargs):
+    """
+    Delete notifications
+    kwargs: dict {notification_type, app_name, created, course_id}
+    """
+    batch_size = settings.EXPIRED_NOTIFICATIONS_DELETE_BATCH_SIZE
+    total_deleted = 0
+    kwargs = clean_arguments(kwargs)
+    logger.info(f'Running delete with kwargs {kwargs}')
+    while True:
+        ids_to_delete = Notification.objects.filter(
+            **kwargs
+        ).values_list('id', flat=True)[:batch_size]
+        ids_to_delete = list(ids_to_delete)
+        if not ids_to_delete:
+            break
+        delete_queryset = Notification.objects.filter(
+            id__in=ids_to_delete
+        )
+        delete_count, _ = delete_queryset.delete()
+        total_deleted += delete_count
+        logger.info(f'Deleted in batch {delete_count}')
+    logger.info(f'Total deleted: {total_deleted}')
+
+
 @shared_task(ignore_result=True)
 @set_code_owner_attribute
 def delete_expired_notifications():
diff --git a/openedx/core/djangoapps/notifications/tests/test_tasks.py b/openedx/core/djangoapps/notifications/tests/test_tasks.py
index 4ef627b7496a..83ac9d17633e 100644
--- a/openedx/core/djangoapps/notifications/tests/test_tasks.py
+++ b/openedx/core/djangoapps/notifications/tests/test_tasks.py
@@ -4,6 +4,7 @@
 
 from unittest.mock import patch
 
+import datetime
 import ddt
 from django.core.exceptions import ValidationError
 from django.conf import settings
@@ -14,9 +15,15 @@
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
+from .utils import create_notification
 from ..config.waffle import ENABLE_NOTIFICATIONS
 from ..models import CourseNotificationPreference, Notification
-from ..tasks import create_notification_pref_if_not_exists, send_notifications, update_user_preference
+from ..tasks import (
+    create_notification_pref_if_not_exists,
+    delete_notifications,
+    send_notifications,
+    update_user_preference
+)
 
 
 @patch('openedx.core.djangoapps.notifications.models.COURSE_NOTIFICATION_CONFIG_VERSION', 1)
@@ -362,3 +369,67 @@ def test_preference_enabled_in_batch_audience(self, notification_type,
         content_url = 'https://example.com/'
         send_notifications(user_ids, str(self.course.id), app_name, notification_type, context, content_url)
         self.assertEqual(len(Notification.objects.all()), generated_count)
+
+
+class TestDeleteNotificationTask(ModuleStoreTestCase):
+    """
+    Tests delete_notification_function
+    """
+    def setUp(self):
+        """
+        Setup
+        """
+        super().setUp()
+        self.user = UserFactory()
+        self.course_1 = CourseFactory.create(org='org', number='num', run='run_01')
+        self.course_2 = CourseFactory.create(org='org', number='num', run='run_02')
+        Notification.objects.all().delete()
+
+    def test_app_name_param(self):
+        """
+        Tests if app_name parameter works as expected
+        """
+        assert not Notification.objects.all()
+        create_notification(self.user, self.course_1.id, app_name='discussion', notification_type='new_comment')
+        create_notification(self.user, self.course_1.id, app_name='updates', notification_type='course_update')
+        delete_notifications({'app_name': 'discussion'})
+        assert not Notification.objects.filter(app_name='discussion')
+        assert Notification.objects.filter(app_name='updates')
+
+    def test_notification_type_param(self):
+        """
+        Tests if notification_type parameter works as expected
+        """
+        assert not Notification.objects.all()
+        create_notification(self.user, self.course_1.id, app_name='discussion', notification_type='new_comment')
+        create_notification(self.user, self.course_1.id, app_name='discussion', notification_type='new_response')
+        delete_notifications({'notification_type': 'new_comment'})
+        assert not Notification.objects.filter(notification_type='new_comment')
+        assert Notification.objects.filter(notification_type='new_response')
+
+    def test_created_param(self):
+        """
+        Tests if created parameter works as expected
+        """
+        assert not Notification.objects.all()
+        create_notification(self.user, self.course_1.id, created=datetime.datetime(2024, 2, 10))
+        create_notification(self.user, self.course_2.id, created=datetime.datetime(2024, 3, 12, 5))
+        kwargs = {
+            'created': {
+                'created__gte': datetime.datetime(2024, 3, 12, 0, 0, 0),
+                'created__lte': datetime.datetime(2024, 3, 12, 23, 59, 59),
+            }
+        }
+        delete_notifications(kwargs)
+        self.assertEqual(Notification.objects.all().count(), 1)
+
+    def test_course_id_param(self):
+        """
+        Tests if course_id parameter works as expected
+        """
+        assert not Notification.objects.all()
+        create_notification(self.user, self.course_1.id)
+        create_notification(self.user, self.course_2.id)
+        delete_notifications({'course_id': self.course_1.id})
+        assert not Notification.objects.filter(course_id=self.course_1.id)
+        assert Notification.objects.filter(course_id=self.course_2.id)
diff --git a/openedx/core/djangoapps/notifications/tests/utils.py b/openedx/core/djangoapps/notifications/tests/utils.py
new file mode 100644
index 000000000000..b34d934c85f3
--- /dev/null
+++ b/openedx/core/djangoapps/notifications/tests/utils.py
@@ -0,0 +1,29 @@
+"""
+Utils for tests
+"""
+from openedx.core.djangoapps.notifications.models import Notification
+
+
+def create_notification(user, course_key, **kwargs):
+    """
+    Create a  notification
+    """
+    params = {
+        'user': user,
+        'course_id': course_key,
+        'app_name': 'discussion',
+        'notification_type': 'new_comment',
+        'content_url': '',
+        'content_context': {
+            "replier_name": "replier",
+            "username": "username",
+            "author_name": "author_name",
+            "post_title": "post_title",
+            "course_update_content": "Course update content",
+            "content_type": 'post',
+            "content": "post_title"
+        }
+    }
+    params.update(kwargs)
+    notification = Notification.objects.create(**params)
+    return notification
diff --git a/openedx/core/djangoapps/notifications/utils.py b/openedx/core/djangoapps/notifications/utils.py
index c3ce819c0011..e1e25901e134 100644
--- a/openedx/core/djangoapps/notifications/utils.py
+++ b/openedx/core/djangoapps/notifications/utils.py
@@ -154,3 +154,16 @@ def remove_preferences_with_no_access(preferences: dict, user) -> dict:
         user_forum_roles
     )
     return preferences
+
+
+def clean_arguments(kwargs):
+    """
+    Returns query arguments from command line arguments
+    """
+    clean_kwargs = {}
+    for key in ['app_name', 'notification_type', 'course_id']:
+        if kwargs.get(key):
+            clean_kwargs[key] = kwargs[key]
+    if kwargs.get('created', {}):
+        clean_kwargs.update(kwargs.get('created'))
+    return clean_kwargs