diff --git a/app/api/api.rb b/app/api/api.rb
index a32114685..eb4f79245 100644
--- a/app/api/api.rb
+++ b/app/api/api.rb
@@ -12,3 +12,4 @@
 require_relative 'internal/transactions'
 require_relative 'internal/utilization'
 require_relative 'internal/service_tokens'
+require_relative 'internal/stats'
diff --git a/app/api/internal/stats.rb b/app/api/internal/stats.rb
new file mode 100644
index 000000000..1418c6b65
--- /dev/null
+++ b/app/api/internal/stats.rb
@@ -0,0 +1,25 @@
+module ThreeScale
+  module Backend
+    module API
+      internal_api '/services/:service_id/stats' do
+        before do
+          respond_with_404('service not found') unless Service.exists?(params[:service_id])
+        end
+
+        delete '' do |service_id|
+          delete_stats_job_attrs = api_params Stats::DeleteJobDef
+          delete_stats_job_attrs[:service_id] = service_id
+          delete_stats_job_attrs[:from] = delete_stats_job_attrs[:from].to_i
+          delete_stats_job_attrs[:to] = delete_stats_job_attrs[:to].to_i
+          begin
+            Stats::DeleteJobDef.new(delete_stats_job_attrs).run_async
+          rescue DeleteServiceStatsValidationError => e
+            [400, headers, { status: :error, error: e.message }.to_json]
+          else
+            { status: :to_be_deleted }.to_json
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/3scale/backend.rb b/lib/3scale/backend.rb
index d66107a44..3df4b4d1d 100644
--- a/lib/3scale/backend.rb
+++ b/lib/3scale/backend.rb
@@ -42,7 +42,7 @@
 require '3scale/backend/queue_storage'
 require '3scale/backend/transaction_storage'
 require '3scale/backend/errors'
-require '3scale/backend/stats/aggregator'
+require '3scale/backend/stats'
 require '3scale/backend/usage_limit'
 require '3scale/backend/user'
 require '3scale/backend/alerts'
diff --git a/lib/3scale/backend/errors.rb b/lib/3scale/backend/errors.rb
index 5f86b29ce..1bf86b021 100644
--- a/lib/3scale/backend/errors.rb
+++ b/lib/3scale/backend/errors.rb
@@ -385,5 +385,10 @@ def initialize(id)
       end
     end
 
+    class DeleteServiceStatsValidationError < Error
+      def initialize(service_id, msg)
+        super "Delete stats job context validation error. Service: #{service_id}. Error: #{msg}"
+      end
+    end
   end
 end
diff --git a/lib/3scale/backend/stats.rb b/lib/3scale/backend/stats.rb
new file mode 100644
index 000000000..d5726fa27
--- /dev/null
+++ b/lib/3scale/backend/stats.rb
@@ -0,0 +1,3 @@
+require '3scale/backend/stats/aggregator'
+require '3scale/backend/stats/partition_generator_job'
+require '3scale/backend/stats/delete_job_def'
diff --git a/lib/3scale/backend/stats/delete_job_def.rb b/lib/3scale/backend/stats/delete_job_def.rb
new file mode 100644
index 000000000..51a7ca692
--- /dev/null
+++ b/lib/3scale/backend/stats/delete_job_def.rb
@@ -0,0 +1,65 @@
+module ThreeScale
+  module Backend
+    module Stats
+      class DeleteJobDef
+        ATTRIBUTES = %i[service_id applications metrics users from to context_info].freeze
+        private_constant :ATTRIBUTES
+        attr_reader(*ATTRIBUTES)
+
+        def self.attribute_names
+          ATTRIBUTES
+        end
+
+        def initialize(params = {})
+          ATTRIBUTES.each do |key|
+            instance_variable_set("@#{key}".to_sym, params[key]) unless params[key].nil?
+          end
+          validate
+        end
+
+        def run_async
+          Resque.enqueue(PartitionGeneratorJob, Time.now.getutc.to_f, service_id, applications,
+                         metrics, users, from, to, context_info)
+        end
+
+        def to_json
+          to_hash.to_json
+        end
+
+        def to_hash
+          Hash[ATTRIBUTES.collect { |key| [key, send(key)] }]
+        end
+
+        private
+
+        def validate
+          # from and to valid epoch times
+          raise_validation_error('from field not integer') unless from.is_a? Integer
+          raise_validation_error('from field is zero') if from.zero?
+          raise_validation_error('to field not integer') unless to.is_a? Integer
+          raise_validation_error('to field is zero') if to.zero?
+          raise_validation_error('from < to fields') if Time.at(to) < Time.at(from)
+          # application is array
+          raise_validation_error('applications field') unless applications.is_a? Array
+          raise_validation_error('applications values') unless applications.all? do |x|
+            x.is_a?(String) || x.is_a?(Integer)
+          end
+          # metrics is array
+          raise_validation_error('metrics field') unless metrics.is_a? Array
+          raise_validation_error('metrics values') unless metrics.all? do |x|
+            x.is_a?(String) || x.is_a?(Integer)
+          end
+          # users is array
+          raise_validation_error('users field') unless users.is_a? Array
+          raise_validation_error('users values') unless users.all? do |x|
+            x.is_a?(String) || x.is_a?(Integer)
+          end
+        end
+
+        def raise_validation_error(msg)
+          raise DeleteServiceStatsValidationError.new(service_id, msg)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/3scale/backend/stats/partition_generator_job.rb b/lib/3scale/backend/stats/partition_generator_job.rb
new file mode 100644
index 000000000..8d9063475
--- /dev/null
+++ b/lib/3scale/backend/stats/partition_generator_job.rb
@@ -0,0 +1,20 @@
+module ThreeScale
+  module Backend
+    module Stats
+      class PartitionGeneratorJob < BackgroundJob
+        # low priority queue
+        @queue = :main
+
+        class << self
+          def perform_logged(_enqueue_time, service_id, applications, metrics, users, from, to, context_info = {})  end
+
+          private
+
+          def enqueue_time
+            @args[0]
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/acceptance/api/internal/stats_api_spec.rb b/spec/acceptance/api/internal/stats_api_spec.rb
new file mode 100644
index 000000000..8ed605c37
--- /dev/null
+++ b/spec/acceptance/api/internal/stats_api_spec.rb
@@ -0,0 +1,68 @@
+require_relative '../../../spec_helpers/acceptance_spec_helper'
+
+resource 'Stats (prefix: /services/:service_id/stats)' do
+  header 'Accept', 'application/json'
+  header 'Content-Type', 'application/json'
+
+  let(:existing_service_id) { '10000' }
+  let(:service_id) { existing_service_id }
+  let(:provider_key) { 'statsfoo' }
+  let(:applications) { %w[1 2 3] }
+  let(:metrics) { %w[10 20 30] }
+  let(:users) { %w[100 200 300] }
+  let(:from) { Time.new(2002, 10, 31).to_i }
+  let(:to) { Time.new(2003, 10, 31).to_i }
+  let(:req_body) do
+    {
+      deletejobdef: {
+        applications: applications,
+        metrics: metrics,
+        users: users,
+        from: from,
+        to: to
+      }
+    }
+  end
+  # From and To fields are sent as string, even though they are integers in req_body
+  let(:raw_post) { req_body }
+
+  before do
+    ThreeScale::Backend::Service.save!(provider_key: provider_key, id: existing_service_id)
+  end
+
+  delete '/services/:service_id/stats' do
+    parameter :service_id, 'Service ID', required: true
+
+    context 'PartitionGeneratorJob is enqueued' do
+      before do
+        ResqueSpec.reset!
+      end
+
+      example_request 'Deleting stats' do
+        expect(status).to eq 200
+        expect(response_json['status']).to eq 'to_be_deleted'
+        expect(ThreeScale::Backend::Stats::PartitionGeneratorJob).to have_queued(anything,
+                                                                                 existing_service_id,
+                                                                                 applications,
+                                                                                 metrics, users,
+                                                                                 from, to, nil)
+      end
+    end
+
+    context 'service does not exist' do
+      let(:service_id) { existing_service_id + 'foo' }
+
+      example_request 'Deleting stats' do
+        expect(status).to eq 404
+      end
+    end
+
+    context 'invalid param sent' do
+      let(:from) { 'adfsadfasd' }
+
+      example_request 'Deleting stats' do
+        expect(status).to eq 400
+      end
+    end
+  end
+end
diff --git a/spec/unit/stats/delete_job_def_spec.rb b/spec/unit/stats/delete_job_def_spec.rb
new file mode 100644
index 000000000..ad8beac4a
--- /dev/null
+++ b/spec/unit/stats/delete_job_def_spec.rb
@@ -0,0 +1,165 @@
+require_relative '../../spec_helper'
+
+RSpec.shared_examples 'job hash is correct' do
+  it 'has service_id' do
+    expect(job).to include(service_id: service_id)
+  end
+
+  it 'has applications' do
+    expect(job).to include(applications: applications)
+  end
+
+  it 'has metrics' do
+    expect(job).to include(metrics: metrics)
+  end
+
+  it 'has users' do
+    expect(job).to include(users: users)
+  end
+
+  it 'has from' do
+    expect(job).to include(from: from)
+  end
+
+  it 'has to' do
+    expect(job).to include(to: to)
+  end
+end
+
+RSpec.shared_examples 'validation error' do
+  it 'raise validation error' do
+    expect { subject }.to raise_error(ThreeScale::Backend::DeleteServiceStatsValidationError)
+  end
+end
+
+RSpec.describe ThreeScale::Backend::Stats::DeleteJobDef do
+  let(:service_id) { 'some_service_id' }
+  let(:applications) { %w[1 2 3] }
+  let(:metrics) { %w[10 20 30] }
+  let(:users) { %w[100 200 300] }
+  let(:from) { Time.new(2002, 10, 31).to_i }
+  let(:to) { Time.new(2003, 10, 31).to_i }
+  let(:params) do
+    {
+      service_id: service_id,
+      applications: applications,
+      metrics: metrics,
+      users: users,
+      from: from,
+      to: to
+    }
+  end
+  subject { described_class.new params }
+
+  context '#initialize' do
+    context 'happy path' do
+      it 'does not raise' do should_not be_nil end
+    end
+
+    context 'from field is nil' do
+      let(:from) { nil }
+      include_examples 'validation error'
+    end
+
+    context 'from field is string' do
+      let(:from) { '12345' }
+      include_examples 'validation error'
+    end
+
+    context 'from field is zero' do
+      let(:from) { 0 }
+      include_examples 'validation error'
+    end
+
+    context 'to field is nil' do
+      let(:to) { nil }
+      include_examples 'validation error'
+    end
+
+    context 'to field is string' do
+      let(:to) { '12345' }
+      include_examples 'validation error'
+    end
+
+    context 'to field is zero' do
+      let(:to) { 0 }
+      include_examples 'validation error'
+    end
+
+    context 'to field happens before from field' do
+      let(:from) { Time.new(2005, 10, 31).to_i }
+      let(:to) { Time.new(2003, 10, 31).to_i }
+      include_examples 'validation error'
+    end
+
+    context 'applicatoins field is nil' do
+      let(:applications) { nil }
+      include_examples 'validation error'
+    end
+
+    context 'applicatoins field is not array' do
+      let(:applications) { 3 }
+      include_examples 'validation error'
+    end
+
+    context 'applicatoins field constains bad element' do
+      let(:applications) { ['3', {}, '4'] }
+      include_examples 'validation error'
+    end
+
+    context 'metrics field is nil' do
+      let(:metrics) { nil }
+      include_examples 'validation error'
+    end
+
+    context 'metrics field is not array' do
+      let(:metrics) { 3 }
+      include_examples 'validation error'
+    end
+
+    context 'metrics field constains element bad string' do
+      let(:metrics) { ['3', [], '4'] }
+      include_examples 'validation error'
+    end
+
+    context 'users field is nil' do
+      let(:users) { nil }
+      include_examples 'validation error'
+    end
+
+    context 'users field is not array' do
+      let(:users) { 3 }
+      include_examples 'validation error'
+    end
+
+    context 'users field constains element bad string' do
+      let(:users) { ['3', [], '4'] }
+      include_examples 'validation error'
+    end
+  end
+
+  context '#run_async' do
+    before do
+      ResqueSpec.reset!
+    end
+
+    it 'partition generator job is queued' do
+      subject.run_async
+      expect(ThreeScale::Backend::Stats::PartitionGeneratorJob).to have_queued(anything,
+                                                                               service_id,
+                                                                               applications,
+                                                                               metrics, users,
+                                                                               from, to, nil)
+    end
+  end
+
+  context '#to_json' do
+    let(:job) { JSON.parse(subject.to_json, symbolize_names: true) }
+    include_examples 'job hash is correct'
+  end
+
+  context '#to_hash' do
+    let(:job) { subject.to_hash }
+    include_examples 'job hash is correct'
+  end
+end